github-actions[bot] commented on code in PR #63763:
URL: https://github.com/apache/doris/pull/63763#discussion_r3627538910
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunctionTest.java:
##########
@@ -338,6 +368,2672 @@ 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 testNotMatchWhenOuterOnlyRelationOutputIsPruned() throws
Exception {
+ createTable("CREATE TABLE tpch.fact_window_pruned (\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_window_pruned (\n"
Review Comment:
[P2] Make this negative test reach the intended guard
`dim_window_pruned.k` is nullable on a `DUPLICATE KEY(did)` table and has no
PRIMARY/UNIQUE constraint, so `checkUniqueCorrelatedTable()` rejects this plan
regardless of the projection/ExprId condition this test is meant to cover. The
SQL also still projects `d.k` as `t.k`. This assertion therefore stays green if
the intended guard is removed. Please use a non-null unique correlated key,
prove the corresponding control shape rewrites, and then construct the
genuinely pruned/mismatched shape so this test fails for only the intended
reason.
##########
fe/fe-core/src/test/java/org/apache/doris/nereids/rules/rewrite/AggScalarSubQueryToWindowFunctionTest.java:
##########
@@ -338,6 +368,2672 @@ 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 testNotMatchWhenOuterOnlyRelationOutputIsPruned() throws
Exception {
+ createTable("CREATE TABLE tpch.fact_window_pruned (\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_window_pruned (\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 t.id, t.k, t.v "
+ + "FROM ("
+ + " SELECT f.id, d.k, f.v "
+ + " FROM fact_window_pruned f, dim_window_pruned d "
+ + " WHERE f.k = d.k"
+ + ") t "
+ + "WHERE t.v * 2 > ("
+ + " SELECT SUM(f2.v) "
+ + " FROM fact_window_pruned f2 "
+ + " WHERE f2.k = t.k"
+ + " )";
+
+ Plan plan = PlanChecker.from(createCascadesContext(sql))
+ .analyze(sql)
+ .applyBottomUp(new PullUpProjectUnderApply())
+ .applyTopDown(new PushDownFilterThroughProject())
+ .customRewrite(new EliminateUnnecessaryProject())
+ .customRewrite(new AggScalarSubQueryToWindowFunction())
+ .getPlan();
+
+ Assertions.assertFalse(plan.anyMatch(LogicalWindow.class::isInstance));
+ }
+
+ @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
Review Comment:
[P2] Assert that hoisted predicates are preserved
This only proves that a shared-only conjunct is absent below the window; it
still passes if the rewrite drops `v > 6` entirely. The adjacent volatile test
has the same blind spot for `random() > 0.5`. Please locate each exact target
conjunct outside the window subtree (ideally exactly once) in addition to
asserting its absence below, so these tests cover preservation as well as
placement.
--
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]
