This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c1bff0d27c2 [fix](ivm) Fail aggregate IVM refresh when the previous
refresh txn is not visible yet (#67646)
c1bff0d27c2 is described below
commit c1bff0d27c2b1327de70ac583d322db7561878de
Author: yujun <[email protected]>
AuthorDate: Thu Sep 10 11:08:46 2026 +0800
[fix](ivm) Fail aggregate IVM refresh when the previous refresh txn is not
visible yet (#67646)
This PR contains two follow-ups from the #62606 review round:
1. **Derive IVM hidden column names from `IVM_HIDDEN_COLUMN_PREFIX`** —
IVM hidden column names are now derived from the shared prefix constant
instead of duplicated literals.
2. **Fail aggregate IVM refresh when the previous refresh txn is not
visible yet** — an aggregate delta joins the MV's old rows with the new
delta; when the previous refresh txn committed but its data is not
visible yet (an MV partition's committed version is ahead of its visible
version), that join misses the old rows and permanently loses the delta.
The delta rewriter now fails a non-empty aggregate delta with
`MV_COMMIT_NOT_VISIBLE` and the refresh falls back to a rebuild that
recomputes from the base tables without reading old MV state.
Adds a nonConcurrent regression that holds a refresh txn in COMMITTED
via the new `DatabaseTransactionMgr.finishTransaction.block_visible`
debug point, then verifies the next strict incremental refresh fails
while a COMPLETE refresh can still report SUCCESS with its txn not yet
visible, and that everything converges once the stuck txns publish.
Trace issue: https://github.com/apache/doris/issues/65418
---
.../main/java/org/apache/doris/catalog/Column.java | 12 +-
.../apache/doris/mtmv/ivm/IvmDeltaRewriter.java | 35 ++++
.../apache/doris/mtmv/ivm/IvmFailureReason.java | 2 +
.../doris/mtmv/ivm/IvmIncrRefreshManager.java | 2 +-
.../apache/doris/mtmv/ivm/IvmRewriteContext.java | 69 +++++--
.../trees/plans/commands/RefreshMTMVCommand.java | 7 +-
.../doris/transaction/DatabaseTransactionMgr.java | 23 +++
.../doris/mtmv/ivm/IvmAggDeltaHandlerTest.java | 6 +-
.../doris/mtmv/ivm/IvmDeltaRewriterTest.java | 15 +-
.../doris/mtmv/ivm/IvmJoinDeltaHandlerTest.java | 2 +-
.../doris/mtmv/ivm/IvmLinearDeltaHandlerTest.java | 2 +-
.../rules/analysis/IvmIncrRefreshMTMVTest.java | 18 +-
.../rules/analysis/IvmNormalizeMTMVTest.java | 6 +-
.../test_ivm_agg_previous_commit_not_visible.out | 26 +++
...test_ivm_agg_previous_commit_not_visible.groovy | 219 +++++++++++++++++++++
15 files changed, 394 insertions(+), 50 deletions(-)
diff --git a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
index 593a853b235..e7a22663d48 100644
--- a/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-catalog/src/main/java/org/apache/doris/catalog/Column.java
@@ -51,17 +51,16 @@ public class Column implements GsonPostProcessable {
public static final String HIDDEN_COLUMN_PREFIX = "__DORIS_";
// all shadow indexes should have this prefix in name
public static final String SHADOW_NAME_PREFIX = "__doris_shadow_";
- public static final String IVM_HIDDEN_COLUMN_PREFIX = "__DORIS_IVM_";
// NOTE: you should name hidden column start with '__DORIS_'
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
public static final String DELETE_SIGN = "__DORIS_DELETE_SIGN__";
public static final String WHERE_SIGN = "__DORIS_WHERE_SIGN__";
public static final String SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
public static final String GLOBAL_ROWID_COL = "__DORIS_GLOBAL_ROWID_COL__";
- public static final String IVM_ROW_ID_COL = "__DORIS_IVM_ROW_ID_COL__";
- public static final String IVM_AGG_COUNT_COL =
"__DORIS_IVM_AGG_COUNT_COL__";
- public static final String IVM_DML_FACTOR_COL =
"__DORIS_IVM_DML_FACTOR_COL__";
- public static final String IVM_BASE_OP_COL = "__DORIS_IVM_BASE_OP_COL__";
- public static final String IVM_DELTA_GROUP_COUNT_COL =
"__DORIS_IVM_DELTA_GROUP_COUNT_COL__";
+ public static final String IVM_HIDDEN_COLUMN_PREFIX = HIDDEN_COLUMN_PREFIX
+ "IVM_";
+ public static final String IVM_ROW_ID_COL = IVM_HIDDEN_COLUMN_PREFIX +
"ROW_ID_COL__";
+ public static final String IVM_AGG_COUNT_COL = IVM_HIDDEN_COLUMN_PREFIX +
"AGG_COUNT_COL__";
+ public static final String IVM_DML_FACTOR_COL = IVM_HIDDEN_COLUMN_PREFIX +
"DML_FACTOR_COL__";
+ public static final String IVM_DELTA_GROUP_COUNT_COL =
IVM_HIDDEN_COLUMN_PREFIX + "DELTA_GROUP_COUNT_COL__";
// Prefix for sink-level IVM identity key hidden columns
(__DORIS_IVM_KEY_).
public static final String IVM_KEY_COL_PREFIX = IVM_HIDDEN_COLUMN_PREFIX +
"KEY_";
// Prefix for union arm-index columns (__DORIS_IVM_UNION_ARM_INDEX_).
@@ -301,7 +300,6 @@ public class Column implements GsonPostProcessable {
false, null, null, Sets.newHashSet(), null);
}
-
public Column(String name, Type type, boolean isKey, AggregateType
aggregateType, boolean isAllowNull,
String defaultValue, String comment, boolean visible, int
colUniqueId) {
this(name, type, isKey, aggregateType, isAllowNull, -1, defaultValue,
comment, visible, null, colUniqueId, null,
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 59fb144da83..a6894a64149 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
@@ -18,9 +18,12 @@
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.catalog.info.TableNameInfo;
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.MTMVPartitionUtil;
@@ -70,6 +73,20 @@ public class IvmDeltaRewriter {
refreshContext.getConnectContext().getStatementContext().getNextRelationId(),
sinkChild.getOutput());
}
+ // A non-empty delta reaches this point, and only an aggregate delta
joins the MV's
+ // old rows with the new delta. If the previous refresh txn committed
but its data is
+ // not yet visible (an MV partition's committed version is ahead of
its visible
+ // version), that join misses the old rows and the delta is
permanently lost. Fail
+ // here; the fallback chain recomputes from the base tables without
reading old MV
+ // state. EXPLAIN REFRESH is exempt: it only produces a plan and
neither executes it
+ // nor reads MV data.
+ if (!rewriteContext.isExplain()
+ && rewriteResult.isAggMv()
+ && hasUnpublishedCommittedMvData(rewriteContext.getMtmv())) {
+ throw new IvmException(IvmFailureReason.MV_COMMIT_NOT_VISIBLE,
+ "previous refresh txn committed but its MV data is not
visible yet; "
+ + "aggregate delta would join stale old MV rows");
+ }
Plan deltaPlan = deltaResult.get().plan;
IvmDeltaRewriteResult result = deltaResult.get();
IvmDeltaRewriteResult mergedResult = new
IvmDeltaRewriteResult(deltaPlan,
@@ -89,6 +106,24 @@ public class IvmDeltaRewriter {
return visitor.rewritePlan(plan, ctx);
}
+ /**
+ * A refresh txn assigns the next partition version when it commits, while
the
+ * partition's visible version only advances when the txn publishes. Any MV
+ * partition whose committed version is still ahead of its visible version
+ * therefore holds unpublished refresh data.
+ */
+ private boolean hasUnpublishedCommittedMvData(MTMV mtmv) {
+ if (Config.isCloudMode()) {
+ return false;
+ }
+ for (Partition partition : mtmv.getPartitions()) {
+ if (partition.getCommittedVersion() >
partition.getVisibleVersion()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
static Pair<Plan, Map<Slot, Slot>> preSnapshot(Plan plan,
IvmDeltaRewriteState rewriteState) {
return rewriteSnapshot(plan, rewriteState, true);
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
index bacf16e17bd..61754e1556a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureReason.java
@@ -32,6 +32,8 @@ public enum IvmFailureReason {
MIN_MAX_BOUNDARY_HIT,
BITMAP_AGG_DELETE,
PLAN_SIGNATURE_MISMATCH,
+ /** A previous IVM refresh txn committed but its data is not yet visible.
*/
+ MV_COMMIT_NOT_VISIBLE,
MV_PARTITION_NOT_FOUND;
public boolean requiresCompleteRefresh() {
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 16d795c1b67..886d6ee6c23 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
@@ -92,7 +92,7 @@ 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,
false)));
+
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.incremental(mtmv)));
// 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 f3fb77e44d1..414a1df3d51 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
@@ -51,12 +51,28 @@ public class IvmRewriteContext {
FULL
}
+ /**
+ * How the rewrite result is consumed. {@link #DRY_RUN} only applies to
incremental
+ * refresh; {@link #EXPLAIN} applies to both incremental and complete
refresh and only
+ * produces a plan (no execution and no MV data read).
+ */
+ public enum ExecutionKind {
+ /** Real refresh: the rewrite result is executed and written back. */
+ EXECUTE,
+ /** Dry run: execute the delta query and return its rows, but do not
write the MV. */
+ DRY_RUN,
+ /** Explain: only generate the plan for display. */
+ EXPLAIN
+ }
+
private final Mode mode;
private final MTMV mtmv;
// The MTMV object does not exist yet during CREATE, so keep its name
separately for diagnostics.
private final String createMtmvName;
+ // Only used by EXPLAIN REFRESH ... ALL (kind == EXPLAIN); false for every
other kind.
private final boolean includeExhaustedStreams;
- private final boolean dryRun;
+ private final ExecutionKind executionKind;
+ // Present only for DRY_RUN (incremental refresh); empty otherwise.
private final Optional<IvmDryRunLimit> dryRunLimit;
private final Map<BaseTableInfo, Set<Long>> fullRefreshResetPartitionIds;
private final Optional<StreamReadMode> fullRefreshNonPctReadMode;
@@ -64,20 +80,15 @@ public class IvmRewriteContext {
// Null when the rewrite context is created outside the analyzeQuery flow.
private Boolean useFullKeys;
- public IvmRewriteContext(Mode mode, MTMV mtmv, boolean
includeExhaustedStreams) {
- this(mode, mtmv, null, includeExhaustedStreams, false,
Optional.empty(),
- Collections.emptyMap(), Optional.empty());
- }
-
private IvmRewriteContext(Mode mode, MTMV mtmv, String createMtmvName,
boolean includeExhaustedStreams,
- boolean dryRun, Optional<IvmDryRunLimit> dryRunLimit,
+ ExecutionKind executionKind, Optional<IvmDryRunLimit> dryRunLimit,
Map<BaseTableInfo, Set<Long>> fullRefreshResetPartitionIds,
Optional<StreamReadMode> fullRefreshNonPctReadMode) {
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;
this.includeExhaustedStreams = includeExhaustedStreams;
- this.dryRun = dryRun;
+ this.executionKind = Objects.requireNonNull(executionKind,
"executionKind can not be null");
this.dryRunLimit = Objects.requireNonNull(dryRunLimit, "dryRunLimit
can not be null");
Map<BaseTableInfo, Set<Long>> resetPartitionIds = new HashMap<>();
Objects.requireNonNull(fullRefreshResetPartitionIds,
"fullRefreshResetPartitionIds can not be null")
@@ -91,30 +102,47 @@ public class IvmRewriteContext {
public static IvmRewriteContext create(String mtmvName) {
return new IvmRewriteContext(Mode.CREATE, null,
Objects.requireNonNull(mtmvName, "mtmvName can not be null"),
false,
- false, Optional.empty(), Collections.emptyMap(),
Optional.empty());
+ ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
}
public static IvmRewriteContext normalize(MTMV mtmv) {
- return new IvmRewriteContext(Mode.NORMALIZE,
Objects.requireNonNull(mtmv, "mtmv can not be null"), false);
+ return new IvmRewriteContext(Mode.NORMALIZE,
Objects.requireNonNull(mtmv, "mtmv can not be null"),
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
+ }
+
+ public static IvmRewriteContext incremental(MTMV mtmv) {
+ return new IvmRewriteContext(Mode.INCREMENTAL,
Objects.requireNonNull(mtmv, "mtmv can not be null"),
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
}
- public static IvmRewriteContext incremental(MTMV mtmv, boolean
includeExhaustedStreams) {
- return new IvmRewriteContext(Mode.INCREMENTAL, mtmv,
includeExhaustedStreams);
+ /** 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());
}
public static IvmRewriteContext incrementalDryRun(MTMV mtmv,
Optional<IvmDryRunLimit> dryRunLimit) {
return new IvmRewriteContext(Mode.INCREMENTAL, mtmv, null, false,
- true, dryRunLimit, Collections.emptyMap(), Optional.empty());
+ ExecutionKind.DRY_RUN, dryRunLimit, Collections.emptyMap(),
Optional.empty());
+ }
+
+ /** 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());
}
public static IvmRewriteContext full(MTMV mtmv) {
- return new IvmRewriteContext(Mode.FULL, mtmv, false);
+ return new IvmRewriteContext(Mode.FULL, Objects.requireNonNull(mtmv,
"mtmv can not be null"),
+ null, false, ExecutionKind.EXECUTE, Optional.empty(),
Collections.emptyMap(), Optional.empty());
}
public static IvmRewriteContext full(MTMV mtmv,
Map<BaseTableInfo, Set<Long>> resetPartitionIds,
StreamReadMode nonPctReadMode) {
- return new IvmRewriteContext(Mode.FULL, mtmv, null, false, false,
Optional.empty(), resetPartitionIds,
+ return new IvmRewriteContext(Mode.FULL, mtmv, null, false,
ExecutionKind.EXECUTE, Optional.empty(),
+ resetPartitionIds,
Optional.of(Objects.requireNonNull(nonPctReadMode,
"nonPctReadMode can not be null")));
}
@@ -138,12 +166,17 @@ public class IvmRewriteContext {
return includeExhaustedStreams;
}
- // True for REFRESH ... INCREMENTAL WITH DRY RUN: the root plan must be a
LogicalResultSink.
+ /** True for REFRESH ... INCREMENTAL WITH DRY RUN: the root plan must be a
LogicalResultSink. */
public boolean isDryRun() {
- return dryRun;
+ return executionKind == ExecutionKind.DRY_RUN;
+ }
+
+ /** True for EXPLAIN REFRESH (incremental or complete): only the plan is
produced. */
+ public boolean isExplain() {
+ return executionKind == ExecutionKind.EXPLAIN;
}
- // Present only for REFRESH ... INCREMENTAL WITH DRY RUN; empty otherwise.
+ // Present only for DRY_RUN (incremental refresh); empty otherwise.
public Optional<IvmDryRunLimit> getDryRunLimit() {
return dryRunLimit;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
index 06ac74257b9..583b5b2551a 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/RefreshMTMVCommand.java
@@ -186,13 +186,13 @@ public class RefreshMTMVCommand extends Command
implements Forward, Explainable
"EXPLAIN REFRESH INCREMENTAL only supports IVM
materialized views");
}
statementContext.setIvmRewriteContext(Optional.of(
- IvmRewriteContext.incremental(mtmv,
includeExhaustedStreams)));
+ IvmRewriteContext.incrementalExplain(mtmv,
includeExhaustedStreams)));
// Excluded trigger tables must not be validated for binlog /
key-type support.
statementContext.setExcludedTriggerTables(mtmv.getExcludedTriggerTables());
return createIvmIncrRefreshManager().buildInsertCommand(mtmv);
case COMPLETE:
if (mtmv.isIvm()) {
-
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.full(mtmv)));
+
statementContext.setIvmRewriteContext(Optional.of(IvmRewriteContext.fullExplain(mtmv)));
}
statementContext.setExcludedTriggerTables(mtmv.getExcludedTriggerTables());
return UpdateMvByPartitionCommand.from(
@@ -216,6 +216,9 @@ public class RefreshMTMVCommand extends Command implements
Forward, Explainable
if (explainPlan != null) {
return;
}
+ // createRefreshCommand installs an EXPLAIN-kind rewrite context for
both the
+ // INCREMENTAL and COMPLETE branches, so guards that protect real
execution (and
+ // dry-run data reads) skip plan-only generation.
LogicalPlan refreshCommand = createRefreshCommand(mtmv,
statementContext);
if (refreshCommand instanceof Explainable) {
Explainable explainable = (Explainable) refreshCommand;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
index 007df317605..43b6a073c56 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java
@@ -124,6 +124,14 @@ public class DatabaseTransactionMgr {
// the max number of txn that can be remove per round.
// set it to avoid holding lock too long when removing too many txns per
round.
private static final int MAX_REMOVE_TXN_PER_ROUND = 10000;
+
+ // Test hook: when enabled with the MV table name as the debug point's
"value" param,
+ // finishTransaction returns without turning transactions that write that
MV table
+ // VISIBLE, so IVM regression tests can hold a refresh txn in COMMITTED
while other
+ // tables keep publishing normally.
+ public static final String DEBUG_POINT_FINISH_TRANSACTION_BLOCK_VISIBLE =
+ "DatabaseTransactionMgr.finishTransaction.block_visible";
+
// ConfigBase replaces the array on every update, so its identity is the
cache version.
private static volatile String[] cachedResourceGroupSuccQuorumConfig;
private static volatile Map<String, Integer> cachedResourceGroupSuccQuorum
= Map.of();
@@ -1258,6 +1266,11 @@ public class DatabaseTransactionMgr {
if (LOG.isDebugEnabled()) {
LOG.debug("finish transaction {} with tables {}", transactionId,
tableIdList);
}
+ String blockedTableName = DebugPointUtil.getDebugParamOrDefault(
+ DEBUG_POINT_FINISH_TRANSACTION_BLOCK_VISIBLE, "");
+ if (!blockedTableName.isEmpty() && transactionWritesTableNamed(db,
tableIdList, blockedTableName)) {
+ return;
+ }
List<? extends TableIf> tableList =
db.getTablesOnIdOrderIfExist(tableIdList);
if (!MetaLockUtils.tryWriteLockTablesIfExist(tableList, 10,
TimeUnit.SECONDS)) {
LOG.warn("finish transaction {} failed, get lock timeout with
tables {}", transactionId, tableIdList);
@@ -3256,4 +3269,14 @@ public class DatabaseTransactionMgr {
((BaseTableStream)
tableIf).unprotectedUpdateStreamUpdate(info.getUpdate(), ts);
}
}
+
+ private static boolean transactionWritesTableNamed(Database db, List<Long>
tableIdList, String tableName) {
+ for (Long tableId : tableIdList) {
+ Table table = db.getTableNullable(tableId);
+ if (table != null && table.getName().equals(tableName)) {
+ return true;
+ }
+ }
+ return false;
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
index 0ca4e1920a6..9f6fb5c5b89 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmAggDeltaHandlerTest.java
@@ -70,7 +70,7 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
mtmv.getIvmInfo().advanceRefreshVersion();
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
bundle.normalizedPlan, bundle.rewriteResult,
- IvmRewriteContext.incremental(mtmv, false),
bundle.connectContext);
+ IvmRewriteContext.incremental(mtmv), bundle.connectContext);
Assertions.assertNotNull(rewritten);
InsertIntoTableCommand command = new IvmIncrRefreshManager()
.buildInsertCommand((LogicalPlan) rewritten, mtmv);
@@ -89,7 +89,7 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
mtmv.getIvmInfo().advanceRefreshVersion();
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
bundle.normalizedPlan, bundle.rewriteResult,
- IvmRewriteContext.incremental(mtmv, false),
bundle.connectContext);
+ IvmRewriteContext.incremental(mtmv), bundle.connectContext);
Assertions.assertNotNull(rewritten);
InsertIntoTableCommand command = new IvmIncrRefreshManager()
.buildInsertCommand((LogicalPlan) rewritten, mtmv);
@@ -247,7 +247,7 @@ class IvmAggDeltaHandlerTest extends IvmDeltaTestBase {
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
bundle.normalizedPlan, bundle.rewriteResult,
- IvmRewriteContext.incremental(mtmv, false),
bundle.connectContext);
+ IvmRewriteContext.incremental(mtmv), bundle.connectContext);
Assertions.assertTrue(rewritten.anyMatch(node -> node instanceof
LogicalProject
&& ((LogicalProject<?>)
node).getProjects().stream().anyMatch(this::containsNonDeterministicGuard)));
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 07d0d906846..ac8b48a4d44 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
@@ -69,7 +69,7 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
ConnectContext connectContext, IvmRewriteResult rewriteResult) {
ensureStatementContext(connectContext);
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
- sinkChild, rewriteResult, IvmRewriteContext.incremental(mtmv,
false), connectContext);
+ sinkChild, rewriteResult, IvmRewriteContext.incremental(mtmv),
connectContext);
Assertions.assertNotNull(rewritten);
return new IvmIncrRefreshManager().buildInsertCommand(
(org.apache.doris.nereids.trees.plans.logical.LogicalPlan)
rewritten, mtmv);
@@ -97,8 +97,13 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
mtmv.setMvProperties(ImmutableMap.of(
PropertyAnalyzer.PROPERTIES_EXCLUDED_TRIGGER_TABLES,
excludedTriggerTables));
}
- return new IvmDeltaRewriter().generateIncrRefreshPlan(normalizedPlan,
rewriteResult,
- IvmRewriteContext.incremental(mtmv, includeExhaustedStreams),
connectContext);
+ // includeExhaustedStreams is an EXPLAIN REFRESH ... ALL option;
exercising it through
+ // the explain-kind context keeps the factory surface aligned with the
semantics.
+ IvmRewriteContext context = includeExhaustedStreams
+ ? IvmRewriteContext.incrementalExplain(mtmv, true)
+ : IvmRewriteContext.incremental(mtmv);
+ return new IvmDeltaRewriter().generateIncrRefreshPlan(normalizedPlan,
rewriteResult, context,
+ connectContext);
}
private LogicalJoin<LogicalOlapScan, LogicalOlapScan> crossJoin(
@@ -220,7 +225,7 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
bundle.normalizedPlan, bundle.rewriteResult,
- IvmRewriteContext.incremental(mtmv, false),
bundle.connectContext);
+ IvmRewriteContext.incremental(mtmv), bundle.connectContext);
new
CheckAfterRewrite().checkTreeAllSlotReferenceFromChildren(rewritten);
}
@@ -245,7 +250,7 @@ class IvmDeltaRewriterTest extends IvmDeltaTestBase {
PlanBundle bundle = normalizePlan(buildScanPlan(scan).child());
MTMV mtmv = buildMtmvFromPlan(bundle.normalizedPlan.getOutput());
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
- bundle.normalizedPlan, bundle.rewriteResult,
IvmRewriteContext.incremental(mtmv, false),
+ bundle.normalizedPlan, bundle.rewriteResult,
IvmRewriteContext.incremental(mtmv),
bundle.connectContext);
Assertions.assertInstanceOf(LogicalProject.class, rewritten);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandlerTest.java
index ca163312a8c..cf5b9d53859 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmJoinDeltaHandlerTest.java
@@ -197,7 +197,7 @@ class IvmJoinDeltaHandlerTest extends IvmDeltaTestBase {
IvmIncrRefreshContext ctx = newRefreshContext(normalizedPlan,
rewriteResult);
Plan mergedPlan = runWithIvmRewriteContext(ctx, () -> new
IvmDeltaRewriter().generateIncrRefreshPlan(
normalizedPlan, rewriteResult,
- IvmRewriteContext.incremental(ctx.getMtmv(), true),
ctx.getConnectContext()));
+ IvmRewriteContext.incrementalExplain(ctx.getMtmv(), true),
ctx.getConnectContext()));
LogicalUnion union = findOnlyUnion(mergedPlan);
Assertions.assertEquals(2, union.children().size());
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmLinearDeltaHandlerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmLinearDeltaHandlerTest.java
index 8fe294ca132..e82d9839ad3 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmLinearDeltaHandlerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmLinearDeltaHandlerTest.java
@@ -129,7 +129,7 @@ class IvmLinearDeltaHandlerTest extends IvmDeltaTestBase {
private InsertIntoTableCommand buildIncrementalInsertCommand(Plan
sinkChild, MTMV mtmv) {
PlanBundle bundle = normalizePlan(sinkChild);
Plan rewritten = new IvmDeltaRewriter().generateIncrRefreshPlan(
- bundle.normalizedPlan, bundle.rewriteResult,
IvmRewriteContext.incremental(mtmv, false),
+ bundle.normalizedPlan, bundle.rewriteResult,
IvmRewriteContext.incremental(mtmv),
bundle.connectContext);
Assertions.assertNotNull(rewritten);
return new
IvmIncrRefreshManager().buildInsertCommand((org.apache.doris.nereids.trees.plans.logical.LogicalPlan)
rewritten,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmIncrRefreshMTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmIncrRefreshMTMVTest.java
index 46c8d19aa68..fff4cf2864d 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmIncrRefreshMTMVTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmIncrRefreshMTMVTest.java
@@ -77,7 +77,7 @@ class IvmIncrRefreshMTMVTest {
void testCreateRewriteContextKeepsPlanUnchanged() {
LogicalOlapTableSink<Plan> sink = newSink(mtmv, scan);
RecordingRule rule = new RecordingRule(scan);
- IvmRewriteContext context = new
IvmRewriteContext(IvmRewriteContext.Mode.CREATE, null, false);
+ IvmRewriteContext context = IvmRewriteContext.create("test_mtmv");
Plan result = rule.rewriteRoot(sink, newJobContext(sink, context,
newRewriteResult(SIGNATURE)));
@@ -92,7 +92,7 @@ class IvmIncrRefreshMTMVTest {
IvmException exception = Assertions.assertThrows(IvmException.class,
() -> rule.rewriteRoot(sink, newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, false), null)));
+ IvmRewriteContext.incremental(mtmv), null)));
Assertions.assertEquals(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED,
exception.getFailureReason());
Assertions.assertEquals(0, rule.rewriter.callCount);
@@ -104,7 +104,7 @@ class IvmIncrRefreshMTMVTest {
IvmException exception = Assertions.assertThrows(IvmException.class,
() -> rule.rewriteRoot(scan, newJobContext(scan,
- IvmRewriteContext.incremental(mtmv, false),
newRewriteResult(SIGNATURE))));
+ IvmRewriteContext.incremental(mtmv),
newRewriteResult(SIGNATURE))));
Assertions.assertEquals(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED,
exception.getFailureReason());
Assertions.assertTrue(exception.getMessage().contains("LogicalOlapTableSink"));
@@ -178,7 +178,7 @@ class IvmIncrRefreshMTMVTest {
IvmException exception = Assertions.assertThrows(IvmException.class,
() -> rule.rewriteRoot(sink, newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, false),
newRewriteResult(SIGNATURE))));
+ IvmRewriteContext.incremental(mtmv),
newRewriteResult(SIGNATURE))));
Assertions.assertEquals(IvmFailureReason.PLAN_PATTERN_UNSUPPORTED,
exception.getFailureReason());
Assertions.assertTrue(exception.getMessage().contains("target table
mismatch"));
@@ -194,7 +194,7 @@ class IvmIncrRefreshMTMVTest {
RecordingRule rule = new RecordingRule(deltaPlan);
IvmRewriteResult rewriteResult = newRewriteResult(SIGNATURE);
JobContext jobContext = newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, true), rewriteResult);
+ IvmRewriteContext.incrementalExplain(mtmv, true),
rewriteResult);
Plan result = rule.rewriteRoot(sink, jobContext);
@@ -234,7 +234,7 @@ class IvmIncrRefreshMTMVTest {
RecordingRule rule = new RecordingRule(deltaPlan);
Plan result = rule.rewriteRoot(sink, newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, false),
newRewriteResult(SIGNATURE)));
+ IvmRewriteContext.incremental(mtmv),
newRewriteResult(SIGNATURE)));
LogicalOlapTableSink<?> rewrittenSink = (LogicalOlapTableSink<?>)
result;
Assertions.assertSame(deltaPlan, rewrittenSink.child());
@@ -251,7 +251,7 @@ class IvmIncrRefreshMTMVTest {
RecordingRule rule = new RecordingRule(deltaPlan);
IvmRewriteResult rewriteResult = newRewriteResult(SIGNATURE);
JobContext jobContext = newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, false), rewriteResult);
+ IvmRewriteContext.incremental(mtmv), rewriteResult);
Plan firstResult = rule.rewriteRoot(sink, jobContext);
Plan secondResult = rule.rewriteRoot(firstResult, jobContext);
@@ -266,7 +266,7 @@ class IvmIncrRefreshMTMVTest {
RecordingRule rule = new RecordingRule(null);
Plan result = rule.rewriteRoot(sink, newJobContext(sink,
- IvmRewriteContext.incremental(mtmv, false),
newRewriteResult(SIGNATURE)));
+ IvmRewriteContext.incremental(mtmv),
newRewriteResult(SIGNATURE)));
Assertions.assertInstanceOf(LogicalOlapTableSink.class, result);
LogicalOlapTableSink<?> rewrittenSink = (LogicalOlapTableSink<?>)
result;
@@ -278,7 +278,7 @@ class IvmIncrRefreshMTMVTest {
@Test
void testIncrementalContextRejectsNullMtmv() {
Assertions.assertThrows(NullPointerException.class,
- () -> IvmRewriteContext.incremental(null, false));
+ () -> IvmRewriteContext.incremental(null));
}
private JobContext newJobContext(Plan root, IvmRewriteContext
rewriteContext,
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMVTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMVTest.java
index 515f6999cc9..34898b049b8 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMVTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/analysis/IvmNormalizeMTMVTest.java
@@ -128,7 +128,7 @@ class IvmNormalizeMTMVTest {
@Test
void testIvmRewriteContextEnablesNormalizeWithoutSessionVariable() {
JobContext jobContext = newJobContextForRoot(scan, false,
Collections.emptySet(),
- Optional.of(new
IvmRewriteContext(IvmRewriteContext.Mode.CREATE, null, false)));
+ Optional.of(IvmRewriteContext.create("test_mtmv")));
Plan result = new IvmNormalizeMTMV().rewriteRoot(scan, jobContext);
Assertions.assertInstanceOf(LogicalProject.class, result);
@@ -142,7 +142,7 @@ class IvmNormalizeMTMVTest {
void testIncrementalSignatureMismatchThrowsAtNormalize() {
MTMV mtmv = mockIvmMtmv("stored-signature");
JobContext jobContext = newJobContextForRoot(scan, true,
Collections.emptySet(),
- Optional.of(IvmRewriteContext.incremental(mtmv, false)));
+ Optional.of(IvmRewriteContext.incremental(mtmv)));
IvmException exception = Assertions.assertThrows(IvmException.class,
() -> new IvmNormalizeMTMV().rewriteRoot(scan, jobContext));
@@ -162,7 +162,7 @@ class IvmNormalizeMTMVTest {
MTMV mtmv = mockIvmMtmv(signature);
JobContext jobContext = newJobContextForRoot(scan, true,
Collections.emptySet(),
- Optional.of(IvmRewriteContext.incremental(mtmv, false)));
+ Optional.of(IvmRewriteContext.incremental(mtmv)));
Plan result = new IvmNormalizeMTMV().rewriteRoot(scan, jobContext);
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.out
new file mode 100644
index 00000000000..6ebb79b1122
--- /dev/null
+++
b/regression-test/data/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.out
@@ -0,0 +1,26 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !prev_commit_baseline --
+1 1 10
+2 1 20
+
+-- !prev_commit_after_blocked_refresh --
+1 1 10
+2 1 20
+
+-- !prev_commit_unchanged_after_failed_refresh --
+1 1 10
+2 1 20
+
+-- !prev_commit_converged_after_publish --
+1 1 10
+2 1 20
+3 1 30
+4 1 40
+
+-- !prev_commit_converged --
+1 1 10
+2 1 20
+3 1 30
+4 1 40
+5 1 50
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.groovy
new file mode 100644
index 00000000000..0448142511d
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_agg_previous_commit_not_visible.groovy
@@ -0,0 +1,219 @@
+// 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
+
+suite("test_ivm_agg_previous_commit_not_visible", "nonConcurrent") {
+ // The finish-transaction debug point only fires on the local
(shared-nothing) txn
+ // path; the cloud transaction manager never hits it.
+ if (isCloudMode()) {
+ logger.info("skip test_ivm_agg_previous_commit_not_visible on cloud
mode: " +
+ "finishTransaction debug point only fires on the local txn
path")
+ return
+ }
+ def blockVisible = "DatabaseTransactionMgr.finishTransaction.block_visible"
+ def mvName = "ivm_prev_commit_mv"
+ def baseName = "ivm_prev_commit_t"
+ // Id of the newest task of this MV; refreshed before every REFRESH so the
wait
+ // below can tell the task just submitted apart from its predecessors.
+ def prevTaskId = ""
+
+ def disableDebugPoints = {
+ GetDebugPoint().disableDebugPointForAllFEs(blockVisible)
+ }
+ def enableBlockMvVisible = {
+ GetDebugPoint().enableDebugPointForAllFEs(blockVisible, [value:
mvName])
+ }
+
+ // The newest task of the MV before a refresh is submitted. tasks()
transiently misses
+ // a task that just finished: AbstractJob.onTaskSuccess() removes it from
the job's
+ // running list before MTMVTask.after() appends it to the MV history, and
while that
+ // window is open the newest row of the MV is its previous task.
Remembering the id
+ // and ignoring it while polling keeps the wait from latching onto that
task's already
+ // terminal status.
+ def newestTaskId = {
+ def taskResult = sql_return_maparray("""
+ SELECT TaskId 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()
+ }
+
+ def latestTask = { String excludeTaskId ->
+ def taskResult
+ Awaitility.await().atMost(180, 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 = '${mvName}'
+ ORDER BY CreateTime DESC, TaskId DESC LIMIT 1
+ """)
+ return !taskResult.isEmpty()
+ && taskResult[0].TaskId.toString() != excludeTaskId
+ && taskResult[0].Status.toString() != 'PENDING'
+ && taskResult[0].Status.toString() != 'RUNNING'
+ })
+ return taskResult[0]
+ }
+
+ def waitTaskSuccess = { String excludeTaskId ->
+ def task = latestTask(excludeTaskId)
+ if (task.Status.toString() != "SUCCESS") {
+ logger.info("refresh task ${task.TaskId} is ${task.Status}, error:
${task.ErrorMsg}")
+ }
+ assertEquals("SUCCESS", task.Status.toString())
+ return task
+ }
+
+ try {
+ disableDebugPoints()
+ sql """drop materialized view if exists ${mvName}"""
+ sql """drop table if exists ${baseName}"""
+
+ sql """
+ CREATE TABLE ${baseName} (
+ k1 INT,
+ v1 INT
+ )
+ UNIQUE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "binlog.enable" = "true",
+ "binlog.format" = "ROW",
+ "binlog.need_historical_value" = "true",
+ "enable_unique_key_merge_on_write" = "true"
+ )
+ """
+ sql """INSERT INTO ${baseName} VALUES (1, 10), (2, 20)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ${mvName}
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY RANDOM BUCKETS 1
+ PROPERTIES ('replication_num' = '1')
+ AS SELECT k1, COUNT(*) AS cnt, SUM(v1) AS sum_v1
+ FROM ${baseName} GROUP BY k1
+ """
+
+ // Initial INCREMENTAL refresh (no debug point): consumes the
historical binlog
+ // and establishes the baseline snapshot.
+ prevTaskId = newestTaskId()
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ waitTaskSuccess(prevTaskId)
+ order_qt_prev_commit_baseline """
+ SELECT k1, cnt, sum_v1 FROM ${mvName} ORDER BY k1
+ """
+
+ // Batch B is visible before the debug point blocks the next MV
refresh.
+ sql """INSERT INTO ${baseName} VALUES (3, 30)"""
+ enableBlockMvVisible()
+
+ // R1: the delta txn commits but its finish (VISIBLE) is blocked by
the debug
+ // point, so the refresh txn stays COMMITTED. The insert times out
waiting for
+ // publish and the task still reports SUCCESS (committed mode). The
refresh task
+ // runs in an internal ConnectContext that clones the global session
variables,
+ // so shorten insert_visible_timeout_ms globally for the wait.
+ prevTaskId = newestTaskId()
+ setGlobalVarTemporary([insert_visible_timeout_ms: 3000], {
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ waitTaskSuccess(prevTaskId)
+ })
+ order_qt_prev_commit_after_blocked_refresh """
+ SELECT k1, cnt, sum_v1 FROM ${mvName} ORDER BY k1
+ """
+
+ // EXPLAIN REFRESH only produces a plan (no execution and no MV data
read), so it
+ // must still succeed while the previous refresh txn is committed but
not visible;
+ // this holds for INCREMENTAL and COMPLETE alike.
+ sql """EXPLAIN REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ sql """EXPLAIN REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+
+ // Batch C commits and publishes normally (the debug point only
matches the MV),
+ // so the next incremental refresh has a real delta while the previous
refresh
+ // txn on the MV is still not visible. The aggregate delta would join
stale old
+ // MV rows, so the refresh must fail instead of corrupting the MV.
+ sql """INSERT INTO ${baseName} VALUES (4, 40)"""
+ prevTaskId = newestTaskId()
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def failedTask = latestTask(prevTaskId)
+ assertEquals("FAILED", failedTask.Status.toString())
+
assertTrue(failedTask.ErrorMsg.toString().contains("MV_COMMIT_NOT_VISIBLE"))
+ assertTrue(failedTask.ErrorMsg.toString().contains("not visible yet"))
+ order_qt_prev_commit_unchanged_after_failed_refresh """
+ SELECT k1, cnt, sum_v1 FROM ${mvName} ORDER BY k1
+ """
+
+ // A COMPLETE refresh recomputes from the base tables and never joins
old MV rows,
+ // so it is not stopped by the guard. But the debug point is still on,
so its own
+ // write txn can not turn VISIBLE either: the task reports SUCCESS
(committed
+ // mode) while readers still see none of the new data.
+ prevTaskId = newestTaskId()
+ setGlobalVarTemporary([insert_visible_timeout_ms: 3000], {
+ sql """REFRESH MATERIALIZED VIEW ${mvName} COMPLETE"""
+ waitTaskSuccess(prevTaskId)
+ })
+ def completeStillInvisibleRows = sql """SELECT COUNT(*) FROM ${mvName}
WHERE k1 >= 3"""
+ assertEquals("0", completeStillInvisibleRows.get(0).get(0).toString())
+
+ // The COMPLETE refresh above reported SUCCESS while its txn is still
COMMITTED,
+ // so a subsequent strict incremental with a real delta is refused by
the guard
+ // just like after a stuck incremental: readers must never see MV
state computed
+ // on top of rows that are committed but not visible.
+ sql """INSERT INTO ${baseName} VALUES (5, 50)"""
+ prevTaskId = newestTaskId()
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ def failedAfterCompleteTask = latestTask(prevTaskId)
+ assertEquals("FAILED", failedAfterCompleteTask.Status.toString())
+
assertTrue(failedAfterCompleteTask.ErrorMsg.toString().contains("MV_COMMIT_NOT_VISIBLE"))
+ def stillInvisibleRows = sql """SELECT COUNT(*) FROM ${mvName} WHERE
k1 >= 3"""
+ assertEquals("0", stillInvisibleRows.get(0).get(0).toString())
+
+ // Drop the debug point: the committed-but-unpublished txns
auto-publish (their
+ // rowsets are already on the BE, only the FE finish was held back),
the MV
+ // partition commit and visible versions converge, and reads now
return the full
+ // aggregate — the deferred failure lost nothing.
+ disableDebugPoints()
+
+ // The stuck txns auto-publish (their rowsets are already on the BE,
only the FE
+ // finish was held back), so the MV partition commit and visible
versions converge
+ // and readers see the COMPLETE content (base rows through (4,40)).
+ Awaitility.await().atMost(120, SECONDS).pollInterval(2, SECONDS).until
{
+ sql("SELECT COUNT(*) FROM ${mvName}").get(0).get(0).toString() ==
"4"
+ }
+ order_qt_prev_commit_converged_after_publish """
+ SELECT k1, cnt, sum_v1 FROM ${mvName} ORDER BY k1
+ """
+
+ // (5,50) was inserted after the stuck COMPLETE, so it needs one more
successful
+ // INCREMENTAL refresh; the guard is clear now that the stuck txns
have published.
+ prevTaskId = newestTaskId()
+ sql """REFRESH MATERIALIZED VIEW ${mvName} INCREMENTAL"""
+ waitTaskSuccess(prevTaskId)
+ order_qt_prev_commit_converged """
+ SELECT k1, cnt, sum_v1 FROM ${mvName} ORDER BY k1
+ """
+ } finally {
+ disableDebugPoints()
+ sql """drop materialized view if exists ${mvName}"""
+ sql """drop table if exists ${baseName}"""
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
