yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4371842841
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java:
##########
@@ -505,45 +521,273 @@ private Expression rewriteShuttledExpression(StructInfo
queryStructInfo, Express
* select b, count(*) from t1 group by b, after rewritten by materialized
view successfully
* If mv part partition is invalid, can not compensate union all, because
result is wrong after
* compensate union all.
+ *
+ * <p>The compensation unions two finished aggregates: what the
materialized view answers for the partitions
+ * it keeps, and what the base table answers for the partitions it does
not. It is right while no group is
+ * answerable by both, and only a grouping that reaches the column the
view is partitioned on keeps the two
+ * apart -- the view's partitions are disjoint in that column, by value
for a list partitioned view and by
+ * range for a range partitioned one. A view partitioned by an expression
over the column is disjoint in
+ * that expression's value instead, so grouping by the same expression
splits the branches the same way:
+ * {@code date_trunc(d, 'month')} over a view partitioned by it answers
one month per group, all of it in
+ * one branch. Grouping by another expression over the column does not:
{@code date_trunc(d, 'year')} over
+ * that view answers one year out of both branches, the view's months of
it and the base table's, and the
+ * two rows come out beside each other. Nor does a grouping set that
leaves it out -- the {@code ()} total
+ * of {@code ROLLUP}/{@code CUBE}/{@code GROUPING SETS} puts every
partition value in one group, which both
+ * branches answer. And the view's own transform keeps them apart only
while each partition it keeps holds
+ * the whole transform bucket: a window cut or a partition boundary a
fraction of a second past the bucket's
+ * end leaves part of a bucket in the base branch, which
+ * {@code columnsSharingRetainedBucket} reports.
*/
@Override
protected boolean canUnionRewrite(Plan queryPlan,
AsyncMaterializationContext context,
- CascadesContext cascadesContext) {
- // Check query plan is contain the partition column
- // Query plan in the current rule must contain aggregate node, because
the rule pattern is
+ CascadesContext cascadesContext, Set<BaseColInfo>
columnsSharingRetainedBucket) {
+ // The union is built over the whole query plan, so the outermost
aggregate is the one whose groups
+ // the two branches answer. The plan holds one, since the rule's
pattern requires it.
Optional<LogicalAggregate<Plan>> logicalAggregateOptional =
queryPlan.collectFirst(planTreeNode -> planTreeNode instanceof
LogicalAggregate);
if (!logicalAggregateOptional.isPresent()) {
return true;
}
- List<Expression> groupByExpressions =
logicalAggregateOptional.get().getGroupByExpressions();
+ LogicalAggregate<Plan> queryAggregate = logicalAggregateOptional.get();
+ List<Expression> groupByExpressions =
queryAggregate.getGroupByExpressions();
if (groupByExpressions.isEmpty()) {
// Scalar aggregate can not compensate union all
return false;
}
- MTMVPartitionInfo mvPartitionInfo =
context.getMtmv().getMvPartitionInfo();
- List<BaseColInfo> pctInfos = mvPartitionInfo.getPctInfos();
- boolean canUnionRewrite = false;
- // Check the query plan group by expression contains partition col or
not
- List<? extends Expression> groupByShuttledExpressions =
-
ExpressionUtils.shuttleExpressionWithLineage(groupByExpressions, queryPlan);
- for (Expression expression : groupByShuttledExpressions) {
- canUnionRewrite = !expression.collectToSet(expr -> {
- if (!(expr instanceof SlotReference) || !((SlotReference)
expr).isColumnFromTable()) {
- return false;
+ MTMVPartitionInfo partitionInfo =
context.getMtmv().getMvPartitionInfo();
+ List<BaseColInfo> pctInfos = partitionInfo.getPctInfos();
+ // The view has one partition column but it can be tracked by several
base columns at once: an inner
+ // join on A.d = B.d records both, and they are one dimension as long
as the query holds them equal.
+ Map<BaseColInfo, Set<BaseColInfo>> equalColumns =
columnsTheQueryHoldsEqual(queryPlan);
+ Optional<String> partitionTimeUnit =
viewPartitionTimeUnit(partitionInfo);
+ Optional<LogicalRepeat<?>> logicalRepeatOptional =
queryAggregate.getSourceRepeat();
+ if (!logicalRepeatOptional.isPresent()) {
+ return groupingCoversPartitionColumns(groupByExpressions,
queryPlan, pctInfos, equalColumns,
+ partitionTimeUnit, columnsSharingRetainedBucket);
+ }
+ // The group by list holds every grouping set's expressions beside the
grouping id, so it can not tell
+ // which set a partition column was read from; each set is asked on
its own instead.
+ for (List<Expression> groupingSet :
logicalRepeatOptional.get().getGroupingSets()) {
+ if (!groupingCoversPartitionColumns(groupingSet, queryPlan,
pctInfos, equalColumns,
+ partitionTimeUnit, columnsSharingRetainedBucket)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * A group whose key names the whole partition key of the table by itself
belongs to one of its base
+ * partitions, and the two branches read disjoint base partitions: the MV
partitions that stay read the
+ * ones their snapshots record, the compensation the uncovered ones.
Grouping by the whole key therefore
+ * keeps the branches' groups apart even where the keys meet in the MV's
partition columns.
+ */
+ @Override
+ protected boolean groupingSeparatesCompensatedBranches(Plan queryPlan,
+ Map<MTMVRelatedTableIf, Set<String>> uncoveredNarrowedByTable) {
+ Optional<LogicalAggregate<Plan>> logicalAggregateOptional =
+ queryPlan.collectFirst(planTreeNode -> planTreeNode instanceof
LogicalAggregate);
+ if (!logicalAggregateOptional.isPresent()) {
+ return false;
+ }
+ LogicalAggregate<Plan> queryAggregate = logicalAggregateOptional.get();
+ Map<BaseColInfo, Set<BaseColInfo>> equalColumns =
columnsTheQueryHoldsEqual(queryPlan);
+ Optional<LogicalRepeat<?>> logicalRepeatOptional =
queryAggregate.getSourceRepeat();
+ for (MTMVRelatedTableIf table : uncoveredNarrowedByTable.keySet()) {
+ BaseTableInfo tableInfo = new BaseTableInfo(table);
+ Set<BaseColInfo> partitionColumns = ((OlapTable)
table).getPartitionColumns().stream()
+ .map(column -> new BaseColInfo(column.getName(),
tableInfo))
+ .collect(Collectors.toSet());
+ if (logicalRepeatOptional.isPresent()) {
+ for (List<Expression> groupingSet :
logicalRepeatOptional.get().getGroupingSets()) {
+ if (!groupingCoversColumns(groupingSet, queryPlan,
partitionColumns, equalColumns)) {
+ return false;
+ }
}
- String columnName = ((SlotReference)
expr).getOriginalColumn().map(Column::getName).orElse(null);
- BaseTableInfo baseTableInfo = ((SlotReference)
expr).getOriginalTable().map(BaseTableInfo::new)
- .orElse(null);
- return pctInfos.stream().anyMatch(colInfo ->
colInfo.getTableInfo().equals(baseTableInfo)
- && colInfo.getColName().equals(columnName));
- }
- ).isEmpty();
- if (canUnionRewrite) {
- break;
+ } else if
(!groupingCoversColumns(queryAggregate.getGroupByExpressions(), queryPlan,
+ partitionColumns, equalColumns)) {
+ return false;
}
}
- return canUnionRewrite;
+ return true;
+ }
+
+ /** Whether these grouping expressions name every one of these columns by
itself, or through an equality. */
+ private static boolean groupingCoversColumns(List<Expression>
groupingExpressions, Plan queryPlan,
+ Set<BaseColInfo> columns, Map<BaseColInfo, Set<BaseColInfo>>
equalColumns) {
+ if (groupingExpressions.isEmpty()) {
+ return false;
+ }
+ List<? extends Expression> groupingShuttledExpressions =
+
ExpressionUtils.shuttleExpressionWithLineage(groupingExpressions, queryPlan);
+ Set<BaseColInfo> groupedByItself = Sets.newHashSet();
+ for (Expression expression : groupingShuttledExpressions) {
+ columnGroupedByItself(expression).ifPresent(groupedByItself::add);
+ }
+ return equalColumnsClosure(groupedByItself,
equalColumns).containsAll(columns);
+ }
+
+ /**
+ * The time unit of the view's own partition transform, empty when it
partitions by the bare column. An
+ * expression partitioned view always is a date_trunc, the only function
its partition clause takes, and
+ * the second argument is the unit.
+ */
+ private static Optional<String> viewPartitionTimeUnit(MTMVPartitionInfo
partitionInfo) {
+ Expr partitionExpression = partitionInfo.getExpr();
+ if (partitionExpression == null) {
+ return Optional.empty();
+ }
+ List<Expr> arguments = ((FunctionCallExpr)
partitionExpression).getParams().exprs();
+ return Optional.of(((StringLiteral)
arguments.get(1)).getStringValue());
+ }
+
+ /**
+ * Whether these grouping expressions reach every column the view is
partitioned on, each of them by itself
+ * or equal to one that is -- through another column too, since an
equality chain is one equality -- or
+ * under the view's own partition transform. A tracked column that is
neither grouped nor tied to a grouped
+ * one by the query varies freely within a group.
+ *
+ * <p>The transform counts only while the view's partition holds the whole
transform bucket: over a column
+ * whose compensation shares a retained bucket it forms a group out of
both branches, so it covers nothing.
+ * The columns the query holds equal to such a column share the group's
value, so a transform over any of
+ * them is no better.
+ */
+ private static boolean groupingCoversPartitionColumns(List<Expression>
groupingExpressions, Plan queryPlan,
+ List<BaseColInfo> pctInfos, Map<BaseColInfo, Set<BaseColInfo>>
equalColumns,
+ Optional<String> partitionTimeUnit, Set<BaseColInfo>
columnsSharingRetainedBucket) {
+ List<? extends Expression> groupingShuttledExpressions =
+
ExpressionUtils.shuttleExpressionWithLineage(groupingExpressions, queryPlan);
+ Set<BaseColInfo> partitionColumns = Sets.newHashSet(pctInfos);
+ Set<BaseColInfo> groupedByItself = Sets.newHashSet();
+ Set<BaseColInfo> groupedByTransform = Sets.newHashSet();
+ for (Expression expression : groupingShuttledExpressions) {
+ Optional<BaseColInfo> grouped = columnGroupedByItself(expression);
+ if (grouped.isPresent()) {
+ if (partitionColumns.contains(grouped.get())) {
+ groupedByItself.add(grouped.get());
+ }
+ continue;
+ }
+ Optional<BaseColInfo> transformed =
columnUnderPartitionTransform(expression, partitionTimeUnit);
+ if (transformed.isPresent() &&
partitionColumns.contains(transformed.get())) {
+ groupedByTransform.add(transformed.get());
+ }
+ }
+ Set<BaseColInfo> groupedTogether =
equalColumnsClosure(groupedByItself, equalColumns);
+ for (BaseColInfo transformed : groupedByTransform) {
+ if (groupedTogether.contains(transformed)) {
+ continue;
+ }
+ Set<BaseColInfo> transformedClass =
equalColumnsClosure(Sets.newHashSet(transformed), equalColumns);
+ if (!Sets.intersection(transformedClass,
columnsSharingRetainedBucket).isEmpty()) {
+ continue;
+ }
+ groupedTogether.addAll(transformedClass);
+ }
+ return groupedTogether.containsAll(partitionColumns);
+ }
+
+ /** The columns reachable from these through the equalities the query
holds, the seeds included. */
+ private static Set<BaseColInfo> equalColumnsClosure(Set<BaseColInfo> seeds,
+ Map<BaseColInfo, Set<BaseColInfo>> equalColumns) {
+ Set<BaseColInfo> groupedTogether = Sets.newHashSet();
+ Deque<BaseColInfo> unvisited = new ArrayDeque<>(seeds);
+ while (!unvisited.isEmpty()) {
+ BaseColInfo column = unvisited.poll();
+ if (!groupedTogether.add(column)) {
+ continue;
+ }
+ for (BaseColInfo equal : equalColumns.getOrDefault(column,
Collections.emptySet())) {
+ if (!groupedTogether.contains(equal)) {
+ unvisited.add(equal);
+ }
+ }
+ }
+ return groupedTogether;
+ }
+
+ /**
+ * The base column this expression groups by under the view's own
partition transform, if it is that
+ * transform itself: the same date_trunc, in the same unit, over a column
of a table. A group it forms
+ * lies within one of the view's partitions, which are disjoint in the
transform's value, while a coarser
+ * truncation rolls several of them up into a group both branches answer.
+ */
+ private static Optional<BaseColInfo>
columnUnderPartitionTransform(Expression expression,
+ Optional<String> partitionTimeUnit) {
+ if (!partitionTimeUnit.isPresent() || !(expression instanceof
DateTrunc)) {
+ return Optional.empty();
+ }
+ DateTrunc dateTrunc = (DateTrunc) expression;
+ int dateArgumentIndex = dateTrunc.getMonotonicFunctionChildIndex();
+ // A timezone aware value is truncated in the session timezone, while
the view's partition boundaries
+ // are fixed in UTC, so the two can disagree on which of the view's
buckets a row's group falls in.
+ if (dateTrunc.getArgument(dateArgumentIndex).getDataType() instanceof
TimeStampTzType) {
+ return Optional.empty();
+ }
+ Expression timeUnit = dateTrunc.getArgument(1 - dateArgumentIndex);
+ if (!(timeUnit instanceof StringLikeLiteral)
+ || !((StringLikeLiteral) timeUnit).getStringValue()
+ .equalsIgnoreCase(partitionTimeUnit.get())) {
+ return Optional.empty();
+ }
+ return columnGroupedByItself(dateTrunc.getArgument(dateArgumentIndex));
+ }
+
+ /**
+ * The base column this expression is by itself: a slot of a column of a
table, with nothing over it.
+ */
+ private static Optional<BaseColInfo> columnGroupedByItself(Expression
expression) {
+ if (!(expression instanceof SlotReference) || !((SlotReference)
expression).isColumnFromTable()) {
+ return Optional.empty();
+ }
+ SlotReference slotReference = (SlotReference) expression;
+ String columnName =
slotReference.getOriginalColumn().map(Column::getName).orElse(null);
+ BaseTableInfo tableInfo =
slotReference.getOriginalTable().map(BaseTableInfo::new).orElse(null);
+ if (columnName == null || tableInfo == null) {
+ return Optional.empty();
+ }
+ return Optional.of(new BaseColInfo(columnName, tableInfo));
Review Comment:
Fixed in e9fbf0640bc. The proof now runs over column slots keyed by ExprId,
so an equality only reaches the very slots it names: `A.k = b1.x` no longer
proves that a grouping of `b2.x` covers `A.k`, and your fixture is answered
from the base table (`notContains "UNION"`, sums 3). Regression: the
`non_pct_alias` block in `mtmv_p0/test_mtmv_base_partition_read_scope`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java:
##########
@@ -270,13 +323,78 @@ private MTMV getMtmv() throws AnalysisException,
MetaNotFoundException {
TableType.MATERIALIZED_VIEW);
}
+ /**
+ * One engine snapshot per external input of this MV, also installed on
{@code statementContext} so
+ * that predicate construction and scan binding reuse the pins the mapping
is built from, as
+ * {@code MTMVTask#beforeMTMVRefresh} does for a refresh.
+ */
+ @VisibleForTesting
+ Map<MvccTableInfo, MvccSnapshot> pinBaseTableSnapshots(MTMV mtmv,
StatementContext statementContext)
+ throws AnalysisException {
+ Map<MvccTableInfo, MvccSnapshot> snapshots = Maps.newHashMap();
+ MTMVRelation relation = mtmv.getRelation();
+ if (relation == null) {
+ // A legacy MV has no relation until its first refresh regenerates
one.
+ return snapshots;
+ }
+ for (BaseTableInfo tableInfo :
relation.getBaseTablesOneLevelAndFromView()) {
+ TableIf table = MTMVUtil.getTable(tableInfo);
Review Comment:
Fixed in e9fbf0640bc. `pinBaseTableSnapshots` now resolves the inputs from
the current definition (`MTMVPlanUtil.getBaseTableFromQuery` +
`generateMTMVRelation`), as `MTMVTask.run` regenerates the relation, so a view
altered and a table dropped since the last refresh preview against the inputs
the next refresh reads, and those are the ones pinned. Regression: the
`view_inputs` block in `mtmv_p0/test_mtmv_base_partition_read_scope` — ALTER
VIEW, DROP the old table, EXPLAIN reads the new one, then COMPLETE rebuilds
from it.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java:
##########
@@ -270,13 +323,78 @@ private MTMV getMtmv() throws AnalysisException,
MetaNotFoundException {
TableType.MATERIALIZED_VIEW);
}
+ /**
+ * One engine snapshot per external input of this MV, also installed on
{@code statementContext} so
+ * that predicate construction and scan binding reuse the pins the mapping
is built from, as
+ * {@code MTMVTask#beforeMTMVRefresh} does for a refresh.
+ */
+ @VisibleForTesting
+ Map<MvccTableInfo, MvccSnapshot> pinBaseTableSnapshots(MTMV mtmv,
StatementContext statementContext)
+ throws AnalysisException {
+ Map<MvccTableInfo, MvccSnapshot> snapshots = Maps.newHashMap();
+ MTMVRelation relation = mtmv.getRelation();
+ if (relation == null) {
+ // A legacy MV has no relation until its first refresh regenerates
one.
+ return snapshots;
+ }
+ for (BaseTableInfo tableInfo :
relation.getBaseTablesOneLevelAndFromView()) {
+ TableIf table = MTMVUtil.getTable(tableInfo);
+ if (!(table instanceof MvccTable)) {
+ continue;
+ }
+ MvccTableInfo mvccTableInfo = new MvccTableInfo(table);
+ MvccSnapshot snapshot = ((MvccTable)
table).loadSnapshot(Optional.empty(), Optional.empty());
+ snapshots.put(mvccTableInfo, snapshot);
+ statementContext.setSnapshot(mvccTableInfo, snapshot);
+ }
+ return snapshots;
+ }
+
private Set<String> getCompleteRefreshPartitions(MTMV mtmv) {
if (mtmv.getMvPartitionInfo().getPartitionType() ==
MTMVPartitionType.SELF_MANAGE) {
return Sets.newHashSet();
}
return Sets.newHashSet(mtmv.getPartitionNames());
}
+ /**
+ * The partitions a COMPLETE refresh will rebuild, once the alignment it
runs before reading anything has
+ * dropped the partitions the base partitions no longer describe.
+ *
+ * <p>Execution applies that alignment and then takes the set it rebuilds
from the MV
+ * (MTMVTask#executeCompleteAttempt), so a preview that kept a dropped
partition would show a read the
+ * refresh does not make: that partition's own key range, and -- while the
partition is undescribed -- the
+ * whole batch unscoped, taking every other partition's scope with it (see
+ * MTMVPartitionUtil#mappedBasePartitionsByMvPartition). The alignment
computed here is the one the next
+ * refresh would run, against the snapshots the preview is pinned to, and
nothing is applied.
+ *
+ * <p>Only names that exist stay in the preview. A partition the alignment
has yet to create has no name
+ * the sink can target and no rows until the refresh creates it, so it
stays out; one the alignment drops
+ * and recreates under its own name stays in, because that name is what
the refresh rebuilds and leaving
+ * it out would hide a read the refresh makes.
+ */
+ private static Set<String> partitionsAfterPendingAlignment(MTMV mtmv,
+ Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots, Set<String>
partitionNames)
+ throws AnalysisException {
+ if (mtmv.getMvPartitionInfo().getPartitionType() ==
MTMVPartitionType.SELF_MANAGE) {
+ return partitionNames;
+ }
+ Pair<List<String>, List<PartitionKeyDesc>> alignment =
+ MTMVPartitionUtil.alignMvPartition(mtmv, pinnedSnapshots);
+ Set<String> retained = Sets.newHashSet(partitionNames);
+ retained.removeAll(alignment.first);
+ for (PartitionKeyDesc partitionKeyDesc : alignment.second) {
+ String recreatedName =
MTMVPartitionUtil.generatePartitionName(partitionKeyDesc);
+ if (partitionNames.contains(recreatedName)) {
Review Comment:
Fixed in e9fbf0640bc. The preview now takes the alignment
(`computePartitionAlignment` + `pendingPartitions`): the sink still targets
only names the MV holds (or recreates under the same name), while the
partitions the alignment creates are read through their items and read scopes.
Your fixture now explains `partitions=3/3 (p1,p3,p12)` — the replacement p1+p12
beside the retained p3 — instead of p3 alone, and the COMPLETE refresh
materializes the committed d=1/d=2 rows. Regression: the `replacement` block in
`mtmv_p0/test_mtmv_base_partition_read_scope`.
--
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]
