Repository navigation
[fix](ivm) Fall back to COMPLETE when an IVM stream offset has expired - #69772
Merged
Merged
Conversation
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
An IVM materialized view reads each unique-key (merge-on-write) base table
through an internal table stream created with MIN_DELTA. Planning a MIN_DELTA
read refuses to start from an offset that the base table's
`changelog.ttl_seconds` has expired: `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, and this message was not one of
them. `IvmIncrRefreshManager.doRefresh` rethrew it, so the task failed even when
the refresh allowed a fallback (`REFRESH ... AUTO`, `INCREMENTAL FALLBACK`, the
scheduled refresh of a `REFRESH INCREMENTAL FALLBACK` MV). The offset does not
move on a failed refresh, so while the TTL stays as it is every later refresh
fails the same way, until someone runs a manual COMPLETE refresh.
`IvmFailureReason.CHANGELOG_BROKEN` had no producer in production code, although
`MTMVTask.handleIvmFallbackResult` already sends it straight to COMPLETE.
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, and that refresh reads
the tables the MV does not partition by through their streams at the offsets
(`StreamReadMode.SNAPSHOT`). With such an offset expired, the rebuild failed
with the same message before the delta ran, and that failure went straight out
of the attempt as well.
How to reproduce: create a base table with the changelog enabled and a large
TTL, build an IVM MV on it, then set `changelog.ttl_seconds` to 1, wait two
seconds, insert a row and run `REFRESH MATERIALIZED VIEW mv AUTO`.
| Refresh after the offset expired | Before | After |
| --- | --- | --- |
| `REFRESH ... INCREMENTAL` (strict) | FAILED, `IvmFallbackReason` empty | FAILED, `IvmFallbackReason = CHANGELOG_BROKEN`, the error says `reason=CHANGELOG_BROKEN` and keeps the offset and the cutoff TSO |
| `REFRESH ... AUTO` | FAILED, the 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` (a strict request fails with that reason) |
The fix classifies that message as `CHANGELOG_BROKEN`, matching the constant
`ChangelogUtils` throws with rather than a copy of it, and the rebuild before
the delta answers that failure through `handleIvmFallbackResult` the way the
delta does; any other rebuild failure still fails the attempt. COMPLETE is the only
attempt that resets the stream baselines, which is why the reason goes there
directly instead of trying a partition refresh first: a partition refresh still
reads the streams at their offsets. The comment and the log of that branch said
"previous run incomplete", which is not what the reason means; they now describe
the expired changelog, and the log carries the detail with both TSOs.
### Release note
Fix an IVM materialized view refresh failing on every run after a base
table's `changelog.ttl_seconds` expired the stream offset. A refresh that allows
fallback now rebuilds the MV with COMPLETE; a strict INCREMENTAL refresh fails
with reason `CHANGELOG_BROKEN`.
### Check List (For Author)
- Test: Regression test / Unit Test
- Regression test: `mtmv_p0/ivm/test_ivm_changelog_offset_expired` (new,
nonConcurrent), covering the delta and the rebuild before it. Run on a
local shared-nothing cluster. The delta case was first run on the base
commit, where strict and AUTO both FAILED; the rebuild case was reproduced
by hand on a build that had only the classifier change (AUTO FAILED on the
rebuild). With the whole fix the suite passes, twice in a row against the
generated `.out`. Not run on a cloud cluster.
- Unit Test: `IvmIncrRefreshManagerTest` (new case for the expired offset),
`MTMVTaskTest` (two new cases for the rebuild, strict and fallback),
`IvmFailureReasonTest`, `ChangelogUtilsTest`: 94 tests pass.
- Behavior changed: Yes. A refresh that allows fallback recovers with COMPLETE
instead of failing; a strict INCREMENTAL refresh reports `CHANGELOG_BROKEN`.
- Does this need documentation: No
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Contributor
Author
|
/review |
Contributor
Author
|
run buildall |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Reviewed all seven changed files at cc473a3 against the supplied authoritative diff. No new substantiated findings and no existing inline blockers. Review converged after one round: the production, test, and separate recovery-risk reviewers each returned NO_NEW_VALUABLE_FINDINGS; the main review verified and closed every candidate and risk focus.
The inspected flow is: expired MIN_DELTA read in the delta or pre-delta partition rebuild -> preserved root-cause marker -> CHANGELOG_BROKEN -> strict failure, or skip PARTITIONS and rebuild with COMPLETE when fallback is allowed.
Critical checkpoint conclusions:
- Goal and evidence: the classifier and pre-delta handler cover both intended expired-offset failure sites. Added unit cases and regression scenarios check strict refusal, AUTO recovery, and a subsequent incremental refresh; these were inspected, not executed.
- Scope and reuse: the production change is focused and reuses the existing expiration constant and fallback handler.
- Concurrency and locks: no new threads, shared mutable state, lock acquisitions, or lock-order changes. Existing per-MV job serialization, cancellation checks, and epoch ceilings remain intact.
- Lifecycle: execution-context cleanup and rebuilt-partition accounting retain their finally blocks; no new resource or static-initialization lifetime.
- Configuration: none added. The existing table TTL is read during planning; tests shorten and restore it.
- Compatibility: no persisted field, enum value, function interface, or storage-format change; CHANGELOG_BROKEN already exists.
- Thrift/protobuf: no changes.
- Parallel paths: checked delta, pre-delta rebuild, COMPLETE RESET/SNAPSHOT, manual strict/fallback, scheduled policy resolution, cloud and append-only branches. Standalone PARTITIONS retains its explicitly unchanged prior behavior.
- Conditions and errors: only recognized expiration is converted by the new catch; unrelated rebuild errors propagate. Strict requests are rejected before recovery, with reason and original TSO detail.
- Test coverage: focused unit and end-to-end coverage matches the intended fix. Scheduled triggers, append-only sources, concurrent invalidation, partial commits, and failover were examined statically rather than exercised by the new suite.
- Test results: all 15 expected-output sections match the SQL changes and task reporting. Ordered row checks and normalized task fields avoid variable TSO output. Output generation was not rerun.
- Observability: the existing task fallback-reason field and failure metrics remain; the recovery warning now includes the original detail with MV/task identifiers.
- Persistence and failover: existing durable rebuild requirements precede stream reset, and task-result replay preserves newer invalidations. No new journal or replay path.
- Writes and atomicity: COMPLETE includes the full partition scope, discards held older-image records, and records completed groups only. No new data-loss, double-application, or false-success path was substantiated.
- FE/BE propagation: no new transmitted variables or protocol fields.
- Performance and other concerns: marker matching is limited to an exception path; no new hot-path work or unbounded retry. No BE memory ownership, visible-version, or delete-bitmap logic changes. No additional user focus was supplied.
Validation limits: static inspection only, as required by this review contract. No builds, tests, benchmarks, cluster execution, or runtime fault injection were run. No unresolved PR-specific candidates remain.
Review converged after 1 round(s).
yujun777
reviewed
Oct 28, 2026
| // 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) |
Contributor
There was a problem hiding this comment.
This check can state the invariant instead of enumerating the reason, and pass the reason through:
Optional<IvmFailureReason> reason = IvmFailureClassifier.classifyExecutionFailure(detail)
.filter(IvmFailureReason::requiresCompleteRefresh);
if (reason.isEmpty()) {
throw e;
}
return handleIvmFallbackResult(IvmIncrRefreshResult.fallback(reason.get(), detail), request);requiresCompleteRefresh() is exactly the set whose only repair is COMPLETE, which is what makes it valid to turn a rebuild failure into a fallback from inside the rebuild: there is no meaningful PARTITIONS fallback here (the rebuild already is the partition refresh) and the delta retry machinery (MV_PARTITION_NOT_FOUND <-> sync + retry) does not exist on this path. Behaviour is unchanged today, since the classifier can only produce CHANGELOG_BROKEN here; the difference is that the code says why the conversion is safe rather than which reason happened to be reachable, and the synthesized result carries its real reason instead of a hard-coded one.
| @@ -1184,11 +1196,12 @@ private AttemptResultType handleIvmFallbackResult(IvmIncrRefreshResult ivmResult | |||
| + ", detail=" + ivmResult.getDetailMessage()); | |||
| } | |||
| if (ivmResult.getFailureReason() == IvmFailureReason.CHANGELOG_BROKEN) { | |||
Contributor
There was a problem hiding this comment.
This branch is redundant now: requiresCompleteRefresh() already returns true for CHANGELOG_BROKEN (IvmFailureReason.java:41-47), and the branch below returns the same FALLBACK_TO_COMPLETE while its log already prints reason={} and detail={}. The only difference is the log wording, so I would drop this branch entirely and keep the single requiresCompleteRefresh() branch. The explanation of why no partition attempt is tried before COMPLETE (a partition refresh still reads the streams at their offsets) still reads well as a comment on that generic branch, or next to the classifier entry that produces the reason. If the catch above is also filtered on requiresCompleteRefresh() (see the other comment), nothing remains here that is specific to CHANGELOG_BROKEN.
yujun777
self-requested a review
October 28, 2026 13:12
Contributor
TPC-H: Total hot run time: 28432 ms |
Contributor
TPC-DS: Total hot run time: 151080 ms |
Contributor
ClickBench: Total hot run time: 24.28 s |
morningman
added a commit
that referenced
this pull request
Oct 29, 2026
…cate-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.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary:
In short. Once a base table's
changelog.ttl_secondsexpires 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 asCHANGELOG_BROKEN, so a refresh that may fall back rebuilds the MV with COMPLETE (the only attempt that resets the stream baselines) and a strictINCREMENTALrefresh 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 aMIN_DELTAread refuses an offset that is older than what the table still retains:ChangelogUtils.effectiveStartTsothrowsChangelog offset has expired according to changelog.ttl_seconds. The incremental refresh turns a failure into a fallback only whenIvmFailureClassifierrecognizes its message.The problem, and what it cost.
IvmIncrRefreshManager.doRefreshrethrew it and the task failed even forREFRESH ... AUTO,INCREMENTAL FALLBACK, or the scheduled refresh of aREFRESH INCREMENTAL FALLBACKMV.TRUNCATEof 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_BROKENhad no producer in production code, althoughMTMVTask.handleIvmFallbackResultalready sends it straight to COMPLETE; the comment and log of that branch described it as "previous run incomplete".How this PR fixes it.
IvmFailureClassifiermaps a failure carryingChangelogUtils.CHANGELOG_OFFSET_EXPIREDtoCHANGELOG_BROKEN(matching the constant the read throws with, not a copy of it).MTMVTask.executeIvmAttemptanswers that failure of the rebuild before the delta throughhandleIvmFallbackResult, the way the delta's failure is answered; any other rebuild failure still fails the attempt.CHANGELOG_BROKENbranch ofhandleIvmFallbackResultnow 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 ... INCREMENTAL(strict)IvmFallbackReasonemptyIvmFallbackReason = CHANGELOG_BROKEN, error saysreason=CHANGELOG_BROKENwith both TSOsREFRESH ... AUTOIvmFallbackReason = CHANGELOG_BROKENREFRESH ... AUTOREFRESH ... AUTOthat first rebuilds a dirty MV partition while a joined table's offset has expiredIvmFallbackReason = CHANGELOG_BROKENClasses, and how they call each other.
MTMVTask.run→executeIvmAttempt→ (dirty partitions)executePartitionBasedRefresh→ on an expired offsethandleIvmFallbackResult(CHANGELOG_BROKEN)executeIvmAttempt→executeSingleIvmAttempt→IvmIncrRefreshManager.doRefresh→IvmFailureClassifier.classifyExecutionFailure→IvmIncrRefreshResult.fallback(CHANGELOG_BROKEN)→handleIvmFallbackResulthandleIvmFallbackResult: strict request →JobExceptionwith the reason; otherwiseFALLBACK_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.Release note
Fix an IVM materialized view refresh failing on every run after a base table's
changelog.ttl_secondsexpired the stream offset. A refresh that allows fallback now rebuilds the MV with COMPLETE; a strict INCREMENTAL refresh fails with reasonCHANGELOG_BROKEN.Check List (For Author)
mtmv_p0/ivm/test_ivm_changelog_offset_expired(new, nonConcurrent), covering the delta and the rebuild before it. Local shared-nothing cluster: the delta case was first run on the base commit (strict and AUTO both FAILED); the rebuild case was reproduced by hand on a build with only the classifier change (AUTO FAILED on the rebuild). With the whole fix the suite passes twice in a row against the generated.out. Not run on a cloud cluster.IvmIncrRefreshManagerTest(new case),MTMVTaskTest(two new cases, strict and fallback),IvmFailureReasonTest,ChangelogUtilsTest: 94 tests pass; each new case fails when its half of the fix is removed.CHANGELOG_BROKEN.