Distributed request to tables with Object Storage Engines

src/Core/SettingsChangesHistory.cpp

@@ -73,6 +73,7 @@ static std::initializer_list<std::pair<ClickHouseVersion, SettingsChangesHistory
             {"least_greatest_legacy_null_behavior", true, false, "New setting"},
             {"object_storage_cluster", "", "", "New setting"},
             {"object_storage_max_nodes", 0, 0, "New setting"},
+            {"input_format_parquet_use_metadata_cache", false, false, "New setting"},
         }
     },
     {"24.11",

src/Databases/Iceberg/DatabaseIceberg.cpp

@@ -12,6 +12,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>
@@ -37,10 +38,12 @@ namespace DatabaseIcebergSetting
     extern const DatabaseIcebergSettingsString storage_endpoint;
     extern const DatabaseIcebergSettingsString oauth_server_uri;
     extern const DatabaseIcebergSettingsBool vended_credentials;
+    extern const DatabaseIcebergSettingsString object_storage_cluster;
 }
 namespace Setting
 {
     extern const SettingsBool allow_experimental_database_iceberg;
+    extern const SettingsString object_storage_cluster;
 }
 
 namespace ErrorCodes
@@ -235,19 +238,37 @@ StoragePtr DatabaseIceberg::tryGetTable(const String & name, ContextPtr context_
     /// no table structure in table definition AST.
     StorageObjectStorage::Configuration::initialize(*configuration, args, context_, /* with_table_structure */false, std::move(storage_settings));
 
-    return std::make_shared<StorageObjectStorage>(
-        configuration,
-        configuration->createObjectStorage(context_, /* is_readonly */ false),
-        context_,
-        StorageID(getDatabaseName(), name),
-        /* columns */columns,
-        /* constraints */ConstraintsDescription{},
-        /* comment */"",
-        getFormatSettings(context_),
-        LoadingStrictnessLevel::CREATE,
-        /* distributed_processing */false,
-        /* partition_by */nullptr,
-        /* lazy_init */true);
+    auto cluster_name = settings[DatabaseIcebergSetting::object_storage_cluster].value;
+    if (cluster_name.empty())
+        cluster_name = context_->getSettingsRef()[Setting::object_storage_cluster].value;
+
+    if (cluster_name.empty())
+    {
+        return std::make_shared<StorageObjectStorage>(
+            configuration,
+            configuration->createObjectStorage(context_, /* is_readonly */ false),
+            context_,
+            StorageID(getDatabaseName(), name),
+            /* columns */columns,
+            /* constraints */ConstraintsDescription{},
+            /* comment */"",
+            getFormatSettings(context_),
+            LoadingStrictnessLevel::CREATE,
+            /* distributed_processing */false,
+            /* partition_by */nullptr,
+            /* lazy_init */true);
+    }
+    else
+    {
+        return std::make_shared<StorageObjectStorageCluster>(
+            cluster_name,
+            configuration,
+            configuration->createObjectStorage(context_, /* is_readonly */ false),
+            StorageID(getDatabaseName(), name),
+            columns,
+            ConstraintsDescription{},
+            context_);
+    }
 }
 
 DatabaseTablesIteratorPtr DatabaseIceberg::getTablesIterator(

src/Databases/Iceberg/DatabaseIcebergSettings.cpp

@@ -23,6 +23,7 @@ namespace ErrorCodes
     DECLARE(String, warehouse, "", "Warehouse name inside the catalog", 0)             \
     DECLARE(String, auth_header, "", "Authorization header of format 'Authorization: <scheme> <auth_info>'", 0)             \
     DECLARE(String, storage_endpoint, "", "Object storage endpoint", 0) \
+    DECLARE(String, object_storage_cluster, "", "Cluster for distributed requests", 0) \
 
 #define LIST_OF_DATABASE_ICEBERG_SETTINGS(M, ALIAS) \
     DATABASE_ICEBERG_RELATED_SETTINGS(M, ALIAS)

src/Parsers/FunctionSecretArgumentsFinder.h

@@ -102,7 +102,7 @@ class FunctionSecretArgumentsFinder
         }
         else if ((function->name() == "s3") || (function->name() == "cosn") || (function->name() == "oss") ||
                  (function->name() == "deltaLake") || (function->name() == "hudi") || (function->name() == "iceberg") ||
-                 (function->name() == "gcs"))
+                 (function->name() == "icebergS3") || (function->name() == "gcs"))
         {
             /// s3('url', 'aws_access_key_id', 'aws_secret_access_key', ...)
             findS3FunctionSecretArguments(/* is_cluster_function= */ false);

src/Storages/IStorageCluster.h

@@ -43,6 +43,8 @@ class IStorageCluster : public IStorage
     bool supportsOptimizationToSubcolumns() const override { return false; }
     bool supportsTrivialCountOptimization(const StorageSnapshotPtr &, ContextPtr) const override { return true; }
 
+    const String & getClusterName() const { return cluster_name; }
+
 protected:
     virtual void updateBeforeRead(const ContextPtr &) {}
     virtual void updateQueryToSendIfNeeded(ASTPtr & /*query*/, const StorageSnapshotPtr & /*storage_snapshot*/, const ContextPtr & /*context*/) {}

src/Storages/ObjectStorage/Azure/Configuration.cpp

@@ -154,6 +154,14 @@ void StorageAzureConfiguration::fromNamedCollection(const NamedCollection & coll
     compression_method = collection.getOrDefault<String>("compression_method", collection.getOrDefault<String>("compression", "auto"));
 
     blobs_paths = {blob_path};
+    if (account_name && account_key)
+    {
+        if (saved_params.empty())
+        {
+            saved_params.push_back(*account_name);
+            saved_params.push_back(*account_key);
+        }
+    }
     connection_params = getConnectionParams(connection_url, container_name, account_name, account_key, context);
 }
 
@@ -173,7 +181,6 @@ void StorageAzureConfiguration::fromAST(ASTs & engine_args, ContextPtr context,
 
     std::unordered_map<std::string_view, size_t> engine_args_to_idx;
 
-
     String connection_url = checkAndGetLiteralArgument<String>(engine_args[0], "connection_string/storage_account_url");
     String container_name = checkAndGetLiteralArgument<String>(engine_args[1], "container");
     blob_path = checkAndGetLiteralArgument<String>(engine_args[2], "blobpath");
@@ -279,6 +286,14 @@ void StorageAzureConfiguration::fromAST(ASTs & engine_args, ContextPtr context,
     }
 
     blobs_paths = {blob_path};
+    if (account_name && account_key)
+    {
+        if (saved_params.empty())
+        {
+            saved_params.push_back(*account_name);
+            saved_params.push_back(*account_key);
+        }
+    }
     connection_params = getConnectionParams(connection_url, container_name, account_name, account_key, context);
 }
 
@@ -444,6 +459,22 @@ void StorageAzureConfiguration::addStructureAndFormatToArgsIfNeeded(
     }
 }
 
+void StorageAzureConfiguration::getTableFunctionArguments(ASTs & args) const
+{
+    if (!args.empty())
+    { /// Just check
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Arguments are not empty");
+    }
+
+    args.push_back(std::make_shared<ASTLiteral>(connection_params.endpoint.storage_account_url));
+    args.push_back(std::make_shared<ASTIdentifier>(connection_params.endpoint.container_name));
+    args.push_back(std::make_shared<ASTLiteral>(blob_path));
+    for (const auto & arg : saved_params)
+    {
+        args.push_back(std::make_shared<ASTLiteral>(arg));
+    }
+}
+
 }
 
 #endif

src/Storages/ObjectStorage/Azure/Configuration.h

@@ -79,13 +79,16 @@ class StorageAzureConfiguration : public StorageObjectStorage::Configuration
         ContextPtr context,
         bool with_structure) override;
 
+    void getTableFunctionArguments(ASTs & args) const override;
+
 protected:
     void fromNamedCollection(const NamedCollection & collection, ContextPtr context) override;
     void fromAST(ASTs & args, ContextPtr context, bool with_structure) override;
 
     std::string blob_path;
     std::vector<String> blobs_paths;
     AzureBlobStorage::ConnectionParams connection_params;
+    std::vector<std::string> saved_params;
 };
 
 }

src/Storages/ObjectStorage/HDFS/Configuration.cpp

@@ -235,6 +235,16 @@ void StorageHDFSConfiguration::addStructureAndFormatToArgsIfNeeded(
     }
 }
 
+void StorageHDFSConfiguration::getTableFunctionArguments(ASTs & args) const
+{
+    if (!args.empty())
+    { /// Just check
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Arguments are not empty");
+    }
+
+    args.push_back(std::make_shared<ASTLiteral>(url + path));
+}
+
 }
 
 #endif

src/Storages/ObjectStorage/HDFS/Configuration.h

@@ -65,6 +65,8 @@ class StorageHDFSConfiguration : public StorageObjectStorage::Configuration
         ContextPtr context,
         bool with_structure) override;
 
+    void getTableFunctionArguments(ASTs & args) const override;
+
 private:
     void fromNamedCollection(const NamedCollection &, ContextPtr context) override;
     void fromAST(ASTs & args, ContextPtr, bool /* with_structure */) override;

src/Storages/ObjectStorage/S3/Configuration.cpp

@@ -363,11 +363,11 @@ void StorageS3Configuration::fromAST(ASTs & args, ContextPtr context, bool with_
 
     if (engine_args_to_idx.contains("format"))
     {
-        format = checkAndGetLiteralArgument<String>(args[engine_args_to_idx["format"]], "format");
+        auto format_ = checkAndGetLiteralArgument<String>(args[engine_args_to_idx["format"]], "format");
         /// Set format to configuration only of it's not 'auto',
         /// because we can have default format set in configuration.
-        if (format != "auto")
-            format = format;
+        if (format_ != "auto")
+            format = format_;
     }
 
     if (engine_args_to_idx.contains("structure"))
@@ -585,6 +585,31 @@ void StorageS3Configuration::addStructureAndFormatToArgsIfNeeded(
     }
 }
 
+void StorageS3Configuration::getTableFunctionArguments(ASTs & args) const
+{
+    if (!args.empty())
+    { /// Just check
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Arguments are not empty");
+    }
+
+    args.push_back(std::make_shared<ASTLiteral>(url.uri_str));
+    if (auth_settings[S3AuthSetting::no_sign_request])
+    {
+        args.push_back(std::make_shared<ASTLiteral>("NOSIGN"));
+    }
+    else
+    {
+        args.push_back(std::make_shared<ASTLiteral>(auth_settings[S3AuthSetting::access_key_id].value));
+        args.push_back(std::make_shared<ASTLiteral>(auth_settings[S3AuthSetting::secret_access_key].value));
+        if (!auth_settings[S3AuthSetting::session_token].value.empty())
+            args.push_back(std::make_shared<ASTLiteral>(auth_settings[S3AuthSetting::session_token].value));
+        if (format != "auto")
+            args.push_back(std::make_shared<ASTLiteral>(format));
+        if (!compression_method.empty())
+            args.push_back(std::make_shared<ASTLiteral>(compression_method));
+    }
+}
+
 }
 
 #endif

src/Storages/ObjectStorage/S3/Configuration.h

@@ -94,6 +94,8 @@ class StorageS3Configuration : public StorageObjectStorage::Configuration
         ContextPtr context,
         bool with_structure) override;
 
+    void getTableFunctionArguments(ASTs & args) const override;
+
 private:
     void fromNamedCollection(const NamedCollection & collection, ContextPtr context) override;
     void fromAST(ASTs & args, ContextPtr context, bool with_structure) override;

src/Storages/ObjectStorage/StorageObjectStorage.cpp

@@ -60,6 +60,9 @@ String StorageObjectStorage::getPathSample(ContextPtr context)
     if (context->getSettingsRef()[Setting::use_hive_partitioning])
         local_distributed_processing = false;
 
+    if (!configuration->isArchive() && !configuration->isPathWithGlobs() && !local_distributed_processing)
+        return configuration->getPath();
+
     auto file_iterator = StorageObjectStorageSource::createFileIterator(
         configuration,
         query_settings,
@@ -72,9 +75,6 @@ String StorageObjectStorage::getPathSample(ContextPtr context)
         {} // file_progress_callback
     );
 
-    if (!configuration->isArchive() && !configuration->isPathWithGlobs() && !local_distributed_processing)
-        return configuration->getPath();
-
     if (auto file = file_iterator->next(0))
         return file->getPath();
     return "";

src/Storages/ObjectStorage/StorageObjectStorage.h

@@ -247,6 +247,10 @@ class StorageObjectStorage::Configuration
 
     virtual void update(ObjectStoragePtr object_storage, ContextPtr local_context);
 
+    virtual void getTableFunctionArguments(ASTs & /* args */) const
+    {
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Method getTableFunctionArguments is not supported by storage {}", getEngineName());
+    }
 
 protected:
     virtual void fromNamedCollection(const NamedCollection & collection, ContextPtr context) = 0;