Antalya-26:6 - iceberg v3: rewriteFiles - Added logic to merge small files into bins

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()

src/Core/Settings.cpp

@@ -8386,6 +8386,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.

src/Core/SettingsChangesHistory.cpp

@@ -39,6 +39,12 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         /// controls new feature and it's 'true' by default, use 'false' as previous_value).
         /// It's used to implement `compatibility` setting (see https://github.com/ClickHouse/ClickHouse/issues/35972)
         /// Note: please check if the key already exists to prevent duplicate entries.
+        addSettingsChanges(settings_changes_history, "26.6.8.20003.altinityantalya",
+        {
+            {"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.4.20001.altinityantalya",
         {
             {"input_format_parquet_footer_read_size", 65536, 0, "New setting to override the adaptive Parquet footer initial read size with a fixed number of bytes; 0 keeps the adaptive behavior. Before this setting the reader always read a fixed 64 KiB tail."},

src/Storages/ObjectStorage/DataLakes/DataLakeConfiguration.h

@@ -394,10 +394,16 @@ class DataLakeConfiguration : public BaseStorageConfiguration, public std::enabl
 #endif
     }
 
-    bool optimize(ObjectStoragePtr object_storage, const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional<FormatSettings> & format_settings) override
+    bool optimize(
+        ObjectStoragePtr object_storage,
+        const StorageMetadataPtr & metadata_snapshot,
+        ContextPtr context,
+        const std::optional<FormatSettings> & format_settings,
+        std::shared_ptr<DataLake::ICatalog> catalog,
+        const StorageID & storage_id) override
     {
         lazyInitializeIfNeeded(object_storage, context);
-        return getMetadata()->optimize(metadata_snapshot, context, format_settings);
+        return getMetadata()->optimize(metadata_snapshot, context, format_settings, catalog, storage_id);
     }
 
     void addDeleteTransformers(ObjectInfoPtr object_info, QueryPipelineBuilder & builder, const std::optional<FormatSettings> & format_settings, FormatParserSharedResourcesPtr parser_shared_resources, ContextPtr local_context) const override
@@ -867,8 +873,14 @@ class StorageIcebergConfiguration : public StorageObjectStorageConfiguration, pu
     std::shared_ptr<DataLake::ICatalog> getCatalog(ContextPtr context, const StorageID & table_id) const override
         { return getImpl().getCatalog(context, table_id); }
 
-    bool optimize(ObjectStoragePtr object_storage, const StorageMetadataPtr & metadata_snapshot, ContextPtr context, const std::optional<FormatSettings> & format_settings) override
-        { return getImpl().optimize(object_storage, metadata_snapshot, context, format_settings); }
+    bool optimize(
+        ObjectStoragePtr object_storage,
+        const StorageMetadataPtr & metadata_snapshot,
+        ContextPtr context,
+        const std::optional<FormatSettings> & format_settings,
+        std::shared_ptr<DataLake::ICatalog> catalog,
+        const StorageID & storage_id) override
+        { return getImpl().optimize(object_storage, metadata_snapshot, context, format_settings, catalog, storage_id); }
 
     bool supportsPrewhere() const override { return getImpl().supportsPrewhere(); }
 

src/Storages/ObjectStorage/DataLakes/IDataLakeMetadata.h

@@ -242,7 +242,11 @@ class IDataLakeMetadata : boost::noncopyable
         }
 
     virtual bool optimize(
-        const StorageMetadataPtr & /*metadata_snapshot*/, ContextPtr /*context*/, const std::optional<FormatSettings> & /*format_settings*/)
+        const StorageMetadataPtr & /*metadata_snapshot*/,
+        ContextPtr /*context*/,
+        const std::optional<FormatSettings> & /*format_settings*/,
+        std::shared_ptr<DataLake::ICatalog> /*catalog*/,
+        const StorageID & /*storage_id*/)
     {
         return false;
     }