`OPTIMIZE TABLE ... MANIFEST` can delete a published snapshot and drop manifest statistics

[Selfeer]

Backport of Iceberg manifest compaction (OPTIMIZE TABLE ... MANIFEST) in Altinity/ClickHouse#2193. Format version 2. Compaction runs when the current manifest list has strictly more manifests than iceberg_manifest_min_count_to_compact (default 30).

Commit-failure cleanup deletes manifests after the new snapshot is already visible

Description

After compacted metadata is published, a later catalog failure still runs cleanup, which deletes the new manifest files and the new manifest list. The published snapshot keeps those paths, so later reads fail to plan or scan the table.

Cleanup removes only the rewritten manifests and the new manifest list. It leaves the metadata file and version-hint.text in place, and it does not check whether the catalog commit already succeeded.

Two commit layouts hit this:

• Glue, or any non-transactional catalog, together with a filesystem version hint. ClickHouse writes vN.metadata.json and updates version-hint.text first. If the following catalog update throws or returns false, cleanup deletes the files vN already names. Readers that follow the version hint or the latest metadata file open a broken snapshot. The catalog pointer can still name the previous metadata.
• Transactional REST catalog. ClickHouse skips the storage-side metadata write. RestCatalog::updateMetadata treats HTTP 500, 408, 409, and 429 as a retryable failure and returns false ("commit state unknown"). If the catalog did apply the replace snapshot, cleanup still deletes the manifests that snapshot now names. Compaction then retries, up to 100 times, and each failed attempt deletes the files it just wrote.

Steps to reproduce

Environment: a 26.6 Antalya build that includes PR 2193, format-version 2, allow_experimental_iceberg_compaction = 1, iceberg_manifest_min_count_to_compact = 1. Create the table in Spark with manifest merging disabled so each insert is its own manifest, then attach it in ClickHouse.

CREATE TABLE db.events (id BIGINT, data STRING)
USING iceberg
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

INSERT INTO db.events VALUES (1, 'a');
INSERT INTO db.events VALUES (2, 'b');
INSERT INTO db.events VALUES (3, 'c');

SET allow_experimental_database_iceberg = 1;

CREATE DATABASE lake
ENGINE = DataLakeCatalog('<catalog-url>', '<key>', '<secret>')
SETTINGS catalog_type = 'rest',
         warehouse = 's3://warehouse/',
         storage_endpoint = '<s3-endpoint>',
         auth_header = 'Authorization: Bearer <token>';

REST catalog. Put a proxy in front of the catalog. On the table-update request sent by OPTIMIZE (the commit that adds a replace snapshot), forward it to the catalog, wait until the catalog returns success, then answer ClickHouse with HTTP 500 or 408.

OPTIMIZE TABLE lake.`db.events` MANIFEST
SETTINGS
    allow_experimental_iceberg_compaction = 1,
    iceberg_manifest_min_count_to_compact = 1;

Read the table from Spark and from ClickHouse. Take the manifest-list path and the manifest paths from the current snapshot and check that those objects still exist in object storage.

Glue (or another non-transactional catalog). Use the same table and the same OPTIMIZE, with catalog_type = 'glue'. Make the Glue UpdateTable call fail after ClickHouse has already written the new metadata file (deny glue:UpdateTable, or point the catalog client at an endpoint that returns an error). Then read the table with iceberg_use_version_hint = 1, and open the metadata file named by version-hint.text.

Expected result

The current snapshot stays readable from ClickHouse and from Spark, and returns the same rows as before OPTIMIZE.

Every manifest file and the manifest list named by that snapshot still exist in object storage. A failed or ambiguous catalog response leaves those files in place when the new snapshot, the new metadata file, or version-hint.text is already visible. If the catalog commit did not apply, the previous snapshot stays current and version-hint.text names metadata whose files are still present.

Manifest-list partition bounds are dropped for common partition types

Description

The rewrite fills partitions[*].lower_bound and upper_bound on each compacted manifest-list entry only when canDumpIcebergStats accepts the partition type. That check allows Int32, Int64, dates, times, String, and decimals. dumpFieldToBytes already serializes UInt8, UInt16, UInt32, UInt64, Float32, and Float64, and those types never reach it.

Ordinary ClickHouse partition-transform results fall in the rejected set:

• bucket[N] → UInt32
• year → UInt16
• month → UInt32
• day / hour → UInt32
• identity float / double → Float32 / Float64

After compaction those summaries keep null bounds. Spark, and any reader that prunes from manifest-list partition stats, cannot prune those partitions and opens extra manifests. Row data is unchanged. The rewritten manifest list no longer matches the source manifests.

A NaN float partition hits a second failure on the same path: isNaNPartitionValue reads the value with safeGet<Float64>() for both Float32 and Float64. A NaN float partition throws during the rewrite instead of recording contains_nan.

Steps to reproduce

Same server settings as above. For each table, disable Spark manifest merging, insert at least three times into the same partition values (more manifests than partition groups), attach the table in ClickHouse, and record partition_summaries from db.<table>.manifests before compaction.

-- bucket
CREATE TABLE db.by_bucket (id BIGINT, data STRING) USING iceberg
PARTITIONED BY (bucket(4, id))
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

-- year, month, day, hour: repeat with PARTITIONED BY (year(ts)), (month(ts)), (days(ts)), (hours(ts))
CREATE TABLE db.by_year (id BIGINT, ts TIMESTAMP) USING iceberg
PARTITIONED BY (year(ts))
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

-- identity float and double
CREATE TABLE db.by_float (id BIGINT, x FLOAT) USING iceberg
PARTITIONED BY (x)
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

CREATE TABLE db.by_double (id BIGINT, x DOUBLE) USING iceberg
PARTITIONED BY (x)
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

Insert finite values several times so each table has more data manifests than distinct partition values. Example for the year table:

INSERT INTO db.by_year VALUES (1, TIMESTAMP '2024-01-15 00:00:00');
INSERT INTO db.by_year VALUES (2, TIMESTAMP '2024-06-01 00:00:00');
INSERT INTO db.by_year VALUES (3, TIMESTAMP '2024-12-01 00:00:00');

In Spark, before compaction:

SELECT path, partition_summaries FROM db.by_year.manifests;

In ClickHouse:

OPTIMIZE TABLE lake.`db.by_year` MANIFEST
SETTINGS
    allow_experimental_iceberg_compaction = 1,
    iceberg_manifest_min_count_to_compact = 1;

Repeat for by_bucket, by_float, by_double, and the month, day, and hour tables. Read partition_summaries again from the manifests of the current snapshot.

For the NaN float case, use a separate table partitioned by identity float, and insert NaN on several commits into that same partition:

CREATE TABLE db.by_float_nan (id BIGINT, x FLOAT) USING iceberg
PARTITIONED BY (x)
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

INSERT INTO db.by_float_nan VALUES (1, CAST('NaN' AS FLOAT));
INSERT INTO db.by_float_nan VALUES (2, CAST('NaN' AS FLOAT));
INSERT INTO db.by_float_nan VALUES (3, CAST('NaN' AS FLOAT));

Run OPTIMIZE TABLE ... MANIFEST with the settings above, then read partition_summaries for the current snapshot.

Expected result

OPTIMIZE succeeds for every table above, including the NaN float partition.

On each rewritten manifest-list entry, lower_bound and upper_bound are present and equal the partition value carried by the source manifests, for bucket, year, month, day, hour, and identity float / double. A reader that prunes on those summaries skips manifests whose partition range does not match the filter.

For the NaN float partition, the rewritten summary has contains_nan = true. SELECT returns the same rows as before compaction.

Rewritten data manifests drop optional per-file fields

Description

The manifest-only rewrite copies column_sizes, value_counts, null_value_counts, lower and upper bounds, file_format, sort_order_id, and entry lineage. It does not round-trip other optional Iceberg data-file fields that are present in the Avro schema, in particular nan_value_counts and split_offsets.

AvroForIcebergDeserializer never loads those fields, so the writer cannot put them back. After compaction, Spark sees no NaN column counts and no file split offsets on the rewritten snapshot. Query results stay the same. Scan planning uses one split per file and weaker NaN statistics.

Steps to reproduce

CREATE TABLE db.measures (id BIGINT, x DOUBLE) USING iceberg
TBLPROPERTIES ('format-version' = '2', 'commit.manifest-merge.enabled' = 'false');

INSERT INTO db.measures
SELECT id, CASE WHEN id % 5 = 0 THEN CAST('NaN' AS DOUBLE) ELSE CAST(id AS DOUBLE) END
FROM range(0, 1000);

INSERT INTO db.measures
SELECT id, CASE WHEN id % 5 = 0 THEN CAST('NaN' AS DOUBLE) ELSE CAST(id AS DOUBLE) END
FROM range(1000, 2000);

INSERT INTO db.measures
SELECT id, CASE WHEN id % 5 = 0 THEN CAST('NaN' AS DOUBLE) ELSE CAST(id AS DOUBLE) END
FROM range(2000, 3000);

In Spark, before compaction, read the current data files:

SELECT file_path, nan_value_counts, split_offsets
FROM db.measures.files;

Confirm nan_value_counts has an entry for x and split_offsets is non-empty. Attach the table in ClickHouse and run:

OPTIMIZE TABLE lake.`db.measures` MANIFEST
SETTINGS
    allow_experimental_iceberg_compaction = 1,
    iceberg_manifest_min_count_to_compact = 1;

Read db.measures.files again for the current snapshot, and compare each rewritten data file with the same file_path from the pre-compaction manifests.

Expected result

Each rewritten data-file entry keeps the source file's nan_value_counts and split_offsets. Spark's scan of the compacted snapshot uses those split offsets and NaN counts. SELECT * and SELECT count() return the same rows as before OPTIMIZE.