Repository navigation
[fix](ivm) Keep the offsets of idle IVM stream partitions alive - #69831
morningman wants to merge 7 commits into
Conversation
### What problem does this PR solve? Issue Number: None Related PR: apache#69772, apache#69785 Problem Summary: An IVM stream records an offset for a partition when a refresh consumes it, and a refresh only consumes the partitions that have new data, so the offset of a partition nobody writes stays at its last change. The changelog TTL check (`ChangelogUtils.effectiveStartTso`) only compares that offset with the cutoff, `now - changelog.ttl_seconds` (one day by default). Once a partition has been idle for longer than the TTL, the next write to it makes the refresh after it reject the offset as expired, and since apache#69772 the refresh falls back to COMPLETE and rebuilds the whole MV, although nothing it needs is gone: the only changes after the offset are the ones just written, all inside the retention. A dimension table updated weekly, a backfill into an older partition or a quiet weekend is enough. With apache#69785 this also holds for an IVM MV over a duplicate-key table, whose expired offset used to be clipped. This commit records a keep-alive TSO K for such partitions, meaning the partition holds no change in (offset, K]: - `IvmStreamKeepAliveDaemon`, a Master daemon, finds the IVM streams from the MV side (`IvmUtil.getIvmStreams`, which `mv_infos()` now uses too) and runs a round every `ivm_stream_keepalive_interval_second` (600). It checks every second whether the interval has passed, so a change of the interval applies at once. `enable_ivm_stream_keepalive` (true) turns it off. Both configs are mutable. - A partition qualifies when the base table applies a TTL, the stream has a real offset for it past the initial snapshot, the partition holds no change after the offset, and the newer of the offset and K has less than half the TTL left. - Shared-nothing mode (`TableStreamManager.keepAliveIdlePartitions`): K is a TSO taken from the TSO service while the base table read lock is held. A commit takes its TSO inside the base table write lock, so every commit the partition has not had yet gets a larger one. The locks are taken in the order the stale-offset cleanup takes them, and a partition with a committed transaction that is not visible yet does not qualify. - Cloud mode (`TableStreamManager.keepAliveIdleCloudPartitions`): the offsets live in MetaService. K is the committed TSO watermark of the TSO service, taken before the offsets are read, in batches of up to 10000 partitions, through `get_table_stream_offset`, which waits for pending transactions. Every commit at or below the watermark has finished before the read, so a partition whose offset is at its end in the read holds no change up to the watermark, and a commit that lands later gets a larger TSO. MetaService is not changed. - K is recorded in FE metadata in both modes, with the offset it was taken against: a new GSON field of `OlapTableStream`, logged in a new optional entry list of `OP_TABLE_STREAM_CLEANUP` and replayed the same way. A read uses K only when it starts from that offset (`OlapTableStreamWrapper.getKeepAliveTso`), which in cloud mode is the offset the installed read state carries. An offset that moved, or one MetaService reports differently after a restore, leaves K out. K taken against the same offset only grows, and is pruned with the offsets of a dropped partition. - `NormalizeOlapTableStreamScan.applyChangelogTtl` moves the start of a read to after K before the expiration check. The read returns the same rows, since the partition has no change in between, and reads retained changelog only. ### Release note An IVM materialized view no longer falls back to a COMPLETE refresh when a base table partition is written after being idle for longer than `changelog.ttl_seconds`. New FE configs `enable_ivm_stream_keepalive` and `ivm_stream_keepalive_interval_second`. ### Check List (For Author) - Test: Regression test / Unit Test - `test_ivm_stream_keepalive` (new, nonConcurrent): an aggregate MV over a merge-on-write table and an append-only detail MV over a duplicate-key table, a 60-second TTL and a 5-second check. Idle for 70 seconds and then written, the AUTO refresh of both is incremental with no fallback; with the keep-alive off it falls back to COMPLETE as CHANGELOG_BROKEN; a write landing while a round is held between what it read and what it records is read by the refresh after it. Run locally in shared-nothing mode; the whole `mtmv_p0/ivm` directory (116 suites) passes. Not run in cloud mode locally. - `TableStreamManagerKeepAliveTest` (new, 16 cases): the conditions of both modes (a change after the offset, a committed transaction that is not visible, the initial snapshot, the margin, no TTL or one that never expires an offset; in cloud mode a state that is not at its end or has no real offset, a watermark not past the offset, no calibrated TSO, a failed read, reads in batches), K taken against another offset not counting, the wrapper using K only for a read from its offset in both modes, replay, pruning, and state written before K. - `ExplainTableStreamPlanTest`: an IVM read of an expired offset with a fresh K is admitted and starts after K; one whose K expired too, or was taken against another offset, is rejected. - The IVM unit tests, every `catalog/stream` test class and the other classes using the changed code (47 classes, 677 cases) pass; checkstyle passes for fe-core and fe-common. - Behavior changed: Yes. A refresh reads an idle partition incrementally where it used to fall back to COMPLETE (or fail, if strict). - Does this need documentation: Yes, the two FE configs. Co-Authored-By: Claude Opus 5.5 <[email protected]>
|
Thank you for your contribution to Apache Doris. Please clearly describe your PR:
|
…second against changelog.ttl_seconds ### What problem does this PR solve? Issue Number: None Related PR: apache#69831 Problem Summary: The keep-alive daemon renews a partition once less than half of its base table's changelog.ttl_seconds is left, and only runs a round every ivm_stream_keepalive_interval_second, so a keep-alive TSO gets as old as half the TTL plus one interval before it is renewed. It stays valid only while the interval is at most half the TTL. With the defaults (a TTL of a day, an interval of ten minutes) that holds, but a table whose TTL is shorter than twice the interval sees its keep-alive TSO expire between two rounds, and a refresh after an idle period falls back to COMPLETE as it did before the keep-alive. Nothing in the config said so. This commit says so in the description of the config: keep it at most half the shortest changelog.ttl_seconds of the IVM base tables, what happens otherwise, that a shorter interval costs more checks but not more edit logs, and that a change applies within a second. ### Release note None ### Check List (For Author) - Test: No need to test (a config description only); fe-common compiles and its checkstyle passes. - Behavior changed: No - Does this need documentation: Yes, the same rule goes into the user documentation of the config. Co-Authored-By: Claude Opus 5.5 <[email protected]>
|
/review |
|
run buildall |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
Reviewed all 14 changed files at head 754db89. Request changes for the cloud timeout path that can certify an interval containing a later committed change. Also found an unsynchronized keep-alive-map read that can cause unnecessary full refreshes. Both findings are inline; no existing comments need duplication or blocking-ID carryover. No additional user focus was supplied.
Applicable code-review checkpoints:
- Goal and evidence: the normal idle-partition path records (C,K) without changing C and uses next(K) for the changelog lower bound. Added planner, manager and regression cases cover the intended benefit; the cloud timeout counterexample still violates committed-data visibility.
- Scope and reuse: the change is focused on keep-alive, reuses the cleanup journal operation and TSO arithmetic, and extracts existing stream-discovery logic without changing its metadata output.
- Concurrency: reviewed daemon, planner, commit/publish and cleanup interactions. Local base-read -> bounded stream-write locking follows cleanup order; cloud RPC and journal awaits occur outside metadata locks. The synthesized reader lacks its stream lock, and cloud timeout abandonment breaks the watermark proof, as detailed inline.
- Lifecycle: Env starts the daemon with master-only services; immutable keep-alive values are replaced, and dropped-partition cleanup prunes the map. No new static initialization or ownership cycle was found.
- Configuration: both settings are mutable and master-only; the one-second polling loop observes updates. The interval versus TTL constraint is documented, and disabling renewal intentionally retains existing K values.
- Compatibility: optional Gson fields default safely for older metadata. The IVM guide explicitly exempts pre-January-2027 view compatibility; no new wire format or function-system compatibility issue was identified.
- Thrift/protobuf IDs: not applicable; no gensrc change.
- Parallel paths: checked shared-nothing/cloud, MOW aggregate/DUP detail, incremental/history/reset/pre-snapshot reads, excluded triggers and Explain/dry-run. Offset mismatch correctly suppresses K; the cloud timeout path remains unsafe.
- Conditional checks: initial/sentinel offsets, pending invisible versions, changed offsets, TTL margin, missing watermark and disabled/stale/dropped streams have concrete purposes and corresponding negative cases or traced guards.
- Test coverage: inspected the 16 new manager tests, planner cases and regression suite. None exercises an in-flight FDB commit after a timeout response or the map-capture race. The debug-point regression also relies on sleeps rather than acknowledging keeper completion.
- Test results: the new expected aggregate totals, detail rows and refresh outcomes agree with the test statements. Result-file generation provenance was not independently verified.
- Observability: renewal/failure logs include stream identity, count and K; existing daemon exception and fatal journal handling remain available. No separate logging or metric defect was found.
- Persistence and failover: traced Gson/image coverage and OP_TABLE_STREAM_CLEANUP decode, write and replay. Replay uses the same mutation; normal cloud epoch fencing is present, but it does not close the same-epoch timeout gap described inline.
- Data mutations and crashes: real consumption offsets and user-data write protocols remain unchanged. Persisting a false K makes the cloud finding survive later reads/replay; no additional atomicity or metadata-leak issue was verified.
- FE/BE propagation: no new transmitted variable or scan-bound coordinate; stored TSOs remain inclusive and scan ranges half-open.
- Performance: scans/copies are linear in streams and partitions, with TTL-based renewal throttling. Whole-stream cloud batches can exceed the nominal 10000-partition target, but no concrete failure from that alone was established. The reader race can trigger an expensive COMPLETE fallback.
- Other concerns: no further verified defect from the complete diff and upstream/downstream review. Transient cloud batch failures publish no proof; initial empty partitions have initialized version/TSO metadata.
Validation was static only: no builds, tests, benchmarks, source edits or submodule initialization. Convergence and delivery are tracked separately in this invocation; final convergence is declared by the submission metadata after the verification round.
Review converged after 2 round(s).
| LOG.info("skip keeping IVM stream offsets alive because the committed TSO is not available"); | ||
| return Collections.emptyList(); | ||
| } | ||
| long keepAliveTso = tsoStatus.getCommittedTso(); |
There was a problem hiding this comment.
[P1] Fence timed-out commits before using the cloud watermark
getCommittedTso() does not currently guarantee that every earlier commit has stopped being in flight. A configured FDB transaction timeout (1031) becomes TXN_TIMEOUT -> KV_TXN_COMMIT_ERR; CloudGlobalTransactionMgr.commitTxn() removes that transaction with abandonCommitTso() (lines 974-975), without the fence used for RPC exceptions. FoundationDB explicitly allows a timed-out commit to finish later.
For an idle partition at C, commit T>C can time out, be unregistered, and let the durable watermark advance to K>=T. If this round reads C=endTso before that commit finishes, it persists (C,K). Once the delayed commit publishes T, the next refresh still matches C but applyChangelogTtl() starts after K, silently dropping that committed change; an expired C that previously forced COMPLETE now passes the TTL check. Fence indeterminate commit responses before unregistering them, or establish an equivalent MS fence before using K as a permanent no-change proof. Please cover the timeout-response path, not only RPC exceptions.
There was a problem hiding this comment.
Confirmed. MetaService sets fdb_txn_timeout_ms on every FDB transaction, a 1031 during the commit comes back as KV_TXN_COMMIT_ERR, and commitTxn() unregisters that attempt with abandonCommitTso(), so the watermark can pass a commit that still lands. A round that reads the offsets in that window records a K that hides the commit, and the refresh after it skips the commit for good.
Fixed in 19b8e52: keepAliveIdleCloudPartitions now advances MetaService's TSO fence to K (advanceTsoFence) before it reads any offset. A commit reads the fence key in its own FDB transaction and is rejected at or below the fence, so once the fence is advanced, every commit at or below K has either landed, and the read that follows shows it, or can no longer land. A round whose fence fails records nothing. The cost is one fence write a round: commits being applied at that moment conflict on the key once and are retried through the existing KV_TXN_CONFLICT loop. TableStreamManagerKeepAliveTest now checks that the fence is advanced to the watermark once a round, before any read, and that a round whose fence fails reads and records nothing; both cases fail on the code before the fix. The javadoc and the PR description no longer claim that every commit at or below the watermark has finished.
The commit path itself still treats that timeout as a definite failure. That also affects the bounded cloud @incr reads, which admit an end at or below the watermark, and it predates this PR (#67820), so it is left to the commit path and will be followed up separately. The keep-alive round no longer depends on it.
There was a problem hiding this comment.
Follow-up on the commit path: #69925 makes commitTxn() advance the TSO fence before it gives up the TSO of a commit that MetaService answers with KV_TXN_COMMIT_ERR, as it already does when the RPC fails, so the watermark no longer passes a commit that can still land. The keep-alive round keeps its own fence; it does not rely on that fix.
| this.outputUpdateMap = buildOutputUpdateMap(selectedPartitionIds); | ||
| this.keepAlives = Maps.newHashMap(); | ||
| for (Long partitionId : selectedPartitionIds) { | ||
| OlapTableStream.PartitionKeepAlive keepAlive = stream.getKeepAlive(partitionId); |
There was a problem hiding this comment.
[P2] Snapshot keep-alives under the stream read lock
The IVM refresh starts with a plan such as:
LogicalOlapTableSink(mv)
LogicalAggregate(...)
LogicalOlapScan(base)
CollectRelation/StatementContext.lock() lock the base and target tables, but the stream is looked up later by IvmDeltaRewriteState when it constructs this wrapper. That path never acquires the stream read lock. Meanwhile the keeper writes OlapTableStream.partitionKeepAlive, a plain HashMap, under the stream write lock while holding a compatible base-table read lock.
A first keep-alive for another partition can resize that map during this get, transiently hiding an existing fresh K. The wrapper permanently captures its absence; when C is older than the TTL, applyChangelogTtl() rejects it and AUTO needlessly rebuilds the whole MV (strict INCREMENTAL fails). Capture the offsets and keep-alives under the stream read lock, or publish a concurrency-safe snapshot; explicitly bound SQL streams are locked, but these synthesized IVM readers are not.
There was a problem hiding this comment.
Confirmed. IvmDeltaRewriteState looks the IVM stream up and builds the wrapper after CollectRelation has collected the statement's tables, so the stream is never locked there, and the keeper's put can resize the HashMap under that get. The effect is the one you describe: a needless COMPLETE fallback, or a failed strict refresh, when the missed K is what kept an expired offset readable. No row is read wrong.
Fixed in 6cb029c: partitionKeepAlive is now declared as a ConcurrentMap, so a get never misses an existing entry while another one is added, and the planner path takes no new lock. The field is declared as one, not only initialized as one, so GSON also reads it back from an image as a concurrent map, as it does Database.name2Function. testKeepAliveTsoReadsBackFromTheImage checks that, and it fails with the field declared as a plain Map. The race itself has no deterministic test.
FE UT Coverage ReportIncrement line coverage |
TPC-H: Total hot run time: 28885 ms |
TPC-DS: Total hot run time: 152959 ms |
FE Regression Coverage ReportIncrement line coverage |
ClickBench: Total hot run time: 24.77 s |
### What problem does this PR solve? Issue Number: None Related PR: apache#69831 Problem Summary: `OlapTableStream.partitionKeepAlive` was a `HashMap` that the keep-alive daemon writes under the stream write lock. The delta scan of an IVM refresh reads it without the stream lock: `IvmDeltaRewriteState` looks the stream up and creates its `OlapTableStreamWrapper` after the statement has locked its tables, so the stream is never among them, and the base table read lock the daemon also holds in shared-nothing mode does not exclude the refresh, which holds it too. A put that adds a partition and resizes the map can make a concurrent get miss the keep-alive TSO of another partition. The wrapper keeps the miss for the whole refresh, and when that partition's offset is older than the TTL the read is refused: an AUTO refresh falls back to COMPLETE and a strict INCREMENTAL refresh fails, although a fresh keep-alive TSO was recorded. No row is read wrong. The field is now declared as a `ConcurrentMap`, so a get never misses an entry while another one is added. Declared as such, GSON reads it back from an image as a concurrent map too, as it does `Database.name2Function`. ### Release note None ### Check List (For Author) - Test: Unit Test - `TableStreamManagerKeepAliveTest`: a keep-alive TSO written to an image reads back, into a concurrent map; the cases that set the field directly set a concurrent map. The race itself has no deterministic test. - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <[email protected]>
…ve round reads the offsets ### What problem does this PR solve? Issue Number: None Related PR: apache#69831, apache#67820 Problem Summary: In cloud mode a keep-alive round of the IVM streams takes the committed TSO watermark as the keep-alive TSO K, reads the offsets from MetaService, and records K for every partition whose offset is at the end of the partition, as a lasting proof that the partition holds no change in (offset, K]. The watermark only says that no commit at or below it is still registered with the TSO service, and a commit can be unregistered while it may still land: a MetaService transaction that hits the FDB transaction timeout (1031) after its commit was sent comes back as `KV_TXN_COMMIT_ERR`, and `CloudGlobalTransactionMgr.commitTxn` unregisters it without the fence it uses for an RPC failure, though FoundationDB documents that such a commit can be committed later. If a round reads the offsets in that window, it records a K past the commit. Once the commit lands, the next refresh starts after K and skips it, and its offset then moves past it: the MV silently misses that write until a COMPLETE refresh. Before the keep-alive, a refresh whose offset had expired fell back to COMPLETE and read the write. A round now advances MetaService's TSO fence to K (`advance_tso_fence`) before it reads any offset. A commit reads the fence in its own FDB transaction and is rejected at or below it, so after the fence every commit at or below K has either landed, and the read shows it, or can no longer land. A round whose fence fails records nothing. The fence is one MetaService write a round; the commits being applied at that moment conflict on it once and are retried, as on any commit conflict. That the commit path unregisters such a commit without a fence also affects the bounded cloud `@incr` reads, which admit an end at or below the watermark; that is left to the commit path. ### Release note None ### Check List (For Author) - Test: Unit Test - `TableStreamManagerKeepAliveTest`: the fence is advanced to the watermark once a round, before any read; a round whose fence fails reads and records nothing; the other cloud cases run with the fence taken. - Behavior changed: Yes. In cloud mode a keep-alive round advances MetaService's TSO fence to the watermark before it reads the offsets, and records nothing when the fence fails. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <[email protected]>
|
run buildall |
TPC-H: Total hot run time: 28299 ms |
TPC-DS: Total hot run time: 150759 ms |
ClickBench: Total hot run time: 24.23 s |
FE UT Coverage ReportIncrement line coverage |
FE Regression Coverage ReportIncrement line coverage |
|
/review |
There was a problem hiding this comment.
Review completed at PR head 19b8e52192321fc929bbd3b0a5a17babbe69717c; convergence was reached after 2 bounded review rounds.
The change achieves its main goal for normal operation: idle IVM stream partitions can retain a no-change interval through shared-nothing locks or the cloud TSO fence, and the persisted keep-alive state is replay-compatible. The shared-nothing lock ordering, cloud fence/read-state ordering, offset matching, TTL successor arithmetic, duplicate-key and unique-key paths, stream enumeration, and partition pruning were all traced. No data-loss, stale-proof, deadlock, persistence-format, or non-IVM leakage issue was substantiated.
One new lifecycle issue is reported inline: the IVM daemon starts before the TSO daemon and both pass the readiness gate concurrently. A first cycle can run before TSO calibration, set lastRoundMs, and then suppress retry for the configured 600-second interval. This can leave an offset near its TTL unrenewed after restart, causing strict incremental refresh to fail or AUTO refresh to fall back to COMPLETE. The fix should gate on TSO readiness or advance the timestamp only after a usable round.
The earlier cloud timeout-fence P1 (4410495902) and keep-alive HashMap race P2 (4410495912) were independently revalidated as fixed at this head, so neither is reposted. The user supplied no additional focus beyond the complete review. Persistence compatibility is additive, dynamic configuration is polled, and the added unit/regression tests were inspected but not executed because this invocation is static-only.
Review converged after 2 round(s).
| if (now - lastRoundMs < Config.ivm_stream_keepalive_interval_second * 1000L) { | ||
| return; | ||
| } | ||
| lastRoundMs = now; |
There was a problem hiding this comment.
[P2] Retry keep-alive after TSO startup readiness
startMasterOnlyDaemonThreads() starts this daemon before tsoService, and both threads only leave MasterDaemon's readiness wait after Env.isReady is set. The first keep-alive cycle can therefore run while TSOService.isInitialized() is still false: local getTSO() throws (cloud returns an uninitialized status). Because lastRoundMs = now is assigned before keepAliveIdlePartitions(), the one-second daemon cycle then suppresses retries until ivm_stream_keepalive_interval_second (600 seconds by default). An idle offset close to its TTL can miss renewal after an FE restart and make the next strict incremental refresh fail or AUTO refresh fall back to COMPLETE. Gate the round on TSO readiness or only advance lastRoundMs after a successful round so calibration failures retry promptly.
There was a problem hiding this comment.
Confirmed. startMasterOnlyDaemonThreads() starts this daemon before tsoService, both leave the readiness wait together, and the TSO service calibrates itself in its own first cycle: it logs the next allocation window and, in cloud mode, advances MetaService's TSO fence. A first check that came before that set lastRoundMs and then lost the round (in shared-nothing mode getTSO() threw at the first stream with a partition due for renewal, in cloud mode the uninitialized status skipped it), so the next round came a whole ivm_stream_keepalive_interval_second later. With the default TTL of a day that delay is harmless, but with a changelog.ttl_seconds close to twice the interval, which the config description allows, an idle offset could expire in between: an AUTO refresh falls back to COMPLETE and a strict INCREMENTAL refresh fails. No row is read wrong.
Fixed in cf84aba: runAfterCatalogReady now returns before it sets lastRoundMs while TSOService.getStatusSnapshot().isInitialized() is false, so the first one-second check after calibration starts the round. IvmStreamKeepAliveDaemonTest covers it: no round while the TSO service is not initialized, a round at the first check after it is, and none again before the interval; it fails without the check.
There was a problem hiding this comment.
Confirmed. The ordering makes this real rather than theoretical: Env.startMasterOnlyDaemonThreads() starts this daemon before tsoService, and MasterDaemon.runOneCycle() only waits for Env.isReady(), which does not include TSO calibration. The first cycle after an FE restart or a leadership change can therefore run while the service is still uncalibrated: local getTSO() throws ("TSO timestamp is not calibrated"), and in cloud getStatusSnapshot().isInitialized() is false. runAfterCatalogReady() assigns lastRoundMs = now before keepAliveIdlePartitions(), so that skipped cycle suppresses the next attempt for the whole ivm_stream_keepalive_interval_second, and Daemon.run() swallows the throwable and keeps the thread alive, which is what makes it silent.
The impact is bounded but not zero: the lost round only matters when an offset is already inside its renewal margin and the remaining TTL is shorter than the interval. With the default 24h TTL that is 12h of margin against 600s, so it is harmless; it bites a table with a short changelog.ttl_seconds, where the first refresh after the restart fails as strict INCREMENTAL or falls back to COMPLETE although the keep-alive would have covered it.
Suggested fix, so that a round the service could not run counts as "not run", and readiness is checked before any lock is taken:
long now = System.currentTimeMillis();
if (now - lastRoundMs < Config.ivm_stream_keepalive_interval_second * 1000L) {
return;
}
if (!Env.getCurrentEnv().getTSOService().getStatusSnapshot().isInitialized()) {
return; // do not advance lastRoundMs: retry on the next one-second cycle
}
lastRoundMs = now;
keepAliveIdlePartitions();Gating up front also avoids two consequences of taking the TSO mid-round: getTSO() retries with sleeps (200ms per attempt, or tso_service_update_interval_ms) while keepAliveIdlePartitions holds the base-table read lock, so a failure there stalls writers on that table for up to tso_max_get_retry_count sleeps; and one stream's throw aborts the round for every stream after it, because the loop in IvmStreamKeepAliveDaemon.keepAliveIdlePartitions() has no per-stream guard.
There was a problem hiding this comment.
Thanks. That gate is what cf84aba added, checked before the interval so that the first one-second check after calibration starts the round; IvmStreamKeepAliveDaemonTest covers it. One small correction on "silent": Daemon.run() logs the throwable at ERROR before it carries on, so the lost round did show in fe.log.
On the two other points, at the current head:
getTSO()does not sleep while the service is uncalibrated: it throws before its retry loop. Once calibrated, the only sleep a round of this Master daemon can meet is a logical counter overflow within one physical tick, and then every commit runs the same loop inside the table write lock, so the writers already wait on their own TSOs; a round takes one TSO per stream. K cannot move out of the base table read lock either: a commit takes its TSO inside the write lock, and that is what makes every commit the partition has not had yet get a larger TSO than K.- With the gate, the only call on the per-stream path that throws is
getTSO(), and its failures (disabled, uncalibrated, retries exhausted) are failures of the whole TSO service, so every stream after it would fail the same way; a per-stream catch would turn one ERROR into one per stream. Anything else that throws there is a bug we want to see rather than step over. A lost round is left to the next one, and the config description now counts that against the margin (00befb0).
### What problem does this PR solve? Issue Number: None Related PR: apache#69831 Problem Summary: `startMasterOnlyDaemonThreads()` starts `IvmStreamKeepAliveDaemon` before the TSO service, and both wait for `Env.isReady` before their first cycle. The TSO service calibrates itself in that first cycle, which writes the next allocation window to the edit log and, in cloud mode, advances MetaService's TSO fence, so the first keep-alive check after an FE restart or a master change can come before it. The daemon set `lastRoundMs` before running the round, so such a round was spent for nothing: in shared-nothing mode `getTSO()` threw "TSO timestamp is not calibrated" at the first stream with a partition due for renewal, which ended the round, and in cloud mode the round was skipped because the TSO status was not initialized. The next round then came `ivm_stream_keepalive_interval_second` (600 seconds by default) later. With a `changelog.ttl_seconds` close to twice the interval, which the config description allows, an idle offset could expire before that round: an AUTO refresh fell back to COMPLETE and a strict INCREMENTAL refresh failed. No row was read wrong. A round now waits until the TSO service is initialized, without setting `lastRoundMs`, so the first check after the TSO service is ready starts it. ### Release note None ### Check List (For Author) - Test: Unit Test - `IvmStreamKeepAliveDaemonTest`: no round starts while the TSO service is not initialized, the first check after it is starts one, and the next round waits for the interval. - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <[email protected]>
### What problem does this PR solve? Issue Number: None Related PR: apache#69831 Problem Summary: `OlapTableStream` kept what it holds for a partition in four maps keyed by partition id: the offset (`po`), the time of the last consumption (`pct`), the TSO of the initial rows still to be read (`hpt`), and, since this PR, the keep-alive TSO (`pka`). Each place that adds, prunes or collects the partitions of a stream had to walk all four, and a read made without the stream lock, as the delta scan of an IVM refresh does, could see one map updated and another not. They are now one `ConcurrentMap<Long, PartitionStreamState>` (GSON `ps`). A `PartitionStreamState` holds the offset, the consumption time, the historical TSO and the keep-alive TSO of one partition, and is replaced as a whole under the stream write lock, never changed in place, so a read without the lock sees all of one state. A keep-alive TSO still records the offset it was taken against: in cloud mode the offsets live in MetaService, and the state of a partition holds only its keep-alive TSO. Table Stream is not released, so the old fields are not read: a stream from an image written by an earlier build reads back with no partition state, as if it had consumed nothing, and has to be dropped and recreated. ### Release note None ### Check List (For Author) - Test: Unit Test - The table stream and IVM unit tests, and the other classes using the changed code (57 classes, 931 cases), pass; the tests that set the maps directly set the partition states. `TableStreamManagerKeepAliveTest` reads an offset and its keep-alive TSO back from an image, into a concurrent map; the case for an image without the keep-alive TSO is gone, and the one for a cleanup entry written before it stays. - Behavior changed: Yes. The state of the partitions of a table stream is persisted in a new form, and a stream created by an earlier build loses its offsets. - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <[email protected]>
|
/review |
|
run buildall |
There was a problem hiding this comment.
Static review completed for PR #69831 at exact head bcbfdd134b65179aadcfcb7766a8c10454d2c695, with convergence after 2 bounded rounds.
Verified findings:
- The
OlapTableStreamrefactor replaces the parent image fieldspo,pct, andhptwithpswithout Gson alternates or migration. Existing user and IVM streams therefore lose offsets, historical TSOs, and consumption timestamps after an FE image/restart; later refreshes can replay retained data or reread rows into an already populated MV. - Initialized cloud keep-alive rounds consume
lastRoundMseven when the committed watermark is unavailable, fencing fails, or a read-state RPC fails. A transient failure can therefore defer retry for the full configured interval and let a near-expiry offset expire, causing strict incremental failure or AUTO COMPLETE fallback.
Existing inline findings were revalidated against this head and are fixed, so no blocking comment IDs are carried: cloud watermark fencing (4410495902), the wrapper keep-alive map race (4410495912), and TSO startup readiness (4419362329). No additional user focus was supplied. The Nereids TTL/exclusive-bound, historical/empty partition, local locking, cloud response validation/fencing, replay, stream discovery, metadata, and test changes were inspected with no further verified defects. Validation was static only; builds and tests were not executed.
Review converged after 2 round(s).
| // partition id -> what the stream holds for the partition. A state is replaced as a whole under the stream write | ||
| // lock, never changed in place, and it is also read without the lock: an IVM refresh reads the stream through a | ||
| // scan it creates after the statement has locked its tables (OlapTableStreamWrapper). So it is a concurrent map, | ||
| // and such a read sees one whole state of a partition. |
There was a problem hiding this comment.
[P1] Preserve table-stream partition state across image upgrades
OlapTableStream used to persist consumed offsets, historical TSOs, and consumption times in po, pct, and hpt. This head reads only the new ps field and has no alternate annotations or migration, so an image/checkpoint written by the parent commit deserializes every existing stream with an empty partitionStates map. After an FE restart or rolling upgrade, a user stream can replay retained changes and an IVM stream is treated as unconsumed; the next refresh can reread current rows into an already populated MV, while expired history is no longer represented. Replay only fills partitions that consume after the restart. Preserve the old fields through Gson migration before exposing the stream state.
There was a problem hiding this comment.
The format change is intended. The PR description says so under Persistence: "Table Stream is not released, so the old fields are not read: a stream from an image written by an earlier build reads back with no partition state and has to be dropped and recreated."
No release carries Table Stream: OlapTableStream is on none of branch-4.0, branch-4.1 and branch-4.2, and enable_table_stream is experimental, off by default and not mutable. So there is no rolling upgrade from a release to cover. A cluster that ran an earlier master build with the switch on drops and recreates its streams and IVM MVs; we do not carry a migration for a format that was never released.
The consequence you describe is right, and it is why such a stream has to be recreated rather than kept. In shared-nothing mode, where the offsets live only in FE metadata, an IVM refresh takes a partition without an offset for one the MV has not read yet: it reads the partition from the base table and appends it, so the MV gets those rows a second time and the refresh still succeeds. A user stream reads its retained changelog again.
| // on it once and are retried, as on any commit conflict. | ||
| long fenceTso; | ||
| try { | ||
| fenceTso = Env.getCurrentEnv().getGlobalTransactionMgr().advanceTsoFence(keepAliveTso); |
There was a problem hiding this comment.
[P2] Retry failed cloud keep-alive rounds
lastRoundMs is advanced before the cloud keep-alive work, but this path treats several failures as a completed round: an initialized TSO service with committedTso <= 0, a failed advanceTsoFence, and a getReadStates UserException all return without recording a keep-alive. The daemon then waits the full ivm_stream_keepalive_interval_second before trying again. With the documented interval up to half the table TTL, an offset that entered the renewal margin just before a transient MetaService failure can expire before the retry, making strict INCREMENTAL fail or AUTO fall back to COMPLETE. Propagate round success/failure and do not consume the interval on an unsuccessful cloud round (with suitable backoff).
There was a problem hiding this comment.
The mechanism is right: lastRoundMs is set before the round, the three paths you list record nothing (the first two for the whole round, a failed getReadStates for its batch), and the next attempt comes one ivm_stream_keepalive_interval_second later. Shared-nothing mode has the same shape (a getTSO() failure, a busy lock).
Leaving a failed round to the next one is intended. Such a round records nothing (testCloudKeepAliveSkipsRoundWhenTheFenceFails, testCloudKeepAliveSkipsRoundWhenReadStatesFail), and the next round that succeeds renews a partition that is still idle even if its K has expired by then: neither mode's condition asks whether K is within the TTL, and "nothing was committed after the offset" holds however old K is. What a failed round can cost is the refresh that reads a write made after it, which falls back to COMPLETE as before this PR (a strict INCREMENTAL refresh fails); no row is read wrong.
How much that costs is a matter of margin. When an idle partition is written, the newer of its offset and K is at most TTL/2 + one interval old, and one interval older for each round that could not renew it. After the write the partition is not renewed any more, and the refresh that reads the write has to start before that TSO is a TTL old. So that refresh is incremental if it starts within TTL/2 - interval after the write, one interval less per failed round. With the defaults (a TTL of a day, 600 seconds) that is 11 h 50 min, and a refresh right after the write still reads incrementally after 71 failed rounds in a row.
What this comment does show is that the bound in the config description was wrong: "at most half the shortest changelog.ttl_seconds". At half, no time is left even when every round succeeds, and a refresh shortly after the write falls back; retrying failed rounds sooner would not change that. Fixed in 00befb0: the description of ivm_stream_keepalive_interval_second now states the time a refresh has after a write, less one interval for each check that could not renew the partition, and asks for an interval well under half the TTL. The PR description says the same.
A faster retry is not added. A cloud round advances MetaService's TSO fence before it reads, so retrying sooner while MetaService is failing writes to it more often, and the commits being applied at that moment conflict on each fence once. The interval is the retry period; more margin comes from a shorter interval, which costs more checks but no more edit logs.
FE UT Coverage ReportIncrement line coverage |
…ive_interval_second ### What problem does this PR solve? Issue Number: None Related PR: apache#69831 Problem Summary: The description of `ivm_stream_keepalive_interval_second` asked for an interval of at most half the shortest `changelog.ttl_seconds` of the IVM base tables, a bound that leaves a refresh no time. A partition is renewed at the first check after less than half the TTL is left, so when an idle partition is written, the newer of its offset and its keep-alive TSO is up to half the TTL plus one interval old, and one interval older for each check that could not renew it: a round that cannot reach MetaService, finds a lock busy or fails to get a TSO records nothing and leaves the partition to the next one. After the write the partition is not renewed any more, and the refresh that reads the write falls back to COMPLETE (a strict INCREMENTAL refresh fails) once that TSO is older than the TTL. So a refresh is sure to be incremental only if it starts within half the TTL minus the interval after the write, one interval less for each check that could not renew the partition: 11 hours 50 minutes with the defaults (a TTL of a day, 600 seconds), and no time at all at the bound the description gave. The description now states that time and asks for an interval well under half the TTL. A failed round is still left to the next one: retrying it sooner would not change the bound, and a cloud round first advances MetaService's TSO fence, so a faster retry while MetaService is failing would write to it more often. ### Release note None ### Check List (For Author) - Test: No need to test (only the description of a config changes); checkstyle passes for fe-common. - Behavior changed: No - Does this need documentation: No Co-Authored-By: Claude Opus 5.5 <[email protected]>
TPC-H: Total hot run time: 28893 ms |
TPC-DS: Total hot run time: 151307 ms |
ClickBench: Total hot run time: 24.16 s |
|
run buildall |
|
/review |
There was a problem hiding this comment.
Static review completed for PR #69831 at exact head 00befb02e5e9d25a6e42e333baa97365fbf47d4f. No new verified inline findings remain, so this delivery contains no new comments and carries no existing blocking IDs.
The change achieves its stated goal: idle IVM stream partitions can persist a no-change interval and use the exclusive successor of that keep-alive TSO when planning an incremental changelog read. The shared-nothing path snapshots offsets, checks partition visibility under the base-table read lock, captures TSO before releasing it, and rechecks the stream offset before publication. The cloud path fences the committed watermark before complete MetaService read-state batches and records nothing for failed batches. The wrapper applies a keep-alive only when its recorded offset matches the actual read offset.
Critical checkpoint conclusions:
- Goal and evidence: the changed manager, planner, FE unit, and regression cases cover local/cloud, unique-key aggregate and duplicate-key detail, TTL expiry, replay, and the race boundary. Tests were inspected but not executed because this invocation is static-only.
- Scope and reuse: the implementation is focused on keep-alive, reuses the existing cleanup journal and TSO utilities, and centralizes IVM stream discovery in
IvmUtil. - Concurrency: state values are replaced as whole partition snapshots in a concurrent map; local base-table/stream locking and cloud fence/read ordering were traced, with no deadlock or stale-proof defect verified.
- Lifecycle/configuration: the daemon is constructed and started in the master-only lifecycle, waits for catalog and TSO readiness, and polls both mutable settings on each tick.
- Persistence/compatibility: keep-alive cleanup entries are optional and replay through the existing journal operation. The
OlapTableStreamfield-format break is intentional for unreleased experimental table streams under the repository IVM compatibility guidance. No thrift/protobuf change is present. - Parallel paths and errors: incremental, snapshot, reset, append-only, MOW, cloud, and shared-nothing paths were traced; malformed cloud responses and failed fences produce no keep-alive proof, and non-IVM/user streams are excluded.
- Observability/performance: renewal and failure paths retain stream identity and TSO logging; the scan/discovery work is bounded by streams and partitions.
Existing review context was revalidated at this head: cloud fence finding 4410495902, wrapper map race 4410495912, and TSO-startup readiness 4419362329 are fixed in the current code. The old-image concern 4422394915 is intentionally unsupported for this pre-release experimental format. The failed-round retry concern 4422394921 remains known context and is not duplicated; it is a P2 margin/behavior discussion rather than a new blocking finding. No additional user focus was supplied.
Validation was static only: no builds, tests, benchmarks, source edits, or submodule initialization. Review coverage converged after 1 bounded round: all assigned subagents completed their full scopes and rechecks, and every candidate and risk focus has an explicit disposition.
Review converged after 1 round(s).
TPC-H: Total hot run time: 28745 ms |
TPC-DS: Total hot run time: 151486 ms |
ClickBench: Total hot run time: 24.29 s |
What problem does this PR solve?
Issue Number: #69907
Related PR: #69772, #69785
Problem Summary:
In short: an IVM materialized view rebuilds itself completely whenever one of its base table partitions is written after being idle for longer than
changelog.ttl_seconds(one day by default), although nothing the refresh needs has expired. This PR has the Master record, for such idle partitions, a keep-alive TSO up to which the partition is known to hold no change, and the refresh checks the TTL against it. Both shared-nothing and cloud mode are covered; MetaService is not changed.Background
An IVM MV reads each base table through an internal table stream, which keeps a consumption offset (a TSO) per partition. A refresh reads the partition's changelog after the offset and moves the offset to what it read. A table's changelog is kept for
changelog.ttl_seconds; a read whose offset is older thannow - TTLmay have lost changes, so an IVM read refuses it (CHANGELOG_OFFSET_EXPIRED), and since #69772 an AUTO refresh answers that asCHANGELOG_BROKENand falls back to COMPLETE (a strict INCREMENTAL refresh fails). Since #69785 the same holds for an IVM MV over a duplicate-key table, whose expired offset used to be clipped silently.The problem, and what it cost
How this PR fixes it
NormalizeOlapTableStreamScan.applyChangelogTtl). The rows read are the same, and only retained changelog is read. The offset itself does not move, so the invariants built on "offset == partition TSO means caught up" stay as they are.IvmStreamKeepAliveDaemon, a Master daemon, runs a round everyivm_stream_keepalive_interval_second(600) over the IVM streams, found from the MV side. A round waits until the TSO service is initialized: a new Master calibrates it in the TSO service's own first cycle, and a round started before that would be lost, with the next one an interval later. A partition qualifies when the base table applies a TTL, the stream has a real offset for it past the initial snapshot, nothing is committed to it after the offset, and the newer of the offset and K has less than half the TTL left. A keep-alive TSO thus gets as old as half the TTL plus one interval before it is renewed, and one interval older for each round that cannot renew it (MetaService unreachable, a lock busy, the TSO service failing), which records nothing and leaves the partition to the next round. A partition is no longer renewed once it is written, so the refresh that reads the write is incremental if it starts within half the TTL minus the interval after the write, one interval less for each such round; with the defaults (a TTL of a day, 600 seconds) that is 11 hours 50 minutes, less ten minutes a round. A refresh that starts later may fall back to COMPLETE, as before this PR. At an interval of half the TTL no time is left at all, so the interval should be well under half the shortestchangelog.ttl_secondsof the IVM base tables. The config's description says so.get_table_stream_offset(which waits for the partitions' pending transactions), it advances MetaService's TSO fence to K (advance_tso_fence, the fence [fix](binlog) Track committed TSO and fence uncertain commits for bounded incremental reads #67820 added for commits whose outcome the FE does not know). Every commit at or below the watermark is unregistered from the TSO service by then, but one the FE gave up on after a response it may outlive, such as a MetaService transaction that hit the FDB transaction timeout after its commit was sent, can still land. A commit reads the fence in its own FDB transaction and is rejected at or below it, so after the fence every commit at or below K has either landed, and the read shows it, or can no longer land; a commit that lands later gets a larger TSO. A partition whose offset is at its end in the read thus holds no change up to the watermark, for good. A round whose fence fails records nothing. The fence is one MetaService write a round; the commits being applied at that moment conflict on it once and are retried. The read goes in batches of about 10000 partitions; a stream with more partitions is read in one request.PartitionStreamStatein one concurrent map ofOlapTableStream(GSONps), in place of the mapspo,pctandhptand a fourth one for K. A state is replaced as a whole under the stream write lock, so the delta scan of an IVM refresh, which reads the stream without that lock, sees all of one state. Table Stream is not released, so the old fields are not read: a stream from an image written by an earlier build reads back with no partition state and has to be dropped and recreated. The K entries are logged in a new optional list ofOP_TABLE_STREAM_CLEANUPand replayed the same way; K of a dropped partition is pruned with its offsets.Results
The new regression suite
mtmv_p0/ivm/test_ivm_stream_keepalive(shared-nothing, run locally): an aggregate MV over a merge-on-write table and an append-only detail MV over a duplicate-key table, both base tables withchangelog.ttl_seconds = 60and a five-second check. Each row is whattasks()reports for the refresh (Status,RefreshMode,IvmFallbackReason), for both MVs, and every MV matches its query over the base table.REFRESH ... AUTOSUCCESS COMPLETE CHANGELOG_BROKEN: the whole MV is rebuiltSUCCESS, incremental, no fallbackThe FE log of that run shows the keep-alive TSO renewed every thirty seconds, half the TTL, while the partitions were idle, and nothing recorded for the held stream once released.
Classes, and how they call each other
IvmStreamKeepAliveDaemon(new): the Master daemon. It checks every second whether the TSO service is initialized andivm_stream_keepalive_interval_secondhas passed since its last round, so that a change of the interval applies at once and the first round after a restart is not lost, and finds the IVM streams throughIvmUtil.getIvmStreams(new, whichmv_infos()now uses too).TableStreamManager.keepAliveIdlePartitions/keepAliveIdleCloudPartitions(new): one round in each mode; the shared-nothing one takes the locks in the ordercleanupStalePartitionOffsetstakes them.OlapTableStream:partitionStates(GSONps), one concurrent map ofPartitionStreamState(offset, consumptionTime, historicalTso, keepAlive)in place of the four maps, itskeepAliveaPartitionKeepAlive(offset, keepAliveTso);unprotectedKeepAlive,effectiveOffset. A state is replaced as a whole under the stream write lock, but the delta scan of an IVM refresh is created after the statement has locked its tables and reads it without the stream lock.OlapTableStreamUpdate.checkPartitionOffsetreads the states.TableStreamCleanupInfo: the new optionalkeepAliveEntriesofOP_TABLE_STREAM_CLEANUP.OlapTableStreamWrapper.getKeepAliveTsoandNormalizeOlapTableStreamScan.applyChangelogTtl: the read side.Not changed: MetaService, the protos, the offsets themselves, how a user stream is read and consumed (its state moves into the same map, but it gets no K), and the direct incremental query (
BindRelation.applyChangelogTtl), which has no offset.Release note
An IVM materialized view no longer falls back to a COMPLETE refresh when a base table partition is written after being idle for longer than
changelog.ttl_seconds. New FE configsenable_ivm_stream_keepalive(default true) andivm_stream_keepalive_interval_second(default 600), both mutable.Check List (For Author)
Test
Regression test
Unit Test
Manual test (add detailed scripts or steps below)
No need to test or manual test. Explain why:
test_ivm_stream_keepalive(new, nonConcurrent): the cases in Results. Run locally in shared-nothing mode, together with the wholemtmv_p0/ivmdirectory (116 suites, all pass). Not run locally in cloud mode;mtmv_p0is part of the cloud pipeline, where the same suite takes the cloud branch. Rerun after the TSO readiness fix and the map merge, with the wholemtmv_p0/ivmdirectory (116 suites),table_stream_p0(16 suites) andchangelog_p0/test_changelog_publish_conflict: all pass; the commit after them, which corrects the description ofivm_stream_keepalive_interval_second, changes no code. The TSO fence only runs in cloud mode and is covered by the unit tests below.TableStreamManagerKeepAliveTest(new, 19 cases): the conditions of both modes (a change after the offset, a committed transaction that is not visible, the initial snapshot, the margin, no TTL or one that never expires an offset; in cloud mode a state that is not at its end or has no real offset, a watermark not past the offset, no calibrated TSO, a failed read, reads in batches, the fence advanced to the watermark once a round before any read, and a round that records nothing when the fence fails), K taken against another offset not counting, the wrapper using K only for a read from its offset in both modes, replay, pruning, a cleanup entry written before K, and an offset with its K read back from an image into a concurrent map. The three cases of the review fixes fail on the code before them.IvmStreamKeepAliveDaemonTest(new): no round while the TSO service is not initialized, a round at the first check after it is, and none again before the interval. It fails without the check, and with the check placed after the round timer is set.ExplainTableStreamPlanTest: an IVM read of an expired offset with a fresh K is admitted and starts after K; one whose K expired too, or was taken against another offset, is rejected.The IVM unit tests, every
catalog/streamtest class and the other classes using the changed code (57 classes, 931 cases) pass; checkstyle passes for fe-core and fe-common.Behavior changed:
Does this need documentation?
Check List (For Reviewer who merge this PR)