diff --git a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java index a02915f1002939..45c9f965bb5e63 100644 --- a/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java +++ b/flink-table/flink-table-api-java/src/main/java/org/apache/flink/table/api/config/OptimizerConfigOptions.java @@ -400,6 +400,17 @@ public class OptimizerConfigOptions { .withDescription( "Strategy for optimizing the delta-join. Only AUTO, FORCE or NONE can be set. Default it AUTO."); + @Documentation.TableOption(execMode = Documentation.ExecMode.BATCH_STREAMING) + public static final ConfigOption + TABLE_OPTIMIZER_PROJECT_FILTER_TRANSPOSE_RULE = + key("table.optimizer.project-filter-transpose-rule") + .enumType(ProjectFilterTransposeRule.class) + .defaultValue(ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE) + .withDescription( + "Selects which Calcite ProjectFilterTransposeRule variant the optimizer " + + "uses when pushing a Project past a Filter in the " + + "PROJECT_REWRITE phase."); + /** Strategy for handling non-deterministic updates. */ @PublicEvolving public enum NonDeterministicUpdateStrategy { @@ -496,4 +507,37 @@ public InlineElement getDescription() { return description; } } + + /** + * Variant of Calcite's ProjectFilterTransposeRule used when pushing a Project past a Filter. + */ + @PublicEvolving + public enum ProjectFilterTransposeRule implements DescribedEnum { + PROJECT_FILTER_TRANSPOSE( + text( + "Default variant. Pushes a Project past a Filter, splitting expressions " + + "so that only the columns referenced by the Filter remain below it.")), + PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS( + text( + "Pushes the Project past the Filter without splitting expressions. " + + "Each top-level Project expression is preserved as a whole and " + + "either pushed below the Filter or kept above it." + + "NOTE: Required for proper nested projection + filtering")), + PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS( + text( + "Pushes the Project past the Filter only when the Project as a whole can be " + + "moved below the Filter; the Project is treated as atomic and is " + + "never split.")); + + private final InlineElement description; + + ProjectFilterTransposeRule(InlineElement description) { + this.description = description; + } + + @Override + public InlineElement getDescription() { + return description; + } + } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkBatchProgram.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkBatchProgram.scala index cc46f3aeedc562..794ae4e4d7f587 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkBatchProgram.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkBatchProgram.scala @@ -22,7 +22,12 @@ import org.apache.flink.table.api.config.OptimizerConfigOptions import org.apache.flink.table.planner.plan.nodes.FlinkConventions import org.apache.flink.table.planner.plan.rules.FlinkBatchRuleSets +import org.apache.calcite.plan.RelOptRule import org.apache.calcite.plan.hep.HepMatchOrder +import org.apache.calcite.rel.rules.CoreRules +import org.apache.calcite.tools.{RuleSet, RuleSets} + +import scala.collection.JavaConverters._ /** Defines a sequence of programs to optimize flink batch table plan. */ object FlinkBatchProgram { @@ -45,6 +50,7 @@ object FlinkBatchProgram { def buildProgram(tableConfig: ReadableConfig): FlinkChainedProgram[BatchOptimizeContext] = { val chainedProgram = new FlinkChainedProgram[BatchOptimizeContext]() + val projectFilterTransposeRule = pickProjectFilterTransposeRule(tableConfig) chainedProgram.addLast( // rewrite sub-queries to joins @@ -239,17 +245,23 @@ object FlinkBatchProgram { .build() ) - // window rewrite + // In PROJECT_REWRITE we utilize user desired transposition method chainedProgram.addLast( PROJECT_REWRITE, FlinkHepRuleSetProgramBuilder.newBuilder .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_COLLECTION) .setHepMatchOrder(HepMatchOrder.BOTTOM_UP) - .add(FlinkBatchRuleSets.PROJECT_RULES) + .add( + substituteRule( + FlinkBatchRuleSets.PROJECT_RULES, + CoreRules.PROJECT_FILTER_TRANSPOSE, + projectFilterTransposeRule)) .build() ) // optimize the logical plan + // Note: the project-filter-transpose variant is intentionally NOT appended to the Volcano + // LOGICAL phase. Prevents infinite oscillation with optimization from PROJECT_REWRITE chainedProgram.addLast( LOGICAL, FlinkVolcanoProgramBuilder.newBuilder @@ -298,4 +310,23 @@ object FlinkBatchProgram { chainedProgram } + + private def pickProjectFilterTransposeRule(tableConfig: ReadableConfig): RelOptRule = { + tableConfig.get(OptimizerConfigOptions.TABLE_OPTIMIZER_PROJECT_FILTER_TRANSPOSE_RULE) match { + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE => + CoreRules.PROJECT_FILTER_TRANSPOSE + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS => + CoreRules.PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS => + CoreRules.PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS + } + } + + private def substituteRule(base: RuleSet, oldRule: RelOptRule, newRule: RelOptRule): RuleSet = { + if (oldRule eq newRule) { + base + } else { + RuleSets.ofList((base.asScala.filterNot(_ eq oldRule) ++ Seq(newRule)).asJava) + } + } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala index 8ec32c242fa48a..0079414e535fdb 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/optimize/program/FlinkStreamProgram.scala @@ -24,7 +24,12 @@ import org.apache.flink.table.planner.plan.rules.FlinkStreamRuleSets import org.apache.flink.table.planner.plan.rules.logical.EventTimeTemporalJoinRewriteRule import org.apache.flink.table.planner.plan.rules.physical.stream.{FlinkDuplicateChangesTraitInitProgram, FlinkMarkChangelogNormalizeProgram} +import org.apache.calcite.plan.RelOptRule import org.apache.calcite.plan.hep.HepMatchOrder +import org.apache.calcite.rel.rules.CoreRules +import org.apache.calcite.tools.{RuleSet, RuleSets} + +import scala.collection.JavaConverters._ /** Defines a sequence of programs to optimize for stream table plan. */ object FlinkStreamProgram { @@ -45,6 +50,7 @@ object FlinkStreamProgram { def buildProgram(tableConfig: ReadableConfig): FlinkChainedProgram[StreamOptimizeContext] = { val chainedProgram = new FlinkChainedProgram[StreamOptimizeContext]() + val projectFilterTransposeRule = pickProjectFilterTransposeRule(tableConfig) // rewrite sub-queries to joins chainedProgram.addLast( @@ -248,17 +254,23 @@ object FlinkStreamProgram { .build() ) - // project rewrite + // In PROJECT_REWRITE we utilize user desired transposition method chainedProgram.addLast( PROJECT_REWRITE, FlinkHepRuleSetProgramBuilder.newBuilder .setHepRulesExecutionType(HEP_RULES_EXECUTION_TYPE.RULE_COLLECTION) .setHepMatchOrder(HepMatchOrder.BOTTOM_UP) - .add(FlinkStreamRuleSets.PROJECT_RULES) + .add( + substituteRule( + FlinkStreamRuleSets.PROJECT_RULES, + CoreRules.PROJECT_FILTER_TRANSPOSE, + projectFilterTransposeRule)) .build() ) // optimize the logical plan + // Note: the project-filter-transpose variant is intentionally NOT appended to the Volcano + // LOGICAL phase. Prevents infinite oscillation with optimization from PROJECT_REWRITE chainedProgram.addLast( LOGICAL, FlinkVolcanoProgramBuilder.newBuilder @@ -360,4 +372,23 @@ object FlinkStreamProgram { chainedProgram } + + private def pickProjectFilterTransposeRule(tableConfig: ReadableConfig): RelOptRule = { + tableConfig.get(OptimizerConfigOptions.TABLE_OPTIMIZER_PROJECT_FILTER_TRANSPOSE_RULE) match { + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE => + CoreRules.PROJECT_FILTER_TRANSPOSE + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS => + CoreRules.PROJECT_FILTER_TRANSPOSE_WHOLE_EXPRESSIONS + case OptimizerConfigOptions.ProjectFilterTransposeRule.PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS => + CoreRules.PROJECT_FILTER_TRANSPOSE_WHOLE_PROJECT_EXPRESSIONS + } + } + + private def substituteRule(base: RuleSet, oldRule: RelOptRule, newRule: RelOptRule): RuleSet = { + if (oldRule eq newRule) { + base + } else { + RuleSets.ofList((base.asScala.filterNot(_ eq oldRule) ++ Seq(newRule)).asJava) + } + } } diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala index 7ff8a16a70e4a6..6edaa79a089be3 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkBatchRuleSets.scala @@ -204,9 +204,8 @@ object FlinkBatchRuleSets { CoreRules.AGGREGATE_VALUES ) - /** RuleSet about project */ val PROJECT_RULES: RuleSet = RuleSets.ofList( - // push a projection past a filter + // push a projection past a filter (safe default; HEP swaps in the configured variant) CoreRules.PROJECT_FILTER_TRANSPOSE, // push a projection to the children of a non semi/anti join // push all expressions to handle the time indicator correctly diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala index 05de12a6950a98..5d970257aa68c3 100644 --- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala +++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/FlinkStreamRuleSets.scala @@ -202,9 +202,8 @@ object FlinkStreamRuleSets { CoreRules.AGGREGATE_VALUES ) - /** RuleSet about project */ val PROJECT_RULES: RuleSet = RuleSets.ofList( - // push a projection past a filter + // push a projection past a filter (safe default; HEP swaps in the configured variant) CoreRules.PROJECT_FILTER_TRANSPOSE, // push a projection to the children of a non semi/anti join // push all expressions to handle the time indicator correctly