[Bug] stream load 的时候报错 [INTERNAL_ERROR]unknown stream load id

[330570902]

Search before asking

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

Version

doirs 版本 2.1.11

What's Wrong?

错误日志：
I20260529 08:21:25.307652 198092 tablet.cpp:1980] cumulative compaction meet delete rowset, increase cumu point without other operation.|tablet id:=20711468|after cumulative compaction, cumu point:=2
I20260529 08:21:27.409749 199486 stream_load.cpp:208] new income streaming load request.id=ee4ad23d0593a0dd-b8e586767dbe51b2, job_id=-1, txn_id=-1, label=datax_doris_writer_55e3d16f-47bb-4282-9a82-603110fbbd10_0, elapse(s)=0, db=qxk, tbl=qxk_datahub_ysk_xj_gat_wlmqgj_wlmq_smzkxx_new, group_commit=0
I20260529 08:21:27.415176 199486 stream_load_executor.cpp:72] begin to execute stream load. label=datax_doris_writer_55e3d16f-47bb-4282-9a82-603110fbbd10_0, txn_id=11142908, query_id=ee4ad23d0593a0dd-b8e586767dbe51b2
I20260529 08:21:27.416150 199486 stream_load.cpp:214] finished to handle HTTP header, id=ee4ad23d0593a0dd-b8e586767dbe51b2, job_id=-1, txn_id=11142908, label=datax_doris_writer_55e3d16f-47bb-4282-9a82-603110fbbd10_0, elapse(s)=0
W20260529 08:21:27.417343 201197 status.h:415] meet error status: [INTERNAL_ERROR]unknown stream load id: ee4ad23d0593a0dd-b8e586767dbe51b2

...skipping one line
1# doris::vectorized::NewJsonReader::_open_file_reader(bool) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
2# doris::vectorized::NewJsonReader::_simdjson_init_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
3# doris::vectorized::NewJsonReader::init_reader(std::unordered_map<std::__cxx11::basic_string<char, std::char_traits, std::allocator >, std::shared_ptrdoris::vectorized::VExprContext, std::hash<std::__cxx11::basic_string<char, std::char_traits, std::allocator > >, std::equal_to<std::__cxx11::basic_string<char, std::char_traits, std::allocator > >, std::allocator<std::pair<std::__cxx11::basic_string<char, std::char_traits, std::allocator > const, std::shared_ptrdoris::vectorized::VExprContext > > > const&) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
4# doris::vectorized::VFileScanner::_get_next_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
5# doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
6# doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
7# doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
8# doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
9# doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptrdoris::vectorized::ScannerContext, std::shared_ptrdoris::vectorized::ScanTask) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
10# std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptrdoris::vectorized::ScannerContext, std::shared_ptrdoris::vectorized::ScanTask)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
11# doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
12# doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
13# start_thread
14# __clone
W20260529 08:21:27.417537 201197 status.h:415] meet error status: [INTERNAL_ERROR]failed to init reader, err: [INTERNAL_ERROR]unknown stream load id: ee4ad23d0593a0dd-b8e586767dbe51b2

    0#  doris::FileFactory::create_pipe_reader(doris::TUniqueId const&, std::shared_ptr<doris::io::FileReader>*, doris::RuntimeState*, bool) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::NewJsonReader::_open_file_reader(bool) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    2#  doris::vectorized::NewJsonReader::_simdjson_init_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::NewJsonReader::init_reader(std::unordered_map<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::shared_ptr<doris::vectorized::VExprContext>, std::hash<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::equal_to<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::allocator<std::pair<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const, std::shared_ptr<doris::vectorized::VExprContext> > > > const&) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    4#  doris::vectorized::VFileScanner::_get_next_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    5#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
    6#  doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    7#  doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
    8#  doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
    9#  doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    10# std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
    11# doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
    12# doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
    13# start_thread
    14# __clone


    0#  doris::vectorized::VFileScanner::_get_next_reader() at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
    2#  doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
    4#  doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
    5#  doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    6#  std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
    7#  doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
    8#  doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
    9#  start_thread
    10# __clone

W20260529 08:21:27.417586 201197 scanner_scheduler.cpp:283] Scan thread read VScanner failed: [INTERNAL_ERROR]failed to init reader, err: [INTERNAL_ERROR]unknown stream load id: ee4ad23d0593a0dd-b8e586767dbe51b2

    0#  doris::FileFactory::create_pipe_reader(doris::TUniqueId const&, std::shared_ptr<doris::io::FileReader>*, doris::RuntimeState*, bool) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::NewJsonReader::_open_file_reader(bool) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    2#  doris::vectorized::NewJsonReader::_simdjson_init_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::NewJsonReader::init_reader(std::unordered_map<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::shared_ptr<doris::vectorized::VExprContext>, std::hash<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::equal_to<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::allocator<std::pair<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const, std::shared_ptr<doris::vectorized::VExprContext> > > > const&) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    4#  doris::vectorized::VFileScanner::_get_next_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    5#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489

...skipping one line
7# doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
8# doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
9# doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptrdoris::vectorized::ScannerContext, std::shared_ptrdoris::vectorized::ScanTask) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
10# std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptrdoris::vectorized::ScannerContext, std::shared_ptrdoris::vectorized::ScanTask)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
11# doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
12# doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
13# start_thread
14# __clone
. cur path:

    0#  doris::vectorized::VFileScanner::_get_next_reader() at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
    2#  doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
    4#  doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
    5#  doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    6#  std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
    7#  doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
    8#  doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
    9#  start_thread
    10# __clone

W20260529 08:21:27.417747 201068 task_scheduler.cpp:361] Pipeline task failed. query_id: ee4ad23d0593a0dd-b8e586767dbe51b2|ee4ad23d0593a0dd-b8e586767dbe51b3 reason: [INTERNAL_ERROR]failed to init reader, err: [INTERNAL_ERROR]unknown stream load id: ee4ad23d0593a0dd-b8e586767dbe51b2

    0#  doris::FileFactory::create_pipe_reader(doris::TUniqueId const&, std::shared_ptr<doris::io::FileReader>*, doris::RuntimeState*, bool) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::NewJsonReader::_open_file_reader(bool) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    2#  doris::vectorized::NewJsonReader::_simdjson_init_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::NewJsonReader::init_reader(std::unordered_map<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >, std::shared_ptr<doris::vectorized::VExprContext>, std::hash<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::equal_to<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > >, std::allocator<std::pair<std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const, std::shared_ptr<doris::vectorized::VExprContext> > > > const&) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    4#  doris::vectorized::VFileScanner::_get_next_reader() at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    5#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
    6#  doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    7#  doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
    8#  doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
    9#  doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    10# std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
    11# doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
    12# doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
    13# start_thread
    14# __clone

. cur path:

    0#  doris::vectorized::VFileScanner::_get_next_reader() at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187
    1#  doris::vectorized::VFileScanner::_get_block_wrapped(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:489
    2#  doris::vectorized::VFileScanner::_get_block_impl(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:494
    3#  doris::vectorized::VScanner::get_block(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:0
    4#  doris::vectorized::VScanner::get_block_after_projects(doris::RuntimeState*, doris::vectorized::Block*, bool*) at /home/zcp/repo_center/doris_release/doris/be/src/vec/exec/scan/vscanner.cpp:102
    5#  doris::vectorized::ScannerScheduler::_scanner_scan(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>) at /home/zcp/repo_center/doris_release/doris/be/src/common/status.h:380
    6#  std::_Function_handler<void (), doris::vectorized::ScannerScheduler::submit(std::shared_ptr<doris::vectorized::ScannerContext>, std::shared_ptr<doris::vectorized::ScanTask>)::$_1::operator()() const::{lambda()#1}>::_M_invoke(std::_Any_data const&) at /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701
    7#  doris::ThreadPool::dispatch_thread() at /home/zcp/repo_center/doris_release/doris/be/src/util/threadpool.cpp:0
    8#  doris::Thread::supervise_thread(void*) at /var/local/ldb-toolchain/bin/../usr/include/pthread.h:562
    9#  start_thread
    10# __clone

I20260529 08:21:27.417892 201068 pipeline_fragment_context.cpp:175] PipelineFragmentContext::cancel|query_id=ee4ad23d0593a0dd-b8e586767dbe51b2|fragment_id=1|instance_id=ee4ad23d0593a0dd-b8e586767dbe51b3|reason=INTERNAL_ERROR|message=failed to init reader, err: [INTERNAL_ERROR]unknown stream load id: ee4ad23d0593a0dd-b8e586767dbe51b2

What You Expected?

可以通过 stream load 加载数据

How to Reproduce?

No response

Anything Else?

No response

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_377b20f15c47

Initial maintainer triage for apache/doris#63917.

I checked the live issue metadata and the Doris 2.1.11-rc01 code path that matches the reported affected version. The issue currently has no labels. The report is credible, but the current material is not enough to produce a one-command reproducer yet.

What the supplied log proves:

• The stream load request was accepted by BE and a transaction was opened: txn_id=11142908, query_id=ee4ad23d0593a0dd-b8e586767dbe51b2.
• The failure happens while the JSON scanner is opening the stream reader, before JSON parsing or tablet writing can proceed.
• The preceding cumulative-compaction log line is very likely unrelated. The failing stack is in the stream-load file reader path.

Code-path evidence in 2.1.11:

• StreamLoadAction::_process_put() creates a StreamLoadPipe, stores it in ctx->body_sink / ctx->pipe, registers the context through new_load_stream_mgr()->put(ctx->id, ctx), then starts StreamLoadExecutor::execute_plan_fragment(ctx) for streaming formats.
• NewJsonReader::_open_file_reader() handles TFileType::FILE_STREAM by calling FileFactory::create_pipe_reader(_range.load_id, ...).
• FileFactory::create_pipe_reader() does only one lookup for this error: ExecEnv::GetInstance()->new_load_stream_mgr()->get(load_id). If that returns null, it emits the exact error in the issue: unknown stream load id: <query_id>.
• The normal removal paths for this same context include the stream-load fragment finish callback in StreamLoadExecutor::execute_plan_fragment() and HTTP request cleanup in StreamLoadAction::free_handler_ctx().

Maintainer judgment:

This should be treated as a BE stream-load context lifetime/ordering issue unless the client disconnected before sending the request body. The error is not caused by invalid JSON content by itself. For this stack to happen, the scan side has a FILE_STREAM range with the correct load id, but the in-memory stream-load context has already disappeared from NewLoadStreamMgr by the time NewJsonReader opens the pipe reader.

The highest-risk area to inspect is the race between:

1. StreamLoadAction::_process_put() registering the context and asynchronously starting the pipeline fragment;
2. StreamLoadAction::free_handler_ctx() cancelling/removing the context when the HTTP request is freed or the sender is gone;
3. the finish callback in StreamLoadExecutor::execute_plan_fragment() removing the same load id before all scan-side open paths are done.

Recommended next steps:

1. Ask the reporter for the complete BE log around this query_id / load id, especially lines before and after the shown snippet, and the final stream-load HTTP response body if one was returned.
2. Ask for the exact stream-load request headers generated by DataX: format, read_json_by_line, strip_outer_array, Content-Length vs Transfer-Encoding: chunked, timeout, and whether the client connection closed or retried at that timestamp.
3. Ask for the FE log around streamLoadPut for the same load id and the relevant FE/BE config values, especially pipeline-load related settings and stream-load timeout/body-size settings.
4. For a code fix, focus on making the stream-load pipe registration lifetime cover all scanner initialization paths. A robust direction is to avoid removing the NewLoadStreamMgr entry from a per-fragment/per-request cleanup path until the stream-load handle path has reached the final state, or to pass/hold a direct shared reference to the stream-load context/pipe so scanner initialization does not depend on a late global lookup by load id.

Missing information for a definitive reproducer:

• Minimal table DDL and stream-load command or DataX job config.
• Whether the error is stable or intermittent.
• Full BE/FE logs for the same query_id.
• The HTTP response body returned to the client.
• Whether the client uses chunked transfer or a fixed Content-Length, and whether it closes the connection early.