Fix deltaLakeS3/deltaLakeAzure

Author: ianton-ru

src/Access/Common/AccessType.h

@@ -23,7 +23,6 @@ namespace AccessTypeObjects
 enum class Source : uint8_t
 {
 #define APPLY_FOR_SOURCE(M) \
-    M(UNDEFINED, "") \
     M(FILE, "File") \
     M(URL, "") \
     M(REMOTE, "Distributed") \

src/Storages/ObjectStorage/StorageObjectStorageCluster.cpp

@@ -288,6 +288,8 @@ void StorageObjectStorageCluster::updateQueryForDistributedEngineIfNeeded(ASTPtr
         {"IcebergAzure", "icebergAzure"},
         {"IcebergHDFS", "icebergHDFS"},
         {"DeltaLake", "deltaLake"},
+        {"DeltaLakeS3", "deltaLakeS3"},
+        {"DeltaLakeAzure", "deltaLakeAzure"},
         {"Hudi", "hudi"}
     };
 
@@ -407,6 +409,8 @@ void StorageObjectStorageCluster::updateQueryToSendIfNeeded(
             {"icebergAzure", "icebergAzureCluster"},
             {"icebergHDFS", "icebergHDFSCluster"},
             {"deltaLake", "deltaLakeCluster"},
+            {"deltaLakeS3", "deltaLakeS3Cluster"},
+            {"deltaLakeAzure", "deltaLakeAzureCluster"},
             {"hudi", "hudiCluster"},
         };
 

src/Storages/ObjectStorage/StorageObjectStorageDefinitions.h

@@ -162,6 +162,20 @@ struct DeltaLakeClusterDefinition
     static constexpr auto non_clustered_storage_engine_name = DeltaLakeDefinition::storage_engine_name;
 };
 
+struct DeltaLakeS3ClusterDefinition
+{
+    static constexpr auto name = "deltaLakeS3Cluster";
+    static constexpr auto storage_engine_name = "DeltaLakeS3Cluster";
+    static constexpr auto non_clustered_storage_engine_name = DeltaLakeS3Definition::storage_engine_name;
+};
+
+struct DeltaLakeAzureClusterDefinition
+{
+    static constexpr auto name = "deltaLakeAzureCluster";
+    static constexpr auto storage_engine_name = "DeltaLakeAzureCluster";
+    static constexpr auto non_clustered_storage_engine_name = DeltaLakeAzureDefinition::storage_engine_name;
+};
+
 struct HudiClusterDefinition
 {
     static constexpr auto name = "hudiCluster";

src/Storages/ObjectStorage/registerStorageObjectStorage.cpp

@@ -220,7 +220,7 @@ void registerStorageIceberg(StorageFactory & factory)
             .supports_settings = true,
             .supports_sort_order = true,
             .supports_schema_inference = true,
-            .source_access_type = AccessTypeObjects::Source::UNDEFINED,
+            .source_access_type = AccessTypeObjects::Source::S3,
             .has_builtin_setting_fn = DataLakeStorageSettings::hasBuiltin,
         });
 
@@ -298,7 +298,7 @@ void registerStorageIceberg(StorageFactory & factory)
 #if USE_PARQUET && USE_DELTA_KERNEL_RS
 void registerStorageDeltaLake(StorageFactory & factory)
 {
-#if USE_AWS_S3
+#   if USE_AWS_S3
     factory.registerStorage(
         DeltaLakeDefinition::storage_engine_name,
         [&](const StorageFactory::Arguments & args)

src/TableFunctions/TableFunctionObjectStorage.cpp

@@ -312,6 +312,11 @@ template class TableFunctionObjectStorage<IcebergHDFSClusterDefinition, StorageH
 
 #if USE_PARQUET && USE_AWS_S3 && USE_DELTA_KERNEL_RS
 template class TableFunctionObjectStorage<DeltaLakeClusterDefinition, StorageS3DeltaLakeConfiguration, true>;
+template class TableFunctionObjectStorage<DeltaLakeS3ClusterDefinition, StorageS3DeltaLakeConfiguration, true>;
+#endif
+
+#if USE_PARQUET && USE_AWS_S3 && USE_DELTA_KERNEL_RS
+template class TableFunctionObjectStorage<DeltaLakeAzureClusterDefinition, StorageAzureDeltaLakeConfiguration, true>;
 #endif
 
 #if USE_AWS_S3

src/TableFunctions/TableFunctionObjectStorage.h

@@ -138,13 +138,13 @@ using TableFunctionIcebergHDFS = TableFunctionObjectStorage<IcebergHDFSDefinitio
 using TableFunctionIcebergLocal = TableFunctionObjectStorage<IcebergLocalDefinition, StorageLocalIcebergConfiguration, true>;
 #endif
 #if USE_PARQUET && USE_DELTA_KERNEL_RS
-#if USE_AWS_S3
+#    if USE_AWS_S3
 using TableFunctionDeltaLake = TableFunctionObjectStorage<DeltaLakeDefinition, StorageS3DeltaLakeConfiguration, true>;
 using TableFunctionDeltaLakeS3 = TableFunctionObjectStorage<DeltaLakeS3Definition, StorageS3DeltaLakeConfiguration, true>;
-#endif
-#if USE_AZURE_BLOB_STORAGE
+#    endif
+#    if USE_AZURE_BLOB_STORAGE
 using TableFunctionDeltaLakeAzure = TableFunctionObjectStorage<DeltaLakeAzureDefinition, StorageAzureDeltaLakeConfiguration, true>;
-#endif
+#    endif
 // New alias for local Delta Lake table function
 using TableFunctionDeltaLakeLocal = TableFunctionObjectStorage<DeltaLakeLocalDefinition, StorageLocalDeltaLakeConfiguration, true>;
 #endif

src/TableFunctions/TableFunctionObjectStorageCluster.cpp

@@ -184,19 +184,37 @@ void registerTableFunctionIcebergCluster(TableFunctionFactory & factory)
 }
 #endif
 
-#if USE_AWS_S3
 #if USE_PARQUET && USE_DELTA_KERNEL_RS
 void registerTableFunctionDeltaLakeCluster(TableFunctionFactory & factory)
 {
+#   if USE_AWS_S3
     factory.registerFunction<TableFunctionDeltaLakeCluster>(
         {.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{{DeltaLakeClusterDefinition::name, "SELECT * FROM deltaLakeCluster(cluster, url, access_key_id, secret_access_key)", ""}},
             .category = FunctionDocumentation::Category::TableFunction},
          .allow_readonly = false});
+
+    factory.registerFunction<TableFunctionDeltaLakeS3Cluster>(
+        {.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{{DeltaLakeS3ClusterDefinition::name, "SELECT * FROM deltaLakeS3Cluster(cluster, url, access_key_id, secret_access_key)", ""}},
+            .category = FunctionDocumentation::Category::TableFunction},
+         .allow_readonly = false});
+#   endif
+
+#   if USE_AZURE_BLOB_STORAGE
+    factory.registerFunction<TableFunctionDeltaLakeAzureCluster>(
+        {.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{{DeltaLakeAzureClusterDefinition::name, "SELECT * FROM deltaLakeAzureCluster(cluster, url, access_key_id, secret_access_key)", ""}},
+            .category = FunctionDocumentation::Category::TableFunction},
+         .allow_readonly = false});
+#   endif
 }
 #endif
 
+#if USE_AWS_S3
 void registerTableFunctionHudiCluster(TableFunctionFactory & factory)
 {
     factory.registerFunction<TableFunctionHudiCluster>(

src/TableFunctions/TableFunctionObjectStorageCluster.h

@@ -78,6 +78,11 @@ using TableFunctionIcebergHDFSCluster = TableFunctionObjectStorageCluster<Iceber
 
 #if USE_AWS_S3 && USE_PARQUET && USE_DELTA_KERNEL_RS
 using TableFunctionDeltaLakeCluster = TableFunctionObjectStorageCluster<DeltaLakeClusterDefinition, StorageS3DeltaLakeConfiguration, true>;
+using TableFunctionDeltaLakeS3Cluster = TableFunctionObjectStorageCluster<DeltaLakeS3ClusterDefinition, StorageS3DeltaLakeConfiguration, true>;
+#endif
+
+#if USE_AZURE_BLOB_STORAGE && USE_PARQUET && USE_DELTA_KERNEL_RS
+using TableFunctionDeltaLakeAzureCluster = TableFunctionObjectStorageCluster<DeltaLakeAzureClusterDefinition, StorageAzureDeltaLakeConfiguration, true>;
 #endif
 
 #if USE_AWS_S3

src/TableFunctions/TableFunctionObjectStorageClusterFallback.cpp

@@ -74,6 +74,20 @@ struct DeltaLakeClusterFallbackDefinition
     static constexpr auto storage_engine_cluster_name = "DeltaLakeS3Cluster";
 };
 
+struct DeltaLakeS3ClusterFallbackDefinition
+{
+    static constexpr auto name = "deltaLakeS3";
+    static constexpr auto storage_engine_name = "S3";
+    static constexpr auto storage_engine_cluster_name = "DeltaLakeS3Cluster";
+};
+
+struct DeltaLakeAzureClusterFallbackDefinition
+{
+    static constexpr auto name = "deltaLakeAzure";
+    static constexpr auto storage_engine_name = "Azure";
+    static constexpr auto storage_engine_cluster_name = "DeltaLakeAzureCluster";
+};
+
 struct HudiClusterFallbackDefinition
 {
     static constexpr auto name = "hudi";
@@ -165,6 +179,11 @@ using TableFunctionIcebergHDFSClusterFallback = TableFunctionObjectStorageCluste
 
 #if USE_AWS_S3 && USE_PARQUET && USE_DELTA_KERNEL_RS
 using TableFunctionDeltaLakeClusterFallback = TableFunctionObjectStorageClusterFallback<DeltaLakeClusterFallbackDefinition, TableFunctionDeltaLakeCluster>;
+using TableFunctionDeltaLakeS3ClusterFallback = TableFunctionObjectStorageClusterFallback<DeltaLakeS3ClusterFallbackDefinition, TableFunctionDeltaLakeS3Cluster>;
+#endif
+
+#if USE_AZURE_BLOB_STORAGE && USE_PARQUET && USE_DELTA_KERNEL_RS
+using TableFunctionDeltaLakeAzureClusterFallback = TableFunctionObjectStorageClusterFallback<DeltaLakeAzureClusterFallbackDefinition, TableFunctionDeltaLakeAzureCluster>;
 #endif
 
 #if USE_AWS_S3
@@ -337,7 +356,8 @@ void registerTableFunctionObjectStorageClusterFallback(TableFunctionFactory & fa
     );
 #endif
 
-#if USE_AWS_S3 && USE_PARQUET && USE_DELTA_KERNEL_RS
+#if USE_PARQUET && USE_DELTA_KERNEL_RS
+#   if USE_AWS_S3
     factory.registerFunction<TableFunctionDeltaLakeClusterFallback>(
     {
         .documentation = {
@@ -358,6 +378,49 @@ void registerTableFunctionObjectStorageClusterFallback(TableFunctionFactory & fa
         .allow_readonly = false
     }
     );
+    factory.registerFunction<TableFunctionDeltaLakeS3ClusterFallback>(
+    {
+        .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 or from single node.)",
+            .examples{
+                {
+                    "deltaLakeS3",
+                    "SELECT * FROM deltaLakeS3(url, access_key_id, secret_access_key)", ""
+                },
+                {
+                    "deltaLakeS3",
+                    "SELECT * FROM deltaLakeS3(url, access_key_id, secret_access_key) "
+                    "SETTINGS object_storage_cluster='cluster'", ""
+                },
+            },
+            .category = FunctionDocumentation::Category::TableFunction
+        },
+        .allow_readonly = false
+    }
+    );
+#   endif
+#   if USE_AZURE_BLOB_STORAGE
+    factory.registerFunction<TableFunctionDeltaLakeAzureClusterFallback>(
+    {
+        .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 or from single node.)",
+            .examples{
+                {
+                    "deltaLakeAzure",
+                    "SELECT * FROM deltaLakeAzure(url, access_key_id, secret_access_key)", ""
+                },
+                {
+                    "deltaLakeAzure",
+                    "SELECT * FROM deltaLakeAzure(url, access_key_id, secret_access_key) "
+                    "SETTINGS object_storage_cluster='cluster'", ""
+                },
+            },
+            .category = FunctionDocumentation::Category::TableFunction
+        },
+        .allow_readonly = false
+    }
+    );
+#   endif
 #endif
 
 #if USE_AWS_S3

tests/integration/test_storage_iceberg/test.py

@@ -1,4 +1,5 @@
 import logging
+import json
 import os
 import subprocess
 import time