github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4371693356
##########
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:
[P1] Preserve relation identity in the aggregate grouping proof
This merges different aliases into the same equality vertex. Reuse the
`LIST(d,k)` kept/bridge/old fixture: kept={(2020-01-01,1),(2038-01-01,1)},
bridge={(2038-01-01,2)}, old={(2020-01-01,2)}, with a 2 YEAR sync window. Give
A an extra z=0 column, v=2 at the kept 2020 key and v=1 at the old key. Let
unpartitioned DUP table B(x,z) contain (1,0),(2,0). Build the d-partitioned MV
over `A a JOIN B b1 ON a.k=b1.x JOIN B b2 ON a.z=b2.z`, grouped by a.d,a.k,b2.x
with SUM(a.v). Query the same joins grouped only by a.d,b2.x, filtering
a.d='2020-01-01':
```text
Aggregate[a.d,b2.x; SUM(a.v)]
Filter[a.d=2020-01-01]
InnerJoin[a.z=b2.z]
InnerJoin[a.k=b1.x](Scan A a[kept,old], Scan B b1)
Scan B b2
```
`groupingCoversColumns` converts grouped b2.x to B.x, while
`columnsTheQueryHoldsEqual` converts a.k=b1.x to A.k=B.x. Closure therefore
falsely proves that the grouping contains A.k. Only A is PCT, so the
multiple-PCT-input safeguard does not reject this. The admitted reduced plan is
`UnionAll(Aggregate[d,bx,SUM(s)] -> MV[kept component], Aggregate[d,bx,SUM(v)]
-> Join(A[old],B b1,B b2))`: each bx gets separate sums 2 and 1 instead of one
3. Keep relation/ExprId identity in this proof and add this non-PCT-alias
compensation regression. This is distinct from the existing plain LIST
group-split and safe full-key findings: the new full-key exception is accepted
through an equality belonging to another alias.
##########
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:
[P2] Resolve current view inputs before pinning snapshots
This resolves the relation saved by the last refresh, while execution
regenerates it from the current query first (`MTMVTask.run`). For a SELF_MANAGE
MV over a view selecting k,v from table a, refresh once, ALTER that view to
select the same columns/types from b, then DROP a. `EXPLAIN REFRESH ...
COMPLETE` now throws resolving obsolete a here, although the next COMPLETE
refresh analyzes the current view over b and succeeds. ALTER VIEW invalidates
the MV but does not replace its stored relation until a refresh succeeds.
Derive the current query's input closure before pinning, so newly referenced
external inputs also receive pins. This differs from the existing external-pin
consistency thread: no concurrent metadata change or connector is needed.
##########
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:
[P2] Include replacement partitions in the COMPLETE preview
A pending LIST widening drops a live target from this preview without
representing its replacement. Start with populated LIST(d,k) partitions
p1={(1,1)}, p3={(3,1)} and a manual `PARTITION BY(d) AS SELECT d,k,SUM(v) ...
GROUP BY d,k` MV. After refresh, ADD p12={(1,2),(2,2)}. Alignment drops MV p_1
and creates p_1_2, retaining p_3. This condition refuses p_1_2 because it is
not an existing name, so the preview returns only p_3:
```text
EXPLAIN: Aggregate -> Filter(d=3 AND k=1) -> Scan base[p3]
COMPLETE: rebuild p_1_2 from base[p1,p12], and p_3 from base[p3]
```
`MTMVTask.syncPartitionsIfNeeded` creates the replacement before
`executeCompleteAttempt` enumerates targets, so real COMPLETE reads the
committed d=1/2 rows that EXPLAIN now hides. Preview the desired
descriptors/read scopes without requiring future sink names to exist. Unlike
the existing obsolete-expired-target thread, this alignment adds a new target
with live data as well as dropping the old one; add an ADD->EXPLAIN->COMPLETE
regression with another unchanged target.
--
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]
