all integration work

Author: zvonand

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

@@ -1085,7 +1085,7 @@ void IcebergMetadata::addDeleteTransformers(
             }
             /// Then we read the content of the delete file.
             auto mutable_columns_for_set = block_for_set.cloneEmptyColumns();
-            std::unique_ptr<ReadBuffer> data_read_buffer = createReadBuffer(delete_file_object, object_storage, local_context, log);
+            std::unique_ptr<ReadBuffer> data_read_buffer = createReadBuffer(delete_file_object, delete_storage_to_use, local_context, log);
             CompressionMethod compression_method = chooseCompressionMethod(delete_file.file_path, "auto");
             auto delete_format = FormatFactory::instance().getInput(
                 delete_file.file_format,

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

@@ -167,15 +167,19 @@ std::optional<DeleteFileWriteResultWithStats> writeDataFiles(
                     Field cur_value;
                     col_data_filename.column->get(i, cur_value);
 
+                    String original_path = cur_value.safeGet<String>();
                     String path_without_namespace;
-                    if (cur_value.safeGet<String>().starts_with(configuration->getNamespace()))
-                        path_without_namespace = cur_value.safeGet<String>().substr(configuration->getNamespace().size());
 
-                    if (!path_without_namespace.starts_with(configuration->getPathForRead().path))
+                    if (original_path.starts_with(configuration->getNamespace()))
+                        path_without_namespace = original_path.substr(configuration->getNamespace().size());
+                    else
+                        path_without_namespace = original_path;
+
+                    if (!path_without_namespace.empty() && !path_without_namespace.starts_with(configuration->getPathForRead().path))
                     {
                         if (path_without_namespace.starts_with('/'))
                             path_without_namespace = path_without_namespace.substr(1);
-                        else
+                        else if (!path_without_namespace.empty())
                             path_without_namespace = "/" + path_without_namespace;
                     }
                     col_data_filename_without_namespaces->insert(path_without_namespace);

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

@@ -67,7 +67,7 @@ void IcebergPositionDeleteTransform::initializeDeleteSources()
     {
         /// Skip position deletes that do not match the data file path.
         if (position_deletes_object.reference_data_file_path.has_value()
-            && position_deletes_object.reference_data_file_path != iceberg_data_path)
+            && position_deletes_object.reference_data_file_path.value() != iceberg_data_path)
             continue;
 
         /// Resolve the position delete file path to get the correct storage and key
@@ -171,10 +171,8 @@ void IcebergBitmapPositionDeleteTransform::initialize()
         while (auto delete_chunk = delete_source->read())
         {
             int position_index = getColumnIndex(delete_source, IcebergPositionDeleteTransform::positions_column_name);
-            int filename_index = getColumnIndex(delete_source, IcebergPositionDeleteTransform::data_file_path_column_name);
 
             auto position_column = delete_chunk.getColumns()[position_index];
-            auto filename_column = delete_chunk.getColumns()[filename_index];
 
             for (size_t i = 0; i < delete_chunk.getNumRows(); ++i)
             {