try to fix deletes

Author: zvonand

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/IcebergMetadata.cpp

@@ -1042,7 +1042,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 +1053,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();
     }