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 468afdc2a52 [fix](ivm) Reject an expired stream offset for an IVM read
of a duplicate-key table (#69785)
468afdc2a52 is described below
commit 468afdc2a52d696ef974d7e4761cc47e86ce0be4
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 29 04:00:20 2026 +0000
[fix](ivm) Reject an expired stream offset for an IVM read of a
duplicate-key table (#69785)
### What problem does this PR solve?
Issue Number: None
Related PR: #69772
Problem Summary:
**In short.** An IVM materialized view over a duplicate-key base table
silently misses rows once the table's `changelog.ttl_seconds` has
expired its stream offset: the refresh succeeds, the rows written
between the offset and the cutoff are never read, and a base-table query
of the MV's shape that transparent rewrite answers from the MV returns
the same wrong result. This PR makes every read of an IVM refresh reject
an expired offset, whatever its scan type, so the refresh fails with
`CHANGELOG_BROKEN` or, when it may fall back, rebuilds the MV with
COMPLETE (#69772). It also carries the two follow-ups yujun777 suggested
on #69772.
**Background.** An IVM MV reads each base table through an internal
table stream. A duplicate-key table is read `APPEND_ONLY`:
`OlapTableStreamWrapper.getStreamScanType()` returns `APPEND_ONLY` for
`DUP_KEYS`, whatever type the stream declares.
`NormalizeOlapTableStreamScan.applyChangelogTtl` decides what an expired
start means: a `MIN_DELTA` read is rejected (#69772 turns that into
`CHANGELOG_BROKEN` and COMPLETE), any other read is clipped to the first
TSO the changelog retains. The clipping is right for a user stream,
which reads the changelog it retains.
**The problem, and what it cost.**
- For an IVM refresh the clipping loses data without a trace. The rows
written between the offset and the cutoff are never read, the refresh
succeeds, the offset moves past them, and no later incremental refresh
brings them back.
- It hits every IVM MV over a duplicate-key table, the aggregate ones
(unique-key MV) and the append-only detail ones alike, whenever the MV
is not refreshed within the TTL of a write.
**How this PR fixes it.**
1. `applyChangelogTtl` rejects an expired start for every read of an IVM
statement (the statement carries an `IvmRewriteContext`), whatever the
scan type. The condition is the statement, not the type the stream
declares: a user stream on a duplicate-key table also declares
`MIN_DELTA` by default, and it keeps its clipping. An IVM statement
never reads a user stream: an MV query that scans a stream is rejected
when the MV is created. The snapshot read of a duplicate-key table is
bounded by the offset on the base table itself and reads no changelog,
so it is unchanged.
2. The follow-ups from the review of #69772, in their own commit and
without a behavior change:
- the rebuild an incremental attempt runs before its delta now turns a
failure into a fallback when its reason's only repair is COMPLETE
(`IvmFailureReason.requiresCompleteRefresh()`), and passes that reason
through, instead of naming `CHANGELOG_BROKEN`;
- the `CHANGELOG_BROKEN` branch of `handleIvmFallbackResult`, which
returned the same result as the `requiresCompleteRefresh()` branch after
it, is removed; why only COMPLETE repairs an expired changelog now sits
on `IvmFailureReason.CHANGELOG_BROKEN`.
**Results** (local shared-nothing cluster; a duplicate-key table, a row
written after the MV's offset, then `changelog.ttl_seconds` set to 1 and
two seconds of waiting):
| Refresh after the offset expired | Before | After |
| --- | --- | --- |
| `REFRESH ... INCREMENTAL` (strict) | SUCCESS, the row missing from the
MV | FAILED, `IvmFallbackReason = CHANGELOG_BROKEN`, the MV keeps its
rows |
| `REFRESH ... AUTO` | SUCCESS with nothing left to read; the row stays
missing | SUCCESS through COMPLETE, `IvmFallbackReason =
CHANGELOG_BROKEN`, the MV matches the base table |
| the next refresh, with the TTL back to a day | incremental; the row
missing for good | incremental |
| a user stream on the same table | reads from the first retained TSO |
the same |
Both an aggregate MV and an append-only detail MV over the table behave
as above.
**Classes, and how they call each other.**
- `IvmIncrRefreshMTMV` → `IvmDeltaRewriteState.createDeltaScan` →
`LogicalOlapTableStreamScan` →
`NormalizeOlapTableStreamScan.makeTableStreamScan` →
`makeIncrementalScanFromChangelog` → `applyChangelogTtl` →
`ChangelogUtils.effectiveStartTso`, which now rejects for an IVM
statement.
- The rejection reaches `IvmIncrRefreshManager.doRefresh` →
`IvmFailureClassifier.classifyExecutionFailure` → `CHANGELOG_BROKEN` →
`MTMVTask.handleIvmFallbackResult` → strict: failure with the reason;
otherwise `FALLBACK_TO_COMPLETE`.
- The rebuild before the delta: `MTMVTask.executeIvmAttempt` →
`executePartitionBasedRefresh` → on a failure whose reason
`requiresCompleteRefresh()`, `handleIvmFallbackResult` with that reason.
Not changed: user streams, direct incremental queries
(`BindRelation.applyChangelogTtl`), the snapshot and reset reads, and
anything persisted. A partition with no stream offset is a separate
problem, fixed by #69778.
---
.../apache/doris/job/extensions/mtmv/MTMVTask.java | 21 +--
.../apache/doris/mtmv/ivm/IvmFailureReason.java | 10 +-
.../rewrite/NormalizeOlapTableStreamScan.java | 8 +-
.../java/org/apache/doris/mtmv/MTMVTaskTest.java | 37 ++++-
.../trees/plans/ExplainTableStreamPlanTest.java | 55 ++++++
.../ivm/test_ivm_changelog_offset_expired_dup.out | 75 +++++++++
.../test_ivm_changelog_offset_expired_dup.groovy | 184 +++++++++++++++++++++
7 files changed, 366 insertions(+), 24 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 86d8661db1d..302ecd0972b 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
@@ -832,14 +832,16 @@ public class MTMVTask extends AbstractTask {
} 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.
+ // below. Only a failure whose one repair is COMPLETE becomes
a fallback here: the rebuild already
+ // is the partition refresh, and the sync-and-retry the delta
has for a missing MV partition does
+ // not exist on this path. Any other failure still fails the
attempt.
String detail = Util.getRootCauseMessage(e);
- if (IvmFailureClassifier.classifyExecutionFailure(detail)
- .filter(reason -> reason ==
IvmFailureReason.CHANGELOG_BROKEN).isEmpty()) {
+ Optional<IvmFailureReason> reason =
IvmFailureClassifier.classifyExecutionFailure(detail)
+ .filter(IvmFailureReason::requiresCompleteRefresh);
+ if (reason.isEmpty()) {
throw e;
}
- return handleIvmFallbackResult(
-
IvmIncrRefreshResult.fallback(IvmFailureReason.CHANGELOG_BROKEN, detail),
request);
+ return
handleIvmFallbackResult(IvmIncrRefreshResult.fallback(reason.get(), detail),
request);
} finally {
recordRebuiltPartitions(request);
}
@@ -1195,15 +1197,6 @@ public class MTMVTask extends AbstractTask {
+ ", reason=" + ivmResult.getFailureReason()
+ ", detail=" + ivmResult.getDetailMessage());
}
- if (ivmResult.getFailureReason() == IvmFailureReason.CHANGELOG_BROKEN)
{
- // 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()) {
LOG.warn("IVM refresh fell back for mv={}, reason={}, detail={},
taskId={}. "
+ "Continuing with COMPLETE refresh.",
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 9b2a5154548..af2f5b358aa 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,7 +20,11 @@ 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. */
+ /**
+ * The changelog a stream has to read from is outside the retention:
changelog.ttl_seconds expired its offset.
+ * A partition refresh still reads the streams at their offsets, so only
COMPLETE, the one refresh that resets
+ * the stream baselines, repairs it.
+ */
CHANGELOG_BROKEN,
// ATTN(changelog-rename): formerly "BINLOG_NOT_ENABLED"
CHANGELOG_NOT_ENABLED,
@@ -39,6 +43,10 @@ public enum IvmFailureReason {
MV_COMMIT_NOT_VISIBLE,
MV_PARTITION_NOT_FOUND;
+ /**
+ * Whether COMPLETE is the only refresh that repairs this failure, so that
a refresh allowed to fall back goes
+ * to it directly instead of trying a partition refresh first.
+ */
public boolean requiresCompleteRefresh() {
return this == CHANGELOG_BROKEN
|| this == STREAM_UNSUPPORTED
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
index 6b8092c9f97..22e2f02d7b5 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/NormalizeOlapTableStreamScan.java
@@ -239,10 +239,16 @@ public class NormalizeOlapTableStreamScan extends
OneRewriteRuleFactory {
long cutoffTso = TSOTimestamp.calculateCutoff(
cascadesContext.getStatementContext().getChangelogReferenceTso(),
table.getChangelogConfig().getTtlSeconds());
+ // An expired start is rejected for MIN_DELTA and clipped to the first
retained TSO otherwise. An IVM read
+ // rejects it whatever its scan type: a duplicate-key base table is
read APPEND_ONLY, and a clipped start
+ // would skip the rows written between the offset and the cutoff while
the refresh succeeds. The rejection
+ // is what sends the refresh to COMPLETE. A user stream keeps what its
scan type does.
+ boolean rejectExpiredStart = scanType ==
BaseTableStream.StreamScanType.MIN_DELTA
+ || cascadesContext.getStatementContext().isIvmMTMVRewrite();
return
offsetMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, entry
-> {
Pair<Long, Long> range = entry.getValue();
return Pair.of(ChangelogUtils.effectiveStartTso(range.first,
cutoffTso,
- range.first != null && scanType ==
BaseTableStream.StreamScanType.MIN_DELTA), range.second);
+ range.first != null && rejectExpiredStart), range.second);
}));
}
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 006ff1fd780..f14e0b45d40 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
@@ -45,6 +45,7 @@ import
org.apache.doris.mtmv.MTMVPartitionInfo.MTMVPartitionType;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVRefreshState;
import org.apache.doris.mtmv.MTMVRefreshEnum.MTMVState;
import org.apache.doris.mtmv.MTMVRefreshEnum.RefreshMethod;
+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;
@@ -1787,7 +1788,7 @@ public class MTMVTaskTest {
MTMVTask task = new MTMVTask(mtmv, relation,
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
RefreshMode.INCREMENTAL, true, null));
- Object result = runIvmAttemptWhoseRebuildReadsAnExpiredOffset(task);
+ Object result = runIvmAttemptWhoseRebuildFails(task,
expiredOffsetFailure());
Assertions.assertEquals("FALLBACK_TO_COMPLETE", result.toString());
Assertions.assertEquals(IvmFailureReason.CHANGELOG_BROKEN.name(),
@@ -1801,7 +1802,7 @@ public class MTMVTaskTest {
RefreshMode.INCREMENTAL, false, null));
JobException failure = Assertions.assertThrows(JobException.class,
- () -> runIvmAttemptWhoseRebuildReadsAnExpiredOffset(task));
+ () -> runIvmAttemptWhoseRebuildFails(task,
expiredOffsetFailure()));
Assertions.assertTrue(failure.getMessage().contains("reason=" +
IvmFailureReason.CHANGELOG_BROKEN));
Assertions.assertTrue(failure.getMessage().contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED));
@@ -1810,10 +1811,32 @@ public class MTMVTaskTest {
}
/**
- * 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.
+ * Only a rebuild failure whose one repair is COMPLETE becomes a fallback.
A missing MV partition is not one: the
+ * sync-and-retry that answers it exists on the delta's path only, so it
still fails the attempt, even for a
+ * request that may fall back.
*/
- private Object runIvmAttemptWhoseRebuildReadsAnExpiredOffset(MTMVTask
task) throws Exception {
+ @Test
+ public void testRebuildFailureCompleteDoesNotRepairStillFailsTheAttempt() {
+ MTMVTask task = new MTMVTask(mtmv, relation,
MTMVTaskContext.of(MTMVTaskTriggerMode.MANUAL, null,
+ RefreshMode.INCREMENTAL, true, null));
+
+ JobException failure = Assertions.assertThrows(JobException.class, ()
-> runIvmAttemptWhoseRebuildFails(
+ task, new
RuntimeException(IvmFailureClassifier.MV_PARTITION_NOT_FOUND_MSG)));
+
+
Assertions.assertTrue(failure.getMessage().contains(IvmFailureClassifier.MV_PARTITION_NOT_FOUND_MSG));
+ Assertions.assertNull(Deencapsulation.getField(task,
"ivmFallbackReason"));
+ }
+
+ // The failure planning raises for a MIN_DELTA read from an expired offset.
+ private static RuntimeException expiredOffsetFailure() {
+ return Assertions.assertThrows(RuntimeException.class, () ->
ChangelogUtils.effectiveStartTso(99L, 100L, true));
+ }
+
+ /**
+ * Runs the incremental attempt with p1 needing a rebuild and p2 caught
up, the rebuild's read failing with
+ * {@code failure}. The delta must not be reached.
+ */
+ private Object runIvmAttemptWhoseRebuildFails(MTMVTask task,
RuntimeException failure) throws Exception {
Mockito.when(mtmv.isIvm()).thenReturn(true);
Mockito.when(mtmv.getName()).thenReturn("test_mv");
Mockito.when(mtmv.hasRefreshSnapshot()).thenReturn(true);
@@ -1828,8 +1851,6 @@ public class MTMVTaskTest {
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));
@@ -1847,7 +1868,7 @@ public class MTMVTaskTest {
.thenReturn(command);
mtmvPlanUtilStatic.when(() -> MTMVPlanUtil.executeCommand(
Mockito.eq(mtmvCtx), Mockito.eq(command),
Mockito.any(StatementContext.class),
- Mockito.anyString(),
Mockito.any(Consumer.class))).thenThrow(expired);
+ Mockito.anyString(),
Mockito.any(Consumer.class))).thenThrow(failure);
Object request = Deencapsulation.invoke(task,
"resolveRefreshRequest");
try {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
index 1ff0b84099f..9b25c8d442c 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/ExplainTableStreamPlanTest.java
@@ -182,6 +182,19 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
"create stream if not exists test_stream.s_dup on table
test_stream.tbl_dup_stream_base\n"
+ "properties('show_initial_rows' = 'false')";
createTable(createDuplicateStream);
+
+ // A stream on a duplicate-key table with a TTL, and data after its
offset. bumpPartitionsAndReplicas
+ // stamps a wall-clock millisecond as the partition TSO, far below a
TSO composed from the current time,
+ // so under such a reference the offset is expired.
+ createTable("create table test_stream.tbl_dup_stream_ttl_offset (k1
int, k2 int) "
+ + "duplicate key(k1) distributed by hash(k1) buckets 1
properties("
+ + "\"replication_num\"=\"1\",\"changelog.enable\"=\"true\","
+ + "\"changelog.ttl_seconds\"=\"10\")");
+ OlapTable duplicateTtlTable = (OlapTable)
db.getTableOrMetaException("tbl_dup_stream_ttl_offset");
+ bumpPartitionsAndReplicas(duplicateTtlTable, 1001L);
+ createTable("create stream if not exists test_stream.s_dup_ttl"
+ + " on table test_stream.tbl_dup_stream_ttl_offset
properties('show_initial_rows' = 'false')");
+ bumpPartitionsAndReplicas(duplicateTtlTable, 2002L);
}
private static void bumpPartitionsAndReplicas(OlapTable table, long
newVersion) {
@@ -797,6 +810,48 @@ public class ExplainTableStreamPlanTest extends
TestWithFeService {
Assertions.assertTrue(exception.getMessage().contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED));
}
+ @Test
+ public void testChangelogTtlClipsExpiredAppendOnlyStreamOffset() {
+ // A user stream on a duplicate-key table is read APPEND_ONLY: its
expired offset is clipped to the first
+ // retained TSO, and the read goes on from there.
+ long referenceTso =
TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis());
+ long firstRetainedTso =
TSOTimestamp.nextTso(TSOTimestamp.calculateCutoff(referenceTso, 10));
+ Plan plan = normalizeStreamRead("select * from test_stream.s_dup_ttl",
referenceTso, Optional.empty());
+
+ Set<LogicalOlapScan> changelogScans = plan.collect(node -> node
instanceof LogicalOlapScan
+ && ((LogicalOlapScan) node).getTable() instanceof
OlapChangelogTableWrapper);
+ Assertions.assertEquals(1, changelogScans.size());
+ OlapChangelogTableWrapper wrapper = (OlapChangelogTableWrapper)
changelogScans.iterator().next().getTable();
+ Assertions.assertFalse(wrapper.getPartitionIds().isEmpty());
+ for (Long partitionId : wrapper.getPartitionIds()) {
+ Assertions.assertEquals(firstRetainedTso,
wrapper.getPartitionOffset(partitionId).first);
+ }
+ }
+
+ @Test
+ public void testChangelogTtlRejectsExpiredAppendOnlyStreamOffsetForIvm() {
+ // The same read inside an IVM refresh is rejected: clipped, it would
leave out the rows written between
+ // the offset and the cutoff while the refresh succeeds.
+ long referenceTso =
TSOTimestamp.composePhysicalTimestamp(System.currentTimeMillis());
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class,
+ () -> normalizeStreamRead("select * from
test_stream.s_dup_ttl", referenceTso,
+
Optional.of(IvmRewriteContext.incremental(Mockito.mock(MTMV.class)))));
+
Assertions.assertTrue(exception.getMessage().contains(ChangelogUtils.CHANGELOG_OFFSET_EXPIRED),
+ exception.getMessage());
+ }
+
+ private Plan normalizeStreamRead(String sql, long referenceTso,
Optional<IvmRewriteContext> ivmRewriteContext) {
+ PlanChecker checker = PlanChecker.from(connectContext).parse(sql);
+ StatementContext statementContext =
checker.getCascadesContext().getStatementContext();
+ // Direct rewrite tests bypass the planner's statement reference
initialization.
+ statementContext.getOrRegisterChangelogReferenceTso(() ->
referenceTso);
+ checker.analyze();
+ // Installed after the analysis, whose IVM rules would rewrite the
query as an MV's: only how the stream
+ // read treats its offset inside an IVM statement is under test.
+ statementContext.setIvmRewriteContext(ivmRewriteContext);
+ return checker.applyTopDown(new
NormalizeOlapTableStreamScan()).getCascadesContext().getRewritePlan();
+ }
+
@Test
public void testChangelogTtlDisablesQueryCache() throws Exception {
ConnectContext ctx = createDefaultCtx();
diff --git
a/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.out
b/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.out
new file mode 100644
index 00000000000..69a07645164
--- /dev/null
+++ b/regression-test/data/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.out
@@ -0,0 +1,75 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !agg_baseline_task --
+SUCCESS COMPLETE NONE false false
+
+-- !agg_baseline_mv --
+1 1 10
+2 1 20
+
+-- !detail_baseline_task --
+SUCCESS COMPLETE NONE false false
+
+-- !detail_baseline_mv --
+1 1 10
+2 2 20
+
+-- !stream_before_expiry --
+3 1 30
+
+-- !stream_after_expiry --
+
+-- !agg_strict_task --
+FAILED NONE CHANGELOG_BROKEN true true
+
+-- !agg_mv_after_strict --
+1 1 10
+2 1 20
+
+-- !agg_fallback_task --
+SUCCESS COMPLETE CHANGELOG_BROKEN false false
+
+-- !agg_mv_after_fallback --
+1 2 40
+2 1 20
+
+-- !detail_strict_task --
+FAILED NONE CHANGELOG_BROKEN true true
+
+-- !detail_mv_after_strict --
+1 1 10
+2 2 20
+
+-- !detail_fallback_task --
+SUCCESS COMPLETE CHANGELOG_BROKEN false false
+
+-- !detail_mv_after_fallback --
+1 1 10
+2 2 20
+3 1 30
+
+-- !agg_task_after_fallback --
+SUCCESS NONE NONE false false
+
+-- !agg_mv_after_a_further_refresh --
+1 2 40
+2 2 60
+
+-- !detail_task_after_fallback --
+SUCCESS NONE NONE false false
+
+-- !detail_mv_after_a_further_refresh --
+1 1 10
+2 2 20
+3 1 30
+4 2 40
+
+-- !base_agg --
+1 2 40
+2 2 60
+
+-- !base_detail --
+1 1 10
+2 2 20
+3 1 30
+4 2 40
+
diff --git
a/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.groovy
b/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.groovy
new file mode 100644
index 00000000000..34dd356b12d
--- /dev/null
+++
b/regression-test/suites/mtmv_p0/ivm/test_ivm_changelog_offset_expired_dup.groovy
@@ -0,0 +1,184 @@
+// 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 of an IVM MV over a duplicate-key base table whose
stream offset the table's
+ * changelog.ttl_seconds has expired.
+ *
+ * <p>A duplicate-key table is read through its stream APPEND_ONLY. A user
stream read that way clips an expired
+ * start to the changelog the table still retains and reads on from there. An
IVM read may not: the rows written
+ * between the offset and the cutoff would be missing from an MV whose refresh
succeeded. The read refuses the
+ * offset instead, the way a MIN_DELTA read does, and the refresh answers it
as CHANGELOG_BROKEN, which only
+ * COMPLETE can repair.
+ *
+ * <p>Cases pinned here, for an aggregate MV and for an append-only detail MV
over the same table:
+ * <ol>
+ * <li>a strict INCREMENTAL refresh fails with that reason and leaves the MV
as it was;</li>
+ * <li>an AUTO refresh, which may fall back, rebuilds the MV with COMPLETE,
the row written before the offset
+ * expired included, and records the reason;</li>
+ * <li>with the TTL back to a day, the refresh after it is incremental
again.</li>
+ * </ol>
+ * A user stream on the table still reads past its expired offset, from the
changelog retained.
+ *
+ * <p>nonConcurrent: the offsets expire by sleeping past a one-second TTL.
+ */
+suite("test_ivm_changelog_offset_expired_dup", "nonConcurrent") {
+ // The base-table queries below check the MVs, so they must not be
answered from an MV; and a base-table
+ // insert must not be redirected into one, or the incremental path under
test goes unused.
+ sql "SET enable_materialized_view_rewrite = false"
+ sql "SET enable_dml_materialized_view_rewrite = false"
+
+ 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_dup_agg_mv"""
+ sql """DROP MATERIALIZED VIEW IF EXISTS ivm_offset_expired_dup_detail_mv"""
+ sql """DROP STREAM IF EXISTS ivm_offset_expired_dup_stream FORCE"""
+ sql """DROP TABLE IF EXISTS ivm_offset_expired_dup_t"""
+
+ sql """
+ CREATE TABLE ivm_offset_expired_dup_t (
+ id BIGINT NOT NULL,
+ grp INT NOT NULL,
+ v1 BIGINT
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "changelog.enable" = "true",
+ "changelog.ttl_seconds" = "86400"
+ )
+ """
+ sql """INSERT INTO ivm_offset_expired_dup_t VALUES (1, 1, 10), (2, 2,
20)"""
+
+ // The aggregate keeps this MV on the unique-key model; the plain
projection makes the other an append-only
+ // detail MV. Both read the table through an internal stream of their own.
+ sql """
+ CREATE MATERIALIZED VIEW ivm_offset_expired_dup_agg_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY HASH(grp) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT grp, COUNT(*) AS cnt, SUM(v1) AS total FROM
ivm_offset_expired_dup_t GROUP BY grp
+ """
+ sql """
+ CREATE MATERIALIZED VIEW ivm_offset_expired_dup_detail_mv
+ BUILD DEFERRED REFRESH INCREMENTAL ON MANUAL
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("replication_num" = "1")
+ AS SELECT id, grp, v1 FROM ivm_offset_expired_dup_t
+ """
+ sql """
+ CREATE STREAM ivm_offset_expired_dup_stream ON TABLE
ivm_offset_expired_dup_t
+ PROPERTIES ("show_initial_rows" = "false")
+ """
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_agg_mv
INCREMENTAL"""
+ def aggTask = waitForNewTask('ivm_offset_expired_dup_agg_mv', null)
+ qt_agg_baseline_task taskQuery(aggTask.TaskId.toString())
+ order_qt_agg_baseline_mv """SELECT grp, cnt, total FROM
ivm_offset_expired_dup_agg_mv"""
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_detail_mv
INCREMENTAL"""
+ def detailTask = waitForNewTask('ivm_offset_expired_dup_detail_mv', null)
+ qt_detail_baseline_task taskQuery(detailTask.TaskId.toString())
+ order_qt_detail_baseline_mv """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_detail_mv"""
+
+ // The row written after the offsets is what a read from the retained
changelog leaves out once the TTL is
+ // shorter than its age.
+ sql """INSERT INTO ivm_offset_expired_dup_t VALUES (3, 1, 30)"""
+ order_qt_stream_before_expiry """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_stream"""
+ sql """ALTER TABLE ivm_offset_expired_dup_t SET ("changelog.ttl_seconds" =
"1")"""
+ sleep(1500)
+
+ // A user stream reads past its expired offset: the read starts at the
first retained TSO, which is after
+ // the row it has not consumed.
+ order_qt_stream_after_expiry """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_stream"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_agg_mv
INCREMENTAL"""
+ aggTask = waitForNewTask('ivm_offset_expired_dup_agg_mv',
aggTask.TaskId.toString())
+ qt_agg_strict_task taskQuery(aggTask.TaskId.toString())
+ order_qt_agg_mv_after_strict """SELECT grp, cnt, total FROM
ivm_offset_expired_dup_agg_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_agg_mv AUTO"""
+ aggTask = waitForNewTask('ivm_offset_expired_dup_agg_mv',
aggTask.TaskId.toString())
+ qt_agg_fallback_task taskQuery(aggTask.TaskId.toString())
+ order_qt_agg_mv_after_fallback """SELECT grp, cnt, total FROM
ivm_offset_expired_dup_agg_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_detail_mv
INCREMENTAL"""
+ detailTask = waitForNewTask('ivm_offset_expired_dup_detail_mv',
detailTask.TaskId.toString())
+ qt_detail_strict_task taskQuery(detailTask.TaskId.toString())
+ order_qt_detail_mv_after_strict """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_detail_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_detail_mv AUTO"""
+ detailTask = waitForNewTask('ivm_offset_expired_dup_detail_mv',
detailTask.TaskId.toString())
+ qt_detail_fallback_task taskQuery(detailTask.TaskId.toString())
+ order_qt_detail_mv_after_fallback """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_detail_mv"""
+
+ // With the TTL back to a day, the offsets the rebuilds left are well
inside the retention again.
+ sql """ALTER TABLE ivm_offset_expired_dup_t SET ("changelog.ttl_seconds" =
"86400")"""
+ sql """INSERT INTO ivm_offset_expired_dup_t VALUES (4, 2, 40)"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_agg_mv AUTO"""
+ aggTask = waitForNewTask('ivm_offset_expired_dup_agg_mv',
aggTask.TaskId.toString())
+ qt_agg_task_after_fallback taskQuery(aggTask.TaskId.toString())
+ order_qt_agg_mv_after_a_further_refresh """SELECT grp, cnt, total FROM
ivm_offset_expired_dup_agg_mv"""
+
+ sql """REFRESH MATERIALIZED VIEW ivm_offset_expired_dup_detail_mv AUTO"""
+ detailTask = waitForNewTask('ivm_offset_expired_dup_detail_mv',
detailTask.TaskId.toString())
+ qt_detail_task_after_fallback taskQuery(detailTask.TaskId.toString())
+ order_qt_detail_mv_after_a_further_refresh """SELECT id, grp, v1 FROM
ivm_offset_expired_dup_detail_mv"""
+
+ // What the base table answers for the two MV queries.
+ order_qt_base_agg """SELECT grp, COUNT(*), SUM(v1) FROM
ivm_offset_expired_dup_t GROUP BY grp"""
+ order_qt_base_detail """SELECT id, grp, v1 FROM ivm_offset_expired_dup_t"""
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
