This is an automated email from the ASF dual-hosted git repository.
morningman 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 86d62ae00c9 [fix](ivm) Fall back to COMPLETE when an IVM stream offset
has expired (#69772)
86d62ae00c9 is described below
commit 86d62ae00c9cc0d7f8644de2912e0e13469b280e
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Oct 28 15:22:14 2026 +0000
[fix](ivm) Fall back to COMPLETE when an IVM stream offset has expired
(#69772)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
**In short.** Once a base table's `changelog.ttl_seconds` expires the
offset of an IVM materialized view's internal stream, every refresh of
that MV fails, including refreshes that are allowed to fall back to
COMPLETE, until someone runs a manual COMPLETE refresh. This PR
classifies that failure as `CHANGELOG_BROKEN`, so a refresh that may
fall back rebuilds the MV with COMPLETE (the only attempt that resets
the stream baselines) and a strict `INCREMENTAL` refresh fails with that
reason.
**Background.** An IVM MV reads each unique-key (merge-on-write) base
table through an internal table stream created with `MIN_DELTA`.
Planning a `MIN_DELTA` read refuses an offset that is older than what
the table still retains: `ChangelogUtils.effectiveStartTso` throws
`Changelog offset has expired according to changelog.ttl_seconds`. The
incremental refresh turns a failure into a fallback only when
`IvmFailureClassifier` recognizes its message.
**The problem, and what it cost.**
- The expired-offset message was not one the classifier knew, so
`IvmIncrRefreshManager.doRefresh` rethrew it and the task failed even
for `REFRESH ... AUTO`, `INCREMENTAL FALLBACK`, or the scheduled refresh
of a `REFRESH INCREMENTAL FALLBACK` MV.
- A failed refresh does not move the offset, so while the TTL stays as
it is every later refresh fails the same way, and the MV keeps serving
old rows.
- The delta is not the only read of that offset. When some MV partitions
need a rebuild (for example after a `TRUNCATE` of a base partition), the
incremental attempt first rebuilds them with a partition refresh, which
reads the tables the MV does not partition by through their streams at
the offsets (`StreamReadMode.SNAPSHOT`). With such an offset expired
that rebuild failed the same way, before the delta ran.
- `IvmFailureReason.CHANGELOG_BROKEN` had no producer in production
code, although `MTMVTask.handleIvmFallbackResult` already sends it
straight to COMPLETE; the comment and log of that branch described it as
"previous run incomplete".
**How this PR fixes it.**
- `IvmFailureClassifier` maps a failure carrying
`ChangelogUtils.CHANGELOG_OFFSET_EXPIRED` to `CHANGELOG_BROKEN`
(matching the constant the read throws with, not a copy of it).
- `MTMVTask.executeIvmAttempt` answers that failure of the rebuild
before the delta through `handleIvmFallbackResult`, the way the delta's
failure is answered; any other rebuild failure still fails the attempt.
- The `CHANGELOG_BROKEN` branch of `handleIvmFallbackResult` now
explains why it goes directly to COMPLETE (a partition refresh still
reads the streams at their offsets) and logs the detail with the offset
and cutoff TSO.
**Results** (local shared-nothing cluster):
| Refresh after the offset expired | Before | After |
| --- | --- | --- |
| `REFRESH ... INCREMENTAL` (strict) | FAILED, `IvmFallbackReason` empty
| FAILED, `IvmFallbackReason = CHANGELOG_BROKEN`, error says
`reason=CHANGELOG_BROKEN` with both TSOs |
| `REFRESH ... AUTO` | FAILED, MV keeps the old rows | SUCCESS through
COMPLETE, `IvmFallbackReason = CHANGELOG_BROKEN` |
| the next `REFRESH ... AUTO` | fails again while the TTL is unchanged |
SUCCESS, incremental |
| `REFRESH ... AUTO` that first rebuilds a dirty MV partition while a
joined table's offset has expired | FAILED | SUCCESS through COMPLETE,
`IvmFallbackReason = CHANGELOG_BROKEN` |
**Classes, and how they call each other.**
- `MTMVTask.run` → `executeIvmAttempt` → (dirty partitions)
`executePartitionBasedRefresh` → on an expired offset
`handleIvmFallbackResult(CHANGELOG_BROKEN)`
- `executeIvmAttempt` → `executeSingleIvmAttempt` →
`IvmIncrRefreshManager.doRefresh` →
`IvmFailureClassifier.classifyExecutionFailure` →
`IvmIncrRefreshResult.fallback(CHANGELOG_BROKEN)` →
`handleIvmFallbackResult`
- `handleIvmFallbackResult`: strict request → `JobException` with the
reason; otherwise `FALLBACK_TO_COMPLETE` → `executeCompleteAttempt`,
which resets the stream baselines.
Not changed: the planner's TTL check itself, the PARTITIONS attempt (it
reads a changed table at an old offset only after a partition refresh
that failed part-way: a plan covering every partition reads the joined
tables with `RESET`), and the persisted task fields.
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 23 ++-
.../doris/mtmv/ivm/IvmFailureClassifier.java | 7 +
.../apache/doris/mtmv/ivm/IvmFailureReason.java | 1 +
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 86 ++++++++
.../doris/mtmv/ivm/IvmIncrRefreshManagerTest.java | 21 ++
.../ivm/test_ivm_changelog_offset_expired.out | 57 ++++++
.../ivm/test_ivm_changelog_offset_expired.groovy | 221 +++++++++++++++++++++
7 files changed, 411 insertions(+), 5 deletions(-)
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 6de3a886bea..86d8661db1d 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
@@ -66,6 +66,7 @@ import org.apache.doris.mtmv.MTMVRefreshPartitionSnapshot;
import org.apache.doris.mtmv.MTMVRelatedTableIf;
import org.apache.doris.mtmv.MTMVRelation;
import org.apache.doris.mtmv.MTMVUtil;
+import org.apache.doris.mtmv.ivm.IvmFailureClassifier;
import org.apache.doris.mtmv.ivm.IvmFailureReason;
import org.apache.doris.mtmv.ivm.IvmIncrRefreshContext;
import org.apache.doris.mtmv.ivm.IvmIncrRefreshManager;
@@ -828,6 +829,17 @@ public class MTMVTask extends AbstractTask {
this.refreshMode = generateRefreshMode(toRebuild);
try {
executePartitionBasedRefresh(refreshContext,
RefreshMode.PARTITIONS, ctx, toRebuild);
+ } catch (JobException e) {
+ // The rebuild reads the tables the MV does not partition by
through their streams at the
+ // offsets, so an offset changelog.ttl_seconds has expired
stops it the way it stops the delta
+ // below -- and it is the same failure of this attempt,
answered the same way.
+ String detail = Util.getRootCauseMessage(e);
+ if (IvmFailureClassifier.classifyExecutionFailure(detail)
+ .filter(reason -> reason ==
IvmFailureReason.CHANGELOG_BROKEN).isEmpty()) {
+ throw e;
+ }
+ return handleIvmFallbackResult(
+
IvmIncrRefreshResult.fallback(IvmFailureReason.CHANGELOG_BROKEN, detail),
request);
} finally {
recordRebuiltPartitions(request);
}
@@ -1184,11 +1196,12 @@ public class MTMVTask extends AbstractTask {
+ ", detail=" + ivmResult.getDetailMessage());
}
if (ivmResult.getFailureReason() == IvmFailureReason.CHANGELOG_BROKEN)
{
- // The previous task already entered the IVM execution phase. If
- // fallback is allowed, jump directly to COMPLETE recovery instead
of
- // trying PARTITIONS first.
- LOG.warn("IVM previous run incomplete for mv={}, taskId={}.
Continuing with COMPLETE recovery.",
- mtmv.getName(), getTaskId());
+ // changelog.ttl_seconds has expired the offset of a stream this
attempt reads, so no incremental
+ // attempt can catch the MV up. A partition refresh still reads
the streams at their offsets, so
+ // go directly to COMPLETE: it is the only attempt that resets the
stream baselines.
+ LOG.warn("IVM stream changelog expired for mv={}, detail={},
taskId={}. "
+ + "Continuing with COMPLETE refresh.",
+ mtmv.getName(), ivmResult.getDetailMessage(), getTaskId());
return AttemptResultType.FALLBACK_TO_COMPLETE;
}
if (ivmResult.getFailureReason().requiresCompleteRefresh()) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureClassifier.java
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureClassifier.java
index c58fc4f3394..e0c7244ac78 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureClassifier.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/mtmv/ivm/IvmFailureClassifier.java
@@ -17,6 +17,8 @@
package org.apache.doris.mtmv.ivm;
+import org.apache.doris.changelog.ChangelogUtils;
+
import java.util.Objects;
import java.util.Optional;
@@ -48,6 +50,11 @@ public final class IvmFailureClassifier {
if (detail.contains(MV_PARTITION_NOT_FOUND_MSG)) {
return Optional.of(IvmFailureReason.MV_PARTITION_NOT_FOUND);
}
+ // Planning a MIN_DELTA read refuses an offset that
changelog.ttl_seconds has expired: the changelog
+ // the stream has to read from is gone, so no delta can catch the MV
up.
+ if (detail.contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED)) {
+ return Optional.of(IvmFailureReason.CHANGELOG_BROKEN);
+ }
return Optional.empty();
}
}
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 1db33550941..9b2a5154548 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
@@ -20,6 +20,7 @@ package org.apache.doris.mtmv.ivm;
/** Reasons IVM analysis or execution can fail. */
public enum IvmFailureReason {
// ATTN(changelog-rename): formerly "BINLOG_BROKEN"
+ /** The changelog a stream has to read from is outside the retention:
changelog.ttl_seconds expired its offset. */
CHANGELOG_BROKEN,
// ATTN(changelog-rename): formerly "BINLOG_NOT_ENABLED"
CHANGELOG_NOT_ENABLED,
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 3bed5bd078f..006ff1fd780 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
@@ -27,6 +27,7 @@ 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.changelog.ChangelogUtils;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
@@ -1775,6 +1776,91 @@ public class MTMVTaskTest {
}
}
+ /**
+ * The rebuild an incremental attempt runs first, for the MV partitions
that need one, reads the tables the MV
+ * does not partition by through their streams at the offsets. An offset
changelog.ttl_seconds has expired stops
+ * it the way it stops the delta, so the answer is the same: a request
that may fall back goes to COMPLETE, the
+ * only attempt that resets the stream baselines.
+ */
+ @Test
+ public void
testExpiredOffsetInTheRebuildBeforeTheDeltaFallsBackToComplete() throws
Exception {
+ MTMVTask task = new MTMVTask(mtmv, relation,
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
+ RefreshMode.INCREMENTAL, true, null));
+
+ Object result = runIvmAttemptWhoseRebuildReadsAnExpiredOffset(task);
+
+ Assertions.assertEquals("FALLBACK_TO_COMPLETE", result.toString());
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
+ }
+
+ /** A request that may not fall back fails instead, with the reason it
would have fallen back for. */
+ @Test
+ public void
testExpiredOffsetInTheRebuildBeforeTheDeltaFailsAStrictRequest() {
+ MTMVTask task = new MTMVTask(mtmv, relation,
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
+ RefreshMode.INCREMENTAL, false, null));
+
+ JobException failure = Assertions.assertThrows(JobException.class,
+ () -> runIvmAttemptWhoseRebuildReadsAnExpiredOffset(task));
+
+ Assertions.assertTrue(failure.getMessage().contains("reason=" +
IvmFailureReason.CHANGELOG_BROKEN));
+
Assertions.assertTrue(failure.getMessage().contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED));
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
+ Deencapsulation.getField(task, "ivmFallbackReason"));
+ }
+
+ /**
+ * Runs the incremental attempt with p1 needing a rebuild and p2 caught
up, the rebuild's read refused the way
+ * planning refuses a MIN_DELTA read from an expired offset. The delta
must not be reached.
+ */
+ private Object runIvmAttemptWhoseRebuildReadsAnExpiredOffset(MTMVTask
task) throws Exception {
+ Mockito.when(mtmv.isIvm()).thenReturn(true);
+ Mockito.when(mtmv.getName()).thenReturn("test_mv");
+ Mockito.when(mtmv.hasRefreshSnapshot()).thenReturn(true);
+ Mockito.when(mtmv.getRefreshPartitionNum()).thenReturn(1);
+
Mockito.when(mtmvPartitionInfo.getPctInfos()).thenReturn(Collections.emptyList());
+
Mockito.when(mtmv.getPartitionNames()).thenReturn(Sets.newHashSet(poneName,
ptwoName));
+
Mockito.when(mtmv.getPartitionStates()).thenReturn(Maps.newHashMap(Map.of(
+ poneName, new MTMVPartitionState(1, 2),
+ ptwoName, new MTMVPartitionState(2, 2))));
+ MTMVRefreshContext refreshContext =
Mockito.mock(MTMVRefreshContext.class);
+
Mockito.when(refreshContext.getByPartitionName(Mockito.anyString())).thenReturn(Collections.emptyMap());
+ mtmvPartitionUtilStatic.when(() ->
MTMVPartitionUtil.generatePartitionSnapshots(
+ Mockito.same(refreshContext), Mockito.anySet(),
Mockito.anySet()))
+ .thenReturn(Collections.emptyMap());
+ RuntimeException expired =
Assertions.assertThrows(RuntimeException.class,
+ () -> ChangelogUtils.effectiveStartTso(99L, 100L, true));
+
+ ConnectContext mtmvCtx = new ConnectContext();
+ mtmvCtx.setQueryId(new TUniqueId(1L, 2L));
+ mtmvCtx.setThreadLocalInfo();
+ UpdateMvByPartitionCommand command =
Mockito.mock(UpdateMvByPartitionCommand.class);
+ try (MockedStatic<MTMVPlanUtil> mtmvPlanUtilStatic =
Mockito.mockStatic(MTMVPlanUtil.class);
+ MockedStatic<UpdateMvByPartitionCommand> updateMvStatic
+ = Mockito.mockStatic(UpdateMvByPartitionCommand.class);
+ MockedConstruction<IvmIncrRefreshManager> delta
+ =
Mockito.mockConstruction(IvmIncrRefreshManager.class)) {
+ mtmvPlanUtilStatic.when(() ->
MTMVPlanUtil.createMTMVContext(Mockito.eq(mtmv), Mockito.anyList()))
+ .thenReturn(mtmvCtx);
+ updateMvStatic.when(() -> UpdateMvByPartitionCommand.from(
+ Mockito.eq(mtmv), Mockito.anySet(), Mockito.anyMap(),
Mockito.any(StatementContext.class)))
+ .thenReturn(command);
+ mtmvPlanUtilStatic.when(() -> MTMVPlanUtil.executeCommand(
+ Mockito.eq(mtmvCtx), Mockito.eq(command),
Mockito.any(StatementContext.class),
+ Mockito.anyString(),
Mockito.any(Consumer.class))).thenThrow(expired);
+
+ Object request = Deencapsulation.invoke(task,
"resolveRefreshRequest");
+ try {
+ return Deencapsulation.invoke(task, "executeIvmAttempt",
refreshContext, request, mtmvCtx,
+ Lists.newArrayList());
+ } finally {
+ Assertions.assertTrue(delta.constructed().isEmpty());
+ }
+ } finally {
+ ConnectContext.remove();
+ }
+ }
+
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
index 37d49924fa3..ef318686683 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/mtmv/ivm/IvmIncrRefreshManagerTest.java
@@ -20,9 +20,11 @@ package org.apache.doris.mtmv.ivm;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.MTMV;
import org.apache.doris.catalog.info.TableNameInfo;
+import org.apache.doris.changelog.ChangelogUtils;
import org.apache.doris.mtmv.MTMVPlanUtil;
import org.apache.doris.nereids.StatementContext;
import org.apache.doris.nereids.analyzer.UnboundTableSink;
+import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.trees.plans.commands.Command;
import
org.apache.doris.nereids.trees.plans.commands.insert.InsertIntoTableCommand;
import org.apache.doris.qe.ConnectContext;
@@ -172,6 +174,25 @@ public class IvmIncrRefreshManagerTest {
IvmFailureClassifier.NON_DETERMINISTIC_ROW_ID_MSG_PREFIX);
}
+ @Test
+ public void
testManagerReturnsChangelogBrokenFallbackForExpiredStreamOffset() throws
Exception {
+ // What planning a MIN_DELTA read throws once changelog.ttl_seconds
has expired the stream offset.
+ AnalysisException expired =
Assertions.assertThrows(AnalysisException.class,
+ () -> ChangelogUtils.effectiveStartTso(99L, 100L, true));
+ MTMV mtmv = mockMtmv();
+ Command cmd = Mockito.mock(Command.class);
+ TestIvmIncrRefreshManager manager = new
TestIvmIncrRefreshManager(newContext(mtmv), makeCommands(cmd));
+ manager.failureMessage = expired.getMessage();
+
+ IvmIncrRefreshResult result = manager.doRefresh(manager.context);
+
+ Assertions.assertFalse(result.isSuccess());
+ Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN,
result.getFailureReason());
+ // A strict refresh reports this detail, so it has to keep the offset
and the cutoff.
+
Assertions.assertTrue(result.getDetailMessage().contains(expired.getMessage()));
+ Assertions.assertTrue(manager.executeCalled);
+ }
+
@Test
public void testManagerReturnsChangelogNotEnabledFallbackOnIvmException()
throws Exception {
MTMV mtmv = mockMtmv();
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired.out
new file mode 100644
index 00000000000..d5b8e92972c
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired.out
@@ -0,0 +1,57 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !baseline_task --
+SUCCESS COMPLETE NONE false false
+
+-- !baseline_mv --
+1 10
+2 20
+
+-- !strict_task --
+FAILED NONE CHANGELOG_BROKEN true true
+
+-- !mv_after_strict --
+1 10
+2 20
+
+-- !fallback_task --
+SUCCESS COMPLETE CHANGELOG_BROKEN false false
+
+-- !mv_after_fallback --
+1 10
+2 20
+3 30
+
+-- !task_after_fallback --
+SUCCESS NONE NONE false false
+
+-- !mv_after_a_further_refresh --
+1 10
+2 20
+3 30
+4 40
+
+-- !partitioned_baseline_mv --
+1 2026-01-10 100 10
+2 2026-02-10 200 10
+
+-- !rebuild_strict_task --
+FAILED PARTIAL CHANGELOG_BROKEN true true
+
+-- !partitioned_mv_after_strict --
+1 2026-01-10 100 10
+2 2026-02-10 200 10
+
+-- !rebuild_fallback_task --
+SUCCESS COMPLETE CHANGELOG_BROKEN false false
+
+-- !partitioned_mv_after_fallback --
+1 2026-01-10 100 20
+2 2026-02-10 200 20
+
+-- !partitioned_task_after_fallback --
+SUCCESS NONE NONE false false
+
+-- !partitioned_mv_after_a_further_refresh --
+1 2026-01-10 100 30
+2 2026-02-10 200 20
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired.groovy
new file mode 100644
index 00000000000..6bba1acb236
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired.groovy
@@ -0,0 +1,221 @@
+// 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
+
+/**
+ * An incremental refresh whose stream offset the base table's
changelog.ttl_seconds has expired.
+ *
+ * <p>An IVM MV reads a unique-key base table through an internal stream
created with MIN_DELTA, and planning
+ * such a read refuses to start from an offset older than the changelog the
table still retains. That refusal is
+ * the stream being broken for the incremental attempt, not a failure of the
refresh: the attempt reports it as
+ * CHANGELOG_BROKEN, and only COMPLETE, which resets the stream baselines, can
bring the MV back.
+ *
+ * <p>Cases pinned here:
+ * <ol>
+ * <li>the delta reads the expired offset: a strict INCREMENTAL refresh
fails with that reason and leaves the
+ * MV as it was, an AUTO refresh, which may fall back, rebuilds the MV
with COMPLETE and records the
+ * reason, and the refresh after it is incremental again;</li>
+ * <li>the rebuild an incremental attempt runs first for an MV partition
that needs one reads the expired
+ * offset of a table the MV does not partition by: the same answer,
before the delta runs.</li>
+ * </ol>
+ *
+ * <p>nonConcurrent: the offsets expire by sleeping past a one-second TTL.
+ */
+suite("test_ivm_changelog_offset_expired", "nonConcurrent") {
+ def waitForNewTask = { String mvName, 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]
+ }
+
+ // How a refresh ended, in one row. Unset RefreshMode / IvmFallbackReason
come back as the literal
+ // two-character string "\N", which does not survive the .out round trip,
so fold them into a printable
+ // token; a reason is an enum name, which "\N" is not. The error text
carries TSOs, so only whether it
+ // names the expired offset and the reason is kept.
+ def taskQuery = { String taskId ->
+ """
+ SELECT Status,
+ CASE WHEN RefreshMode IN ('COMPLETE', 'PARTIAL',
'NOT_REFRESH')
+ THEN RefreshMode ELSE 'NONE' END,
+ CASE WHEN IvmFallbackReason REGEXP '^[A-Z_]+\$'
+ THEN IvmFallbackReason ELSE 'NONE' END,
+ ErrorMsg LIKE '%Changelog offset has expired according to
changelog.ttl_seconds%',
+ ErrorMsg LIKE '%reason=CHANGELOG_BROKEN%'
+ FROM tasks('type'='mv')
+ WHERE TaskId = '${taskId}'
+ """
+ }
+
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_offset_expired_mv"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_offset_expired_pmv"""
+ sql """DROP TABLE IF EXISTS ivm_offset_expired_t"""
+ sql """DROP TABLE IF EXISTS ivm_offset_expired_f"""
+ sql """DROP TABLE IF EXISTS ivm_offset_expired_d"""
+
+ // 1. The delta reads the expired offset.
+ sql """
+ CREATE TABLE ivm_offset_expired_t (
+ id BIGINT NOT NULL,
+ v1 BIGINT
+ )
+ UNIQUE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ sql """INSERT INTO ivm_offset_expired_t VALUES (1, 10), (2, 20)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ivm_offset_expired_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT id, v1 FROM ivm_offset_expired_t
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_mv INCREMENTAL"""
+ def task = waitForNewTask('ivm_offset_expired_mv', null)
+ qt_baseline_task taskQuery(task.TaskId.toString())
+ order_qt_baseline_mv """SELECT id, v1 FROM ivm_offset_expired_mv ORDER BY
id"""
+
+ // The offset the baseline recorded falls out of the retention once the
TTL is shorter than its age. The row
+ // written after that is what gives the stream something to read, so the
next refresh plans the read.
+ sql """ALTER TABLE ivm_offset_expired_t SET ("changelog.ttl_seconds" =
"1")"""
+ sleep(1500)
+ sql """INSERT INTO ivm_offset_expired_t VALUES (3, 30)"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_mv INCREMENTAL"""
+ task = waitForNewTask('ivm_offset_expired_mv', task.TaskId.toString())
+ qt_strict_task taskQuery(task.TaskId.toString())
+ order_qt_mv_after_strict """SELECT id, v1 FROM ivm_offset_expired_mv ORDER
BY id"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_mv AUTO"""
+ task = waitForNewTask('ivm_offset_expired_mv', task.TaskId.toString())
+ qt_fallback_task taskQuery(task.TaskId.toString())
+ order_qt_mv_after_fallback """SELECT id, v1 FROM ivm_offset_expired_mv
ORDER BY id"""
+
+ // With the TTL back to a day, the offsets the rebuild left are well
inside the retention again.
+ sql """ALTER TABLE ivm_offset_expired_t SET ("changelog.ttl_seconds" =
"86400")"""
+ sql """INSERT INTO ivm_offset_expired_t VALUES (4, 40)"""
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_mv AUTO"""
+ task = waitForNewTask('ivm_offset_expired_mv', task.TaskId.toString())
+ qt_task_after_fallback taskQuery(task.TaskId.toString())
+ order_qt_mv_after_a_further_refresh """SELECT id, v1 FROM
ivm_offset_expired_mv ORDER BY id"""
+
+ // 2. The rebuild before the delta reads the expired offset. The MV is
partitioned by the fact table and joins a
+ // dimension: rebuilding some of its partitions is a read of a subset of
the MV, which reads the dimension
+ // through its stream at the offset.
+ sql """
+ CREATE TABLE ivm_offset_expired_f (
+ k BIGINT NOT NULL,
+ dt DATE NOT NULL,
+ amount INT
+ )
+ UNIQUE KEY(k, dt)
+ PARTITION BY RANGE(dt) (
+ PARTITION p202601 VALUES [('2026-01-01'), ('2026-02-01')),
+ PARTITION p202602 VALUES [('2026-02-01'), ('2026-03-01'))
+ )
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ sql """
+ CREATE TABLE ivm_offset_expired_d (
+ k BIGINT NOT NULL,
+ v INT
+ )
+ UNIQUE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "enable_unique_key_merge_on_write" = "true",
+ "changelog.enable" = "true",
+ "changelog.need_historical_value" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ sql """INSERT INTO ivm_offset_expired_f VALUES (1, '2026-01-10', 100), (2,
'2026-02-10', 200)"""
+ sql """INSERT INTO ivm_offset_expired_d VALUES (1, 10), (2, 10)"""
+
+ sql """
+ CREATE MATERIALIZED VIEW ivm_offset_expired_pmv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ KEY(k, dt)
+ PARTITION BY(dt)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT f.k, f.dt, f.amount, d.v
+ FROM ivm_offset_expired_f f JOIN ivm_offset_expired_d d ON f.k = d.k
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_pmv COMPLETE"""
+ task = waitForNewTask('ivm_offset_expired_pmv', null)
+ order_qt_partitioned_baseline_mv """SELECT k, dt, amount, v FROM
ivm_offset_expired_pmv ORDER BY k"""
+
+ // The dimension changes after its offset and the offset expires; the
truncate is a change no delta can
+ // repair, so the MV partition it reaches needs a rebuild.
+ sql """INSERT INTO ivm_offset_expired_d VALUES (1, 20), (2, 20)"""
+ sql """ALTER TABLE ivm_offset_expired_d SET ("changelog.ttl_seconds" =
"1")"""
+ sleep(1500)
+ sql """TRUNCATE TABLE ivm_offset_expired_f PARTITION(p202601)"""
+ sql """INSERT INTO ivm_offset_expired_f VALUES (1, '2026-01-10', 100)"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_pmv INCREMENTAL"""
+ task = waitForNewTask('ivm_offset_expired_pmv', task.TaskId.toString())
+ qt_rebuild_strict_task taskQuery(task.TaskId.toString())
+ order_qt_partitioned_mv_after_strict """SELECT k, dt, amount, v FROM
ivm_offset_expired_pmv ORDER BY k"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_pmv AUTO"""
+ task = waitForNewTask('ivm_offset_expired_pmv', task.TaskId.toString())
+ qt_rebuild_fallback_task taskQuery(task.TaskId.toString())
+ order_qt_partitioned_mv_after_fallback """SELECT k, dt, amount, v FROM
ivm_offset_expired_pmv ORDER BY k"""
+
+ sql """ALTER TABLE ivm_offset_expired_d SET ("changelog.ttl_seconds" =
"86400")"""
+ sql """INSERT INTO ivm_offset_expired_d VALUES (1, 30)"""
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_pmv AUTO"""
+ task = waitForNewTask('ivm_offset_expired_pmv', task.TaskId.toString())
+ qt_partitioned_task_after_fallback taskQuery(task.TaskId.toString())
+ order_qt_partitioned_mv_after_a_further_refresh """
+ SELECT k, dt, amount, v FROM ivm_offset_expired_pmv ORDER BY k
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
