From c332fefccbc116c19ca9f04be1decd04496044e4 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Thu, 15 Oct 2026 17:16:21 +0200 Subject: [PATCH 1/3] Pass `ExternalStorageCache` into Iceberg DROP PARTITION metadata reads `getManifestFileEntriesHandle` and `generateManifestList` take the cache after external-path support, but `AlterDropPartitionExecutor` still used the old signatures, so the amd binary build failed. https: //altinity-build-artifacts.s3.amazonaws.com/json.html?PR=2575&sha=de846f87651f582609e740036d2a9ba9db179950&name_0=PR&name_1=Build%20%28amd_binary%29 https: //github.com/Altinity/ClickHouse/pull/2575 Co-authored-by: Cursor --- .../DataLakes/Iceberg/AlterDropPartitionExecutor.cpp | 9 +++++++-- .../DataLakes/Iceberg/AlterDropPartitionExecutor.h | 3 +++ .../ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp | 1 + 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp index af38b06ed41aa..faf3b30e2a176 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.cpp @@ -215,6 +215,7 @@ AlterDropPartitionExecutor::AlterDropPartitionExecutor( const IcebergMetadata & metadata_, ContextPtr context_, ObjectStoragePtr object_storage_, + ExternalStorageCache & external_storages_, const PersistentTableComponents & components_, const DataLakeStorageSettings & data_lake_settings_, String write_format_, @@ -223,6 +224,7 @@ AlterDropPartitionExecutor::AlterDropPartitionExecutor( , metadata(metadata_) , context(context_) , object_storage(std::move(object_storage_)) + , external_storages(external_storages_) , components(components_) , data_lake_settings(data_lake_settings_) , write_format(std::move(write_format_)) @@ -365,7 +367,8 @@ AlterDropPartitionExecutor::discoverTargetFilePaths(const SnapshotState & state, for (const auto & manifest_key : state.snapshot->manifest_list_entries) { - auto handle = getManifestFileEntriesHandle(object_storage, components, context, log, manifest_key, state.schema_id); + auto handle = getManifestFileEntriesHandle( + object_storage, components, context, log, manifest_key, state.schema_id, external_storages); collect(handle.getFilesWithoutDeleted(FileContentType::DATA)); collect(handle.getFilesWithoutDeleted(FileContentType::POSITION_DELETE)); @@ -460,7 +463,8 @@ AlterDropPartitionExecutor::buildDropPlan(const SnapshotState & state, const Tar for (const auto & manifest_key : state.snapshot->manifest_list_entries) { - auto handle = getManifestFileEntriesHandle(object_storage, components, context, log, manifest_key, state.schema_id); + auto handle = getManifestFileEntriesHandle( + object_storage, components, context, log, manifest_key, state.schema_id, external_storages); size_t entries_to_keep = 0; size_t entries_to_remove = 0; @@ -575,6 +579,7 @@ AlterDropPartitionExecutor::ManifestListWriteResult AlterDropPartitionExecutor:: components.path_resolver, state.metadata_object, object_storage, + external_storages, context, /*manifest_entry_names=*/{}, new_snapshot, diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h index de47a35ab9ec4..a462b94962a57 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/AlterDropPartitionExecutor.h @@ -30,6 +30,7 @@ namespace DB class FileNamesGenerator; class IcebergMetadata; +struct ExternalStorageCache; namespace Iceberg { @@ -42,6 +43,7 @@ class AlterDropPartitionExecutor const IcebergMetadata & metadata_, ContextPtr context_, ObjectStoragePtr object_storage_, + ExternalStorageCache & external_storages_, const PersistentTableComponents & components_, const DataLakeStorageSettings & data_lake_settings_, String write_format_, @@ -106,6 +108,7 @@ class AlterDropPartitionExecutor const IcebergMetadata & metadata; ContextPtr context; ObjectStoragePtr object_storage; + ExternalStorageCache & external_storages; const PersistentTableComponents & components; const DataLakeStorageSettings & data_lake_settings; String write_format; diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 2eb3feca93452..b892c9b105ae7 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -973,6 +973,7 @@ void IcebergMetadata::alterPartitionDropImpl(const PartitionCommand & command, C *this, context, object_storage, + *external_storages, persistent_components, data_lake_settings, write_format, From 5b5e0110ee3fb875cecf1df57097dbacb4c613f6 Mon Sep 17 00:00:00 2001 From: Anton Ivashkin Date: Fri, 16 Oct 2026 11:37:31 +0200 Subject: [PATCH 2/3] Send Iceberg file statistics and the task retry delay in the Antalya protocol trailer. Frontport of #2482 and #2489. On `antalya-26.8` version 2 already means absolute Iceberg paths, so `DataFileMetaInfo` and `retry_after_us` are version 3. A peer below that version gets a plain object path, and `lock_object_storage_task_distribution_ms` is ignored for the query. Co-authored-by: Cursor --- docs/en/antalya/protocol.md | 14 +- .../sql-reference/distribution-on-cluster.md | 2 + .../interfaces/specs/NativeProtocol.mdx | 2 +- src/Client/Connection.h | 2 + src/Client/IConnections.h | 2 + src/Client/MultiplexedConnections.h | 5 + src/Core/AntalyaProtocol.h | 5 +- src/Interpreters/ClusterFunctionReadTask.cpp | 39 +++- src/Interpreters/ClusterFunctionReadTask.h | 11 +- .../gtest_cluster_function_read_task.cpp | 174 +++++++++++++++++- src/QueryPipeline/RemoteQueryExecutor.cpp | 4 +- src/QueryPipeline/RemoteQueryExecutor.h | 2 +- src/Server/TCPHandler.cpp | 2 +- .../StorageObjectStorageCluster.cpp | 26 +-- .../StorageObjectStorageSource.cpp | 47 +++-- .../StorageObjectStorageSource.h | 5 + ...rageObjectStorageStableTaskDistributor.cpp | 53 +++--- ...torageObjectStorageStableTaskDistributor.h | 23 ++- .../tests/gtest_rendezvous_hashing.cpp | 26 +-- src/Storages/StorageFileCluster.cpp | 2 +- src/Storages/StorageURLCluster.cpp | 2 +- .../configs/config.d/cluster.xml | 32 ++++ .../test.py | 70 +++++++ 23 files changed, 466 insertions(+), 84 deletions(-) create mode 100644 tests/integration/test_iceberg_mixed_antalya_upstream/configs/config.d/cluster.xml create mode 100644 tests/integration/test_iceberg_mixed_antalya_upstream/test.py diff --git a/docs/en/antalya/protocol.md b/docs/en/antalya/protocol.md index d66326b0f1cb5..a385e22db77a1 100644 --- a/docs/en/antalya/protocol.md +++ b/docs/en/antalya/protocol.md @@ -14,8 +14,8 @@ upstream ClickHouse cannot reach, defined in `src/Core/AntalyaProtocol.h`. Both the name string of their `Hello`, on every connection: ```text -client -> server "ClickHouse client (antalya:2)" -server -> client "ClickHouse (antalya:2)" +client -> server "ClickHouse client (antalya:3)" +server -> client "ClickHouse (antalya:3)" ``` Each side parses the peer's suffix, caps the value with `min(own, peer)` and keeps the result. `0` @@ -30,6 +30,14 @@ outside the table location, including files on another object storage. For a wor the initiator strips the scheme and authority as before, and refuses to dispatch a data file that needs another storage or whose delete-file keys differ from the resolved keys. +Version 3 appends two optional fields to `ReadTaskResponse`, after the upstream cluster-processing +payload (including `read_source_index` when that cluster-protocol version writes it). The first is +Iceberg column statistics (`DataFileMetaInfo`). The second is `retry_after_us`, sent when +`lock_object_storage_task_distribution_ms` asks a worker to wait before taking another node's file. +If any worker negotiated a lower version, the initiator ignores that setting for the query and +assigns files immediately, so every task path stays an object key. On `antalya-26.6` these two +fields used version 2; here version 2 is the absolute-path layout, so the fields are version 3. + ## Adding an Antalya-only wire change {#adding-a-wire-change} - Bump `DBMS_ANTALYA_PROTOCOL_VERSION` by one and gate the change on the negotiated value. @@ -37,7 +45,7 @@ needs another storage or whose delete-file keys differ from the resolved keys. `DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION` for a feature upstream does not have. - Keep the counter cumulative. A backport takes the whole contiguous range up to the value it needs, or does not bump at all - the `min(own, server)` cap is only sound for a cumulative feature set. -- Update this page, and update `docs/en/interfaces/specs/NativeProtocol.md` when the change alters +- Update this page, and update `docs/reference/interfaces/specs/NativeProtocol.mdx` when the change alters a packet layout described there. ## Why a counter of our own {#why-a-counter-of-our-own} diff --git a/docs/en/sql-reference/distribution-on-cluster.md b/docs/en/sql-reference/distribution-on-cluster.md index bbc6407547d42..82e4c3fcae949 100644 --- a/docs/en/sql-reference/distribution-on-cluster.md +++ b/docs/en/sql-reference/distribution-on-cluster.md @@ -16,6 +16,8 @@ This improves cache efficiency by minimizing data movement among nodes. Each node begins processing files for which it is the primary node. After completing its assigned files, a node may take tasks from other nodes, either immediately or after waiting for `lock_object_storage_task_distribution_ms` milliseconds if the primary node does not request new files during that interval. The default value of `lock_object_storage_task_distribution_ms` is 500 milliseconds. This setting balances between caching efficiency and workload redistribution when nodes are imbalanced. +The wait is a field of `ReadTaskResponse` and is sent only when every worker negotiated Antalya protocol version 3. If any worker negotiated a lower version, including an upstream node, the setting is ignored for that query and files are assigned immediately. + ## `SYSTEM STOP SWARM MODE` command {#system-stop-swarm-mode} If a node needs to shut down gracefully, the command `SYSTEM STOP SWARM MODE` prevents the node from receiving new tasks for *Cluster-family queries. The node finishes processing already assigned files before it can safely shut down without errors. diff --git a/docs/reference/interfaces/specs/NativeProtocol.mdx b/docs/reference/interfaces/specs/NativeProtocol.mdx index 3480f44075162..a01f74cb88037 100644 --- a/docs/reference/interfaces/specs/NativeProtocol.mdx +++ b/docs/reference/interfaces/specs/NativeProtocol.mdx @@ -904,7 +904,7 @@ Below version `8` the announcement is fire-and-forget regardless of mode, and th | 6 | KeepAlive | not specified | Connection keepalive | | 7 | Scalar | not specified | Scalar data block | | 8 | IgnoredPartUUIDs | not specified | Parts to exclude from query | -| 9 | ReadTaskResponse | not specified | S3 cluster read response | +| 9 | ReadTaskResponse | not specified | S3 cluster read response. When both peers negotiated [Antalya protocol](/antalya/protocol) version 3 or newer, the initiator appends optional Iceberg column statistics and then optional `retry_after_us` for `lock_object_storage_task_distribution_ms` after the upstream payload. If any worker negotiated a lower Antalya version, `retry_after_us` is omitted and that setting is ignored for the query. | | 10 | MergeTreeReadTaskResponse | not specified | Parallel read task response | | 11 | SSHChallengeRequest | [SSH auth](#ssh-authentication) | SSH auth challenge request | | 12 | SSHChallengeResponse | [SSH auth](#ssh-authentication) | SSH auth challenge response | diff --git a/src/Client/Connection.h b/src/Client/Connection.h index ac5d4f57e84c4..bdaa422407d5b 100644 --- a/src/Client/Connection.h +++ b/src/Client/Connection.h @@ -117,6 +117,8 @@ class Connection : public IServerConnection UInt64 getServerRevision(const ConnectionTimeouts & timeouts) override; + UInt64 getServerAntalyaProtocolVersion() const { return server_antalya_protocol_version; } + const String & getServerTimezone(const ConnectionTimeouts & timeouts) override; const String & getServerDisplayName(const ConnectionTimeouts & timeouts) override; diff --git a/src/Client/IConnections.h b/src/Client/IConnections.h index 19c864306a160..70c9e49e10e1e 100644 --- a/src/Client/IConnections.h +++ b/src/Client/IConnections.h @@ -30,6 +30,8 @@ class IConnections : boost::noncopyable virtual void sendQueryPlan(const QueryPlan & query_plan) = 0; virtual void sendClusterFunctionReadTaskResponse(const ClusterFunctionReadTaskResponse &) = 0; + + virtual UInt64 getServerAntalyaProtocolVersion() const { return 0; } virtual void sendMergeTreeReadTaskResponse(const ParallelReadResponse & response) = 0; virtual void sendMergeTreeAllRangesAnnouncementResponse(const InitialAllRangesAnnouncementResponse & response) = 0; diff --git a/src/Client/MultiplexedConnections.h b/src/Client/MultiplexedConnections.h index 80164397d7da5..420ea1a474f26 100644 --- a/src/Client/MultiplexedConnections.h +++ b/src/Client/MultiplexedConnections.h @@ -42,6 +42,11 @@ class MultiplexedConnections final : public IConnections void sendQueryPlan(const QueryPlan & query_plan) override; void sendClusterFunctionReadTaskResponse(const ClusterFunctionReadTaskResponse & response) override; + + UInt64 getServerAntalyaProtocolVersion() const override + { + return current_connection ? current_connection->getServerAntalyaProtocolVersion() : 0; + } void sendMergeTreeReadTaskResponse(const ParallelReadResponse & response) override; void sendMergeTreeAllRangesAnnouncementResponse(const InitialAllRangesAnnouncementResponse & response) override; diff --git a/src/Core/AntalyaProtocol.h b/src/Core/AntalyaProtocol.h index d422c1698b940..d88ecd7a5cd1c 100644 --- a/src/Core/AntalyaProtocol.h +++ b/src/Core/AntalyaProtocol.h @@ -9,8 +9,11 @@ namespace DB /// Bump for every Antalya-only wire protocol change. See `docs/en/antalya/protocol.md`. /// 2: the `*Cluster` read-task payload carries Iceberg delete-file paths as absolute URIs. +/// 3: `ReadTaskResponse` then carries optional `DataFileMetaInfo`, then optional `retry_after_us`. +/// Both fields ship together. Version 2 stays the absolute-path layout. static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH = 2; -static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = DBMS_ANTALYA_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH; +static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS = 3; +static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS; namespace AntalyaProtocol { diff --git a/src/Interpreters/ClusterFunctionReadTask.cpp b/src/Interpreters/ClusterFunctionReadTask.cpp index 31b4022cf0816..05a7519bfbe7b 100644 --- a/src/Interpreters/ClusterFunctionReadTask.cpp +++ b/src/Interpreters/ClusterFunctionReadTask.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -213,9 +214,28 @@ void ClusterFunctionReadTaskResponse::serialize(WriteBuffer & out, size_t worker protocol_version, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION_WITH_READ_SOURCE_INDEX); } + + if (worker_antalya_protocol_version >= DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS) + { + if (file_meta_info && *file_meta_info) + { + writeVarUInt(1, out); + (*file_meta_info)->serialize(out); + } + else + writeVarUInt(0, out); + + if (retry_after_us) + { + writeVarUInt(1, out); + writeVarUInt(*retry_after_us, out); + } + else + writeVarUInt(0, out); + } } -void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in) +void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in, UInt64 antalya_protocol_version) { size_t protocol_version = 0; readVarUInt(protocol_version, in); @@ -278,6 +298,23 @@ void ClusterFunctionReadTaskResponse::deserialize(ReadBuffer & in) read_source_index = value; } } + + if (antalya_protocol_version >= DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS) + { + UInt64 has_file_meta_info = 0; + readVarUInt(has_file_meta_info, in); + if (has_file_meta_info) + file_meta_info = std::make_shared(DataFileMetaInfo::deserialize(in)); + + UInt64 has_retry_after_us = 0; + readVarUInt(has_retry_after_us, in); + if (has_retry_after_us) + { + UInt64 value = 0; + readVarUInt(value, in); + retry_after_us = value; + } + } } } diff --git a/src/Interpreters/ClusterFunctionReadTask.h b/src/Interpreters/ClusterFunctionReadTask.h index 6d479258573ee..e9cce22680b75 100644 --- a/src/Interpreters/ClusterFunctionReadTask.h +++ b/src/Interpreters/ClusterFunctionReadTask.h @@ -27,20 +27,25 @@ struct ClusterFunctionReadTaskResponse std::optional iceberg_info; /// File's columns info std::optional file_meta_info; + /// Microseconds the worker should wait before asking for another task. + /// Set only when `path` is empty. Written when the negotiated Antalya version is at least + /// `DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS`. + std::optional retry_after_us; /// Convert received response into ObjectInfo. ObjectInfoPtr getObjectInfo() const; /// Whether response is empty. - /// It is used to identify an end of processing. - bool isEmpty() const { return path.empty(); } + /// It is used to identify an end of processing. A retry request has no path and is not the end. + bool isEmpty() const { return path.empty() && !retry_after_us; } /// Serialize according to the protocol version. /// `worker_antalya_protocol_version` is the Antalya protocol version negotiated with the worker. void serialize(WriteBuffer & out, size_t worker_protocol_version, UInt64 worker_antalya_protocol_version) const; /// Deserialize. Protocol version will be received from `in` /// and the result will be deserialized accordingly. - void deserialize(ReadBuffer & in); + /// `antalya_protocol_version` is the version negotiated with the peer that wrote the packet. + void deserialize(ReadBuffer & in, UInt64 antalya_protocol_version = 0); }; using ClusterFunctionReadTaskResponsePtr = std::shared_ptr; diff --git a/src/Interpreters/tests/gtest_cluster_function_read_task.cpp b/src/Interpreters/tests/gtest_cluster_function_read_task.cpp index 7dfc232b62df2..63b0c0dfb8d1f 100644 --- a/src/Interpreters/tests/gtest_cluster_function_read_task.cpp +++ b/src/Interpreters/tests/gtest_cluster_function_read_task.cpp @@ -76,7 +76,7 @@ TEST(ClusterFunctionReadTaskResponse, RoundTripsSchemaTransformOnSupportedProtoc ReadBufferFromString in(serialized); ClusterFunctionReadTaskResponse deserialized; - deserialized.deserialize(in); + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION); ASSERT_TRUE(deserialized.data_lake_metadata.schema_transform); EXPECT_FALSE(deserialized.data_lake_metadata.schema_transform->getInputs().empty()); @@ -131,7 +131,7 @@ TEST(ClusterFunctionReadTaskResponse, RoundTripsExcludedRowsOnSupportedProtocol) ReadBufferFromString in(serialized); ClusterFunctionReadTaskResponse deserialized; - deserialized.deserialize(in); + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION); ASSERT_TRUE(deserialized.data_lake_metadata.excluded_rows); EXPECT_EQ(deserialized.data_lake_metadata.excluded_rows->size(), 2u); @@ -246,7 +246,7 @@ TEST(ClusterFunctionReadTaskResponse, RoundTripsIcebergDeletesOnSupportedProtoco ReadBufferFromString in(serialized); ClusterFunctionReadTaskResponse deserialized; - deserialized.deserialize(in); + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION); ASSERT_TRUE(deserialized.iceberg_info.has_value()); ASSERT_EQ(deserialized.iceberg_info->equality_deletes_objects.size(), 1u); @@ -304,7 +304,7 @@ TEST(ClusterFunctionReadTaskResponse, RoundTripsFileBucketInfoOnSupportedProtoco ReadBufferFromString in(serialized); ClusterFunctionReadTaskResponse deserialized; - deserialized.deserialize(in); + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION); ASSERT_TRUE(deserialized.file_bucket_info); auto * parquet_buckets = dynamic_cast(deserialized.file_bucket_info.get()); @@ -313,3 +313,169 @@ TEST(ClusterFunctionReadTaskResponse, RoundTripsFileBucketInfoOnSupportedProtoco } #endif + +namespace +{ + +ClusterFunctionReadTaskResponse makeResponseWithFileMeta() +{ + ClusterFunctionReadTaskResponse response; + response.path = "data/file.parquet"; + auto meta = std::make_shared(); + DataFileMetaInfo::ColumnInfo column; + column.rows_count = 10; + column.nulls_count = 0; + column.hyperrectangle = Range(Field(Int64(1)), true, Field(Int64(5)), true); + meta->columns_info.emplace("id", std::move(column)); + response.file_meta_info = std::move(meta); + return response; +} + +} + +TEST(ClusterFunctionReadTaskResponse, OmitsFileMetaInfoForUpstreamPeer) +{ + auto with_meta = makeResponseWithFileMeta(); + ClusterFunctionReadTaskResponse without_meta; + without_meta.path = with_meta.path; + + String with_meta_bytes; + String without_meta_bytes; + String version_two_bytes; + { + WriteBufferFromString out(with_meta_bytes); + with_meta.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, /*antalya_protocol_version*/ 0); + out.finalize(); + } + { + WriteBufferFromString out(without_meta_bytes); + without_meta.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, /*antalya_protocol_version*/ 0); + out.finalize(); + } + { + WriteBufferFromString out(version_two_bytes); + with_meta.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_ICEBERG_ABSOLUTE_PATH); + out.finalize(); + } + EXPECT_EQ(with_meta_bytes, without_meta_bytes); + EXPECT_EQ(with_meta_bytes, version_two_bytes); + + ReadBufferFromString in(with_meta_bytes); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, /*antalya_protocol_version*/ 0); + EXPECT_FALSE(deserialized.file_meta_info.has_value()); + EXPECT_EQ(deserialized.path, "data/file.parquet"); + EXPECT_TRUE(in.eof()); +} + +TEST(ClusterFunctionReadTaskResponse, RoundTripsFileMetaInfoOnAntalyaProtocol) +{ + auto response = makeResponseWithFileMeta(); + + String serialized; + WriteBufferFromString out(serialized); + response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + out.finalize(); + + ReadBufferFromString in(serialized); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + + ASSERT_TRUE(deserialized.file_meta_info && *deserialized.file_meta_info); + const auto & column = (*deserialized.file_meta_info)->columns_info.at("id"); + ASSERT_TRUE(column.rows_count.has_value()); + ASSERT_TRUE(column.nulls_count.has_value()); + EXPECT_EQ(*column.rows_count, 10); + EXPECT_EQ(*column.nulls_count, 0); + ASSERT_TRUE(column.hyperrectangle.has_value()); + EXPECT_EQ(column.hyperrectangle->left.safeGet(), 1); + EXPECT_EQ(column.hyperrectangle->right.safeGet(), 5); + EXPECT_TRUE(in.eof()); +} + +TEST(ClusterFunctionReadTaskResponse, AntalyaProtocolWithoutFileMetaInfoConsumesPresenceFlag) +{ + ClusterFunctionReadTaskResponse response; + response.path = "data/file.parquet"; + + String serialized; + WriteBufferFromString out(serialized); + response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + out.finalize(); + + ReadBufferFromString in(serialized); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + EXPECT_FALSE(deserialized.file_meta_info.has_value()); + EXPECT_TRUE(in.eof()); +} + +TEST(ClusterFunctionReadTaskResponse, OmitsRetryAfterUsForUpstreamPeer) +{ + ClusterFunctionReadTaskResponse with_retry; + with_retry.path = "data/file.parquet"; + with_retry.retry_after_us = 1000; + + ClusterFunctionReadTaskResponse without_retry; + without_retry.path = with_retry.path; + + String with_retry_bytes; + String without_retry_bytes; + { + WriteBufferFromString out(with_retry_bytes); + with_retry.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, /*antalya_protocol_version*/ 0); + out.finalize(); + } + { + WriteBufferFromString out(without_retry_bytes); + without_retry.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, /*antalya_protocol_version*/ 0); + out.finalize(); + } + EXPECT_EQ(with_retry_bytes, without_retry_bytes); + + ReadBufferFromString in(with_retry_bytes); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, /*antalya_protocol_version*/ 0); + EXPECT_FALSE(deserialized.retry_after_us.has_value()); + EXPECT_EQ(deserialized.path, "data/file.parquet"); + EXPECT_TRUE(in.eof()); +} + +TEST(ClusterFunctionReadTaskResponse, RoundTripsRetryAfterUsOnAntalyaProtocol) +{ + ClusterFunctionReadTaskResponse response; + response.retry_after_us = 2500; + EXPECT_FALSE(response.isEmpty()); + + String serialized; + WriteBufferFromString out(serialized); + response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + out.finalize(); + + ReadBufferFromString in(serialized); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + ASSERT_TRUE(deserialized.retry_after_us.has_value()); + EXPECT_EQ(*deserialized.retry_after_us, 2500); + EXPECT_TRUE(deserialized.path.empty()); + EXPECT_FALSE(deserialized.isEmpty()); + EXPECT_TRUE(in.eof()); +} + +TEST(ClusterFunctionReadTaskResponse, AntalyaProtocolWithoutRetryConsumesPresenceFlag) +{ + ClusterFunctionReadTaskResponse response; + response.path = "data/file.parquet"; + + String serialized; + WriteBufferFromString out(serialized); + response.serialize(out, DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + out.finalize(); + + ReadBufferFromString in(serialized); + ClusterFunctionReadTaskResponse deserialized; + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS); + EXPECT_FALSE(deserialized.retry_after_us.has_value()); + EXPECT_EQ(deserialized.path, "data/file.parquet"); + EXPECT_TRUE(in.eof()); +} diff --git a/src/QueryPipeline/RemoteQueryExecutor.cpp b/src/QueryPipeline/RemoteQueryExecutor.cpp index b6bf2f73d840a..ce06d7be101bd 100644 --- a/src/QueryPipeline/RemoteQueryExecutor.cpp +++ b/src/QueryPipeline/RemoteQueryExecutor.cpp @@ -904,7 +904,9 @@ void RemoteQueryExecutor::processReadTaskRequest() ProfileEvents::increment(ProfileEvents::ReadTaskRequestsReceived); - auto response = (*extension->task_iterator)(extension->replica_info->number_of_current_replica); + auto response = (*extension->task_iterator)( + extension->replica_info->number_of_current_replica, + connections->getServerAntalyaProtocolVersion()); connections->sendClusterFunctionReadTaskResponse(*response); } diff --git a/src/QueryPipeline/RemoteQueryExecutor.h b/src/QueryPipeline/RemoteQueryExecutor.h index 8cd35f2a18fa4..d4ea5e53db4d2 100644 --- a/src/QueryPipeline/RemoteQueryExecutor.h +++ b/src/QueryPipeline/RemoteQueryExecutor.h @@ -51,7 +51,7 @@ class TaskIterator { throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method rescheduleTasksFromReplica is not implemented"); } - virtual ClusterFunctionReadTaskResponsePtr operator()(size_t number_of_current_replica) const = 0; + virtual ClusterFunctionReadTaskResponsePtr operator()(size_t number_of_current_replica, UInt64 /*antalya_protocol_version*/) const = 0; }; /// This class allows one to launch queries on remote replicas of one shard and get results diff --git a/src/Server/TCPHandler.cpp b/src/Server/TCPHandler.cpp index fbe5abae9ed6f..29fb7c01668eb 100644 --- a/src/Server/TCPHandler.cpp +++ b/src/Server/TCPHandler.cpp @@ -2467,7 +2467,7 @@ ClusterFunctionReadTaskResponsePtr TCPHandler::receiveClusterFunctionReadTaskRes case Protocol::Client::ReadTaskResponse: { auto task = std::make_shared(); - task->deserialize(*in); + task->deserialize(*in, client_antalya_protocol_version); return task; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp index c85036edd08e8..28cb81fe2fd20 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp @@ -15,6 +15,7 @@ #include #include +#include #include #include #include @@ -52,7 +53,6 @@ namespace Setting extern const SettingsInt64 delta_lake_snapshot_start_version; extern const SettingsInt64 delta_lake_snapshot_end_version; extern const SettingsUInt64 lock_object_storage_task_distribution_ms; - extern const SettingsBool allow_experimental_iceberg_read_optimization; } namespace ErrorCodes @@ -644,14 +644,12 @@ class TaskDistributor : public TaskIterator std::vector && ids_of_hosts, bool send_over_whole_archive, uint64_t lock_object_storage_task_distribution_ms, - ContextPtr context_, - bool iceberg_read_optimization_enabled) + ContextPtr context_) : task_distributor( iterator, std::move(ids_of_hosts), send_over_whole_archive, - lock_object_storage_task_distribution_ms, - iceberg_read_optimization_enabled) + lock_object_storage_task_distribution_ms) , context(context_) {} ~TaskDistributor() override = default; bool supportRerunTask() const override { return true; } @@ -660,16 +658,23 @@ class TaskDistributor : public TaskIterator task_distributor.rescheduleTasksFromReplica(number_of_current_replica); } - ClusterFunctionReadTaskResponsePtr operator()(size_t number_of_current_replica) const override + ClusterFunctionReadTaskResponsePtr operator()(size_t number_of_current_replica, UInt64 antalya_protocol_version) const override { fiu_do_on(FailPoints::storage_cluster_read_sleep, { sleepForSeconds(10); }); - auto task = task_distributor.getNextTask(number_of_current_replica); - if (task) - return std::make_shared(std::move(task), context); + const bool worker_supports_task_retry = antalya_protocol_version >= DBMS_ANTALYA_PROTOCOL_VERSION_WITH_READ_TASK_EXTENSIONS; + auto task = task_distributor.getNextTask(number_of_current_replica, worker_supports_task_retry); + if (task.retry_after_us) + { + auto response = std::make_shared(); + response->retry_after_us = task.retry_after_us; + return response; + } + if (task.object) + return std::make_shared(std::move(task.object), context); return std::make_shared(); } @@ -742,8 +747,7 @@ RemoteQueryExecutor::Extension StorageObjectStorageCluster::getTaskIteratorExten std::move(ids_of_hosts), /* send_over_whole_archive */!local_context->getSettingsRef()[Setting::cluster_function_process_archive_on_multiple_nodes], lock_object_storage_task_distribution_ms, - local_context, - /* iceberg_read_optimization_enabled */local_context->getSettingsRef()[Setting::allow_experimental_iceberg_read_optimization]); + local_context); return RemoteQueryExecutor::Extension{ .task_iterator = std::move(callback) }; } diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp index 426f4c97169f7..f76d7a1b5411e 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -2379,10 +2380,7 @@ StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator( for (size_t i = 0; i < max_threads_count; ++i) objects.push_back(pool_scheduler([this]() -> ObjectInfoPtr { - auto task = callback(); - if (!task || task->isEmpty()) - return nullptr; - return task->getObjectInfo(); + return takeNextObject(/*wait_for_retry=*/ false); }, Priority{})); pool.wait(); @@ -2391,10 +2389,7 @@ StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator( { auto object = object_future.get(); if (object) - { - resolveObjectStorageIfNeeded(object); buffer.push_back(object); - } } } @@ -2417,13 +2412,12 @@ static size_t getKnownArchiveSize(const ObjectInfoPtr & object_info) return object_metadata->size_bytes; } -ObjectInfoPtr StorageObjectStorageSource::ReadTaskIterator::next(size_t) +ObjectInfoPtr StorageObjectStorageSource::ReadTaskIterator::takeNextObject(bool wait_for_retry) { - size_t current_index = index.fetch_add(1, std::memory_order_relaxed); - - ObjectInfoPtr object_info; - if (current_index >= buffer.size()) + while (true) { + /// Recheck after each `retry_after_us` wait. `SYSTEM STOP SWARM MODE` during that wait + /// must not be followed by another task request. if (!getContext()->isSwarmModeEnabled()) { LOG_DEBUG(getLogger("StorageObjectStorageSource"), "STOP SWARM MODE called, stop getting new tasks"); @@ -2438,8 +2432,35 @@ ObjectInfoPtr StorageObjectStorageSource::ReadTaskIterator::next(size_t) if (!task || task->isEmpty()) return nullptr; - object_info = task->getObjectInfo(); + if (task->retry_after_us) + { + if (!wait_for_retry) + return nullptr; + + /// The wait stays on the worker so the initiator thread is not blocked. + /// TODO: wait asynchronously instead of sleeping in this thread. + auto wait_time = std::min(100000, *task->retry_after_us); + ProfileEvents::increment(ProfileEvents::ObjectStorageClusterWaitingMicroseconds, wait_time); + sleepForMicroseconds(wait_time); + continue; + } + + auto object_info = task->getObjectInfo(); resolveObjectStorageIfNeeded(object_info); + return object_info; + } +} + +ObjectInfoPtr StorageObjectStorageSource::ReadTaskIterator::next(size_t) +{ + size_t current_index = index.fetch_add(1, std::memory_order_relaxed); + + ObjectInfoPtr object_info; + if (current_index >= buffer.size()) + { + object_info = takeNextObject(/*wait_for_retry=*/ true); + if (!object_info) + return nullptr; } else { diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.h b/src/Storages/ObjectStorage/StorageObjectStorageSource.h index f0b2c082892ee..196beb0e6bbf8 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageSource.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageSource.h @@ -205,6 +205,11 @@ class StorageObjectStorageSource::ReadTaskIterator : public IObjectIterator, pri const std::string & path_in_archive, std::optional read_source_index); + /// Fetch the next object. When `wait_for_retry` is set, a `retry_after_us` response is waited out + /// on this thread and another task is requested. The constructor prefetch passes false so one + /// waiting slot cannot hold back tasks the other slots already received. + ObjectInfoPtr takeNextObject(bool wait_for_retry); + void resolveObjectStorageIfNeeded(const ObjectInfoPtr & object); ClusterFunctionReadTaskCallback callback; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp index b4f26536966cc..f92481368a81f 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp +++ b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp @@ -51,15 +51,13 @@ StorageObjectStorageStableTaskDistributor::StorageObjectStorageStableTaskDistrib std::shared_ptr iterator_, std::vector && ids_of_nodes_, bool send_over_whole_archive_, - uint64_t lock_object_storage_task_distribution_ms_, - bool iceberg_read_optimization_enabled_) + uint64_t lock_object_storage_task_distribution_ms_) : iterator(std::move(iterator_)) , send_over_whole_archive(send_over_whole_archive_) , connection_to_files(ids_of_nodes_.size()) , ids_of_nodes(std::move(ids_of_nodes_)) , lock_object_storage_task_distribution_us(lock_object_storage_task_distribution_ms_ * 1000) , iterator_exhausted(false) - , iceberg_read_optimization_enabled(iceberg_read_optimization_enabled_) { Poco::Timestamp now; size_t nodes = ids_of_nodes.size(); @@ -70,10 +68,21 @@ StorageObjectStorageStableTaskDistributor::StorageObjectStorageStableTaskDistrib } } -ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getNextTask(size_t number_of_current_replica) +DistributedReadTask StorageObjectStorageStableTaskDistributor::getNextTask(size_t number_of_current_replica, bool worker_supports_task_retry) { LOG_TRACE(log, "Received request from replica {} looking for a file", number_of_current_replica); + /// One peer that cannot be asked to wait makes the lock useless: that peer would take another + /// node's file immediately. Run the rest of the query as if the setting were 0. + if (!worker_supports_task_retry && !task_retry_disabled.exchange(true, std::memory_order_relaxed)) + { + LOG_INFO( + log, + "Replica {} does not support task retry, ignoring `lock_object_storage_task_distribution_ms` for this query", + number_of_current_replica); + } + const bool allow_task_retry = worker_supports_task_retry && !task_retry_disabled.load(std::memory_order_relaxed); + saveLastNodeActivity(number_of_current_replica); { @@ -94,7 +103,12 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getNextTask(size_t numb file = getMatchingFileFromIterator(number_of_current_replica); // 3. Process unprocessed files if iterator is exhausted if (!file) - file = getAnyUnprocessedFile(number_of_current_replica); + { + auto pending = getAnyUnprocessedFile(number_of_current_replica, allow_task_retry); + if (pending.retry_after_us) + return pending; + file = std::move(pending.object); + } if (file) { @@ -111,7 +125,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getNextTask(size_t numb processed_file_list_ptr->second.push_back(file); } - return file; + return DistributedReadTask{.object = std::move(file), .retry_after_us = std::nullopt}; } size_t StorageObjectStorageStableTaskDistributor::getReplicaForFile(const String & file_path) @@ -225,17 +239,6 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter file_identifier = getSchedulingIdentifier(object_info, send_over_whole_archive); } - if (iceberg_read_optimization_enabled) - { - auto file_meta_info = object_info->relative_path_with_metadata.getFileMetaInfo(); - if (file_meta_info.has_value()) - { - auto file_path = send_over_whole_archive ? object_info->getPathOrPathToArchiveIfArchive() : object_info->getPath(); - object_info->relative_path_with_metadata.command.setFilePath(file_path); - object_info->relative_path_with_metadata.command.setFileMetaInfo(file_meta_info.value()); - } - } - size_t file_replica_idx; { @@ -272,7 +275,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getMatchingFileFromIter return {}; } -ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(size_t number_of_current_replica) +DistributedReadTask StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(size_t number_of_current_replica, bool allow_task_retry) { /// Limit time of node activity to keep task in queue Poco::Timestamp activity_limit; @@ -291,6 +294,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(s auto number_of_matched_replica = it->second.second; auto last_activity = last_node_activity.find(number_of_matched_replica); if (lock_object_storage_task_distribution_us <= 0 // file deferring is turned off + || !allow_task_retry // this request must not wait: peer is old, or the query already saw one || it->second.second == number_of_current_replica // file is matching with current replica || last_activity == last_node_activity.end() // msut never be happen, last_activity is filled for each replica on start || activity_limit > last_activity->second) // matched replica did not ask for a new files for a while @@ -308,7 +312,7 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(s ); ProfileEvents::increment(ProfileEvents::ObjectStorageClusterSentToNonMatchedReplica); - return next_file; + return DistributedReadTask{.object = std::move(next_file), .retry_after_us = std::nullopt}; } oldest_activity = std::min(oldest_activity, last_activity->second); @@ -322,11 +326,12 @@ ObjectInfoPtr StorageObjectStorageStableTaskDistributor::getAnyUnprocessedFile(s oldest_activity - activity_limit ); - /// All unprocessed files owned by alive replicas with recenlty activity - /// Need to retry after (oldest_activity - activity_limit) microseconds - RelativePathWithMetadata::CommandInTaskResponse response; - response.setRetryAfterUs(oldest_activity - activity_limit); - return std::make_shared(response.toString()); + /// All unprocessed files owned by alive replicas with recently activity. + /// The worker waits `(oldest_activity - activity_limit)` microseconds, then asks again. + auto retry_after_us = oldest_activity - activity_limit; + if (retry_after_us < 0) + retry_after_us = 0; + return DistributedReadTask{.object = nullptr, .retry_after_us = static_cast(retry_after_us)}; } return {}; diff --git a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h index 0cd10ac7188ed..9955f81f78da9 100644 --- a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h +++ b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h @@ -14,10 +14,20 @@ #include #include #include +#include +#include namespace DB { +struct DistributedReadTask +{ + ObjectInfoPtr object; + /// Microseconds the worker should wait before asking again. + /// Set only when `object` is empty and the worker understands Antalya protocol version 3. + std::optional retry_after_us; +}; + class StorageObjectStorageStableTaskDistributor { public: @@ -25,10 +35,12 @@ class StorageObjectStorageStableTaskDistributor std::shared_ptr iterator_, std::vector && ids_of_nodes_, bool send_over_whole_archive_, - uint64_t lock_object_storage_task_distribution_ms_, - bool iceberg_read_optimization_enabled_); + uint64_t lock_object_storage_task_distribution_ms_); - ObjectInfoPtr getNextTask(size_t number_of_current_replica); + /// `worker_supports_task_retry` is false for a peer that did not negotiate Antalya protocol version 3. + /// The first such peer disables `lock_object_storage_task_distribution_ms` for the rest of the query: + /// every replica is then given a file immediately, with no `retry_after_us`. + DistributedReadTask getNextTask(size_t number_of_current_replica, bool worker_supports_task_retry = true); /// Insert objects back to unprocessed files void rescheduleTasksFromReplica(size_t number_of_current_replica); @@ -37,7 +49,7 @@ class StorageObjectStorageStableTaskDistributor size_t getReplicaForFile(const String & file_path); ObjectInfoPtr getPreQueuedFile(size_t number_of_current_replica); ObjectInfoPtr getMatchingFileFromIterator(size_t number_of_current_replica); - ObjectInfoPtr getAnyUnprocessedFile(size_t number_of_current_replica); + DistributedReadTask getAnyUnprocessedFile(size_t number_of_current_replica, bool allow_task_retry); void saveLastNodeActivity(size_t number_of_current_replica); @@ -55,7 +67,8 @@ class StorageObjectStorageStableTaskDistributor std::mutex mutex; bool iterator_exhausted = false; - bool iceberg_read_optimization_enabled = false; + /// Set when any worker in this query cannot carry `retry_after_us`. Further tasks skip the lock. + std::atomic task_retry_disabled{false}; LoggerPtr log = getLogger("StorageClusterTaskDistributor"); }; diff --git a/src/Storages/ObjectStorage/tests/gtest_rendezvous_hashing.cpp b/src/Storages/ObjectStorage/tests/gtest_rendezvous_hashing.cpp index 50ce5453c4c97..a9c8bddf92d2c 100644 --- a/src/Storages/ObjectStorage/tests/gtest_rendezvous_hashing.cpp +++ b/src/Storages/ObjectStorage/tests/gtest_rendezvous_hashing.cpp @@ -62,7 +62,7 @@ namespace { for (size_t i = 0; i < n; ++i) { - ObjectInfoPtr object = distributor.getNextTask(replica_id); + ObjectInfoPtr object = distributor.getNextTask(replica_id).object; if (!object) return false; auto path = object->getPath(); @@ -79,7 +79,7 @@ namespace { for (size_t i = 0; i < n; ++i) { - ObjectInfoPtr object = distributor.getNextTask(replica_id); + ObjectInfoPtr object = distributor.getNextTask(replica_id).object; if (!object) return false; read_source_indices.emplace_back(object->relative_path_with_metadata.read_source_index); @@ -127,7 +127,7 @@ TEST(RendezvousHashing, SingleNode) { auto iterator = makeIterator(); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 0, 10)); ASSERT_TRUE(checkHead(paths, {6})); @@ -136,7 +136,7 @@ TEST(RendezvousHashing, SingleNode) { auto iterator = makeIterator(); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 1, 10)); ASSERT_TRUE(checkHead(paths, {0, 2, 4})); @@ -145,7 +145,7 @@ TEST(RendezvousHashing, SingleNode) { auto iterator = makeIterator(); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 2, 10)); ASSERT_TRUE(checkHead(paths, {1, 5, 7, 8})); @@ -154,7 +154,7 @@ TEST(RendezvousHashing, SingleNode) { auto iterator = makeIterator(); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 3, 10)); ASSERT_TRUE(checkHead(paths, {3, 9})); @@ -165,7 +165,7 @@ TEST(RendezvousHashing, MultipleNodes) { auto iterator = makeIterator(); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); { std::vector paths; @@ -197,7 +197,7 @@ TEST(RendezvousHashing, SingleNodeReducedCluster) { auto iterator = makeIterator(); std::vector replicas = {"replica2", "replica1"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 0, 10)); ASSERT_TRUE(checkHead(paths, {1, 5, 6, 7, 8, 9})); @@ -206,7 +206,7 @@ TEST(RendezvousHashing, SingleNodeReducedCluster) { auto iterator = makeIterator(); std::vector replicas = {"replica2", "replica1"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths; ASSERT_TRUE(extractNForReplica(distributor, paths, 1, 10)); ASSERT_TRUE(checkHead(paths, {0, 2, 3, 4})); @@ -217,7 +217,7 @@ TEST(RendezvousHashing, MultipleNodesReducedCluster) { auto iterator = makeIterator(); std::vector replicas = {"replica2", "replica1"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); { std::vector paths; @@ -236,7 +236,7 @@ TEST(RendezvousHashing, MultipleNodesReducedClusterOneByOne) { auto iterator = makeIterator(); std::vector replicas = {"replica2", "replica1"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector paths0; std::vector paths1; @@ -267,7 +267,7 @@ TEST(RendezvousHashing, DoesNotDeduplicateSamePathFromDifferentReadSources) auto iterator = std::make_shared(paths); std::vector replicas = {"replica0", "replica1", "replica2", "replica3"}; - StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0, false); + StorageObjectStorageStableTaskDistributor distributor(iterator, std::move(replicas), false, 0); std::vector> read_source_indices; ASSERT_TRUE(extractNForReplica(distributor, read_source_indices, 0, 2)); @@ -311,7 +311,7 @@ TEST(ClusterFunctionReadTaskResponse, PreservesReadSourceIndex) ReadBufferFromString in(serialized); ClusterFunctionReadTaskResponse deserialized; - deserialized.deserialize(in); + deserialized.deserialize(in, DBMS_ANTALYA_PROTOCOL_VERSION); auto object_info = deserialized.getObjectInfo(); ASSERT_TRUE(object_info->relative_path_with_metadata.read_source_index.has_value()); diff --git a/src/Storages/StorageFileCluster.cpp b/src/Storages/StorageFileCluster.cpp index 214c1bcaef2a6..b9999d217eb27 100644 --- a/src/Storages/StorageFileCluster.cpp +++ b/src/Storages/StorageFileCluster.cpp @@ -150,7 +150,7 @@ class FileTaskIterator : public TaskIterator ~FileTaskIterator() override = default; - ClusterFunctionReadTaskResponsePtr operator()(size_t /* number_of_current_replica */) const override + ClusterFunctionReadTaskResponsePtr operator()(size_t /* number_of_current_replica */, UInt64 /* antalya_protocol_version */) const override { auto file = iterator.next(); if (file.empty()) diff --git a/src/Storages/StorageURLCluster.cpp b/src/Storages/StorageURLCluster.cpp index 2a24377786141..1ba03ce762484 100644 --- a/src/Storages/StorageURLCluster.cpp +++ b/src/Storages/StorageURLCluster.cpp @@ -155,7 +155,7 @@ class UrlTaskIterator : public TaskIterator ~UrlTaskIterator() override = default; - ClusterFunctionReadTaskResponsePtr operator()(size_t /* number_of_current_replica */) const override + ClusterFunctionReadTaskResponsePtr operator()(size_t /* number_of_current_replica */, UInt64 /* antalya_protocol_version */) const override { auto url = iterator.next(); if (url.empty()) diff --git a/tests/integration/test_iceberg_mixed_antalya_upstream/configs/config.d/cluster.xml b/tests/integration/test_iceberg_mixed_antalya_upstream/configs/config.d/cluster.xml new file mode 100644 index 0000000000000..f01efc87a6399 --- /dev/null +++ b/tests/integration/test_iceberg_mixed_antalya_upstream/configs/config.d/cluster.xml @@ -0,0 +1,32 @@ + + + + + + antalya + 9000 + + + + + + + upstream + 9000 + + + + + + + antalya + 9000 + + + upstream + 9000 + + + + + diff --git a/tests/integration/test_iceberg_mixed_antalya_upstream/test.py b/tests/integration/test_iceberg_mixed_antalya_upstream/test.py new file mode 100644 index 0000000000000..bf0531478c4a7 --- /dev/null +++ b/tests/integration/test_iceberg_mixed_antalya_upstream/test.py @@ -0,0 +1,70 @@ +import pytest + +from helpers.cluster import ClickHouseCluster +from helpers.config_cluster import minio_access_key, minio_secret_key + +# Four-part tag. `latest` and `26.6` move when a new build is published. +UPSTREAM_IMAGE = "clickhouse/clickhouse-server" +UPSTREAM_TAG = "26.6.8.7" + +cluster = ClickHouseCluster(__file__) +antalya = cluster.add_instance( + "antalya", + main_configs=["configs/config.d/cluster.xml"], + with_minio=True, +) +upstream = cluster.add_instance( + "upstream", + main_configs=["configs/config.d/cluster.xml"], + image=UPSTREAM_IMAGE, + tag=UPSTREAM_TAG, + with_installed_binary=True, + with_remote_database_disk=False, +) + +TABLE_URL = "http://minio1:9001/root/mixed_antalya_upstream/" + + +@pytest.fixture(scope="module") +def started_cluster(): + try: + cluster.start() + antalya.query( + f""" + CREATE TABLE mixed_proto (id Int32, tag String) + ENGINE = IcebergS3('{TABLE_URL}', '{minio_access_key}', '{minio_secret_key}') + """ + ) + # Two files, each with a constant `tag`, so the Antalya read optimization has + # something to send. The other side must still return these values. + antalya.query( + "INSERT INTO mixed_proto VALUES (1, 'alpha')", + settings={"allow_insert_into_iceberg": 1}, + ) + antalya.query( + "INSERT INTO mixed_proto VALUES (2, 'beta')", + settings={"allow_insert_into_iceberg": 1}, + ) + yield cluster + finally: + cluster.shutdown() + + +@pytest.mark.parametrize( + ("initiator_name", "cluster_name"), + [ + pytest.param("antalya", "upstream_only", id="antalya_initiator_upstream_worker"), + pytest.param("upstream", "antalya_only", id="upstream_initiator_antalya_worker"), + # Both workers, default `lock_object_storage_task_distribution_ms`. The Antalya initiator + # must not put a retry command into the object path: the upstream worker would open that text as a key. + pytest.param("antalya", "mixed", id="antalya_initiator_mixed_workers"), + ], +) +def test_mixed_iceberg_cluster_read(started_cluster, initiator_name, cluster_name): + expected = antalya.query(f"SELECT id, tag FROM icebergS3('{TABLE_URL}', '{minio_access_key}', '{minio_secret_key}') ORDER BY id") + initiator = started_cluster.instances[initiator_name] + got = initiator.query( + f"SELECT id, tag FROM icebergS3Cluster('{cluster_name}', '{TABLE_URL}', '{minio_access_key}', '{minio_secret_key}') ORDER BY id", + ) + assert got == expected + assert got == "1\talpha\n2\tbeta\n" From e1e35b89855f0dd9bd540782d6e90fceaea0cacd Mon Sep 17 00:00:00 2001 From: blau-ai Date: Sun, 18 Oct 2026 11:23:10 +0000 Subject: [PATCH 3/3] Add __init__.py to test_iceberg_mixed_antalya_upstream integration test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes a pytest 'import file mismatch' collection error in the 'Integration tests (..., db disk, old analyzer, 4/8)' shard: without an __init__.py the new test.py is imported under the bare top-level module name 'test', which collides with another test.py (e.g. test_export_merge_tree_part_to_iceberg/test.py) collected in the same pytest process. Adding an empty __init__.py — the convention followed by ~all other tests/integration/* dirs — makes it import as the package-qualified module 'test_iceberg_mixed_antalya_upstream.test', which cannot collide regardless of collection order. Co-Authored-By: Claude Opus 4.8 --- tests/integration/test_iceberg_mixed_antalya_upstream/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 tests/integration/test_iceberg_mixed_antalya_upstream/__init__.py diff --git a/tests/integration/test_iceberg_mixed_antalya_upstream/__init__.py b/tests/integration/test_iceberg_mixed_antalya_upstream/__init__.py new file mode 100644 index 0000000000000..e69de29bb2d1d