github-actions[bot] commented on code in PR #63763:
URL: https://github.com/apache/doris/pull/63763#discussion_r3679443205
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunction.java:
##########
@@ -299,6 +439,125 @@ private void createSlotMapping(List<CatalogRelation>
outerTables, List<CatalogRe
}
}
+ /**
+ * Verify that every pair of outer/inner scans sharing the same table ID
+ * has equivalent row-domain semantics. Table.getId() equality alone is
+ * insufficient — wrappers (RowBinlogTableWrapper, OlapTableStreamWrapper)
+ * share the base table ID but read different rows, and even two scans of
+ * the same base table can differ by {@code @incr} params, partition/tablet
+ * selection, index, snapshot, or table sample.
+ */
+ private static boolean isSameScanDomain(List<CatalogRelation> outerTables,
+ List<CatalogRelation> innerTables)
{
+ for (CatalogRelation outerTable : outerTables) {
+ for (CatalogRelation innerTable : innerTables) {
+ if (innerTable.getTable().getId() !=
outerTable.getTable().getId()) {
+ continue;
+ }
+ // Different table objects with the same ID means one is a
+ // wrapper (RowBinlogTableWrapper, OlapTableStreamWrapper, …)
+ // and the other is the base table. They read different rows.
+ if (innerTable.getTable() != outerTable.getTable()) {
+ return false;
+ }
+ // If both are LogicalOlapScan, compare scan-domain properties.
+ if (innerTable instanceof LogicalOlapScan
+ && outerTable instanceof LogicalOlapScan) {
+ LogicalOlapScan innerScan = (LogicalOlapScan) innerTable;
+ LogicalOlapScan outerScan = (LogicalOlapScan) outerTable;
+ // Different selected index → different row set.
+ if (innerScan.getSelectedIndexId() !=
outerScan.getSelectedIndexId()) {
+ return false;
+ }
+ // Different scan params (e.g. @incr) → different row set.
+ if (!Objects.equals(
+ innerScan.getScanParams(),
outerScan.getScanParams())) {
+ return false;
+ }
+ // Different table sample → different row set.
+ if (!Objects.equals(
Review Comment:
[P1] Reject equal non-repeatable samples
A reduced accepted plan is:
```text
Filter(f.k = d.k, f.v > scalar_sum)
Apply(correlation = d.k)
CrossJoin(Scan fact TABLESAMPLE(n ROWS) f, Scan unique_dim d)
Aggregate(SUM(f2.v))
Filter(f2.k = d.k)
Scan fact TABLESAMPLE(n ROWS) f2
```
Without `REPEATABLE`, `LogicalPlanBuilder` gives both samples `seek = -1`,
so their `TableSample` values compare equal here. At execution, however, each
`OlapScanNode` resolves `-1` with its own `new SecureRandom()` partition/tablet
seek. The original outer and scalar-subquery scans can therefore read different
samples, while this rewrite removes the inner scan and computes the window only
over the outer sample, changing the result.
This is distinct from the earlier `@incr` row-domain thread: the scan
descriptors are equal here but the materialized runtime row domains are not.
Please reject non-repeatable samples (or otherwise share/prove the same sampled
row set) and add a negative case.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunction.java:
##########
@@ -180,21 +216,47 @@ private boolean checkAggregate() {
if (functions.size() != 1) {
return false;
}
- return functions.stream().allMatch(f -> f instanceof
SupportWindowAnalytic && !f.isDistinct());
+ if (!functions.stream().allMatch(f -> f instanceof
SupportWindowAnalytic && !f.isDistinct())) {
+ return false;
+ }
+ // Reject aggregates whose arguments contain volatile or
+ // NoneMovableFunction expressions (e.g. COUNT(assert_true(...))).
+ // The rewrite places some outer predicates below the window,
+ // changing which rows reach the side-effecting function inside
+ // the aggregate and potentially suppressing expected errors.
+ return functions.stream().noneMatch(
Review Comment:
[P1] Validate unsafe wrappers around the aggregate output
This guard only walks the collected `AggregateFunction`, so it misses an
unsafe ancestor in the aggregate output:
```text
Filter(d.tag > 0, f.k = d.k, TRUE = scalar_flag)
Apply(correlation = d.k)
CrossJoin(Scan fact f, Scan unique_dim d)
Aggregate(assert_true(SUM(f2.v) > 0, 'bad') AS scalar_flag)
Filter(f2.k = d.k)
Scan fact f2
```
`checkAggregate()` sees only the safe `SUM`, but `rewrite()` takes the
complete `aggOut.child(0)`, replaces `SUM` with the window slot, and inlines
`assert_true(window_sum > 0, 'bad')` into the top comparison. Meanwhile `d.tag
> 0` and the matched join predicate go below the window. For a failing key
whose dimension row has `tag = 0`, the original Apply evaluates the scalar
aggregate output and raises before the outer filter; the rewritten lower filter
removes that key before the assertion, suppressing the error.
This is distinct from the existing `COUNT(assert_true(...))` thread: there
the unsafe call is an aggregate argument and this new subtree check catches it;
here it wraps `SUM` and lies outside the inspected subtree. Please validate the
complete aggregate output expression (or reject unsafe wrappers) and add a
negative regression.
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunctionTest.java:
##########
@@ -338,6 +368,2846 @@ public void testNotMatchTheRule() {
}
}
+ @Test
+ public void testWindowPartitionsByOuterOnlyRelationSlots() throws
Exception {
+ // Use TPC-H Q17: correlated table is part (p_partkey is unique via
constraint),
+ // fact table is lineitem. The window PARTITION BY should contain all
output
+ // columns of the correlated table (part).
+ String sql = TPCHUtils.Q17;
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ Assertions.assertEquals(1, windows.size());
+
+ LogicalWindow<Plan> window = windows.get(0);
+ List<NamedExpression> windowExpressions =
window.getWindowExpressions();
+ Assertions.assertEquals(1, windowExpressions.size());
+
+ WindowExpression windowExpression = (WindowExpression)
windowExpressions.get(0).child(0);
+ Set<String> partitionKeys =
windowExpression.getPartitionKeys().stream()
+ .map(Expression.class::cast)
+ .filter(Slot.class::isInstance)
+ .map(Slot.class::cast)
+ .map(Slot::getName)
+ .collect(Collectors.toSet());
+ // The window PARTITION BY should include the correlated column
(p_partkey)
+ Assertions.assertTrue(partitionKeys.contains("p_partkey"),
+ "Expected partition keys to contain p_partkey, got: " +
partitionKeys);
+ }
+
+ @Test
+ public void testSharedTablePredicatesStayAboveWindow() throws Exception {
+ createTable("CREATE TABLE fact_split (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_split (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ // Add UNIQUE constraint so the rule matches
+ addConstraint("alter table dim_split add constraint uq_dim_split_k
unique (k)");
+
+ // Query with extra predicate on shared table (f.v > 6).
+ // This predicate must stay ABOVE the window, otherwise the window
+ // function would aggregate fewer rows than the original scalar
subquery.
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_split f, dim_split d "
+ + "WHERE f.k = d.k "
+ + " AND f.v > 6"
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_split f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule should match and produce a window
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule should produce a window for this query");
+
+ // Collect the ExprIds of the shared table (fact_split – appears in
both
+ // outer and inner plans). Predicates that reference ONLY these
ExprIds
+ // (shared-table-only filters) must stay ABOVE the window, otherwise
the
+ // window function would see fewer rows than the original scalar
subquery.
+ List<CatalogRelation> rels =
plan.collectToList(CatalogRelation.class::isInstance);
+ Set<ExprId> sharedExprIds = rels.stream()
+ .filter(r -> r.getTable().getName().equals("fact_split"))
+ .flatMap(r -> r.getOutputExprIdSet().stream())
+ .collect(Collectors.toSet());
+
+ // Verify that no filter below the window contains a predicate whose
+ // input ExprIds are all from the shared table. Such predicates
+ // (e.g. f.v > 6) must have been placed above the window.
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ LogicalWindow<Plan> window = windows.get(0);
+ Plan belowWindow = window.child(0);
+ List<LogicalFilter<Plan>> belowFilters = belowWindow
+ .collectToList(LogicalFilter.class::isInstance);
+ for (LogicalFilter<Plan> f : belowFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (!conjExprIds.isEmpty() &&
sharedExprIds.containsAll(conjExprIds)) {
+ Assertions.fail(
+ "Shared-table-only predicate should not be below
window: " + conj.toSql());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testMixedSharedOuterPredicatesStayAboveWindow() throws
Exception {
+ // Mixed predicates that reference both shared-table and
outer-only-table
+ // columns (e.g. f.v > d.tag) must stay ABOVE the window. Pushing them
+ // below would restrict the rows seen by the window function, producing
+ // a different aggregate than the original scalar subquery.
+ //
+ // Input plan shape:
+ // Filter(f.v > d.tag, f.v * 2 > sum_alias) ← mixed +
correlated
+ // Apply(correlation: d.k)
+ // CrossJoin
+ // Scan fact f
+ // Scan dim d -- d.k is unique (constraint)
+ // Aggregate(sum(f2.v) AS sum_alias)
+ // Filter(f2.k = d.k)
+ // Scan fact f2
+ //
+ // Output plan shape:
+ // Filter(f.v > d.tag, f.v * 2 > sum_over_window) ← mixed stays
ABOVE
+ // Window(sum(v) OVER (PARTITION BY d.k))
+ // Filter(f.k = d.k) ← join cond BELOW
+ // CrossJoin
+ // Scan fact f
+ // Scan dim d
+ createTable("CREATE TABLE fact_mixed (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_mixed (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_mixed add constraint uq_dim_mixed_k
unique (k)");
+
+ // Mixed predicate: f.v > d.tag references both shared (f.v) and
+ // outer-only (d.tag) columns. It must stay ABOVE the window.
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_mixed f, dim_mixed d "
+ + "WHERE f.k = d.k "
+ + " AND f.v > d.tag"
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_mixed f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule should match and produce a window
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule should produce a window for this query");
+
+ // Collect shared table (fact_mixed) ExprIds.
+ // A mixed predicate like f.v > d.tag references BOTH shared and
+ // outer-only sets.
+ List<CatalogRelation> rels =
plan.collectToList(CatalogRelation.class::isInstance);
+ Set<ExprId> sharedExprIds = rels.stream()
+ .filter(r -> r.getTable().getName().equals("fact_mixed"))
+ .flatMap(r -> r.getOutputExprIdSet().stream())
+ .collect(Collectors.toSet());
+
+ // Verify: the mixed predicate f.v > d.tag must NOT be below the
window.
+ // A mixed predicate references slots from BOTH shared and outer-only
+ // tables. We detect it by checking that at least one ExprId is in the
+ // shared set AND at least one is outside the shared set.
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ LogicalWindow<Plan> window = windows.get(0);
+ Plan belowWindow = window.child(0);
+ List<LogicalFilter<Plan>> belowFilters = belowWindow
+ .collectToList(LogicalFilter.class::isInstance);
+ for (LogicalFilter<Plan> f : belowFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (conjExprIds.isEmpty()) {
+ continue;
+ }
+ boolean hasShared = false;
+ boolean hasNonShared = false;
+ for (ExprId id : conjExprIds) {
+ if (sharedExprIds.contains(id)) {
+ hasShared = true;
+ } else {
+ hasNonShared = true;
+ }
+ }
+ if (hasShared && hasNonShared) {
+ // Join conditions (f.k = d.k) are expected below the
+ // window — they are matched from the inner filter and
+ // needed for the join. Only flag non-equality mixed
+ // predicates like f.v > d.tag.
+ if (!(conj instanceof
org.apache.doris.nereids.trees.expressions.EqualPredicate)) {
+ Assertions.fail(
+ "Mixed shared+outer predicate should not be
below window: "
+ + conj.toSql());
+ }
+ }
+ if (!hasNonShared) {
+ // Shared-table-only predicate also belongs ABOVE.
+ Assertions.fail(
+ "Shared-table-only predicate should not be below
window: "
+ + conj.toSql());
+ }
+ }
+ }
+
+ // Verify the mixed predicate IS present in a filter ABOVE the window.
+ // Collect all filters above the window by excluding below-window
filters.
+ List<LogicalFilter<Plan>> allFilters = plan
+ .collectToList(LogicalFilter.class::isInstance);
+ List<LogicalFilter<Plan>> aboveFilters = allFilters.stream()
+ .filter(f -> !belowFilters.contains(f))
+ .collect(Collectors.toList());
+ boolean foundMixedAbove = false;
+ for (LogicalFilter<Plan> f : aboveFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ boolean hasShared = false;
+ boolean hasNonShared = false;
+ for (ExprId id : conjExprIds) {
+ if (sharedExprIds.contains(id)) {
+ hasShared = true;
+ } else {
+ hasNonShared = true;
+ }
+ }
+ if (hasShared && hasNonShared) {
+ foundMixedAbove = true;
+ }
+ }
+ }
+ Assertions.assertTrue(foundMixedAbove,
+ "Mixed predicate f.v > d.tag should be above the window");
+ }
+
+ @Test
+ public void testEnsureProjectOutputExpandsPrunedColumn() throws Exception {
+ // When a nested shared-table filter is extracted and hoisted above the
+ // window, any pruning project inside apply.left() that dropped a
column
+ // referenced by the extracted conjunct must be expanded by
+ // ensureProjectOutput(). Otherwise the reinserted predicate would
have
+ // a dangling slot reference.
+ //
+ // Plan shape:
+ // Filter(sf.k = d.k, sf.k * 2 > sum_alias) ← sf.k comparison
+ // Apply(correlation: d.k)
+ // CrossJoin
+ // SubQueryAlias sf
+ // Project(k) ← prunes v
+ // Filter(v > 6) ← nested shared
filter
+ // Scan fact(k, v)
+ // Scan dim_unique d
+ // Aggregate(SUM(f2.k) AS sum_alias)
+ // Filter(f2.k = d.k)
+ // Scan fact f2
+ //
+ // The subquery SELECTs only k; the outer query uses sf.k for both join
+ // and comparison. The column v appears ONLY in the nested filter v >
6.
+ // After extracting v > 6, ensureProjectOutput() must expand Project(k)
+ // to Project(k, v) so the hoisted predicate has access to v.
+ createTable("CREATE TABLE fact_ensure_proj (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_ensure_proj (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_ensure_proj add constraint uq_dim_ep_k
unique (k)");
+
+ // The subquery SELECTs only k, pruning v. v > 6 is a nested
shared-table
+ // filter that must be extracted. The rule must produce a window with
+ // v > 6 hoisted above it, and ensureProjectOutput must expand the
+ // pruning project to carry v through.
+ String sql = "SELECT sf.k, d.did "
+ + "FROM (SELECT k FROM fact_ensure_proj WHERE v > 6) sf, "
+ + " dim_ensure_proj d "
+ + "WHERE sf.k = d.k "
+ + " AND sf.k * 2 > ("
+ + " SELECT SUM(f2.k) "
+ + " FROM fact_ensure_proj f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule must match and produce a window.
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule must produce a window for ensureProjectOutput test");
+
+ // Walk every filter in the plan and verify that every slot referenced
+ // by each conjunct is produced by the filter's child. If
+ // ensureProjectOutput() did not expand the project, the reinserted
+ // predicate v > 6 would have a dangling reference to v.
+ List<LogicalFilter<Plan>> allFilters = plan
+ .collectToList(LogicalFilter.class::isInstance);
+ for (LogicalFilter<Plan> f : allFilters) {
+ Set<ExprId> childOutput = f.child().getOutputExprIdSet();
+ for (Expression conj : f.getConjuncts()) {
+ for (ExprId id : conj.getInputSlotExprIds()) {
+ Assertions.assertTrue(childOutput.contains(id),
+ "Filter conjunct slot " + id
+ + " must be produced by filter's child. "
+ + "ensureProjectOutput() may not have expanded "
+ + "a pruning project. Conjunct: " + conj.toSql());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testNotMatchWhenCorrelatedTableNotUnique() throws Exception {
+ createTable("CREATE TABLE tpch.fact_dup (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE tpch.dim_dup (\n"
+ + " did INT,\n"
+ + " k INT,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_dup f, dim_dup d "
+ + "WHERE f.k = d.k "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_dup f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // DUPLICATE KEY table does not guarantee uniqueness, rule should not
match
+ Assertions.assertFalse(plan.anyMatch(LogicalWindow.class::isInstance));
+ }
+
+ @Test
+ public void testCompositeUniqueConstraintTriggersRewrite() throws
Exception {
+ // A table with a composite UNIQUE(a, b) constraint guarantees that
+ // (a, b) pairs are unique and non-null. When both columns are used
+ // as correlated slots, the window-rewrite is safe: PARTITION BY a, b
+ // produces exactly one row per outer row, matching the scalar
+ // subquery semantics.
+ //
+ // This also validates findSlotsByColumn() in LogicalCatalogRelation:
+ // only declared constraints whose FULL column set is present in the
+ // scan output are propagated as uniqueness slots. A rollup that
+ // drops column b from UNIQUE(a, b) would NOT see {a} as unique.
+ createTable("CREATE TABLE fact_comp (\n"
+ + " id INT,\n"
+ + " a INT,\n"
+ + " b INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_comp (\n"
+ + " did INT,\n"
+ + " a INT NOT NULL,\n"
+ + " b INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_comp add constraint uq_dim_comp_ab
unique (a, b)");
+
+ // Correlate on both columns of the composite constraint.
+ String sql = "SELECT d.did, d.a, d.b, d.tag, f.id, f.v "
+ + "FROM fact_comp f, dim_comp d "
+ + "WHERE f.a = d.a AND f.b = d.b "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_comp f2 "
+ + " WHERE f2.a = d.a AND f2.b = d.b"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Composite UNIQUE(a,b) -- both columns correlated -- should match.
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Composite UNIQUE(a,b) with both columns correlated must
trigger the rewrite");
+ }
+
+ @Test
+ public void testNotMatchWhenCorrelatedKeyIsNullableUnique() throws
Exception {
+ // A nullable column with a UNIQUE constraint is still unsafe for
+ // the window rewrite when the correlation uses null-safe equality
+ // (<=>). PARTITION BY groups all NULL-key outer rows into one
+ // partition, so those rows can join the same NULL-key inner rows
+ // and multiply the window aggregate — while the original scalar
+ // subquery is evaluated per outer row independently.
+ // isUniqueAndNotNull() must reject this because the key is nullable.
+ createTable("CREATE TABLE fact_nullable_uk (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ // k is nullable but has a UNIQUE constraint — DataTrait sees
+ // uniqueness without non-null, so isUniqueAndNotNull() is false.
+ createTable("CREATE TABLE dim_nullable_uk (\n"
+ + " did INT,\n"
+ + " k INT,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_nullable_uk add constraint
uq_dim_nullable_uk_k unique (k)");
+
+ // Use <=> (null-safe equals) correlation so NULL keys can join.
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_nullable_uk f, dim_nullable_uk d "
+ + "WHERE f.k <=> d.k "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_nullable_uk f2 "
+ + " WHERE f2.k <=> d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Nullable unique key is unsafe — all null keys partition together.
+ Assertions.assertFalse(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Nullable unique key with <=> correlation must not be
rewritten");
+ }
+
+ @Test
+ public void testUniqueKeyModelTriggersRewrite() throws Exception {
+ // UNIQUE KEY model tables guarantee uniqueness + non-null on the key
+ // column. DataTrait recognizes this even without an explicit
+ // ADD CONSTRAINT, so the rule should fire.
+ createTable("CREATE TABLE tpch.fact_ukey (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ // dim_ukey has UNIQUE KEY(k) with k INT NOT NULL — this implies
+ // unique + non-null without needing an explicit constraint.
+ createTable("CREATE TABLE tpch.dim_ukey (\n"
+ + " k INT NOT NULL,\n"
+ + " did INT,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "UNIQUE KEY(k)\n"
+ + "DISTRIBUTED BY HASH(k) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_ukey f, dim_ukey d "
+ + "WHERE f.k = d.k "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_ukey f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // UNIQUE KEY model alone provides uniqueness + non-null.
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "UNIQUE KEY model should trigger the window rewrite");
+ }
+
+ @Test
+ public void testOuterOnlyRelationOutputPreserved() throws Exception {
+ // When the outer query references the dim table directly (no
+ // SubQueryAlias wrapping it), the Apply's correlation slot
+ // ExprId IS the scan's original. checkRelation() finds it
+ // in the outer-only table's output, so the rewrite can proceed
+ // when other guards (uniqueness, filters) pass.
+ createTable("CREATE TABLE fact_oor (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_oor (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_oor add constraint uq_dim_oor_k unique
(k)");
+
+ // Direct table reference — no SubQueryAlias wrapping dim_oor.
+ // The Apply's correlation slot ExprId matches the scan output.
+ String sql = "SELECT d.did, d.k, d.tag, f.id, f.v "
+ + "FROM fact_oor f, dim_oor d "
+ + "WHERE f.k = d.k "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_oor f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // With the scan's original ExprId preserved, the rewrite should fire.
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rewrite must succeed when the correlated slot ExprId "
+ + "is the scan's original (direct table reference)");
+ }
+
+ @Test
+ public void testNotMatchWhenOuterOnlyRelationOutputIsPruned() throws
Exception {
+ // An aliased projection (d.k AS new_k) creates a new ExprId in
+ // the outer scope that is NOT present in the outer-only table's
+ // scan output. checkRelation() compares each correlated slot's
+ // ExprId against the scan's output ExprIdSet — the aliased slot
+ // is genuinely missing, so the rewrite must be rejected for this
+ // reason alone (not because of a missing uniqueness constraint).
+ createTable("CREATE TABLE fact_oor2 (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_oor2 (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_oor2 add constraint uq_dim_oor2_k
unique (k)");
+
+ // d.k AS new_k creates an Alias with a new ExprId. The inner
+ // query references t.new_k, so the Apply's correlation slot is
+ // this new ExprId. The scan of dim_oor2 outputs d.k with its
+ // original ExprId — the check in checkRelation() fails.
+ String sql = "SELECT t.id, t.new_k, t.v "
+ + "FROM ("
+ + " SELECT f.id, d.k AS new_k, f.v "
+ + " FROM fact_oor2 f, dim_oor2 d "
+ + " WHERE f.k = d.k"
+ + ") t "
+ + "WHERE t.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_oor2 f2 "
+ + " WHERE f2.k = t.new_k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // The aliased ExprId is not in the scan's output — must reject.
+ Assertions.assertFalse(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rewrite must be rejected when the correlated slot ExprId "
+ + "is not in the outer-only relation's scan output "
+ + "(e.g. d.k AS new_k creates a new ExprId)");
+ }
+
+ @Test
+ public void testNestedOuterFilterHoistedAboveWindow() throws Exception {
+ // When the outer child of Apply contains a nested LogicalFilter
+ // (e.g. a filter pushed into the FROM subquery like
+ // FROM (SELECT * FROM fact WHERE v > 6) sf), the rule must extract
+ // the nested conjuncts, classify them, and hoist shared-table
+ // predicates ABOVE the window. Otherwise the window would aggregate
+ // over a filtered subset, while the original scalar subquery computes
+ // over ALL fact rows for the key.
+ createTable("CREATE TABLE fact_nested (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_nested (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_nested add constraint uq_dim_nested_k
unique (k)");
+
+ // The FROM-subquery with WHERE v > 6 produces a LogicalFilter(f.v > 6)
+ // nested inside apply.left() under the LogicalSubQueryAlias.
+ String sql = "SELECT d.did, sf.id, sf.k, sf.v "
+ + "FROM (SELECT id, k, v FROM fact_nested WHERE v > 6) sf,
dim_nested d "
+ + "WHERE sf.k = d.k "
+ + " AND sf.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_nested f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule should match and produce a window
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule should produce a window when nested outer filter is
present");
+
+ // The nested shared-table predicate (v > 6) must be hoisted ABOVE the
+ // window. Verify it is both absent below AND preserved above.
+ List<CatalogRelation> rels =
plan.collectToList(CatalogRelation.class::isInstance);
+ Set<ExprId> factExprIds = rels.stream()
+ .filter(r -> r.getTable().getName().equals("fact_nested"))
+ .flatMap(r -> r.getOutputExprIdSet().stream())
+ .collect(Collectors.toSet());
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ LogicalWindow<Plan> window = windows.get(0);
+ Plan belowWindow = window.child(0);
+ List<LogicalFilter<Plan>> belowFilters = belowWindow
+ .collectToList(LogicalFilter.class::isInstance);
+ for (LogicalFilter<Plan> f : belowFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (!conjExprIds.isEmpty() &&
factExprIds.containsAll(conjExprIds)) {
+ Assertions.fail(
+ "Nested shared-table predicate should be hoisted
above window: "
+ + conj.toSql());
+ }
+ }
+ }
+
+ // Preservation: the hoisted predicate must appear exactly once in
+ // a filter ABOVE the window (not just absent below).
+ List<LogicalFilter<Plan>> allFilters = plan
+ .collectToList(LogicalFilter.class::isInstance);
+ List<LogicalFilter<Plan>> aboveFilters = allFilters.stream()
+ .filter(f -> !belowFilters.contains(f))
+ .collect(Collectors.toList());
+ int sharedOnlyAboveCount = 0;
+ for (LogicalFilter<Plan> f : aboveFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (!conjExprIds.isEmpty() &&
factExprIds.containsAll(conjExprIds)) {
+ sharedOnlyAboveCount++;
+ }
+ }
+ }
+ Assertions.assertEquals(1, sharedOnlyAboveCount,
+ "Hoisted shared-table predicate (v > 6) must be preserved "
+ + "exactly once above the window, not silently dropped");
+ }
+
+ @Test
+ public void testVolatilePredicateStaysAboveWindow() throws Exception {
+ // Volatile predicates like random() > 0.5 have no table column
+ // references but are non-deterministic. They must stay ABOVE the
+ // window, otherwise the window function would aggregate over a
+ // different set of rows per partition than the original scalar
+ // subquery. This follows the same principle as
+ // PushDownFilterThroughWindow.canPushDown().
+ createTable("CREATE TABLE fact_volatile (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_volatile (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_volatile add constraint
uq_dim_volatile_k unique (k)");
+
+ // random() > 0.5 is a volatile predicate with no input slots.
+ // It must be kept ABOVE the window.
+ String sql = "SELECT d.did, f.id, f.k, f.v "
+ + "FROM fact_volatile f, dim_volatile d "
+ + "WHERE f.k = d.k "
+ + " AND random() > 0.5"
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_volatile f2 "
+ + " WHERE f2.k = d.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule should match and produce a window
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule should produce a window for volatile predicate query");
+
+ // Verify the volatile predicate is NOT below the window.
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ LogicalWindow<Plan> window = windows.get(0);
+ Plan belowWindow = window.child(0);
+ List<LogicalFilter<Plan>> belowFilters = belowWindow
+ .collectToList(LogicalFilter.class::isInstance);
+ for (LogicalFilter<Plan> f : belowFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Assertions.assertFalse(conj.containsVolatileExpression(),
+ "Volatile predicate should stay above window: " +
conj.toSql());
+ }
+ }
+
+ // Preservation: the volatile predicate must appear exactly once in
+ // a filter ABOVE the window (not just absent below).
+ List<LogicalFilter<Plan>> allFilters = plan
+ .collectToList(LogicalFilter.class::isInstance);
+ List<LogicalFilter<Plan>> aboveFilters = allFilters.stream()
+ .filter(f -> !belowFilters.contains(f))
+ .collect(Collectors.toList());
+ int volatileAboveCount = 0;
+ for (LogicalFilter<Plan> f : aboveFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ if (conj.containsVolatileExpression()) {
+ volatileAboveCount++;
+ }
+ }
+ }
+ Assertions.assertEquals(1, volatileAboveCount,
+ "Volatile predicate (random() > 0.5) must be preserved "
+ + "exactly once above the window, not silently dropped");
+ }
+
+ @Test
+ public void testInnerFilterConjunctsStayBelowWindow() throws Exception {
+ // Regression test: shared-table predicates that were matched against
+ // inner subquery filter conjuncts must stay BELOW the window.
+ //
+ // Without the matchedInnerFilterConjuncts tracking, f.v < 10 would be
+ // classified as shared-table-only and placed ABOVE the window, letting
+ // the window aggregate over rows the original scalar subquery
excluded.
+ createTable("CREATE TABLE fact_inner_filter (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_inner_filter (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ // UNIQUE constraint so the rule matches
+ addConstraint("alter table dim_inner_filter add constraint uq_dim_if_k
unique (k)");
+
+ // Inner subquery has f2.v < 10 as a filter. After checkFilter matches
+ // it against outer f.v < 10, the outer conjunct must go BELOW the
window
+ // because it is semantically part of the inner aggregate's filter.
+ String sql = "SELECT d.did, f.id, f.k, f.v "
+ + "FROM fact_inner_filter f, dim_inner_filter d "
+ + "WHERE f.k = d.k "
+ + " AND f.v < 10"
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_inner_filter f2 "
+ + " WHERE f2.k = d.k"
+ + " AND f2.v < 10"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ // Rule should match and produce a window
+ Assertions.assertTrue(plan.anyMatch(LogicalWindow.class::isInstance),
+ "Rule should produce a window for this query");
+
+ // Collect the ExprIds of the shared table (fact_inner_filter).
+ List<CatalogRelation> allRels =
plan.collectToList(CatalogRelation.class::isInstance);
+ Set<ExprId> sharedExprIds = allRels.stream()
+ .filter(r ->
r.getTable().getName().equals("fact_inner_filter"))
+ .flatMap(r -> r.getOutputExprIdSet().stream())
+ .collect(Collectors.toSet());
+
+ // The conjunct f.v < 10, which was matched from the inner filter, must
+ // appear in a filter BELOW the window (it is part of the aggregate
+ // computation). Verify it exists there and is NOT above.
+ List<LogicalWindow<Plan>> windows =
plan.collectToList(LogicalWindow.class::isInstance);
+ LogicalWindow<Plan> window = windows.get(0);
+
+ // Check below-window filters: there MUST be at least one
shared-table-only
+ // conjunct (f.v < 10) — this is the matched inner-filter predicate.
+ Plan belowWindow = window.child(0);
+ List<LogicalFilter<Plan>> belowFilters = belowWindow
+ .collectToList(LogicalFilter.class::isInstance);
+ boolean foundSharedOnlyBelow = false;
+ for (LogicalFilter<Plan> f : belowFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (!conjExprIds.isEmpty() &&
sharedExprIds.containsAll(conjExprIds)) {
+ foundSharedOnlyBelow = true;
+ }
+ }
+ }
+ Assertions.assertTrue(foundSharedOnlyBelow,
+ "Matched inner-filter conjunct f.v < 10 must be below the
window");
+
+ // Check above-window filters: there should NOT be a shared-table-only
+ // conjunct that is NOT the window comparison. f.v < 10 should not
leak
+ // above.
+ List<LogicalFilter<Plan>> allFilters = plan
+ .collectToList(LogicalFilter.class::isInstance);
+ List<LogicalFilter<Plan>> aboveFilters = allFilters.stream()
+ .filter(f -> !belowFilters.contains(f))
+ .collect(Collectors.toList());
+ for (LogicalFilter<Plan> f : aboveFilters) {
+ for (Expression conj : f.getConjuncts()) {
+ Set<ExprId> conjExprIds = conj.getInputSlotExprIds();
+ if (!conjExprIds.isEmpty() &&
sharedExprIds.containsAll(conjExprIds)) {
+ Assertions.fail(
+ "Unexpected shared-table-only predicate above
window: " + conj.toSql());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testSplitInnerFilterFromPushDown() throws Exception {
+ // Regression: PushDownFilterThroughProject splits the inner WHERE
clause
+ // into multiple LogicalFilter nodes when a correlated predicate cannot
+ // be pushed through a project (references a correlated slot not in the
+ // project output) but a non-correlated predicate can.
+ //
+ // Before the fix, checkFilter() required exactly one inner filter and
+ // would reject plans where the filter was split. Now it collects
+ // conjuncts from ALL inner filters.
+ createTable("CREATE TABLE fact_split2 (\n"
+ + " id INT,\n"
+ + " k INT,\n"
+ + " v INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(id)\n"
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ createTable("CREATE TABLE dim_split2 (\n"
+ + " did INT,\n"
+ + " k INT NOT NULL,\n"
+ + " tag INT\n"
+ + ") ENGINE=OLAP\n"
+ + "DUPLICATE KEY(did)\n"
+ + "DISTRIBUTED BY HASH(did) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1')");
+ addConstraint("alter table dim_split2 add constraint uq_dim_split2_k
unique (k)");
+
+ String sql = "SELECT d.did, f.id, f.k, f.v "
+ + "FROM fact_split2 f, dim_split2 d "
+ + "WHERE f.k = d.k "
+ + " AND f.v < 10 "
+ + " AND f.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_split2 f2 "
+ + " WHERE f2.k = d.k"
+ + " AND f2.v < 10"
+ + " )";
+
+ // Full regression-style pipeline: PushDownFilterThroughProject splits
+ // the inner filter, MergeFilters merges adjacent filters, then the
+ // custom rule rewrites.
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyTopDown(new PushDownFilterThroughProject())
Review Comment:
[P2] Make the multi-filter test create the claimed split
`PushDownFilterThroughProject` only splits an exact `Filter -> Project`, but
this test runs it before `PullUpProjectUnderApply`; the scalar aggregate
predicates remain in the ordinary `Aggregate -> Filter -> Scan` path.
`MergeFilters` only combines adjacent filters. The test then asserts only that
a window exists and that some `v < 10` predicate is below it, both of which the
previous one-filter implementation already satisfies.
The production rewrite schedule uses `PullUpProjectUnderApply` first and
then `PushDownFilterThroughProject`/`MergeFilters`. Please capture the plan
immediately before `AggScalarSubQueryToWindowFunction`, assert that the target
Apply's right subtree actually contains the intended two filter nodes with
separated predicates, and only then assert the rewrite result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
