yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4269221120
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +142,90 @@ 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));
+ 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. Which of the rows that
+ // range reaches sit in an explicit partition and which in the
default one is not something a
+ // predicate on the partition columns can say -- ADD PARTITION
claims a key without moving the
+ // rows a default partition already holds under it -- so this
table is not scoped at all, and
+ // the mapping is kept whole for it for the same reason: see
+ // MTMVRelatedPartitionDescSyncLimitGenerator, which leaves
such a table out of the window.
+ // 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;
+ }
+ if (items.iterator().next() instanceof ListPartitionItem) {
+ // A key of a list partition is a value the MV partition's own
key set was written from, so
+ // the whole key pinned here is inside the range this batch
writes and nothing is added to it.
+ builder.put(table, constructPredicatesOfBasePartitions(items,
olapTable, colName));
+ continue;
+ }
+ // A partition of a base range table can end past the MV partition
it is rolled up into, and the
+ // roll-up keeps such a partition: MTMVPartitionExprDateTrunc
takes the last second off the base's
+ // upper bound before truncating it, so a DATETIME(3) partition
ending at 2025-01-01 00:00:00.123
+ // is rolled up into the year 2024. This batch writes the year
2024 partition alone, and a row of
+ // that partition just past its upper bound -- 2025-01-01
00:00:00.100 -- is then read by a batch
+ // that has no partition to put it in: the sink counts it as
filtered and `enable_insert_strict`,
+ // on by default, fails the refresh for it. So the mapped
partitions' bounds are read together
+ // with the MV partitions' own ranges -- the term every other
table is read through -- and what is
+ // read is the rows of the recorded partitions that these MV
partitions hold.
+ Expression mvRange =
ExpressionUtils.or(constructPredicates(mvItems, colName));
+ Set<Expression> scoped = Sets.newHashSet();
+ for (Expression partition :
constructPredicatesOfBasePartitions(items, olapTable, colName)) {
+ scoped.add(ExpressionUtils.and(partition, mvRange));
Review Comment:
Fixed in 17d80797034. The two sides are now one disjunction each, built
once: `AND(OR(baseRanges), OR(mvRanges))`. It is the same read (`(b1 ∧ MV) ∨
(b2 ∧ MV) … = (b1 ∨ b2 …) ∧ MV`), with the MV side written once instead of N
times, so the expression the analyzer rebinds is linear rather than O(N^2). The
scoped case in `UpdateMvByPartitionCommandTest` asserts the single term and
that it names every mapped partition's bounds; the DATETIME(3) boundary case
still pins the read to the MV partition's own bound. 14 tests pass.
Note on the range below: the `OR` I write here is over one term per mapped
base partition, not over a list's key set, so the case you describe
(`refresh_partition_num = N` with N daily partitions read as one batch) is
exactly what the shape now avoids -- that batch's predicate is `AND(OR of N day
bounds, OR of the MV buckets)`.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelatedPartitionDescOnePartitionColGenerator.java:
##########
@@ -60,18 +60,14 @@ public void apply(MTMVPartitionInfo mvPartitionInfo,
Map<String, String> mvPrope
for (Entry<MTMVRelatedTableIf, Map<String, PartitionItem>> entry :
relatedPartitionItems.entrySet()) {
int relatedColPos = mvPartitionInfo.getPctColPos(entry.getKey());
Map<PartitionKeyDesc, Set<String>> onePctRes = Maps.newHashMap();
- Set<String> queryUsedPartitions =
queryUsedPartitionMap.get(entry.getKey().getFullQualifiers());
for (Entry<String, PartitionItem> onePctEntry :
entry.getValue().entrySet()) {
- if (queryUsedPartitions != null &&
!queryUsedPartitions.contains(onePctEntry.getKey())) {
- continue;
- }
PartitionKeyDesc partitionKeyDesc =
onePctEntry.getValue().toPartitionKeyDesc(relatedColPos);
- if (onePctRes.containsKey(partitionKeyDesc)) {
- onePctRes.get(partitionKeyDesc).add(onePctEntry.getKey());
- } else {
- onePctRes.put(partitionKeyDesc,
Sets.newHashSet(onePctEntry.getKey()));
- }
+ onePctRes.computeIfAbsent(partitionKeyDesc, k ->
Sets.newHashSet()).add(onePctEntry.getKey());
Review Comment:
Fixed in cc5e5f3355e. You are right that the early filter belongs back for
range, and the reason it can be there is the expansion: `MTMVPartitionExpander`
(through `MTMV#getEffectiveQueryUsedBaseTablePartitionMap`, called by
`calculatePartitionMappings`) turns the query's base partitions into all the
base partitions of the MV buckets the query touches, so what the early filter
leaves out is a bucket the query reads nothing of.
`RangePartitionItem#toPartitionKeyDesc(int)` also states that a range base
table of an MV has a single partition column, so one partition is one desc and
the desc the transfer generator would drop is the one dropped here -- without
the `rollUpRange` date work behind it. A list partition keeps its desc: its
keys meet the other partitions' of the component, and those are merged after
the grouping, in the transfer generator, which then filters the partitions per
table.
The two sides are covered by
`MTMVRelatedPartitionDescGeneratorTest#testAQueryUnusedRangePartitionIsLeftOutBeforeTheRollUp`
(one desc where the query reads one, three where the query filter is empty; it
fails with the early filter removed) and
`testAQueryUnusedListPartitionIsKeptForTheMerge` (the list descs of a filtered
run are the ones of an unfiltered one). `MTMVRewriteUtilTest`, `MTMVTest`,
`MTMVPartitionUtilTest`, `MTMVTaskTest`,
`MTMVRelatedPartitionDescSyncLimitGeneratorTest`,
`ExplainRefreshMTMVCommandTest`, `RefreshMTMVCommandTest`,
`PartitionCompensatorTest`, `AsyncMaterializationContextTest` and
`CheckRowPolicyTest` pass as well.
--
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]
