This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7bd89a0795c [fix](ivm) Stop the incremental delta from reading
partitions the MV dropped (#67814)
7bd89a0795c is described below
commit 7bd89a0795ce7aa43d715fa57a591d0a7f4f8035
Author: yujun <[email protected]>
AuthorDate: Fri Sep 11 17:22:28 2026 +0800
[fix](ivm) Stop the incremental delta from reading partitions the MV
dropped (#67814)
`partition_sync_limit` keeps only a recent slice of each base table's
partitions, so an MV can be missing a partition its base table still
has. The incremental delta read that base partition anyway, through the
delta scan and through the join-opposite snapshot. A change to a
non-partitioned dimension therefore produced delta rows for dates the MV
has no partition for, and the insert failed with `no partition for this
tuple`.
Re-syncing cannot bring an expired partition back, so the retry loop in
`executeIvmAttempt` could never recover: the task failed after
exhausting its attempts, and because the write is atomic, the partitions
the MV does keep were left unrepaired as well. A strict `REFRESH ...
INCREMENTAL`, and a scheduled refresh of an MV declared without
`FALLBACK`, fail outright.
### What changed
The incremental delta now reads only the base partitions the MV's
partition definition keeps.
- `MTMVPartitionUtil.generateRelatedBasePartitionIds` returns that
partition set per base table partitioned by the MV's partition column,
and an empty value when `partition_sync_limit` is not set, which is the
only property that can leave the MV without a base partition.
- `IvmIncrRefreshManager` passes it to `IvmRewriteContext.incremental`,
and `IvmDeltaRewriter` applies it as an upper bound on the partitions
each base table may be read from, intersected with
`ivm_partition_window_limit` when that is set too.
- The bound is the partition set the MV is aligned to rather than the
partitions it already has, so a base partition whose MV partition has
not been added yet stays readable: the refresh still reports the missing
partition and recovers it by syncing, which is what keeps a newly added
base partition working.
- A base table absent from the set keeps its full read, since the MV
partition column does not come from it and limiting it would change join
results without narrowing the target MV partitions. A set covering every
partition of its table leaves the plan untouched, so an MV that mirrors
all of its base partitions is unaffected.
- Only olap tables are reachable: the delta reads them through their
stream and restricts a scan by partition id, which a connector table has
no equivalent of.
The COMPLETE paths are unchanged.
The PR also removes the `nonConcurrent` group from seven `mtmv_p0/ivm`
suites that use neither debug points nor global variables or config, so
they run in the parallel pool again.
Trace issue: https://github.com/apache/doris/issues/65418
---
.../org/apache/doris/mtmv/MTMVPartitionUtil.java | 57 ++++++++
.../org/apache/doris/mtmv/MTMVPropertyUtil.java | 13 ++
.../apache/doris/mtmv/ivm/IvmDeltaRewriter.java | 64 ++++++++-
.../doris/mtmv/ivm/IvmIncrRefreshManager.java | 16 ++-
.../apache/doris/mtmv/ivm/IvmRewriteContext.java | 51 +++++--
.../apache/doris/mtmv/MTMVPartitionUtilTest.java | 23 +++
.../apache/doris/mtmv/MTMVPropertyUtilTest.java | 13 ++
.../doris/mtmv/ivm/IvmDeltaRewriterTest.java | 73 ++++++++++
.../apache/doris/mtmv/ivm/IvmDeltaTestBase.java | 21 ++-
.../mtmv_p0/ivm/test_ivm_partition_sync_limit.out | 14 ++
.../test_ivm_partition_sync_limit_with_window.out | 15 ++
.../ivm/test_ivm_bitmap_runtime_fallback.groovy | 2 +-
.../test_ivm_drop_column_fallback_reason.groovy | 2 +-
.../ivm/test_ivm_minmax_runtime_fallback.groovy | 2 +-
.../mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy | 2 +-
.../ivm/test_ivm_partition_baseline_rebuild.groovy | 2 +-
.../ivm/test_ivm_partition_sync_limit.groovy | 159 +++++++++++++++++++++
...est_ivm_partition_sync_limit_with_window.groovy | 157 ++++++++++++++++++++
.../mtmv_p0/ivm/test_ivm_rewrite_projection.groovy | 2 +-
...t_ivm_strict_failure_partition_atomicity.groovy | 2 +-
20 files changed, 670 insertions(+), 20 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
index 7a80da85109..d0bbc341aa5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPartitionUtil.java
@@ -237,6 +237,63 @@ public class MTMVPartitionUtil {
return result.getRes();
}
+ /**
+ * Base partitions the MV's partition definition keeps, per base table
partitioned by the MV's
+ * partition column. This is the partition set the MV is aligned to, so it
still contains a base
+ * partition whose MV partition the next {@link #alignMvPartition} has yet
to add, and no longer
+ * contains one the partition properties filter out, such as an expired
partition_sync_limit.
+ *
+ * <p>Empty when no restriction applies at all: without {@code
partition_sync_limit} the MV
+ * mirrors every base partition, so there is nothing to restrict. When
present, the map holds an
+ * entry for every base table the MV partition column comes from, each
with the partitions that
+ * table may be read from — an empty set meaning that table may not be
read at all. Keeping the
+ * empty set explicit is what separates "this table is restricted to
nothing" from "this table
+ * is not one of the MV partition column's sources".
+ *
+ * <p>Read without pinned snapshots, like {@link #alignMvPartition}. A
base partition that
+ * partition sync adds or drops at the same moment can therefore sit
outside the returned set
+ * for one refresh; the same refresh either recovers the partition by
syncing again or leaves
+ * its binlog pending for the next one.
+ *
+ * @return baseTableInfo ==> base partition ids, empty when no restriction
applies
+ */
+ public static Optional<Map<BaseTableInfo, Set<Long>>>
generateRelatedBasePartitionIds(MTMV mtmv)
+ throws AnalysisException {
+ MTMVPartitionInfo mvPartitionInfo = mtmv.getMvPartitionInfo();
+ if (mvPartitionInfo == null
+ || mvPartitionInfo.getPartitionType() ==
MTMVPartitionType.SELF_MANAGE
+ ||
!MTMVPropertyUtil.hasPartitionSyncLimit(mtmv.getMvProperties())) {
+ return Optional.empty();
+ }
+ Map<BaseTableInfo, Set<Long>> res = Maps.newHashMap();
+ // Only olap tables are restricted: the delta rewrite reads them
through their stream and
+ // selects partitions by id, which an external table has no equivalent
of. A connector
+ // table therefore stays out of the scope and keeps its full read, so
supporting one as an
+ // IVM base table means giving the scope a partition identity it can
express, not just
+ // widening these types.
+ for (MTMVRelatedTableIf pctTable : mvPartitionInfo.getPctTables()) {
+ if (pctTable instanceof OlapTable) {
+ res.put(new BaseTableInfo((OlapTable) pctTable),
Sets.newHashSet());
+ }
+ }
+ Map<PartitionKeyDesc, Map<MTMVRelatedTableIf, Set<String>>>
relatedDescs = generateRelatedPartitionDescs(
+ mvPartitionInfo, mtmv.getMvProperties(),
mtmv.getPartitionColumns(), Maps.newHashMap());
+ for (Map<MTMVRelatedTableIf, Set<String>> relatedPartitions :
relatedDescs.values()) {
+ for (Entry<MTMVRelatedTableIf, Set<String>> entry :
relatedPartitions.entrySet()) {
+ if (!(entry.getKey() instanceof OlapTable)) {
+ continue;
+ }
+ OlapTable baseTable = (OlapTable) entry.getKey();
+ Set<Long> partitionIds = res.computeIfAbsent(
+ new BaseTableInfo(baseTable), key ->
Sets.newHashSet());
+ for (String partitionName : entry.getValue()) {
+
partitionIds.add(baseTable.getPartitionOrAnalysisException(partitionName).getId());
+ }
+ }
+ }
+ return Optional.of(res);
+ }
+
/**
* check if table is sync with all baseTables
*
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
index d1b8a8765b1..ebd4a6f166d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPropertyUtil.java
@@ -218,6 +218,19 @@ public class MTMVPropertyUtil {
PropertyAnalyzer.PROPERTIES_IVM_PARTITION_WINDOW_LIMIT));
}
+ /**
+ * True when the MV keeps only a recent slice of each base table's
partitions. This is the only
+ * property that can leave the MV without a partition the base table still
has, so it is also
+ * the only case where the incremental delta has to be told which base
partitions it may read.
+ */
+ public static boolean hasPartitionSyncLimit(Map<String, String>
mvProperties) {
+ if (mvProperties == null) {
+ return false;
+ }
+ String value =
mvProperties.get(PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT);
+ return !StringUtils.isEmpty(value) && Integer.parseInt(value) > 0;
+ }
+
/**
* Look up the window limit configured for a base table, mirroring the
* excluded_trigger_tables name-matching semantics (empty db/ctl wildcard).
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
index a6894a64149..1d97e87796c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriter.java
@@ -26,6 +26,7 @@ import org.apache.doris.catalog.stream.OlapTableStream;
import org.apache.doris.common.Config;
import org.apache.doris.common.Pair;
import org.apache.doris.info.TableNameInfoUtils;
+import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.mtmv.MTMVPartitionUtil;
import org.apache.doris.mtmv.MTMVPropertyUtil;
import org.apache.doris.nereids.trees.expressions.Slot;
@@ -37,6 +38,7 @@ import
org.apache.doris.nereids.trees.plans.logical.LogicalProject;
import org.apache.doris.nereids.types.DataType;
import org.apache.doris.qe.ConnectContext;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -66,7 +68,8 @@ public class IvmDeltaRewriter {
Pair<Plan, List<LogicalProject<?>>> prefixChain =
helper.detachAdaptProjectChain(sinkChild);
Plan rootPlan = prefixChain.first;
long refreshVersion = refreshContext.getMtmv().getNextRefreshVersion();
- IvmDeltaRewriteState rewriteState = createDeltaRewriteState(rootPlan,
refreshContext, refreshVersion);
+ IvmDeltaRewriteState rewriteState = createDeltaRewriteState(rootPlan,
refreshContext, refreshVersion,
+ rewriteContext.getIncrementalScopePartitionIds());
Optional<IvmDeltaRewriteResult> deltaResult = rewriteDelta(rootPlan,
refreshContext, rewriteState);
if (!deltaResult.isPresent()) {
return new LogicalEmptyRelation(
@@ -153,7 +156,8 @@ public class IvmDeltaRewriter {
return IvmDeltaRewriteHelper.INSTANCE.freshPlan(rewritten);
}
- private IvmDeltaRewriteState createDeltaRewriteState(Plan plan,
IvmIncrRefreshContext ctx, long refreshVersion) {
+ private IvmDeltaRewriteState createDeltaRewriteState(Plan plan,
IvmIncrRefreshContext ctx, long refreshVersion,
+ Map<BaseTableInfo, Set<Long>> scopePartitionIds) {
Map<OlapTable, OlapTableStream> streams = new HashMap<>();
// Window limits apply to every base table in the plan, including
excluded
// trigger tables (their snapshot side is windowed too, so the
property saves
@@ -181,11 +185,67 @@ public class IvmDeltaRewriter {
new TableNameInfo(table.getFullQualifiers()),
windowLimits));
}
}
+ applyScopePartitionIds(windowPartitionIdsByTable, planTables,
scopePartitionIds);
return new IvmDeltaRewriteState(streams,
ctx.isIncludeExhaustedStreams(), refreshVersion,
DataType.fromCatalogType(ctx.getMtmv().getColumn(Column.SEQUENCE_COL).getType()),
windowPartitionIdsByTable);
}
+ /**
+ * Limits every base table whose partition column feeds the MV's partition
column to the base
+ * partitions the MV's partition definition keeps. The delta and the
join-opposite snapshot
+ * would otherwise also read base partitions the MV does not keep, expired
by
+ * partition_sync_limit, and then try to write their rows into MV
partitions that do not exist,
+ * which fails the whole insert with "no partition for this tuple" and
takes the partitions
+ * that do exist down with it.
+ *
+ * <p>The limit is the partition set the MV is aligned to rather than the
partitions it already
+ * has, so a base partition whose MV partition has not been added yet
stays readable: the
+ * refresh still reports the missing partition and recovers it by syncing.
+ *
+ * <p>Tables outside {@code scopePartitionIds} keep their full read: the
MV partition column
+ * does not come from them, so limiting them would change join results
without narrowing the
+ * set of MV partitions the delta can target. A scope that covers every
partition of its table
+ * is left alone as well, so an MV that mirrors all of its base partitions
keeps the plan it
+ * had before this restriction existed. An entry with no partitions is the
opposite case and is
+ * applied as it stands: the scan then reads nothing, which is how the
delta and the snapshot
+ * side both stay out of a table the MV has no partition for.
+ *
+ * <p>The scope is keyed by base table identity, but only olap tables are
reachable here: the
+ * delta reads them through {@code OlapTableStream} and restricts a scan
by partition id.
+ */
+ private static void applyScopePartitionIds(Map<OlapTable, List<Long>>
windowPartitionIdsByTable,
+ Set<OlapTable> planTables, Map<BaseTableInfo, Set<Long>>
scopePartitionIds) {
+ if (scopePartitionIds.isEmpty()) {
+ return;
+ }
+ for (OlapTable table : planTables) {
+ Set<Long> tableScope = scopePartitionIds.get(new
BaseTableInfo(table));
+ if (tableScope == null ||
tableScope.containsAll(table.getPartitionIds())) {
+ continue;
+ }
+ List<Long> windowPartitionIds =
windowPartitionIdsByTable.get(table);
+ List<Long> readablePartitionIds;
+ if (windowPartitionIds == null) {
+ readablePartitionIds = new ArrayList<>(tableScope);
+ } else {
+ // Both limits apply, so only their intersection is readable.
One keeps the
+ // partitions above a time cutoff and the other the last N by
partition value, so
+ // each is a suffix of the same value order and the two cannot
be disjoint.
+ readablePartitionIds = new ArrayList<>();
+ for (Long partitionId : windowPartitionIds) {
+ if (tableScope.contains(partitionId)) {
+ readablePartitionIds.add(partitionId);
+ }
+ }
+ }
+ // Sorted like every other partition selection handed to a scan,
so the plan shape does
+ // not depend on the order the two limits were combined in.
+ readablePartitionIds.sort(Long::compareTo);
+ windowPartitionIdsByTable.put(table, readablePartitionIds);
+ }
+ }
+
boolean isExcludedTriggerTable(LogicalOlapScan scan, Set<TableNameInfo>
excludedTriggerTables) {
if (excludedTriggerTables == null || excludedTriggerTables.isEmpty()) {
return false;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
index 886d6ee6c23..4559fd54937 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManager.java
@@ -22,6 +22,8 @@ import org.apache.doris.common.util.DebugPointUtil;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.common.util.Util;
import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.mtmv.BaseTableInfo;
+import org.apache.doris.mtmv.MTMVPartitionUtil;
import org.apache.doris.mtmv.MTMVPlanUtil;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundTableSink;
@@ -39,9 +41,12 @@ import com.google.common.collect.ImmutableList;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
/**
* Minimal orchestration entry point for incremental refresh.
@@ -92,7 +97,16 @@ public class IvmIncrRefreshManager {
MTMV mtmv = context.getMtmv();
StatementContext statementContext = new StatementContext(
context.getConnectContext(), new
OriginStatement(mtmv.getQuerySql(), 0));
-
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.incremental(mtmv)));
+ // The delta may only read the base partitions the MV's partition
definition keeps. A base
+ // partition outside that set, expired by partition_sync_limit, would
otherwise still be
+ // read through the delta and the join-opposite snapshot, and its rows
would have no MV
+ // partition to land in. A base partition inside the set stays
readable even when its MV
+ // partition is not there yet, so that the refresh still reports the
missing partition and
+ // recovers it by syncing.
+ Map<BaseTableInfo, Set<Long>> scopePartitionIds =
+
MTMVPartitionUtil.generateRelatedBasePartitionIds(mtmv).orElse(Collections.emptyMap());
+ statementContext.setIvmRewriteContext(
+ Optional.of(IvmRewriteContext.incremental(mtmv,
scopePartitionIds)));
// Excluded trigger tables do not produce delta and must not be
validated for
// binlog / key-type support during the incremental analyze.
statementContext.setExcludedTriggerTables(mtmv.getExcludedTriggerTables());
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRewriteContext.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRewriteContext.java
index 414a1df3d51..a6fefda42c9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRewriteContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmRewriteContext.java
@@ -76,6 +76,11 @@ public class IvmRewriteContext {
private final Optional<IvmDryRunLimit> dryRunLimit;
private final Map<BaseTableInfo, Set<Long>> fullRefreshResetPartitionIds;
private final Optional<StreamReadMode> fullRefreshNonPctReadMode;
+ // Base partitions the incremental delta may read, per base table. A table
that is absent is
+ // read in full, so an empty map keeps the unrestricted behaviour. Unlike
+ // fullRefreshResetPartitionIds, which names the partitions a COMPLETE
refresh must reset,
+ // every entry here is an upper bound on what the delta is allowed to read.
+ private final Map<BaseTableInfo, Set<Long>> incrementalScopePartitionIds;
// Set by MTMVPlanUtil before normalization: true means the MV unique keys
include identity key columns.
// Null when the rewrite context is created outside the analyzeQuery flow.
private Boolean useFullKeys;
@@ -83,7 +88,8 @@ public class IvmRewriteContext {
private IvmRewriteContext(Mode mode, MTMV mtmv, String createMtmvName,
boolean includeExhaustedStreams,
ExecutionKind executionKind, Optional<IvmDryRunLimit> dryRunLimit,
Map<BaseTableInfo, Set<Long>> fullRefreshResetPartitionIds,
- Optional<StreamReadMode> fullRefreshNonPctReadMode) {
+ Optional<StreamReadMode> fullRefreshNonPctReadMode,
+ Map<BaseTableInfo, Set<Long>> incrementalScopePartitionIds) {
this.mode = Objects.requireNonNull(mode, "mode can not be null");
this.mtmv = mode == Mode.CREATE ? mtmv : Objects.requireNonNull(mtmv,
"mtmv can not be null");
this.createMtmvName = createMtmvName;
@@ -97,45 +103,66 @@ public class IvmRewriteContext {
this.fullRefreshResetPartitionIds =
Collections.unmodifiableMap(resetPartitionIds);
this.fullRefreshNonPctReadMode = Objects.requireNonNull(
fullRefreshNonPctReadMode, "fullRefreshNonPctReadMode can not
be null");
+ Map<BaseTableInfo, Set<Long>> scopePartitionIds = new HashMap<>();
+ Objects.requireNonNull(incrementalScopePartitionIds,
"incrementalScopePartitionIds can not be null")
+ .forEach((baseTableInfo, partitionIds) ->
scopePartitionIds.put(baseTableInfo,
+ Collections.unmodifiableSet(new
HashSet<>(partitionIds))));
+ this.incrementalScopePartitionIds =
Collections.unmodifiableMap(scopePartitionIds);
}
public static IvmRewriteContext create(String mtmvName) {
return new IvmRewriteContext(Mode.CREATE, null,
Objects.requireNonNull(mtmvName, "mtmvName can not be null"),
false,
- ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty(),
+ Collections.emptyMap());
}
public static IvmRewriteContext normalize(MTMV mtmv) {
return new IvmRewriteContext(Mode.NORMALIZE,
Objects.requireNonNull(mtmv, "mtmv can not be null"),
- null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty(),
+ Collections.emptyMap());
}
public static IvmRewriteContext incremental(MTMV mtmv) {
+ return incremental(mtmv, Collections.emptyMap());
+ }
+
+ /**
+ * Incremental refresh whose delta may only read the given base
partitions. A base partition
+ * outside the scope is read neither through its stream nor through the
join-opposite
+ * snapshot, so a base partition the MV no longer mirrors cannot produce
delta rows that the
+ * MV has no target partition for.
+ */
+ public static IvmRewriteContext incremental(MTMV mtmv, Map<BaseTableInfo,
Set<Long>> scopePartitionIds) {
return new IvmRewriteContext(Mode.INCREMENTAL,
Objects.requireNonNull(mtmv, "mtmv can not be null"),
- null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty(),
+ scopePartitionIds);
}
/** EXPLAIN REFRESH INCREMENTAL [ALL]: only a plan is produced. */
public static IvmRewriteContext incrementalExplain(MTMV mtmv, boolean
includeExhaustedStreams) {
return new IvmRewriteContext(Mode.INCREMENTAL,
Objects.requireNonNull(mtmv, "mtmv can not be null"),
null, includeExhaustedStreams, ExecutionKind.EXPLAIN,
- Optional.empty(), Collections.emptyMap(), Optional.empty());
+ Optional.empty(), Collections.emptyMap(), Optional.empty(),
Collections.emptyMap());
}
public static IvmRewriteContext incrementalDryRun(MTMV mtmv,
Optional<IvmDryRunLimit> dryRunLimit) {
return new IvmRewriteContext(Mode.INCREMENTAL, mtmv, null, false,
- ExecutionKind.DRY_RUN, dryRunLimit, Collections.emptyMap(),
Optional.empty());
+ ExecutionKind.DRY_RUN, dryRunLimit, Collections.emptyMap(),
Optional.empty(),
+ Collections.emptyMap());
}
/** EXPLAIN REFRESH COMPLETE: only a plan is produced. */
public static IvmRewriteContext fullExplain(MTMV mtmv) {
return new IvmRewriteContext(Mode.FULL, Objects.requireNonNull(mtmv,
"mtmv can not be null"),
- null, false, ExecutionKind.EXPLAIN, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ null, false, ExecutionKind.EXPLAIN, Optional.empty(),
Collections.emptyMap(), Optional.empty(),
+ Collections.emptyMap());
}
public static IvmRewriteContext full(MTMV mtmv) {
return new IvmRewriteContext(Mode.FULL, Objects.requireNonNull(mtmv,
"mtmv can not be null"),
- null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty(),
+ Collections.emptyMap());
}
public static IvmRewriteContext full(MTMV mtmv,
@@ -143,7 +170,8 @@ public class IvmRewriteContext {
StreamReadMode nonPctReadMode) {
return new IvmRewriteContext(Mode.FULL, mtmv, null, false,
ExecutionKind.EXECUTE, Optional.empty(),
resetPartitionIds,
- Optional.of(Objects.requireNonNull(nonPctReadMode,
"nonPctReadMode can not be null")));
+ Optional.of(Objects.requireNonNull(nonPctReadMode,
"nonPctReadMode can not be null")),
+ Collections.emptyMap());
}
public Mode getMode() {
@@ -189,6 +217,11 @@ public class IvmRewriteContext {
return
Optional.ofNullable(fullRefreshResetPartitionIds.get(baseTableInfo)).map(HashSet::new);
}
+ /** Empty when the incremental delta is not limited to a partition subset.
*/
+ public Map<BaseTableInfo, Set<Long>> getIncrementalScopePartitionIds() {
+ return incrementalScopePartitionIds;
+ }
+
public Optional<StreamReadMode> getFullRefreshNonPctReadMode() {
return fullRefreshNonPctReadMode;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
index dad6c3b5b1c..a96caf01d94 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPartitionUtilTest.java
@@ -312,6 +312,29 @@ public class MTMVPartitionUtilTest {
Assertions.assertFalse(MTMVPartitionUtil.isTableNamelike(new
TableNameInfo("ctl1"), tableNameToCheck));
}
+ @Test
+ public void testGenerateRelatedBasePartitionIdsWithoutSyncLimit() throws
AnalysisException {
+ // Without partition_sync_limit the MV mirrors every base partition,
so the incremental
+ // delta has nothing to restrict and must be left alone.
+
Mockito.when(mtmvPartitionInfo.getPartitionType()).thenReturn(MTMVPartitionType.FOLLOW_BASE_TABLE);
+ Mockito.when(mtmv.getMvProperties()).thenReturn(Maps.newHashMap());
+
Assertions.assertFalse(MTMVPartitionUtil.generateRelatedBasePartitionIds(mtmv).isPresent());
+ }
+
+ @Test
+ public void testGenerateRelatedBasePartitionIdsOnSelfManageMv() throws
AnalysisException {
+ // setUp leaves the mocked MV on SELF_MANAGE: it decides its own
partitions, so there is no
+ // base partition mapping to restrict the delta to.
+
Assertions.assertFalse(MTMVPartitionUtil.generateRelatedBasePartitionIds(mtmv).isPresent());
+ }
+
+ @Test
+ public void testGenerateRelatedBasePartitionIdsWithoutMvPartitionInfo()
throws AnalysisException {
+ MTMV mvWithoutPartitionInfo = Mockito.mock(MTMV.class);
+ Assertions.assertFalse(
+
MTMVPartitionUtil.generateRelatedBasePartitionIds(mvWithoutPartitionInfo).isPresent());
+ }
+
@Test
public void testGetBaseVersionsUsesMappedPartitions() throws
AnalysisException {
Map<String, Map<MTMVRelatedTableIf, Set<String>>> partitionMappings =
Maps.newHashMap();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPropertyUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPropertyUtilTest.java
index 4593e498e1d..2fbc5a06359 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPropertyUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPropertyUtilTest.java
@@ -106,6 +106,19 @@ class MTMVPropertyUtilTest {
Assertions.assertTrue(MTMVPropertyUtil.getIvmPartitionWindowLimit(null).isEmpty());
}
+ @Test
+ void testHasPartitionSyncLimit() {
+ String key = PropertyAnalyzer.PROPERTIES_PARTITION_SYNC_LIMIT;
+ Assertions.assertFalse(MTMVPropertyUtil.hasPartitionSyncLimit(null));
+
Assertions.assertFalse(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of()));
+
Assertions.assertFalse(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of(key,
"")));
+ // A limit that keeps nothing is not a limit: no base partition is
filtered out by it.
+
Assertions.assertFalse(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of(key,
"0")));
+
Assertions.assertFalse(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of(key,
"-1")));
+
Assertions.assertTrue(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of(key,
"1")));
+
Assertions.assertTrue(MTMVPropertyUtil.hasPartitionSyncLimit(ImmutableMap.of(key,
"2")));
+ }
+
@Test
void testGetPartitionWindowLimitNameMatching() {
Map<TableNameInfo, Integer> windowLimits =
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriterTest.java
index ac8b48a4d44..e3f4a074efd 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaRewriterTest.java
@@ -19,7 +19,10 @@ package org.apache.doris.mtmv.ivm;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.MTMV;
+import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.Partition;
import org.apache.doris.common.util.PropertyAnalyzer;
+import org.apache.doris.mtmv.BaseTableInfo;
import org.apache.doris.nereids.analyzer.UnboundTableSink;
import org.apache.doris.nereids.jobs.JobContext;
import org.apache.doris.nereids.rules.analysis.CheckAfterRewrite;
@@ -50,8 +53,12 @@ import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
class IvmDeltaRewriterTest extends IvmDeltaTestBase {
@@ -112,6 +119,25 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
new JoinReorderContext());
}
+ private Plan generateMergedDeltaWithScope(Plan plan, Map<BaseTableInfo,
Set<Long>> scopePartitionIds) {
+ ConnectContext connectContext = newConnectContext();
+ JobContext jobContext = newJobContextForRoot(plan, connectContext);
+ Plan normalizedPlan = new IvmNormalizeMTMV().rewriteRoot(plan,
jobContext);
+ IvmRewriteResult rewriteResult =
jobContext.getCascadesContext().getIvmRewriteResult().get();
+ MTMV mtmv = buildMtmvFromPlan(normalizedPlan.getOutput());
+ return new IvmDeltaRewriter().generateIncrRefreshPlan(normalizedPlan,
rewriteResult,
+ IvmRewriteContext.incremental(mtmv, scopePartitionIds),
connectContext);
+ }
+
+ /** The partition selection of every olap scan in the plan, in plan order.
*/
+ private List<List<Long>> collectedPartitionSelections(Plan plan) {
+ List<List<Long>> selections = new ArrayList<>();
+ for (LogicalOlapScan scan : collectScans(plan)) {
+ selections.add(new ArrayList<>(scan.getSelectedPartitionIds()));
+ }
+ return selections;
+ }
+
private List<LogicalOlapScan> collectScans(Plan plan) {
return plan.collectToList(n -> n instanceof LogicalOlapScan);
}
@@ -297,6 +323,53 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
Assertions.assertTrue(IvmDeltaRewriteHelper.INSTANCE.isIncrementalDeltaScan(collectScans(rewritten).get(0)));
}
+ @Test
+ void testIncrementalScopeRestrictsDeltaToScopePartitions() {
+ LogicalOlapScan scan = buildScanForTableWithTwoPartitions(120,
"scope_subset");
+ OlapTable table = scan.getTable();
+ bumpBaseTableTso(table, 20);
+ setStreamOffset(table, getRegisteredStream(table, 1L), 10);
+ long keptPartitionId = table.getPartition("p2").getId();
+
+ Plan rewritten = generateMergedDeltaWithScope(scan,
+ ImmutableMap.of(new BaseTableInfo(table),
Sets.newHashSet(keptPartitionId)));
+
+ List<LogicalOlapScan> scans = collectScans(rewritten);
+ Assertions.assertFalse(scans.isEmpty());
+ for (LogicalOlapScan rewrittenScan : scans) {
+ Assertions.assertEquals(ImmutableList.of(keptPartitionId),
+ rewrittenScan.getSelectedPartitionIds());
+ }
+ }
+
+ @Test
+ void testIncrementalScopeCoveringAllPartitionsLeavesScanUnchanged() {
+ LogicalOlapScan scan = buildScanForTableWithTwoPartitions(121,
"scope_all");
+ OlapTable table = scan.getTable();
+ bumpBaseTableTso(table, 20);
+ setStreamOffset(table, getRegisteredStream(table, 1L), 10);
+
+ Plan withoutScope = generateMergedDelta(scan, false);
+ Plan withScope = generateMergedDeltaWithScope(scan,
ImmutableMap.of(new BaseTableInfo(table),
+
table.getPartitions().stream().map(Partition::getId).collect(Collectors.toSet())));
+
+ Assertions.assertEquals(collectedPartitionSelections(withoutScope),
+ collectedPartitionSelections(withScope));
+ }
+
+ @Test
+ void testIncrementalScopeWithoutPartitionsProducesEmptyRelation() {
+ LogicalOlapScan scan = buildScanForTableWithTwoPartitions(122,
"scope_empty");
+ OlapTable table = scan.getTable();
+ bumpBaseTableTso(table, 20);
+ setStreamOffset(table, getRegisteredStream(table, 1L), 10);
+
+ Plan rewritten = generateMergedDeltaWithScope(scan,
+ ImmutableMap.of(new BaseTableInfo(table), Sets.newHashSet()));
+
+ Assertions.assertInstanceOf(LogicalEmptyRelation.class, rewritten);
+ }
+
@Test
void testRecursiveJoinDeltaMergesBothScanDeltas() {
LogicalOlapScan left = buildScanForTable(301, "recursive_left");
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
index 6e6e5fd80a6..32c78df1adc 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmDeltaTestBase.java
@@ -480,7 +480,11 @@ abstract class IvmDeltaTestBase {
return;
}
long partitionId = table.getId() * 100 + 1;
- Partition partition = new Partition(partitionId, "p1",
+ addTestPartition(table, "p1", partitionId);
+ }
+
+ private void addTestPartition(OlapTable table, String name, long
partitionId) {
+ Partition partition = new Partition(partitionId, name,
new MaterializedIndex(table.getBaseIndexId(),
MaterializedIndex.IndexState.NORMAL),
new RandomDistributionInfo(1));
partition.setVisibleVersionAndTime(Partition.PARTITION_INIT_VERSION +
1,
@@ -488,4 +492,19 @@ abstract class IvmDeltaTestBase {
partition.setNextVersion(Partition.PARTITION_INIT_VERSION + 2);
table.addPartition(partition);
}
+
+ /**
+ * Builds a scan for a table with two partitions, p1 and p2, so that a
partition subset is
+ * available to restrict a scan to.
+ */
+ protected LogicalOlapScan buildScanForTableWithTwoPartitions(long tableId,
String tableName) {
+ OlapTable table = PlanConstructor.newOlapTable(tableId, tableName, 0);
+ addTestPartition(table);
+ addTestPartition(table, "p2", tableId * 100 + 2);
+ enableRowBinlog(table);
+ table.setQualifiedDbName("test_db");
+ registerTestStreams(table);
+ return new LogicalOlapScan(PlanConstructor.getNextRelationId(), table,
+ ImmutableList.of("test_db"));
+ }
}
diff --git a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit.out
new file mode 100644
index 00000000000..ba46dd58702
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit.out
@@ -0,0 +1,14 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !complete_task --
+SUCCESS COMPLETE NONE
+
+-- !complete_mv --
+2 20 \N
+3 30 known
+
+-- !incremental_task --
+SUCCESS NONE NONE
+
+-- !incremental_mv --
+2 20 late-arriving
+3 30 known
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.out
new file mode 100644
index 00000000000..f6eb8f2e733
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.out
@@ -0,0 +1,15 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !complete_task --
+SUCCESS COMPLETE NONE
+
+-- !complete_mv --
+2 20 \N
+3 30 \N
+
+-- !incremental_task --
+SUCCESS NONE NONE
+
+-- !incremental_mv --
+2 20 \N
+3 30 late-arriving
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_bitmap_runtime_fallback.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_bitmap_runtime_fallback.groovy
index fcbe8689e88..c49540bdc44 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_bitmap_runtime_fallback.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_bitmap_runtime_fallback.groovy
@@ -18,7 +18,7 @@
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
-suite("test_ivm_bitmap_runtime_fallback", "nonConcurrent") {
+suite("test_ivm_bitmap_runtime_fallback") {
sql """drop materialized view if exists ivm_bm_fb_mv"""
sql """drop table if exists ivm_bm_fb_t"""
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_column_fallback_reason.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_column_fallback_reason.groovy
index 2019247d816..5d63b4bdf7b 100644
---
a/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_column_fallback_reason.groovy
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_column_fallback_reason.groovy
@@ -18,7 +18,7 @@
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
-suite("test_ivm_drop_column_fallback_reason", "nonConcurrent") {
+suite("test_ivm_drop_column_fallback_reason") {
if (isCloudMode()) {
return
}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_runtime_fallback.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_runtime_fallback.groovy
index 1ee2153a53e..4bc1f069d78 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_runtime_fallback.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_minmax_runtime_fallback.groovy
@@ -18,7 +18,7 @@
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
-suite("test_ivm_minmax_runtime_fallback", "nonConcurrent") {
+suite("test_ivm_minmax_runtime_fallback") {
sql """drop materialized view if exists ivm_mm_fb_mv"""
sql """drop table if exists ivm_mm_fb_t"""
diff --git a/regression-test/suites/mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy
index 5e272898f06..94828aed649 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-suite("test_ivm_mtmv_row_binlog", "nonConcurrent") {
+suite("test_ivm_mtmv_row_binlog") {
if (isCloudMode()) {
return
}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild.groovy
index 150856e3f16..488e7bd44c3 100644
---
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild.groovy
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild.groovy
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-suite("test_ivm_partition_baseline_rebuild", "nonConcurrent") {
+suite("test_ivm_partition_baseline_rebuild") {
def tableName = "ivm_part_rebuild_t"
def mvName = "ivm_part_rebuild_mv"
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit.groovy
new file mode 100644
index 00000000000..24009d05987
--- /dev/null
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit.groovy
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/**
+ * An MV that keeps only part of the base table's partitions must not break
the incremental
+ * refresh when a dimension change touches base rows that live outside that
window.
+ *
+ * <p>The MV below is partitioned by the fact table's dt and keeps only
partitions whose range
+ * upper bound is greater than the start of the current year
(partition_sync_limit=1 with
+ * partition_sync_time_unit=YEAR), so p_dropped is filtered out of the MV
while p_kept stays.
+ * Its partition set is therefore a strict subset of the base table's.
+ *
+ * <p>A late-arriving dimension row for key 99 produces a delta that joins the
dimension events
+ * against the fact snapshot; the snapshot covers every fact partition, so it
also matches the
+ * fact row stored in p_dropped and emits a delta row for a date the MV has no
partition for.
+ * The insert then fails with "no partition for this tuple", and because the
write is atomic the
+ * in-window repair is lost as well. The refresh must instead ignore the parts
of the delta that
+ * fall outside the MV's partition set and still repair the partition it does
keep.
+ *
+ * <p>All dates are literals and every partition is created by hand: no
current_date() and no
+ * dynamic partition scheduler, so the expectation does not depend on the run
date. Only the
+ * YEAR unit is used to keep the kept/dropped split stable for any run date in
this century.
+ */
+suite("test_ivm_partition_sync_limit") {
+ def factTable = "ivm_pwld_f"
+ def dimTable = "ivm_pwld_d"
+ def mvName = "ivm_pwld_mv"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+
+ sql """
+ CREATE TABLE ${factTable} (
+ order_id BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_id INT,
+ amount INT
+ )
+ UNIQUE KEY(order_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ // Empty PARTITION BY range plus manual ADD PARTITION: the partition
layout does not depend
+ // on the date the suite runs.
+ sql """ALTER TABLE ${factTable} ADD PARTITION p_dropped VALUES
[('2019-01-01'), ('2020-01-01'))"""
+ sql """ALTER TABLE ${factTable} ADD PARTITION p_kept VALUES
[('2026-01-01'), ('2099-01-01'))"""
+
+ sql """
+ CREATE TABLE ${dimTable} (
+ dimension_id INT NOT NULL,
+ dimension_name VARCHAR(32)
+ )
+ UNIQUE KEY(dimension_id)
+ DISTRIBUTED BY HASH(dimension_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """INSERT INTO ${dimTable} VALUES (10, 'known')"""
+ // order_id 1 and 2 share dimension key 99, which has no dimension row
yet: order_id 1 sits in
+ // the partition the MV does not keep, order_id 2 in the one it does.
+ sql """INSERT INTO ${factTable} VALUES
+ (1, '2019-06-01', 99, 10),
+ (2, '2026-06-15', 99, 20),
+ (3, '2026-06-16', 10, 30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(order_id, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "partition_sync_limit" = "1",
+ "partition_sync_time_unit" = "YEAR"
+ )
+ AS SELECT f.order_id, f.dt, f.amount, d.dimension_name
+ FROM ${factTable} f
+ LEFT JOIN ${dimTable} d ON f.dimension_id = d.dimension_id
+ """
+
+ // Waiting through the framework helper rather than by reading the newest
row: tasks() can
+ // briefly miss a task that just finished, and this suite reuses the MV
name across runs, so
+ // the newest row can be another run's task. Only the id is taken here;
the row the .out
+ // compares comes from taskQuery below.
+ def refreshAndGetTaskId = { String mode ->
+ sql """REFRESH MATERIALIZED VIEW ${mvName} ${mode}"""
+ waitingMTMVTaskFinishedByMvName(mvName)
+ def rows = sql_return_maparray("""
+ SELECT TaskId FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ assert !rows.isEmpty(): "no refresh task for ${mode} on ${mvName}"
+ return rows[0].TaskId.toString()
+ }
+
+ // Unset RefreshMode / IvmFallbackReason come back as the literal
two-character string "\N",
+ // which does not survive the .out round trip, so fold the unset value
into a printable token.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END,
+ CASE WHEN IvmFallbackReason = 'BINLOG_BROKEN'
+ THEN IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ // The MV keeps p_kept only, so order_id 1 is never part of it.
+ def taskId = refreshAndGetTaskId("COMPLETE")
+ qt_complete_task taskQuery(taskId)
+ order_qt_complete_mv """
+ SELECT order_id, amount, dimension_name FROM ${mvName} ORDER BY
order_id
+ """
+
+ // A late-arriving dimension row for key 99. Its delta must repair
order_id 2 (in p_kept) and
+ // must not try to write order_id 1 into a partition the MV does not have.
+ sql """INSERT INTO ${dimTable} VALUES (99, 'late-arriving')"""
+ taskId = refreshAndGetTaskId("INCREMENTAL")
+ qt_incremental_task taskQuery(taskId)
+ order_qt_incremental_mv """
+ SELECT order_id, amount, dimension_name FROM ${mvName} ORDER BY
order_id
+ """
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.groovy
new file mode 100644
index 00000000000..2713db9333f
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_sync_limit_with_window.groovy
@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+/**
+ * partition_sync_limit and ivm_partition_window_limit together: the base
partitions the
+ * incremental delta may read are the INTERSECTION of the two.
+ *
+ * <p>The MV keeps the partitions above the start of the current year
(partition_sync_limit=1 with
+ * YEAR), which is p_mid and p_new, while p_dropped is filtered out. The
compute window keeps the
+ * last partition by value, which is p_new alone.
+ *
+ * <p>So only p_new may be read: p_dropped must stay unread, or the delta
would emit rows for a
+ * date the MV has no partition for, and p_mid must stay unread too, or the
window would be
+ * ignored and the MV would be maintained outside it. A dimension change
touches one fact row in
+ * each partition, so the expectation separates the three: p_new is repaired,
p_mid keeps the
+ * value the window says not to maintain, p_dropped is not part of the MV at
all.
+ *
+ * <p>All dates are literals and every partition is created by hand: no
current_date() and no
+ * dynamic partition scheduler. The YEAR unit and the far-future range upper
bounds keep the
+ * kept/dropped split stable for any run date in this century.
+ */
+suite("test_ivm_partition_sync_limit_with_window") {
+ def factTable = "ivm_pslw_f"
+ def dimTable = "ivm_pslw_d"
+ def mvName = "ivm_pslw_mv"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+
+ sql """
+ CREATE TABLE ${factTable} (
+ order_id BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ dimension_id INT,
+ amount INT
+ )
+ UNIQUE KEY(order_id, dt)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+ sql """ALTER TABLE ${factTable} ADD PARTITION p_dropped VALUES
[('2019-01-01'), ('2020-01-01'))"""
+ sql """ALTER TABLE ${factTable} ADD PARTITION p_mid VALUES
[('2026-01-01'), ('2099-01-01'))"""
+ sql """ALTER TABLE ${factTable} ADD PARTITION p_new VALUES
[('2099-01-01'), ('2199-01-01'))"""
+
+ sql """
+ CREATE TABLE ${dimTable} (
+ dimension_id INT NOT NULL,
+ dimension_name VARCHAR(32)
+ )
+ UNIQUE KEY(dimension_id)
+ DISTRIBUTED BY HASH(dimension_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true"
+ )
+ """
+
+ sql """INSERT INTO ${dimTable} VALUES (10, 'known')"""
+ // One fact row per partition, all on dimension key 99, which has no
dimension row yet.
+ sql """INSERT INTO ${factTable} VALUES
+ (1, '2019-06-01', 99, 10),
+ (2, '2026-06-15', 99, 20),
+ (3, '2099-06-15', 99, 30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(order_id, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(order_id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "partition_sync_limit" = "1",
+ "partition_sync_time_unit" = "YEAR",
+ "ivm_partition_window_limit" = "${factTable}:1"
+ )
+ AS SELECT f.order_id, f.dt, f.amount, d.dimension_name
+ FROM ${factTable} f
+ LEFT JOIN ${dimTable} d ON f.dimension_id = d.dimension_id
+ """
+
+ // Waiting through the framework helper rather than by reading the newest
row: tasks() can
+ // briefly miss a task that just finished, and this suite reuses the MV
name across runs, so
+ // the newest row can be another run's task. Only the id is taken here;
the row the .out
+ // compares comes from taskQuery below.
+ def refreshAndGetTaskId = { String mode ->
+ sql """REFRESH MATERIALIZED VIEW ${mvName} ${mode}"""
+ waitingMTMVTaskFinishedByMvName(mvName)
+ def rows = sql_return_maparray("""
+ SELECT TaskId FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ assert !rows.isEmpty(): "no refresh task for ${mode} on ${mvName}"
+ return rows[0].TaskId.toString()
+ }
+
+ // Unset RefreshMode / IvmFallbackReason come back as the literal
two-character string "\N",
+ // which does not survive the .out round trip, so fold the unset value
into a printable token.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END,
+ CASE WHEN IvmFallbackReason = 'BINLOG_BROKEN'
+ THEN IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ // The MV keeps p_mid and p_new, so order_id 1 is never part of it. The
window is an
+ // incremental-path property, so the complete refresh still builds both MV
partitions.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ def taskId = refreshAndGetTaskId("COMPLETE")
+ qt_complete_task taskQuery(taskId)
+ order_qt_complete_mv """
+ SELECT order_id, amount, dimension_name FROM ${mvName} ORDER BY
order_id
+ """
+
+ // A late-arriving dimension row for key 99 touches the fact row in every
partition.
+ sql """INSERT INTO ${dimTable} VALUES (99, 'late-arriving')"""
+ taskId = refreshAndGetTaskId("INCREMENTAL")
+ qt_incremental_task taskQuery(taskId)
+ order_qt_incremental_mv """
+ SELECT order_id, amount, dimension_name FROM ${mvName} ORDER BY
order_id
+ """
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${factTable}"""
+ sql """DROP TABLE IF EXISTS ${dimTable}"""
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_rewrite_projection.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_rewrite_projection.groovy
index cf8d21b7831..fd9f41e168d 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_rewrite_projection.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_rewrite_projection.groovy
@@ -15,7 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-suite("test_ivm_rewrite_projection", "nonConcurrent") {
+suite("test_ivm_rewrite_projection") {
sql """drop materialized view if exists rewrite_projection_ivm;"""
sql """drop table if exists rewrite_projection_base;"""
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_strict_failure_partition_atomicity.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_strict_failure_partition_atomicity.groovy
index 1b7acd59eca..13b9edc7e5c 100644
---
a/regression-test/suites/mtmv_p0/ivm/test_ivm_strict_failure_partition_atomicity.groovy
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_strict_failure_partition_atomicity.groovy
@@ -18,7 +18,7 @@
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
-suite("test_ivm_strict_failure_partition_atomicity", "nonConcurrent") {
+suite("test_ivm_strict_failure_partition_atomicity") {
sql """DROP MATERIALIZED VIEW IF EXISTS ivm_strict_atomicity_mv"""
sql """DROP TABLE IF EXISTS ivm_strict_atomicity_t"""
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
