#include "config.h" #include #include #include #include #include #include #include #include #include namespace DB { template StoragePtr TableFunctionObjectStorageCluster::executeImpl( const ASTPtr & /*function*/, ContextPtr context, const std::string & table_name, ColumnsDescription cached_columns, bool is_insert_query) const { auto configuration = Base::getConfiguration(); ColumnsDescription columns; if (configuration->structure != "auto") columns = parseColumnsListFromString(configuration->structure, context); else if (!Base::structure_hint.empty()) columns = Base::structure_hint; else if (!cached_columns.empty()) columns = cached_columns; auto object_storage = Base::getObjectStorage(context, !is_insert_query); StoragePtr storage; if (context->getClientInfo().query_kind == ClientInfo::QueryKind::SECONDARY_QUERY) { /// On worker node this filename won't contains globs storage = std::make_shared( configuration, object_storage, context, StorageID(Base::getDatabaseName(), table_name), columns, ConstraintsDescription{}, /* comment */ String{}, /* format_settings */ std::nullopt, /// No format_settings /* mode */ LoadingStrictnessLevel::CREATE, /* catalog*/nullptr, /* if_not_exists*/false, /* is_datalake_query*/ false, /* distributed_processing */ true, /* partition_by_ */Base::partition_by, /* is_table_function */true, /* lazy_init */ true); } else { storage = std::make_shared( ITableFunctionCluster::cluster_name, configuration, object_storage, StorageID(Base::getDatabaseName(), table_name), columns, ConstraintsDescription{}, Base::partition_by, context); } storage->startup(); return storage; } void registerTableFunctionObjectStorageCluster(TableFunctionFactory & factory) { #if USE_AWS_S3 factory.registerFunction( { .documentation = { .description=R"(The table function can be used to read the data stored on S3 in parallel for many nodes in a specified cluster.)", .examples{{"s3Cluster", "SELECT * FROM s3Cluster(cluster, url, format, structure)", ""}}, .category = FunctionDocumentation::Category::TableFunction }, .allow_readonly = false } ); #endif #if USE_AZURE_BLOB_STORAGE factory.registerFunction( { .documentation = { .description=R"(The table function can be used to read the data stored on Azure Blob Storage in parallel for many nodes in a specified cluster.)", .examples{{ "azureBlobStorageCluster", "SELECT * FROM azureBlobStorageCluster(cluster, connection_string|storage_account_url, container_name, blobpath, " "[account_name, account_key, format, compression, structure])", ""}}, .category = FunctionDocumentation::Category::TableFunction }, .allow_readonly = false } ); #endif #if USE_HDFS factory.registerFunction( { .documentation = { .description=R"(The table function can be used to read the data stored on HDFS in parallel for many nodes in a specified cluster.)", .examples{{"HDFSCluster", "SELECT * FROM HDFSCluster(cluster, uri, format)", ""}}, .category = FunctionDocumentation::Category::TableFunction }, .allow_readonly = false } ); #endif UNUSED(factory); } #if USE_AVRO void registerTableFunctionIcebergCluster(TableFunctionFactory & factory) { UNUSED(factory); #if USE_AWS_S3 factory.registerFunction( {.documentation = {.description = R"(The table function can be used to read the Iceberg table stored on S3 object store in parallel for many nodes in a specified cluster.)", .examples{{"icebergS3Cluster", "SELECT * FROM icebergS3Cluster(cluster, url, [, NOSIGN | access_key_id, secret_access_key, [session_token]], format, [,compression])", ""}}, .category = FunctionDocumentation::Category::TableFunction}, .allow_readonly = false}); #endif #if USE_AZURE_BLOB_STORAGE factory.registerFunction( {.documentation = {.description = R"(The table function can be used to read the Iceberg table stored on Azure object store in parallel for many nodes in a specified cluster.)", .examples{{"icebergAzureCluster", "SELECT * FROM icebergAzureCluster(cluster, connection_string|storage_account_url, container_name, blobpath, [account_name, account_key, format, compression])", ""}}, .category = FunctionDocumentation::Category::TableFunction}, .allow_readonly = false}); #endif #if USE_HDFS factory.registerFunction( {.documentation = {.description = R"(The table function can be used to read the Iceberg table stored on HDFS virtual filesystem in parallel for many nodes in a specified cluster.)", .examples{{"icebergHDFSCluster", "SELECT * FROM icebergHDFSCluster(cluster, uri, [format], [structure], [compression_method])", ""}}, .category = FunctionDocumentation::Category::TableFunction}, .allow_readonly = false}); #endif } #endif #if USE_AWS_S3 #if USE_PARQUET && USE_DELTA_KERNEL_RS void registerTableFunctionDeltaLakeCluster(TableFunctionFactory & factory) { factory.registerFunction( {.documentation = {.description = R"(The table function can be used to read the DeltaLake table stored on object store in parallel for many nodes in a specified cluster.)", .examples{{"deltaLakeCluster", "SELECT * FROM deltaLakeCluster(cluster, url, access_key_id, secret_access_key)", ""}}, .category = FunctionDocumentation::Category::TableFunction}, .allow_readonly = false}); } #endif void registerTableFunctionHudiCluster(TableFunctionFactory & factory) { factory.registerFunction( {.documentation = {.description = R"(The table function can be used to read the Hudi table stored on object store in parallel for many nodes in a specified cluster.)", .examples{{"hudiCluster", "SELECT * FROM hudiCluster(cluster, url, access_key_id, secret_access_key)", ""}}, .category = FunctionDocumentation::Category::TableFunction}, .allow_readonly = false}); } #endif void registerDataLakeClusterTableFunctions(TableFunctionFactory & factory) { UNUSED(factory); #if USE_AVRO registerTableFunctionIcebergCluster(factory); #endif #if USE_AWS_S3 #if USE_PARQUET && USE_DELTA_KERNEL_RS registerTableFunctionDeltaLakeCluster(factory); #endif registerTableFunctionHudiCluster(factory); #endif } }