Merge branch 'antalya-25.6.5' into improvement/antalya/874

Author: zvonand

src/Analyzer/FunctionSecretArgumentsFinderTreeNode.h

@@ -71,8 +71,14 @@ class FunctionTreeNodeImpl : public AbstractFunction
     {
     public:
         explicit ArgumentsTreeNode(const QueryTreeNodes * arguments_) : arguments(arguments_) {}
-        size_t size() const override { return arguments ? arguments->size() : 0; }
-        std::unique_ptr<Argument> at(size_t n) const override { return std::make_unique<ArgumentTreeNode>(arguments->at(n).get()); }
+        size_t size() const override
+        { /// size withous skipped indexes
+            return arguments ? arguments->size() - skippedSize() : 0;
+        }
+        std::unique_ptr<Argument> at(size_t n) const override
+        { /// n is relative index, some can be skipped
+            return std::make_unique<ArgumentTreeNode>(arguments->at(getRealIndex(n)).get());
+        }
     private:
         const QueryTreeNodes * arguments = nullptr;
     };

src/Core/Settings.cpp

@@ -6841,6 +6841,15 @@ Enable PRQL - an alternative to SQL.
 )", EXPERIMENTAL) \
     DECLARE(Bool, enable_adaptive_memory_spill_scheduler, false, R"(
 Trigger processor to spill data into external storage adpatively. grace join is supported at present.
+)", EXPERIMENTAL) \
+    DECLARE(String, object_storage_cluster, "", R"(
+Cluster to make distributed requests to object storages with alternative syntax.
+)", EXPERIMENTAL) \
+    DECLARE(UInt64, object_storage_max_nodes, 0, R"(
+Limit for hosts used for request in object storage cluster table functions - azureBlobStorageCluster, s3Cluster, hdfsCluster, etc.
+Possible values:
+- Positive integer.
+- 0 — All hosts in cluster.
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_delta_kernel_rs, true, R"(
 Allow experimental delta-kernel-rs implementation.

src/Core/SettingsChangesHistory.cpp

@@ -67,13 +67,14 @@ 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.6.5.2000",
         {
             {"allow_experimental_database_iceberg", false, true, "Turned ON by default for Antalya"},
             {"allow_experimental_database_unity_catalog", false, true, "Turned ON by default for Antalya"},
             {"allow_experimental_database_glue_catalog", false, true, "Turned ON by default for Antalya"},
             {"output_format_parquet_enum_as_byte_array", true, true, "Enable writing Enum as byte array in Parquet by default"},
+            {"object_storage_cluster", "", "", "New setting"},
+            {"object_storage_max_nodes", 0, 0, "New setting"},
         });
         addSettingsChanges(settings_changes_history, "25.6",
         {

src/Databases/DataLake/DatabaseDataLake.cpp

@@ -18,6 +18,7 @@
 #include <Storages/ConstraintsDescription.h>
 #include <Storages/StorageNull.h>
 #include <Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h>
+#include <Storages/ObjectStorage/StorageObjectStorageCluster.h>
 
 #include <Interpreters/evaluateConstantExpression.h>
 #include <Interpreters/Context.h>
@@ -43,6 +44,7 @@ namespace DatabaseDataLakeSetting
     extern const DatabaseDataLakeSettingsString storage_endpoint;
     extern const DatabaseDataLakeSettingsString oauth_server_uri;
     extern const DatabaseDataLakeSettingsBool vended_credentials;
+    extern const DatabaseDataLakeSettingsString object_storage_cluster;
 
 
     extern const DatabaseDataLakeSettingsString aws_access_key_id;
@@ -428,21 +430,22 @@ StoragePtr DatabaseDataLake::tryGetTableImpl(const String & name, ContextPtr con
 
     /// with_table_structure = false: because there will be
     /// no table structure in table definition AST.
-    StorageObjectStorage::Configuration::initialize(*configuration, args, context_copy, /* with_table_structure */false);
+    configuration->initialize(args, context_copy, /* with_table_structure */false);
 
-    return std::make_shared<StorageObjectStorage>(
+    auto cluster_name = settings[DatabaseDataLakeSetting::object_storage_cluster].value;
+    
+    return std::make_shared<StorageObjectStorageCluster>(
+        cluster_name,
         configuration,
         configuration->createObjectStorage(context_copy, /* is_readonly */ false),
-        context_copy,
         StorageID(getDatabaseName(), name),
         /* columns */columns,
         /* constraints */ConstraintsDescription{},
+        /* partition_by */nullptr,
+        context_copy,
         /* comment */"",
         getFormatSettings(context_copy),
         LoadingStrictnessLevel::CREATE,
-        /* distributed_processing */false,
-        /* partition_by */nullptr,
-        /* is_table_function */false,
         /* lazy_init */true);
 }
 
@@ -488,7 +491,7 @@ DatabaseTablesIteratorPtr DatabaseDataLake::getLightweightTablesIterator(
     const FilterByNameFunction & filter_by_table_name,
     bool skip_not_loaded) const
 {
-     Tables tables;
+    Tables tables;
     auto catalog = getCatalog();
     DB::Names iceberg_tables;
 

src/Databases/DataLake/GlueCatalog.cpp

@@ -441,7 +441,7 @@ bool GlueCatalog::classifyTimestampTZ(const String & column_name, const TableMet
         auto storage_settings = std::make_shared<DB::DataLakeStorageSettings>();
         storage_settings->loadFromSettingsChanges(settings.allChanged());
         auto configuration = std::make_shared<DB::StorageS3IcebergConfiguration>(storage_settings);
-        DB::StorageObjectStorage::Configuration::initialize(*configuration, args, getContext(), false);
+        configuration->initialize(args, getContext(), false);
 
         auto object_storage = configuration->createObjectStorage(getContext(), true);
         const auto & read_settings = getContext()->getReadSettings();

src/Databases/DataLake/ICatalog.cpp

@@ -70,13 +70,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: {}",

src/Disks/DiskType.cpp

@@ -12,7 +12,7 @@ namespace ErrorCodes
     extern const int LOGICAL_ERROR;
 }
 
-MetadataStorageType metadataTypeFromString(const String & type)
+MetadataStorageType metadataTypeFromString(const std::string & type)
 {
     auto check_type = Poco::toLower(type);
     if (check_type == "local")
@@ -60,25 +60,49 @@ std::string DataSourceDescription::toString() const
         case DataSourceType::RAM:
             return "memory";
         case DataSourceType::ObjectStorage:
-        {
-            switch (object_storage_type)
-            {
-                case ObjectStorageType::S3:
-                    return "s3";
-                case ObjectStorageType::HDFS:
-                    return "hdfs";
-                case ObjectStorageType::Azure:
-                    return "azure_blob_storage";
-                case ObjectStorageType::Local:
-                    return "local_blob_storage";
-                case ObjectStorageType::Web:
-                    return "web";
-                case ObjectStorageType::None:
-                    return "none";
-                case ObjectStorageType::Max:
-                    throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected object storage type: Max");
-            }
-        }
+            return DB::toString(object_storage_type);
     }
 }
+
+ObjectStorageType objectStorageTypeFromString(const std::string & type)
+{
+    auto check_type = Poco::toLower(type);
+    if (check_type == "s3")
+        return ObjectStorageType::S3;
+    if (check_type == "hdfs")
+        return ObjectStorageType::HDFS;
+    if (check_type == "azure_blob_storage" || check_type == "azure")
+        return ObjectStorageType::Azure;
+    if (check_type == "local_blob_storage" || check_type == "local")
+        return ObjectStorageType::Local;
+    if (check_type == "web")
+        return ObjectStorageType::Web;
+    if (check_type == "none")
+        return ObjectStorageType::None;
+
+    throw Exception(ErrorCodes::UNKNOWN_ELEMENT_IN_CONFIG,
+        "Unknown object storage type: {}", type);
+}
+
+std::string toString(ObjectStorageType type)
+{
+    switch (type)
+    {
+        case ObjectStorageType::S3:
+            return "s3";
+        case ObjectStorageType::HDFS:
+            return "hdfs";
+        case ObjectStorageType::Azure:
+            return "azure_blob_storage";
+        case ObjectStorageType::Local:
+            return "local_blob_storage";
+        case ObjectStorageType::Web:
+            return "web";
+        case ObjectStorageType::None:
+            return "none";
+        case ObjectStorageType::Max:
+            throw Exception(ErrorCodes::LOGICAL_ERROR, "Unexpected object storage type: Max");
+    }
+}
+
 }

src/Disks/DiskType.h

@@ -36,8 +36,10 @@ enum class MetadataStorageType : uint8_t
     Memory,
 };
 
-MetadataStorageType metadataTypeFromString(const String & type);
-String toString(DataSourceType data_source_type);
+MetadataStorageType metadataTypeFromString(const std::string & type);
+
+ObjectStorageType objectStorageTypeFromString(const std::string & type);
+std::string toString(ObjectStorageType type);
 
 struct DataSourceDescription
 {

src/IO/S3/Client.cpp

@@ -384,7 +384,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())
@@ -589,7 +589,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(
@@ -869,12 +868,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();

src/IO/S3/Client.h

@@ -279,7 +279,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;