[Bug] INSERT OVERWRITE ... PARTITION(*) hangs forever due to a receiver-side load-stream close race

[lzydmxy]

Search before asking

• I had searched in the issues and found no similar issues.

Version

master (also reproduces on branch-4.0 / 4.0.x). Introduced by #56120; latest master still has the buggy code.

What's Wrong?

INSERT OVERWRITE ... PARTITION(*) — and, more generally, any load that opens incremental streams at runtime — can hang forever. The FE MarkedCountDownLatch never releases, so the query blocks until the 4h insert_load_default_timeout_second. It is frequent under concurrent loads.

Which usages are affected (both go through load stream v2 / move-memtable, i.e. VTabletWriterV2, and both open incremental streams mid-load via on_partitions_created → _incremental_open_streams):

1. auto-partition table creates a new partition at runtime (createPartition RPC):
   • INSERT INTO auto_tbl SELECT/VALUES ... where data hits a not-yet-existing partition
   • Stream Load / Broker Load / Routine Load into an auto-partition table that produces a new partition
2. auto-detect overwrite (replacePartition RPC):
   • INSERT OVERWRITE TABLE t PARTITION(*) SELECT/VALUES ... (a static partitioned table also qualifies; with enable_auto_create_when_overwrite=true new keys additionally take path 1)

Not affected: plain loads into existing partitions, and INSERT OVERWRITE TABLE t PARTITION(p1, p2) with named partitions (normal path, no incremental stream).

An additional amplifier: _incremental_open_streams only opens a stream to a newly-appearing backend. So the hang is more likely on larger clusters / more BEs / replicas spread out, where a new partition is more likely to land on a BE that had no stream yet.

Root cause

The bug is in the receiver-side load-stream close path — _dispatch's handling of CLOSE_LOAD in be/src/load/channel/load_stream.cpp (older path: be/src/runtime/load_stream.cpp). It split one logical operation across two separate lock scopes, with a _report_result network IO (several ms) in between:

bool all_closed = close(...);          // lock scope A: count, decide all_closed
_report_result(...);                   // network IO (several ms), NO lock
std::lock_guard lock(_lock);           // lock scope B: a DIFFERENT acquisition
if (num_incremental_streams > 0) _closing_stream_ids.push_back(id);
else                             brpc::StreamClose(id);
if (all_closed) {                      // iterate + clear
    for (auto& cid : _closing_stream_ids) brpc::StreamClose(cid);
    _closing_stream_ids.clear();
}

What You Expected?

INSERT OVERWRITE ... PARTITION(*) and auto-partition loads that create partitions at runtime should complete normally; no stream should be left un-closed regardless of CLOSE_LOAD interleaving on the receiver, and no stream should be closed before its own EOS is sent.

How to Reproduce?

The race window is only microseconds on a fast single-host cluster (streams close almost instantly), so brute-force concurrency will not open it — it needs real inter-BE latency, or a deterministic delay.

Deterministic reproduction with a BE debug point (added alongside the fix, inert unless enable_debug_points=true): LoadStream.close_load.delay_incremental_register delays a non-last stream between close() and its deferred-close registration, so the last stream drains+clears the list first and orphans the delayed one.

A docker regression suite reproduces it on a 3-BE cluster (replication_num=1 + BUCKETS 1, so new partitions land on fresh backends as incremental streams):

suite("test_iot_overwrite_partition_star_hang", "docker") {
    def options = new ClusterOptions()
    options.feNum = 1; options.beNum = 3
    options.beConfigs += ['enable_debug_points=true']
    docker(options) {
        sql """create table iot_star_hang(k0 int null) auto partition by list (k0)
                 (PARTITION p1 values in ((0))) DISTRIBUTED BY HASH(k0) BUCKETS 1
                 properties("replication_num"="1");"""
        sql "insert into iot_star_hang values (0);"                 // base partition on ONE BE
        GetDebugPoint().enableDebugPointForAllBEs("LoadStream.close_load.delay_incremental_register")
        // one load opens non-incremental (key 0) + incremental (keys 1..49) streams
        def t = Thread.start {
            /* insert overwrite table iot_star_hang partition(*)
               select number from numbers("number"="50"); */
        }
        t.join(60000)
        if (t.isAlive()) throw new Exception("hung: orphaned incremental stream never closed")
    }
}

Result: on the buggy binary the load hangs and the suite times out; on the fixed binary the load completes.

Anything Else?

Fix approach (in the PR): a new mark_eos_sent_and_collect() merges register + all-received check + collect-to-close under a single lock; a stream is registered only after its own EOS was sent; and a write-once _all_close_load_received latch makes any thread arriving after all-received drain the remaining streams, so no orphan survives any interleaving. Fencing (closing incremental streams together, after non-incremental ones) is preserved.

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

[zclllyybb]

Breakwater-GitHub-Analysis-Slot: slot_19cb3209792b
This content is generated by AI for reference only.

Initial triage for #65582.

Summary:

• Live issue metadata matches the report: the issue is open, has no labels/comments at the time of triage, and a matching open PR exists: [fix](load_stream) fix receiver-side stream leak race on close_load #65584, [fix](load_stream) fix receiver-side stream leak race on close_load.
• I verified the code path on current upstream/master (052880662b061000694f338d682c741f947353ac) and current upstream/branch-4.0 (54e9da0b631ba8e77510d7b1704319d4bc58fa56). I did not run the docker reproduction locally.
• The receiver-side race described in the issue is real from code inspection. On master, be/src/load/channel/load_stream.cpp handles CLOSE_LOAD by calling close() under _lock, then doing _report_result(..., eos=true) outside the lock, then reacquiring _lock to register an incremental stream in _closing_stream_ids or close a non-incremental stream, and finally draining _closing_stream_ids if the earlier close() result said all streams had closed. Branch-4.0 has the same structure in be/src/runtime/load_stream.cpp.
• The bad interleaving is credible: a non-last incremental stream can be counted by close() but delayed before _closing_stream_ids.push_back(id). The last stream can then observe all CLOSE_LOADs, drain and clear the current list, and return. When the delayed stream later registers itself, no later thread is guaranteed to close it, so on_closed never fires on that stream and the sender-side close wait can hang until the load timeout.
• The related Stream closed without EOS hazard is also plausible from the same split ordering: the all-closed thread can close another deferred stream before that stream has sent its own EOS response.
• The producer-side trigger path is consistent with the issue. VTabletWriterV2::on_partitions_created() calls _incremental_open_streams(), which opens stream-map entries with get_or_create(dst_id, true). LoadStreamMap::get_or_create() increments _num_incremental_streams only when a newly appearing backend stream entry is created, and close_load() sends that count in PStreamHeader. The auto-detect overwrite path calls FE replacePartition, updates the partition mapping, and invokes the same create-partition callback, so INSERT OVERWRITE ... PARTITION(*) can reach the incremental stream path.
• Plain loads into already-opened streams, and named-partition overwrite that does not create runtime incremental streams, should not hit this specific receiver-side deferred-close race.
• PR [fix](load_stream) fix receiver-side stream leak race on close_load #65584 appears directionally correct for this root cause: it adds mark_eos_sent_and_collect(), registers a stream only after its own EOS response is sent, combines EOS-sent registration / all-close latch / collection under one lock, and adds a deterministic docker regression with a debug point.

Missing information / validation:

• Need CI/regression evidence for [fix](load_stream) fix receiver-side stream leak race on close_load #65584, especially the new docker regression on current master.
• If this fix is intended for 4.0.x, request branch-4.0 validation or a branch-4.0-specific backport patch. Current branch-4.0 has the same receiver-side race in the older be/src/runtime/load_stream.cpp path.
• For any production occurrence, ask for BE logs around the affected load id showing received CLOSE_LOAD, close-wait behavior, and any sender-side Stream closed without EOS messages. These logs are not required to accept the root-cause mechanism, but they would help correlate real incidents.

Recommended next steps:

1. Review PR [fix](load_stream) fix receiver-side stream leak race on close_load #65584 as the likely fix for this issue.
2. Add the appropriate branch-pick label(s) if maintainers confirm the 4.0/4.0.x support scope.
3. In review, focus on the invariant that a stream must only be closed after its own EOS response is sent, and once all CLOSE_LOADs are counted, every later EOS-sent registration must also drain and close.
4. Keep the issue open until the new regression passes on master and the backport target decision is made.