lzydmxy opened a new issue, #65582: URL: https://github.com/apache/doris/issues/65582
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/doris/issues?q=is%3Aissue) 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: ```cpp 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): ```groovy 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? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
