Antalya 26.8, frontports: Send Iceberg file statistics and task retry delay in the Antalya protocol trailer

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,14 +30,22 @@ 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.
 - Never bump `DBMS_TCP_PROTOCOL_VERSION`, and never take a slot in
   `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}

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.

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 |

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;
 

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;
 

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;
 

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
 {

src/Interpreters/ClusterFunctionReadTask.cpp

@@ -4,6 +4,7 @@
 #include <AggregateFunctions/AggregateFunctionGroupBitmapData.h>
 #include <Core/Settings.h>
 #include <Core/ProtocolDefines.h>
+#include <Core/AntalyaProtocol.h>
 #include <Common/Exception.h>
 #include <Common/logger_useful.h>
 #include <IO/WriteHelpers.h>
@@ -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>(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;
+        }
+    }
 }
 
 }

src/Interpreters/ClusterFunctionReadTask.h

@@ -27,20 +27,25 @@ struct ClusterFunctionReadTaskResponse
     std::optional<Iceberg::IcebergObjectSerializableInfo> iceberg_info;
     /// File's columns info
     std::optional<DataFileMetaInfoPtr> 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<UInt64> 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<ClusterFunctionReadTaskResponse>;

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<ParquetFileBucketInfo *>(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>();
+    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<Int64>(), 1);
+    EXPECT_EQ(column.hyperrectangle->right.safeGet<Int64>(), 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());
+}