Antalya-26.8: Iceberg: support external paths in tables

docs/en/antalya/protocol.md

@@ -14,15 +14,21 @@ 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:1)"
-server -> client   "ClickHouse (antalya:1)"
+client -> server   "ClickHouse client (antalya:2)"
+server -> client   "ClickHouse (antalya:2)"
 ```
 
 Each side parses the peer's suffix, caps the value with `min(own, peer)` and keeps the result. `0`
 means the peer is not an Antalya build. Negotiation is per hop and not transitive: initiator to
 worker and worker to worker negotiate independently.
 
-Version 1 is the advertisement itself. Nothing is gated on it yet.
+Version 1 is the advertisement itself.
+
+Version 2 lets the `*Cluster` read-task payload (the `ReadTaskResponse` body) carry the Iceberg
+delete-file paths as absolute URIs, as the manifest spells them, so that a worker can read files
+outside the table location, including files on another object storage. For a worker below version 2
+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.
 
 ## Adding an Antalya-only wire change {#adding-a-wire-change}
 

docs/reference/interfaces/specs/NativeProtocol.mdx

@@ -146,7 +146,7 @@ When a feature is active, its fields **must** be present on the wire. The protoc
 | JWT_IN_INTERSERVER              | 54476   | ClientInfo             | Adds a JWT-presence UInt8 + optional `String jwt` at the tail of ClientInfo. External clients (no JWT) send byte `0x00`. (Spelled `DBMS_MIN_REVISON_WITH_JWT_IN_INTERSERVER` in C++ — note the typo in the constant name.) |
 | QUERY_PLAN_SERIALIZATION        | 54477   | ServerHello, QueryPlan packet | ServerHello appends `VarUInt query_plan_serialization_version` after server settings. Also introduces `ClientPacket::QueryPlan` (code `13`) for inter-server delivery of pre-built query plans — external clients never send. |
 | PARALLEL_BLOCK_MARSHALLING      | 54478   | Block (Column)         | Server may wrap columns in `ColumnBLOB` (compressed inline) for parallel processing. Gated on the query having compression enabled AND `rows > 1`; otherwise the regular column wire format applies. Clients that never enable compression on outgoing Query packets see no wire change. |
-| VERSIONED_CLUSTER_FUNCTION_PROTOCOL | 54479 | ServerHello           | Adds `VarUInt cluster_function_protocol_version` at the tail of ServerHello. Used for `*Cluster` table functions (`s3Cluster`, etc.). Current value: `8` (`DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION`); version `7` is reserved for a private-repository feature (Iceberg compaction), and `8` adds an optional `read_source_index` to the inter-server cluster read-task payload (the `ReadTaskResponse` body, which stays unspecified here — see below). External clients decode and ignore. |
+| VERSIONED_CLUSTER_FUNCTION_PROTOCOL | 54479 | ServerHello           | Adds `VarUInt cluster_function_protocol_version` at the tail of ServerHello. Used for `*Cluster` table functions (`s3Cluster`, etc.). Current value: `9` (`DBMS_CLUSTER_PROCESSING_PROTOCOL_VERSION`); version `7` is reserved for a private-repository feature (Iceberg compaction), `8` adds an optional `read_source_index` to the inter-server cluster read-task payload (the `ReadTaskResponse` body, which stays unspecified here — see below), and `9` adds Iceberg identity partition columns. External clients decode and ignore. |
 | OUT_OF_ORDER_BUCKETS_IN_AGGREGATION | 54480 | BlockInfo              | Adds field 3 (`out_of_order_buckets: Vec<Int32>`) to BlockInfo's field-tagged stream. Decoded as `[VarUInt count][Int32]*count`. External clients don't emit this themselves; the decoder reads any non-empty list the server sends. |
 | COMPRESSED_LOGS_PROFILE_EVENTS_COLUMNS | 54481 | Log, ProfileEvents, TableColumns | Server may wrap [`Log`](#log), [`ProfileEvents`](#profileevents), and [`TableColumns`](#tablecolumns) packet bodies in the [compression frame](/reference/interfaces/specs/NativeFormat#compression-frame). At this version all three bodies travel through the same optionally-compressed output path, which becomes a real compression frame only when the query has `compression = true`. Clients that never enable compression on outgoing Query packets see no wire change. |
 | REPLICATED_SERIALIZATION        | 54482   | Block (Column)         | Server may emit columns with kind_stack `0x04 = REPLICATED` — a dictionary-style compact form for repeated values — see [kind_stack and sparse encoding](/reference/interfaces/specs/NativeFormat#kind-stack-and-sparse-encoding). Below this version the writer expanded such columns before sending. Decoded via index lookup (`elements[indexes[i]]` per row); leaf types plus `Nullable`/`Array`/`Tuple`/`Map`/`Nested`/`LowCardinality` inners supported. |
@@ -463,7 +463,7 @@ Server → Client. The reply to ClientHello on successful authentication.
 | 11 | nonce            | UInt64  | inter-server | INTERSERVER_SECRET_V2 (v54462) | 8-byte LE random nonce. The server's inter-server query-signing scheme uses it. External clients MUST decode it (to keep the stream aligned) and SHOULD ignore the value. |
 | 12 | server_settings  | Setting[] | universal | SERVER_SETTINGS (v54474)        | Server's non-default settings broadcast. Format: zero or more `(String key, VarUInt flags, String value)` triples, terminated by an empty key. Same as the [Query packet's settings list](#setting). |
 | 13 | query_plan_serialization_version | VarUInt | universal | QUERY_PLAN_SERIALIZATION (v54477) | Server's supported query-plan serialization version. External clients decode and ignore. |
-| 14 | cluster_function_protocol_version | VarUInt | universal | VERSIONED_CLUSTER_FUNCTION_PROTOCOL (v54479) | Server's `*Cluster` table-function protocol version. Current: `8`. The value gates additive fields in the inter-server cluster read-task payload (the otherwise-unspecified `ReadTaskResponse` body); version `7` is reserved for a private-repository feature (Iceberg compaction), and `8` adds an optional `read_source_index`. External clients do not participate in cluster reads — they decode and ignore this field. |
+| 14 | cluster_function_protocol_version | VarUInt | universal | VERSIONED_CLUSTER_FUNCTION_PROTOCOL (v54479) | Server's `*Cluster` table-function protocol version. Current: `9`. The value gates additive fields in the inter-server cluster read-task payload (the otherwise-unspecified `ReadTaskResponse` body); version `7` is reserved for a private-repository feature (Iceberg compaction), `8` adds an optional `read_source_index`, and `9` adds Iceberg identity partition columns. External clients do not participate in cluster reads — they decode and ignore this field. |
 
 **Rule** — an element of `password_complexity_rules`:
 

src/Access/ContextAccess.cpp

@@ -18,10 +18,13 @@
 #include <Common/re2.h>
 #include <Core/Settings.h>
 #include <IO/WriteHelpers.h>
+#include <Poco/Exception.h>
 #include <Poco/Logger.h>
+#include <Poco/URI.h>
 #include <Common/logger_useful.h>
 #include <boost/algorithm/string/join.hpp>
 #include <boost/range/algorithm/set_algorithm.hpp>
+#include <filesystem>
 #include <unordered_set>
 
 
@@ -40,6 +43,23 @@ namespace ErrorCodes
 
 namespace
 {
+    String normalizeAccessURI(std::string_view uri)
+    {
+        if (uri.empty())
+            return {};
+
+        try
+        {
+            Poco::URI parsed{String{uri}};
+            parsed.normalize();
+            return parsed.toString();
+        }
+        catch (const Poco::Exception &)
+        {
+            return {};
+        }
+    }
+
     AccessRights mixAccessRightsFromUserAndRoles(const User & user, const EnabledRolesInfo & roles_info)
     {
         AccessRights res = user.access;
@@ -1100,13 +1120,16 @@ bool ContextAccess::isGrantedWithFilter(const ContextPtr & context, const Access
     if (isGranted(context, flags, parameter))
         return true;
 
-    if (!to_check_by_filter.empty())
+    const String normalized_filter_target = parameter == AccessTypeObjects::toStringSource(AccessTypeObjects::Source::FILE)
+        ? std::filesystem::path{String{to_check_by_filter}}.lexically_normal().string()
+        : normalizeAccessURI(to_check_by_filter);
+    if (!normalized_filter_target.empty())
     {
         auto access_rights = getAccessRights();
         auto filters = access_rights->getFilters(parameter);
         for (const auto & filter : filters)
         {
-            if (re2::RE2::FullMatch(to_check_by_filter, filter.path) && filter.access_flags.contains(flags))
+            if (re2::RE2::FullMatch(normalized_filter_target, filter.path) && filter.access_flags.contains(flags))
                 return true;
         }
     }

src/Access/ContextAccess.h

@@ -133,6 +133,7 @@ class ContextAccess : public std::enable_shared_from_this<ContextAccess>
     void checkGranteesAreAllowed(const std::vector<UUID> & grantee_ids) const;
 
     /// Checks access of grants with parameter where a filter can be applied.
+    /// The URI checked against the filter is normalized first; an invalid URI cannot match a filter.
     /// For example, for `GRANT READ ON S3('s3://foo.*') TO user` calling `checkAccess(READ, "S3")` will throw an error
     /// because we are checking for `READ` permissions for all possible URLs.
     ///
@@ -333,6 +334,7 @@ class ContextAccessWrapper : public std::enable_shared_from_this<ContextAccessWr
     ALWAYS_INLINE void checkGranteesAreAllowed(const std::vector<UUID> & grantee_ids) const { access->checkGranteesAreAllowed(grantee_ids); }
 
     /// Checks access of grants with parameter where a filter can be applied.
+    /// The URI checked against the filter is normalized first; an invalid URI cannot match a filter.
     ALWAYS_INLINE void checkAccessWithFilter(const AccessFlags & flags, std::string_view parameter, std::string_view filter) const { access->checkAccessWithFilter(context, flags, parameter, filter); }
 
     /// Checks access of grants with parameter where a filter can be applied, without throwing.

src/Backups/BackupFactory.cpp

@@ -3,8 +3,6 @@
 #include <Interpreters/Context.h>
 #include <Common/Exception.h>
 
-#include <Poco/URI.h>
-
 #include <fmt/format.h>
 
 #include <iterator>
@@ -25,24 +23,6 @@ namespace
     {
         fmt::format_to(std::back_inserter(identity), ":{}:{}", component.size(), component);
     }
-
-    /// Must normalize identically to `ITableFunction::getFunctionURINormalized`, or the same regex
-    /// grant matches a table function and not a backup. An empty result requires a whole-source grant.
-    String normalizeAccessURI(const String & uri)
-    {
-        if (uri.empty())
-            return uri;
-        try
-        {
-            Poco::URI parsed(uri);
-            parsed.normalize();
-            return parsed.toString();
-        }
-        catch (const Poco::Exception &)
-        {
-            return "";
-        }
-    }
 }
 
 
@@ -82,7 +62,7 @@ void BackupFactory::checkSourceAccess(const BackupInfo & backup_info, ContextPtr
 
     if (auto target = it->second.source_access(backup_info, context, open_mode))
         context->getAccess()->checkAccessWithFilter(
-            target->flags, AccessTypeObjects::toStringSource(target->source), normalizeAccessURI(target->uri));
+            target->flags, AccessTypeObjects::toStringSource(target->source), target->uri);
 }
 
 BackupMutablePtr BackupFactory::createBackup(const CreateParams & params) const

src/Client/Connection.cpp

@@ -1284,7 +1284,7 @@ void Connection::sendData(const Block & block, const String & name, bool scalar)
 void Connection::sendClusterFunctionReadTaskResponse(const ClusterFunctionReadTaskResponse & response)
 {
     writeVarUInt(Protocol::Client::ReadTaskResponse, *out);
-    response.serialize(*out, worker_cluster_function_protocol_version);
+    response.serialize(*out, worker_cluster_function_protocol_version, server_antalya_protocol_version);
     out->finishChunk();
     out->next();
 }

src/Common/maskSensitiveQueryParameters.cpp

@@ -2,6 +2,7 @@
 
 #include <Common/StringUtils.h>
 #include <base/hex.h>
+#include <Common/maskURIPassword.h>
 
 #include <array>
 #include <string>
@@ -143,4 +144,11 @@ std::string maskSensitiveQueryParametersInURI(const std::string & uri)
     return result;
 }
 
+std::string maskCredentialsInURI(const std::string & uri)
+{
+    std::string result = maskSensitiveQueryParametersInURI(uri);
+    maskURIPassword(&result);
+    return result;
+}
+
 }

src/Common/maskSensitiveQueryParameters.h

@@ -2,7 +2,6 @@
 
 #include <string>
 
-
 namespace DB
 {
 
@@ -21,4 +20,7 @@ namespace DB
 /// If the URI has no query string, the original string is returned unchanged.
 std::string maskSensitiveQueryParametersInURI(const std::string & uri);
 
+/// Also accepts strings containing a URI; the first '?' starts the query string and any suffix belongs to it.
+std::string maskCredentialsInURI(const std::string & uri);
+
 }

src/Common/tests/gtest_mask_sensitive_query_parameters.cpp

@@ -2,6 +2,7 @@
 
 #include <gtest/gtest.h>
 
+using DB::maskCredentialsInURI;
 using DB::maskSensitiveQueryParametersInURI;
 
 TEST(MaskSensitiveQueryParameters, NoQueryString)
@@ -87,3 +88,20 @@ TEST(MaskSensitiveQueryParameters, EmptyQueryString)
 {
     EXPECT_EQ(maskSensitiveQueryParametersInURI("/?"), "/?");
 }
+
+TEST(MaskCredentialsInURI, ObjectPathsUnchanged)
+{
+    EXPECT_EQ(maskCredentialsInURI(""), "");
+    EXPECT_EQ(maskCredentialsInURI("s3://bucket/t/data/x.parquet"), "s3://bucket/t/data/x.parquet");
+    EXPECT_EQ(maskCredentialsInURI("/table/data/x.parquet"), "/table/data/x.parquet");
+}
+
+TEST(MaskCredentialsInURI, PasswordAndQueryParameters)
+{
+    for (const auto & prefix : {"", "7:"})
+    {
+        EXPECT_EQ(
+            maskCredentialsInURI(std::string(prefix) + "s3://alice:hunter2@minio:9000/bucket/f.parquet?X-Amz-Signature=abc"),
+            std::string(prefix) + "s3://alice:[HIDDEN]@minio:9000/bucket/f.parquet?X-Amz-Signature=[HIDDEN]");
+    }
+}

src/Core/AntalyaProtocol.h

@@ -8,7 +8,9 @@ namespace DB
 {
 
 /// Bump for every Antalya-only wire protocol change. See `docs/en/antalya/protocol.md`.
-static constexpr auto DBMS_ANTALYA_PROTOCOL_VERSION = 1;
+/// 2: the `*Cluster` read-task payload carries Iceberg delete-file paths as absolute URIs.
+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;
 
 namespace AntalyaProtocol
 {

src/Core/Settings.cpp

@@ -691,6 +691,10 @@ Use multiple threads for azure multipart upload.
 )", 0) \
     DECLARE(Bool, s3_throw_on_zero_files_match, false, R"(
 Throw an error, when ListObjects request cannot match any files
+)", 0) \
+    DECLARE(Bool, object_storage_propagate_credentials_to_other_storages, false, R"(
+Reuse the base storage's `S3` credentials when accessing Iceberg table files located in a different endpoint.
+Ignored if `s3_allow_server_credentials_in_user_queries` is disabled.
 )", 0) \
     DECLARE(Bool, hdfs_throw_on_zero_files_match, false, R"(
 Throw an error if matched zero files according to glob expansion rules.

src/Core/SettingsChangesHistory.cpp

@@ -219,6 +219,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya",
         {
             {"data_lake_delete_data_on_drop", false, false, "New setting that unifies dropping of data lake data; the released `iceberg_delete_data_on_drop` is kept as an alias for it."},
+            {"object_storage_propagate_credentials_to_other_storages", false, false, "New setting, allowing to reuse the base storage's `S3` credentials for a secondary object storage, even if the target endpoint differs from the base one. Disabled by default. Ignored while `s3_allow_server_credentials_in_user_queries` is disabled."},
             {"datalake_ignore_unsupported_table_properties", false, false, "New setting: allow `CREATE TABLE` and `SHOW CREATE TABLE` in a `DataLakeCatalog` to omit unsupported table properties."},
             {"object_storage_cluster_fallback_to_local_if_empty", false, false, "New setting"},
             {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."},

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -864,7 +864,8 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
             static_credentials->addCredentialsToEngineArgs(args);
             static_credentials_applied = true;
         }
-        else if (!lightweight && table_metadata.requiresCredentials() && std::find(vended_credentials_catalogs.begin(), vended_credentials_catalogs.end(), catalog->getCatalogType()) == vended_credentials_catalogs.end())
+        else if (!lightweight && table_metadata.requiresCredentials() && std::find(vended_credentials_catalogs.begin(), vended_credentials_catalogs.end(), catalog->getCatalogType()) == vended_credentials_catalogs.end()
+            && table_metadata.getStorageType() != DatabaseDataLakeStorageType::Local)
         {
             throw Exception(
                ErrorCodes::BAD_ARGUMENTS,

src/IO/S3/URI.cpp

@@ -18,10 +18,12 @@ namespace DB
 
 struct URIConverter
 {
-    static void modifyURI(Poco::URI & uri, NameToNameMap mapper)
+    static void modifyURI(Poco::URI & uri, NameToNameMap mapper, bool enable_url_encoding = true)
     {
         Macros macros({{"bucket", uri.getHost()}});
-        uri = macros.expand(mapper[uri.getScheme()]).empty() ? uri : Poco::URI(macros.expand(mapper[uri.getScheme()]) + uri.getPathAndQuery());
+        uri = macros.expand(mapper[uri.getScheme()]).empty()
+            ? uri
+            : Poco::URI(macros.expand(mapper[uri.getScheme()]) + uri.getPathAndQuery(), enable_url_encoding);
     }
 };
 
@@ -33,7 +35,7 @@ namespace ErrorCodes
 namespace S3
 {
 
-URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_presigned_query_parameters, S3UriStyle uri_style)
+URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_presigned_query_parameters, S3UriStyle uri_style, bool enable_url_encoding)
 {
     /// Case when AWS Private Link Interface is being used
     /// E.g. (bucket.vpce-07a1cd78f1bd55c5f-j3a3vg6w.s3.us-east-1.vpce.amazonaws.com/bucket-name/key)
@@ -45,9 +47,9 @@ URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_pre
     else
         uri_str = uri_;
 
-    uri = Poco::URI(uri_str);
+    uri = Poco::URI(uri_str, enable_url_encoding);
     /// Keep a copy of how Poco parsed the original string before any mapping
-    Poco::URI original_uri(uri_str);
+    Poco::URI original_uri(uri_str, enable_url_encoding);
     bool looks_like_presigned = false;
     for (const auto & [qk, qv] : original_uri.getQueryParameters())
     {
@@ -93,7 +95,7 @@ URI::URI(const std::string & uri_, bool allow_archive_path_syntax, bool keep_pre
         }
 
         if (!mapper.empty())
-            URIConverter::modifyURI(uri, mapper);
+            URIConverter::modifyURI(uri, mapper, enable_url_encoding);
     }
 
     storage_name = "S3";

src/IO/S3/URI.h

@@ -41,7 +41,8 @@ struct URI
         const std::string & uri_,
         bool allow_archive_path_syntax = false,
         bool keep_presigned_query_parameters = true,
-        S3UriStyle uri_style = S3UriStyle::AUTO);
+        S3UriStyle uri_style = S3UriStyle::AUTO,
+        bool enable_url_encoding = true);
     void addRegionToURI(const std::string & region);
 
     static void validateBucket(const std::string & bucket, const Poco::URI & uri);