github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4367115732
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/AbstractMaterializedViewAggregateRule.java:
##########
@@ -505,45 +519,191 @@ 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.
*/
@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
+ // 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);
+ }
+ // 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)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * 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.
+ */
+ private static boolean groupingCoversPartitionColumns(List<Expression>
groupingExpressions, Plan queryPlan,
+ List<BaseColInfo> pctInfos, Map<BaseColInfo, Set<BaseColInfo>>
equalColumns,
+ Optional<String> partitionTimeUnit) {
+ List<? extends Expression> groupingShuttledExpressions =
+
ExpressionUtils.shuttleExpressionWithLineage(groupingExpressions, queryPlan);
+ Set<BaseColInfo> partitionColumns = Sets.newHashSet(pctInfos);
+ Set<BaseColInfo> groupedTogether = Sets.newHashSet();
+ Deque<BaseColInfo> unvisited = new ArrayDeque<>();
+ for (Expression expression : groupingShuttledExpressions) {
+ Optional<BaseColInfo> grouped = columnGroupedByItself(expression);
+ if (!grouped.isPresent()) {
+ grouped = columnUnderPartitionTransform(expression,
partitionTimeUnit);
Review Comment:
[P1] Check window cuts before accepting the partition transform
Matching the MV transform does not prove disjoint groups:
`partition_sync_limit` removes base RANGE partitions before roll-up, and the
scoped refresh can retain only part of one year. Reuse the first
`base_read_scope` fixture on a same-year day and query `k1 IN (expiredDay,
keptDay)` grouped only by `date_trunc(k1,'year')`. The fresh year MV records
`p_kept,p_today`; `p_expired` is unrecorded. `calcInvalidPartitions`
compensates it without removing the MV, the LIST-only overlap check returns
empty, and this branch accepts the year grouping:
```text
UnionAll
Aggregate(year(k1), SUM(total))
Filter(k1 IN(expiredDay,keptDay)) -> Scan MV[year] // (year,2)
Aggregate(year(k1), SUM(value))
Filter(query AND bounds(p_expired)) -> Scan base // (year,1)
```
No aggregate above this union combines the rows, so the result is two year
groups instead of `(year,3)`. This differs from the safe whole-bucket transform
case in #4366249927: the read window splits the transform bucket itself. Check
compensated ranges against retained groups in transform space, or merge the
aggregate states, before permitting this rewrite; add this window-cut case
alongside `pct_month`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +143,112 @@ private static List<String>
constructPartsForMv(Set<String> partitionNames) {
return Lists.newArrayList(partitionNames);
}
+ /**
+ * The predicate every base table of the MV definition is read through.
+ *
+ * <p>A table the caller scopes is read from exactly the base partitions
it named. Those are the ones
+ * the refresh is about to record as this MV partition's, and the read is
what has to match the record:
+ * reading the MV partition's own key range instead also reads base
partitions no snapshot describes,
+ * and a later silent change to one of them -- dropped, with the base
partition set back to what it
+ * was -- leaves the rows it put in this MV partition behind while the
partition is still judged
+ * synchronized, so the transparent rewrite serves them and no refresh
plans it again.
+ *
+ * <p>Every other table keeps the MV partition's own key range, which is
what the tables the caller
+ * does not scope were always read through, and a scoped range table keeps
it as well: a partition of
+ * a base range table can hold a row past the MV partition it is rolled up
into, see below. Scoped
+ * tables are the MV's partition tables, olap or external, and their
partitions are resolved from the
+ * snapshot pinned in the statement context -- the same one the mapping
was built from, so the read is
+ * pinned to the partitions the batch records even when the connector
moves between the two.
+ */
private static Map<TableIf, Set<Expression>>
constructTableWithPredicates(MTMV mv,
- Set<String> partitionNames, Map<TableIf, String> tableWithPartKey)
throws AnalysisException {
- Set<PartitionItem> items = Sets.newHashSet();
+ Set<String> partitionNames, Map<TableIf, String> tableWithPartKey,
+ StatementContext statementContext,
+ Map<BaseTableInfo, Set<String>> readableBasePartitions) throws
AnalysisException {
+ Set<PartitionItem> mvItems = Sets.newHashSet();
for (String partitionName : partitionNames) {
- PartitionItem partitionItem =
mv.getPartitionItemOrAnalysisException(partitionName);
- items.add(partitionItem);
+ mvItems.add(mv.getPartitionItemOrAnalysisException(partitionName));
}
ImmutableMap.Builder<TableIf, Set<Expression>> builder = new
ImmutableMap.Builder<>();
- tableWithPartKey.forEach((table, colName) ->
- builder.put(table, constructPredicates(items, colName))
- );
+ for (Map.Entry<TableIf, String> entry : tableWithPartKey.entrySet()) {
+ TableIf table = entry.getKey();
+ String colName = entry.getValue();
+ Set<String> readable = readableBasePartitions == null ? null
+ : readableBasePartitions.get(new BaseTableInfo(table));
+ if (readable == null) {
+ builder.put(table, constructPredicates(mvItems, colName));
+ continue;
+ }
+ MTMVRelatedTableIf relatedTable = (MTMVRelatedTableIf) table;
+ Optional<MvccSnapshot> snapshot =
statementContext.getSnapshot(table);
+ List<Column> partitionColumns =
relatedTable.getPartitionColumns(snapshot);
+ Map<String, PartitionItem> itemsByName =
relatedTable.getAndCopyPartitionItems(snapshot);
Review Comment:
[P2] Avoid copying every base partition for each refresh batch
`OlapTable.getAndCopyPartitionItems` takes the table read lock, walks every
base partition and allocates a full map. This call runs for every command built
by `MTMVTask`'s batch loop, whose default `refresh_partition_num` is 1. A
COMPLETE refresh of a direct N-partition MV therefore adds N full N-entry
copies, even though each batch reads one partition: 10,000 partitions entail
roughly 100 million metadata entries visited/copied before predicate planning.
External tables copy their entire pinned map here too. Resolve the selected
OLAP names directly, or prepare and reuse a frozen partition-item map once per
refresh context while preserving the external snapshot. This is separate from
the earlier quadratic predicate-expansion reports; the repeated work is across
singleton batches and needs no default LIST partition.
--
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]
