Skip to content

[fix](mtmv) Check agg/window partition keys only for tracked partition outputs #65393

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
foxtail463 wants to merge 2 commits into
base: master
Choose a base branch
from
+152 −64
Open

[fix](mtmv) Check agg/window partition keys only for tracked partition outputs #65393

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f478271
[fix](mtmv) Check agg/window partition keys only for tracked partitio…
Jul 8, 2026
3823fa6
Fix regression
Jul 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 51 additions & 13 deletions ...main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionIncrementMaintainer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@
import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import org.apache.doris.nereids.trees.plans.logical.LogicalLimit;
import org.apache.doris.nereids.trees.plans.logical.LogicalPartitionTopN;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
import org.apache.doris.nereids.trees.plans.logical.LogicalResultSink;
Expand Down Expand Up @@ -410,6 +411,9 @@ && new BaseTableInfo(((LogicalCatalogRelation) relation).getTable())
@Override
public Void visitLogicalAggregate(LogicalAggregate<? extends Plan> aggregate,
PartitionIncrementCheckContext context) {
if (!planOutputContainsPartitionColumnToCheck(aggregate, context)) {
return super.visitLogicalAggregate(aggregate, context);
}
Set<Expression> groupByExprSet = new HashSet<>(aggregate.getGroupByExpressions());
if (groupByExprSet.isEmpty()) {
context.addFailReason("group by sets is empty, doesn't contain the target partition");
Expand All @@ -426,6 +430,9 @@ public Void visitLogicalAggregate(LogicalAggregate<? extends Plan> aggregate,

@Override
public Void visitLogicalWindow(LogicalWindow<? extends Plan> window, PartitionIncrementCheckContext context) {
if (!planOutputContainsPartitionColumnToCheck(window, context)) {
return super.visitLogicalWindow(window, context);
}
List<NamedExpression> windowExpressions = window.getWindowExpressions();
if (windowExpressions.isEmpty()) {
context.addFailReason("window expression is empty, doesn't contain the target partition");
Expand All @@ -444,6 +451,20 @@ public Void visitLogicalWindow(LogicalWindow<? extends Plan> window, PartitionIn
return super.visitLogicalWindow(window, context);
}

@Override
public Void visitLogicalPartitionTopN(LogicalPartitionTopN<? extends Plan> partitionTopN,
PartitionIncrementCheckContext context) {
if (!planOutputContainsPartitionColumnToCheck(partitionTopN, context)) {
return super.visitLogicalPartitionTopN(partitionTopN, context);

@morrySnow morrySnow Jul 9, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The error message says "window partition sets" but this method handles LogicalPartitionTopN, not a window. Consider changing to "PartitionTopN partition keys does not contain the target partition" to avoid misleading debug output.

}
if (!checkPartitionKeysContainPartitionToCheck(partitionTopN.getPartitionKeys(), context)) {
context.addFailReason("window partition sets doesn't contain the target partition");
context.collectFailedTableSet(partitionTopN);
context.setFailFast(true);
}
return super.visitLogicalPartitionTopN(partitionTopN, context);
}

@Override
public Void visit(Plan plan, PartitionIncrementCheckContext context) {
if (plan instanceof LogicalProject
Expand All @@ -456,6 +477,7 @@ public Void visit(Plan plan, PartitionIncrementCheckContext context) {
|| plan instanceof LogicalWindow
|| (plan instanceof LogicalUnion
&& ((LogicalUnion) plan).getQualifier() == SetOperation.Qualifier.ALL)
|| plan instanceof LogicalPartitionTopN
|| plan instanceof LogicalCTEAnchor
|| plan instanceof LogicalCTEConsumer
|| plan instanceof LogicalCTEProducer
Expand All @@ -476,25 +498,41 @@ private boolean checkWindowPartition(Expression expression, PartitionIncrementCh
expression.collectToList(expressionTreeNode -> expressionTreeNode instanceof WindowExpression);
for (Object windowExpressionObj : windowExpressions) {
WindowExpression windowExpression = (WindowExpression) windowExpressionObj;
List<Expression> partitionKeys = windowExpression.getPartitionKeys();
Set<Column> originalPartitionbyExprSet = new HashSet<>();
partitionKeys.forEach(groupExpr -> {
if (groupExpr instanceof SlotReference && groupExpr.isColumnFromTable()) {
originalPartitionbyExprSet.add(((SlotReference) groupExpr).getOriginalColumn().get());
}
});
Set<SlotReference> contextPartitionColumnSet = getPartitionColumnsToCheck(context);
if (contextPartitionColumnSet.isEmpty()) {
return false;
}
if (contextPartitionColumnSet.stream().noneMatch(
partition -> originalPartitionbyExprSet.contains(partition.getOriginalColumn().get()))) {
if (!checkPartitionKeysContainPartitionToCheck(windowExpression.getPartitionKeys(), context)) {
return false;
}
}
return true;
}

private boolean checkPartitionKeysContainPartitionToCheck(List<Expression> partitionKeys,
PartitionIncrementCheckContext context) {
Set<Column> originalPartitionByExprSet = new HashSet<>();
partitionKeys.forEach(partitionKey -> {
if (partitionKey instanceof SlotReference && partitionKey.isColumnFromTable()) {
originalPartitionByExprSet.add(((SlotReference) partitionKey).getOriginalColumn().get());
}
});
Set<SlotReference> contextPartitionColumnSet = getPartitionColumnsToCheck(context);
if (contextPartitionColumnSet.isEmpty()) {
return false;
}
return contextPartitionColumnSet.stream().anyMatch(
partition -> originalPartitionByExprSet.contains(partition.getOriginalColumn().get()));
}

private boolean planOutputContainsPartitionColumnToCheck(Plan plan, PartitionIncrementCheckContext context) {
Set<Slot> outputSet = plan.getOutputSet();
for (NamedExpression namedExpression : context.getPartitionAndRefExpressionMap().keySet()) {
// Plan outputs are slots, so only tracked partition expressions already resolved
// to slots can match here.
if (namedExpression instanceof Slot && outputSet.contains(namedExpression)) {
return true;
}
}
return false;
}

private Set<SlotReference> getPartitionColumnsToCheck(PartitionIncrementCheckContext context) {
Set<NamedExpression> partitionExpressionSet = context.getPartitionAndRefExpressionMap().keySet();
Set<SlotReference> partitionSlotSet = new HashSet<>();
Expand Down
56 changes: 53 additions & 3 deletions ...rc/test/java/org/apache/doris/nereids/rules/exploration/mv/MaterializedViewUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -637,6 +637,54 @@ public void getRelatedTableInfoTestWithWindowButNotPartitionTest() {
});
}

@Test
public void getRelatedTableInfoTestWithUnrelatedRightWindowTest() {
PlanChecker.from(connectContext)
.checkExplain("SELECT l.L_SHIPDATE, l.L_ORDERKEY, o.O_ORDERDATE "
+ "FROM lineitem as l "
+ "LEFT JOIN ("
+ "SELECT O_ORDERKEY, O_ORDERDATE, "
+ "ROW_NUMBER() OVER (PARTITION BY O_ORDERKEY ORDER BY O_ORDERDATE DESC) AS rn "
+ "FROM orders"
+ ") as o "
+ "ON l.L_ORDERKEY = o.O_ORDERKEY AND o.rn = 1",
nereidsPlanner -> {
Plan rewrittenPlan = nereidsPlanner.getRewrittenPlan();
RelatedTableInfo relatedTableInfo =
MaterializedViewUtils.getRelatedTableInfo("L_SHIPDATE", null,
rewrittenPlan, nereidsPlanner.getCascadesContext());
Assertions.assertTrue(relatedTableInfo.isPctPossible(), relatedTableInfo.getFailReason());
checkRelatedTableInfo(relatedTableInfo,
"lineitem",
"L_SHIPDATE",
true);
});
}

@Test
public void getRelatedTableInfoTestWithUnrelatedRightAggregateTest() {
PlanChecker.from(connectContext)
.checkExplain("SELECT l.L_SHIPDATE, l.L_ORDERKEY, o.max_orderdate "
+ "FROM lineitem as l "
+ "LEFT JOIN ("
+ "SELECT O_ORDERKEY, max(O_ORDERDATE) AS max_orderdate "
+ "FROM orders "
+ "GROUP BY O_ORDERKEY"
+ ") as o "
+ "ON l.L_ORDERKEY = o.O_ORDERKEY",
nereidsPlanner -> {
Plan rewrittenPlan = nereidsPlanner.getRewrittenPlan();
RelatedTableInfo relatedTableInfo =
MaterializedViewUtils.getRelatedTableInfo("L_SHIPDATE", null,
rewrittenPlan, nereidsPlanner.getCascadesContext());
Assertions.assertTrue(relatedTableInfo.isPctPossible(), relatedTableInfo.getFailReason());
checkRelatedTableInfo(relatedTableInfo,
"lineitem",
"L_SHIPDATE",
true);
});
}

@Test
public void getRelatedTableInfoTestWithLimitTest() {
PlanChecker.from(connectContext)
Expand Down Expand Up @@ -875,9 +923,11 @@ public void getRelatedTableInfoWhenMultiBaseTablePartition() {
RelatedTableInfo relatedTableInfo =
MaterializedViewUtils.getRelatedTableInfo("upgrade_day", null,
rewrittenPlan, nereidsPlanner.getCascadesContext());
Assertions.assertTrue(relatedTableInfo.getFailReason().contains(
"partition column is not in group by or window partition by"));
Assertions.assertFalse(relatedTableInfo.isPctPossible());
Assertions.assertTrue(relatedTableInfo.isPctPossible(), relatedTableInfo.getFailReason());
checkRelatedTableInfo(relatedTableInfo,
"test1",
"upgrade_day",
true);
});
}

Expand Down
8 changes: 4 additions & 4 deletions ...t/suites/nereids_rules_p0/mv/increment_create/cross_join_list_str_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -345,12 +345,12 @@ suite("cross_join_list_str_increment_create", "increment_create") {

def sql_all_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12,
mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8, mv_sql_14]
def sql_complete_list = []

// change left table data
// create mv base on left table with partition col
def sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
def sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -375,9 +375,9 @@ suite("cross_join_list_str_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_increment_list = []
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8]
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8, mv_sql_14]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list, partition_by_part_col, slave_tb_change, is_complete_change)

// create mv base on left table with no partition col
Expand Down
8 changes: 4 additions & 4 deletions ...suites/nereids_rules_p0/mv/increment_create/cross_join_range_date_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,12 +336,12 @@ suite("cross_join_range_date_increment_create", "increment_create") {

def sql_all_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12,
mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_13]
def sql_complete_list = []

// change left table data
// create mv base on left table with partition col
def sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
def sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -366,9 +366,9 @@ suite("cross_join_range_date_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_increment_list = []
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7]
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_13]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, slave_tb_change, is_complete_change)

Expand Down
6 changes: 3 additions & 3 deletions ...ites/nereids_rules_p0/mv/increment_create/cross_join_range_number_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -351,7 +351,7 @@ suite("cross_join_range_number_increment_create", "increment_create") {

// change left table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -376,9 +376,9 @@ suite("cross_join_range_number_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_increment_list = []
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8]
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8, mv_sql_14]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, slave_tb_change, is_complete_change)

Expand Down
6 changes: 3 additions & 3 deletions ...suites/nereids_rules_p0/mv/increment_create/inner_join_range_date_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,7 +361,7 @@ suite("inner_join_range_date_increment_create", "increment_create") {

// change left table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -386,9 +386,9 @@ suite("inner_join_range_date_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_13, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12, mv_sql_14, mv_sql_15, mv_sql_16, mv_sql_17, mv_sql_18]
sql_increment_list = [mv_sql_1, mv_sql_3]
sql_complete_list = [mv_sql_4, mv_sql_6, mv_sql_7]
sql_complete_list = [mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_13]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, slave_tb_change, is_complete_change)

Expand Down
8 changes: 4 additions & 4 deletions ...ites/nereids_rules_p0/mv/increment_create/left_anti_join_list_str_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -302,12 +302,12 @@ suite("left_anti_join_list_str_increment_create", "increment_create") {
}

def sql_all_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8]
def sql_increment_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8, mv_sql_11]
def sql_complete_list = []

// change left table data
// create mv base on left table with partition col
def sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12]
def sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_12]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -328,9 +328,9 @@ suite("left_anti_join_list_str_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12]
sql_error_list = [mv_sql_7, mv_sql_9, mv_sql_10, mv_sql_12]
sql_increment_list = []
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8]
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_8, mv_sql_11]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, slave_tb_change, is_complete_change)

Expand Down
6 changes: 3 additions & 3 deletions ...es/nereids_rules_p0/mv/increment_create/left_anti_join_range_date_increment_create.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -298,7 +298,7 @@ suite("left_anti_join_range_date_increment_create", "increment_create") {

// change left table data
// create mv base on left table with partition col
def sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12]
def sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_11, mv_sql_12]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, primary_tb_change, is_complete_change)

Expand All @@ -319,9 +319,9 @@ suite("left_anti_join_range_date_increment_create", "increment_create") {

// change right table data
// create mv base on left table with partition col
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_10, mv_sql_11, mv_sql_12]
sql_error_list = [mv_sql_8, mv_sql_9, mv_sql_11, mv_sql_12]
sql_increment_list = []
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7]
sql_complete_list = [mv_sql_1, mv_sql_3, mv_sql_4, mv_sql_6, mv_sql_7, mv_sql_10]
list_judgement(sql_all_list, sql_increment_list, sql_complete_list, sql_error_list,
partition_by_part_col, slave_tb_change, is_complete_change)

Expand Down
Loading
Loading