This is an automated email from the ASF dual-hosted git repository. snuyanzin pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit f353726e9dbd8f7e5000eaf1fc56998591b2ef3e Author: Sergey Nuyanzin <[email protected]> AuthorDate: Sat Jun 13 21:08:58 2026 +0200 [FLINK-36602][table] Enable CALCITE-6044 optimization --- .../calcite/rel/metadata/RelMdPredicates.java | 35 +++++++++++++++++++--- .../planner/plan/rules/FlinkBatchRuleSets.scala | 4 +-- .../planner/plan/rules/FlinkStreamRuleSets.scala | 4 +-- .../table/planner/plan/batch/sql/CalcTest.xml | 2 +- .../table/planner/plan/batch/sql/TableSinkTest.xml | 6 ++-- .../table/planner/plan/batch/sql/ValuesTest.xml | 3 +- .../rules/logical/FlinkPruneEmptyRulesTest.xml | 6 ++-- .../table/planner/plan/stream/sql/ValuesTest.xml | 3 +- .../rules/logical/FlinkPruneEmptyRulesTest.scala | 6 ++-- 9 files changed, 50 insertions(+), 19 deletions(-) diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java index de3d4a6deaa..c18ec7dc6a0 100644 --- a/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java +++ b/flink-table/flink-table-planner/src/main/java/org/apache/calcite/rel/metadata/RelMdPredicates.java @@ -75,10 +75,12 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; +import java.util.Objects; import java.util.Set; import java.util.SortedMap; import java.util.TreeMap; import java.util.stream.Collectors; +import java.util.stream.IntStream; import static java.util.Objects.requireNonNull; @@ -123,8 +125,6 @@ import static java.util.Objects.requireNonNull; * The class contains CALCITE-6599 fix from 1.39.0, should be dropped after upgrade to that version * * <p>Lines 181 ~ 217 - * - * <p>Lines 560 ~ 562 disable CALCITE-6044 optimization */ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicates> { public static final RelMetadataProvider SOURCE = @@ -557,9 +557,36 @@ public class RelMdPredicates implements MetadataHandler<BuiltInMetadata.Predicat * null)} are {@code 'y = 2'} and {@code 'z is null'}. */ public RelOptPredicateList getPredicates(Values values, RelMetadataQuery mq) { - // FLINK MODIFICATION BEGIN + ImmutableList<ImmutableList<RexLiteral>> tuples = values.tuples; + if (tuples.size() > 0) { + Set<Integer> constants = new HashSet<>(); + IntStream.range(0, tuples.size()).boxed().forEach(constants::add); + List<RexLiteral> firstTuple = new ArrayList<>(tuples.get(0)); + for (ImmutableList<RexLiteral> tuple : tuples) { + if (constants.isEmpty()) { + break; + } + for (int i = 0; i < tuple.size(); i++) { + if (!Objects.equals(tuple.get(i), firstTuple.get(i))) { + constants.remove(i); + } + } + } + RexBuilder rexBuilder = values.getCluster().getRexBuilder(); + List<RexNode> predicates = new ArrayList<>(); + for (int i = 0; i < firstTuple.size(); i++) { + if (constants.contains(i)) { + RexLiteral literal = firstTuple.get(i); + predicates.add( + rexBuilder.makeCall( + SqlStdOperatorTable.EQUALS, + rexBuilder.makeInputRef(literal.getType(), i), + literal)); + } + } + return RelOptPredicateList.of(rexBuilder, predicates); + } return RelOptPredicateList.EMPTY; - // FLINK MODIFICATION END } // CHECKSTYLE: IGNORE 1 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 f102f593ac3..58aeac3b0fa 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 @@ -195,11 +195,11 @@ object FlinkBatchRuleSets { /** RuleSet to prune empty results rules */ val PRUNE_EMPTY_RULES: RuleSet = RuleSets.ofList( + PruneEmptyRules.PROJECT_INSTANCE, + PruneEmptyRules.FILTER_INSTANCE, FlinkPruneEmptyRules.UNION_INSTANCE, PruneEmptyRules.INTERSECT_INSTANCE, FlinkPruneEmptyRules.MINUS_INSTANCE, - PruneEmptyRules.PROJECT_INSTANCE, - PruneEmptyRules.FILTER_INSTANCE, PruneEmptyRules.SORT_INSTANCE, PruneEmptyRules.AGGREGATE_INSTANCE, PruneEmptyRules.JOIN_LEFT_INSTANCE, 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 25ad791f53c..4f07715e78c 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 @@ -191,11 +191,11 @@ object FlinkStreamRuleSets { /** RuleSet to prune empty results rules */ val PRUNE_EMPTY_RULES: RuleSet = RuleSets.ofList( + PruneEmptyRules.PROJECT_INSTANCE, + PruneEmptyRules.FILTER_INSTANCE, FlinkPruneEmptyRules.UNION_INSTANCE, PruneEmptyRules.INTERSECT_INSTANCE, FlinkPruneEmptyRules.MINUS_INSTANCE, - PruneEmptyRules.PROJECT_INSTANCE, - PruneEmptyRules.FILTER_INSTANCE, PruneEmptyRules.SORT_INSTANCE, PruneEmptyRules.AGGREGATE_INSTANCE, PruneEmptyRules.JOIN_LEFT_INSTANCE, diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/CalcTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/CalcTest.xml index ec4dfbdbdaf..77be840b0ce 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/CalcTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/CalcTest.xml @@ -115,7 +115,7 @@ LogicalProject(EXPR$0=[$0], EXPR$1=[CAST($1):VARCHAR(2147483647) CHARACTER SET " </Resource> <Resource name="optimized exec plan"> <![CDATA[ -Calc(select=[a AS EXPR$0, CAST(b AS VARCHAR(2147483647)) AS EXPR$1]) +Calc(select=[3 AS EXPR$0, CAST(b AS VARCHAR(2147483647)) AS EXPR$1]) +- Values(tuples=[[{ 3, _UTF-16LE'c' }]], values=[a, b]) ]]> </Resource> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/TableSinkTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/TableSinkTest.xml index 9b802a08292..e27d56c2db9 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/TableSinkTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/TableSinkTest.xml @@ -30,14 +30,16 @@ Sink(table=[default_catalog.default_database.MyCtasSource], fields=[b, c, d]) +- Calc(select=[b, c, d]) +- Sort(orderBy=[a ASC]) +- Exchange(distribution=[single]) - +- Values(tuples=[[{ 1, 1, 2, _UTF-16LE'd1' }]], values=[a, b, c, d]) + +- Calc(select=[b, c, d, 1 AS a]) + +- Values(tuples=[[{ 1, 1, 2, _UTF-16LE'd1' }]], values=[a, b, c, d]) == Optimized Execution Plan == Sink(table=[default_catalog.default_database.MyCtasSource], fields=[b, c, d]) +- Calc(select=[b, c, d]) +- Sort(orderBy=[a ASC]) +- Exchange(distribution=[single]) - +- Values(tuples=[[{ 1, 1, 2, _UTF-16LE'd1' }]], values=[a, b, c, d]) + +- Calc(select=[b, c, d, 1 AS a]) + +- Values(tuples=[[{ 1, 1, 2, _UTF-16LE'd1' }]], values=[a, b, c, d]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/ValuesTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/ValuesTest.xml index 8a1d0b9f3e3..1e47b06db00 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/ValuesTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/ValuesTest.xml @@ -118,7 +118,8 @@ LogicalProject(a=[$0], b=[$1], c=[$2]) </Resource> <Resource name="optimized exec plan"> <![CDATA[ -Values(tuples=[[{ 1, 2, 3 }]], values=[a, b, c]) +Calc(select=[1 AS a, b, c]) ++- Values(tuples=[[{ 1, 2, 3 }]], values=[a, b, c]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.xml index abf88a5994a..f5c44f2083f 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.xml @@ -188,9 +188,9 @@ LogicalMinus(all=[false]) <Resource name="optimized rel plan"> <![CDATA[ LogicalMinus(all=[false]) -:- LogicalProject(x=[$0], y=[$1]) +:- LogicalProject(x=[30], y=[$1]) : +- LogicalValues(tuples=[[{ 30, 3 }]]) -+- LogicalProject(EXPR$0=[$0], EXPR$1=[$1]) ++- LogicalProject(EXPR$0=[40], EXPR$1=[$1]) +- LogicalValues(tuples=[[{ 40, 4 }]]) ]]> </Resource> @@ -216,7 +216,7 @@ LogicalMinus(all=[false]) <Resource name="optimized rel plan"> <![CDATA[ LogicalAggregate(group=[{0, 1}]) -+- LogicalProject(X=[$0], Y=[$1]) ++- LogicalProject(X=[30], Y=[3]) +- LogicalValues(tuples=[[{ 30, 3 }, { 30, 3 }]]) ]]> </Resource> diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ValuesTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ValuesTest.xml index 098ecd36f34..a934bbecf9b 100644 --- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ValuesTest.xml +++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/stream/sql/ValuesTest.xml @@ -100,7 +100,8 @@ LogicalProject(a=[$0], b=[$1], c=[$2]) </Resource> <Resource name="optimized exec plan"> <![CDATA[ -Values(tuples=[[{ 1, 2, 3 }]]) +Calc(select=[1 AS a, b, c]) ++- Values(tuples=[[{ 1, 2, 3 }]]) ]]> </Resource> </TestCase> diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.scala index 74e9ad050fe..d35c83a1a6f 100644 --- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.scala +++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/rules/logical/FlinkPruneEmptyRulesTest.scala @@ -47,14 +47,14 @@ class FlinkPruneEmptyRulesTest extends TableTestBase { CoreRules.FILTER_REDUCE_EXPRESSIONS, CoreRules.PROJECT_REDUCE_EXPRESSIONS, CoreRules.FILTER_SET_OP_TRANSPOSE, - CoreRules.FILTER_PROJECT_TRANSPOSE, + FlinkFilterProjectTransposeRule.INSTANCE, CoreRules.PROJECT_MERGE, CoreRules.PROJECT_FILTER_VALUES_MERGE, + PruneEmptyRules.PROJECT_INSTANCE, + PruneEmptyRules.FILTER_INSTANCE, FlinkPruneEmptyRules.UNION_INSTANCE, PruneEmptyRules.INTERSECT_INSTANCE, FlinkPruneEmptyRules.MINUS_INSTANCE, - PruneEmptyRules.PROJECT_INSTANCE, - PruneEmptyRules.FILTER_INSTANCE, PruneEmptyRules.SORT_INSTANCE, PruneEmptyRules.AGGREGATE_INSTANCE, PruneEmptyRules.JOIN_LEFT_INSTANCE,
