yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4369631203
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/mv/PartitionCompensator.java:
##########
@@ -144,19 +212,338 @@ 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 (!takeMvPartitionsOut) {
+ // The MV partitions stay, and the uncovered partitions are added
to the base branch beside them.
+ // Such an uncovered partition can answer a group an MV partition
also answers when its keys
+ // meet, in the MV's partition columns, the keys of the base
partitions that MV partition is
+ // recorded with: a merged LIST component {kept (2020,1), bridge
(2038,2)} projected onto d
+ // holds 2020 and 2038, and an uncovered old (2020,2) is read
beside it, so the union would
+ // answer d=2020 twice. A RANGE partition whose range no retained
MV partition covers answers
+ // its own groups and is not recorded here. Only the MV partitions
that stay count: one the
+ // staleness pass has already scheduled for removal is re-read
through the base branch too, so
+ // its keys are the compensation's own, not a retained group the
uncovered partition would meet.
+ Set<String> removedMvPartitions = Sets.newHashSet();
+
mvPartitionNeedRemoveNameMap.values().forEach(removedMvPartitions::addAll);
+ Set<String> retainedMvPartitions = Sets.difference(
+ rewrittenPlanUsePartitionNameSet, removedMvPartitions);
+ for (Entry<MTMVRelatedTableIf, Set<String>> entry :
uncoveredByTable.entrySet()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new
BaseTableInfo(entry.getKey()));
+ if (pctInfo == null) {
+ continue;
+ }
+
uncoveredNarrowedBasePartitions.addAll(overlappingUncoveredListPartitions(
+ entry.getKey(), pctInfo, entry.getValue(),
retainedMvPartitions,
+ mtmvRelatedTableIfMapMap));
+ }
+ addColumnsWhoseCompensationSharesRetainedBucket(mtmv, pctInfoMap,
+ baseTablePartitionNeedUnionNameMap, uncoveredByTable,
retainedMvPartitions,
+ columnsSharingRetainedBucket);
+ }
+ 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);
+ }
+ allCompensateIsNull = false;
+ }
+
// merge all partition to delete or union
Set<String> needRemovePartitionSet = new HashSet<>();
mvPartitionNeedRemoveNameMap.values().forEach(needRemovePartitionSet::addAll);
mvPartitionNeedRemoveNameMap.replaceAll((k, v) ->
needRemovePartitionSet);
- // consider multi base table partition name not same, how to handle it?
- Set<String> needUnionPartitionSet = new HashSet<>();
-
baseTablePartitionNeedUnionNameMap.values().forEach(needUnionPartitionSet::addAll);
- baseTablePartitionNeedUnionNameMap.replaceAll((k, v) ->
needUnionPartitionSet);
+ // The partitions to union stay with the table they belong to: a
partition name means that table's
+ // keys, and two tables of a multi-table MV can use the same name for
different keys -- the merge of the
+ // cross-table LIST descriptors makes such MVs buildable. Unioning the
names across the tables would
+ // read the other table's partitions too, and with the union rewrite
the MV's own branch already
+ // supplies their rows, so the answer would count them twice.
return Pair.of(mvPartitionNeedRemoveNameMap,
baseTablePartitionNeedUnionNameMap);
}
+ /**
+ * The uncovered partitions of this table whose keys meet, in the MV's
partition columns, the keys of the
+ * base partitions a retained MV partition is recorded with. Only a LIST
partitioned table can have them:
+ * a merged component's MV partition holds several keys of the MV's
partition column, and an uncovered
+ * partition can carry one of them. A RANGE base partition is either
covered by a retained MV partition
+ * -- then it is not uncovered -- or outside its range, where it answers
groups of its own.
+ */
+ private static Set<String> overlappingUncoveredListPartitions(
+ MTMVRelatedTableIf table, BaseColInfo pctInfo, Set<String>
uncovered,
+ Set<String> retainedMvPartitions,
+ Map<MTMVRelatedTableIf, Map<String, Set<String>>> tableMappings) {
+ if (!(table instanceof OlapTable)
+ || !(((OlapTable) table).getPartitionInfo() instanceof
ListPartitionInfo)) {
+ return Sets.newHashSet();
+ }
+ OlapTable olapTable = (OlapTable) table;
+ List<Column> partitionColumns = olapTable.getPartitionColumns();
+ int keyPosition = -1;
+ for (int i = 0; i < partitionColumns.size(); i++) {
+ if
(partitionColumns.get(i).getName().equalsIgnoreCase(pctInfo.getColName())) {
+ keyPosition = i;
+ break;
+ }
+ }
+ if (keyPosition < 0) {
+ return Sets.newHashSet();
+ }
+ Set<LiteralExpr> retainedValues = Sets.newHashSet();
+ Map<String, Set<String>> mvToBasePartitions = tableMappings.get(table);
+ if (mvToBasePartitions != null) {
+ for (String mvPartition : retainedMvPartitions) {
+ for (String basePartition :
mvToBasePartitions.getOrDefault(mvPartition, Sets.newHashSet())) {
+ addProjectedListKeys(olapTable, basePartition,
keyPosition, retainedValues);
+ }
+ }
+ }
+ Set<String> res = Sets.newHashSet();
+ for (String partition : uncovered) {
+ Set<LiteralExpr> values = Sets.newHashSet();
+ addProjectedListKeys(olapTable, partition, keyPosition, values);
+ if (values.stream().anyMatch(retainedValues::contains)) {
+ res.add(partition);
+ }
+ }
+ return res;
+ }
+
+ private static void addProjectedListKeys(OlapTable table, String
partitionName, int keyPosition,
+ Set<LiteralExpr> out) {
+ Partition partition = table.getPartition(partitionName);
+ if (partition == null) {
+ return;
+ }
+ PartitionItem item =
table.getPartitionInfo().getItem(partition.getId());
+ if (!(item instanceof ListPartitionItem)) {
+ return;
+ }
+ for (PartitionKey key : ((ListPartitionItem) item).getItems()) {
+ if (keyPosition < key.getKeys().size()) {
+ out.add(key.getKeys().get(keyPosition));
+ }
+ }
+ }
+
+ /**
+ * The base columns whose compensation reads a partition that falls in the
same transform bucket as an MV
+ * partition that stays.
+ *
+ * <p>The aggregate rule lets a grouping by the view's own transform keep
the two branches apart because
+ * the view's partitions are disjoint in that transform's value. That
holds only while each retained
+ * partition holds the whole bucket: {@code partition_sync_limit} drops
base partitions before the roll-up,
+ * so a retained year partition can hold the days the window kept while an
expired day of the same year is
+ * compensated -- grouping by {@code date_trunc(d, 'year')} then answers
that year from both branches.
+ * Likewise a retained partition can hold a recorded base partition that
runs past its own upper bound: the
+ * roll-up of a {@code DATETIME(3)} partition ending a fraction of a
second after a bucket boundary takes
+ * off one second and lands in the bucket before it, while the rows past
the boundary are grouped into the
+ * next one. Such a column is recorded here, so the aggregate rule
declines the transform coverage and
+ * the query is answered from the base table instead.
+ */
+ private static void addColumnsWhoseCompensationSharesRetainedBucket(MTMV
mtmv,
+ Map<BaseTableInfo, BaseColInfo> pctInfoMap,
+ Map<BaseColInfo, Set<String>> baseTablePartitionNeedUnionNameMap,
+ Map<MTMVRelatedTableIf, Set<String>> uncoveredByTable,
+ Set<String> retainedMvPartitions,
+ Set<BaseColInfo> columnsSharingRetainedBucket) throws
AnalysisException {
+ MTMVPartitionInfo mvPartitionInfo = mtmv.getMvPartitionInfo();
+ if (mvPartitionInfo.getPartitionType() != MTMVPartitionType.EXPR ||
retainedMvPartitions.isEmpty()) {
+ return;
+ }
+ MTMVPartitionExprService exprService;
+ try {
+ exprService =
MTMVPartitionExprFactory.getExprService(mvPartitionInfo.getExpr());
+ } catch (AnalysisException e) {
+ // The view was created from an expression this factory reads, so
it cannot happen.
+ throw new IllegalStateException("Cannot read the partition
expression of " + mtmv.getName(), e);
+ }
+ Map<String, PartitionItem> mvItems = mtmv.getAndCopyPartitionItems();
+ Map<PartitionKeyDesc, String> retainedDescs = Maps.newHashMap();
+ for (String mvPartitionName : retainedMvPartitions) {
+ PartitionItem mvItem = mvItems.get(mvPartitionName);
+ if (mvItem != null) {
+ retainedDescs.put(mvItem.toPartitionKeyDesc(),
mvPartitionName);
+ }
+ }
+ MTMVRefreshSnapshot refreshSnapshot = mtmv.getRefreshSnapshot();
+ for (MTMVRelatedTableIf pctTable : mvPartitionInfo.getPctTables()) {
+ BaseColInfo pctInfo = pctInfoMap.get(new BaseTableInfo(pctTable));
+ if (pctInfo == null) {
+ continue;
+ }
+ Set<String> compensated = Sets.newHashSet(
+ baseTablePartitionNeedUnionNameMap.getOrDefault(pctInfo,
Sets.newHashSet()));
+ compensated.addAll(uncoveredByTable.getOrDefault(pctTable,
Sets.newHashSet()));
+ if (compensated.isEmpty()) {
+ continue;
+ }
+ Optional<MvccSnapshot> snapshot =
MvccUtil.getSnapshotFromContext(pctTable);
+ // The partitions are read once and shared by both checks below:
each takes the table map, and
+ // the map is every partition of the table, not just the ones
either check looks up.
+ Map<String, PartitionItem> baseItems =
pctTable.getAndCopyPartitionItems(snapshot);
+ boolean sharesRetainedBucket;
+ try {
+ sharesRetainedBucket =
compensatedSharesRetainedBucket(pctTable, baseItems, snapshot,
+ exprService, mvPartitionInfo, compensated,
retainedDescs)
+ || (refreshSnapshot != null &&
retainedPartitionRunsPastItsUpperBound(
+ pctTable, pctInfo.getColName(), snapshot,
baseItems, refreshSnapshot,
+ retainedMvPartitions, mvItems));
+ } catch (AnalysisException e) {
+ // Which bucket one of them falls in cannot be told, so the
transform is not trusted to keep
+ // the branches apart.
+ sharesRetainedBucket = true;
+ }
+ if (sharesRetainedBucket) {
+ columnsSharingRetainedBucket.add(pctInfo);
+ }
+ }
+ }
+
+ private static boolean compensatedSharesRetainedBucket(MTMVRelatedTableIf
pctTable,
+ Map<String, PartitionItem> items, Optional<MvccSnapshot> snapshot,
+ MTMVPartitionExprService exprService, MTMVPartitionInfo
mvPartitionInfo, Set<String> compensated,
+ Map<PartitionKeyDesc, String> retainedDescs) throws
AnalysisException {
+ int colPos = mvPartitionInfo.getPctColPos(pctTable);
+ for (String partitionName : compensated) {
+ PartitionItem item = items.get(partitionName);
+ if (item == null) {
+ continue;
+ }
+ PartitionKeyDesc rollUpDesc =
exprService.generateRollUpPartitionKeyDesc(
+ item.toPartitionKeyDesc(colPos), mvPartitionInfo,
pctTable, snapshot);
+ if (retainedDescs.containsKey(rollUpDesc)) {
Review Comment:
Good catch. A compensated partition can reach into a later bucket by itself,
with nothing on the retained side to show it; fixed in f9e11e7feaa.
For a `DATETIME(3)` base the exact contiguous layout cannot be created:
`p_kept=[2024-02-01 00:00:00.123, 2024-03-01)` fails the roll-up with
`partition values not equal, beginTruncTime: 2024-02-01 00:00:00.000,
endTruncTime: 2024-02-01 00:00:00` (the truncated bounds compare as literals of
different scale), and overlapping base ranges are rejected anyway. The same
miss is reachable with a gap: `p_prev=[2024-01-01 00:00:00.123, 2024-02-01
00:00:00.123)` followed by `[2024-02-05, 2024-03-01)`. `p_prev` rolls up to
January and the refresh clips its `.100` row in February, while the view's
February partition is fresh and records only the later partition, so
`retainedPartitionRunsPastItsUpperBound` sees no overrun and the descriptor
equality compares January with retained February.
Reproduced on a live cluster: the rewritten February query returned two
rows, `(2024-02, 1000)` from the view and `(2024-02, 1)` from the compensation,
while the base answer is one row with 1001.
The fix records a compensated partition whose actual upper bound reaches
past the upper bound its rolled descriptor names as sharing a retained bucket.
The aggregate rule then declines the transform coverage and the query is
answered from the base table. A partition ending exactly on the bucket boundary
is unaffected, so ordinary DATE and DATETIME(0) roll-ups keep the transform.
Regression: `mtmv_p0/test_mtmv_base_partition_read_scope`, new `range_spill`
section -- the view's partition holds no row of the clipped fraction, and the
February query returns a single 1001 row with no union in the plan.
--
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]
