github-actions[bot] commented on code in PR #68648:
URL: https://github.com/apache/doris/pull/68648#discussion_r4143622789
##########
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()) {
+ if
(olapTable.getPartitionItemOrAnalysisException(partitionName).isDefaultPartition())
{
+ res.put(pctTable, partitionName);
+ break;
+ }
+ }
+ }
+ return res;
+ }
+
+ /**
+ * One MV partition's mapping, with every base table's default list
partition named in it.
+ *
+ * <p>Such a partition holds rows for every key its table can be read by,
so it belongs to every MV
+ * partition that reads the table -- not only to the one its own key, the
sentinel those rows were placed
+ * by, maps to. Naming it everywhere is what the read and the record have
to agree on: the refresh reads
+ * the rows of it that belong to the MV partition being refreshed, and the
partition is recorded among the
+ * ones that partition is read through, so an insert into it leaves that
MV partition out of sync instead
+ * of changing nothing the MV compares.
+ */
+ private Map<MTMVRelatedTableIf, Set<String>> withDefaultListPartitions(
+ Map<MTMVRelatedTableIf, Set<String>> mapping,
Map<MTMVRelatedTableIf, String> defaultListPartitions) {
+ if (defaultListPartitions.isEmpty()) {
+ return mapping;
+ }
+ Map<MTMVRelatedTableIf, Set<String>> res = Maps.newHashMap(mapping);
+ for (Entry<MTMVRelatedTableIf, String> entry :
defaultListPartitions.entrySet()) {
+ Set<String> partitions =
Sets.newHashSet(res.getOrDefault(entry.getKey(), Sets.newHashSet()));
+ partitions.add(entry.getValue());
Review Comment:
[P1] Cover default LIST rows during base-side union compensation. In a
two-table join MV partitioned by left keys 1 and 3, let the right table's
default partition hold rows for both keys. After a key-3 insert, refresh only
MV key1, then query both keys with union rewrite enabled. This fan-out makes
key1 valid and key3 stale, so compensation removes MV key3 and unions
`r_default`; `PredicateAdder` filters that base scan with the default item's
synthetic `MIN` key (`right.k IN (MIN)`), making the key-3 join branch empty.
This is distinct from the earlier direct-rewrite inverse issue. Filter default
rows by the stale MV key range or reject this partial union rewrite.
##########
fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRewriteUtil.java:
##########
@@ -166,25 +166,35 @@ private static Set<String>
getMtmvPartitionsByRelatedPartitions(MTMV mtmv, MTMVR
}
Set<String> pctPartitions = entry.getValue();
for (String pctPartition : pctPartitions) {
- String mvPartition = relatedToMv.get(Pair.of(tableIf,
pctPartition));
- if (mvPartition != null) {
- res.add(mvPartition);
+ Set<String> mvPartitions = relatedToMv.get(Pair.of(tableIf,
pctPartition));
+ if (mvPartitions != null) {
Review Comment:
[P1] Reject MV-only rewrites that leave an expired base partition uncovered.
With `partition_sync_limit=2 DAY`, the new scoped refresh omits p_expired but
retains p_kept in the same year MV partition. A query over both base partitions
still reaches that MV year through p_kept here while p_expired has no inverse
edge; with `enable_materialized_view_union_rewrite=false`,
`AbstractMaterializedViewRule` skips `calcInvalidPartitions` and rewrites to
the MV alone, dropping p_expired's committed rows. Require coverage of every
query-used base partition for an MV-only rewrite, or force base compensation
for the missing partitions.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +139,66 @@ 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;
+ }
+ 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. That read can be seen to
+ // be too wide -- it is the one this scope exists to narrow --
rather than one that drops
+ // rows belonging to the MV partition being refreshed. 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,
Review Comment:
[P1] Keep excluded explicit LIST rows out of the default fallback. With
`LIST(d,region)` partitions p_old=(2020,US), p_kept=(2020,EU),(2038,EU), plus
p_default, a 2 YEAR sync limit excludes p_old but retains p_kept. Fan-out adds
p_default to the MV partition mapped from p_kept, so this branch falls back to
its projected `d IN (2020,2038)` and rereads p_old even though its snapshot
names only p_kept and p_default. Dropping p_old then leaves its US row in an MV
partition still judged synchronized. This is the default branch bypassing the
full-tuple fix from the earlier thread; preserve exact retained-partition reads
while covering default rows, and test the combined shape.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -160,30 +223,121 @@ 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());
Review Comment:
[P1] Preserve DATETIME(3) scale in union compensation. With the added
`fractional_key` shape, insert a new `.123` row into p_fraction after refresh
while p_whole remains valid, then query both timestamps with union rewrite
enabled. Compensation removes stale MV p_fraction and reads base p_fraction
through `PredicateAdder`, which calls this overload and passes
`Optional.empty()`; `Type.fromPrimitiveType(DATETIMEV2)` then makes the key
`.000`, so its `ts IN` predicate misses the committed `.123` row. The earlier
scale fixes cover refresh predicates, but this separate base-side union path
still needs the column's full Type.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/UpdateMvByPartitionCommand.java:
##########
@@ -130,17 +139,66 @@ 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;
+ }
+ 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. That read can be seen to
+ // be too wide -- it is the one this scope exists to narrow --
rather than one that drops
+ // rows belonging to the MV partition being refreshed. 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;
+ }
+ builder.put(table, constructPredicatesOfBasePartitions(items,
olapTable, colName));
Review Comment:
[P1] Use full LIST tuples for base-side union compensation too. In the added
`list_scope` shape, p_old=(2020,US) is expired while p_kept=(2020,EU),(2038,EU)
remains in the MV. For a grouped query with `d=2020` and union rewrite enabled,
compensation selects p_old, but `PredicateAdder` still builds only `d IN
(2020)` for the base branch. That rereads p_kept's EU row, which the valid MV
branch already supplies, so `UNION ALL` duplicates it. The new full-tuple
refresh predicate here does not reach this parallel compensation path; scope
that path by the full tuple or exact base partition identity 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]
