github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4369568661
##########
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:
[P1] Check compensated ranges that spill into the next retained bucket
For a `DATETIME(3)` base, use `p_prev=[2024-01-01 00:00:00.123,2024-02-01
00:00:00.123)`, `p_kept=[2024-02-01 00:00:00.123,2024-03-01)`, and an unrelated
third partition. Refresh a monthly MV (`PARTITION BY(date_trunc(ts,'month')) AS
SELECT ts,SUM(v) s FROM t GROUP BY ts`), then query the rows at February 1
`.100` (v=1, in `p_prev`) and `.200` (v=2, in `p_kept`), grouped by
`date_trunc(ts,'month')`.
`getUpperValue` subtracts one second, so `p_prev` rolls to January and
refresh clips its February tail. Query mapping leaves it uncovered, while
February's MV partition remains fresh. This equality compares January with
retained February and misses the overlap;
`retainedPartitionRunsPastItsUpperBound` checks only `p_kept`. The month
transform is therefore accepted, yielding this reduced plan inferred from the
code:
```text
UnionAll
Aggregate(month(ts), SUM(s)) -> Scan MV[February] // (February,2)
Aggregate(month(ts), SUM(v)) -> Filter(query AND p_prev bounds) -> Scan t
// (February,1)
```
The full-precision compensation predicate now reads `.100`, but no aggregate
above the union combines the two February groups into the required sum 3.
Unlike the existing sync-window-cut report, the compensated partition here
rolls to a *different* descriptor and spills into the retained one. Check
actual compensated range overlap, or reject transform coverage when a
compensated range overruns its rolled bucket, and add this fractional-boundary
regression.
--
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]
