proper handling of paths in snapshot

Author: zvonand

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

@@ -69,6 +69,7 @@ struct Plan
     IcebergHistory history;
     std::unordered_map<String, Int64> manifest_file_to_first_snapshot;
     std::unordered_map<String, std::vector<String>> manifest_list_to_manifest_files;
+    std::unordered_map<String, std::unordered_set<Int64>> manifest_file_to_snapshots;
     std::unordered_map<Int64, std::vector<std::shared_ptr<DataFilePlan>>> snapshot_id_to_data_files;
     std::unordered_map<String, std::shared_ptr<DataFilePlan>> path_to_data_file;
     FileNamesGenerator generator;
@@ -156,7 +157,9 @@ Plan getPlan(
         {
             plan.manifest_list_to_manifest_files[snapshot.manifest_list_absolute_path].push_back(manifest_file.manifest_file_absolute_path);
             if (!plan.manifest_file_to_first_snapshot.contains(manifest_file.manifest_file_absolute_path))
+            {
                 plan.manifest_file_to_first_snapshot[manifest_file.manifest_file_absolute_path] = snapshot.snapshot_id;
+            }
             auto manifest_file_content = getManifestFile(
                 object_storage,
                 configuration,
@@ -174,6 +177,8 @@ Plan getPlan(
                 manifest_files[manifest_file.manifest_file_absolute_path]->path = manifest_file.manifest_file_absolute_path;
             }
             manifest_files[manifest_file.manifest_file_absolute_path]->manifest_lists_path.push_back(snapshot.manifest_list_path);
+            // Track which snapshots this manifest file belongs to
+            plan.manifest_file_to_snapshots[manifest_file.manifest_file_absolute_path].insert(snapshot.snapshot_id);
             auto data_files = manifest_file_content->getFilesWithoutDeleted(FileContentType::DATA);
             auto positional_delete_files = manifest_file_content->getFilesWithoutDeleted(FileContentType::POSITION_DELETE);
             for (const auto & pos_delete_file : positional_delete_files)
@@ -403,6 +408,9 @@ void writeMetadataFiles(
         {
             manifest_entry->patched_path = plan.generator.generateManifestEntryName();
             manifest_file_renamings[manifest_entry->path] = manifest_entry->patched_path.path_in_metadata;
+
+            std::vector<String> unique_data_filenames(data_filenames.begin(), data_filenames.end());
+
             auto buffer_manifest_entry = object_storage->writeObject(
                 StoredObject(manifest_entry->patched_path.path_in_storage),
                 WriteMode::Rewrite,
@@ -420,7 +428,7 @@ void writeMetadataFiles(
                 partition_columns,
                 plan.partition_encoder.getPartitionValue(grouped_by_manifest_files_partitions[manifest_entry]),
                 ChunkPartitioner(fields_from_partition_spec, current_schema, context, sample_block_).getResultTypes(),
-                std::vector(data_filenames.begin(), data_filenames.end()),
+                unique_data_filenames,
                 manifest_entry->statistics,
                 sample_block_,
                 snapshot,
@@ -453,12 +461,21 @@ void writeMetadataFiles(
         auto initial_manifest_entries = plan.manifest_list_to_manifest_files[initial_manifest_list_name];
         auto renamed_manifest_list = manifest_list_renamings[plan.history[i].manifest_list_path];
         std::vector<String> renamed_manifest_entries;
+        std::unordered_set<String> seen_manifest_entries; // Deduplicate manifest entries
         Int32 total_manifest_file_sizes = 0;
         for (const auto & initial_manifest_entry : initial_manifest_entries)
         {
             auto renamed_manifest_entry = manifest_file_renamings[initial_manifest_entry];
             if (!renamed_manifest_entry.empty())
             {
+                auto it = plan.manifest_file_to_snapshots.find(initial_manifest_entry);
+                if (it != plan.manifest_file_to_snapshots.end() && !it->second.contains(plan.history[i].snapshot_id))
+                    continue;
+
+                if (seen_manifest_entries.contains(renamed_manifest_entry))
+                    continue;
+
+                seen_manifest_entries.insert(renamed_manifest_entry);
                 renamed_manifest_entries.push_back(renamed_manifest_entry);
                 total_manifest_file_sizes += manifest_file_sizes[renamed_manifest_entry];
             }