Repository navigation
[Bug] Table Stream offset takes effect at COMMITTED, so consumers that read their own target table can lose updates #70281
Description
Search before asking
- I had searched in the issues and found no similar issues.
Version
master @ 72e9189 (2026-11-03). The code paths below are unchanged on master as of 2026-11-05.
What's Wrong?
In shared-nothing mode, the Table Stream offset is advanced in updateCatalogAfterCommitted() when the consuming transaction commits (DatabaseTransactionMgr.java#L2433-L2446). The rows it wrote become readable only after publish. If publish is slow or times out, INSERT INTO ... SELECT ... FROM <stream> still returns, with the transaction left COMMITTED (OlapInsertExecutor.java#L236-L251).
A common consumption pattern reads its own target table in the same statement, for example accumulating deltas into a state table with a scheduled job:
INSERT INTO state
SELECT d.id, COALESCE(s.amount, 0) + d.delta
FROM (SELECT id, SUM(amount) AS delta FROM detail_stream GROUP BY id) d
LEFT JOIN state s ON s.id = d.id;If round N is COMMITTED but not yet VISIBLE when round N+1 runs:
- round N+1 reads the stream from the advanced offset, so round N's delta is no longer pending;
- round N+1 reads
statewithout round N's rows; - round N+1's upsert later overwrites round N's result, and round N's delta is lost permanently.
IVM already guards against exactly this for aggregate MVs. IvmDeltaRewriter fails the refresh with MV_COMMIT_NOT_VISIBLE when the previous refresh is committed but not visible, and its comment says that otherwise "that join misses the old rows and the delta is permanently lost" (IvmDeltaRewriter.java#L76-L90). User-defined stream consumers get no equivalent protection.
What You Expected?
A stream consumer should not observe an advanced offset while the transaction that advanced it is still invisible. Possible approaches:
- reject or wait when the stream's previous consumption is COMMITTED but not yet VISIBLE;
- make the advanced offset effective only at VISIBLE.
At minimum, the Table Stream documentation should warn about read-modify-write consumers.
How to Reproduce?
- Create a changelog-enabled detail table, an
append_onlystream on it, and a MoWstatetable. - Run the statement above once so that
statehas a baseline. - Enable the FE debug point
DatabaseTransactionMgr.finishTransaction.block_visible(also used by test_ivm_agg_previous_commit_not_visible.groovy), load a batch into the detail table, and run the statement (round N stays COMMITTED). - Load another batch for the same ids and run the statement again (round N+1).
- Disable the debug point and compare
statewithSELECT id, SUM(amount) FROM detail GROUP BY id. Suspected: round N's amounts are missing.
Anything Else?
Found by code reading while documenting Table Stream consumption patterns; not yet reproduced. In shared-storage mode a commit is visible as soon as it succeeds, so that mode should not be affected.
Are you willing to submit PR?
- Yes I am willing to submit a PR!
Code of Conduct
- I agree to follow this project's Code of Conduct
Activity
Doris-Breakwater commentedon Nov 5, 2026
Assessment: This warrants a data-correctness investigation for shared-nothing read-modify-write stream consumers. At
72e91896bbdd8760f8bd4e488e5672406b40ce52, the reported timing window is supported by the code, and the existing offset-conflict check does not close it. Advancing an offset at COMMITTED is established transaction behavior; the additional concern here is consuming the next delta while reading stale target state. I have not executed the reproduction, so the final lost-update outcome remains a code-supported hypothesis, not a reproduced result. The issue is currently open with no labels.Verified code path at the reported commit:
InsertIntoTableCommandattaches the stream update information to the consuming transaction.updateCatalogAfterCommitted()applies it at commit.OlapTableStream.getStreamUpdate()then exposes that offset as the next scan's starting point.GlobalTransactionMgr.commitAndPublishTransaction()releases the target/stream write locks before waiting for visibility. Those locks therefore do not serialize the next consumer behind publish.OlapTableStreamUpdate.checkPartitionOffset()checks expected offsets, not the previous consumer transaction's visibility. A statement planned after N's offset advance can pass this check.partition.getVisibleVersion(). Thus it can read the pre-N state even though its stream scan starts after N's consumed range. MoW's duplicate-key handling operates on materialized input rows; it does not reruns.amount + d.deltaagainst subsequently published state.IvmDeltaRewriterexplicitly guards aggregate refreshes against this stale-state dependency. That guard is specific to IVM; it does not protect the ordinary INSERT shown here.Reproduction refinement: The debug point requires the target table name as its value. Follow the existing regression test: enable
DatabaseTransactionMgr.finishTransaction.block_visibleon all FEs with[value: "state"]. Enabling it without a nonempty value does not block this path. Targetingstateallows both detail-table batches to become visible normally.Use one key, a fully visible baseline of 100, and successive detail deltas of +10 and +20. Use a plain MoW target without a sequence-column override or another writer. Set a short
insert_visible_timeout_msand explicitly selectinsert_visible_timeout_return_mode=committed. After N returns, verify its transaction is COMMITTED, its stream offset advanced, andstatestill reads 100. Make +20 visible indetail, then run N+1 and capture its transaction/result before disabling the point. After cleanup, wait until both consumer transactions are VISIBLE before comparing: the hypothesized result is 120 versus the correct 130. An immediate comparison while either transaction remains COMMITTED would demonstrate only delayed visibility. Always disable the point infinally.Evidence still needed: Complete table/stream DDL and inserts; relevant session settings; both INSERT responses including labels/transaction IDs; transaction status and target committed/visible versions at each boundary; stream consumption offsets before/after each round; and the final ordered target/source aggregate comparison after both transactions become VISIBLE. FE transaction/publish log excerpts correlated by transaction ID would establish the ordering. The exact revision is already supplied; no query profile is needed for this initial correctness check.
Next steps: Add a focused shared-nothing regression with the above assertions. If it confirms the mismatch, preserve the invariant that a dependent consumer cannot start its target read from an advanced stream position until the preceding consumption's writes are visible. A wait/rejection needs to happen before fixing the read snapshot. Simply moving the existing offset update to VISIBLE also requires retaining commit-time offset reservation/conflict protection, otherwise two consumers could commit the same range; persistence/replay and concurrent-consumer tests are needed for that design.
Until semantics are clarified, document this constraint and have orchestration wait for the preceding transaction to reach VISIBLE before starting the next dependent statement. Waiting only for statement completion, or increasing the publish timeout, does not establish that guarantee.
insert_visible_timeout_return_mode=errorchanges the response but does not roll back the committed transaction. The reported FE COMMITTED-to-VISIBLE window is specific to the shared-nothing path; this investigation does not establish a blanket guarantee for every shared-storage failure mode. The hypothesized loss affects accumulated target state; retained detail rows can still support a controlled rebuild.Breakwater-GitHub-Analysis-Slot: slot_b79f743f618d