still ugly but seems to be working

Author: zvonand

src/Storages/ObjectStorage/DataLakes/Iceberg/Compaction.cpp

@@ -147,7 +147,7 @@ Plan getPlan(
     std::unordered_map<String, std::shared_ptr<ManifestFilePlan>> manifest_files;
     for (const auto & snapshot : snapshots_info)
     {
-        auto [manifest_list_storage, key_in_storage] = resolveObjectStorageForPath("", snapshot.manifest_list_path, object_storage, secondary_storages, context);
+        auto [manifest_list_storage, key_in_storage] = resolveObjectStorageForPath(persistent_table_components.table_location, snapshot.manifest_list_path, object_storage, secondary_storages, context);
 
         auto manifest_list
             = getManifestList(manifest_list_storage, configuration, persistent_table_components, context, key_in_storage, snapshot.manifest_list_path, log);
@@ -163,10 +163,10 @@ Plan getPlan(
                 persistent_table_components,
                 context,
                 log,
-                manifest_file.manifest_file_path,
                 manifest_file.manifest_file_absolute_path,
                 manifest_file.added_sequence_number,
-                manifest_file.added_snapshot_id);
+                manifest_file.added_snapshot_id,
+                secondary_storages);
 
             if (!manifest_files.contains(manifest_file.manifest_file_absolute_path))
             {

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.cpp

@@ -56,6 +56,29 @@ IcebergDataObjectInfo::IcebergDataObjectInfo(Iceberg::ManifestFileEntry data_man
     }
 }
 
+IcebergDataObjectInfo::IcebergDataObjectInfo(
+    Iceberg::ManifestFileEntry data_manifest_file_entry_,
+    ObjectStoragePtr resolved_storage,
+    const String & resolved_key)
+    : PathWithMetadata(resolved_key, std::nullopt, data_manifest_file_entry_.file_path, resolved_storage)
+    , data_object_file_path_key(data_manifest_file_entry_.file_path_key)
+    , underlying_format_read_schema_id(data_manifest_file_entry_.schema_id)
+    , sequence_number(data_manifest_file_entry_.added_sequence_number)
+{
+    auto toupper = [](String & str)
+    {
+        std::transform(str.begin(), str.end(), str.begin(), ::toupper);
+        return str;
+    };
+    if (!position_deletes_objects.empty() && toupper(data_manifest_file_entry_.file_format) != "PARQUET")
+    {
+        throw Exception(
+            ErrorCodes::NOT_IMPLEMENTED,
+            "Position deletes are only supported for data files of Parquet format in Iceberg, but got {}",
+            data_manifest_file_entry_.file_format);
+    }
+}
+
 std::shared_ptr<ISimpleTransform> IcebergDataObjectInfo::getPositionDeleteTransformer(
     ObjectStoragePtr object_storage,
     const SharedHeader & header,

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDataObjectInfo.h

@@ -21,6 +21,12 @@ struct IcebergDataObjectInfo : public PathWithMetadata, std::enable_shared_from_
     /// It is used to filter position deletes objects by data file path.
     /// It is also used to create a filter for the data object in the position delete transform.
     explicit IcebergDataObjectInfo(Iceberg::ManifestFileEntry data_manifest_file_entry_);
+    
+    /// Constructor with resolved storage and key for files that may be outside table location
+    explicit IcebergDataObjectInfo(
+        Iceberg::ManifestFileEntry data_manifest_file_entry_,
+        ObjectStoragePtr resolved_storage,
+        const String & resolved_key);
 
     std::shared_ptr<ISimpleTransform> getPositionDeleteTransformer(
         ObjectStoragePtr object_storage,

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -42,6 +42,7 @@
 #include <Storages/ObjectStorage/DataLakes/Iceberg/PositionDeleteTransform.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/Snapshot.h>
 #include <Storages/ObjectStorage/DataLakes/Iceberg/Utils.h>
+#include <Storages/ObjectStorage/Utils.h>
 
 #include <Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.h>
 
@@ -124,16 +125,17 @@ std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
                 ++manifest_file_index;
                 continue;
             }
+            std::map<String, ObjectStoragePtr> empty_secondary_storages;
             current_manifest_file_content = Iceberg::getManifestFile(
                 object_storage,
                 configuration.lock(),
                 persistent_components,
                 local_context,
                 log,
-                data_snapshot->manifest_list_entries[manifest_file_index].manifest_file_path,
                 data_snapshot->manifest_list_entries[manifest_file_index].manifest_file_absolute_path,
                 data_snapshot->manifest_list_entries[manifest_file_index].added_sequence_number,
-                data_snapshot->manifest_list_entries[manifest_file_index].added_snapshot_id);
+                data_snapshot->manifest_list_entries[manifest_file_index].added_snapshot_id,
+                empty_secondary_storages);
             internal_data_index = 0;
         }
         auto files = files_generator(current_manifest_file_content);
@@ -237,6 +239,7 @@ IcebergIterator::IcebergIterator(
     PersistentTableComponents persistent_components_)
     : filter_dag(filter_dag_ ? std::make_unique<ActionsDAG>(filter_dag_->clone()) : nullptr)
     , object_storage(std::move(object_storage_))
+    , local_context(local_context_)
     , data_files_iterator(
           object_storage,
           local_context_,
@@ -327,7 +330,13 @@ ObjectInfoPtr IcebergIterator::next(size_t)
     Iceberg::ManifestFileEntry manifest_file_entry;
     if (blocking_queue.pop(manifest_file_entry))
     {
-        IcebergDataObjectInfoPtr object_info = std::make_shared<IcebergDataObjectInfo>(manifest_file_entry);
+        // Resolve the data file path to get the correct storage and key
+        std::map<String, ObjectStoragePtr> empty_secondary_storages;
+        auto [storage_to_use, resolved_key] = resolveObjectStorageForPath(
+            persistent_components.table_location, manifest_file_entry.file_path, object_storage, empty_secondary_storages, local_context);
+        
+        // Create IcebergDataObjectInfo with resolved storage
+        IcebergDataObjectInfoPtr object_info = std::make_shared<IcebergDataObjectInfo>(manifest_file_entry, storage_to_use, resolved_key);
         for (const auto & position_delete : defineDeletesSpan(manifest_file_entry, position_deletes_files, false))
         {
             object_info->addPositionDeleteObject(position_delete);

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.h

@@ -100,6 +100,7 @@ class IcebergIterator : public IObjectIterator
 private:
     std::unique_ptr<ActionsDAG> filter_dag;
     ObjectStoragePtr object_storage;
+    ContextPtr local_context;
     Iceberg::SingleThreadIcebergKeysIterator data_files_iterator;
     Iceberg::SingleThreadIcebergKeysIterator deletes_iterator;
     ConcurrentBoundedQueue<Iceberg::ManifestFileEntry> blocking_queue;

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp

@@ -497,7 +497,7 @@ void IcebergMetadata::updateSnapshot(ContextPtr local_context, Poco::JSON::Objec
 
             auto [partition_key, sorting_key] = extractIcebergKeys(metadata_object);
 
-            auto [storage_to_use, key_in_storage] = resolveObjectStorageForPath("", snapshot->getValue<String>(f_manifest_list), object_storage, secondary_storages, local_context);
+            auto [storage_to_use, key_in_storage] = resolveObjectStorageForPath(persistent_components.table_location, snapshot->getValue<String>(f_manifest_list), object_storage, secondary_storages, local_context);
 
             relevant_snapshot = std::make_shared<IcebergDataSnapshot>(
                 getManifestList(
@@ -914,10 +914,10 @@ std::optional<size_t> IcebergMetadata::totalRows(ContextPtr local_context) const
             persistent_components,
             local_context,
             log,
-            manifest_list_entry.manifest_file_path,
             manifest_list_entry.manifest_file_absolute_path,
             manifest_list_entry.added_sequence_number,
-            manifest_list_entry.added_snapshot_id);
+            manifest_list_entry.added_snapshot_id,
+            secondary_storages);
         auto data_count = manifest_file_ptr->getRowsCountInAllFilesExcludingDeleted(FileContentType::DATA);
         auto position_deletes_count = manifest_file_ptr->getRowsCountInAllFilesExcludingDeleted(FileContentType::POSITION_DELETE);
         if (!data_count.has_value() || !position_deletes_count.has_value())
@@ -955,10 +955,10 @@ std::optional<size_t> IcebergMetadata::totalBytes(ContextPtr local_context) cons
             persistent_components,
             local_context,
             log,
-            manifest_list_entry.manifest_file_path,
             manifest_list_entry.manifest_file_absolute_path,
             manifest_list_entry.added_sequence_number,
-            manifest_list_entry.added_snapshot_id);
+            manifest_list_entry.added_snapshot_id,
+            secondary_storages);
         auto count = manifest_file_ptr->getBytesCountInAllDataFilesExcludingDeleted();
         if (!count.has_value())
             return {};

src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.cpp

@@ -73,10 +73,10 @@ Iceberg::ManifestFilePtr getManifestFile(
     const PersistentTableComponents & persistent_table_components,
     ContextPtr local_context,
     LoggerPtr log,
-    const String & key_in_storage,
     const String & absolute_path,
     Int64 inherited_sequence_number,
-    Int64 inherited_snapshot_id)
+    Int64 inherited_snapshot_id,
+    std::map<String, ObjectStoragePtr> & secondary_storages)
 {
     auto log_level = local_context->getSettingsRef()[Setting::iceberg_metadata_log_level].value;
 
@@ -85,19 +85,23 @@ Iceberg::ManifestFilePtr getManifestFile(
 
     auto create_fn = [&, use_iceberg_metadata_cache]()
     {
-        PathWithMetadata manifest_object_info(key_in_storage, std::nullopt, absolute_path, object_storage);
+        // Resolve the absolute path to get the correct storage and key_in_storage
+        auto [storage_to_use, resolved_key_in_storage] = resolveObjectStorageForPath(
+            persistent_table_components.table_location, absolute_path, object_storage, secondary_storages, local_context);
+        
+        PathWithMetadata manifest_object_info(resolved_key_in_storage, std::nullopt, absolute_path, storage_to_use);
 
         auto read_settings = local_context->getReadSettings();
         /// Do not utilize filesystem cache if more precise cache enabled
         if (use_iceberg_metadata_cache)
             read_settings.enable_filesystem_cache = false;
 
-        auto buffer = createReadBuffer(manifest_object_info, object_storage, local_context, log, read_settings);
-        Iceberg::AvroForIcebergDeserializer manifest_file_deserializer(std::move(buffer), key_in_storage, getFormatSettings(local_context));
+        auto buffer = createReadBuffer(manifest_object_info, storage_to_use, local_context, log, read_settings);
+        Iceberg::AvroForIcebergDeserializer manifest_file_deserializer(std::move(buffer), resolved_key_in_storage, getFormatSettings(local_context));
 
         return std::make_shared<Iceberg::ManifestFileContent>(
             manifest_file_deserializer,
-            key_in_storage,
+            resolved_key_in_storage,
             persistent_table_components.format_version,
             configuration->getPathForRead().path,
             *persistent_table_components.schema_processor,

src/Storages/ObjectStorage/DataLakes/Iceberg/StatelessMetadataFileGetter.h

@@ -29,10 +29,10 @@ Iceberg::ManifestFilePtr getManifestFile(
     const PersistentTableComponents & persistent_table_components,
     ContextPtr local_context,
     LoggerPtr log,
-    const String & key_in_storage,
     const String & absolute_path,
     Int64 inherited_sequence_number,
-    Int64 inherited_snapshot_id);
+    Int64 inherited_snapshot_id,
+    std::map<String, ObjectStoragePtr> & secondary_storages);
 
 
 ManifestFileCacheKeys getManifestList(

src/Storages/ObjectStorage/StorageObjectStorageSource.cpp

@@ -1282,7 +1282,10 @@ StorageObjectStorageSource::ReadTaskIterator::ReadTaskIterator(
             {
                 auto [storage_to_use, key] = resolveObjectStorageForPath("", object->getAbsolutePath().value(), object_storage, secondary_storages, getContext());
                 if (!key.empty())
+                {
                     object->object_storage_to_use = storage_to_use;
+                    object->relative_path = key;
+                }
             }
             buffer.push_back(object);
         }
@@ -1308,7 +1311,11 @@ StorageObjectStorage::ObjectInfoPtr StorageObjectStorageSource::ReadTaskIterator
 
         object_info = raw->getObjectInfo();
 
-        if (raw->absolute_path.has_value())
+        if (object_info->getObjectStorage().has_value())
+        {
+
+        }
+        else if (raw->absolute_path.has_value())
         {
             auto [storage_to_use, key]
                 = resolveObjectStorageForPath("", raw->absolute_path.value(), object_storage, secondary_storages, getContext());

src/Storages/ObjectStorage/Utils.cpp

@@ -12,6 +12,9 @@
 #include <Disks/ObjectStorages/ObjectStorageFactory.h>
 #include <Poco/Util/MapConfiguration.h>
 #include <IO/S3/URI.h>
+#if USE_AWS_S3
+#include <Disks/ObjectStorages/S3/S3ObjectStorage.h>
+#endif
 
 
 namespace DB
@@ -54,6 +57,7 @@ static std::string factoryTypeForScheme(const std::string & normalized_scheme)
 
 }
 
+// TODO: handle https://s3.amazonaws.com/bucketname/... properly
 SchemeAuthorityKey::SchemeAuthorityKey(const std::string & uri)
 {
     if (uri.empty())
@@ -304,34 +308,65 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
     const std::string base_norm = normalizeSchema(base.scheme);
     const std::string target_norm = normalizeSchema(target.scheme);
 
-    // For S3 URIs, use S3::URI to properly URLs: https://s3.amazonaws.com/.....
+    // For S3 URIs, use S3::URI to properly handle all kinds of URIs, e.g. https://s3.amazonaws.com/bucket/... == s3://bucket/...
     #if USE_AWS_S3
     if (target_norm == "s3" || target_norm == "https" || target_norm == "http")
     {
         try
         {
             S3::URI s3_uri(path);
 
-            if (base_norm == "s3" || base_norm == "https" || base_norm == "http")
+            // Check if base storage is S3 and matches the path's bucket/endpoint
+            bool use_base_storage = false;
+            if (base_storage->getType() == ObjectStorageType::S3)
+            {
+                #if USE_AWS_S3
+                if (auto s3_storage = std::dynamic_pointer_cast<S3ObjectStorage>(base_storage))
+                {
+                    const std::string base_bucket = s3_storage->getObjectsNamespace();
+                    const std::string base_endpoint = s3_storage->getDescription();
+                    
+                    // If bucket matches, use base storage. For s3:// URIs (which don't have explicit endpoint),
+                    // we should use base storage if bucket matches, regardless of endpoint.
+                    // For explicit http(s):// URIs, we require both bucket and endpoint to match.
+                    bool bucket_matches = (s3_uri.bucket == base_bucket);
+                    bool endpoint_matches = (s3_uri.endpoint == base_endpoint);
+                    bool is_generic_s3_uri = (target_norm == "s3");
+                    
+                    if (bucket_matches && (endpoint_matches || is_generic_s3_uri))
+                    {
+                        use_base_storage = true;
+                    }
+                }
+                #endif
+            }
+            
+            // 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);
 
-                if (s3_uri.bucket == base_s3_uri.bucket && s3_uri.endpoint == base_s3_uri.endpoint)
+                // 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);
+                bool endpoint_matches = (s3_uri.endpoint == base_s3_uri.endpoint);
+                bool is_generic_s3_uri = (target_norm == "s3");
+                
+                if (bucket_matches && (endpoint_matches || is_generic_s3_uri))
                 {
-                    return {base_storage, s3_uri.key};
+                    use_base_storage = true;
                 }
             }
+            
+            if (use_base_storage)
+                return {base_storage, s3_uri.key};
 
             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};
-            }
 
             /// TODO: maybe do not invent new configuration. Use old one and clean up later
             Poco::AutoPtr<Poco::Util::MapConfiguration> cfg(new Poco::Util::MapConfiguration);
-
             const std::string config_prefix = "object_storages." + cache_key;
 
             cfg->setString(config_prefix + ".object_storage_type", "s3");
@@ -342,14 +377,41 @@ std::pair<DB::ObjectStoragePtr, std::string> resolveObjectStorageForPath(
                 : s3_uri.endpoint;
             cfg->setString(config_prefix + ".endpoint", endpoint);
 
+            // Copy credentials from base storage if it's also S3
+            if (base_storage->getType() == ObjectStorageType::S3)
+            {
+                #if USE_AWS_S3
+                if (auto s3_storage = std::dynamic_pointer_cast<S3ObjectStorage>(base_storage))
+                {
+                    if (auto s3_client = s3_storage->tryGetS3StorageClient())
+                    {
+                        const auto credentials = s3_client->getCredentials();
+                        const std::string access_key_id = credentials.GetAWSAccessKeyId();
+                        const std::string secret_access_key = credentials.GetAWSSecretKey();
+                        const std::string session_token = credentials.GetSessionToken();
+                        const std::string region = s3_client->getRegion();
+
+                        if (!access_key_id.empty())
+                            cfg->setString(config_prefix + ".access_key_id", access_key_id);
+                        if (!secret_access_key.empty())
+                            cfg->setString(config_prefix + ".secret_access_key", secret_access_key);
+                        if (!session_token.empty())
+                            cfg->setString(config_prefix + ".session_token", session_token);
+                        if (!region.empty())
+                            cfg->setString(config_prefix + ".region", region);
+                    }
+                }
+                #endif
+            }
+
             auto & factory = DB::ObjectStorageFactory::instance();
 
             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};
         }
-        catch (...)
+        catch (const Exception &) // NOLINT
         {
             // If S3::URI parsing fails, fall back to the old logic
         }