Cleanup

Author: zvonand

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

@@ -177,7 +177,6 @@ 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);
@@ -195,19 +194,18 @@ Plan getPlan(
 
                 IcebergDataObjectInfoPtr data_object_info = std::make_shared<IcebergDataObjectInfo>(data_file, resolved_storage, resolved_key);
                 std::shared_ptr<DataFilePlan> data_file_ptr;
-                std::string storage_identifier = resolved_storage->getDescription() + ":" + resolved_storage->getObjectsNamespace();
-                std::string composite_key = storage_identifier + "|" + resolved_key;
-                if (!plan.path_to_data_file.contains(composite_key))
+                std::string path_identifier = resolved_storage->getDescription() + ":" + resolved_storage->getObjectsNamespace() + "|" + resolved_key;
+                if (!plan.path_to_data_file.contains(path_identifier))
                 {
                     data_file_ptr = std::make_shared<DataFilePlan>(DataFilePlan{
                         .data_object_info = data_object_info,
                         .manifest_list = manifest_files[manifest_file.manifest_file_absolute_path],
                         .patched_path = plan.generator.generateDataFileName()});
-                    plan.path_to_data_file[composite_key] = data_file_ptr;
+                    plan.path_to_data_file[path_identifier] = data_file_ptr;
                 }
                 else
                 {
-                    data_file_ptr = plan.path_to_data_file[composite_key];
+                    data_file_ptr = plan.path_to_data_file[path_identifier];
                 }
                 plan.partitions[partition_index].push_back(data_file_ptr);
                 plan.snapshot_id_to_data_files[snapshot.snapshot_id].push_back(plan.partitions[partition_index].back());

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

@@ -37,13 +37,22 @@ namespace Setting
 extern const SettingsBool use_roaring_bitmap_iceberg_positional_deletes;
 };
 
+namespace
+{
+String toupper(String & str)
+{
+    std::transform(str.begin(), str.end(), str.begin(), ::toupper);
+    return str;
+}
+}
+
 IcebergDataObjectInfo::IcebergDataObjectInfo(Iceberg::ManifestFileEntry data_manifest_file_entry_)
     : PathWithMetadata(data_manifest_file_entry_.file_path, std::nullopt, data_manifest_file_entry_.file_path_key)
     , 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)
 {
-    if (!position_deletes_objects.empty() && Poco::toUpper(data_manifest_file_entry_.file_format) != "PARQUET")
+    if (!position_deletes_objects.empty() && toupper(data_manifest_file_entry_.file_format) != "PARQUET")
     {
         throw Exception(
             ErrorCodes::NOT_IMPLEMENTED,

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

@@ -22,7 +22,7 @@ struct IcebergDataObjectInfo : public PathWithMetadata, std::enable_shared_from_
     /// 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
+    /// Sometimes data files are located outside the table location and even in a different storage.
     explicit IcebergDataObjectInfo(
         Iceberg::ManifestFileEntry data_manifest_file_entry_,
         ObjectStoragePtr resolved_storage,

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

@@ -135,12 +135,12 @@ std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
                 data_snapshot->manifest_list_entries[manifest_file_index].added_sequence_number,
                 data_snapshot->manifest_list_entries[manifest_file_index].added_snapshot_id,
                 secondary_storages);
-            current_files = files_generator(current_manifest_file_content);
             internal_data_index = 0;
         }
-        while (internal_data_index < current_files.size())
+        auto files = files_generator(current_manifest_file_content);
+        while (internal_data_index < files.size())
         {
-            const auto & manifest_file_entry = current_files[internal_data_index++];
+            const auto & manifest_file_entry = files[internal_data_index++];
             if ((manifest_file_entry.schema_id != previous_entry_schema) && (use_partition_pruning))
             {
                 previous_entry_schema = manifest_file_entry.schema_id;
@@ -175,7 +175,6 @@ std::optional<ManifestFileEntry> SingleThreadIcebergKeysIterator::next()
             }
         }
         current_manifest_file_content = nullptr;
-        current_files.clear();
         current_pruner = std::nullopt;
         ++manifest_file_index;
         internal_data_index = 0;
@@ -340,16 +339,14 @@ ObjectInfoPtr IcebergIterator::next(size_t)
         auto [storage_to_use, resolved_key] = resolveObjectStorageForPath(
             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);
+        
         for (const auto & position_delete : defineDeletesSpan(manifest_file_entry, position_deletes_files, false))
-        {
             object_info->addPositionDeleteObject(position_delete);
-        }
+
         for (const auto & equality_delete : defineDeletesSpan(manifest_file_entry, equality_deletes_files, true))
-        {
             object_info->addEqualityDeleteObject(equality_delete);
-        }
+        
         object_info->setFileMetaInfo(std::make_shared<DataFileMetaInfo>(
                                     *persistent_components.schema_processor,
                                     table_schema_id, /// current schema id to use current column names

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

@@ -69,7 +69,6 @@ class SingleThreadIcebergKeysIterator
     size_t manifest_file_index = 0;
     size_t internal_data_index = 0;
     Iceberg::ManifestFilePtr current_manifest_file_content;
-    std::vector<ManifestFileEntry> current_files;
     Int32 previous_entry_schema = -1;
     std::optional<Iceberg::ManifestFilesPruner> current_pruner;
 

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

@@ -1063,8 +1063,7 @@ void IcebergMetadata::addDeleteTransformers(
         {
             /// get header of delete file
             Block delete_file_header;
-            // 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);
 

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

@@ -74,7 +74,7 @@ struct IcebergMetadataFilesCacheCell : private boost::noncopyable
          size_t total_size = 0;
          for (const auto & entry: manifest_file_cache_keys)
          {
-             total_size += sizeof(ManifestFileCacheKey) + entry.manifest_file_absolute_path.capacity();
+             total_size += sizeof(ManifestFileCacheKey) + entry.manifest_file_path.capacity() + entry.manifest_file_absolute_path.capacity();
          }
          return total_size;
     }

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

@@ -85,7 +85,6 @@ Iceberg::ManifestFilePtr getManifestFile(
 
     auto create_fn = [&, use_iceberg_metadata_cache]()
     {
-        // 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);
 

src/Storages/ObjectStorage/StorageObjectStorageSource.h

@@ -179,7 +179,7 @@ class StorageObjectStorageSource::ReadTaskIterator : public IObjectIterator, pri
     std::atomic_size_t index = 0;
     bool is_archive;
     ObjectStoragePtr object_storage;
-    mutable std::map<String, ObjectStoragePtr> secondary_storages; // Sometimes data can be located on a different storage
+    std::map<String, ObjectStoragePtr> secondary_storages; // Sometimes data can be located on a different storage
     /// path_to_archive -> archive reader.
     std::unordered_map<std::string, std::shared_ptr<IArchiveReader>> archive_readers;
     std::mutex archive_readers_mutex;