foxtail463 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4305780645
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +142,98 @@ 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 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));
Review Comment:
IVM 那边能不能直接传空 Map？null 和空 Map 在这里效果一样，调用方统一保证非 null，就能省掉这一层判断。查不到表时仍沿用 MV
范围，查到空 Set 时仍表示不读数据，这两个语义保留就行。
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java:
##########
@@ -58,6 +58,21 @@ public void apply(MTMVPartitionInfo mvPartitionInfo,
Map<String, String> mvPrope
Optional<String> dateFormat = config.getDateFormat();
Map<MTMVRelatedTableIf, Map<String, PartitionItem>> res =
Maps.newHashMap();
for (Entry<MTMVRelatedTableIf, Map<String, PartitionItem>> entry :
partitionItems.entrySet()) {
+ if (hasDefaultListPartition(entry.getValue())) {
+ // The window is what makes a refresh read fewer base
partitions than the MV partition's own
+ // key range covers. It cannot be applied to a table whose
list partitions have a default
+ // one: that partition takes the rows no other partition of
the table claims, and which of
+ // those rows belong to an MV partition is decided by their
own key rather than by the
+ // partition they were placed in -- ADD PARTITION can claim a
key without moving the rows the
+ // default partition already holds under it, so which side of
the window a row is on is not
+ // something a predicate on the partition columns can say. A
read that is to keep only part of
+ // the MV partition's key range cannot be written for such a
table, so the key range is kept
+ // whole and the mapping names every partition it covers,
which is what the refresh reads for
+ // it. What is given up is the window for this table, not the
correspondence between the read
+ // and the record.
+ res.put(entry.getKey(), entry.getValue());
Review Comment:
IVM 过程中，LIST 分区的基表的 default 分区插入数据，mv 没有对应的分区，无法执行。收集分区 id 的逻辑走了这里，放行了
default
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -1599,6 +1599,23 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>>
calculatePartitionMappi
public Map<String, Map<MTMVRelatedTableIf, Set<String>>>
calculatePartitionMappings(
Map<List<String>, Set<String>> queryUsedBaseTablePartitionMap,
Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots) throws
AnalysisException {
+ return calculatePartitionMappings(queryUsedBaseTablePartitionMap,
pinnedSnapshots, Sets.newHashSet());
+ }
+
+ /**
+ * The same, with every MV partition no base partition describes added to
{@code undescribedMvPartitions}.
+ *
+ * <p>Such a partition is one the alignment has not caught up with: a base
{@code ADD PARTITION} widens the
+ * keys an MV partition covers, so the alignment replaces it with an empty
partition of the wider keys, and
+ * until that is done its own descriptor is one the base partitions no
longer produce. Reading it as no
+ * table feeding it would write no rows over the rows it holds, so a
refresh has to tell it apart from a
+ * partition that really is fed by nothing -- and the default list
partition of a table is named in every
+ * MV partition's mapping, which is why the mapping being non-empty says
nothing about this.
+ */
+ public Map<String, Map<MTMVRelatedTableIf, Set<String>>>
calculatePartitionMappings(
+ Map<List<String>, Set<String>> queryUsedBaseTablePartitionMap,
+ Map<MvccTableInfo, MvccSnapshot> pinnedSnapshots,
+ Set<String> undescribedMvPartitions) throws AnalysisException {
Review Comment:
这里的 undescribedMvPartitions 也是方法算出来的结果，能不能和映射一起返回？现在一个通过返回值拿到，另一个靠修改传进来的
Set，调用方还得把两份结果再放进 Context。用个简单的结果对象装在一起，读起来会更清楚。
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java:
##########
@@ -58,6 +58,21 @@ public void apply(MTMVPartitionInfo mvPartitionInfo,
Map<String, String> mvPrope
Optional<String> dateFormat = config.getDateFormat();
Map<MTMVRelatedTableIf, Map<String, PartitionItem>> res =
Maps.newHashMap();
for (Entry<MTMVRelatedTableIf, Map<String, PartitionItem>> entry :
partitionItems.entrySet()) {
+ if (hasDefaultListPartition(entry.getValue())) {
+ // The window is what makes a refresh read fewer base
partitions than the MV partition's own
+ // key range covers. It cannot be applied to a table whose
list partitions have a default
+ // one: that partition takes the rows no other partition of
the table claims, and which of
+ // those rows belong to an MV partition is decided by their
own key rather than by the
+ // partition they were placed in -- ADD PARTITION can claim a
key without moving the rows the
+ // default partition already holds under it, so which side of
the window a row is on is not
+ // something a predicate on the partition columns can say. A
read that is to keep only part of
+ // the MV partition's key range cannot be written for such a
table, so the key range is kept
+ // whole and the mapping names every partition it covers,
which is what the refresh reads for
+ // it. What is given up is the window for this table, not the
correspondence between the read
+ // and the record.
+ res.put(entry.getKey(), entry.getValue());
Review Comment:
感觉含默认LIST 分区的应该 ban 掉吧，我看物化改写的时候是 ban 掉了，保持设计一致
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescSyncLimitGenerator.java:
##########
@@ -58,6 +58,21 @@ public void apply(MTMVPartitionInfo mvPartitionInfo,
Map<String, String> mvPrope
Optional<String> dateFormat = config.getDateFormat();
Map<MTMVRelatedTableIf, Map<String, PartitionItem>> res =
Maps.newHashMap();
for (Entry<MTMVRelatedTableIf, Map<String, PartitionItem>> entry :
partitionItems.entrySet()) {
+ if (hasDefaultListPartition(entry.getValue())) {
+ // The window is what makes a refresh read fewer base
partitions than the MV partition's own
+ // key range covers. It cannot be applied to a table whose
list partitions have a default
+ // one: that partition takes the rows no other partition of
the table claims, and which of
+ // those rows belong to an MV partition is decided by their
own key rather than by the
+ // partition they were placed in -- ADD PARTITION can claim a
key without moving the rows the
+ // default partition already holds under it, so which side of
the window a row is on is not
+ // something a predicate on the partition columns can say. A
read that is to keep only part of
+ // the MV partition's key range cannot be written for such a
table, so the key range is kept
+ // whole and the mapping names every partition it covers,
which is what the refresh reads for
+ // it. What is given up is the window for this table, not the
correspondence between the read
+ // and the record.
+ res.put(entry.getKey(), entry.getValue());
Review Comment:
话说是不是应该禁止创建这种 mv 啊，带默认 list 分区的基表
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,15 +170,112 @@ public static Pair<Map<BaseTableInfo, Set<String>>,
Map<BaseColInfo, Set<String>
if (allCompensateIsNull) {
return null;
}
+ // A base partition the query reads that no MV partition of this plan
is named from -- one the
+ // partition_sync_limit window left out, say -- is one the MV does not
answer for: a partition of the MV
+ // reads a table through the partitions it does name, and the mapping
has no entry for this one. Its
+ // rows have to come from the base table, and two things follow from
that, decided apart.
+ //
+ // The base branch of the union reads those partitions. A table the
mapping mentions not at all is
+ // uncovered in every partition the query uses, so it is read for the
query's own partitions there
+ // rather than skipped -- without this the plan would answer for that
table out of the MV alone.
+ //
+ // And the MV partitions this plan uses have to be taken out when they
hold rows of such a partition: a
+ // partition the window has since left out can have been materialized
in one of them by an earlier
+ // refresh, and the union would then count its rows twice. The refresh
snapshot names exactly the
+ // partitions a refresh read into each MV partition, so that is what
this is asked of. (It is not the
+ // keys: a list partition's default one holds the keys no other
partition of the table claims, so a
+ // partition can be read into an MV partition whose keys do not cover
it.)
+ Map<MTMVRelatedTableIf, Set<String>> uncoveredByTable =
Maps.newLinkedHashMap();
+ boolean uncoveredRecorded = false;
+ for (MTMVRelatedTableIf pctTable :
mtmv.getMvPartitionInfo().getPctTables()) {
+ Set<String> queryUsed =
queryUsedBaseTablePartitionMap.get(pctTable.getFullQualifiers());
+ if (CollectionUtils.isEmpty(queryUsed)) {
+ continue;
+ }
+ // The mapping of a table the query's filtering left out entirely
is empty, so a table it does not
+ // mention at all is a table every query-used partition of which
is uncovered.
+ Map<String, Set<String>> tableMapping =
mtmvRelatedTableIfMapMap.getOrDefault(pctTable, Maps.newHashMap());
+ Set<String> named = Sets.newHashSet();
+ for (Set<String> names : tableMapping.values()) {
+ named.addAll(names);
+ }
+ Set<String> uncovered = Sets.difference(queryUsed, named);
+ if (uncovered.isEmpty()) {
+ continue;
+ }
+ uncoveredByTable.put(pctTable, uncovered);
+ Set<String> recorded = Sets.newHashSet();
+ MTMVRefreshSnapshot refreshSnapshot = mtmv.getRefreshSnapshot();
+ if (refreshSnapshot != null) {
+ for (String mvPartitionName :
rewrittenPlanUsePartitionNameSet) {
+
recorded.addAll(refreshSnapshot.getPctSnapshots(mvPartitionName, new
BaseTableInfo(pctTable)));
+ }
+ }
+ if (!Sets.intersection(uncovered, recorded).isEmpty()) {
+ uncoveredRecorded = true;
+ }
+ }
+ // Narrowing a table's read keeps the MV partitions answering for the
rest, and that only holds while
+ // one input of the query reads the tables it narrows: the pass that
turns the compensation into a plan
+ // puts the filters it is given on one join, so two narrowed inputs
read the pairs both filters reach
+ // and lose the mixed ones -- an a_old x b_kept row, say, when A is
narrowed for a partition the
+ // mapping does not name and B for one the MV no longer holds, or when
one table is narrowed while a
+ // self-join reads it through two inputs, both of which the filter
reaches. Nor can a table be left as
+ // it is while the MV partitions that answered for it are taken out.
So the MV partitions are taken
+ // out, and every table the MV is partitioned on is read for the
query's own partitions, whenever more
+ // than one table is narrowed, a single table is narrowed while the
query reads such a table through
+ // more inputs than that, or a refresh read one of the uncovered
partitions into an MV partition.
+ int narrowedTables = 0;
+ for (MTMVRelatedTableIf pctTable :
mtmv.getMvPartitionInfo().getPctTables()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new BaseTableInfo(pctTable));
+ Set<String> queryUsed =
queryUsedBaseTablePartitionMap.get(pctTable.getFullQualifiers());
+ Set<String> stale = pctInfo == null ? null :
baseTablePartitionNeedUnionNameMap.get(pctInfo);
+ boolean narrowedByStalePass = CollectionUtils.isNotEmpty(stale) &&
!stale.containsAll(queryUsed);
+ if (narrowedByStalePass || uncoveredByTable.containsKey(pctTable))
{
+ narrowedTables++;
+ }
+ }
+ boolean takeMvPartitionsOut = uncoveredRecorded || narrowedTables > 1
+ || (narrowedTables > 0 && pctInputsInQuery > narrowedTables);
+ if (!uncoveredByTable.isEmpty()) {
+ for (MTMVRelatedTableIf pctTable :
mtmv.getMvPartitionInfo().getPctTables()) {
+ Set<String> queryUsed =
queryUsedBaseTablePartitionMap.get(pctTable.getFullQualifiers());
+ if (CollectionUtils.isEmpty(queryUsed)) {
+ continue;
+ }
+ BaseColInfo pctInfo = pctInfoMap.get(new
BaseTableInfo(pctTable));
+ if (pctInfo == null) {
+ continue;
+ }
+ if (takeMvPartitionsOut) {
+ baseTablePartitionNeedUnionNameMap.put(pctInfo, new
HashSet<>(queryUsed));
+ continue;
+ }
+ Set<String> uncovered = uncoveredByTable.get(pctTable);
+ if (CollectionUtils.isEmpty(uncovered)) {
+ continue;
+ }
+ baseTablePartitionNeedUnionNameMap
+ .computeIfAbsent(pctInfo, k -> new
HashSet<>()).addAll(uncovered);
+ }
+ if (takeMvPartitionsOut &&
!rewrittenPlanUsePartitionNameSet.isEmpty()) {
+ mvPartitionNeedRemoveNameMap
+ .computeIfAbsent(new BaseTableInfo(mtmv), k -> new
HashSet<>())
+ .addAll(rewrittenPlanUsePartitionNameSet);
Review Comment:
MVScan（没有任何分区） UnionAll 完整原查询的深拷贝
进了优化器会一直递归，直到 memo 爆了
mv 没任分区应该直接拒绝该候选
--
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]
