github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4141908573
##########
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()) {
Review Comment:
[P1] Check default LIST coverage before treating an empty mapping as FALSE.
In a supported two-PCT MV, A has an explicit LIST partition for key 1, B has an
explicit partition for key 2 plus a default partition holding key 1, and the MV
joins A and B on that key. The key-1 MV partition is created from A; B has no
mapped name for it because its default projects the MIN sentinel. This branch
injects FALSE into B's scan, making the join empty and dropping the committed
key-1 row on the first refresh. The earlier default-partition thread covers the
nonempty-map branch. Apply the MV-key fallback here too, with matching snapshot
dependencies, and test the two-PCT case.
##########
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:
[P2] Resolve default-LIST presence once from stable partition metadata.
getIdToItem(false) returns a mutable HashMap and requires the table read lock,
but this call runs after buildRefreshContext released its locks; concurrent
ADD/DROP PARTITION can make anyMatch throw an unretried
ConcurrentModificationException. With no default, it also walks all N
partitions for every refresh batch; the default one-partition batch size adds
O(N²) metadata visits for N one-to-one MV partitions. Capture this fact under
the table lock and reuse it across batches.
##########
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:
[P1] Track default LIST partitions in every MV partition that reads them. In
the new list_default fixture, this fallback puts the p_default row with
d=2020-01-01 into the explicit 2020 MV partition, but its snapshot records only
p_explicit because the default sentinel maps elsewhere. A later insert into
p_default changes none of the versions checked for 2020, so automatic refresh
skips it and transparent rewrite can serve stale rows. The existing
default-partition thread covers the initial read; this is the missing
dependency after that read. Add a post-refresh default insert test.
##########
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()
+ .anyMatch(PartitionItem::isDefaultPartition);
+ }
+
+ /**
+ * One partition of a list partitioned table, pinned to the whole of each
key it holds: the keys are
+ * what tells it apart from a partition that shares a key with it, and the
value of a key a row does not
+ * have is asked for as {@code IS NULL}, since no comparison to it is ever
true.
+ *
+ * <p>A list partitioned table's default partition is not one of these:
its key is the sentinel the rows
+ * no other partition claims are placed by rather than a value, and a
partition of the MV takes the rows
+ * whose own key falls in it wherever the base table put them. What it
holds cannot be said with a
+ * predicate on the partition columns, so a table that has one is read the
way an unscoped one is, see
+ * {@code constructTableWithPredicates}.
+ */
+ private static Expression convertListPartitionToKey(PartitionItem item,
List<Slot> partitionSlots,
+ List<Type> partitionColumnTypes) {
+ List<Expression> keys = new ArrayList<>();
+ for (PartitionKey key : ((ListPartitionItem) item).getItems()) {
+ List<Expression> oneKey = new ArrayList<>();
+ for (int pos = 0; pos < partitionSlots.size(); pos++) {
+ Expression value = convertPartitionKeyToLiteral(key, pos,
+ Optional.of(partitionColumnTypes.get(pos)));
+ oneKey.add(value instanceof NullLiteral ? new
IsNull(partitionSlots.get(pos))
+ : new EqualTo(partitionSlots.get(pos), value));
+ }
+ keys.add(ExpressionUtils.and(oneKey));
+ }
+ Preconditions.checkState(!keys.isEmpty(), "a list partition holds at
least one key: %s", item);
+ return ExpressionUtils.or(keys);
+ }
+
+ /**
+ * A partition key is a value of the partition column it is written
against, and what tells two of them
+ * apart can be a scale the key's primitive type does not carry: a literal
rounded to a coarser one is
+ * one no row of the partition compares equal to, and the rows of the
partition are then read as none.
+ * The callers that have the column pass its type; the ones that do not
leave the key's own primitive
+ * type, which is what a reader of these predicates was given before.
+ */
+ private static Expression convertPartitionKeyToLiteral(PartitionKey key,
int keyPos,
+ Optional<Type> columnType) {
+ return Literal.fromLegacyLiteral(key.getKeys().get(keyPos),
+ columnType.orElseGet(() ->
Type.fromPrimitiveType(key.getTypes().get(keyPos))));
}
private static Expression convertListPartitionToIn(PartitionItem item,
Slot col) {
List<Expression> inValues = ((ListPartitionItem)
item).getItems().stream()
- .map(UpdateMvByPartitionCommand::convertPartitionKeyToLiteral)
+ .map(key -> convertPartitionKeyToLiteral(key, 0,
Optional.empty()))
Review Comment:
[P1] Preserve DATETIMEV2 scale when the new default-LIST fallback reaches
this conversion. A LIST(ts DATETIME(3)) base table with an explicit key
2024-02-01 00:00:00.123 and a default partition sends its PARTITION BY ts MV
through constructPredicates(mvItems, colName). This Optional.empty() uses
Type.fromPrimitiveType(DATETIMEV2) (scale 0), so the generated ts IN predicate
compares against .000 and omits the committed .123 row. The earlier scale
thread covered the scoped base-partition path; the fallback bypasses that fix.
Add a combined fractional/default 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]
