From 71316548d4780ad7cca4e16d0effd79474875305 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 23 Sep 2026 18:19:10 +0200 Subject: [PATCH 1/5] Added logic to merge small files into bins --- src/Core/Settings.cpp | 10 + src/Core/SettingsChangesHistory.cpp | 2 + .../DataLakes/Iceberg/BinPackRewrite.cpp | 787 ++++++++++++++++++ .../DataLakes/Iceberg/BinPackRewrite.h | 42 + .../DataLakes/Iceberg/IcebergMetadata.cpp | 52 +- .../DataLakes/Iceberg/IcebergWrites.cpp | 11 +- .../DataLakes/Iceberg/IcebergWrites.h | 3 + .../DataLakes/Iceberg/MetadataGenerator.cpp | 115 +++ .../DataLakes/Iceberg/MetadataGenerator.h | 13 + .../Iceberg/tests/gtest_bin_pack_rewrite.cpp | 225 +++++ .../test_bin_pack_rewrite.py | 181 ++++ 11 files changed, 1420 insertions(+), 21 deletions(-) create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.cpp create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h create mode 100644 src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp create mode 100644 tests/integration/test_storage_iceberg_no_spark/test_bin_pack_rewrite.py diff --git a/src/Core/Settings.cpp b/src/Core/Settings.cpp index d0725772d8aa..b3ad58187dce 100644 --- a/src/Core/Settings.cpp +++ b/src/Core/Settings.cpp @@ -8370,6 +8370,16 @@ Allow to explicitly use 'OPTIMIZE' for iceberg tables. Minimum number of manifest files required to trigger manifest-only compaction via OPTIMIZE TABLE ... MANIFEST. If the current number of manifest files is less than or equal to this threshold, compaction is skipped. Requires allow_experimental_iceberg_compaction to be enabled. +)", EXPERIMENTAL) \ + DECLARE(UInt64, iceberg_target_data_file_size_bytes, 536870912, R"( +Target data file size in bytes for Iceberg bin-packing compaction via OPTIMIZE TABLE. +Small files are merged until the result approaches this size. Default is 512 MiB. +Requires allow_experimental_iceberg_compaction to be enabled. +)", EXPERIMENTAL) \ + DECLARE(UInt64, iceberg_min_data_file_size_bytes, 402653184, R"( +Data files smaller than this threshold are candidates for bin-packing compaction via OPTIMIZE TABLE. +Default is 384 MiB (75% of iceberg_target_data_file_size_bytes). +Requires allow_experimental_iceberg_compaction to be enabled. )", EXPERIMENTAL) \ DECLARE(Bool, allow_iceberg_remove_orphan_files, false, R"( Allow to use 'ALTER TABLE ... EXECUTE remove_orphan_files()' for iceberg tables. diff --git a/src/Core/SettingsChangesHistory.cpp b/src/Core/SettingsChangesHistory.cpp index f093c4054cf6..79870cb4498f 100644 --- a/src/Core/SettingsChangesHistory.cpp +++ b/src/Core/SettingsChangesHistory.cpp @@ -42,6 +42,8 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory() addSettingsChanges(settings_changes_history, "26.6.2.20001.altinityantalya", { {"use_puffin_files_cache", false, true, "Enables cache of parsed Puffin file content such as deletion vectors."}, + {"iceberg_target_data_file_size_bytes", 536870912, 536870912, "Target file size for Iceberg bin-packing compaction (default 512 MiB)."}, + {"iceberg_min_data_file_size_bytes", 402653184, 402653184, "Files below this size are candidates for Iceberg bin-packing compaction (default 384 MiB)."}, }); addSettingsChanges(settings_changes_history, "26.6", diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.cpp new file mode 100644 index 000000000000..fe1ea5b4c797 --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.cpp @@ -0,0 +1,787 @@ +#include + +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#if USE_AVRO + +namespace DB::ErrorCodes +{ + extern const int BAD_ARGUMENTS; + extern const int LOGICAL_ERROR; + extern const int ICEBERG_SPECIFICATION_VIOLATION; +} + +namespace DB::Setting +{ + extern const SettingsUInt64 iceberg_target_data_file_size_bytes; + extern const SettingsUInt64 iceberg_min_data_file_size_bytes; +} + +namespace DB::DataLakeStorageSetting +{ + extern const DataLakeStorageSettingsBool iceberg_use_version_hint; +} + +namespace DB::Iceberg +{ + +namespace +{ + +/// A single small data file selected for bin-packing. +struct SmallFileEntry +{ + IcebergPathFromMetadata file_path; + Int64 record_count; + Int64 file_size_in_bytes; + String file_format; + Row partition_key; + std::optional sort_order_id; + /// Lineage from the source manifest entry. + std::optional snapshot_id; + std::optional sequence_number; + std::optional file_sequence_number; + /// Per-column statistics from the source manifest. + DataFileColumnStatistics column_stats; +}; + +/// A bin: a group of small files from the same partition to be merged. +struct Bin +{ + Row partition_key; + std::vector files; + Int64 total_bytes = 0; + Int64 total_records = 0; +}; + +/// Key for grouping files by partition. +struct PartitionKeyHash +{ + std::hash hasher; + size_t operator()(const Row & row) const + { + size_t result = 0; + FieldVisitorDump dump_visitor; + for (const auto & value : row) + result ^= hasher(applyVisitor(dump_visitor, value)); + return result; + } +}; + +struct PartitionKeyEqual +{ + bool operator()(const Row & a, const Row & b) const + { + if (a.size() != b.size()) + return false; + for (size_t i = 0; i < a.size(); ++i) + if (a[i] != b[i]) + return false; + return true; + } +}; + +/// The plan: which files to rewrite, which manifests to keep. +struct BinPackPlan +{ + /// Bins of small files to merge. + std::vector bins; + /// Manifest paths to carry forward unchanged (delete manifests + data manifests with no small files). + std::unordered_set carry_forward_manifest_paths; + /// Total statistics across removed files. + Int64 removed_data_files = 0; + Int64 removed_records = 0; + Int64 removed_files_size = 0; + /// Number of distinct partitions affected. + Int64 num_partitions = 0; + /// The current snapshot id to use as parent. + Int64 current_snapshot_id = -1; + /// The partition spec to use for new manifests. + Poco::JSON::Object::Ptr partition_spec; + Int64 partition_spec_id = 0; + std::vector partition_columns; + DataTypes partition_types; +}; + + +BinPackPlan buildBinPackPlan( + Poco::JSON::Object::Ptr metadata_object, + const PersistentTableComponents & persistent_table_components, + ObjectStoragePtr object_storage, + SecondaryStorages & secondary_storages, + ContextPtr context, + UInt64 min_file_size, + UInt64 target_file_size) +{ + LoggerPtr log = getLogger("IcebergBinPack::buildPlan"); + BinPackPlan plan; + + if (!metadata_object->has(f_current_snapshot_id)) + return plan; + Int64 current_snapshot_id = metadata_object->getValue(f_current_snapshot_id); + if (current_snapshot_id < 0) + return plan; + plan.current_snapshot_id = current_snapshot_id; + + String current_manifest_list_path; + auto snapshots = metadata_object->get(f_snapshots).extract(); + for (size_t i = 0; i < snapshots->size(); ++i) + { + const auto snapshot = snapshots->getObject(static_cast(i)); + if (snapshot->getValue(f_metadata_snapshot_id) == current_snapshot_id) + { + current_manifest_list_path = snapshot->getValue(f_manifest_list); + break; + } + } + if (current_manifest_list_path.empty()) + return plan; + + auto current_schema_id = metadata_object->getValue(f_current_schema_id); + + /// Resolve partition spec. + auto partition_spec_id = metadata_object->getValue(f_default_spec_id); + auto partitions_specs = metadata_object->getArray(f_partition_specs); + Poco::JSON::Object::Ptr partition_spec; + for (size_t i = 0; i < partitions_specs->size(); ++i) + { + auto candidate = partitions_specs->getObject(static_cast(i)); + if (candidate->getValue(f_spec_id) == partition_spec_id) + { + partition_spec = candidate; + break; + } + } + if (!partition_spec) + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Iceberg metadata does not contain partition spec matching default-spec-id {}", + partition_spec_id); + + plan.partition_spec = partition_spec; + plan.partition_spec_id = partition_spec_id; + + auto spec_fields = partition_spec->getArray(f_fields); + std::vector partition_columns; + for (UInt32 i = 0; i < spec_fields->size(); ++i) + partition_columns.push_back(spec_fields->getObject(i)->getValue(f_name)); + plan.partition_columns = partition_columns; + + /// Resolve partition types. + auto schemas = metadata_object->getArray(f_schemas); + Poco::JSON::Object::Ptr current_schema; + for (size_t i = 0; i < schemas->size(); ++i) + { + if (schemas->getObject(static_cast(i))->getValue(f_schema_id) == current_schema_id) + { + current_schema = schemas->getObject(static_cast(i)); + break; + } + } + if (!current_schema) + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Iceberg metadata does not contain schema matching current-schema-id {}", + current_schema_id); + + /// Build partition types from the partitioner. + for (UInt32 i = 0; i < schemas->size(); ++i) + persistent_table_components.schema_processor->addIcebergTableSchema(schemas->getObject(i), context); + + auto fields_characteristics = persistent_table_components.schema_processor->tryGetFieldsCharacteristics( + static_cast(current_schema_id), {}); + Block spec_sample_block; + for (const auto & nt : fields_characteristics) + spec_sample_block.insert(ColumnWithTypeAndName(nt.type, nt.name)); + auto shared_sample = std::make_shared(std::move(spec_sample_block)); + if (!partition_columns.empty()) + plan.partition_types = ChunkPartitioner(spec_fields, current_schema->getArray(f_fields), context, shared_sample).getResultTypes(); + + /// Scan the current manifest list. + auto manifest_list = getManifestList( + object_storage, persistent_table_components, context, + IcebergPathFromMetadata::deserialize(current_manifest_list_path), + log, secondary_storages); + + /// Collect files per partition. + using PartitionFiles = std::vector; + std::unordered_map partition_files; + /// Track manifest paths with no small files to carry forward. + std::unordered_set manifests_with_only_large_files; + + for (const auto & manifest_file : manifest_list) + { + if (manifest_file.content_type == ManifestFileContentType::DELETE) + { + plan.carry_forward_manifest_paths.insert(manifest_file.manifest_file_path.serialize()); + continue; + } + + auto files_handle = getManifestFileEntriesHandle( + object_storage, persistent_table_components, context, log, + manifest_file, static_cast(current_schema_id), secondary_storages); + + bool has_small_files = false; + for (const auto & data_file : files_handle.getFilesWithoutDeleted(FileContentType::DATA)) + { + const auto & entry = data_file->parsed_entry; + if (static_cast(entry->file_size_in_bytes) < min_file_size) + { + has_small_files = true; + SmallFileEntry small_entry; + small_entry.file_path = entry->file_path_key; + small_entry.record_count = entry->record_count; + small_entry.file_size_in_bytes = entry->file_size_in_bytes; + small_entry.file_format = entry->file_format; + small_entry.partition_key = entry->partition_key_value; + small_entry.sort_order_id = entry->sort_order_id; + small_entry.snapshot_id = entry->parsed_snapshot_id; + if (!small_entry.snapshot_id.has_value()) + small_entry.snapshot_id = manifest_file.added_snapshot_id; + small_entry.sequence_number = entry->parsed_sequence_number; + if (!small_entry.sequence_number.has_value()) + small_entry.sequence_number = manifest_file.added_sequence_number; + small_entry.file_sequence_number = entry->parsed_file_sequence_number; + if (!small_entry.file_sequence_number.has_value()) + small_entry.file_sequence_number = manifest_file.added_sequence_number; + + /// Carry over per-column stats. + for (const auto & [field_id, col_info] : entry->columns_infos) + { + if (col_info.bytes_size.has_value()) + small_entry.column_stats.column_sizes.emplace_back(field_id, *col_info.bytes_size); + if (col_info.rows_count.has_value()) + small_entry.column_stats.value_counts.emplace_back(field_id, *col_info.rows_count); + if (col_info.nulls_count.has_value()) + small_entry.column_stats.null_value_counts.emplace_back(field_id, *col_info.nulls_count); + } + for (const auto & [field_id, bounds] : entry->value_bounds) + { + if (!bounds.first.isNull()) + small_entry.column_stats.lower_bounds.emplace_back(field_id, bounds.first.safeGet()); + if (!bounds.second.isNull()) + small_entry.column_stats.upper_bounds.emplace_back(field_id, bounds.second.safeGet()); + } + + partition_files[entry->partition_key_value].push_back(std::move(small_entry)); + } + } + + if (!has_small_files) + manifests_with_only_large_files.insert(manifest_file.manifest_file_path.serialize()); + } + + /// Carry forward manifests that only have large files. + for (const auto & path : manifests_with_only_large_files) + plan.carry_forward_manifest_paths.insert(path); + + if (partition_files.empty()) + { + LOG_INFO(log, "No small files found below threshold {} bytes; nothing to compact", min_file_size); + return plan; + } + + /// Group small files into bins per partition. + for (auto & [partition_key, files] : partition_files) + { + /// Need at least 2 files to make compaction worthwhile. + if (files.size() < 2) + { + /// Not enough files to merge — the manifest containing these is NOT carried forward as-is + /// because it also holds the small files. The existing compaction logic will write a + /// data manifest for the untouched partition in the new manifest list below. + continue; + } + + Bin current_bin; + current_bin.partition_key = partition_key; + + for (auto & entry : files) + { + if (current_bin.total_bytes + entry.file_size_in_bytes > static_cast(target_file_size) + && !current_bin.files.empty()) + { + plan.bins.push_back(std::move(current_bin)); + current_bin = Bin{}; + current_bin.partition_key = partition_key; + } + + current_bin.total_bytes += entry.file_size_in_bytes; + current_bin.total_records += entry.record_count; + plan.removed_data_files++; + plan.removed_records += entry.record_count; + plan.removed_files_size += entry.file_size_in_bytes; + current_bin.files.push_back(std::move(entry)); + } + + if (current_bin.files.size() >= 2) + plan.bins.push_back(std::move(current_bin)); + else + { + /// Undo the stats for a single-file bin (not worth merging). + for (const auto & f : current_bin.files) + { + plan.removed_data_files--; + plan.removed_records -= f.record_count; + plan.removed_files_size -= f.file_size_in_bytes; + } + } + } + + plan.num_partitions = 0; + { + std::unordered_set affected_partitions; + for (const auto & bin : plan.bins) + affected_partitions.insert(bin.partition_key); + plan.num_partitions = static_cast(affected_partitions.size()); + } + + LOG_INFO(log, "Bin-pack plan: {} bins across {} partitions, {} small files totalling {} bytes", + plan.bins.size(), plan.num_partitions, plan.removed_data_files, plan.removed_files_size); + + return plan; +} + +} // anonymous namespace + + +bool executeBinPackCompaction( + const PersistentTableComponents & persistent_table_components, + ObjectStoragePtr object_storage, + SecondaryStorages & secondary_storages, + const DataLakeStorageSettings & data_lake_settings, + SharedHeader sample_block, + ContextPtr context, + const String & write_format, + std::shared_ptr catalog, + const StorageID & table_id) +{ + LoggerPtr log = getLogger("IcebergBinPack"); + + const auto & settings = context->getSettingsRef(); + UInt64 target_size = settings[Setting::iceberg_target_data_file_size_bytes]; + UInt64 min_size = settings[Setting::iceberg_min_data_file_size_bytes]; + + const auto [metadata_version, metadata_file_path, _] = getLatestOrExplicitMetadataFileAndVersion( + object_storage, + persistent_table_components.table_path, + data_lake_settings, + persistent_table_components.metadata_cache, + context, + log.get(), + persistent_table_components.table_uuid, + persistent_table_components.metadata_compression_method, + /* force_fetch_latest_metadata */ true, + /* ignore_explicit_metadata_file_path */ true); + + auto metadata_object = getMetadataJSONObject( + metadata_file_path, + object_storage, + persistent_table_components.metadata_cache, + context, + log, + persistent_table_components.metadata_compression_method, + persistent_table_components.table_uuid); + + const Int32 format_version = metadata_object->getValue(f_format_version); + if (format_version < 2) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Bin-packing compaction is supported only for Iceberg format_version >= 2"); + + /// Build the plan. + auto plan = buildBinPackPlan( + metadata_object, persistent_table_components, object_storage, + secondary_storages, context, min_size, target_size); + + if (plan.bins.empty()) + { + LOG_INFO(log, "No bins to compact; table is already optimally packed"); + return true; + } + + const auto & path_resolver = persistent_table_components.path_resolver; + CompressionMethod compression_method = persistent_table_components.metadata_compression_method; + + FileNamesGenerator generator( + path_resolver.getTableLocation(), false, compression_method, write_format); + generator.setVersion(metadata_version + 1); + + MetadataGenerator metadata_generator(metadata_object); + + /// Track new files for cleanup on failure. + std::vector new_data_file_paths; + std::vector new_manifest_paths; + IcebergPathFromMetadata manifest_list_path; + + auto cleanup = [&]() + { + for (const auto & p : new_data_file_paths) + { + try { object_storage->removeObjectIfExists(StoredObject(path_resolver.resolve(p))); } + catch (...) { tryLogCurrentException(log, "Cleanup: failed to remove data file"); } + } + for (const auto & p : new_manifest_paths) + { + try { object_storage->removeObjectIfExists(StoredObject(path_resolver.resolve(p))); } + catch (...) { tryLogCurrentException(log, "Cleanup: failed to remove manifest file"); } + } + if (!manifest_list_path.empty()) + { + try { object_storage->removeObjectIfExists(StoredObject(path_resolver.resolve(manifest_list_path))); } + catch (...) { tryLogCurrentException(log, "Cleanup: failed to remove manifest list"); } + } + }; + + try + { + /// Resolve schema for the column mapper. + auto current_schema_id = metadata_object->getValue(f_current_schema_id); + auto schemas = metadata_object->getArray(f_schemas); + Poco::JSON::Object::Ptr current_schema; + for (size_t i = 0; i < schemas->size(); ++i) + { + if (schemas->getObject(static_cast(i))->getValue(f_schema_id) == current_schema_id) + { + current_schema = schemas->getObject(static_cast(i)); + break; + } + } + if (!current_schema) + throw Exception(ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Missing schema for current-schema-id {}", current_schema_id); + + /// Phase 1: Read small files and write merged data files. + /// For each bin, read all source files and write a merged file via MultipleFileWriter. + Int64 total_added_files = 0; + Int64 total_added_records = 0; + Int64 total_added_files_size = 0; + + /// Track info per bin for manifest writing. + struct BinResult + { + Row partition_key; + std::vector merged_file_paths; + std::vector merged_file_row_counts; + std::vector merged_file_byte_counts; + /// The old files that were replaced. + std::vector old_file_paths; + std::vector old_file_row_counts; + std::vector old_file_byte_counts; + std::vector old_file_lineage; + std::vector old_file_stats; + std::vector old_file_formats; + std::vector> old_file_sort_order_ids; + }; + std::vector bin_results; + + for (auto & bin : plan.bins) + { + BinResult result; + result.partition_key = bin.partition_key; + + /// Prepare the MultipleFileWriter for this bin. + MultipleFileWriter writer( + /* max_data_file_num_rows */ 0, /// no row limit; size limit is used + /* max_data_file_num_bytes */ target_size, + current_schema->getArray(f_fields), + generator, + path_resolver, + object_storage, + context, + std::nullopt, /// format_settings + write_format, + sample_block, + [&](const std::string & path) + { + new_data_file_paths.push_back(IcebergPathFromMetadata::deserialize(path)); + }); + + /// Read each source file and feed into the writer. + for (auto & file_entry : bin.files) + { + auto [resolved_storage, resolved_key] = resolveObjectStorageForPath( + persistent_table_components.table_location, + file_entry.file_path.serialize(), + object_storage, secondary_storages, context, + path_resolver); + + RelativePathWithMetadata object_info(resolved_key); + ObjectStoragePtr storage_to_use = resolved_storage ? resolved_storage : object_storage; + auto read_buffer = createReadBuffer(object_info, storage_to_use, context, log); + + auto parser_shared_resources = std::make_shared( + settings, /*num_streams_=*/1); + + auto input_format = FormatFactory::instance().getInput( + file_entry.file_format.empty() ? write_format : file_entry.file_format, + *read_buffer, + *sample_block, + context, + 8192, + std::nullopt, /// format_settings + parser_shared_resources, + std::make_shared(nullptr, context, nullptr, nullptr, nullptr), + true, /// is_remote_fs + CompressionMethod::None, + false); + + while (true) + { + auto chunk = input_format->read(); + if (chunk.empty()) + break; + writer.consume(chunk); + } + + /// Track old file info for the delete manifest. + result.old_file_paths.push_back(file_entry.file_path); + result.old_file_row_counts.push_back(static_cast(file_entry.record_count)); + result.old_file_byte_counts.push_back(static_cast(file_entry.file_size_in_bytes)); + result.old_file_formats.push_back(file_entry.file_format); + result.old_file_sort_order_ids.push_back(file_entry.sort_order_id); + result.old_file_stats.push_back(std::move(file_entry.column_stats)); + + DataFileEntryLineage lineage; + lineage.added_snapshot_id = file_entry.snapshot_id; + lineage.sequence_number = file_entry.sequence_number; + lineage.file_sequence_number = file_entry.file_sequence_number; + lineage.status_override = ManifestEntryStatus::DELETED; + result.old_file_lineage.push_back(lineage); + } + + writer.finalize(); + + result.merged_file_paths = writer.getDataFiles(); + result.merged_file_row_counts = writer.getDataFileRowCounts(); + result.merged_file_byte_counts = writer.getDataFileByteCounts(); + + for (size_t i = 0; i < result.merged_file_paths.size(); ++i) + { + total_added_files++; + total_added_records += static_cast(result.merged_file_row_counts[i]); + total_added_files_size += static_cast(result.merged_file_byte_counts[i]); + } + + bin_results.push_back(std::move(result)); + } + + /// Phase 2: Generate the replace snapshot. + auto generated_metadata_info = generator.generateMetadataPathWithInfo(); + auto snapshot_result = metadata_generator.generateReplaceSnapshot( + generator, + generated_metadata_info.path, + plan.current_snapshot_id, + total_added_files, + total_added_records, + total_added_files_size, + plan.removed_data_files, + plan.removed_records, + plan.removed_files_size, + plan.num_partitions); + + /// Phase 3: Write manifest files. + /// We write two types of manifests: + /// - Delete manifests: DELETED entries for old files (one per bin) + /// - Add manifests: ADDED entries for new merged files (one per bin) + std::vector all_new_manifest_paths; + std::vector all_manifest_sizes; + std::vector all_existing_counts; + std::vector all_entry_partition_spec_ids; + + for (auto & bin_result : bin_results) + { + /// Delete manifest: old files marked DELETED. + { + auto manifest_path = generator.generateManifestEntryName(); + auto storage_path = path_resolver.resolve(manifest_path); + new_manifest_paths.push_back(manifest_path); + + auto buf = object_storage->writeObject( + StoredObject(storage_path), WriteMode::Rewrite, std::nullopt, + DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings()); + + generateManifestFile( + metadata_object, + plan.partition_columns, + bin_result.partition_key, + plan.partition_types, + bin_result.old_file_paths, + bin_result.old_file_row_counts, + bin_result.old_file_byte_counts, + std::nullopt, /// data_file_statistics + sample_block, + snapshot_result.snapshot, + write_format, + plan.partition_spec, + plan.partition_spec_id, + *buf, + FileContentType::DATA, + std::nullopt, /// user_defined_sequence_number + {}, /// per_file_stats + bin_result.old_file_formats, + bin_result.old_file_stats, + bin_result.old_file_sort_order_ids, + bin_result.old_file_lineage); + + buf->finalize(); + Int64 manifest_size = buf->count(); + if (manifest_size == 0) + manifest_size = object_storage->getObjectMetadata(storage_path, false).size_bytes; + + all_new_manifest_paths.push_back(manifest_path); + all_manifest_sizes.push_back(manifest_size); + /// The DELETED manifest has existing (really: deleted) file counts for manifest-list accounting. + Int64 min_seq = std::numeric_limits::max(); + for (const auto & lineage : bin_result.old_file_lineage) + min_seq = std::min(min_seq, lineage.sequence_number.value_or(0)); + all_existing_counts.push_back( + {static_cast(bin_result.old_file_paths.size()), + static_cast(std::accumulate(bin_result.old_file_row_counts.begin(), bin_result.old_file_row_counts.end(), 0UL)), + min_seq}); + all_entry_partition_spec_ids.push_back(plan.partition_spec_id); + } + + /// Add manifest: new merged files. + { + auto manifest_path = generator.generateManifestEntryName(); + auto storage_path = path_resolver.resolve(manifest_path); + new_manifest_paths.push_back(manifest_path); + + auto buf = object_storage->writeObject( + StoredObject(storage_path), WriteMode::Rewrite, std::nullopt, + DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings()); + + generateManifestFile( + metadata_object, + plan.partition_columns, + bin_result.partition_key, + plan.partition_types, + bin_result.merged_file_paths, + bin_result.merged_file_row_counts, + bin_result.merged_file_byte_counts, + std::nullopt, /// data_file_statistics + sample_block, + snapshot_result.snapshot, + write_format, + plan.partition_spec, + plan.partition_spec_id, + *buf, + FileContentType::DATA); + + buf->finalize(); + Int64 manifest_size = buf->count(); + if (manifest_size == 0) + manifest_size = object_storage->getObjectMetadata(storage_path, false).size_bytes; + + all_new_manifest_paths.push_back(manifest_path); + all_manifest_sizes.push_back(manifest_size); + all_existing_counts.push_back({0, 0, 0}); /// New files: no existing counts. + all_entry_partition_spec_ids.push_back(plan.partition_spec_id); + } + } + + /// Phase 4: Write manifest list. + { + auto storage_manifest_list_path = path_resolver.resolve(snapshot_result.manifest_list_path); + manifest_list_path = snapshot_result.manifest_list_path; + + auto buf = object_storage->writeObject( + StoredObject(storage_manifest_list_path), WriteMode::Rewrite, std::nullopt, + DBMS_DEFAULT_BUFFER_SIZE, context->getWriteSettings()); + + generateManifestList( + path_resolver, + metadata_object, + object_storage, + secondary_storages, + context, + all_new_manifest_paths, + snapshot_result.snapshot, + all_manifest_sizes, + *buf, + FileContentType::DATA, + false, /// use_previous_snapshots + {}, /// per_entry_content_types + all_existing_counts, + plan.carry_forward_manifest_paths, + all_entry_partition_spec_ids); + + buf->finalize(); + } + + /// Phase 5: Commit metadata. + { + std::ostringstream oss; // STYLE_CHECK_ALLOW_STD_STRING_STREAM + Poco::JSON::Stringifier::stringify(metadata_object, oss, 4); + std::string json_representation = removeEscapedSlashes(oss.str()); + + auto hint_path = generator.generateVersionHint(); + + const bool catalog_writes_metadata_file = catalog && catalog->isTransactional(); + if (!catalog_writes_metadata_file + && !writeMetadataFileAndVersionHint( + path_resolver, + generated_metadata_info, + json_representation, + hint_path, + object_storage, + context, + data_lake_settings[DataLakeStorageSetting::iceberg_use_version_hint])) + { + LOG_INFO(log, "Bin-pack commit conflict detected, cleaning up"); + cleanup(); + return false; + } + + if (catalog) + { + auto catalog_filename = path_resolver.resolveForCatalog(generated_metadata_info.path); + const auto & [namespace_name, table_name] = DataLake::parseTableName(table_id.getTableName()); + if (!catalog->updateMetadata(namespace_name, table_name, catalog_filename, snapshot_result.snapshot)) + { + LOG_INFO(log, "Bin-pack commit conflict via catalog, cleaning up"); + cleanup(); + return false; + } + } + } + + LOG_INFO(log, "Bin-pack compaction committed: {} bins, {} new files ({} records, {} bytes), " + "{} old files removed ({} records, {} bytes)", + plan.bins.size(), total_added_files, total_added_records, total_added_files_size, + plan.removed_data_files, plan.removed_records, plan.removed_files_size); + return true; + } + catch (...) + { + cleanup(); + throw; + } +} + +} + +#endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h new file mode 100644 index 000000000000..1abc28e5ff7e --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h @@ -0,0 +1,42 @@ +#pragma once + +#include "config.h" + +#if USE_AVRO + +#include +#include +#include +#include +#include + +namespace DataLake +{ +class ICatalog; +} + +namespace DB::Iceberg +{ + +/// Execute bin-packing compaction for an Iceberg table: merge small data files into +/// larger ones, producing a `replace` snapshot that atomically swaps the old files +/// for the merged results. Only data files smaller than `iceberg_min_data_file_size_bytes` +/// are candidates; each bin targets `iceberg_target_data_file_size_bytes`. +/// +/// Leaves all other files, manifests, and snapshot history untouched. +/// +/// Returns true on successful commit, false on commit conflict (caller should retry). +bool executeBinPackCompaction( + const PersistentTableComponents & persistent_table_components, + ObjectStoragePtr object_storage, + SecondaryStorages & secondary_storages, + const DataLakeStorageSettings & data_lake_settings, + SharedHeader sample_block, + ContextPtr context, + const String & write_format, + std::shared_ptr catalog, + const StorageID & table_id); + +} + +#endif diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index 7f7d7211c680..adf54cfb13c5 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -68,6 +68,7 @@ #include #include #include +#include #include #include #include @@ -512,27 +513,42 @@ IcebergMetadata::getIcebergDataSnapshot(Poco::JSON::Object::Ptr metadata_object, bool IcebergMetadata::optimize( const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional & format_settings) { - if (context->getSettingsRef()[Setting::allow_experimental_iceberg_compaction]) - { - const auto sample_block = std::make_shared(metadata_snapshot->getSampleBlock()); - auto snapshots_info = getHistory(context); - compactIcebergTable( - snapshots_info, - persistent_components, - object_storage, - secondary_storages, - data_lake_settings, - format_settings, - sample_block, - context, - write_format); - return true; - } - else - { + if (!context->getSettingsRef()[Setting::allow_experimental_iceberg_compaction]) throw Exception( ErrorCodes::BAD_ARGUMENTS, "Enable 'allow_experimental_iceberg_compaction' setting to call optimize for iceberg tables."); + + static constexpr size_t MAX_BIN_PACK_RETRIES = 100; + + const auto sample_block = std::make_shared(metadata_snapshot->getSampleBlock()); + + for (size_t attempt = 0; attempt < MAX_BIN_PACK_RETRIES; ++attempt) + { + if (attempt > 0) + LOG_INFO(log, "Retrying bin-pack compaction (attempt {}/{})", attempt + 1, MAX_BIN_PACK_RETRIES); + + if (Iceberg::executeBinPackCompaction( + persistent_components, + object_storage, + *secondary_storages, + data_lake_settings, + sample_block, + context, + write_format, + /* catalog */ nullptr, + /* table_id */ StorageID::createEmpty())) + { + if (persistent_components.metadata_cache) + { + persistent_components.metadata_cache->remove(persistent_components.table_path); + if (persistent_components.table_uuid) + persistent_components.metadata_cache->remove(*persistent_components.table_uuid); + } + return true; + } } + + throw Exception(ErrorCodes::LOGICAL_ERROR, + "Bin-pack compaction failed to commit after {} attempts", MAX_BIN_PACK_RETRIES); } bool IcebergMetadata::optimizeManifestFiles( diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp index 0adbf4f2c0e5..9c4fdff05a53 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.cpp @@ -723,9 +723,14 @@ void generateManifestFile( const DataFileEntryLineage * entry_lineage = per_file_entry_lineage.empty() ? nullptr : &per_file_entry_lineage[file_idx]; - manifest.field(Iceberg::f_status) - = avro::GenericDatum(entry_lineage ? static_cast(ManifestEntryStatus::EXISTING) - : static_cast(ManifestEntryStatus::ADDED)); + ManifestEntryStatus entry_status; + if (entry_lineage && entry_lineage->status_override) + entry_status = *entry_lineage->status_override; + else if (entry_lineage) + entry_status = ManifestEntryStatus::EXISTING; + else + entry_status = ManifestEntryStatus::ADDED; + manifest.field(Iceberg::f_status) = avro::GenericDatum(static_cast(entry_status)); Int64 snapshot_id = (entry_lineage && entry_lineage->added_snapshot_id) ? *entry_lineage->added_snapshot_id : new_snapshot->getValue(Iceberg::f_metadata_snapshot_id); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h index a7c6c6a322c5..7c2004c11563 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergWrites.h @@ -61,11 +61,14 @@ struct DataFileColumnStatistics }; /// Per-file manifest-entry lineage (`added_snapshot_id`, data `sequence_number` and `file_sequence_number`) carried over for a manifest-only rewrite. +/// When `status_override` is set, the entry is written with that status instead of the default +/// EXISTING/ADDED logic. Used by bin-packing compaction to produce DELETED entries. struct DataFileEntryLineage { std::optional added_snapshot_id; std::optional sequence_number; std::optional file_sequence_number; + std::optional status_override; }; /// Read a data-file sidecar and return its contents in Iceberg wire format. diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp index 7c232a26b8fe..327fe8347fc6 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.cpp @@ -640,6 +640,121 @@ MetadataGenerator::NextMetadataResult MetadataGenerator::generateManifestOnlySna return {new_snapshot, manifest_list_path}; } +MetadataGenerator::NextMetadataResult MetadataGenerator::generateReplaceSnapshot( + FileNamesGenerator & generator, + const Iceberg::IcebergPathFromMetadata & metadata_file_path, + Int64 parent_snapshot_id, + Int64 added_data_files, + Int64 added_records, + Int64 added_files_size, + Int64 removed_data_files, + Int64 removed_records, + Int64 removed_files_size, + Int64 num_partitions) +{ + int format_version = metadata_object->getValue(Iceberg::f_format_version); + + for (const auto * field : {Iceberg::f_metadata_log, Iceberg::f_snapshot_log}) + if (!metadata_object->has(field)) + metadata_object->set(field, Poco::JSON::Array::Ptr(new Poco::JSON::Array)); + + if (!metadata_object->has(Iceberg::f_snapshots)) + throw Exception( + ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION, + "Metadata has a current snapshot with id {} but no `snapshots` list", + parent_snapshot_id); + + Poco::JSON::Object::Ptr new_snapshot = new Poco::JSON::Object; + if (format_version > 1) + { + auto sequence_number = getMaxSequenceNumber() + 1; + new_snapshot->set(Iceberg::f_metadata_sequence_number, sequence_number); + metadata_object->set(Iceberg::f_last_sequence_number, sequence_number); + } + Int64 snapshot_id = static_cast(dis(gen)); + + auto manifest_list_path = generator.generateManifestListName(snapshot_id, format_version); + new_snapshot->set(Iceberg::f_metadata_snapshot_id, snapshot_id); + new_snapshot->set(Iceberg::f_parent_snapshot_id, parent_snapshot_id); + + auto now = std::chrono::system_clock::now(); + auto ms = duration_cast(now.time_since_epoch()); + Int64 timestamp = ms.count(); + new_snapshot->set(Iceberg::f_timestamp_ms, timestamp); + metadata_object->set(Iceberg::f_last_updated_ms, timestamp); + + auto parent_snapshot = getParentSnapshot(parent_snapshot_id); + + Poco::JSON::Object::Ptr summary = new Poco::JSON::Object; + summary->set(Iceberg::f_operation, Iceberg::f_replace); + summary->set(Iceberg::f_added_data_files, std::to_string(added_data_files)); + summary->set(Iceberg::f_added_records, std::to_string(added_records)); + summary->set(Iceberg::f_added_files_size, std::to_string(added_files_size)); + summary->set(Iceberg::f_deleted_data_files, std::to_string(removed_data_files)); + summary->set(Iceberg::f_removed_data_files, std::to_string(removed_data_files)); + summary->set(Iceberg::f_deleted_records, std::to_string(removed_records)); + summary->set(Iceberg::f_removed_files_size, std::to_string(removed_files_size)); + summary->set(Iceberg::f_changed_partition_count, std::to_string(num_partitions)); + + /// Compute total-* counters: parent totals + added - removed. + setSnapshotTotals( + summary, + parent_snapshot, + /*added_records=*/added_records - removed_records, + /*added_files_size=*/added_files_size - removed_files_size, + /*added_data_files=*/added_data_files - removed_data_files, + /*added_delete_files=*/0, + /*added_position_deletes=*/0, + /*added_equality_deletes=*/0); + new_snapshot->set(Iceberg::f_summary, summary); + + new_snapshot->set(Iceberg::f_schema_id, metadata_object->getValue(Iceberg::f_current_schema_id)); + new_snapshot->set(Iceberg::f_manifest_list, manifest_list_path.serialize()); + + if (format_version >= 3) + { + Int64 next_row_id = metadata_object->has(Iceberg::f_next_row_id) && !metadata_object->isNull(Iceberg::f_next_row_id) + ? metadata_object->getValue(Iceberg::f_next_row_id) + : 0; + new_snapshot->set(Iceberg::f_first_row_id, next_row_id); + new_snapshot->set(Iceberg::f_added_rows, added_records); + metadata_object->set(Iceberg::f_next_row_id, next_row_id + added_records); + } + + getOrCreateArray(metadata_object, Iceberg::f_snapshots)->add(new_snapshot); + metadata_object->set(Iceberg::f_current_snapshot_id, snapshot_id); + + if (!metadata_object->has(Iceberg::f_refs)) + metadata_object->set(Iceberg::f_refs, Poco::JSON::Object::Ptr(new Poco::JSON::Object)); + + if (!metadata_object->getObject(Iceberg::f_refs)->has(Iceberg::f_main)) + { + Poco::JSON::Object::Ptr branch = new Poco::JSON::Object; + branch->set(Iceberg::f_metadata_snapshot_id, snapshot_id); + branch->set(Iceberg::f_type, Iceberg::f_branch); + metadata_object->getObject(Iceberg::f_refs)->set(Iceberg::f_main, branch); + } + else + { + metadata_object->getObject(Iceberg::f_refs)->getObject(Iceberg::f_main)->set(Iceberg::f_metadata_snapshot_id, snapshot_id); + } + + { + Poco::JSON::Object::Ptr new_metadata_item = new Poco::JSON::Object; + new_metadata_item->set(Iceberg::f_metadata_file, metadata_file_path.serialize()); + new_metadata_item->set(Iceberg::f_timestamp_ms, timestamp); + getOrCreateArray(metadata_object, Iceberg::f_metadata_log)->add(new_metadata_item); + } + { + Poco::JSON::Object::Ptr new_snapshot_item = new Poco::JSON::Object; + new_snapshot_item->set(Iceberg::f_metadata_snapshot_id, snapshot_id); + new_snapshot_item->set(Iceberg::f_timestamp_ms, timestamp); + getOrCreateArray(metadata_object, Iceberg::f_snapshot_log)->add(new_snapshot_item); + } + + return {new_snapshot, manifest_list_path}; +} + void MetadataGenerator::generateDropColumnMetadata(const String & column_name) { const auto next_schema_id = getNextSchemaId(metadata_object); diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h index 33ff405a65d7..ccb145a26858 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/MetadataGenerator.h @@ -47,6 +47,19 @@ class MetadataGenerator const Iceberg::IcebergPathFromMetadata & metadata_file_path, Int64 parent_snapshot_id); + /// Create a `replace` snapshot for bin-packing compaction: atomically removes small files and adds merged files. + NextMetadataResult generateReplaceSnapshot( + FileNamesGenerator & generator, + const Iceberg::IcebergPathFromMetadata & metadata_file_path, + Int64 parent_snapshot_id, + Int64 added_data_files, + Int64 added_records, + Int64 added_files_size, + Int64 removed_data_files, + Int64 removed_records, + Int64 removed_files_size, + Int64 num_partitions); + void generateAddColumnMetadata(const String & column_name, DataTypePtr type); void generateDropColumnMetadata(const String & column_name); /// Returns false when the column already has the requested type (no metadata change). diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp new file mode 100644 index 000000000000..484ab231740a --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp @@ -0,0 +1,225 @@ +#include "config.h" + +#if USE_AVRO + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace DB; +using namespace DB::Iceberg; + +namespace +{ + +/// Build minimal metadata suitable for snapshot generation. +Poco::JSON::Object::Ptr makeMetadataForReplace() +{ + auto metadata = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + metadata->set(f_format_version, 2); + metadata->set(f_current_schema_id, 0); + metadata->set(f_last_column_id, 1); + metadata->set(f_default_spec_id, 0); + metadata->set(f_last_sequence_number, Int64(2)); + metadata->set(f_table_uuid, "test-uuid-1234"); + + auto schemas = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto schema = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + schema->set(f_schema_id, 0); + schema->set(f_type, "struct"); + auto fields = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto field = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + field->set(f_id, 1); + field->set(f_name, "x"); + field->set(f_required, true); + field->set(f_type, "int"); + fields->add(field); + schema->set(f_fields, fields); + schemas->add(schema); + metadata->set(f_schemas, schemas); + + /// Partition specs. + auto specs = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto spec = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + spec->set(f_spec_id, 0); + spec->set(f_fields, Poco::JSON::Array::Ptr(new Poco::JSON::Array)); + specs->add(spec); + metadata->set(f_partition_specs, specs); + + /// Create a parent snapshot with known totals. + auto snapshots = Poco::JSON::Array::Ptr(new Poco::JSON::Array); + auto parent_snapshot = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + parent_snapshot->set(f_metadata_snapshot_id, Int64(100)); + parent_snapshot->set(f_timestamp_ms, Int64(1000)); + parent_snapshot->set(f_metadata_sequence_number, Int64(1)); + parent_snapshot->set(f_manifest_list, "s3://bucket/metadata/snap-100-0.avro"); + + auto parent_summary = Poco::JSON::Object::Ptr(new Poco::JSON::Object); + parent_summary->set(f_operation, f_append); + parent_summary->set(f_total_records, "1000"); + parent_summary->set(f_total_files_size, "50000"); + parent_summary->set(f_total_data_files, "10"); + parent_summary->set(f_total_delete_files, "0"); + parent_summary->set(f_total_position_deletes, "0"); + parent_summary->set(f_total_equality_deletes, "0"); + parent_snapshot->set(f_summary, parent_summary); + + snapshots->add(parent_snapshot); + metadata->set(f_snapshots, snapshots); + metadata->set(f_current_snapshot_id, Int64(100)); + + return metadata; +} + +} + + +TEST(IcebergBinPackRewrite, ReplaceSnapshotSummaryCounters) +{ + auto metadata = makeMetadataForReplace(); + MetadataGenerator gen(metadata); + + FileNamesGenerator file_gen("s3://bucket/table/", false, CompressionMethod::None, "Parquet"); + file_gen.setVersion(2); + + auto metadata_path = file_gen.generateMetadataPathWithInfo(); + + auto result = gen.generateReplaceSnapshot( + file_gen, + metadata_path.path, + /*parent_snapshot_id=*/100, + /*added_data_files=*/2, + /*added_records=*/1000, + /*added_files_size=*/40000, + /*removed_data_files=*/8, + /*removed_records=*/800, + /*removed_files_size=*/35000, + /*num_partitions=*/3); + + ASSERT_NE(result.snapshot, nullptr); + + auto summary = result.snapshot->getObject(f_summary); + ASSERT_NE(summary, nullptr); + + /// Operation must be `replace`. + EXPECT_EQ(summary->getValue(f_operation), f_replace); + + /// Added counters. + EXPECT_EQ(summary->getValue(f_added_data_files), "2"); + EXPECT_EQ(summary->getValue(f_added_records), "1000"); + EXPECT_EQ(summary->getValue(f_added_files_size), "40000"); + + /// Removed counters. + EXPECT_EQ(summary->getValue(f_deleted_data_files), "8"); + EXPECT_EQ(summary->getValue(f_removed_data_files), "8"); + EXPECT_EQ(summary->getValue(f_deleted_records), "800"); + EXPECT_EQ(summary->getValue(f_removed_files_size), "35000"); + + /// Partition count. + EXPECT_EQ(summary->getValue(f_changed_partition_count), "3"); + + /// Total-* counters: parent + (added - removed). + /// total_records: 1000 + (1000 - 800) = 1200 + EXPECT_EQ(summary->getValue(f_total_records), "1200"); + /// total_files_size: 50000 + (40000 - 35000) = 55000 + EXPECT_EQ(summary->getValue(f_total_files_size), "55000"); + /// total_data_files: 10 + (2 - 8) = 4 + EXPECT_EQ(summary->getValue(f_total_data_files), "4"); +} + + +TEST(IcebergBinPackRewrite, ReplaceSnapshotSequenceNumberIncremented) +{ + auto metadata = makeMetadataForReplace(); + MetadataGenerator gen(metadata); + + FileNamesGenerator file_gen("s3://bucket/table/", false, CompressionMethod::None, "Parquet"); + file_gen.setVersion(2); + + auto metadata_path = file_gen.generateMetadataPathWithInfo(); + + auto result = gen.generateReplaceSnapshot( + file_gen, + metadata_path.path, + /*parent_snapshot_id=*/100, + /*added_data_files=*/1, + /*added_records=*/500, + /*added_files_size=*/20000, + /*removed_data_files=*/5, + /*removed_records=*/500, + /*removed_files_size=*/25000, + /*num_partitions=*/1); + + /// The new snapshot's sequence number must be > parent's (which was 2). + EXPECT_GT(result.snapshot->getValue(f_metadata_sequence_number), 2); + /// metadata.last-sequence-number must also advance. + EXPECT_GT(metadata->getValue(f_last_sequence_number), 2); +} + + +TEST(IcebergBinPackRewrite, DataFileEntryLineageStatusOverride) +{ + /// When status_override is set, the entry should use that status. + DataFileEntryLineage lineage; + lineage.added_snapshot_id = 42; + lineage.sequence_number = 1; + lineage.file_sequence_number = 1; + lineage.status_override = ManifestEntryStatus::DELETED; + + EXPECT_EQ(lineage.status_override.value(), ManifestEntryStatus::DELETED); +} + + +TEST(IcebergBinPackRewrite, DataFileEntryLineageNoOverrideDefaultsToExisting) +{ + /// When status_override is not set but lineage is present, the entry status + /// should be EXISTING (handled by generateManifestFile logic, but we test the struct). + DataFileEntryLineage lineage; + lineage.added_snapshot_id = 42; + lineage.sequence_number = 1; + lineage.file_sequence_number = 1; + + EXPECT_FALSE(lineage.status_override.has_value()); +} + + +TEST(IcebergBinPackRewrite, SnapshotSummaryReplaceOperationCounters) +{ + /// Build a SnapshotSummary with a replace update and verify totals. + SnapshotSummaryTotals parent_totals{ + .records = 1000, + .files_size = 50000, + .data_files = 10, + .delete_files = 0, + .position_deletes = 0, + .equality_deletes = 0}; + + SnapshotSummary summary( + SnapshotSummaryUpdateReplace{ + .added_files = 3, + .added_records = 800, + .added_files_size = 30000, + .deleted_data_files = 7, + .removed_records = 700, + .removed_files_size = 28000, + .num_partitions = 2}, + parent_totals); + + EXPECT_EQ(summary.getOperation(), SnapshotSummaryOperation::REPLACE); + + auto totals = summary.getTotals(); + /// total_records: 1000 + 800 - 700 = 1100 + EXPECT_EQ(totals.records, 1100); + /// total_files_size: 50000 + 30000 - 28000 = 52000 + EXPECT_EQ(totals.files_size, 52000); + /// total_data_files: 10 + 3 - 7 = 6 + EXPECT_EQ(totals.data_files, 6); +} + +#endif diff --git a/tests/integration/test_storage_iceberg_no_spark/test_bin_pack_rewrite.py b/tests/integration/test_storage_iceberg_no_spark/test_bin_pack_rewrite.py new file mode 100644 index 000000000000..c5472a2b3caa --- /dev/null +++ b/tests/integration/test_storage_iceberg_no_spark/test_bin_pack_rewrite.py @@ -0,0 +1,181 @@ +"""Integration test for Iceberg bin-packing compaction via OPTIMIZE TABLE. + +Creates a table, inserts many small batches to produce many small data files, +runs OPTIMIZE TABLE, and verifies: +- fewer data files after compaction +- same row count and data integrity +- the snapshot has a `replace` operation +""" + +import json +import pytest + +from helpers.iceberg_utils import ( + create_iceberg_table, + get_uuid_str, + default_download_directory, +) + + +def _count_data_files(instance, table_name): + """Return the number of DATA files via system.iceberg_files.""" + result = instance.query( + f"SELECT count() FROM system.iceberg_files " + f"WHERE database = 'default' AND table = '{table_name}' AND content = 'DATA'" + ).strip() + return int(result) + + +def _get_latest_snapshot_summary(instance, table_name): + """Return the summary of the latest snapshot as a dict.""" + raw = instance.query( + f"SELECT toJSONString(summary) " + f"FROM system.iceberg_history " + f"WHERE database = 'default' AND table = '{table_name}' " + f"ORDER BY made_current_at DESC LIMIT 1" + ).strip() + return json.loads(raw) if raw else {} + + +@pytest.mark.parametrize("format_version", [2]) +def test_bin_pack_rewrite(started_cluster_iceberg_no_spark, format_version): + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_bin_pack_rewrite_" + get_uuid_str() + + create_iceberg_table( + "local", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(id Int64, value String)", + format_version=format_version, + ) + + # Insert many small batches — each INSERT produces one data file. + num_batches = 10 + rows_per_batch = 100 + total_rows = num_batches * rows_per_batch + + for batch in range(num_batches): + values = ", ".join( + f"({batch * rows_per_batch + i}, 'row_{batch * rows_per_batch + i}')" + for i in range(rows_per_batch) + ) + instance.query( + f"INSERT INTO {table_name} VALUES {values}", + settings={"allow_insert_into_iceberg": 1}, + ) + + # Verify we have many data files. + files_before = _count_data_files(instance, table_name) + assert files_before == num_batches, ( + f"Expected {num_batches} data files before compaction, got {files_before}" + ) + + # Verify total row count. + count_before = int(instance.query(f"SELECT count() FROM {table_name}").strip()) + assert count_before == total_rows + + # Capture data before compaction for integrity check. + data_before = instance.query( + f"SELECT id, value FROM {table_name} ORDER BY id" + ).strip() + + # Run OPTIMIZE TABLE with tiny thresholds so all files are candidates. + instance.query( + f"OPTIMIZE TABLE {table_name}", + settings={ + "allow_experimental_iceberg_compaction": 1, + "iceberg_target_data_file_size_bytes": 10 * 1024 * 1024, # 10 MB target + "iceberg_min_data_file_size_bytes": 10 * 1024 * 1024, # all files < 10 MB are candidates + }, + ) + + # Drop and recreate the table to pick up the new metadata. + instance.query(f"DROP TABLE IF EXISTS {table_name}") + create_iceberg_table( + "local", + instance, + table_name, + started_cluster_iceberg_no_spark, + ) + + # Verify fewer data files. + files_after = _count_data_files(instance, table_name) + assert files_after < files_before, ( + f"Expected fewer files after compaction: before={files_before}, after={files_after}" + ) + + # Verify same row count. + count_after = int(instance.query(f"SELECT count() FROM {table_name}").strip()) + assert count_after == total_rows, ( + f"Row count mismatch: before={total_rows}, after={count_after}" + ) + + # Verify data integrity. + data_after = instance.query( + f"SELECT id, value FROM {table_name} ORDER BY id" + ).strip() + assert data_after == data_before, "Data mismatch after compaction" + + # Verify the latest snapshot has a `replace` operation. + summary = _get_latest_snapshot_summary(instance, table_name) + assert summary.get("operation") == "replace", ( + f"Expected 'replace' operation in snapshot summary, got: {summary.get('operation')}" + ) + + # Verify summary counters. + assert int(summary.get("added-data-files", 0)) > 0 + assert int(summary.get("deleted-data-files", 0)) == num_batches + + +@pytest.mark.parametrize("format_version", [2]) +def test_bin_pack_noop_when_no_small_files( + started_cluster_iceberg_no_spark, format_version +): + """When there are no small files, OPTIMIZE should be a no-op.""" + instance = started_cluster_iceberg_no_spark.instances["node1"] + table_name = "test_bin_pack_noop_" + get_uuid_str() + + create_iceberg_table( + "local", + instance, + table_name, + started_cluster_iceberg_no_spark, + "(id Int64)", + format_version=format_version, + ) + + # Insert a single batch. + values = ", ".join(f"({i})" for i in range(100)) + instance.query( + f"INSERT INTO {table_name} VALUES {values}", + settings={"allow_insert_into_iceberg": 1}, + ) + + files_before = _count_data_files(instance, table_name) + assert files_before == 1 + + # OPTIMIZE with a very small threshold — but only 1 file, so nothing to merge. + instance.query( + f"OPTIMIZE TABLE {table_name}", + settings={ + "allow_experimental_iceberg_compaction": 1, + "iceberg_target_data_file_size_bytes": 1, + "iceberg_min_data_file_size_bytes": 1024 * 1024 * 1024, # 1 GB — everything is "small" + }, + ) + + # With only 1 file, there's nothing to bin-pack. File count stays. + instance.query(f"DROP TABLE IF EXISTS {table_name}") + create_iceberg_table( + "local", + instance, + table_name, + started_cluster_iceberg_no_spark, + ) + + files_after = _count_data_files(instance, table_name) + assert files_after == files_before, ( + f"Expected same number of files (nothing to compact): before={files_before}, after={files_after}" + ) From 23e7a529b070d8e452bc9eda508fbaafef9b7221 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 23 Sep 2026 21:24:34 +0200 Subject: [PATCH 2/5] Fix compilation error in IcebergMetadata --- .../ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp index adf54cfb13c5..c9a0f61bbaa0 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.cpp @@ -511,7 +511,7 @@ IcebergMetadata::getIcebergDataSnapshot(Poco::JSON::Object::Ptr metadata_object, } bool IcebergMetadata::optimize( - const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional & format_settings) + const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional & /*format_settings*/) { if (!context->getSettingsRef()[Setting::allow_experimental_iceberg_compaction]) throw Exception( From 43c15c4e0f55f59779d446b30ff8fecb002fd24a Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Wed, 23 Sep 2026 23:42:31 +0200 Subject: [PATCH 3/5] Explicitly specify SecondaryStorages in BinPackRewrite.h --- .../ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h index 1abc28e5ff7e..5722ee7849db 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/BinPackRewrite.h @@ -15,6 +15,11 @@ namespace DataLake class ICatalog; } +namespace DB +{ +struct SecondaryStorages; +} + namespace DB::Iceberg { From d10fcddf6dbe0c65faf65f927e7c30eb288f1be3 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 24 Sep 2026 17:27:48 +0200 Subject: [PATCH 4/5] Fix assert in gtest_bin_pack_rewrite --- .../DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp index 484ab231740a..d4dce4c20b95 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/tests/gtest_bin_pack_rewrite.cpp @@ -102,10 +102,10 @@ TEST(IcebergBinPackRewrite, ReplaceSnapshotSummaryCounters) /*removed_files_size=*/35000, /*num_partitions=*/3); - ASSERT_NE(result.snapshot, nullptr); + ASSERT_NE(result.snapshot.get(), nullptr); auto summary = result.snapshot->getObject(f_summary); - ASSERT_NE(summary, nullptr); + ASSERT_NE(summary.get(), nullptr); /// Operation must be `replace`. EXPECT_EQ(summary->getValue(f_operation), f_replace); From 0357f1f8ebfb3b5da20903edcb71340bd1809936 Mon Sep 17 00:00:00 2001 From: Kanthi Subramanian Date: Thu, 24 Sep 2026 19:38:05 +0200 Subject: [PATCH 5/5] Link Avro library to unit_tests_dbms so headers can be included in test files --- src/CMakeLists.txt | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 065d44432d74..e95950afe199 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -956,6 +956,10 @@ if (ENABLE_TESTS) target_link_libraries(unit_tests_dbms PRIVATE ch_contrib::parquet) endif() + if (TARGET ch_contrib::avrocpp) + target_link_libraries(unit_tests_dbms PRIVATE ch_contrib::avrocpp) + endif() + if (TARGET ch_contrib::silk_fibers) target_link_libraries(unit_tests_dbms PRIVATE ch_contrib::silk_fibers) endif()