Join Concurrency with object_storage_cluster_join_mode='global'

[alsugiliazova]

Issue Summary

swarms/tests/joins.py fails only when scenarios run concurrently (Pool(), pool size > 1).
The same scenarios are stable with serial execution (pool size = 1).

Primary errors:

• Code: 32. DB::Exception: Attempt to read after eof ... (ATTEMPT_TO_READ_AFTER_EOF)
• Followed by Code: 210. DB::NetException: Connection refused (localhost:9000) after server instability/crash cascade.

Root Cause

In PR #1527 branch code (src/Storages/IStorageCluster.*), a mutable member field is used to control per-query behavior:

• IStorageCluster::send_external_tables (class field)
• Set in updateQueryWithJoinToSendIfNeeded(...) when GLOBAL rewrite path is taken
• Read later in read(...) to decide whether to forward temporary external tables

This field is query-specific but stored on a shared storage instance, so concurrent reads can race and leak state across unrelated queries.

Risk Pattern

• Query A enables send_external_tables = true
• Query B interleaves on same storage object and reads stale/shared flag state
• External table forwarding becomes nondeterministic per query
• GLOBAL join execution on shards receives inconsistent temporary table state
• Results in protocol/stream failures (ATTEMPT_TO_READ_AFTER_EOF) and possible crash/connection refusal cascade

Code-Level Defect

Defect class: cross-query mutable state + data race in a concurrent read path.

Affected logic:

• src/Storages/IStorageCluster.h: bool send_external_tables = false;
• src/Storages/IStorageCluster.cpp:
  • GLOBAL branch sets send_external_tables = true
  • read(...) checks if (send_external_tables && planner_context ...)

Proposed Fix

Make external-table forwarding state local to each read(...) invocation:

1. Change updateQueryWithJoinToSendIfNeeded(...) to return bool should_send_external_tables.
2. Remove class member send_external_tables.
3. In read(...), capture:
   • const bool send_external_tables = updateQueryWithJoinToSendIfNeeded(...);
4. Use this local bool when extracting planner external tables.

This scopes the decision to the current query and eliminates cross-query races.

Validation Plan

1. Re-run swarms/tests/joins.py with pool > 1 (same workload that currently fails).
2. Confirm absence of:
   • ATTEMPT_TO_READ_AFTER_EOF
   • Connection refused (localhost:9000)
3. Compare with pool=1 baseline behavior.
4. Run targeted GLOBAL-mode join combinations (s3, s3Cluster, iceberg, icebergS3Cluster) under concurrency.

Severity

High (P1) for GLOBAL join mode in concurrent workloads:

• Can cause query failures
• Can destabilize server under stress
• Produces nondeterministic behavior depending on timing/interleaving

Recommendation

Fix should be merged before relying on GLOBAL join mode in parallel regression runs.
This is a server-side correctness/concurrency issue, not a test harness issue.

Trace:

2026.03.27 15:42:15.473719 [ 74 ] {e88529f8-17de-4863-8cca-8b617c66605b} <Error> executeQuery: Code: 32. DB::Exception: Attempt to read after eof. (ATTEMPT_TO_READ_AFTER_EOF) (version 26.1.4.20001.altinityantalya) (from 172.18.0.12:44308) (query 1, line 2) (in query: SELECT __table1.boolean_col AS boolean_col, __table1.long_col AS long_col, __table1.double_col AS double_col, __table1.string_col AS string_col, __table1.timestamp_col AS timestamp_col, __table1.date_col AS date_col, __table1.time_col AS time_col, __table1.timestamptz_col AS timestamptz_col, __table1.integer_col AS integer_col, __table1.float_col AS float_col, __table1.decimal_col AS decimal_col, __table2.boolean_col AS `t2.boolean_col`, __table2.long_col AS `t2.long_col`, __table2.double_col AS `t2.double_col`, __table2.string_col AS `t2.string_col`, __table2.timestamp_col AS `t2.timestamp_col`, __table2.date_col AS `t2.date_col`, __table2.time_col AS `t2.time_col`, __table2.timestamptz_col AS `t2.timestamptz_col`, __table2.integer_col AS `t2.integer_col`, __table2.float_col AS `t2.float_col`, __table2.decimal_col AS `t2.decimal_col` FROM s3Cluster('replicated_cluster', 'http://minio:9000/warehouse/data1/data/**.parquet', 'admin', '[HIDDEN]', 'Parquet', '`boolean_col` Nullable(Bool), `long_col` Nullable(Int64), `double_col` Nullable(Float64), `string_col` Nullable(String), `timestamp_col` Nullable(DateTime64(6, \'UTC\')), `date_col` Nullable(Date32), `time_col` Nullable(DateTime64(6, \'UTC\')), `timestamptz_col` Nullable(DateTime64(6, \'UTC\')), `integer_col` Nullable(Int32), `float_col` Nullable(Float32), `decimal_col` Nullable(Decimal(10, 2))') AS __table1 GLOBAL ALL INNER JOIN _data_11397359597041654076_4211412118624154458 AS __table2 ON __table1.integer_col = __table2.integer_col ORDER BY __table1.boolean_col ASC, __table1.long_col ASC, __table1.double_col ASC, __table1.string_col ASC, __table1.timestamp_col ASC, __table1.date_col ASC, __table1.time_col ASC, __table1.timestamptz_col ASC, __table1.integer_col ASC, __table1.float_col ASC, __table1.decimal_col ASC, __table2.boolean_col ASC, __table2.long_col ASC, __table2.double_col ASC, __table2.string_col ASC, __table2.timestamp_col ASC, __table2.date_col ASC, __table2.time_col ASC, __table2.timestamptz_col ASC, __table2.integer_col ASC, __table2.float_col ASC, __table2.decimal_col ASC), Stack trace (when copying this message, always include the lines below):

0. ./ci/tmp/build/./base/poco/Foundation/src/Exception.cpp:28: Poco::Exception::Exception(String const&, int) @ 0x000000001e6253f8
1. ./ci/tmp/build/./src/Common/Exception.cpp:136: DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x0000000010f20c4c
2. ./src/Common/Exception.h:172: DB::Exception::Exception(String&&, int, String, bool) @ 0x000000000ab91b18
3. ./src/Common/Exception.h:58: DB::Exception::Exception(PreformattedMessage&&, int) @ 0x000000000ab915cc
4. ./src/Common/Exception.h:190: DB::Exception::Exception<>(int, FormatStringHelperImpl<>) @ 0x0000000010f28a14
5. ./ci/tmp/build/./src/IO/VarInt.cpp:13: DB::throwReadAfterEOF() @ 0x000000001101be44
6. ./src/IO/VarInt.h:83: DB::TCPHandler::receiveClusterFunctionReadTaskResponse(DB::QueryState&) @ 0x000000001997f598
7. ./ci/tmp/build/./src/Server/TCPHandler.cpp:714: std::shared_ptr<DB::ClusterFunctionReadTaskResponse> std::__function::__policy_func<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()>::__call_func[abi:ne210105]<DB::TCPHandler::runImpl()::$_5>(std::__function::__policy_storage const*) @ 0x0000000019982700
8. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x00000000150ff764
9. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001510ca4c
10. ./contrib/llvm-project/libcxx/include/future:1601: std::packaged_task<std::shared_ptr<DB::ObjectInfo> ()>::operator()() @ 0x000000001510ce50
11. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001105b728
12. ./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:0: ThreadFromGlobalPoolImpl<false, true>::ThreadFromGlobalPoolImpl<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&&)::'lambda'()::operator()() @ 0x000000001105ffa0
13. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001105966c
14. ./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:0: void* std::__thread_proxy[abi:ne210105]<std::tuple<std::unique_ptr<std::__thread_struct, std::default_delete<std::__thread_struct>>, void (ThreadPoolImpl<std::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::thread>::ThreadFromThreadPool*>>(void*) @ 0x000000001105ea08
15. ? @ 0x000000000007d5b8
16. ? @ 0x00000000000e5edc

2026.03.27 15:42:15.473761 [ 837 ] {f0793fe7-1cb5-49e6-a47a-2a88e3526cf6} <Error> executeQuery: Code: 32. DB::Exception: Attempt to read after eof. (ATTEMPT_TO_READ_AFTER_EOF) (version 26.1.4.20001.altinityantalya) (from 172.18.0.12:44324) (query 1, line 2) (in query: SELECT __table1.boolean_col AS boolean_col, __table1.long_col AS long_col, __table1.double_col AS double_col, __table1.string_col AS string_col, __table1.timestamp_col AS timestamp_col, __table1.date_col AS date_col, __table1.time_col AS time_col, __table1.timestamptz_col AS timestamptz_col, __table1.integer_col AS integer_col, __table1.float_col AS float_col, __table1.decimal_col AS decimal_col, __table2.boolean_col AS `t2.boolean_col`, __table2.long_col AS `t2.long_col`, __table2.double_col AS `t2.double_col`, __table2.string_col AS `t2.string_col`, __table2.timestamp_col AS `t2.timestamp_col`, __table2.date_col AS `t2.date_col`, __table2.time_col AS `t2.time_col`, __table2.timestamptz_col AS `t2.timestamptz_col`, __table2.integer_col AS `t2.integer_col`, __table2.float_col AS `t2.float_col`, __table2.decimal_col AS `t2.decimal_col` FROM icebergS3Cluster('replicated_cluster', 'http://minio:9000/warehouse/data1/', 'admin', '[HIDDEN]', 'Parquet', '`boolean_col` Nullable(Bool), `long_col` Nullable(Int64), `double_col` Nullable(Float64), `string_col` Nullable(String), `timestamp_col` Nullable(DateTime64(6)), `date_col` Nullable(Date32), `time_col` Nullable(Int64), `timestamptz_col` Nullable(DateTime64(6, \'UTC\')), `integer_col` Nullable(Int32), `float_col` Nullable(Float32), `decimal_col` Nullable(Decimal(10, 2))', SETTINGS iceberg_metadata_file_path = 'metadata/00004-f57d3eb2-ec84-4f0d-bd91-4bbb443cf557.metadata.json') AS __table1 GLOBAL ALL INNER JOIN _data_13883430509704433540_14308560915511735178 AS __table2 ON __table1.decimal_col = __table2.decimal_col ORDER BY __table1.boolean_col ASC, __table1.long_col ASC, __table1.double_col ASC, __table1.string_col ASC, __table1.timestamp_col ASC, __table1.date_col ASC, __table1.time_col ASC, __table1.timestamptz_col ASC, __table1.integer_col ASC, __table1.float_col ASC, __table1.decimal_col ASC, __table2.boolean_col ASC, __table2.long_col ASC, __table2.double_col ASC, __table2.string_col ASC, __table2.timestamp_col ASC, __table2.date_col ASC, __table2.time_col ASC, __table2.timestamptz_col ASC, __table2.integer_col ASC, __table2.float_col ASC, __table2.decimal_col ASC), Stack trace (when copying this message, always include the lines below):

0. ./ci/tmp/build/./base/poco/Foundation/src/Exception.cpp:28: Poco::Exception::Exception(String const&, int) @ 0x000000001e6253f8
1. ./ci/tmp/build/./src/Common/Exception.cpp:136: DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x0000000010f20c4c
2. ./src/Common/Exception.h:172: DB::Exception::Exception(String&&, int, String, bool) @ 0x000000000ab91b18
3. ./src/Common/Exception.h:58: DB::Exception::Exception(PreformattedMessage&&, int) @ 0x000000000ab915cc
4. ./src/Common/Exception.h:190: DB::Exception::Exception<>(int, FormatStringHelperImpl<>) @ 0x0000000010f28a14
5. ./ci/tmp/build/./src/IO/VarInt.cpp:13: DB::throwReadAfterEOF() @ 0x000000001101be44
6. ./src/IO/VarInt.h:83: DB::TCPHandler::receiveClusterFunctionReadTaskResponse(DB::QueryState&) @ 0x000000001997f598
7. ./ci/tmp/build/./src/Server/TCPHandler.cpp:714: std::shared_ptr<DB::ClusterFunctionReadTaskResponse> std::__function::__policy_func<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()>::__call_func[abi:ne210105]<DB::TCPHandler::runImpl()::$_5>(std::__function::__policy_storage const*) @ 0x0000000019982700
8. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x00000000150ff764
9. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001510ca4c
10. ./contrib/llvm-project/libcxx/include/future:1601: std::packaged_task<std::shared_ptr<DB::ObjectInfo> ()>::operator()() @ 0x000000001510ce50
11. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001105b728
12. ./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:0: ThreadFromGlobalPoolImpl<false, true>::ThreadFromGlobalPoolImpl<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&&)::'lambda'()::operator()() @ 0x000000001105ffa0
13. ./contrib/llvm-project/libcxx/include/__functional/function.h:508: ? @ 0x000000001105966c
14. ./contrib/llvm-project/libcxx/include/__type_traits/invoke.h:0: void* std::__thread_proxy[abi:ne210105]<std::tuple<std::unique_ptr<std::__thread_struct, std::default_delete<std::__thread_struct>>, void (ThreadPoolImpl<std::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::thread>::ThreadFromThreadPool*>>(void*) @ 0x000000001105ea08
15. ? @ 0x000000000007d5b8
16. ? @ 0x00000000000e5edc

[ianton-ru]

Was fixed in 0cd90a8

[alsugiliazova]

new trace:

2026.03.27 18:39:30.600939 [ 43 ] {6cfafe44-4bb8-41a7-87da-039c03b13286} <Error> executeQuery: Code: 32. DB::Exception: Attempt to read after eof. (ATTEMPT_TO_READ_AFTER_EOF) (version 26.1.4.20001.altinityantalya) (from 172.18.0.6:46586) (query 1, line 2) (in query: SELECT __table1.boolean_col AS boolean_col, __table1.long_col AS long_col, __table1.double_col AS double_col, __table1.string_col AS string_col, __table1.timestamp_col AS timestamp_col, __table1.date_col AS date_col, __table1.time_col AS time_col, __table1.timestamptz_col AS timestamptz_col, __table1.integer_col AS integer_col, __table1.float_col AS float_col, __table1.decimal_col AS decimal_col, __table2.boolean_col AS `t2.boolean_col`, __table2.long_col AS `t2.long_col`, __table2.double_col AS `t2.double_col`, __table2.string_col AS `t2.string_col`, __table2.timestamp_col AS `t2.timestamp_col`, __table2.date_col AS `t2.date_col`, __table2.time_col AS `t2.time_col`, __table2.timestamptz_col AS `t2.timestamptz_col`, __table2.integer_col AS `t2.integer_col`, __table2.float_col AS `t2.float_col`, __table2.decimal_col AS `t2.decimal_col` FROM s3Cluster(replicated_cluster, 'http://minio:9000/warehouse/data1/data/**.parquet', 'admin', '[HIDDEN]', 'Parquet', '`boolean_col` Nullable(Bool), `long_col` Nullable(Int64), `double_col` Nullable(Float64), `string_col` Nullable(String), `timestamp_col` Nullable(DateTime64(6, \'UTC\')), `date_col` Nullable(Date32), `time_col` Nullable(DateTime64(6, \'UTC\')), `timestamptz_col` Nullable(DateTime64(6, \'UTC\')), `integer_col` Nullable(Int32), `float_col` Nullable(Float32), `decimal_col` Nullable(Decimal(10, 2))') AS __table1 GLOBAL ALL INNER JOIN _data_8097750655037348555_2764925514922875233 AS __table2 ON (__table1.boolean_col = __table2.boolean_col) AND (__table1.string_col = __table2.string_col) ORDER BY __table1.boolean_col ASC, __table1.long_col ASC, __table1.double_col ASC, __table1.string_col ASC, __table1.timestamp_col ASC, __table1.date_col ASC, __table1.time_col ASC, __table1.timestamptz_col ASC, __table1.integer_col ASC, __table1.float_col ASC, __table1.decimal_col ASC, __table2.boolean_col ASC, __table2.long_col ASC, __table2.double_col ASC, __table2.string_col ASC, __table2.timestamp_col ASC, __table2.date_col ASC, __table2.time_col ASC, __table2.timestamptz_col ASC, __table2.integer_col ASC, __table2.float_col ASC, __table2.decimal_col ASC), Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x0000000012c77f50
1. DB::Exception::Exception(String&&, int, String, bool) @ 0x000000000c833018
2. DB::Exception::Exception(PreformattedMessage&&, int) @ 0x000000000c83294c
3. DB::Exception::Exception<>(int, FormatStringHelperImpl<>) @ 0x0000000012c76554
4. DB::throwReadAfterEOF() @ 0x0000000012d749a4
5. std::shared_ptr<DB::ClusterFunctionReadTaskResponse> std::__function::__policy_func<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()>::__call_func[abi:ne210105]<DB::TCPHandler::runImpl()::$_5>(std::__function::__policy_storage const*) @ 0x0000000018d13860
6. std::shared_ptr<DB::ObjectInfo> std::__function::__policy_func<std::shared_ptr<DB::ObjectInfo> ()>::__call_func[abi:ne210105]<DB::StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator(std::function<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()> const&, unsigned long, bool, std::shared_ptr<DB::IObjectStorage>, String const&, std::shared_ptr<DB::Context const>)::$_0>(std::__function::__policy_storage const*) @ 0x0000000015d6d5a8
7. std::__packaged_task_func<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda'(), std::allocator<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda'()>, std::shared_ptr<DB::ObjectInfo> ()>::operator()() @ 0x0000000015d6ce08
8. void std::__function::__policy_func<void ()>::__call_func[abi:ne210105]<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda0'()>(std::__function::__policy_storage const*) @ 0x0000000015d6d0c4
9. ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::worker() @ 0x0000000012dba960
10. void std::__function::__policy_func<void ()>::__call_func[abi:ne210105]<ThreadFromGlobalPoolImpl<false, true>::ThreadFromGlobalPoolImpl<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&&)::'lambda'()>(std::__function::__policy_storage const*) @ 0x0000000012dc0d20
11. ThreadPoolImpl<std::thread>::ThreadFromThreadPool::worker() @ 0x0000000012db7f3c
12. void* std::__thread_proxy[abi:ne210105]<std::tuple<std::unique_ptr<std::__thread_struct, std::default_delete<std::__thread_struct>>, void (ThreadPoolImpl<std::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::thread>::ThreadFromThreadPool*>>(void*) @ 0x0000000012dbee1c
13. ? @ 0x000000000007d5b8
14. ? @ 0x00000000000e5edc

2026.03.27 18:39:30.602516 [ 43 ] {} <Error> TCPHandler: Code: 32. DB::Exception: Attempt to read after eof. (ATTEMPT_TO_READ_AFTER_EOF), Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0x0000000012c77f50
1. DB::Exception::Exception(String&&, int, String, bool) @ 0x000000000c833018
2. DB::Exception::Exception(PreformattedMessage&&, int) @ 0x000000000c83294c
3. DB::Exception::Exception<>(int, FormatStringHelperImpl<>) @ 0x0000000012c76554
4. DB::throwReadAfterEOF() @ 0x0000000012d749a4
5. std::shared_ptr<DB::ClusterFunctionReadTaskResponse> std::__function::__policy_func<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()>::__call_func[abi:ne210105]<DB::TCPHandler::runImpl()::$_5>(std::__function::__policy_storage const*) @ 0x0000000018d13860
6. std::shared_ptr<DB::ObjectInfo> std::__function::__policy_func<std::shared_ptr<DB::ObjectInfo> ()>::__call_func[abi:ne210105]<DB::StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator(std::function<std::shared_ptr<DB::ClusterFunctionReadTaskResponse> ()> const&, unsigned long, bool, std::shared_ptr<DB::IObjectStorage>, String const&, std::shared_ptr<DB::Context const>)::$_0>(std::__function::__policy_storage const*) @ 0x0000000015d6d5a8
7. std::__packaged_task_func<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda'(), std::allocator<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda'()>, std::shared_ptr<DB::ObjectInfo> ()>::operator()() @ 0x0000000015d6ce08
8. void std::__function::__policy_func<void ()>::__call_func[abi:ne210105]<std::function<std::future<std::shared_ptr<DB::ObjectInfo>> (std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)> DB::threadPoolCallbackRunnerUnsafe<std::shared_ptr<DB::ObjectInfo>, std::function<std::shared_ptr<DB::ObjectInfo> ()>>(ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>&, DB::ThreadName)::'lambda'(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::operator()(std::function<std::shared_ptr<DB::ObjectInfo> ()>&&, Priority)::'lambda0'()>(std::__function::__policy_storage const*) @ 0x0000000015d6d0c4
9. ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::worker() @ 0x0000000012dba960
10. void std::__function::__policy_func<void ()>::__call_func[abi:ne210105]<ThreadFromGlobalPoolImpl<false, true>::ThreadFromGlobalPoolImpl<void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*>(void (ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool::*&&)(), ThreadPoolImpl<ThreadFromGlobalPoolImpl<false, true>>::ThreadFromThreadPool*&&)::'lambda'()>(std::__function::__policy_storage const*) @ 0x0000000012dc0d20
11. ThreadPoolImpl<std::thread>::ThreadFromThreadPool::worker() @ 0x0000000012db7f3c
12. void* std::__thread_proxy[abi:ne210105]<std::tuple<std::unique_ptr<std::__thread_struct, std::default_delete<std::__thread_struct>>, void (ThreadPoolImpl<std::thread>::ThreadFromThreadPool::*)(), ThreadPoolImpl<std::thread>::ThreadFromThreadPool*>>(void*) @ 0x0000000012dbee1c
13. ? @ 0x000000000007d5b8
14. ? @ 0x00000000000e5edc

[alsugiliazova]

The send_external_tables shared-state bug above was fixed in PR #1527 commit 0cd90a8 (field removed and read-path check changed), but concurrency failures still reproduce .

Likely second root cause is in object storage task distribution under concurrent cluster-function reads:

• StorageObjectStorageCluster::TaskDistributor is shared and serves concurrent read-task callbacks.
• StorageObjectStorageStableTaskDistributor::getNextTask(...) touches replica_to_files_to_be_processed without consistent locking.
• StorageObjectStorageStableTaskDistributor::rescheduleTasksFromReplica(...) mutates/erases the same map under lock when replicas are marked lost.
• Concurrent interleaving can produce inconsistent task state for cluster read-task protocol, surfacing as:
  • ATTEMPT_TO_READ_AFTER_EOF
  • stack traces through TCPHandler::receiveClusterFunctionReadTaskResponse(...)

This explains why fixing send_external_tables alone does not fully resolve the pool>1 failures.

Validation Plan

1. Re-run swarms/tests/joins.py with pool > 1 (same workload that currently fails).
2. Confirm absence of:
   • ATTEMPT_TO_READ_AFTER_EOF
   • Connection refused (localhost:9000)
3. Compare with pool=1 baseline behavior.
4. Run targeted GLOBAL-mode join combinations (s3, s3Cluster, iceberg, icebergS3Cluster) under concurrency.

Severity

High (P1) for GLOBAL join mode in concurrent workloads:

• Can cause query failures
• Can destabilize server under stress
• Produces nondeterministic behavior depending on timing/interleaving

[alsugiliazova]

Current evidence indicates the remaining failure is specific to the PR #1527 GLOBAL path (not the LOCAL path):

• IStorageCluster::updateQueryWithJoinToSendIfNeeded(...) in GLOBAL mode rewrites joins via:
  • rewriteJoinToGlobalJoin(...)
  • buildQueryTreeForShard(..., allow_global_join_for_right_table=true, find_cross_join=true)
• buildQueryTreeForShard(...) materializes right-side inputs into temporary _data_* tables using executeSubqueryNode(...).
• read(...) forwards these temporary external tables to remote shards for execution of GLOBAL ... JOIN _data_*.

Observed failing queries and traces match this exact shape:

• Query text contains GLOBAL ... JOIN _data_*
• Stack trace repeatedly reaches TCPHandler::receiveClusterFunctionReadTaskResponse(...)
• Failure signature is protocol read breakage: ATTEMPT_TO_READ_AFTER_EOF

Because LOCAL mode does not use this _data_* GLOBAL materialization path, LOCAL can pass under the same build and workload while GLOBAL fails.