Merge pull request ClickHouse#79203 from ClickHouse/fix-delta-kernel-auth

Fix delta-kernel with http based endpoints, fix NOSIGN

Author: kssenii

src/Storages/ObjectStorage/DataLakes/DeltaLake/KernelHelper.cpp

@@ -11,6 +11,11 @@ namespace DB::ErrorCodes
     extern const int NOT_IMPLEMENTED;
 }
 
+namespace DB::S3AuthSetting
+{
+    extern const S3AuthSettingsBool no_sign_request;
+}
+
 namespace DeltaLake
 {
 
@@ -23,14 +28,20 @@ class S3KernelHelper final : public IKernelHelper
         const std::string & access_key_id_,
         const std::string & secret_access_key_,
         const std::string & region_,
-        const std::string & token_)
+        const std::string & token_,
+        bool no_sign_)
         : url(url_)
         , access_key_id(access_key_id_)
         , secret_access_key(secret_access_key_)
         , region(region_)
         , token(token_)
+        , no_sign(no_sign_)
         , table_location(getTableLocation(url_))
     {
+        /// Check if user didn't mention any region.
+        /// Same as in S3/Client.cpp (stripping len("https://s3.")).
+        if (url.endpoint.substr(11) == "amazonaws.com")
+            url.addRegionToURI(region);
     }
 
     const std::string & getTableLocation() const override { return table_location; }
@@ -58,27 +69,42 @@ class S3KernelHelper final : public IKernelHelper
 
         /// Supported options
         /// https://github.com/apache/arrow-rs/blob/main/object_store/src/aws/builder.rs#L191
-        set_option("aws_access_key_id", access_key_id);
-        set_option("aws_secret_access_key", secret_access_key);
+        if (!access_key_id.empty())
+            set_option("aws_access_key_id", access_key_id);
+        if (!secret_access_key.empty())
+            set_option("aws_secret_access_key", secret_access_key);
+
+        /// Set even if token is empty to prevent delta-kernel
+        /// from trying to access token api.
         set_option("aws_token", token);
+
+        if (no_sign || (access_key_id.empty() && secret_access_key.empty()))
+            set_option("aws_skip_signature", "true");
+
         set_option("aws_region", region);
+        set_option("aws_bucket", url.bucket);
 
         if (url.uri_str.starts_with("http"))
         {
             set_option("allow_http", "true");
             set_option("aws_endpoint", url.endpoint);
         }
 
-        LOG_TRACE(getLogger("KernelHelper"), "Using region: {}, endpoint: {}, uri: {}", region, url.endpoint, url.uri_str);
+        LOG_TRACE(
+            getLogger("KernelHelper"),
+            "Using endpoint: {}, uri: {}, region: {}, bucket: {}",
+            url.endpoint, url.uri_str, region, url.bucket);
+
         return builder;
     }
 
 private:
-    const DB::S3::URI url;
+    DB::S3::URI url;
     const std::string access_key_id;
     const std::string secret_access_key;
     const std::string region;
     const std::string token;
+    const bool no_sign;
 
     const std::string table_location;
 
@@ -109,6 +135,7 @@ DeltaLake::KernelHelperPtr getKernelHelper(
         case DB::ObjectStorageType::S3:
         {
             const auto * s3_conf = dynamic_cast<const DB::StorageS3Configuration *>(configuration.get());
+            const auto & auth_settings = s3_conf->getAuthSettings();
             const auto & s3_client = object_storage->getS3StorageClient();
             const auto & s3_credentials = s3_client->getCredentials();
             const auto & url = s3_conf->getURL();
@@ -118,7 +145,8 @@ DeltaLake::KernelHelperPtr getKernelHelper(
                 s3_credentials.GetAWSAccessKeyId(),
                 s3_credentials.GetAWSSecretKey(),
                 s3_client->getRegionForBucket(url.bucket),
-                s3_credentials.GetSessionToken());
+                s3_credentials.GetSessionToken(),
+                auth_settings[S3AuthSetting::no_sign_request]);
         }
         default:
         {

src/Storages/ObjectStorage/DataLakes/DeltaLake/TableSnapshot.cpp

@@ -385,10 +385,10 @@ void TableSnapshot::initSnapshotImpl() const
     LOG_TRACE(log, "Initialized scan state");
 
     std::tie(table_schema, physical_names_map) = getTableSchemaFromSnapshot(snapshot.get());
-    LOG_TRACE(log, "Table schema: {}", fmt::join(table_schema.getNames(), ", "));
+    LOG_TRACE(log, "Table logical schema: {}", fmt::join(table_schema.getNames(), ", "));
 
     read_schema = getReadSchemaFromSnapshot(scan_state.get());
-    LOG_TRACE(log, "Read schema: {}", fmt::join(read_schema.getNames(), ", "));
+    LOG_TRACE(log, "Table read schema: {}", fmt::join(read_schema.getNames(), ", "));
 
     partition_columns = getPartitionColumnsFromSnapshot(scan_state.get());
     LOG_TRACE(log, "Partition columns: {}", fmt::join(partition_columns, ", "));

tests/queries/0_stateless/03441_deltalake_clickhouse_public_datasets.reference

@@ -0,0 +1 @@
+99997497

tests/queries/0_stateless/03441_deltalake_clickhouse_public_datasets.sql

@@ -0,0 +1,6 @@
+-- Tags: no-fasttest, no-msan
+-- Tag no-fasttest: Depends on AWS
+-- Tag no-msan: delta-kernel is not built with msan
+
+SELECT count()
+FROM deltaLake('https://clickhouse-public-datasets.s3.amazonaws.com/delta_lake/hits/', SETTINGS allow_experimental_delta_kernel_rs = 1);