try to fix bugs

Author: zvonand

src/Storages/ObjectStorage/Utils.cpp

@@ -16,6 +16,9 @@
 #if USE_AWS_S3
 #include <Disks/ObjectStorages/S3/S3ObjectStorage.h>
 #endif
+#if USE_HDFS
+#include <Disks/ObjectStorages/HDFS/HDFSObjectStorage.h>
+#endif
 
 
 namespace DB
@@ -315,7 +318,17 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
     {
         try
         {
-            S3::URI s3_uri(path);
+            // Normalize s3a:// and s3n:// to s3:// for S3::URI parsing
+            std::string normalized_path = path;
+            if (target.scheme == "s3a" || target.scheme == "s3n")
+            {
+                normalized_path = "s3://" + target.authority + "/" + target.key;
+            }
+            S3::URI s3_uri(normalized_path);
+            
+            // Use the key from SchemeAuthorityKey parsing to ensure we have the full path
+            // S3::URI might extract it differently, so we trust SchemeAuthorityKey which already parsed it correctly
+            std::string key_to_use = target.key;
 
             // Check if base storage is S3 and matches the path's bucket/endpoint
             bool use_base_storage = false;
@@ -345,7 +358,13 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
             // Also check if table_location is provided and matches
             if (!use_base_storage && (base_norm == "s3" || base_norm == "https" || base_norm == "http"))
             {
-                S3::URI base_s3_uri(table_location);
+                // Normalize s3a:// and s3n:// in table_location to s3:// for S3::URI parsing
+                std::string normalized_table_location = table_location;
+                if (base.scheme == "s3a" || base.scheme == "s3n")
+                {
+                    normalized_table_location = "s3://" + base.authority + "/" + base.key;
+                }
+                S3::URI base_s3_uri(normalized_table_location);
 
                 // For s3:// URIs, match by bucket only. For explicit http(s):// URIs, require both bucket and endpoint.
                 bool bucket_matches = (s3_uri.bucket == base_s3_uri.bucket);
@@ -359,12 +378,12 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
             }
 
             if (use_base_storage)
-                return {base_storage, s3_uri.key};
+                return {base_storage, key_to_use};
 
             const std::string cache_key = "s3://" + s3_uri.bucket + "@" + (s3_uri.endpoint.empty() ? "amazonaws.com" : s3_uri.endpoint);
 
             if (auto it = secondary_storages.find(cache_key); it != secondary_storages.end())
-                return {it->second, s3_uri.key};
+                return {it->second, key_to_use};
 
             /// TODO: maybe do not invent new configuration. Use old one and clean up later
             Poco::AutoPtr<Poco::Util::MapConfiguration> cfg(new Poco::Util::MapConfiguration);
@@ -410,7 +429,7 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
             DB::ObjectStoragePtr storage = factory.create(cache_key, *cfg, config_prefix, context, /*skip_access_check*/ true);
 
             secondary_storages.emplace(cache_key, storage);
-            return {storage, s3_uri.key};
+            return {storage, key_to_use};
         }
         catch (const Exception &) // NOLINT
         {
@@ -419,6 +438,50 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
     }
     #endif
 
+    // For HDFS URIs, check if we can reuse base storage
+    if (target_norm == "hdfs")
+    {
+        // Check if base storage is HDFS and matches the path's endpoint
+        bool use_base_storage = false;
+        if (base_storage->getType() == ObjectStorageType::HDFS)
+        {
+            #if USE_HDFS
+            if (auto hdfs_storage = std::dynamic_pointer_cast<HDFSObjectStorage>(base_storage))
+            {
+                const std::string base_url = hdfs_storage->getDescription();
+                // Extract endpoint from base URL (hdfs://namenode:port/path -> hdfs://namenode:port)
+                std::string base_endpoint;
+                if (auto pos = base_url.find('/', base_url.find("//") + 2); pos != std::string::npos)
+                    base_endpoint = base_url.substr(0, pos);
+                else
+                    base_endpoint = base_url;
+                
+                // For HDFS, compare endpoints (namenode addresses)
+                std::string target_endpoint = target_norm + "://" + target.authority;
+                
+                if (base_endpoint == target_endpoint)
+                {
+                    use_base_storage = true;
+                }
+            }
+            #endif
+        }
+        
+        // Also check if table_location is provided and matches
+        if (!use_base_storage && base_norm == "hdfs")
+        {
+            std::string base_endpoint = base_norm + "://" + base.authority;
+            std::string target_endpoint = target_norm + "://" + target.authority;
+            
+            if (base_endpoint == target_endpoint)
+            {
+                use_base_storage = true;
+            }
+        }
+        
+        if (use_base_storage)
+            return {base_storage, target.key};
+    }
 
     // Reuse base storage if scheme and authority (bucket) matches
     if (base_norm == target_norm && base.authority == target.authority)