antalya 26.6: Port of iceberg compaction from upstream

docs/en/engines/table-engines/integrations/iceberg.md

@@ -1,6 +1,6 @@
 ---
-description: 'This engine provides a read-only integration with existing Apache Iceberg
-  tables in Amazon S3, Azure, HDFS and locally stored tables.'
+description: 'This engine provides a read-only data integration with existing Apache Iceberg
+  tables in Amazon S3, Azure, HDFS and locally stored tables, plus experimental metadata-maintenance writes.'
 sidebar_label: 'Iceberg'
 sidebar_position: 90
 slug: /engines/table-engines/integrations/iceberg
@@ -16,7 +16,7 @@ The Iceberg Table Engine is available but may have limitations. ClickHouse wasn'
 For optimal compatibility, we suggest using the Iceberg Table Function while we continue to improve support for the Iceberg Table Engine.
 :::
 
-This engine provides a read-only integration with existing Apache [Iceberg](https://iceberg.apache.org/) tables in Amazon S3, Azure, HDFS and locally stored tables.
+This engine provides a read-only *data* integration with existing Apache [Iceberg](https://iceberg.apache.org/) tables in Amazon S3, Azure, HDFS and locally stored tables.
 
 ## Create table {#create-table}
 
@@ -125,6 +125,23 @@ ClickHouse supports partition pruning during SELECT queries for Iceberg tables,
 
 ClickHouse supports time travel for Iceberg tables, allowing you to query historical data with a specific timestamp or snapshot ID.
 
+## Manifest file compaction {#manifest-compaction}
+
+Over time, frequent writes to an Iceberg table can accumulate a large number of small manifest files in the current snapshot's manifest list. A long manifest list slows down query planning, because every manifest file has to be read to discover the data files. ClickHouse can compact these manifest files into fewer, larger ones using the `OPTIMIZE TABLE ... MANIFEST` statement:
+
+```sql
+OPTIMIZE TABLE example_table MANIFEST SETTINGS allow_experimental_iceberg_compaction = 1;
+```
+
+This produces a new snapshot (a `replace` operation) that references the same data files through a consolidated set of manifest files. No data files are rewritten and no rows are added, deleted, or deduplicated — only the manifest layer is rearranged.
+
+### Requirements and behavior {#manifest-compaction-behavior}
+
+- The feature is experimental and gated behind the `allow_experimental_iceberg_compaction` setting. The statement throws an exception if the setting is not enabled.
+- Compaction is only attempted when the number of manifest files in the current snapshot's manifest list exceeds the threshold given by the `iceberg_manifest_min_count_to_compact` setting (default `30`). If the current count is less than or equal to the threshold, compaction is skipped and no new snapshot is created. Set the threshold lower to compact more eagerly.
+- `OPTIMIZE TABLE ... MANIFEST` is supported only for Iceberg tables. Running it against any other table engine throws an exception.
+- `OPTIMIZE TABLE ... MANIFEST` is supported only for Iceberg format-version 2 tables. Running it against a format-version 1 table throws an exception, and so does running it against a format-version 3 table, because the v3 row-lineage `first_row_id` metadata is not yet round-tripped through the manifest rewrite.
+
 ## Processing of tables with deleted rows {#deleted-rows}
 
 ClickHouse supports reading Iceberg tables that use the following deletion methods:

src/Core/Settings.cpp

@@ -8190,6 +8190,11 @@ Allow to clean up old data files during Iceberg compaction.
 )", EXPERIMENTAL) \
     DECLARE(Bool, allow_experimental_iceberg_compaction, false, R"(
 Allow to explicitly use 'OPTIMIZE' for iceberg tables.
+)", EXPERIMENTAL) \
+    DECLARE(UInt64, iceberg_manifest_min_count_to_compact, 30, R"(
+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(Bool, allow_iceberg_remove_orphan_files, false, R"(
 Allow to use 'ALTER TABLE ... EXECUTE remove_orphan_files()' for iceberg tables.

src/Core/SettingsChangesHistory.cpp

@@ -43,6 +43,7 @@ const VersionToSettingsChangesMap & getSettingsChangesHistory()
         {
             {"analyzer_compatibility_allow_non_aggregate_in_having", false, false, "New compatibility setting. When enabled, the new analyzer mimics the legacy `HAVING`-to-`WHERE` rewrite for non-aggregate AND-conjuncts instead of raising `NOT_AN_AGGREGATE`."},
             {"reserve_memory", 0, 0, "New setting to reserve memory for specific workload before starting a query."},
+            {"iceberg_manifest_min_count_to_compact", 30, 30, "New setting to control manifest compaction for Iceberg tables."},
             {"output_format_image_width", 1024, 1024, "New setting controlling the width of the output image for image output formats such as PNG."},
             {"output_format_image_height", 1024, 1024, "New setting controlling the height of the output image for image output formats such as PNG."},
             {"output_format_image_terminal_mode", "", "", "New setting controlling whether image output formats such as PNG are rendered directly to the terminal using an inline image protocol."},

src/Interpreters/InterpreterOptimizeQuery.cpp

@@ -1,3 +1,5 @@
+#include "config.h"
+
 #include <Storages/IStorage.h>
 #include <Parsers/ASTOptimizeQuery.h>
 #include <Parsers/ASTLiteral.h>
@@ -10,6 +12,11 @@
 #include <Common/typeid_cast.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Storages/MergeTree/MergeTreeData.h>
+#include <Storages/ObjectStorage/StorageObjectStorage.h>
+
+#if USE_AVRO
+#include <Storages/ObjectStorage/DataLakes/Iceberg/IcebergMetadata.h>
+#endif
 
 #include <Interpreters/processColumnTransformers.h>
 
@@ -22,6 +29,7 @@ namespace ErrorCodes
 {
     extern const int BAD_ARGUMENTS;
     extern const int THERE_IS_NO_COLUMN;
+    extern const int NOT_IMPLEMENTED;
 }
 
 
@@ -44,6 +52,28 @@ BlockIO InterpreterOptimizeQuery::execute()
     auto metadata_snapshot = table->getInMemoryMetadataPtr(getContext(), false);
     auto storage_snapshot = table->getStorageSnapshot(metadata_snapshot, getContext());
 
+    /// Handle OPTIMIZE TABLE ... MANIFEST for Iceberg tables
+    if (ast.manifest)
+    {
+        if (ast.final || ast.partition || ast.deduplicate || ast.cleanup || ast.dry_run)
+            throw Exception(ErrorCodes::BAD_ARGUMENTS, "OPTIMIZE MANIFEST is incompatible with FINAL, PARTITION, DEDUPLICATE, CLEANUP, and DRY RUN options");
+
+#if USE_AVRO
+        auto * object_storage_table = dynamic_cast<StorageObjectStorage *>(table.get());
+        if (!object_storage_table)
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "OPTIMIZE MANIFEST is only supported for Iceberg tables");
+
+        auto * iceberg_metadata = dynamic_cast<IcebergMetadata *>(object_storage_table->getExternalMetadata(getContext()));
+        if (!iceberg_metadata)
+            throw Exception(ErrorCodes::NOT_IMPLEMENTED, "OPTIMIZE MANIFEST is only supported for Iceberg tables");
+
+        iceberg_metadata->optimizeManifestFiles(metadata_snapshot, getContext(), object_storage_table->getCatalog(), table_id);
+        return {};
+#else
+        throw Exception(ErrorCodes::NOT_IMPLEMENTED, "OPTIMIZE MANIFEST is only supported for Iceberg tables");
+#endif
+    }
+
     // Empty list of names means we deduplicate by all columns, but user can explicitly state which columns to use.
     Names column_names;
     if (ast.deduplicate_by_columns)

src/Parsers/ASTOptimizeQuery.cpp

@@ -45,6 +45,9 @@ void ASTOptimizeQuery::formatQueryImpl(WriteBuffer & ostr, const FormatSettings
     if (cleanup)
         ostr << " CLEANUP";
 
+    if (manifest)
+        ostr << " MANIFEST";
+
     if (deduplicate_by_columns)
     {
         ostr << " BY ";

src/Parsers/ASTOptimizeQuery.h

@@ -27,10 +27,12 @@ class ASTOptimizeQuery : public ASTQueryWithTableAndOutput, public ASTQueryWithO
     bool dry_run = false;
     /// List of part names for DRY RUN (ASTExpressionList of ASTLiteral strings)
     ASTPtr parts_list;
+    /// Compact manifests only (for Iceberg tables)
+    bool manifest = false;
     /** Get the text that identifies this element. */
     String getID(char delim) const override
     {
-        return "OptimizeQuery" + (delim + getDatabase()) + delim + getTable() + (final ? "_final" : "") + (deduplicate ? "_deduplicate" : "") + (cleanup ? "_cleanup" : "") + (dry_run ? "_dry_run" : "");
+        return "OptimizeQuery" + (delim + getDatabase()) + delim + getTable() + (final ? "_final" : "") + (deduplicate ? "_deduplicate" : "") + (cleanup ? "_cleanup" : "") + (dry_run ? "_dry_run" : "") + (manifest ? "_manifest" : "");
     }
 
     ASTPtr clone() const override

src/Parsers/CommonParsers.h

@@ -325,6 +325,7 @@ namespace DB
     MR_MACROS(LIVE, "LIVE") \
     MR_MACROS(LOCAL, "LOCAL") \
     MR_MACROS(M, "M") \
+    MR_MACROS(MANIFEST, "MANIFEST") \
     MR_MACROS(MASTER_THREAD, "MASTER THREAD") \
     MR_MACROS(MATCH, "MATCH") \
     MR_MACROS(MATERIALIZE_COLUMN, "MATERIALIZE COLUMN") \

src/Parsers/ParserOptimizeQuery.cpp

@@ -33,6 +33,7 @@ bool ParserOptimizeQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expecte
     ParserKeyword s_force(Keyword::FORCE);
     ParserKeyword s_deduplicate(Keyword::DEDUPLICATE);
     ParserKeyword s_cleanup(Keyword::CLEANUP);
+    ParserKeyword s_manifest(Keyword::MANIFEST);
     ParserKeyword s_by(Keyword::BY);
     ParserToken s_dot(TokenType::Dot);
     ParserIdentifier name_p(true);
@@ -46,6 +47,7 @@ bool ParserOptimizeQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expecte
     bool final = false;
     bool deduplicate = false;
     bool cleanup = false;
+    bool manifest = false;
     String cluster_str;
 
     if (!s_optimize_table.ignore(pos, expected))
@@ -90,6 +92,9 @@ bool ParserOptimizeQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expecte
     if (s_cleanup.ignore(pos, expected))
         cleanup = true;
 
+    if (s_manifest.ignore(pos, expected))
+        manifest = true;
+
     ASTPtr deduplicate_by_columns;
     if (deduplicate && s_by.ignore(pos, expected))
     {
@@ -111,6 +116,7 @@ bool ParserOptimizeQuery::parseImpl(Pos & pos, ASTPtr & node, Expected & expecte
     query->deduplicate = deduplicate;
     query->deduplicate_by_columns = deduplicate_by_columns;
     query->cleanup = cleanup;
+    query->manifest = manifest;
     query->database = database;
     query->table = table;
 

src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp

@@ -157,6 +157,16 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
         }
     }
 
+    /// `file_sequence_number` can differ from the data `sequence_number` and, like it, is inherited from the
+    /// manifest's sequence number when null. Keep it raw here; the inherited value is resolved by the caller.
+    std::optional<Int64> file_sequence_number;
+
+    if (format_version > 1 && hasPath(f_file_sequence_number))
+    {
+        const auto file_sequence_number_value = getValueFromRowByName(row_index, f_file_sequence_number);
+        if (!file_sequence_number_value.isNull())
+            file_sequence_number = file_sequence_number_value.safeGet<Int64>();
+    }
 
     const auto file_path_key = IcebergPathFromMetadata::deserialize(
         getValueFromRowByName(row_index, c_data_file_file_path, TypeIndex::String).safeGet<String>());
@@ -251,6 +261,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
                 row_index,
                 status,
                 sequence_number,
+                file_sequence_number,
                 snapshot_id,
                 partition_key_value,
                 columns_infos,
@@ -298,6 +309,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
                 row_index,
                 status,
                 sequence_number,
+                file_sequence_number,
                 snapshot_id,
                 partition_key_value,
                 columns_infos,
@@ -329,6 +341,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
                 row_index,
                 status,
                 sequence_number,
+                file_sequence_number,
                 snapshot_id,
                 partition_key_value,
                 columns_infos,