Merge branch 'antalya-25.3' into backports/25.3/82114_fix_glue_nested_types

Enmk · web-flow · commit 3f43e88b18ef · 2025-07-07T10:35:48.000+02:00
diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp
@@ -6141,6 +6141,9 @@ Cache the list of objects returned by list objects calls in object storage
 )", EXPERIMENTAL) \
     DECLARE(Bool, object_storage_remote_initiator, false, R"(
 Execute request to object storage as remote on one of object_storage_cluster nodes.
+)", EXPERIMENTAL) \
+    DECLARE(UInt64, lock_object_storage_task_distribution_ms, 0, R"(
+In object storage distribution queries do not distibute tasks on non-prefetched nodes until prefetched node is active.
 )", EXPERIMENTAL) \
     \
 
diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp
@@ -66,6 +66,11 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// controls new feature and it's 'true' by default, use 'false' as previous_value).
         /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972)
         /// Note: please check if the key already exists to prevent duplicate entries.
+        addSettingsChanges(settings_changes_history, "25.3.3.20000",
+        {
+            // Altinity Antalya modifications atop of 25.3
+            {"lock_object_storage_task_distribution_ms", 0, 0, "New setting."},
+        });
         addSettingsChanges(settings_changes_history, "25.2.1.20000",
         {
             // Altinity Antalya modifications atop of 25.2
diff --git a/src/Databases/DataLake/ICatalog.cpp b/src/Databases/DataLake/ICatalog.cpp
@@ -67,13 +67,19 @@ void TableMetadata::setLocation(const std::string & location_)
     auto pos_to_path = location_.substr(pos_to_bucket).find('/');
 
     if (pos_to_path == std::string::npos)
-        throw DB::Exception(DB::ErrorCodes::NOT_IMPLEMENTED, "Unexpected location format: {}", location_);
-
-    pos_to_path = pos_to_bucket + pos_to_path;
+    { // empty path
+        location_without_path = location_;
+        path.clear();
+        bucket = location_.substr(pos_to_bucket);
+    }
+    else
+    {
+        pos_to_path = pos_to_bucket + pos_to_path;
 
-    location_without_path = location_.substr(0, pos_to_path);
-    path = location_.substr(pos_to_path + 1);
-    bucket = location_.substr(pos_to_bucket, pos_to_path - pos_to_bucket);
+        location_without_path = location_.substr(0, pos_to_path);
+        path = location_.substr(pos_to_path + 1);
+        bucket = location_.substr(pos_to_bucket, pos_to_path - pos_to_bucket);
+    }
 
     LOG_TEST(getLogger("TableMetadata"),
              "Parsed location without path: {}, path: {}",
diff --git a/src/Disks/ObjectStorages/IObjectStorage.cpp b/src/Disks/ObjectStorages/IObjectStorage.cpp
@@ -8,6 +8,10 @@
 #include <Common/Exception.h>
 #include <Common/ObjectStorageKeyGenerator.h>
 
+#include <Poco/JSON/Object.h>
+#include <Poco/JSON/Parser.h>
+#include <Poco/JSON/JSONException.h>
+
 
 namespace DB
 {
@@ -107,4 +111,36 @@ void RelativePathWithMetadata::loadMetadata(ObjectStoragePtr object_storage)
     }
 }
 
+RelativePathWithMetadata::CommandInTaskResponse::CommandInTaskResponse(const std::string & task)
+{
+    Poco::JSON::Parser parser;
+    try
+    {
+        auto json = parser.parse(task).extract<Poco::JSON::Object::Ptr>();
+        if (!json)
+            return;
+
+        successfully_parsed = true;
+
+        if (json->has("retry_after_us"))
+            retry_after_us = json->getValue<size_t>("retry_after_us");
+    }
+    catch (const Poco::JSON::JSONException &)
+    { /// Not a JSON
+        return;
+    }
+}
+
+std::string RelativePathWithMetadata::CommandInTaskResponse::to_string() const
+{
+    Poco::JSON::Object json;
+    if (retry_after_us.has_value())
+        json.set("retry_after_us", retry_after_us.value());
+
+    std::ostringstream oss;
+    oss.exceptions(std::ios::failbit);
+    Poco::JSON::Stringifier::stringify(json, oss);
+    return oss.str();
+}
+
 }
diff --git a/src/Disks/ObjectStorages/IObjectStorage.h b/src/Disks/ObjectStorages/IObjectStorage.h
@@ -66,15 +66,37 @@ struct ObjectMetadata
 
 struct RelativePathWithMetadata
 {
+    class CommandInTaskResponse
+    {
+    public:
+        CommandInTaskResponse() {}
+        CommandInTaskResponse(const std::string & task);
+
+        bool is_parsed() const { return successfully_parsed; }
+        void set_retry_after_us(Poco::Timestamp::TimeDiff time_us) { retry_after_us = time_us; }
+
+        std::string to_string() const;
+
+        std::optional<Poco::Timestamp::TimeDiff> get_retry_after_us() const { return retry_after_us; }
+
+    private:
+        bool successfully_parsed = false;
+        std::optional<Poco::Timestamp::TimeDiff> retry_after_us;
+    };
+
     String relative_path;
     std::optional<ObjectMetadata> metadata;
+    CommandInTaskResponse command;
 
     RelativePathWithMetadata() = default;
 
-    explicit RelativePathWithMetadata(String relative_path_, std::optional<ObjectMetadata> metadata_ = std::nullopt)
-        : relative_path(std::move(relative_path_))
-        , metadata(std::move(metadata_))
-    {}
+    explicit RelativePathWithMetadata(const String & task_string, std::optional<ObjectMetadata> metadata_ = std::nullopt)
+        : metadata(std::move(metadata_))
+        , command(task_string)
+    {
+        if (!command.is_parsed())
+            relative_path = task_string;
+    }
 
     virtual ~RelativePathWithMetadata() = default;
 
@@ -85,6 +107,8 @@ struct RelativePathWithMetadata
     virtual size_t fileSizeInArchive() const { throw Exception(ErrorCodes::LOGICAL_ERROR, "Not an archive"); }
 
     void loadMetadata(ObjectStoragePtr object_storage);
+
+    const CommandInTaskResponse & getCommand() const { return command; }
 };
 
 struct ObjectKeyWithMetadata
diff --git a/src/IO/S3/Client.cpp b/src/IO/S3/Client.cpp
@@ -373,7 +373,7 @@ Model::HeadObjectOutcome Client::HeadObject(HeadObjectRequest & request) const
     auto bucket_uri = getURIForBucket(bucket);
     if (!bucket_uri)
     {
-        if (auto maybe_error = updateURIForBucketForHead(bucket); maybe_error.has_value())
+        if (auto maybe_error = updateURIForBucketForHead(bucket, request.GetKey()); maybe_error.has_value())
             return *maybe_error;
 
         if (auto region = getRegionForBucket(bucket); !region.empty())
@@ -578,7 +578,6 @@ Client::doRequest(RequestType & request, RequestFn request_fn) const
     if (auto uri = getURIForBucket(bucket); uri.has_value())
         request.overrideURI(std::move(*uri));
 
-
     bool found_new_endpoint = false;
     // if we found correct endpoint after 301 responses, update the cache for future requests
     SCOPE_EXIT(
@@ -813,12 +812,15 @@ std::optional<S3::URI> Client::getURIFromError(const Aws::S3::S3Error & error) c
 }
 
 // Do a list request because head requests don't have body in response
-std::optional<Aws::S3::S3Error> Client::updateURIForBucketForHead(const std::string & bucket) const
+// S3 Tables don't support ListObjects, so made dirty workaroung - changed on GetObject
+std::optional<Aws::S3::S3Error> Client::updateURIForBucketForHead(const std::string & bucket, const std::string & key) const
 {
-    ListObjectsV2Request req;
+    GetObjectRequest req;
     req.SetBucket(bucket);
-    req.SetMaxKeys(1);
-    auto result = ListObjectsV2(req);
+    req.SetKey(key);
+    req.SetRange("bytes=0-1");
+    auto result = GetObject(req);
+
     if (result.IsSuccess())
         return std::nullopt;
     return result.GetError();
diff --git a/src/IO/S3/Client.h b/src/IO/S3/Client.h
@@ -269,7 +269,7 @@ class Client : private Aws::S3::S3Client
 
     void updateURIForBucket(const std::string & bucket, S3::URI new_uri) const;
     std::optional<S3::URI> getURIFromError(const Aws::S3::S3Error & error) const;
-    std::optional<Aws::S3::S3Error> updateURIForBucketForHead(const std::string & bucket) const;
+    std::optional<Aws::S3::S3Error> updateURIForBucketForHead(const std::string & bucket, const std::string & key) const;
 
     std::optional<S3::URI> getURIForBucket(const std::string & bucket) const;
 
diff --git a/src/IO/S3/URI.cpp b/src/IO/S3/URI.cpp
@@ -157,10 +157,72 @@ URI::URI(const std::string & uri_, bool allow_archive_path_syntax)
     }
 }
 
+bool URI::isAWSRegion(std::string_view region)
+{
+    /// List from https://docs.aws.amazon.com/general/latest/gr/s3.html
+    static const std::unordered_set<std::string_view> regions = {
+        "us-east-2",
+        "us-east-1",
+        "us-west-1",
+        "us-west-2",
+        "af-south-1",
+        "ap-east-1",
+        "ap-south-2",
+        "ap-southeast-3",
+        "ap-southeast-5",
+        "ap-southeast-4",
+        "ap-south-1",
+        "ap-northeast-3",
+        "ap-northeast-2",
+        "ap-southeast-1",
+        "ap-southeast-2",
+        "ap-east-2",
+        "ap-southeast-7",
+        "ap-northeast-1",
+        "ca-central-1",
+        "ca-west-1",
+        "eu-central-1",
+        "eu-west-1",
+        "eu-west-2",
+        "eu-south-1",
+        "eu-west-3",
+        "eu-south-2",
+        "eu-north-1",
+        "eu-central-2",
+        "il-central-1",
+        "mx-central-1",
+        "me-south-1",
+        "me-central-1",
+        "sa-east-1",
+        "us-gov-east-1",
+        "us-gov-west-1"
+    };
+
+    /// 's3-us-west-2' is a legacy region format for S3 storage, equals to 'us-west-2'
+    /// See https://docs.aws.amazon.com/AmazonS3/latest/userguide/VirtualHosting.html#VirtualHostingBackwardsCompatibility
+    if (region.substr(0, 3) == "s3-")
+        region = region.substr(3);
+
+    return regions.contains(region);
+}
+
 void URI::addRegionToURI(const std::string &region)
 {
     if (auto pos = endpoint.find(".amazonaws.com"); pos != std::string::npos)
+    {
+        if (pos > 0)
+        { /// Check if region is already in endpoint to avoid add it second time
+            auto prev_pos = endpoint.find_last_of("/.", pos - 1);
+            if (prev_pos == std::string::npos)
+                prev_pos = 0;
+            else
+                ++prev_pos;
+            std::string_view endpoint_region = std::string_view(endpoint).substr(prev_pos, pos - prev_pos);
+            if (isAWSRegion(endpoint_region))
+                return;
+        }
         endpoint = endpoint.substr(0, pos) + "." + region + endpoint.substr(pos);
+    }
 }
 
 void URI::validateBucket(const String & bucket, const Poco::URI & uri)
diff --git a/src/IO/S3/URI.h b/src/IO/S3/URI.h
@@ -41,6 +41,10 @@ struct URI
 
     static void validateBucket(const std::string & bucket, const Poco::URI & uri);
 
+    /// Returns true if 'region' string is an AWS S3 region
+    /// https://docs.aws.amazon.com/general/latest/gr/s3.html
+    static bool isAWSRegion(std::string_view region);
+
 private:
     std::pair<std::string, std::optional<std::string>> getURIAndArchivePattern(const std::string & source);
 };
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp
@@ -487,7 +487,7 @@ void IcebergMetadata::updateSnapshot()
 
             relevant_snapshot = IcebergSnapshot{
                 getManifestList(getProperFilePathFromMetadataInfo(
-                    snapshot->getValue<String>(MANIFEST_LIST_PATH_FIELD), configuration_ptr->getPath(), table_location)),
+                    snapshot->getValue<String>(MANIFEST_LIST_PATH_FIELD), configuration_ptr->getPath(), table_location, configuration_ptr->getNamespace())),
                 relevant_snapshot_id, total_rows, total_bytes};
 
             if (!snapshot->has("schema-id"))
@@ -654,7 +654,7 @@ ManifestListPtr IcebergMetadata::getManifestList(const String & filename) const
         for (size_t i = 0; i < manifest_list_deserializer.rows(); ++i)
         {
             const std::string file_path = manifest_list_deserializer.getValueFromRowByName(i, MANIFEST_FILE_PATH_COLUMN, TypeIndex::String).safeGet<std::string>();
-            const auto manifest_file_name = getProperFilePathFromMetadataInfo(file_path, configuration_ptr->getPath(), table_location);
+            const auto manifest_file_name = getProperFilePathFromMetadataInfo(file_path, configuration_ptr->getPath(), table_location, configuration_ptr->getNamespace());
             Int64 added_sequence_number = 0;
             if (format_version > 1)
                 added_sequence_number = manifest_list_deserializer.getValueFromRowByName(i, SEQUENCE_NUMBER_COLUMN, TypeIndex::Int64).safeGet<Int64>();
@@ -706,6 +706,7 @@ ManifestFilePtr IcebergMetadata::getManifestFile(const String & filename, Int64
             schema_processor,
             inherited_sequence_number,
             table_location,
+            configuration_ptr->getNamespace(),
             context);
     };
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp
@@ -141,6 +141,7 @@ ManifestFileContent::ManifestFileContent(
     const IcebergSchemaProcessor & schema_processor,
     Int64 inherited_sequence_number,
     const String & table_location,
+    const String & common_namespace,
     DB::ContextPtr context)
 {
     this->schema_id = schema_id_;
@@ -205,7 +206,11 @@ ManifestFileContent::ManifestFileContent(
         }
         const auto status = ManifestEntryStatus(manifest_file_deserializer.getValueFromRowByName(i, COLUMN_STATUS_NAME, TypeIndex::Int32).safeGet<UInt64>());
 
-        const auto file_path = getProperFilePathFromMetadataInfo(manifest_file_deserializer.getValueFromRowByName(i, SUBCOLUMN_FILE_PATH_NAME, TypeIndex::String).safeGet<String>(), common_path, table_location);
+        const auto file_path = getProperFilePathFromMetadataInfo(
+            manifest_file_deserializer.getValueFromRowByName(i, SUBCOLUMN_FILE_PATH_NAME, TypeIndex::String).safeGet<String>(),
+            common_path,
+            table_location,
+            common_namespace);
 
         /// NOTE: This is weird, because in manifest file partition looks like this:
         /// {
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h
@@ -96,6 +96,7 @@ class ManifestFileContent
         const DB::IcebergSchemaProcessor & schema_processor,
         Int64 inherited_sequence_number,
         const std::string & table_location,
+        const std::string & common_namespace,
         DB::ContextPtr context);
 
     const std::vector<ManifestFileEntry> & getFiles() const;
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.cpp
@@ -28,7 +28,11 @@ using namespace DB;
 // This function is used to get the file path inside the directory which corresponds to iceberg table from the full blob path which is written in manifest and metadata files.
 // For example, if the full blob path is s3://bucket/table_name/data/00000-1-1234567890.avro, the function will return table_name/data/00000-1-1234567890.avro
 // Common path should end with "<table_name>" or "<table_name>/".
-std::string getProperFilePathFromMetadataInfo(std::string_view data_path, std::string_view common_path, std::string_view table_location)
+std::string getProperFilePathFromMetadataInfo(
+    std::string_view data_path,
+    std::string_view common_path,
+    std::string_view table_location,
+    std::string_view common_namespace)
 {
     auto trim_backward_slash = [](std::string_view str) -> std::string_view
     {
@@ -84,7 +88,20 @@ std::string getProperFilePathFromMetadataInfo(std::string_view data_path, std::s
     }
     else
     {
-        throw ::DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Expected to find '{}' in data path: '{}'", common_path, data_path);
+        /// Data files can have different path
+        pos = data_path.find("://");
+        if (pos == std::string::npos)
+            throw ::DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unexpected data path: '{}'", data_path);
+        pos = data_path.find("/", pos + 3);
+        if (pos == std::string::npos)
+            throw ::DB::Exception(DB::ErrorCodes::BAD_ARGUMENTS, "Unexpected data path: '{}'", data_path);
+        if (data_path.substr(pos + 1).starts_with(common_namespace))
+        {
+            auto new_pos = data_path.find("/", pos + 1);
+            if (new_pos - pos == common_namespace.length() + 1) /// bucket in the path
+                pos = new_pos;
+        }
+        return std::string(data_path.substr(pos));
     }
 }
 
diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/Utils.h
@@ -10,7 +10,11 @@
 namespace Iceberg
 {
 
-std::string getProperFilePathFromMetadataInfo(std::string_view data_path, std::string_view common_path, std::string_view table_location);
+std::string getProperFilePathFromMetadataInfo(
+    std::string_view data_path,
+    std::string_view common_path,
+    std::string_view table_location,
+    std::string_view common_namespace);
 
 }
 
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp b/src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp b/src/Storages/ObjectStorage/StorageObjectStorageSource.cpp
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.cpp
diff --git a/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h b/src/Storages/ObjectStorage/StorageObjectStorageStableTaskDistributor.h
diff --git a/tests/integration/test_s3_cache_locality/test.py b/tests/integration/test_s3_cache_locality/test.py
diff --git a/tests/integration/test_storage_iceberg/test.py b/tests/integration/test_storage_iceberg/test.py

Original file line number	Diff line number	Diff line change
`@@ -6141,6 +6141,9 @@ Cache the list of objects returned by list objects calls in object storage`
`6141`	`6141`	`)", EXPERIMENTAL) \`
`6142`	`6142`	`DECLARE(Bool, object_storage_remote_initiator, false, R"(`
`6143`	`6143`	`Execute request to object storage as remote on one of object_storage_cluster nodes.`
	`6144`	`+)", EXPERIMENTAL) \`
	`6145`	`+ DECLARE(UInt64, lock_object_storage_task_distribution_ms, 0, R"(`
	`6146`	`+In object storage distribution queries do not distibute tasks on non-prefetched nodes until prefetched node is active.`
`6144`	`6147`	`)", EXPERIMENTAL) \`
`6145`	`6148`	`\`
`6146`	`6149`