yujun777 commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4142415638
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +141,63 @@ 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;
+ }
+ 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;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ Set<PartitionItem> items = Sets.newHashSet();
+ for (String partitionName : readable) {
+
items.add(olapTable.getPartitionItemOrAnalysisException(partitionName));
+ }
+ if (hasDefaultListPartition(olapTable)) {
+ // A list partitioned table's default partition takes the rows
no other partition of it
+ // claims, and it is not a partition of that table the MV's
own partition is recorded with:
+ // a partition of the MV takes the rows whose own key falls in
it, wherever the base table
+ // put them, so the rows this refresh is about are the ones
the MV partition's key range
+ // names rather than the ones the base partition it is
recorded with holds. A table that has
+ // such a partition is therefore read the way an unscoped one
is. That read can be seen to be
+ // too wide -- it is the one this scope exists to narrow --
rather than one that drops rows
+ // which belong to the MV partition being refreshed.
+ builder.put(table, constructPredicates(mvItems, colName));
Review Comment:
Fixed in 800af5138a3. The default list partition of a base table is now
named in the mapping of every MV partition that reads it, which is the same set
the read uses and the set the refresh records, so `isSyncWithPartitions`
compares it: an insert into it changes a version the MV partition checks and
the partition stops calling itself synchronized.
Pinned as a case in the read scope suite (`list_default_tracked`): the MV is
refreshed, a row is inserted into the default partition afterwards, and
`SyncWithBaseTables` is `false` for it, where before the change it stayed
`true`.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -170,20 +227,109 @@ public static Set<Expression>
constructPredicates(Set<PartitionItem> partitions,
}
} else {
for (PartitionItem item : partitions) {
- predicates.add(convertRangePartitionToCompare(item, colSlot));
+ predicates.add(convertRangePartitionToCompare(item, colSlot,
Optional.empty()));
}
}
return predicates;
}
- private static Expression convertPartitionKeyToLiteral(PartitionKey key) {
- return Literal.fromLegacyLiteral(key.getKeys().get(0),
- Type.fromPrimitiveType(key.getTypes().get(0)));
+ /**
+ * The predicate a base table is read through when the refresh is to read
exactly these partitions of it.
+ *
+ * <p>A partition of a list partitioned table holds one key per partition
column, and the column the MV
+ * partition is named by is only one of them. A predicate on that column
alone also reaches the
+ * partitions whose other keys differ -- a table partitioned by (d,
region) has one partition of
+ * (d0, 'US') and one of (d0, 'EU'), and `d = d0` reaches both, while only
the second is a partition
+ * this refresh is to read; a later drop of the first would then leave its
rows in the MV partition
+ * while the snapshot, which names only the second, still calls it
synchronized. So a list partition is
+ * pinned to its whole key. A range partition is pinned to its bounds,
which is the same thing: a base
+ * table partitioned by range has a single partition column, see
+ * {@code RangePartitionItem#toPartitionKeyDesc(int)}.
+ *
+ * <p>The partitions are never empty: a table the caller scopes with no
partition is read as nothing
+ * before this is reached, see {@code constructTableWithPredicates}.
+ */
+ private static Set<Expression>
constructPredicatesOfBasePartitions(Set<PartitionItem> partitions,
+ OlapTable baseTable, String colName) throws AnalysisException {
+ List<Column> partitionColumns = baseTable.getPartitionColumns();
+ List<Type> partitionColumnTypes = Lists.transform(partitionColumns,
Column::getType);
+ if (!(partitions.iterator().next() instanceof ListPartitionItem)) {
+ Set<Expression> predicates = new HashSet<>();
+ for (PartitionItem item : partitions) {
+ predicates.add(convertRangePartitionToCompare(item, new
UnboundSlot(colName),
+ Optional.of(partitionColumnTypes.get(0))));
+ }
+ return predicates;
+ }
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(new UnboundSlot(partitionColumn.getName()));
+ }
+ Set<Expression> predicates = new HashSet<>();
+ for (PartitionItem item : partitions) {
+ predicates.add(convertListPartitionToKey(item, partitionSlots,
partitionColumnTypes));
+ }
+ return predicates;
+ }
+
+ /**
+ * Whether this table has a list partition that takes the rows no other
partition of it claims. Such a
+ * partition holds rows for every key its table can be read by, so which
rows of it belong to a partition
+ * of the MV is the MV partition's own question and not the partition's.
+ */
+ private static boolean hasDefaultListPartition(OlapTable table) {
+ PartitionInfo partitionInfo = table.getPartitionInfo();
+ if (!(partitionInfo instanceof ListPartitionInfo)) {
+ return false;
+ }
+ return ((ListPartitionInfo)
partitionInfo).getIdToItem(false).values().stream()
Review Comment:
Fixed in 800af5138a3, in the direction you point at: the fact is no longer
read from `getIdToItem` at all.
Whether a table is read this way is now taken from the mapped partitions a
refresh already has in hand -- one of them being a default partition is the
condition -- so there is no lock-free walk of the table's partition metadata,
no per-batch O(N) scan, and nothing new to serialise against a concurrent ADD
or DROP PARTITION. The underlying fact, which table has such a partition, is
computed once per mapping in `MTMV#calculatePartitionMappings` and merged into
every MV partition's entry there, which is also what makes it available across
batches instead of per batch.
--
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]
