github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4220192680
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -160,30 +218,218 @@ public static Set<Expression>
constructPredicates(Set<PartitionItem> partitions,
*/
@VisibleForTesting
public static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot) {
+ return constructPredicates(partitions, colSlot, Optional.empty());
+ }
+
+ private static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot,
+ Optional<Type> columnType) {
Set<Expression> predicates = new HashSet<>();
if (partitions.isEmpty()) {
return Sets.newHashSet(BooleanLiteral.TRUE);
}
if (partitions.iterator().next() instanceof ListPartitionItem) {
for (PartitionItem item : partitions) {
- predicates.add(convertListPartitionToIn(item, colSlot));
+ predicates.add(convertListPartitionToIn(item, colSlot,
columnType));
}
} else {
for (PartitionItem item : partitions) {
- predicates.add(convertRangePartitionToCompare(item, colSlot));
+ predicates.add(convertRangePartitionToCompare(item, colSlot,
columnType));
+ }
+ }
+ return predicates;
+ }
+
+ /**
+ * 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) {
+ return constructPredicatesOfBasePartitions(partitions, baseTable,
colName, UnboundSlot::new);
+ }
+
+ /**
+ * The same, with the partition columns read through the slots the caller
names them by. A caller that
+ * writes these predicates into a plan that is already bound and is never
bound again -- the union
+ * compensation -- has to pass that plan's own slots: an unbound slot
there fails the rewrite instead of
+ * narrowing the read.
+ */
+ private static Set<Expression>
constructPredicatesOfBasePartitions(Set<PartitionItem> partitions,
+ OlapTable baseTable, String colName, Function<String, Slot>
slotOfColumn) {
+ 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,
slotOfColumn.apply(colName),
+ Optional.of(partitionColumnTypes.get(0))));
}
+ return predicates;
+ }
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(slotOfColumn.apply(partitionColumn.getName()));
+ }
+ Set<Expression> predicates = new HashSet<>();
+ for (PartitionItem item : partitions) {
+ predicates.add(convertListPartitionToKey(item, partitionSlots,
partitionColumnTypes));
}
return predicates;
}
- private static Expression convertPartitionKeyToLiteral(PartitionKey key) {
- return Literal.fromLegacyLiteral(key.getKeys().get(0),
- Type.fromPrimitiveType(key.getTypes().get(0)));
+ /**
+ * The predicate for a table whose mapped partitions include a list
partition's default one.
+ *
+ * <p>That partition takes the rows no other partition of the table
claims, and an MV partition takes the
+ * rows whose own key falls in it wherever the base table placed them --
so those rows are the ones the MV
+ * partition's own key range names, and no predicate on the partition
columns picks them out on its own: a
+ * test on those columns reaches the rows an explicit partition holds as
readily as it reaches theirs.
+ * The read is therefore written the other way round, as the record's own
partitions plus what the key
+ * range holds beyond them:
+ *
+ * <ul>
+ * <li>the partitions this MV partition is recorded with, pinned by
their whole key, and</li>
+ * <li>the rows of the MV partitions' key range that no explicit
partition the mapping does not name
+ * holds, which is the default partition's rows in that range and
nothing else.</li>
+ * </ul>
+ *
+ * <p>An explicit partition the window left out shares the MV partition's
key range with a retained one --
+ * the shape this scope is about, a list partitioned table's partitions
holding several keys of the MV's
+ * partition column -- and reading it back would put its rows in the MV
partition while no snapshot names
+ * it, leaving them there through a later drop of it. Subtracting the
partitions the mapping does not name
+ * is what keeps the read to the record; the partitions it does name are
added back by the first part
+ * whether or not their keys fall in the range, the way every other scoped
read pins them.
+ */
+ private static Set<Expression>
constructPredicatesOfADefaultPartitionTable(Set<PartitionItem> mvItems,
+ Set<PartitionItem> mappedItems, Set<String> readable, OlapTable
baseTable, String colName)
+ throws AnalysisException {
+ List<Column> partitionColumns = baseTable.getPartitionColumns();
+ List<Type> partitionColumnTypes = Lists.transform(partitionColumns,
Column::getType);
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(new UnboundSlot(partitionColumn.getName()));
+ }
+ Set<PartitionItem> notMapped = Sets.newHashSet();
+ for (String partitionName : baseTable.getPartitionNames()) {
+ if (readable.contains(partitionName)) {
+ continue;
+ }
+ PartitionItem item =
baseTable.getPartitionItemOrAnalysisException(partitionName);
+ if (!item.isDefaultPartition()) {
+ notMapped.add(item);
+ }
+ }
+ Expression inRange = ExpressionUtils.or(constructPredicates(mvItems,
colName,
+ Optional.of(partitionColumnType(baseTable, colName))));
+ if (!notMapped.isEmpty()) {
+ inRange = ExpressionUtils.and(inRange, new Not(ExpressionUtils.or(
Review Comment:
[P1] Keep default rows that predate a new explicit key. A LIST(d,k) table
may already hold (2020,1) in p_default when ADD PARTITION p_expired VALUES IN
((2020,1)) runs; ADD creates an empty partition and does not migrate that row.
With p_kept containing (2020,2) and a two-year sync window excluding p_expired,
this NOT(<=>) removes the old row from the mapped default scan even though
p_default is recorded in the MV snapshot. A COMPLETE refresh then silently
omits committed base data from the MV and considers it synchronized. Scope the
exclusion by physical partition or account for rows already in default; cover
insert-before-ADD.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java:
##########
@@ -1619,6 +1621,58 @@ public Map<String, Map<MTMVRelatedTableIf, Set<String>>>
calculatePartitionMappi
return res;
}
+ /**
+ * The list partition each base table of this MV has that takes the rows
no other partition of it claims,
+ * by table, or none for a table that has no such partition.
+ *
+ * <p>Read once per mapping rather than per MV partition: the mapping
describes every MV partition and the
+ * answer is the table's, not the partition's. The partition metadata is
read without a lock, like the
+ * rest of the mapping this is part of.
+ */
+ private Map<MTMVRelatedTableIf, String> defaultListPartitionsOf() throws
AnalysisException {
+ Map<MTMVRelatedTableIf, String> res = Maps.newHashMap();
+ for (MTMVRelatedTableIf pctTable : mvPartitionInfo.getPctTables()) {
+ if (!(pctTable instanceof OlapTable)) {
+ continue;
+ }
+ OlapTable olapTable = (OlapTable) pctTable;
+ if (!(olapTable.getPartitionInfo() instanceof ListPartitionInfo)) {
+ continue;
+ }
+ for (String partitionName : olapTable.getPartitionNames()) {
Review Comment:
[P2] Protect the default LIST metadata walk used by EXPLAIN. EXPLAIN REFRESH
COMPLETE calls MTMVRefreshContext.buildContext before ExplainCommand invokes
planner.plan and takes table read locks. getAndCopyPartitionItems() releases
its own lock before this new getPartitionNames()/item walk, while ADD/DROP
modifies OlapTable's TreeMap under the write lock. DDL can cause a
ConcurrentModificationException, a missing-name AnalysisException, or a mapping
from two metadata states. Derive the default name from the locked
partition-item copy or hold a read lock across this walk.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -160,30 +218,218 @@ public static Set<Expression>
constructPredicates(Set<PartitionItem> partitions,
*/
@VisibleForTesting
public static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot) {
+ return constructPredicates(partitions, colSlot, Optional.empty());
+ }
+
+ private static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot,
+ Optional<Type> columnType) {
Set<Expression> predicates = new HashSet<>();
if (partitions.isEmpty()) {
return Sets.newHashSet(BooleanLiteral.TRUE);
}
if (partitions.iterator().next() instanceof ListPartitionItem) {
for (PartitionItem item : partitions) {
- predicates.add(convertListPartitionToIn(item, colSlot));
+ predicates.add(convertListPartitionToIn(item, colSlot,
columnType));
}
} else {
for (PartitionItem item : partitions) {
- predicates.add(convertRangePartitionToCompare(item, colSlot));
+ predicates.add(convertRangePartitionToCompare(item, colSlot,
columnType));
+ }
+ }
+ return predicates;
+ }
+
+ /**
+ * 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) {
+ return constructPredicatesOfBasePartitions(partitions, baseTable,
colName, UnboundSlot::new);
+ }
+
+ /**
+ * The same, with the partition columns read through the slots the caller
names them by. A caller that
+ * writes these predicates into a plan that is already bound and is never
bound again -- the union
+ * compensation -- has to pass that plan's own slots: an unbound slot
there fails the rewrite instead of
+ * narrowing the read.
+ */
+ private static Set<Expression>
constructPredicatesOfBasePartitions(Set<PartitionItem> partitions,
+ OlapTable baseTable, String colName, Function<String, Slot>
slotOfColumn) {
+ 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,
slotOfColumn.apply(colName),
+ Optional.of(partitionColumnTypes.get(0))));
}
+ return predicates;
+ }
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(slotOfColumn.apply(partitionColumn.getName()));
+ }
+ Set<Expression> predicates = new HashSet<>();
+ for (PartitionItem item : partitions) {
+ predicates.add(convertListPartitionToKey(item, partitionSlots,
partitionColumnTypes));
}
return predicates;
}
- private static Expression convertPartitionKeyToLiteral(PartitionKey key) {
- return Literal.fromLegacyLiteral(key.getKeys().get(0),
- Type.fromPrimitiveType(key.getTypes().get(0)));
+ /**
+ * The predicate for a table whose mapped partitions include a list
partition's default one.
+ *
+ * <p>That partition takes the rows no other partition of the table
claims, and an MV partition takes the
+ * rows whose own key falls in it wherever the base table placed them --
so those rows are the ones the MV
+ * partition's own key range names, and no predicate on the partition
columns picks them out on its own: a
+ * test on those columns reaches the rows an explicit partition holds as
readily as it reaches theirs.
+ * The read is therefore written the other way round, as the record's own
partitions plus what the key
+ * range holds beyond them:
+ *
+ * <ul>
+ * <li>the partitions this MV partition is recorded with, pinned by
their whole key, and</li>
+ * <li>the rows of the MV partitions' key range that no explicit
partition the mapping does not name
+ * holds, which is the default partition's rows in that range and
nothing else.</li>
+ * </ul>
+ *
+ * <p>An explicit partition the window left out shares the MV partition's
key range with a retained one --
+ * the shape this scope is about, a list partitioned table's partitions
holding several keys of the MV's
+ * partition column -- and reading it back would put its rows in the MV
partition while no snapshot names
+ * it, leaving them there through a later drop of it. Subtracting the
partitions the mapping does not name
+ * is what keeps the read to the record; the partitions it does name are
added back by the first part
+ * whether or not their keys fall in the range, the way every other scoped
read pins them.
+ */
+ private static Set<Expression>
constructPredicatesOfADefaultPartitionTable(Set<PartitionItem> mvItems,
+ Set<PartitionItem> mappedItems, Set<String> readable, OlapTable
baseTable, String colName)
+ throws AnalysisException {
+ List<Column> partitionColumns = baseTable.getPartitionColumns();
+ List<Type> partitionColumnTypes = Lists.transform(partitionColumns,
Column::getType);
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(new UnboundSlot(partitionColumn.getName()));
+ }
+ Set<PartitionItem> notMapped = Sets.newHashSet();
+ for (String partitionName : baseTable.getPartitionNames()) {
Review Comment:
[P2] Use the mapping's locked partition metadata to build exclusions.
MTMVTask.buildRefreshContext releases base-table read locks before this
getPartitionNames()/item lookup; concurrent ADD/DROP can abort refresh with
ConcurrentModificationException or AnalysisException, outside the task's
RPC/cloud retry. An ADD for a key in this MV range after the exclusion list is
built but before planner locking can also let the scan read new-partition rows
absent from the recorded mapping and snapshot. The earlier mapped-item lookup
in from() has the same post-lock race for any scoped OLAP table. Carry one
stable partition-item snapshot into both predicate paths.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -160,30 +218,218 @@ public static Set<Expression>
constructPredicates(Set<PartitionItem> partitions,
*/
@VisibleForTesting
public static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot) {
+ return constructPredicates(partitions, colSlot, Optional.empty());
+ }
+
+ private static Set<Expression> constructPredicates(Set<PartitionItem>
partitions, Slot colSlot,
+ Optional<Type> columnType) {
Set<Expression> predicates = new HashSet<>();
if (partitions.isEmpty()) {
return Sets.newHashSet(BooleanLiteral.TRUE);
}
if (partitions.iterator().next() instanceof ListPartitionItem) {
for (PartitionItem item : partitions) {
- predicates.add(convertListPartitionToIn(item, colSlot));
+ predicates.add(convertListPartitionToIn(item, colSlot,
columnType));
}
} else {
for (PartitionItem item : partitions) {
- predicates.add(convertRangePartitionToCompare(item, colSlot));
+ predicates.add(convertRangePartitionToCompare(item, colSlot,
columnType));
+ }
+ }
+ return predicates;
+ }
+
+ /**
+ * 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) {
+ return constructPredicatesOfBasePartitions(partitions, baseTable,
colName, UnboundSlot::new);
+ }
+
+ /**
+ * The same, with the partition columns read through the slots the caller
names them by. A caller that
+ * writes these predicates into a plan that is already bound and is never
bound again -- the union
+ * compensation -- has to pass that plan's own slots: an unbound slot
there fails the rewrite instead of
+ * narrowing the read.
+ */
+ private static Set<Expression>
constructPredicatesOfBasePartitions(Set<PartitionItem> partitions,
+ OlapTable baseTable, String colName, Function<String, Slot>
slotOfColumn) {
+ 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,
slotOfColumn.apply(colName),
+ Optional.of(partitionColumnTypes.get(0))));
}
+ return predicates;
+ }
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(slotOfColumn.apply(partitionColumn.getName()));
+ }
+ Set<Expression> predicates = new HashSet<>();
+ for (PartitionItem item : partitions) {
+ predicates.add(convertListPartitionToKey(item, partitionSlots,
partitionColumnTypes));
}
return predicates;
}
- private static Expression convertPartitionKeyToLiteral(PartitionKey key) {
- return Literal.fromLegacyLiteral(key.getKeys().get(0),
- Type.fromPrimitiveType(key.getTypes().get(0)));
+ /**
+ * The predicate for a table whose mapped partitions include a list
partition's default one.
+ *
+ * <p>That partition takes the rows no other partition of the table
claims, and an MV partition takes the
+ * rows whose own key falls in it wherever the base table placed them --
so those rows are the ones the MV
+ * partition's own key range names, and no predicate on the partition
columns picks them out on its own: a
+ * test on those columns reaches the rows an explicit partition holds as
readily as it reaches theirs.
+ * The read is therefore written the other way round, as the record's own
partitions plus what the key
+ * range holds beyond them:
+ *
+ * <ul>
+ * <li>the partitions this MV partition is recorded with, pinned by
their whole key, and</li>
+ * <li>the rows of the MV partitions' key range that no explicit
partition the mapping does not name
+ * holds, which is the default partition's rows in that range and
nothing else.</li>
+ * </ul>
+ *
+ * <p>An explicit partition the window left out shares the MV partition's
key range with a retained one --
+ * the shape this scope is about, a list partitioned table's partitions
holding several keys of the MV's
+ * partition column -- and reading it back would put its rows in the MV
partition while no snapshot names
+ * it, leaving them there through a later drop of it. Subtracting the
partitions the mapping does not name
+ * is what keeps the read to the record; the partitions it does name are
added back by the first part
+ * whether or not their keys fall in the range, the way every other scoped
read pins them.
+ */
+ private static Set<Expression>
constructPredicatesOfADefaultPartitionTable(Set<PartitionItem> mvItems,
+ Set<PartitionItem> mappedItems, Set<String> readable, OlapTable
baseTable, String colName)
+ throws AnalysisException {
+ List<Column> partitionColumns = baseTable.getPartitionColumns();
+ List<Type> partitionColumnTypes = Lists.transform(partitionColumns,
Column::getType);
+ List<Slot> partitionSlots = Lists.newArrayList();
+ for (Column partitionColumn : partitionColumns) {
+ partitionSlots.add(new UnboundSlot(partitionColumn.getName()));
+ }
+ Set<PartitionItem> notMapped = Sets.newHashSet();
Review Comment:
[P2] Limit default-partition exclusions to keys that can overlap this MV
batch. With N distinct explicit LIST keys plus p_default, a one-partition batch
maps one explicit key, but this loop turns the other N-1 keys into a
NOT(OR(...)) predicate for every batch. COMPLETE refresh therefore builds and
plans about N(N-1) tuple terms, even when those other keys cannot match this
batch's MV range. Prune exclusions by projected MV key or use physical
partition scope so a many-partition MV does not incur quadratic FE planning
work.
--
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]
