try to fix bugs

Author: zvonand

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

@@ -111,7 +111,7 @@ Plan getPlan(
     IcebergHistory snapshots_info,
     const PersistentTableComponents & persistent_table_components,
     ObjectStoragePtr object_storage,
-    std::map<String, DB::ObjectStoragePtr> secondary_storages,
+    std::map<String, DB::ObjectStoragePtr> & secondary_storages,
     StorageObjectStorageConfigurationPtr configuration,
     ContextPtr context,
     CompressionMethod compression_method)
@@ -229,12 +229,14 @@ void writeDataFiles(
     ObjectStoragePtr object_storage,
     const std::optional<FormatSettings> & format_settings,
     ContextPtr context,
-    StorageObjectStorageConfigurationPtr configuration)
+    StorageObjectStorageConfigurationPtr configuration,
+    const String & table_location,
+    std::map<String, ObjectStoragePtr> & secondary_storages)
 {
     for (auto & [_, data_file] : initial_plan.path_to_data_file)
     {
         auto delete_file_transform = std::make_shared<IcebergBitmapPositionDeleteTransform>(
-            sample_block, data_file->data_object_info, object_storage, format_settings, context);
+            sample_block, data_file->data_object_info, object_storage, format_settings, context, table_location, secondary_storages);
 
         StorageObjectStorage::ObjectInfo object_info(data_file->data_object_info->getPath());
         auto read_buffer = createReadBuffer(object_info, object_storage, context, getLogger("IcebergCompaction"));
@@ -513,7 +515,7 @@ void compactIcebergTable(
     IcebergHistory snapshots_info,
     const PersistentTableComponents & persistent_table_components,
     ObjectStoragePtr object_storage_,
-    std::map<String, DB::ObjectStoragePtr> secondary_storages_,
+    std::map<String, DB::ObjectStoragePtr> & secondary_storages_,
     StorageObjectStorageConfigurationPtr configuration_,
     const std::optional<FormatSettings> & format_settings_,
     SharedHeader sample_block_,
@@ -525,7 +527,7 @@ void compactIcebergTable(
     if (plan.need_optimize)
     {
         auto old_files = getOldFiles(object_storage_, configuration_);
-        writeDataFiles(plan, sample_block_, object_storage_, format_settings_, context_, configuration_);
+        writeDataFiles(plan, sample_block_, object_storage_, format_settings_, context_, configuration_, persistent_table_components.table_location, secondary_storages_);
         writeMetadataFiles(plan, object_storage_, configuration_, context_, sample_block_);
         clearOldFiles(object_storage_, old_files);
     }

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

@@ -15,7 +15,7 @@ void compactIcebergTable(
     IcebergHistory snapshots_info,
     const PersistentTableComponents & persistent_table_components,
     DB::ObjectStoragePtr object_storage_,
-    std::map<String, DB::ObjectStoragePtr> secondary_storages_,
+    std::map<String, DB::ObjectStoragePtr> & secondary_storages_,
     DB::StorageObjectStorageConfigurationPtr configuration_,
     const std::optional<DB::FormatSettings> & format_settings_,
     DB::SharedHeader sample_block_,

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

@@ -83,13 +83,15 @@ std::shared_ptr<ISimpleTransform> IcebergDataObjectInfo::getPositionDeleteTransf
     ObjectStoragePtr object_storage,
     const SharedHeader & header,
     const std::optional<FormatSettings> & format_settings,
-    ContextPtr context_)
+    ContextPtr context_,
+    const String & table_location,
+    std::map<String, ObjectStoragePtr> & secondary_storages)
 {
     IcebergDataObjectInfoPtr self = shared_from_this();
     if (!context_->getSettingsRef()[Setting::use_roaring_bitmap_iceberg_positional_deletes].value)
-        return std::make_shared<IcebergStreamingPositionDeleteTransform>(header, self, object_storage, format_settings, context_);
+        return std::make_shared<IcebergStreamingPositionDeleteTransform>(header, self, object_storage, format_settings, context_, table_location, secondary_storages);
     else
-        return std::make_shared<IcebergBitmapPositionDeleteTransform>(header, self, object_storage, format_settings, context_);
+        return std::make_shared<IcebergBitmapPositionDeleteTransform>(header, self, object_storage, format_settings, context_, table_location, secondary_storages);
 }
 }
 

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

@@ -32,7 +32,9 @@ struct IcebergDataObjectInfo : public PathWithMetadata, std::enable_shared_from_
         ObjectStoragePtr object_storage,
         const SharedHeader & header,
         const std::optional<FormatSettings> & format_settings,
-        ContextPtr context_);
+        ContextPtr context_,
+        const String & table_location,
+        std::map<String, ObjectStoragePtr> & secondary_storages);
 
     void addPositionDeleteObject(Iceberg::ManifestFileEntry position_delete_object)
     {

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

@@ -125,7 +125,6 @@ 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(),
@@ -135,7 +134,7 @@ std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
                 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,
-                empty_secondary_storages);
+                secondary_storages);
             internal_data_index = 0;
         }
         auto files = files_generator(current_manifest_file_content);
@@ -202,7 +201,8 @@ SingleThreadIcebergKeysIterator::SingleThreadIcebergKeysIterator(
     const ActionsDAG * filter_dag_,
     Iceberg::IcebergTableStateSnapshotPtr table_snapshot_,
     Iceberg::IcebergDataSnapshotPtr data_snapshot_,
-    PersistentTableComponents persistent_components_)
+    PersistentTableComponents persistent_components_,
+    std::map<String, ObjectStoragePtr> & secondary_storages_)
     : object_storage(object_storage_)
     , filter_dag(filter_dag_ ? std::make_shared<ActionsDAG>(filter_dag_->clone()) : nullptr)
     , local_context(local_context_)
@@ -224,6 +224,7 @@ SingleThreadIcebergKeysIterator::SingleThreadIcebergKeysIterator(
     , persistent_components(persistent_components_)
     , files_generator(files_generator_)
     , log(getLogger("IcebergIterator"))
+    , secondary_storages(secondary_storages_)
     , manifest_file_content_type(manifest_file_content_type_)
 {
 }
@@ -236,7 +237,8 @@ IcebergIterator::IcebergIterator(
     IDataLakeMetadata::FileProgressCallback callback_,
     Iceberg::IcebergTableStateSnapshotPtr table_snapshot_,
     Iceberg::IcebergDataSnapshotPtr data_snapshot_,
-    PersistentTableComponents persistent_components_)
+    PersistentTableComponents persistent_components_,
+    std::map<String, ObjectStoragePtr> & secondary_storages_)
     : filter_dag(filter_dag_ ? std::make_unique<ActionsDAG>(filter_dag_->clone()) : nullptr)
     , object_storage(std::move(object_storage_))
     , local_context(local_context_)
@@ -249,7 +251,8 @@ IcebergIterator::IcebergIterator(
           filter_dag.get(),
           table_snapshot_,
           data_snapshot_,
-          persistent_components_)
+          persistent_components_,
+          secondary_storages_)
     , deletes_iterator(
           object_storage,
           local_context_,
@@ -265,7 +268,8 @@ IcebergIterator::IcebergIterator(
           filter_dag.get(),
           table_snapshot_,
           data_snapshot_,
-          persistent_components_)
+          persistent_components_,
+          secondary_storages_)
     , blocking_queue(100)
     , producer_task(local_context_->getSchedulePool().createTask(
           "IcebergMetaReaderThread",
@@ -305,6 +309,7 @@ IcebergIterator::IcebergIterator(
     , compression_method(configuration_.lock()->getCompressionMethod())
     , persistent_components(persistent_components_)
     , table_schema_id(table_snapshot_->schema_id)
+    , secondary_storages(secondary_storages_)
 {
     auto delete_file = deletes_iterator.next();
     while (delete_file.has_value())
@@ -331,9 +336,8 @@ ObjectInfoPtr IcebergIterator::next(size_t)
     if (blocking_queue.pop(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);
+            persistent_components.table_location, manifest_file_entry.file_path, object_storage, secondary_storages, local_context);
 
         // Create IcebergDataObjectInfo with resolved storage
         IcebergDataObjectInfoPtr object_info = std::make_shared<IcebergDataObjectInfo>(manifest_file_entry, storage_to_use, resolved_key);

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

@@ -45,7 +45,8 @@ class SingleThreadIcebergKeysIterator
         const ActionsDAG * filter_dag_,
         IcebergTableStateSnapshotPtr table_snapshot_,
         IcebergDataSnapshotPtr data_snapshot_,
-        PersistentTableComponents persistent_components);
+        PersistentTableComponents persistent_components,
+        std::map<String, ObjectStoragePtr> & secondary_storages_);
 
     std::optional<DB::Iceberg::ManifestFileEntry> next();
 
@@ -62,7 +63,7 @@ class SingleThreadIcebergKeysIterator
     PersistentTableComponents persistent_components;
     FilesGenerator files_generator;
     LoggerPtr log;
-
+    std::map<String, ObjectStoragePtr> & secondary_storages;
 
     // By Iceberg design it is difficult to avoid storing position deletes in memory.
     size_t manifest_file_index = 0;
@@ -90,7 +91,8 @@ class IcebergIterator : public IObjectIterator
         IDataLakeMetadata::FileProgressCallback callback_,
         Iceberg::IcebergTableStateSnapshotPtr table_snapshot_,
         Iceberg::IcebergDataSnapshotPtr data_snapshot_,
-        Iceberg::PersistentTableComponents persistent_components_);
+        Iceberg::PersistentTableComponents persistent_components_,
+        std::map<String, ObjectStoragePtr> & secondary_storages_);
 
     ObjectInfoPtr next(size_t) override;
 
@@ -114,6 +116,7 @@ class IcebergIterator : public IObjectIterator
     std::mutex exception_mutex;
     Iceberg::PersistentTableComponents persistent_components;
     Int32 table_schema_id;
+    std::map<String, ObjectStoragePtr> & secondary_storages;
 };
 }
 

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

@@ -635,15 +635,15 @@ std::shared_ptr<NamesAndTypesList> IcebergMetadata::getInitialSchemaByPath(Conte
         : nullptr;
 }
 
-std::shared_ptr<const ActionsDAG> IcebergMetadata::getSchemaTransformer(ContextPtr context_, ObjectInfoPtr object_info) const
+std::shared_ptr<const ActionsDAG> IcebergMetadata::getSchemaTransformer(ContextPtr local_context, ObjectInfoPtr object_info) const
 {
     IcebergDataObjectInfo * iceberg_object_info = dynamic_cast<IcebergDataObjectInfo *>(object_info.get());
     SharedLockGuard lock(mutex);
     if (!iceberg_object_info)
         return nullptr;
     return (iceberg_object_info->underlying_format_read_schema_id != relevant_snapshot_schema_id)
         ? persistent_components.schema_processor->getSchemaTransformationDagByIds(
-            context_,
+            local_context,
             iceberg_object_info->underlying_format_read_schema_id,
             relevant_snapshot_schema_id)
         : nullptr;
@@ -1012,7 +1012,8 @@ ObjectIterator IcebergMetadata::iterate(
         callback,
         table_snapshot,
         relevant_snapshot,
-        persistent_components);
+        persistent_components,
+        secondary_storages);
 }
 
 NamesAndTypesList IcebergMetadata::getTableSchema() const
@@ -1042,7 +1043,7 @@ void IcebergMetadata::addDeleteTransformers(
         LOG_DEBUG(log, "Constructing filter transform for position delete, there are {} delete objects", iceberg_object_info->position_deletes_objects.size());
         builder.addSimpleTransform(
             [&](const SharedHeader & header)
-            { return iceberg_object_info->getPositionDeleteTransformer(object_storage, header, format_settings, local_context); });
+            { return iceberg_object_info->getPositionDeleteTransformer(object_storage, header, format_settings, local_context, persistent_components.table_location, secondary_storages); });
     }
     const auto & delete_files = iceberg_object_info->equality_deletes_objects;
     if (!delete_files.empty())
@@ -1053,9 +1054,14 @@ void IcebergMetadata::addDeleteTransformers(
         {
             /// get header of delete file
             Block delete_file_header;
-            ObjectInfo delete_file_object(delete_file.file_path);
+            // Resolve the delete file path to get the correct storage and key
+            // This handles cases where delete files are outside the table location
+            auto [delete_storage_to_use, resolved_delete_key] = resolveObjectStorageForPath(
+                persistent_components.table_location, delete_file.file_path, object_storage, secondary_storages, local_context);
+            
+            PathWithMetadata delete_file_object(resolved_delete_key, std::nullopt, delete_file.file_path, delete_storage_to_use);
             {
-                auto schema_read_buffer = createReadBuffer(delete_file_object, object_storage, local_context, log);
+                auto schema_read_buffer = createReadBuffer(delete_file_object, delete_storage_to_use, local_context, log);
                 auto schema_reader = FormatFactory::instance().getSchemaReader(delete_file.file_format, *schema_read_buffer, local_context);
                 auto columns_with_names = schema_reader->readSchema();
                 ColumnsWithTypeAndName initial_header_data;

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

@@ -70,15 +70,19 @@ void IcebergPositionDeleteTransform::initializeDeleteSources()
             && position_deletes_object.reference_data_file_path != iceberg_data_path)
             continue;
 
-        auto object_path = position_deletes_object.file_path;
-        auto object_metadata = object_storage->getObjectMetadata(object_path);
-        auto object_info = std::make_shared<ObjectInfo>(object_path, object_metadata);
+        /// Resolve the position delete file path to get the correct storage and key
+        /// This handles cases where delete files are outside the table location
+        auto [delete_storage_to_use, resolved_delete_key] = resolveObjectStorageForPath(
+            table_location, position_deletes_object.file_path, object_storage, secondary_storages, context);
+        
+        auto object_metadata = delete_storage_to_use->getObjectMetadata(resolved_delete_key);
+        PathWithMetadata delete_file_object(resolved_delete_key, object_metadata, position_deletes_object.file_path, delete_storage_to_use);
 
 
         String format = position_deletes_object.file_format;
         Block initial_header;
         {
-            std::unique_ptr<ReadBuffer> read_buf_schema = createReadBuffer(*object_info, object_storage, context, log);
+            std::unique_ptr<ReadBuffer> read_buf_schema = createReadBuffer(delete_file_object, delete_storage_to_use, context, log);
             auto schema_reader = FormatFactory::instance().getSchemaReader(format, *read_buf_schema, context);
             auto columns_with_names = schema_reader->readSchema();
             ColumnsWithTypeAndName initial_header_data;
@@ -89,9 +93,9 @@ void IcebergPositionDeleteTransform::initializeDeleteSources()
             initial_header = Block(initial_header_data);
         }
 
-        CompressionMethod compression_method = chooseCompressionMethod(object_path, "auto");
+        CompressionMethod compression_method = chooseCompressionMethod(resolved_delete_key, "auto");
 
-        delete_read_buffers.push_back(createReadBuffer(*object_info, object_storage, context, log));
+        delete_read_buffers.push_back(createReadBuffer(delete_file_object, delete_storage_to_use, context, log));
 
         auto syntax_result = TreeRewriter(context).analyze(where_ast, initial_header.getNamesAndTypesList());
         ExpressionAnalyzer analyzer(where_ast, syntax_result, context);

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

@@ -27,13 +27,17 @@ class IcebergPositionDeleteTransform : public ISimpleTransform
         IcebergDataObjectInfoPtr iceberg_object_info_,
         ObjectStoragePtr object_storage_,
         const std::optional<FormatSettings> & format_settings_,
-        ContextPtr context_)
+        ContextPtr context_,
+        const String & table_location_,
+        std::map<String, ObjectStoragePtr> & secondary_storages_)
         : ISimpleTransform(header_, header_, false)
         , header(header_)
         , iceberg_object_info(iceberg_object_info_)
         , object_storage(object_storage_)
         , format_settings(format_settings_)
         , context(context_)
+        , table_location(table_location_)
+        , secondary_storages(secondary_storages_)
     {
         initializeDeleteSources();
     }
@@ -52,6 +56,8 @@ class IcebergPositionDeleteTransform : public ISimpleTransform
     const ObjectStoragePtr object_storage;
     const std::optional<FormatSettings> format_settings;
     ContextPtr context;
+    const String table_location;
+    std::map<String, ObjectStoragePtr> & secondary_storages;
 
     /// We need to keep the read buffers alive since the delete_sources depends on them.
     std::vector<std::unique_ptr<ReadBuffer>> delete_read_buffers;
@@ -66,8 +72,10 @@ class IcebergBitmapPositionDeleteTransform : public IcebergPositionDeleteTransfo
         IcebergDataObjectInfoPtr iceberg_object_info_,
         ObjectStoragePtr object_storage_,
         const std::optional<FormatSettings> & format_settings_,
-        ContextPtr context_)
-        : IcebergPositionDeleteTransform(header_, iceberg_object_info_, object_storage_, format_settings_, context_)
+        ContextPtr context_,
+        const String & table_location_,
+        std::map<String, ObjectStoragePtr> & secondary_storages_)
+        : IcebergPositionDeleteTransform(header_, iceberg_object_info_, object_storage_, format_settings_, context_, table_location_, secondary_storages_)
     {
         initialize();
     }
@@ -90,8 +98,10 @@ class IcebergStreamingPositionDeleteTransform : public IcebergPositionDeleteTran
         IcebergDataObjectInfoPtr iceberg_object_info_,
         ObjectStoragePtr object_storage_,
         const std::optional<FormatSettings> & format_settings_,
-        ContextPtr context_)
-        : IcebergPositionDeleteTransform(header_, iceberg_object_info_, object_storage_, format_settings_, context_)
+        ContextPtr context_,
+        const String & table_location_,
+        std::map<String, ObjectStoragePtr> & secondary_storages_)
+        : IcebergPositionDeleteTransform(header_, iceberg_object_info_, object_storage_, format_settings_, context_, table_location_, secondary_storages_)
     {
         initialize();
     }