This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch release-2.4
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.4 by this push:
new 937347e4766 [FLINK-40799][table] FlinkMarkChangeLogNormalizeProgram
should cast to `RexCall` only in case of `AND`
937347e4766 is described below
commit 937347e4766e1d5fffb263592f80bd4ebd5ff441
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Thu Nov 5 12:38:36 2026 +0100
[FLINK-40799][table] FlinkMarkChangeLogNormalizeProgram should cast to
`RexCall` only in case of `AND`
---
.../stream/FlinkMarkChangelogNormalizeProgram.java | 4 +-
.../PushCalcPastChangelogNormalizeRuleTest.java | 23 ++
.../PushCalcPastChangelogNormalizeRuleTest.xml | 234 ++++++++++-----------
3 files changed, 142 insertions(+), 119 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java
index 24d3ee495c7..482142706d4 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/rules/physical/stream/FlinkMarkChangelogNormalizeProgram.java
@@ -180,12 +180,12 @@ public class FlinkMarkChangelogNormalizeProgram
changelogNormalizeContexts.stream()
.map(ChangelogNormalizeContext::getConditions)
.collect(Collectors.toList()));
- final RexCall factors = (RexCall) RexUtil.pullFactors(rexBuilder, or);
+ final RexNode factors = RexUtil.pullFactors(rexBuilder, or);
final List<RexNode> commonCondition = new ArrayList<>();
// Since we are interested in factors only then look for AND
if (factors.getKind() == SqlKind.AND) {
- for (RexNode node : factors.getOperands()) {
+ for (RexNode node : ((RexCall) factors).getOperands()) {
// If there is OR on top level then it is not a common factor
anymore
if (node.getKind() == SqlKind.OR) {
break;
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java
index a31f9c574c0..291b6fe16a5 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.java
@@ -31,8 +31,10 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import static org.apache.flink.table.api.DataTypes.BIGINT;
+import static org.apache.flink.table.api.DataTypes.BOOLEAN;
import static org.apache.flink.table.api.DataTypes.INT;
import static org.apache.flink.table.api.DataTypes.STRING;
+import static org.assertj.core.api.Assertions.assertThatCode;
/** Tests for {@link PushCalcPastChangelogNormalizeRule}. */
class PushCalcPastChangelogNormalizeRuleTest extends TableTestBase {
@@ -51,6 +53,7 @@ class PushCalcPastChangelogNormalizeRuleTest extends
TableTestBase {
.column("f1", INT().notNull())
.column("f2", STRING())
.column("f3", BIGINT().notNull())
+ .column("f4", BOOLEAN())
.primaryKey("f1")
.build())
.unboundedScanSource(ChangelogMode.upsert())
@@ -228,6 +231,26 @@ class PushCalcPastChangelogNormalizeRuleTest extends
TableTestBase {
+ " INTERSECT SELECT f3 FROM T WHERE f2 > 0 AND f2 <
10");
}
+ @Test
+ void testSelfJoinWithIdenticalBooleanColumnFilter() {
+ util.tableEnv().createTable("T", sourceDescriptor);
+ final String sql =
+ "SELECT f1.f1 FROM T AS f1 "
+ + "JOIN T AS f2 ON f1.f1 = f2.f1 "
+ + "WHERE f1.f4 AND f2.f4";
+ assertThatCode(() ->
util.tableEnv().explainSql(sql)).doesNotThrowAnyException();
+ }
+
+ @Test
+ void testSelfJoinWithDisjunctiveFilter() {
+ util.tableEnv().createTable("T", sourceDescriptor);
+ final String sql =
+ "SELECT f1.f1 FROM T AS f1 "
+ + "JOIN T AS f2 ON f1.f1 = f2.f1 "
+ + "WHERE (f1.f4 OR f1.f3 < 1) AND (f2.f4 OR f2.f3 <
1)";
+ assertThatCode(() ->
util.tableEnv().explainSql(sql)).doesNotThrowAnyException();
+ }
+
@Test
void testPartialPushDownWithTrimmedFields() {
util.tableEnv().createTable("T", sourceDescriptorWithTwoPrimaryKeys);
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
index 5e89202b442..d0a333ecaee 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/rules/physical/stream/PushCalcPastChangelogNormalizeRuleTest.xml
@@ -46,33 +46,33 @@ Calc(select=[f1, f5])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($0, 10), >($0, 1), IS NOT NULL($0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($0, 2), >($0, 0), IS NOT NULL($0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(<($0, 4), >($0, 2), IS NOT NULL($0))])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
- :- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f0, Sarg[(1..10)])])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
+ :- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f0, Sarg[(1..10)])])
: +- ChangelogNormalize(key=[f1], condition=[f0 IS NOT
NULL])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
- : +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f0, Sarg[(0..2)])])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f0,
Sarg[(0..2)])])
: +- Reused(reference_id=[1])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f0, Sarg[(2..4)])])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f0,
Sarg[(2..4)])])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -86,35 +86,35 @@ GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1,
f2, f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T2]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
+- LogicalTableScan(table=[[default_catalog, default_database, T3]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
:- ChangelogNormalize(key=[f1], condition=[SEARCH(f3, Sarg[(0..1)])])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
: +- ChangelogNormalize(key=[f1], condition=[SEARCH(f3,
Sarg[(0..1)])])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[f0, f1, f2, f3])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[f0, f1, f2, f3, f4])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+- ChangelogNormalize(key=[f1], condition=[SEARCH(f3,
Sarg[(0..1)])])
+- Exchange(distribution=[hash[f1]])
- +- TableSourceScan(table=[[default_catalog,
default_database, T3]], fields=[f0, f1, f2, f3])
+ +- TableSourceScan(table=[[default_catalog,
default_database, T3]], fields=[f0, f1, f2, f3, f4])
]]>
</Resource>
</TestCase>
@@ -127,33 +127,33 @@ INTERSECT SELECT * FROM T WHERE f3 > 0]]>
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[<($3, 1)])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[<($3, 3)])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[>($3, 0)])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
- :- Calc(select=[f0, f1, f2, f3], where=[(f3 < 1)])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
+ :- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 < 1)])
: +- ChangelogNormalize(key=[f1])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
- : +- Calc(select=[f0, f1, f2, f3], where=[(f3 < 3)])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 < 3)])
: +- Reused(reference_id=[1])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Calc(select=[f0, f1, f2, f3], where=[(f3 > 0)])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 > 0)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -167,33 +167,33 @@ GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1,
f2, f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 3), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(>($3, 0), <($3, 10))])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
- :- Calc(select=[f0, f1, f2, f3], where=[(f3 < 1)])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
+ :- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 < 1)])
: +- ChangelogNormalize(key=[f1], condition=[(f3 > 0)])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
- : +- Calc(select=[f0, f1, f2, f3], where=[(f3 < 3)])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 < 3)])
: +- Reused(reference_id=[1])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Calc(select=[f0, f1, f2, f3], where=[(f3 < 10)])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[(f3 < 10)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -207,28 +207,28 @@ GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1,
f2, f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(<($3, 1), >($3, 0))])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
:- ChangelogNormalize(key=[f1], condition=[SEARCH(f3,
Sarg[(0..1)])])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])(reuse_id=[2])
+ : +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])(reuse_id=[2])
: +- Reused(reference_id=[1])
+- Reused(reference_id=[2])
]]>
@@ -249,7 +249,7 @@ LogicalProject(f1=[$1], f2=[$2])
ChangelogNormalize(key=[f1])
+- Exchange(distribution=[hash[f1]])
+- Calc(select=[f1, f2])
- +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3, f4])
]]>
</Resource>
</TestCase>
@@ -365,38 +365,38 @@ GroupAggregate(groupBy=[f3], select=[f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T2]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
+- LogicalTableScan(table=[[default_catalog, default_database, T3]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
:- ChangelogNormalize(key=[f1])
: +- Exchange(distribution=[hash[f1]])
- : +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f1, Sarg[(0..1)])])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f1,
Sarg[(0..1)])])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
: +- ChangelogNormalize(key=[f1])
: +- Exchange(distribution=[hash[f1]])
- : +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f1,
Sarg[(0..1)])])
- : +- TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[f0, f1, f2, f3])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f1,
Sarg[(0..1)])])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, T2]], fields=[f0, f1, f2, f3, f4])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+- ChangelogNormalize(key=[f1])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f1,
Sarg[(0..1)])])
- +- TableSourceScan(table=[[default_catalog,
default_database, T3]], fields=[f0, f1, f2, f3])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f1,
Sarg[(0..1)])])
+ +- TableSourceScan(table=[[default_catalog,
default_database, T3]], fields=[f0, f1, f2, f3, f4])
]]>
</Resource>
</TestCase>
@@ -409,33 +409,33 @@ INTERSECT SELECT * FROM T WHERE f1 > 0]]>
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[<($1, 1)])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[<($1, 3)])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[>($1, 0)])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
- :- Calc(select=[f0, f1, f2, f3], where=[(f1 < 1)])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
+ :- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 < 1)])
: +- ChangelogNormalize(key=[f1])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
- : +- Calc(select=[f0, f1, f2, f3], where=[(f1 < 3)])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 < 3)])
: +- Reused(reference_id=[1])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Calc(select=[f0, f1, f2, f3], where=[(f1 > 0)])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 > 0)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -449,29 +449,29 @@ GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1,
f2, f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
:- ChangelogNormalize(key=[f1])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- Calc(select=[f0, f1, f2, f3], where=[SEARCH(f1, Sarg[(0..1)])])
- : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])(reuse_id=[2])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[SEARCH(f1,
Sarg[(0..1)])])
+ : +- TableSourceScan(table=[[default_catalog, default_database,
T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])(reuse_id=[2])
: +- Reused(reference_id=[1])
+- Reused(reference_id=[2])
]]>
@@ -486,34 +486,34 @@ GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1,
f2, f3])
<Resource name="ast">
<![CDATA[
LogicalUnion(all=[false])
-:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+:- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 1), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
+- LogicalIntersect(all=[false])
- :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ :- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
: +- LogicalFilter(condition=[AND(<($1, 3), >($1, 0))])
: +- LogicalTableScan(table=[[default_catalog, default_database, T]])
- +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+ +- LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[AND(>($1, 0), <($1, 10))])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
</Resource>
<Resource name="optimized exec plan">
<![CDATA[
-GroupAggregate(groupBy=[f0, f1, f2, f3], select=[f0, f1, f2, f3])
-+- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Union(all=[true], union=[f0, f1, f2, f3])
- :- Calc(select=[f0, f1, f2, f3], where=[(f1 < 1)])
+GroupAggregate(groupBy=[f0, f1, f2, f3, f4], select=[f0, f1, f2, f3, f4])
++- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Union(all=[true], union=[f0, f1, f2, f3, f4])
+ :- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 < 1)])
: +- ChangelogNormalize(key=[f1])(reuse_id=[1])
: +- Exchange(distribution=[hash[f1]])
- : +- Calc(select=[f0, f1, f2, f3], where=[(f1 > 0)])
- : +- TableSourceScan(table=[[default_catalog,
default_database, T]], fields=[f0, f1, f2, f3])
- +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30))], select=[f0,
f1, f2, f3], leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
- :- Exchange(distribution=[hash[f0, f1, f2, f3]])
- : +- Calc(select=[f0, f1, f2, f3], where=[(f1 < 3)])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 > 0)])
+ : +- TableSourceScan(table=[[default_catalog,
default_database, T]], fields=[f0, f1, f2, f3, f4])
+ +- Join(joinType=[LeftSemiJoin], where=[(IS NOT DISTINCT FROM(f0, f00)
AND (f1 = f10) AND IS NOT DISTINCT FROM(f2, f20) AND (f3 = f30) AND IS NOT
DISTINCT FROM(f4, f40))], select=[f0, f1, f2, f3, f4],
leftInputSpec=[JoinKeyContainsUniqueKey],
rightInputSpec=[JoinKeyContainsUniqueKey])
+ :- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ : +- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 < 3)])
: +- Reused(reference_id=[1])
- +- Exchange(distribution=[hash[f0, f1, f2, f3]])
- +- Calc(select=[f0, f1, f2, f3], where=[(f1 < 10)])
+ +- Exchange(distribution=[hash[f0, f1, f2, f3, f4]])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[(f1 < 10)])
+- Reused(reference_id=[1])
]]>
</Resource>
@@ -536,7 +536,7 @@ Calc(select=[f1, f1 AS sum])
+- ChangelogNormalize(key=[f1], condition=[OR(IS NULL(f2), >(f1, 3))])
+- Exchange(distribution=[hash[f1]])
+- Calc(select=[f1, f2], where=[<(f1, 10)])
- +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3, f4])
]]>
</Resource>
</TestCase>
@@ -567,7 +567,7 @@ Calc(select=[f0, f1])
</Resource>
<Resource name="ast">
<![CDATA[
-LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
+LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3], f4=[$4])
+- LogicalFilter(condition=[<($1, 1)])
+- LogicalTableScan(table=[[default_catalog, default_database, T]])
]]>
@@ -576,8 +576,8 @@ LogicalProject(f0=[$0], f1=[$1], f2=[$2], f3=[$3])
<![CDATA[
ChangelogNormalize(key=[f1])
+- Exchange(distribution=[hash[f1]])
- +- Calc(select=[f0, f1, f2, f3], where=[<(f1, 1)])
- +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3])
+ +- Calc(select=[f0, f1, f2, f3, f4], where=[<(f1, 1)])
+ +- TableSourceScan(table=[[default_catalog, default_database, T]],
fields=[f0, f1, f2, f3, f4])
]]>
</Resource>
</TestCase>
