This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-incremental-computation
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-incremental-computation
by this push:
new 29b9e48a9bc branch-incremental-computation: pick the merged
incremental-computation PRs from master in merge order (#67753 #67802 #67814
#67837 #67853 #67876) (#68017)
29b9e48a9bc is described below
commit 29b9e48a9bcd4bc4dc86f17e5ca5adfcde9697be
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Tue Sep 15 18:12:09 2026 +0800
branch-incremental-computation: pick the merged incremental-computation PRs
from master in merge order (#67753 #67802 #67814 #67837 #67853 #67876) (#68017)
Cherry-picked from #67753, #67802, #67814, #67837, #67853, #67876
Batch pick of every merged PR carrying the `incremental-computation`
label that `branch-incremental-computation` does not have yet (no
`incremental-computation-picked` label), in the order they landed on
master (`git log --first-parent`). One commit per PR, each created with
`git cherry-pick -x` so the message ends with `(cherry picked from
commit <master sha>)`. Follows the same convention as #67830.
| # | Master commit | PR | Title |
|---|---|---|---|
| 1 | fe39f5b6a42 | #67753 | [fix](ivm) Answer FE-computable dry runs on
the frontend instead of a placeholder backend |
| 2 | f8ed33fa701 | #67802 | [fix](ivm) Refresh the surviving partitions
after an IVM baseline rebuild |
| 3 | 7bd89a0795c | #67814 | [fix](ivm) Stop the incremental delta from
reading partitions the MV dropped |
| 4 | 3050a9ae8ae | #67837 | [fix](ivm) Invalidate the baseline when a
column used by the MV is dropped |
| 5 | 22c95eb8d5c | #67853 | [fix](ivm) Carry the row-binlog hidden
columns in the analyzed MTMV schema |
| 6 | 3390a7a721f | #67876 | [test](ivm) Remove unnecessary cloud skips
from IVM suites |
Not included on purpose:
- The 11 labelled PRs that already carry
`incremental-computation-picked` (#62606 in the fork point, #67508 via
#67712, the nine of #67830).
- #67820 is still open on master; this branch already carries its
content via #67861.
### Prerequisite check
None of the six PRs declares a prerequisite, and none of them needs
another master PR for its behavior. The only master commits that touch
the same files and are not on this branch are unrelated to incremental
computation (#66761 TIMESTAMP_NS, #67545 DLF, #67569 / #67520 / #67835
MySQL-protocol and session refactors, #67186 Hive partition batching,
#67787 SQL cache user variables); they were left out, and two picks
needed a mechanical adaptation because of that:
- **#67753** conflicted in `StmtExecutor.sendMetaData`: master had
already extracted the post-metadata EOF into
`sendMetadataTerminatorIfNeeded(channel)` (#67520, a Connector/J
cursor-fetch fix). The branch keeps its inline EOF block and now sends
it on the given `channel` instead of `context.getMysqlChannel()`, which
is exactly what the extracted helper does on master. Everything else in
the pick is identical to the master commit.
- **#67814** applied cleanly but did not compile: the new
`MTMVPartitionUtil.generateRelatedBasePartitionIds()` returns an
`Optional`, and on master `import java.util.Optional;` came with #67186.
The import was added to the pick commit; that is the only difference
from the master commit.
The other four picks applied without conflicts and are byte-identical to
their master commits (diffs compared with `index`/`@@` lines stripped).
Both adaptations are recorded in the respective commit messages.
### Drift check against master
After the six picks, every touched file is byte-identical to master at
`3390a7a721f` except `MTMV.java`, `MTMVTask.java`,
`MTMVPartitionUtil.java`, `MTMVPartitionUtilTest.java`,
`MTMVTaskTest.java` (differences = #67186 + #67545 + #66761),
`CreateTableInfo.java` (= #67787) and `StmtExecutor.java` (= #67520 +
#67569 + #66761 + the later session refactors #67835 / #67883 + this
branch's #67861). For the first six files, applying those unrelated
master commits on top of the branch's versions reproduces master's files
exactly; for `StmtExecutor.java`, the diff against master right after
#67753 (`fe39f5b6a42`) consists only of #67520 / #67569 / #66761 /
#67861 hunks. So nothing IVM-related is missing. The regression
framework, plugins and the whole `mtmv_p0/ivm` suite/data directories
are identical to master.
### Verification
- FE: `run-fe-ut.sh --run` on this branch (regenerates thrift, compiles
fe-core main + test) with the 17 test classes touched by the picks or
extending the touched `IvmDeltaTestBase`:
17 classes, 404 tests, 0 failures, 0 errors, BUILD SUCCESS (5:19 min) —
`MTMVPlanUtilTest` 24, `IvmAggDeltaHandlerTest` 33,
`IvmDeltaRewriteHelperTest` 17, `IvmNormalizeMTMVJoinTest` 44,
`IvmJoinDeltaHandlerTest` 23, `IvmDeltaRewriteStateTest` 10,
`IvmPlanSignatureGeneratorTest` 22, `IvmBaselineRebuildTest` 28,
`IvmLinearDeltaHandlerTest` 39, `IvmDeltaRewriterTest` 23,
`IvmNormalizeMTMVUnionTest` 10, `MTMVTaskTest` 50,
`MTMVPropertyUtilTest` 13, `MTMVPartitionUtilTest` 16,
`SchemaChangeHandlerTest` 22, `StmtExecutorInternalQueryTest` 3,
`StmtExecutorTest` 27.
- FE checkstyle on fe-core: 0 violations.
- No BE, cloud or thrift changes in this batch.
- All 18 touched groovy files (framework `Suite.groovy`,
`plugin_planner.groovy`, 16 suites) parse cleanly (groovy parser check).
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: yujun <[email protected]>
---
.../main/java/org/apache/doris/catalog/MTMV.java | 35 ++++
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 56 ++++--
.../org/apache/doris/mtmv/MTMVPartitionUtil.java | 58 ++++++
.../java/org/apache/doris/mtmv/MTMVPlanUtil.java | 9 +
.../org/apache/doris/mtmv/MTMVPropertyUtil.java | 13 ++
.../org/apache/doris/mtmv/MTMVRelationManager.java | 69 +++++++-
.../apache/doris/mtmv/ivm/IvmDeltaRewriter.java | 64 ++++++-
.../doris/mtmv/ivm/IvmIncrRefreshManager.java | 16 +-
.../apache/doris/mtmv/ivm/IvmRewriteContext.java | 51 +++++-
.../trees/plans/commands/info/CreateTableInfo.java | 28 ++-
.../java/org/apache/doris/qe/StmtExecutor.java | 71 +++++++-
.../doris/alter/SchemaChangeHandlerTest.java | 8 +
.../apache/doris/mtmv/MTMVPartitionUtilTest.java | 23 +++
.../org/apache/doris/mtmv/MTMVPlanUtilTest.java | 64 +++++++
.../apache/doris/mtmv/MTMVPropertyUtilTest.java | 13 ++
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 42 ++++-
.../doris/mtmv/ivm/IvmBaselineRebuildTest.java | 36 ++++
.../doris/mtmv/ivm/IvmDeltaRewriterTest.java | 73 ++++++++
.../apache/doris/mtmv/ivm/IvmDeltaTestBase.java | 21 ++-
...ivm_drop_referenced_column_baseline_rebuild.out | 8 +
...est_ivm_partition_baseline_rebuild_dup_keys.out | 32 ++++
.../ivm/test_ivm_partition_drop_live_delta.out | 45 +++++
.../mtmv_p0/ivm/test_ivm_partition_sync_limit.out | 14 ++
.../test_ivm_partition_sync_limit_with_window.out | 15 ++
.../data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out | 12 +-
.../ivm/test_ivm_row_binlog_schema_validation.out | 27 +++
.../org/apache/doris/regression/suite/Suite.groovy | 12 +-
regression-test/plugins/plugin_planner.groovy | 21 +++
.../ivm/test_ivm_bitmap_runtime_fallback.groovy | 2 +-
.../test_ivm_drop_column_fallback_reason.groovy | 6 +-
..._drop_referenced_column_baseline_rebuild.groovy | 144 +++++++++++++++
.../test_ivm_fallback_stream_multi_batch.groovy | 4 -
...test_ivm_fallback_stream_multi_batch_dup.groovy | 4 -
.../ivm/test_ivm_minmax_runtime_fallback.groovy | 2 +-
.../mtmv_p0/ivm/test_ivm_mtmv_row_binlog.groovy | 6 +-
.../ivm/test_ivm_partition_baseline_rebuild.groovy | 2 +-
..._ivm_partition_baseline_rebuild_dup_keys.groovy | 123 +++++++++++++
.../ivm/test_ivm_partition_drop_live_delta.groovy | 134 ++++++++++++++
.../ivm/test_ivm_partition_sync_limit.groovy | 159 +++++++++++++++++
...est_ivm_partition_sync_limit_with_window.groovy | 157 +++++++++++++++++
.../mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy | 29 +--
.../mtmv_p0/ivm/test_ivm_rewrite_projection.groovy | 2 +-
.../test_ivm_row_binlog_schema_validation.groovy | 194 +++++++++++++++++++++
...t_ivm_strict_failure_partition_atomicity.groovy | 2 +-
44 files changed, 1811 insertions(+), 95 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
index 40dba304805..f6047d18a48 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/MTMV.java
@@ -643,6 +643,41 @@ public class MTMV extends OlapTable {
editLogItem.await();
}
+ /**
+ * Release the IVM baseline barrier after the partitions it named have
been rebuilt, or after
+ * partition sync removed them (a dropped partition resolves its own
entry: the partition and its
+ * IVM offsets are both gone).
+ *
+ * <p>Guarded by schemaChangeVersion, like {@link
#persistIvmBaselineGuard}: a base-table change
+ * landing while the rebuild runs carries its own barrier entry, and a
blind clear would swallow
+ * it. Failing instead preserves that entry -- the next refresh rebuilds
it together with the
+ * partitions this task handled.
+ *
+ * <p>Journals the new state right away, like every other ivmInfo mutation
here. A task that dies
+ * before {@link #addTaskResult} would otherwise leave the release in
memory only, and a restart
+ * would resurrect the barrier from disk.
+ */
+ public void releaseIvmBaselineRebuild(long expectedSchemaChangeVersion)
throws JobException {
+ EditLogItem editLogItem;
+ writeMvLock();
+ try {
+ if (ivmInfo == null || !ivmInfo.isBaselineRebuildRequired()) {
+ // Nothing to release: skip both the mutation and the journal
entry. Any base-table
+ // change that raced us in is still caught by
validateIvmRefreshStart() below.
+ return;
+ }
+ if (schemaChangeVersion != expectedSchemaChangeVersion) {
+ throw new JobException("Base table metadata changed before IVM
baseline refresh, mv="
+ + getName());
+ }
+ ivmInfo.clearBaselineRebuild();
+ editLogItem = submitIvmInfoChange();
+ } finally {
+ writeMvUnlock();
+ }
+ editLogItem.await();
+ }
+
public void persistIvmBaselineGuard(RefreshMode refreshMode, Set<String>
baselinePartitions,
long expectedSchemaChangeVersion) throws JobException {
EditLogItem editLogItem;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
index f41ba369fe8..0eaa967cce9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/mtmv/MTMVTask.java
@@ -319,9 +319,7 @@ public class MTMVTask extends AbstractTask {
throw new JobException(e.getMessage(), e);
}
MTMVRefreshContext refreshContext = buildRefreshContext(tableIfs);
- if (handlePendingIvmBaselineRebuild(refreshContext, request, ctx))
{
- return;
- }
+ handlePendingIvmBaselineRebuild(refreshContext, request, ctx,
attempts);
boolean disablePartitionRefresh = false;
for (RefreshAttemptType attemptType : attempts) {
switch (attemptType) {
@@ -558,28 +556,56 @@ public class MTMVTask extends AbstractTask {
executePartitionBasedRefresh(context, RefreshMode.COMPLETE, ctx);
}
- private boolean handlePendingIvmBaselineRebuild(MTMVRefreshContext
context, RefreshRequest request,
- ConnectContext ctx)
+ /**
+ * Rebuild the MV partitions whose IVM baseline is broken, before the
normal refresh runs.
+ *
+ * <p>This is a pre-step, not a terminal branch: the caller keeps running
{@code attempts}
+ * afterwards, so a broken baseline no longer skips the refresh entirely.
The list is rewritten
+ * in place when the baseline demands a different set of attempts.
+ *
+ * <p>Partition sync drops the MV partitions whose base partition
disappeared, which is exactly
+ * what the barrier recorded when that base partition was dropped. Those
partitions are resolved
+ * by the drop itself (the partition and its IVM offsets are both gone),
so only the partitions
+ * that still exist need a rebuild. The barrier is released either way,
otherwise the IVM attempt
+ * that follows would be rejected by {@link MTMV#validateIvmRefreshStart}.
+ */
+ private void handlePendingIvmBaselineRebuild(MTMVRefreshContext context,
+ RefreshRequest request, ConnectContext ctx,
List<RefreshAttemptType> attempts)
throws JobException, AnalysisException {
if (!mtmv.isIvm() || request.refreshMode == RefreshMode.COMPLETE
|| !mtmv.getIvmInfo().isBaselineRebuildRequired()) {
- return false;
+ return;
}
ivmFallbackReason = IvmFailureReason.BINLOG_BROKEN.name();
IvmInfo ivmInfo = mtmv.getIvmInfo();
+ // A lone COMPLETE attempt rebuilds every partition anyway, so a
partial pre-rebuild here
+ // would be redundant; it also releases the barrier by itself once it
succeeds.
+ if (attempts.size() == 1 && attempts.get(0) ==
RefreshAttemptType.COMPLETE) {
+ LOG.info("IVM baseline barrier is covered by the pending COMPLETE
attempt, mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ return;
+ }
if (ivmInfo.requiresCompleteBaselineRebuild()) {
- executeCompleteAttempt(context, ctx);
- return true;
+ LOG.warn("IVM baseline requires a complete rebuild, mv={},
taskId={}. "
+ + "Continuing with COMPLETE refresh.", mtmv.getName(),
getTaskId());
+ attempts.clear();
+ attempts.add(RefreshAttemptType.COMPLETE);
+ return;
}
- this.needRefreshPartitions = Lists.newArrayList(Sets.intersection(
+ List<String> baselinePartitions = Lists.newArrayList(Sets.intersection(
ivmInfo.getPendingBaselineRebuildPartitions(),
mtmv.getPartitionNames()));
- this.needRefreshPartitions.sort(String::compareTo);
- this.refreshMode = generateRefreshMode(needRefreshPartitions);
- if (refreshMode == MTMVTaskRefreshMode.NOT_REFRESH) {
- return true;
+ if (baselinePartitions.isEmpty()) {
+ // Partition sync has already dropped every partition the barrier
named, so there is
+ // nothing left to rebuild. The surviving partitions are picked up
by the attempts below.
+ LOG.info("IVM baseline partitions were removed by partition sync,
mv={}, taskId={}",
+ mtmv.getName(), getTaskId());
+ } else {
+ baselinePartitions.sort(String::compareTo);
+ this.needRefreshPartitions = baselinePartitions;
+ this.refreshMode = generateRefreshMode(baselinePartitions);
+ executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
}
- executePartitionBasedRefresh(context, RefreshMode.PARTITIONS, ctx);
- return true;
+ mtmv.releaseIvmBaselineRebuild(mtmvSchemaChangeVersion);
}
private void validateIvmBaselineBeforePartitionSync(RefreshRequest
request) throws JobException {
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 700ab47fa44..4505534a73d 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
@@ -59,6 +59,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -211,6 +212,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/MTMVPlanUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
index 4d708208c5b..11f50909ccc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVPlanUtil.java
@@ -23,6 +23,7 @@ import org.apache.doris.analysis.StatementBase;
import org.apache.doris.analysis.ToSqlParams;
import org.apache.doris.analysis.UserIdentity;
import org.apache.doris.catalog.AggregateType;
+import org.apache.doris.catalog.BinlogConfig;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.DistributionInfo;
@@ -625,6 +626,14 @@ public class MTMVPlanUtil {
properties = CreateTableInfo.addOlapHiddenColumns(
columns, isIvm ? KeysType.UNIQUE_KEYS : KeysType.DUP_KEYS,
isIvm, properties, false);
+ // A row-binlog table carries hidden columns on top of the OLAP
ones above, added by
+ // InternalCatalog#createOlapTable just before the table is built.
The analyzed list has
+ // to carry them too: an MTMV re-validates its schema against it
whenever a base table
+ // changes (MTMVPlanUtil#checkColumnIfChange), and a column
missing here is
+ // indistinguishable from a real schema change. Idempotent, so the
table still gets one.
+ CreateTableInfo.addRowBinlogHiddenColumns(columns,
+ isIvm ? KeysType.UNIQUE_KEYS : KeysType.DUP_KEYS, isIvm,
+ BinlogConfig.fromProperties(properties));
// analyze column
final boolean finalEnableMergeOnWrite = isIvm;
Set<String> keysSet =
Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
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/MTMVRelationManager.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
index dd2806fd88b..3bfa209fd6f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/MTMVRelationManager.java
@@ -86,16 +86,16 @@ public class MTMVRelationManager implements MTMVHookService
{
}
public void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, String
reason) {
- markIvmBaselineRebuild(baseTableInfo, Collections.emptyMap(), reason);
+ markIvmBaselineRebuild(baseTableInfo, true, Collections.emptyMap(),
reason);
}
public void markIvmBaselineRebuildForPartitionChange(BaseTableInfo
baseTableInfo,
Map<String, Long> changedPartitions, String reason) {
Preconditions.checkArgument(!changedPartitions.isEmpty(), "changed
partitions can not be empty");
- markIvmBaselineRebuild(baseTableInfo, changedPartitions, reason);
+ markIvmBaselineRebuild(baseTableInfo, false, changedPartitions,
reason);
}
- private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo,
+ private void markIvmBaselineRebuild(BaseTableInfo baseTableInfo, boolean
allPartitionsChanged,
Map<String, Long> changedPartitions, String reason) {
TableNameInfo baseTableName = new
TableNameInfo(baseTableInfo.getCtlName(),
baseTableInfo.getDbName(), baseTableInfo.getTableName());
@@ -115,7 +115,7 @@ public class MTMVRelationManager implements MTMVHookService
{
if
(MTMVPartitionUtil.isTableExcluded(mtmv.getExcludedTriggerTables(),
baseTableName)) {
continue;
}
- if (changedPartitions.isEmpty()) {
+ if (allPartitionsChanged) {
mtmv.invalidateIvmBaseline();
} else {
mtmv.invalidateIvmBaseline(baseTableInfo, changedPartitions);
@@ -335,7 +335,9 @@ public class MTMVRelationManager implements MTMVHookService
{
*/
@Override
public void dropTable(Table table) {
- processBaseTableChange(new BaseTableInfo(table), "The base table has
been deleted:");
+ // A dropped base table is already caught by the IVM stream guard (the
stream records the
+ // base table id, so it stops being usable once the table is gone), no
need to re-analyze.
+ processBaseTableChange(new BaseTableInfo(table), "The base table has
been deleted:", false);
}
/**
@@ -347,9 +349,56 @@ public class MTMVRelationManager implements
MTMVHookService {
public void alterTable(BaseTableInfo oldTableInfo, Optional<BaseTableInfo>
newTableInfo, boolean isReplace) {
// when replace, need deal two table
if (isReplace) {
- processBaseTableChange(newTableInfo.get(), "The base table has
been updated:");
+ // REPLACE TABLE already invalidates the IVM baseline explicitly,
see Alter#processReplaceTable
+ processBaseTableChange(newTableInfo.get(), "The base table has
been updated:", false);
+ }
+ // A RENAME leaves every column alone, and the failure it does cause
-- the MV query still
+ // spells the old name -- is already reported by the refresh itself
(MTMVTask#run resolves
+ // the base tables from the query before it ever looks at the
baseline). Invalidating here
+ // would only leave a stale flag behind: rename the table back and the
query is analyzable
+ // again, yet every strict INCREMENTAL refresh would stay rejected
until a COMPLETE one ran.
+ boolean renamed = !isReplace && newTableInfo.isPresent()
+ && !Objects.equals(oldTableInfo.getTableName(),
newTableInfo.get().getTableName());
+ processBaseTableChange(oldTableInfo, "The base table has been
updated:", !renamed);
+ }
+
+ /**
+ * An IVM baseline is only valid while the MV query can still be analyzed
against the current
+ * base table schema. Re-analyzing the MV query here (right after the
alter was applied) is what
+ * detects a changed column identity: dropping or renaming a column the MV
uses makes the query
+ * unanalyzable, and a column re-added with the same name is a different
column, so pre-existing
+ * rows read its default value instead.
+ *
+ * <p>Such a change is metadata-only for light schema changes and emits no
binlog, so an
+ * incremental refresh would consume an empty delta and report SUCCESS
while silently keeping the
+ * rows computed under the old column epoch. Invalidating the baseline
makes a strict INCREMENTAL
+ * refresh fail and tell the user to run a COMPLETE refresh instead.
+ *
+ * <p>Only IVM is covered: a plain MTMV keeps its previous behaviour
(status only).
+ */
+ private void invalidateIvmBaselineIfQueryUnusable(BaseTableInfo
baseTableInfo, Table mtmvTable) {
+ if (!(mtmvTable instanceof MTMV) || !((MTMV) mtmvTable).isIvm()) {
+ return;
+ }
+ MTMV mtmv = (MTMV) mtmvTable;
+ // Analyse in a context owned by this check, never the session that
issued the alter: the check
+ // must not disturb the running statement, and it has to work on
threads that have no session.
+ // Setting a thread local is how a context is made current, so restore
the previous one.
+ ConnectContext previousCtx = ConnectContext.get();
+ try {
+ MTMVPlanUtil.ensureMTMVQueryUsable(mtmv,
+ MTMVPlanUtil.createMTMVContext(mtmv,
MTMVPlanUtil.DISABLE_RULES_WHEN_RUN_MTMV_TASK));
+ } catch (Exception e) {
+ LOG.info("Invalidate IVM baseline, the MV query is no longer
usable. baseTable={}, mtmv={}, "
+ + "reason={}", baseTableInfo, mtmv.getName(),
e.getMessage());
+ mtmv.invalidateIvmBaseline();
+ } finally {
+ if (previousCtx != null) {
+ previousCtx.setThreadLocalInfo();
+ } else {
+ ConnectContext.remove();
+ }
}
- processBaseTableChange(oldTableInfo, "The base table has been
updated:");
}
@Override
@@ -411,7 +460,8 @@ public class MTMVRelationManager implements MTMVHookService
{
}
}
- private void processBaseTableChange(BaseTableInfo baseTableInfo, String
msgPrefix) {
+ private void processBaseTableChange(BaseTableInfo baseTableInfo, String
msgPrefix,
+ boolean checkIvmQueryUsable) {
Set<BaseTableInfo> mtmvsByBaseTable =
getMtmvsByBaseTableOneLevelAndFromView(baseTableInfo);
if (CollectionUtils.isEmpty(mtmvsByBaseTable)) {
return;
@@ -424,6 +474,9 @@ public class MTMVRelationManager implements MTMVHookService
{
LOG.warn(e);
continue;
}
+ if (checkIvmQueryUsable) {
+ invalidateIvmBaselineIfQueryUnusable(baseTableInfo, mtmv);
+ }
TableNameInfo tableNameInfo = new
TableNameInfo(mtmv.getQualifiedDbName(),
mtmv.getName());
MTMVStatus status = new MTMVStatus(MTMVState.SCHEMA_CHANGE,
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/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
index ac39d7256a7..c23a95b85c6 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/CreateTableInfo.java
@@ -1646,14 +1646,34 @@ public class CreateTableInfo {
* Add hidden columns required by row binlog.
*/
public void createRowBinlogHiddenColumnsIfNecessary(BinlogConfig
binlogConfig) {
- if (!binlogConfig.isRowFormat()) {
+ addRowBinlogHiddenColumns(columns, keysType, isEnableMergeOnWrite,
binlogConfig);
+ }
+
+ /**
+ * Append the hidden columns a row-binlog table carries. Callers that
build the column list
+ * outside the create-table flow (an analyzed MTMV schema) go through here
as well, so both
+ * sides of {@code MTMVPlanUtil#checkColumnIfChange} agree on the physical
layout.
+ *
+ * <p>Idempotent: a column that is already present is kept once.
+ */
+ public static void addRowBinlogHiddenColumns(List<ColumnDefinition>
columns, KeysType keysType,
+ boolean isEnableMergeOnWrite, BinlogConfig binlogConfig) {
+ if (binlogConfig == null || !binlogConfig.isRowFormat()) {
return;
}
if (keysType.equals(KeysType.DUP_KEYS)) {
-
columns.add(ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
-
columns.add(ColumnDefinition.newRowLsnColumnDefinition(AggregateType.NONE));
+ addIfAbsent(columns,
ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
+ addIfAbsent(columns,
ColumnDefinition.newRowLsnColumnDefinition(AggregateType.NONE));
} else if (keysType.equals(KeysType.UNIQUE_KEYS) &&
isEnableMergeOnWrite) {
-
columns.add(ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
+ addIfAbsent(columns,
ColumnDefinition.newCommitTsoColumnDefinition(AggregateType.NONE));
+ }
+ }
+
+ private static void addIfAbsent(List<ColumnDefinition> columns,
ColumnDefinition columnDefinition) {
+ boolean present = columns.stream()
+ .anyMatch(column ->
column.getName().equalsIgnoreCase(columnDefinition.getName()));
+ if (!present) {
+ columns.add(columnDefinition);
}
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index a2cbd7ec204..a7a28604c5a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1853,11 +1853,16 @@ public class StmtExecutor {
}
private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo>
fieldInfos) throws IOException {
+ sendMetaData(metaData, fieldInfos, context.getMysqlChannel());
+ }
+
+ private void sendMetaData(ResultSetMetaData metaData, List<FieldInfo>
fieldInfos, MysqlChannel channel)
+ throws IOException {
Preconditions.checkState(context.getConnectType() ==
ConnectType.MYSQL);
// sends how many columns
serializer.reset();
serializer.writeVInt(metaData.getColumnCount());
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
// send field one by one
for (int i = 0; i < metaData.getColumns().size(); i++) {
Column col = metaData.getColumn(i);
@@ -1868,16 +1873,16 @@ public class StmtExecutor {
} else {
serializer.writeField(fieldInfos.get(i), col.getType());
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
// When CLIENT_DEPRECATE_EOF is set, the server should not send the
intermediate
// EOF packet after column definitions. The client will go directly
from column
// definitions to reading data rows.
- if (!context.getMysqlChannel().clientDeprecatedEOF()) {
+ if (!channel.clientDeprecatedEOF()) {
serializer.reset();
MysqlEofPacket eofPacket = new MysqlEofPacket(context.getState());
eofPacket.writeTo(serializer);
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
}
@@ -2030,19 +2035,34 @@ public class StmtExecutor {
}
public void sendResultSet(ResultSet resultSet, List<FieldInfo> fieldInfos)
throws IOException {
+ sendResultSet(resultSet, fieldInfos, null);
+ }
+
+ /**
+ * Sends a FE-computed result set to the given mysql channel. Regular
queries use the
+ * executor's own channel; internal queries have to stream to the channel
of the caller
+ * that issued them, because the executor's own channel is not connected
to that client.
+ *
+ * <p>A null channel means the session's own. It is resolved inside the
mysql branch on
+ * purpose: a connection of any other type has no mysql channel, and
asking for one throws,
+ * so a caller must be able to hand a result set over without naming a
channel first.
+ */
+ private void sendResultSet(ResultSet resultSet, List<FieldInfo>
fieldInfos, MysqlChannel channel)
+ throws IOException {
if (context.getConnectType().equals(ConnectType.MYSQL)) {
+ MysqlChannel targetChannel = channel == null ?
context.getMysqlChannel() : channel;
context.updateReturnRows(resultSet.getResultRows().size());
// Send meta data.
- sendMetaData(resultSet.getMetaData(), fieldInfos);
+ sendMetaData(resultSet.getMetaData(), fieldInfos, targetChannel);
// Send result set.
if (isComStmtExecute) {
if (LOG.isDebugEnabled()) {
LOG.debug("Use binary protocol to set result.");
}
- sendBinaryResultRow(resultSet);
+ sendBinaryResultRow(resultSet, targetChannel);
} else {
- sendTextResultRow(resultSet);
+ sendTextResultRow(resultSet, targetChannel);
}
context.getState().setEof();
} else if
(context.getConnectType().equals(ConnectType.ARROW_FLIGHT_SQL)) {
@@ -2056,6 +2076,10 @@ public class StmtExecutor {
}
protected void sendTextResultRow(ResultSet resultSet) throws IOException {
+ sendTextResultRow(resultSet, context.getMysqlChannel());
+ }
+
+ protected void sendTextResultRow(ResultSet resultSet, MysqlChannel
channel) throws IOException {
for (List<String> row : resultSet.getResultRows()) {
serializer.reset();
for (String item : row) {
@@ -2065,11 +2089,15 @@ public class StmtExecutor {
serializer.writeLenEncodedString(item);
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
}
protected void sendBinaryResultRow(ResultSet resultSet) throws IOException
{
+ sendBinaryResultRow(resultSet, context.getMysqlChannel());
+ }
+
+ protected void sendBinaryResultRow(ResultSet resultSet, MysqlChannel
channel) throws IOException {
//
https://dev.mysql.com/doc/dev/mysql-server/latest/page_protocol_binary_resultset.html#sect_protocol_binary_resultset_row_value
ResultSetMetaData metaData = resultSet.getMetaData();
int nullBitmapLength = (metaData.getColumnCount() + 7 + 2) / 8;
@@ -2127,7 +2155,7 @@ public class StmtExecutor {
}
}
}
- context.getMysqlChannel().sendOnePacket(serializer.toByteBuffer());
+ channel.sendOnePacket(serializer.toByteBuffer());
}
}
@@ -2302,6 +2330,31 @@ public class StmtExecutor {
planner = new NereidsPlanner(statementContext);
planner.plan(adapter, context.getSessionVariable().toThrift());
+ // A plan that FE can compute on its own (e.g. the empty delta of
a dry run) must be
+ // answered by the frontend, like a regular query does. Otherwise
the coordinator
+ // would send fragments to the placeholder backend registered when
no backend is
+ // needed (see NereidsPlanner#notNeedBackend), which cannot
resolve. Results go to the
+ // caller's channel: the executor's own channel is not connected
to that client.
+ if (context.supportHandleByFe()) {
+ Optional<ResultSet> resultSet =
planner.handleQueryInFe(adapter);
+ if (resultSet.isPresent()) {
+ boolean sendToChannel = !collectMode;
+ if (sendToChannel) {
+ sendResultSet(resultSet.get(),
adapter.getFieldInfos(), sendChannel);
+ }
+ isHandleQueryInFe = true;
+ if (context.getSessionVariable().enableProfile() &&
profile != null) {
+
profile.getSummaryProfile().setExecutedByFrontend(true);
+ }
+ if (sendToChannel) {
+ return new ArrayList<>();
+ }
+ return resultSet.get().getResultRows().stream()
+ .map(ResultRow::new)
+ .collect(Collectors.toList());
+ }
+ }
+
if (!collectMode) {
executeAndSendResult(false, false, adapter, sendChannel, null,
null);
return new ArrayList<>();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
index 4b3962b9142..1be3cca764a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/alter/SchemaChangeHandlerTest.java
@@ -422,6 +422,14 @@ public class SchemaChangeHandlerTest extends
TestWithFeService {
createTable(create);
expectException("ALTER TABLE test." + tableName + " MODIFY COLUMN v1
BIGINT", "Table With binlog<row>");
+ // 1b) RENAME COLUMN / REORDER COLUMNS are not allowed on row binlog
tables either.
+ // This matters for IVM: dropping a column an MV references
invalidates the IVM baseline, so
+ // every other way of changing a referenced column has to be rejected
here. If one of them
+ // were ever allowed, it would become a new way to leave an MV stale
without being noticed.
+ expectException("ALTER TABLE test." + tableName + " RENAME COLUMN v1
TO v1_renamed",
+ "Table With binlog<row>");
+ expectException("ALTER TABLE test." + tableName + " ORDER BY (k1,
v1)", "Table With binlog<row>");
+
// 2) VARIANT not supported
String createVariant = "CREATE TABLE test.binlog_variant (k1 INT NOT
NULL, v1 VARIANT) "
+ "UNIQUE KEY(k1) DISTRIBUTED BY HASH(k1) BUCKETS 1 "
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 bb484c7ce7b..73a9a092357 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
@@ -253,6 +253,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/MTMVPlanUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPlanUtilTest.java
index 162660ed8e5..c2b0f83d44f 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPlanUtilTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVPlanUtilTest.java
@@ -490,6 +490,70 @@ public class MTMVPlanUtilTest extends SqlTestBase {
incrementalCtx.getStatementContext().getIvmRewriteContext().orElseThrow().getMode());
}
+ @Test
+ public void testEnsureMTMVQueryUsableWithRowBinlogHiddenColumns() throws
Exception {
+ // A table created with row binlog carries hidden columns its query
never produces:
+ // __DORIS_COMMIT_TSO_COL__ for merge-on-write unique keys, plus
__DORIS_ROW_LSN_COL__ for
+ // duplicate keys. InternalCatalog#createOlapTable adds them while the
table is built, so an
+ // MV that enables row binlog itself ends up with them in its physical
schema.
+ // ensureMTMVQueryUsable re-derives the schema from the query and
compares the two
+ // (checkColumnIfChange), so the analyzed column list has to carry
them too -- otherwise
+ // every refresh of such an MV fails with a spurious "column length
not equals".
+ createTable("CREATE TABLE IF NOT EXISTS row_binlog_schema_base (\n"
+ + " k1 int,\n"
+ + " v1 int\n"
+ + ")\n"
+ + "DUPLICATE KEY(k1)\n"
+ + "DISTRIBUTED BY HASH(k1) BUCKETS 1\n"
+ + "PROPERTIES ('replication_num' = '1', 'binlog.enable' =
'true', 'binlog.format' = 'ROW')\n");
+
+ createMvByNereids("create materialized view row_binlog_schema_ivm "
+ + "BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL\n"
+ + " DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + " PROPERTIES ('replication_num' = '1',
'binlog.enable' = 'true', "
+ + "'binlog.format' = 'ROW') \n"
+ + " as select k1, v1 from
test.row_binlog_schema_base;");
+ createMvByNereids("create materialized view row_binlog_schema_dup "
+ + "BUILD DEFERRED REFRESH COMPLETE ON MANUAL\n"
+ + " DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + " PROPERTIES ('replication_num' = '1',
'binlog.enable' = 'true', "
+ + "'binlog.format' = 'ROW') \n"
+ + " as select k1, v1 from
test.row_binlog_schema_base;");
+ createMvByNereids("create materialized view row_binlog_schema_plain "
+ + "BUILD DEFERRED REFRESH COMPLETE ON MANUAL\n"
+ + " DISTRIBUTED BY RANDOM BUCKETS 1\n"
+ + " PROPERTIES ('replication_num' = '1') \n"
+ + " as select k1, v1 from
test.row_binlog_schema_base;");
+
+ Database db =
Env.getCurrentEnv().getInternalCatalog().getDbOrAnalysisException("test");
+ MTMV ivmMv = (MTMV)
db.getTableOrAnalysisException("row_binlog_schema_ivm");
+ MTMV dupMv = (MTMV)
db.getTableOrAnalysisException("row_binlog_schema_dup");
+ MTMV plainMv = (MTMV)
db.getTableOrAnalysisException("row_binlog_schema_plain");
+
+ Assertions.assertEquals(Lists.newArrayList(Column.COMMIT_TSO_COL),
rowBinlogHiddenColumns(ivmMv));
+ Assertions.assertEquals(Lists.newArrayList(Column.COMMIT_TSO_COL,
Column.ROW_LSN_COL),
+ rowBinlogHiddenColumns(dupMv));
+ Assertions.assertTrue(rowBinlogHiddenColumns(plainMv).isEmpty());
+
+ for (MTMV mtmv : Lists.newArrayList(ivmMv, dupMv, plainMv)) {
+ ConnectContext ctx = MTMVPlanUtil.createMTMVContext(mtmv,
+ MTMVPlanUtil.DISABLE_RULES_WHEN_GENERATE_MTMV_CACHE);
+ Assertions.assertDoesNotThrow(() ->
MTMVPlanUtil.ensureMTMVQueryUsable(mtmv, ctx),
+ "analyzed schema must match the physical schema of " +
mtmv.getName());
+ }
+ }
+
+ private static List<String> rowBinlogHiddenColumns(MTMV mtmv) {
+ List<String> hidden = Lists.newArrayList();
+ for (Column column : mtmv.getBaseSchema(true)) {
+ if (column.getName().equalsIgnoreCase(Column.COMMIT_TSO_COL)
+ || column.getName().equalsIgnoreCase(Column.ROW_LSN_COL)) {
+ hidden.add(column.getName().toUpperCase());
+ }
+ }
+ return hidden;
+ }
+
@Test
public void testEnsureMTMVQueryAnalyzeFailed() throws Exception {
createTable("CREATE TABLE IF NOT EXISTS analyze_faild_t_partition (\n"
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/MTMVTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
index 20112a936f3..123b3bcd463 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/mtmv/MTMVTaskTest.java
@@ -816,12 +816,46 @@ public class MTMVTaskTest {
Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
Deencapsulation.invoke(task, "validateIvmBaselineBeforePartitionSync",
request);
- Assertions.assertTrue((Boolean) Deencapsulation.invoke(task,
"handlePendingIvmBaselineRebuild",
- Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext()));
- Assertions.assertEquals(MTMVTask.MTMVTaskRefreshMode.NOT_REFRESH,
- Deencapsulation.getField(task, "refreshMode"));
+ List<Object> attempts = Lists.newArrayList();
+ attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request,
false));
+ Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+
+ Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+ Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext(), attempts);
+
+ // A pending COMPLETE rebuild reshapes the attempt list instead of
rebuilding inline, so
+ // PARTITIONS FALLBACK rebuilds the whole MV through the COMPLETE
attempt it keeps.
+ Assertions.assertEquals("[COMPLETE]", attempts.toString());
+ Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
+ // The barrier is released by the caller once the reshaped attempts
have run.
+ Mockito.verify(mtmv,
Mockito.never()).releaseIvmBaselineRebuild(Mockito.anyLong());
+ }
+
+ @Test
+ public void testDroppedBaselinePartitionsReleaseBarrierWithoutRebuild()
throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ IvmInfo ivmInfo = new IvmInfo();
+ ivmInfo.addPendingBaselineRebuildPartitions(Sets.newHashSet(poneName));
+ Mockito.when(mtmv.getIvmInfo()).thenReturn(ivmInfo);
+ // Partition sync already dropped the partition the barrier named, so
nothing is left to
+ // pre-rebuild and the surviving partitions catch up through the
attempts themselves.
+
Mockito.when(mtmv.getPartitionNames()).thenReturn(Sets.newHashSet(ptwoName));
+ MTMVTask task = new MTMVTask(mtmv, relation, MTMVTaskContext.of(
+ MTMVTaskTriggerMode.MANUAL, null, RefreshMode.PARTITIONS,
true, null));
+ Deencapsulation.setField(task, "mtmvSchemaChangeVersion", 7L);
+ Object request = Deencapsulation.invoke(task, "resolveRefreshRequest");
+
+ List<Object> attempts = Lists.newArrayList();
+ attempts.addAll(Deencapsulation.invoke(task, "buildAttempts", request,
false));
+ Deencapsulation.invoke(task, "handlePendingIvmBaselineRebuild",
+ Mockito.mock(MTMVRefreshContext.class), request, new
ConnectContext(), attempts);
+
+ Assertions.assertEquals("[PARTITIONS, COMPLETE]", attempts.toString());
+ Assertions.assertNull(Deencapsulation.getField(task, "refreshMode"));
Assertions.assertEquals(IvmFailureReason.BINLOG_BROKEN.name(),
Deencapsulation.getField(task, "ivmFallbackReason"));
+ Mockito.verify(mtmv).releaseIvmBaselineRebuild(7L);
}
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
index 2588ca5f532..dfd3d6dc7ef 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmBaselineRebuildTest.java
@@ -104,6 +104,24 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertTrue(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
+ @Test
+ public void testDropColumnMarksBaselineRebuildOnlyWhenReferenced() throws
Exception {
+ String db = "ivm_broken_drop_column";
+ createPartitionedIvmTableAndMv(db);
+ MTMV mtmv = getMtmv(db);
+
+ // ivm_mv selects dt, k1, v1. Dropping a column it does not use must
leave the baseline alone.
+ executeSql("ALTER TABLE ivm_base ADD COLUMN spare int");
+ executeSql("ALTER TABLE ivm_base DROP COLUMN spare");
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+
+ // Dropping a column the MV uses makes the MV query unanalyzable: the
change is metadata-only
+ // and emits no binlog, so an incremental refresh would silently keep
the rows of the old
+ // column. The baseline has to be invalidated instead.
+ executeSql("ALTER TABLE ivm_base DROP COLUMN v1");
+ Assertions.assertTrue(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ }
+
@Test
public void testPublishedPctPartitionUsesPartitionsBaselineRebuild()
throws Exception {
String db = "ivm_partitions_baseline_rebuild";
@@ -217,6 +235,24 @@ public class IvmBaselineRebuildTest extends
TestWithFeService {
Assertions.assertFalse(getMtmv(db).getIvmInfo().isBaselineRebuildRequired());
}
+ @Test
+ public void testRenameTableBackKeepsIncrementalRefreshStartable() throws
Exception {
+ String db = "ivm_broken_rename_table_back";
+ createPartitionedIvmTableAndMv(db);
+
+ executeSql("ALTER TABLE ivm_base RENAME ivm_base_renamed");
+ executeSql("ALTER TABLE ivm_base_renamed RENAME ivm_base");
+
+ // A rename changes no column, so it must not invalidate the baseline
in either direction:
+ // once the table is renamed back, the MV query is analyzable again
and a strict INCREMENTAL
+ // refresh has to be able to start. A "baseline rebuild required" flag
left behind by the
+ // rename would reject every one of them until a COMPLETE refresh had
been run, even though
+ // nothing the MV depends on ever changed.
+ MTMV mtmv = getMtmv(db);
+ Assertions.assertFalse(mtmv.getIvmInfo().isBaselineRebuildRequired());
+ Assertions.assertDoesNotThrow(() ->
mtmv.validateIvmRefreshStart(mtmv.getSchemaChangeVersion()));
+ }
+
@Test
public void testReplaceTableMarksBaselineRebuild() throws Exception {
String db = "ivm_broken_replace_table";
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_drop_referenced_column_baseline_rebuild.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.out
new file mode 100644
index 00000000000..8793c2b56bc
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !mv_rows_baseline --
+10 2 300
+20 1 300
+
+-- !mv_rows_after_aba --
+0 3 600
+
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
new file mode 100644
index 00000000000..271a5f56d8e
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.out
@@ -0,0 +1,32 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS NONE NONE
+
+-- !baseline_base --
+2026-01-10 1 10
+2026-01-10 1 10
+2026-02-10 3 30
+2026-02-10 3 30
+
+-- !baseline_mv --
+2026-01-10 1 10
+2026-01-10 1 10
+2026-02-10 3 30
+2026-02-10 3 30
+
+-- !strict_task --
+FAILED NOT_REFRESH BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS PARTIAL BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10 3 30
+2026-02-10 3 30
+2026-02-15 4 40
+
+-- !fallback_mv --
+2026-02-10 3 30
+2026-02-10 3 30
+2026-02-15 4 40
+
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
new file mode 100644
index 00000000000..fab4869556a
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.out
@@ -0,0 +1,45 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS NONE
+
+-- !baseline_base --
+2026-01-10 1 10
+2026-02-10 2 20
+2026-03-10 3 30
+
+-- !baseline_mv --
+2026-01-10 1 10
+2026-02-10 2 20
+2026-03-10 3 30
+
+-- !strict_task --
+FAILED BINLOG_BROKEN
+
+-- !fallback_task --
+SUCCESS BINLOG_BROKEN
+
+-- !fallback_base --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+
+-- !fallback_mv --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+
+-- !resumed_task --
+SUCCESS NONE
+
+-- !resumed_base --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+2026-03-15 5 50
+
+-- !resumed_mv --
+2026-02-10 2 20
+2026-02-15 4 40
+2026-03-10 3 30
+2026-03-15 5 50
+
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/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
index 1208b149aa3..686e463d2f3 100644
--- a/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_refresh_dry_run.out
@@ -5,8 +5,8 @@
3 1 30
-- !ivm_dry_run_full --
-1 1 1 15 1 1 4097 0
-4 4 1 40 1 1 4097 0
+1 1 1 15 1 1 [regression-fake-sequence] 0
+4 4 1 40 1 1 [regression-fake-sequence] 0
-- !ivm_dry_run_after --
1 1 10
@@ -14,8 +14,8 @@
3 1 30
-- !ivm_dry_run_repeat --
-1 1 1 15 1 1 4097 0
-4 4 1 40 1 1 4097 0
+1 1 1 15 1 1 [regression-fake-sequence] 0
+4 4 1 40 1 1 [regression-fake-sequence] 0
-- !ivm_dry_run_after_refresh --
1 1 15
@@ -23,3 +23,7 @@
3 1 30
4 1 40
+-- !ivm_dry_run_empty --
+
+-- !ivm_dry_run_empty_repeat --
+
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_row_binlog_schema_validation.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_row_binlog_schema_validation.out
new file mode 100644
index 00000000000..9d9acb797cb
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_row_binlog_schema_validation.out
@@ -0,0 +1,27 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !rb_ivm_rows --
+1 10
+2 20
+
+-- !rb_ivm_no_binlog_rows --
+1 10
+2 20
+
+-- !rb_dup_rows --
+1 10
+2 20
+
+-- !rb_dup_no_binlog_rows --
+1 10
+2 20
+
+-- !cascade_l2_baseline --
+10 2 300
+20 1 300
+
+-- !cascade_l1_after_aba --
+0 3 600
+
+-- !cascade_l2_after_aba --
+0 3 600
+
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index d906c610ed8..6ef166d9459 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -1727,8 +1727,10 @@ class Suite implements GroovyInterceptable {
return result
}
- // rowConverter: { row -> convertedRow }
+ // rowConverter: { row -> convertedRow }, or { row, meta -> convertedRow }
to inspect the result
+ // metadata, for example to mask a column whose value depends on the
deployment mode.
void quickRunTest(String tag, Object arg, boolean isOrder = false, Closure
rowConverter = null) {
+ boolean converterNeedsMeta = rowConverter != null &&
rowConverter.maximumNumberOfParameters > 1
if (context.config.generateOutputFile ||
context.config.forceGenerateOutputFile) {
Tuple2<List<List<Object>>, ResultSetMetaData> tupleResult = null
if (arg instanceof PreparedStatement) {
@@ -1763,7 +1765,9 @@ class Suite implements GroovyInterceptable {
}
def (result, meta) = tupleResult
if (rowConverter != null) {
- result = result.collect { rowConverter.call(it) }
+ result = result.collect {
+ converterNeedsMeta ? rowConverter.call(it, meta) :
rowConverter.call(it)
+ }
}
if (isOrder) {
result = sortByToString(result)
@@ -1815,7 +1819,9 @@ class Suite implements GroovyInterceptable {
}
def (realResults, meta) = tupleResult
if (rowConverter != null) {
- realResults = realResults.collect { rowConverter.call(it) }
+ realResults = realResults.collect {
+ converterNeedsMeta ? rowConverter.call(it, meta) :
rowConverter.call(it)
+ }
}
if (isOrder) {
realResults = sortByToString(realResults)
diff --git a/regression-test/plugins/plugin_planner.groovy
b/regression-test/plugins/plugin_planner.groovy
index 0a5a12c402a..eca9594bd2d 100644
--- a/regression-test/plugins/plugin_planner.groovy
+++ b/regression-test/plugins/plugin_planner.groovy
@@ -135,3 +135,24 @@ Suite.metaClass.explainIvmPlan = { String tag, String sql
->
)
}
+// __DORIS_SEQUENCE_COL__ of an IVM dry-run delta encodes the refresh version,
whose base value
+// differs between cloud and shared-nothing deployments. Mask that column so
that both modes can
+// share one suite and one .out file.
+def maskIvmDryRunSequence = { row, meta ->
+ for (int i = 1; i <= meta.getColumnCount(); i++) {
+ if ("__DORIS_SEQUENCE_COL__".equalsIgnoreCase(meta.getColumnLabel(i)))
{
+ def masked = new ArrayList(row)
+ masked.set(i - 1, "[regression-fake-sequence]")
+ return masked
+ }
+ }
+ return row
+}
+
+// Named ivm_dry_run_qt instead of order_qt_*: Suite.invokeMethod intercepts
every method whose
+// name starts with qt_ / order_qt_ and derives the tag from the method name,
so a metaClass
+// method could never be reached under those prefixes.
+Suite.metaClass.ivm_dry_run_qt = { String tag, String sql ->
+ delegate.quickRunTest(tag, sql, true, maskIvmDryRunSequence)
+}
+
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..db9d5be3052 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,11 +18,7 @@
import org.awaitility.Awaitility
import static java.util.concurrent.TimeUnit.SECONDS
-suite("test_ivm_drop_column_fallback_reason", "nonConcurrent") {
- if (isCloudMode()) {
- return
- }
-
+suite("test_ivm_drop_column_fallback_reason") {
sql """DROP MATERIALIZED VIEW IF EXISTS ivm_drop_col_reason_mv"""
sql """DROP TABLE IF EXISTS ivm_drop_col_reason_t"""
sql """
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
new file mode 100644
index 00000000000..26dd37ff2d0
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_drop_referenced_column_baseline_rebuild.groovy
@@ -0,0 +1,144 @@
+// 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.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+// Dropping a column that an IVM references, then re-adding a column with the
same name
+// (schema ABA), used to let a strict INCREMENTAL refresh report SUCCESS while
silently
+// keeping the rows computed under the old column epoch.
+//
+// The base table change is metadata-only (light schema change) and emits no
binlog, so the
+// delta is empty and the refresh has nothing to apply -- the MV baseline is
simply stale.
+//
+// Expected: dropping a referenced column invalidates the IVM baseline, so a
strict
+// INCREMENTAL refresh is rejected and the user is told to run a COMPLETE
refresh.
+// Dropping an unreferenced column must still leave the incremental path
untouched.
+suite("test_ivm_drop_referenced_column_baseline_rebuild") {
+ def tableName = "ivm_drop_ref_col_t"
+ def mvName = "ivm_drop_ref_col_mv"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+
+ sql """
+ CREATE TABLE ${tableName} (
+ id BIGINT NOT NULL,
+ grp INT NULL,
+ amount BIGINT NULL,
+ spare INT NULL
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(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 ${tableName} VALUES (1, 10, 100, 7), (2, 10, 200, 8),
(3, 20, 300, 9)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(grp)
+ DISTRIBUTED BY HASH(grp) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT grp, COUNT(*) AS cnt, SUM(amount) AS total
+ FROM ${tableName} GROUP BY grp
+ """
+
+ def ddlJobCount = { String table ->
+ return sql("""SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'""").size()
+ }
+
+ // `SHOW ALTER TABLE COLUMN` keeps finished jobs, so wait for a *new* job
that is FINISHED.
+ def waitDdlFinished = { String table, int previousJobCount ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def jobs = sql """SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'"""
+ return jobs.size() > previousJobCount
+ && jobs.every({ row -> row[9].toString() == 'FINISHED' })
+ })
+ }
+
+ def lastTaskId = null
+ // tasks('type'='mv') can briefly miss the just-finished task, so require
a *new* TaskId.
+ def waitTerminalTask = { String mv ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status, RefreshMode, IvmFallbackReason, ErrorMsg
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mv}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != lastTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ lastTaskId = taskResult[0].TaskId.toString()
+ return taskResult[0]
+ }
+
+ // ---------------------------------------------------------------- 1.
baseline
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ def task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(), "baseline COMPLETE
refresh: " + task.ErrorMsg)
+ order_qt_mv_rows_baseline "SELECT grp, cnt, total FROM ${mvName}"
+
+ // ------------------------------------- 2. unreferenced column: no
baseline invalidation
+ def before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} DROP COLUMN spare"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(),
+ "dropping an unreferenced column must not invalidate the IVM
baseline: " + task.ErrorMsg)
+
+ // ---------------------------------------- 3. referenced column: strict
INCREMENTAL rejected
+ before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} DROP COLUMN grp"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("FAILED", task.Status.toString(),
+ "dropping a referenced column must reject a strict INCREMENTAL
refresh")
+
+ // -------------------------------- 4. same-name re-add (schema ABA) is
still rejected
+ before = ddlJobCount(tableName)
+ sql """ALTER TABLE ${tableName} ADD COLUMN grp INT NULL DEFAULT '0'"""
+ waitDdlFinished(tableName, before)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ task = waitTerminalTask(mvName)
+ assertEquals("FAILED", task.Status.toString(),
+ "schema ABA must not be silently accepted by a strict INCREMENTAL
refresh")
+ assertTrue(task.ErrorMsg.toString().contains("baseline rebuild is
pending"),
+ "expected a pending baseline rebuild hint, got: " + task.ErrorMsg)
+
+ // ------------------------------------ 5. COMPLETE rebuild reflects
current base semantics
+ // Every pre-existing row now reads the re-added column's default value.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ task = waitTerminalTask(mvName)
+ assertEquals("SUCCESS", task.Status.toString(), "COMPLETE rebuild after
ABA: " + task.ErrorMsg)
+ order_qt_mv_rows_after_aba "SELECT grp, cnt, total FROM ${mvName}"
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch.groovy
index 3d15539f795..dab63c2e2b3 100644
---
a/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch.groovy
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch.groovy
@@ -16,10 +16,6 @@
// under the License.
suite("test_ivm_fallback_stream_multi_batch", "nonConcurrent") {
- if (isCloudMode()) {
- return
- }
-
def forcedFallbackDebugPoint =
"IvmIncrRefreshManager.doRefresh.force_fallback_reason"
GetDebugPoint().disableDebugPointForAllFEs(forcedFallbackDebugPoint)
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch_dup.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch_dup.groovy
index 7e6480c5ef5..18bbd652e57 100644
---
a/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch_dup.groovy
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_fallback_stream_multi_batch_dup.groovy
@@ -16,10 +16,6 @@
// under the License.
suite("test_ivm_fallback_stream_multi_batch_dup", "nonConcurrent") {
- if (isCloudMode()) {
- return
- }
-
def forcedFallbackDebugPoint =
"IvmIncrRefreshManager.doRefresh.force_fallback_reason"
GetDebugPoint().disableDebugPointForAllFEs(forcedFallbackDebugPoint)
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..7c51417417c 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,11 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-suite("test_ivm_mtmv_row_binlog", "nonConcurrent") {
- if (isCloudMode()) {
- return
- }
-
+suite("test_ivm_mtmv_row_binlog") {
sql """DROP MATERIALIZED VIEW IF EXISTS test_ivm_mtmv_row_binlog_mv;"""
sql """DROP TABLE IF EXISTS test_ivm_mtmv_row_binlog_base;"""
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_baseline_rebuild_dup_keys.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy
new file mode 100644
index 00000000000..8f849196d47
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_baseline_rebuild_dup_keys.groovy
@@ -0,0 +1,123 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * Same baseline-rebuild pre-step as test_ivm_partition_drop_live_delta, but
on the branch where the
+ * affected MV partition SURVIVES: TRUNCATE keeps the partition range, so
partition sync leaves the
+ * MV partition in place and the pre-step really has something to rebuild.
+ *
+ * <p>Two things are pinned here. The refreshed partition is picked up again
by the IVM attempt that
+ * follows, which may only apply the remaining delta -- on a duplicate-key MV
a double apply shows up
+ * as extra copies of the same row, not as a wrong value. And the row written
to the surviving
+ * partition after the truncate must still be consumed. Both are checked by
comparing whole result
+ * sets, so row multiplicities are part of the expectation.
+ */
+suite("test_ivm_partition_baseline_rebuild_dup_keys") {
+ def tableName = "ivm_part_dup_t"
+ def mvName = "ivm_part_dup_mv"
+
+ def waitForNewTask = { previousTaskId ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != previousTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[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}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ dt DATE NOT NULL,
+ id INT NOT NULL,
+ v INT
+ )
+ DUPLICATE KEY(dt, id)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW"
+ )
+ """
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+ // Repeated identical rows: a double-applied delta grows the multiplicity
instead of hiding in a
+ // unique key.
+ sql """INSERT INTO ${tableName} VALUES
+ ('2026-01-10', 1, 10), ('2026-01-10', 1, 10),
+ ('2026-02-10', 3, 30), ('2026-02-10', 3, 30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT dt, id, v FROM ${tableName}
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def taskId = waitForNewTask(null)
+ qt_baseline_task taskQuery(taskId)
+ order_qt_baseline_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id, v"""
+ order_qt_baseline_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id,
v"""
+
+ // TRUNCATE replaces the partition, so the MV partition of that range
stays alive and the
+ // baseline pre-step has a real partition to rebuild.
+ sql """TRUNCATE TABLE ${tableName} PARTITION(p202601)"""
+ sql """INSERT INTO ${tableName} VALUES ('2026-02-15', 4, 40)"""
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_strict_task taskQuery(taskId)
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL FALLBACK"""
+ taskId = waitForNewTask(taskId)
+ qt_fallback_task taskQuery(taskId)
+ order_qt_fallback_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id, v"""
+ order_qt_fallback_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id,
v"""
+}
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
new file mode 100644
index 00000000000..bc9f182e306
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_partition_drop_live_delta.groovy
@@ -0,0 +1,134 @@
+// 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.
+
+import org.awaitility.Awaitility
+
+import static java.util.concurrent.TimeUnit.SECONDS
+
+/**
+ * Dropping a base-table partition invalidates the IVM baseline, because the
rows disappear through
+ * metadata rather than through row binlog entries. The MV partition built
from that base partition
+ * is then removed by partition sync, which is exactly what the baseline
barrier recorded.
+ *
+ * <p>The refresh must still consume the delta that accumulated on the
*surviving* partitions: it
+ * may not report SUCCESS while leaving those partitions stale. This case
inserts a row into a
+ * surviving partition after the drop, so an EMPTY baseline-rebuild
intersection cannot be mistaken
+ * for "nothing to do".
+ *
+ * <p>Partitions are managed by hand (no dynamic partition scheduler) and
every dt is a literal, so
+ * the case is fully deterministic.
+ */
+suite("test_ivm_partition_drop_live_delta") {
+ def tableName = "ivm_part_drop_t"
+ def mvName = "ivm_part_drop_mv"
+
+ def waitForNewTask = { previousTaskId ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}'
+ AND MvName = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != previousTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[0].TaskId.toString()
+ }
+
+ // An unset IvmFallbackReason comes 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 IvmFallbackReason = 'BINLOG_BROKEN' THEN
IvmFallbackReason ELSE 'NONE' END
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mvName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """
+ CREATE TABLE ${tableName} (
+ dt DATE NOT NULL,
+ id INT NOT NULL,
+ v INT
+ )
+ UNIQUE KEY(dt, id)
+ PARTITION BY RANGE(dt) ()
+ DISTRIBUTED BY HASH(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 ${tableName} ADD PARTITION p202601 VALUES
[('2026-01-01'), ('2026-02-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202602 VALUES
[('2026-02-01'), ('2026-03-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202603 VALUES
[('2026-03-01'), ('2026-04-01'))"""
+ sql """ALTER TABLE ${tableName} ADD PARTITION p202604 VALUES
[('2026-04-01'), ('2026-05-01'))"""
+ sql """INSERT INTO ${tableName} VALUES
+ ('2026-01-10', 1, 10), ('2026-02-10', 2, 20), ('2026-03-10', 3,
30)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL FALLBACK ON MANUAL
+ KEY(dt, id)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT dt, id, v FROM ${tableName}
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def taskId = waitForNewTask(null)
+ qt_baseline_task taskQuery(taskId)
+ order_qt_baseline_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_baseline_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+
+ sql """ALTER TABLE ${tableName} DROP PARTITION p202601"""
+ sql """INSERT INTO ${tableName} VALUES ('2026-02-15', 4, 40)"""
+
+ // A strict incremental refresh must refuse to run against a broken
baseline.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_strict_task taskQuery(taskId)
+
+ // The fallback reports SUCCESS, so the MV has to match the base table
afterwards: the expired
+ // partition is gone AND the row written to the surviving partition has
been consumed. An MV
+ // that is missing that row means the refresh silently skipped the
surviving partitions' delta.
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL FALLBACK"""
+ taskId = waitForNewTask(taskId)
+ qt_fallback_task taskQuery(taskId)
+ order_qt_fallback_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_fallback_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+
+ // A following strict incremental refresh must be able to continue from
the repaired baseline.
+ sql """INSERT INTO ${tableName} VALUES ('2026-03-15', 5, 50)"""
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ taskId = waitForNewTask(taskId)
+ qt_resumed_task taskQuery(taskId)
+ order_qt_resumed_base """SELECT dt, id, v FROM ${tableName} ORDER BY dt,
id"""
+ order_qt_resumed_mv """SELECT dt, id, v FROM ${mvName} ORDER BY dt, id"""
+}
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_refresh_dry_run.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
index 5cb89692a6f..f2583e14e8d 100644
--- a/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
+++ b/regression-test/suites/mtmv_p0/ivm/test_ivm_refresh_dry_run.groovy
@@ -16,14 +16,9 @@
// under the License.
suite("test_ivm_refresh_dry_run") {
- // Cloud mode: __DORIS_SEQUENCE_COL__ in the dry-run delta rows derives
from cloud txn
- // versioning and differs from local (e.g. 6145 vs 4097), so the .out
values
- // for the sequence column do not apply.
- if (isCloudMode()) {
- logger.info("skip test_ivm_refresh_dry_run on cloud mode: " +
- "__DORIS_SEQUENCE_COL__ differs between cloud and local")
- return
- }
+ // __DORIS_SEQUENCE_COL__ of the dry-run delta rows encodes the refresh
version, whose base
+ // value differs between cloud and shared-nothing deployments (e.g. 6145
vs 4097).
+ // ivm_dry_run_qt masks that column, so cloud and local share this suite
and its .out.
sql "DROP MATERIALIZED VIEW IF EXISTS test_ivm_refresh_dry_run_mv"
sql "DROP TABLE IF EXISTS test_ivm_refresh_dry_run_base"
@@ -65,9 +60,9 @@ suite("test_ivm_refresh_dry_run") {
order_qt_ivm_dry_run_before "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
- order_qt_ivm_dry_run_full """
+ ivm_dry_run_qt("ivm_dry_run_full", """
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
- """
+ """)
// The delta rows picked by a LIMIT depend on scan order (no ORDER BY
before the cap),
// so assert only the returned row count instead of exact rows in the .out
file.
@@ -77,11 +72,21 @@ suite("test_ivm_refresh_dry_run") {
order_qt_ivm_dry_run_after "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
- order_qt_ivm_dry_run_repeat """
+ ivm_dry_run_qt("ivm_dry_run_repeat", """
REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN LIMIT 10
- """
+ """)
sql "REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL"
waitingMTMVTaskFinishedByMvName("test_ivm_refresh_dry_run_mv")
order_qt_ivm_dry_run_after_refresh "SELECT k1, cnt, sum_v1 FROM
test_ivm_refresh_dry_run_mv"
+
+ // Empty delta: the delta query is a LogicalEmptyRelation and the dry run
must return an
+ // empty result instead of sending fragments to a placeholder backend.
Every dry-run
+ // comparison goes through ivm_dry_run_qt so the sequence column is masked
in all of them.
+ ivm_dry_run_qt("ivm_dry_run_empty", """
+ REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
+ """)
+ ivm_dry_run_qt("ivm_dry_run_empty_repeat", """
+ REFRESH MATERIALIZED VIEW test_ivm_refresh_dry_run_mv INCREMENTAL WITH
DRY RUN
+ """)
}
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_row_binlog_schema_validation.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_row_binlog_schema_validation.groovy
new file mode 100644
index 00000000000..b93dff85fdd
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_row_binlog_schema_validation.groovy
@@ -0,0 +1,194 @@
+// 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.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+// A materialized view re-validates its schema against a fresh analysis of its
own query whenever a
+// base table changes (MTMVPlanUtil.ensureMTMVQueryUsable ->
checkColumnIfChange). A table created
+// with row binlog carries hidden columns that the query never produces, and
an MV carries them too
+// when its own properties enable row binlog (which cascade IVM requires). The
analyzed schema has to
+// contain them as well, otherwise every refresh of such an MV fails with
+// "column length not equals, please check whether columns of base table have
changed" -- including a
+// COMPLETE refresh, which is the only way out of a stale baseline.
+//
+// Covered here: the four MV shapes, and the DORIS-28306 shape where a cascade
L1 could not recover
+// through COMPLETE after a schema ABA on a referenced column.
+suite("test_ivm_row_binlog_schema_validation") {
+ def rowBinlogProps = "'replication_num' = '1', 'binlog.enable' = 'true',
'binlog.format' = 'ROW'"
+ def plainProps = "'replication_num' = '1'"
+ // A cascade source keeps historical values so its own downstream MV can
read the binlog.
+ def cascadeProps = rowBinlogProps + ", 'binlog.need_historical_value' =
'true'"
+
+ def ddlJobCount = { String table ->
+ return sql("""SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'""").size()
+ }
+
+ // `SHOW ALTER TABLE COLUMN` keeps finished jobs, so wait for a *new* job
that is FINISHED.
+ def waitDdlFinished = { String table, int previousJobCount ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ def jobs = sql """SHOW ALTER TABLE COLUMN WHERE TableName =
'${table}'"""
+ return jobs.size() > previousJobCount
+ && jobs.every({ row -> row[9].toString() == 'FINISHED' })
+ })
+ }
+
+ def lastTaskId = null
+ // tasks('type'='mv') can briefly miss the just-finished task, so require
a *new* TaskId.
+ def waitTerminalTask = { String mv ->
+ def taskResult
+ Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
+ taskResult = sql_return_maparray("""
+ SELECT TaskId, Status, RefreshMode, ErrorMsg
+ FROM tasks('type'='mv')
+ WHERE MvDatabaseName = '${context.dbName}' AND MvName = '${mv}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != lastTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ lastTaskId = taskResult[0].TaskId.toString()
+ return taskResult[0]
+ }
+
+ def refresh = { String mv, String mode ->
+ sql """REFRESH MATERIALIZED VIEW ${mv} ${mode}"""
+ return waitTerminalTask(mv)
+ }
+
+ // ---------------------------------------------------------------- 1.
four MV shapes
+ // Same query and same base table for all four; what differs is whether
the MV itself enables
+ // row binlog, and (for the IVM ones) whether the MV is a MOW unique table.
+ def cases = [
+ [name: "rb_ivm", mvProps: rowBinlogProps, ivm: true],
+ [name: "rb_ivm_no_binlog", mvProps: plainProps, ivm: true],
+ [name: "rb_dup", mvProps: rowBinlogProps, ivm: false],
+ [name: "rb_dup_no_binlog", mvProps: plainProps, ivm: false],
+ ]
+
+ for (def c : cases) {
+ def table = c.name + "_base"
+ def mv = c.name + "_mv"
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ${mv}"""
+ sql """DROP TABLE IF EXISTS ${table}"""
+ sql """
+ CREATE TABLE ${table} (
+ k1 INT NOT NULL,
+ v1 INT NULL,
+ spare INT NULL
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) 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 ${table} VALUES (1, 10, 1), (2, 20, 1)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mv}
+ BUILD DEFERRED REFRESH ${c.ivm ? 'INCREMENTAL' : 'COMPLETE'} ON
MANUAL
+ KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (${c.mvProps})
+ AS SELECT k1, SUM(v1) AS total FROM ${table} GROUP BY k1
+ """
+ def task = refresh(mv, "COMPLETE")
+ assertEquals("SUCCESS", task.Status.toString(), "${c.name} baseline
refresh: " + task.ErrorMsg)
+
+ // A schema change that leaves the query intact: the MV must stay
refreshable. The dropped
+ // column is not referenced, so nothing about the MV's result changes.
+ def before = ddlJobCount(table)
+ sql """ALTER TABLE ${table} DROP COLUMN spare"""
+ waitDdlFinished(table, before)
+
+ task = refresh(mv, c.ivm ? "INCREMENTAL" : "COMPLETE")
+ assertEquals("SUCCESS", task.Status.toString(),
+ "${c.name} refresh after an unrelated DROP COLUMN: " +
task.ErrorMsg)
+ }
+
+ // The dropped column was never referenced, so all four MVs must still
hold their baseline rows.
+ order_qt_rb_ivm_rows "SELECT k1, total FROM rb_ivm_mv"
+ order_qt_rb_ivm_no_binlog_rows "SELECT k1, total FROM rb_ivm_no_binlog_mv"
+ order_qt_rb_dup_rows "SELECT k1, total FROM rb_dup_mv"
+ order_qt_rb_dup_no_binlog_rows "SELECT k1, total FROM rb_dup_no_binlog_mv"
+
+ // ------------------------------------- 2. DORIS-28306: cascade L1 must
recover via COMPLETE
+ // A cascade forces row binlog onto L1, so L1 is exactly the shape that
used to make every
+ // refresh of L1 fail after a schema change. An ABA on a referenced column
must still leave the
+ // explicit COMPLETE recovery path usable.
+ def cTable = "cascade_aba_base"
+ sql """DROP MATERIALIZED VIEW IF EXISTS cascade_l2"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS cascade_l1"""
+ sql """DROP TABLE IF EXISTS ${cTable}"""
+ sql """
+ CREATE TABLE ${cTable} (
+ id BIGINT NOT NULL,
+ grp INT NULL,
+ amount BIGINT NULL
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(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 ${cTable} VALUES (1, 10, 100), (2, 10, 200), (3, 20,
300)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW cascade_l1
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL KEY(id)
+ PROPERTIES (${cascadeProps})
+ AS SELECT id, grp, amount FROM ${cTable}
+ """
+ def cTask = refresh("cascade_l1", "COMPLETE")
+ assertEquals("SUCCESS", cTask.Status.toString(), "cascade L1 baseline: " +
cTask.ErrorMsg)
+
+ sql """
+ CREATE MATERIALIZED VIEW cascade_l2
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL KEY(grp)
+ PROPERTIES (${cascadeProps})
+ AS SELECT grp, COUNT(*) AS row_count, SUM(amount) AS total_amount FROM
cascade_l1 GROUP BY grp
+ """
+ cTask = refresh("cascade_l2", "COMPLETE")
+ assertEquals("SUCCESS", cTask.Status.toString(), "cascade L2 baseline: " +
cTask.ErrorMsg)
+ order_qt_cascade_l2_baseline "SELECT grp, row_count, total_amount FROM
cascade_l2"
+
+ def cBefore = ddlJobCount(cTable)
+ sql """ALTER TABLE ${cTable} DROP COLUMN grp"""
+ waitDdlFinished(cTable, cBefore)
+ cBefore = ddlJobCount(cTable)
+ sql """ALTER TABLE ${cTable} ADD COLUMN grp INT NULL DEFAULT '0'"""
+ waitDdlFinished(cTable, cBefore)
+
+ // COMPLETE is the recovery path: it has to rebuild L1 from the current
base-table semantics,
+ // where every pre-existing row now reads the re-added column's default
value. Whether the
+ // strict INCREMENTAL in between is rejected is a separate contract, owned
by the suite for the
+ // baseline invalidation itself, so it is deliberately not asserted here
-- this suite has to
+ // hold with or without that change.
+ cTask = refresh("cascade_l1", "COMPLETE")
+ assertEquals("SUCCESS", cTask.Status.toString(), "COMPLETE after schema
ABA: " + cTask.ErrorMsg)
+ order_qt_cascade_l1_after_aba "SELECT grp, COUNT(*) AS c, SUM(amount) AS s
FROM cascade_l1 GROUP BY grp"
+
+ // ... and the downstream level stays refreshable afterwards.
+ cTask = refresh("cascade_l2", "COMPLETE")
+ assertEquals("SUCCESS", cTask.Status.toString(), "cascade L2 after L1
recovery: " + cTask.ErrorMsg)
+ order_qt_cascade_l2_after_aba "SELECT grp, row_count, total_amount FROM
cascade_l2"
+}
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]
