yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4369535871
##########
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:
Fixed in 11140556bc0. PartitionCompensator now records, per base column,
whether a compensated partition rolls up into a retained MV partition's own
descriptor, or whether a retained MV partition records a base partition that
runs past its upper bound; AbstractMaterializedViewAggregateRule lets the
view's transform cover a column only when neither that column nor any column
the query holds equal to it is recorded that way, and otherwise reads the query
from the base table.
Regression: mtmv_p0/test_mtmv_base_partition_read_scope now keeps 1000 rows
in the kept day of a 1 DAY window (one pre-aggregated row, so CBO prefers the
view plus the expired day) and asserts the date_trunc(k1, 'year') grouping is
answered without UNION; the golden carries the single year total.
##########
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:
Fixed in 47caa34126c. The refresh path resolves partition names where they
live: PredicateAdder uses getPartitionItemOrAnalysisException for an olap table
and builds an external table's map once per relation, and
UpdateMvByPartitionCommand's scoped read does the same (olap by name, external
from the pinned snapshot's map). The whole-map copy per batch is gone;
UpdateMvByPartitionCommandTest passes (17 tests).
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -1608,9 +1608,11 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>>
calculatePartitionMappi
Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>>
pctPartitionDescs = MTMVPartitionUtil
.generateRelatedPartitionDescs(mvPartitionInfo, mvProperties,
getPartitionColumns(),
effectiveFilter, pinnedSnapshots);
+ Map<MTMVRelatedTableIf, String> defaultListPartitions =
defaultListPartitionsOf();
for (Entry<String, PartitionItem> entry : mvPartitionItems.entrySet())
{
- res.put(entry.getKey(),
-
pctPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(),
Maps.newHashMap()));
+ res.put(entry.getKey(), withDefaultListPartitions(
+
pctPartitionDescs.getOrDefault(entry.getValue().toPartitionKeyDesc(),
Maps.newHashMap()),
Review Comment:
Fixed in a637adbded5. MTMVTask now calls requireDescribedPartitions whenever
the view has an olap PCT table, IVM included, so an IVM COMPLETE rebuild
refuses a partition its mapping describes with no base partition instead of
overwriting it with no rows, and the next attempt realigns.
Regression: mtmv_p0/ivm/test_ivm_undescribed_partition_refresh skips the
alignment with a debug point after a widening LIST ADD; the rebuild fails with
"is described by no base partition" and keeps the partition's rows, and the
following refresh realigns and reads the widened component.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +139,66 @@ 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. Scoped tables are olap ones;
the partition names are
+ * looked up on one, see the caller.
+ */
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,
+ 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;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ Set<PartitionItem> items = Sets.newHashSet();
+ for (String partitionName : readable) {
+
items.add(olapTable.getPartitionItemOrAnalysisException(partitionName));
+ }
+ if (items.stream().anyMatch(PartitionItem::isDefaultPartition)) {
+ // One of the partitions this MV partition is recorded with is
a list partitioned table's
+ // default partition, which takes the rows no other partition
of it claims. Those rows are
+ // the ones the MV partition's own key range names, wherever
the base table put them, and a
+ // partition of the MV takes them by that key rather than by
the partition they were placed
+ // in. So a table whose mapped partitions include one is read
the way an unscoped one is:
+ // the MV partition's key range, at the partition column's own
type. That read can be seen to
+ // be too wide -- it is the one this scope exists to narrow --
rather than one that drops
+ // rows belonging to the MV partition being refreshed. The
mapping names the default
+ // partition in every MV partition that reads the table, so
this is reached for each of them
+ // and not only for the one the sentinel key maps to.
+ builder.put(table, constructPredicates(mvItems, colName,
+ Optional.of(partitionColumnType(olapTable, colName))));
+ continue;
+ }
+ if (readable.isEmpty()) {
+ // No partition of this table feeds the MV partitions being
refreshed, which is "no row"
+ // rather than "every row": constructPredicates answers the
other way for an empty set,
+ // and that answer would put every row of the table into each
of them.
+ builder.put(table, Sets.newHashSet(BooleanLiteral.FALSE));
+ continue;
+ }
+ builder.put(table, constructPredicatesOfBasePartitions(items,
olapTable, colName));
Review Comment:
Fixed in 47caa34126c. PredicateAdder now writes an external table's
compensation predicate from the whole key of each partition, using the base
table's own partition columns in their own order, exactly as for an olap table;
the MV's column alone is used only for a list default partition, whose own key
cannot pin its rows.
Regression: mtmv_p0/test_mtmv_external_pct_compensation uses a Paimon table
partitioned by (x, d) with the view on d. After appending to the stale
(2021-01-01, 2024-01-01) partition, EXPLAIN shows the base branch `((d =
'2024-01-01') AND (x = '2021-01-01'))` with partition=1/2 (previously `d IN
('2021-01-01', '2022-01-01')`, which reads none), and the rewritten result
carries the appended value in January's total.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -160,30 +223,121 @@ public static Set<Expression>
constructPredicates(Set<PartitionItem> partitions,
*/
@VisibleForTesting
public static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot) {
+ return constructPredicates(partitions, colSlot, Optional.empty());
Review Comment:
Fixed in 47caa34126c. convertPartitionKeyToLiteral now takes the partition
column's own Type instead of Type.fromPrimitiveType(key type), in the whole-key
path and in the default-partition fallback, which also uses the column's
position among the base partition columns instead of position 0.
Coverage note: a temporal external partition column cannot reach this path
in this build. Paimon/Iceberg supply partition values as raw epoch numbers and
PluginDrivenMvccExternalTable.listPartitions drops them ("date literal [...] is
invalid", UNPARTITIONED), so such a table is not a PCT base, and Hive maps
timestamp to DATETIME(6), which cannot express .123. The scale is therefore
pinned at the shared builder in
UpdateMvByPartitionCommandTest#testACompensationOnTheMvColumnKeepsTheKeyScale:
a LIST(k1 DATETIME(3), region) table with a default partition drives the
fallback (the predicate contains no region), and the predicate carries
`2024-01-01 00:00:00.123` and not `2024-01-01 00:00:00.000`. The tuple half is
covered end to end by the Paimon suite above.
--
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]
