Iceberg deletion vector support

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

@@ -131,9 +131,16 @@ ClickHouse supports reading Iceberg tables that use the following deletion metho
 
 - [Position deletes](https://iceberg.apache.org/spec/#position-delete-files)
 - [Equality deletes](https://iceberg.apache.org/spec/#equality-delete-files) (supported from version 25.8+)
+- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) stored in Puffin files (Iceberg v3, read-only)
 
-The following deletion method is **not supported**:
-- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) (introduced in v3)
+The following limitations apply to deletion vectors:
+
+- Only `deletion-vector-v1` Puffin blobs are supported
+- Data files must be in Parquet format
+- Column-scoped deletion vectors (`fields` in the Puffin blob) are not supported
+- Writing deletion vectors is not supported
+
+Parsed deletion vectors can be cached in memory when `use_puffin_files_cache` is enabled and the puffin file has a non-empty `etag`. Empty deletion vectors are cached as well, so repeated reads do not re-fetch the puffin file. The cache can be cleared with `SYSTEM DROP PUFFIN_FILES_CACHE`.
 
 ### Basic usage {#basic-usage}
  ```sql

docs/en/interfaces/formats.md

@@ -110,6 +110,8 @@ The supported formats are:
 | [AvroConfluent](./formats/Avro/AvroConfluent.md)                                                           | ✔   | ✔     |
 | [Parquet](./formats/Parquet/Parquet.md)                                                                    | ✔   | ✔     |
 | [ParquetMetadata](./formats/Parquet/ParquetMetadata.md)                                                    | ✔   | ✗     |
+| [Puffin](./formats/Puffin/Puffin.md)                                                                       | ✔   | ✗     |
+| [PuffinMetadata](./formats/Puffin/PuffinMetadata.md)                                                       | ✔   | ✗     |
 | [Arrow](./formats/Arrow/Arrow.md)                                                                          | ✔   | ✔     |
 | [ArrowStream](./formats/Arrow/ArrowStream.md)                                                              | ✔   | ✔     |
 | [ORC](./formats/ORC.md)                                                                                    | ✔   | ✔     |

docs/en/interfaces/formats/Puffin/Puffin.md

@@ -0,0 +1,50 @@
+---
+description: 'Documentation for the Puffin format'
+input_format: true
+output_format: false
+keywords: ['Puffin']
+sidebar_label: 'Puffin'
+sidebar_position: 1
+slug: /interfaces/formats/Puffin
+title: 'Puffin'
+doc_type: 'reference'
+---
+
+## Description {#description}
+
+Input format for reading [Apache Iceberg Puffin](https://iceberg.apache.org/puffin-spec/) files.
+
+The format exposes deleted row positions from `deletion-vector-v1` blobs. Other blob types (for example `apache-datasketches-theta-v1`) are skipped.
+If a puffin file contains multiple `deletion-vector-v1` blobs, the format outputs one row per such blob.
+
+Fixed output columns:
+- `referenced_data_file` (`String`) - location of the data file the deletion vector applies to (`referenced-data-file` blob property)
+- `deleted_rows` (`Array(UInt64)`) - 64-bit row positions deleted according to the deletion vector roaring bitmap
+
+Deletion vectors whose declared `cardinality` exceeds an absolute materialization ceiling are rejected when `deleted_rows` is requested. Footer `deletion-vector-v1` properties (including that `cardinality` parses as an unsigned integer) are always validated. Selecting only `referenced_data_file` skips on-disk payload I/O and therefore also skips envelope, CRC, roaring deserialize, and the materialization ceiling — intentionally, so a path-only projection does not read up to the blob-size cap.
+
+On-disk `deletion-vector-v1` blob length is bounded by an absolute ceiling (aligned with Iceberg's 2 GiB content-size check). When `deleted_rows` is requested, the reader peeks the envelope header (combined length and magic) before allocating the full payload; CRC is verified after the bounded read.
+
+LZ4-compressed and uncompressed puffin footers are supported. Footer payload size (and declared LZ4 content size) is bounded by a compression ratio where applicable and an absolute ceiling; oversized footers are rejected before allocation.
+
+Only a subset of output columns can be requested. A user-provided structure with unexpected column names or types is rejected when the format is created.
+
+## Example usage {#example-usage}
+
+Read deleted row positions with the referenced data file:
+
+```sql
+SELECT referenced_data_file, deleted_rows
+FROM file(deletes.puffin, Puffin);
+```
+
+Expand deleted positions into individual rows:
+
+```sql
+SELECT referenced_data_file, row_number
+FROM file(deletes.puffin, Puffin)
+ARRAY JOIN deleted_rows AS row_number
+ORDER BY referenced_data_file, row_number;
+```
+
+Use `PuffinMetadata` to inspect footer blob descriptors before reading deletion vectors.

docs/en/interfaces/formats/Puffin/PuffinMetadata.md

@@ -0,0 +1,42 @@
+---
+description: 'Documentation for the PuffinMetadata format'
+input_format: true
+output_format: false
+keywords: ['PuffinMetadata']
+sidebar_label: 'PuffinMetadata'
+sidebar_position: 2
+slug: /interfaces/formats/PuffinMetadata
+title: 'PuffinMetadata'
+doc_type: 'reference'
+---
+
+## Description {#description}
+
+Special input format for reading [Apache Iceberg Puffin](https://iceberg.apache.org/puffin-spec/) file footer metadata.
+It outputs one row per blob entry from the footer `BlobMetadata` list.
+
+Fixed output columns:
+- `blob_type` (`String`) - blob type, for example `deletion-vector-v1`
+- `snapshot_id` (`Int64`) - snapshot id of the blob
+- `sequence_number` (`Int64`) - sequence number of the blob
+- `fields` (`Array(Int32)`) - list of field ids the blob applies to
+- `offset` (`Int64`) - offset of the blob payload in the file
+- `length` (`Int64`) - length of the blob payload in bytes
+- `compression_codec` (`String`) - compression codec of the blob payload, if present
+- `properties` (`Map(String, String)`) - blob-specific properties
+
+Optional top-level `FileMetadata.properties` in the footer (for example `created-by`) are type-checked when present but are not returned as columns. If the key is present it must be a JSON object with string values (null is rejected).
+
+LZ4-compressed and uncompressed puffin footers are supported. Footer payload size (and declared LZ4 content size) is bounded by a compression ratio where applicable and an absolute ceiling; oversized footers are rejected before allocation.
+
+## Example usage {#example-usage}
+
+Inspect footer blobs:
+
+```sql
+SELECT blob_type, snapshot_id, sequence_number, offset, length, compression_codec,
+       mapKeys(properties), mapValues(properties)
+FROM file(deletes.puffin, PuffinMetadata);
+```
+
+Pair with the `Puffin` format to read `deletion-vector-v1` blob payloads.

docs/en/sql-reference/statements/system.md

@@ -109,6 +109,10 @@ Clears the per-URL Confluent Schema Registry caches used by the `AvroConfluent`
 
 Clears the parquet metadata cache.
 
+## SYSTEM DROP PUFFIN_FILES_CACHE {#drop-puffin-files-cache}
+
+Clears the Puffin files cache used for parsed Iceberg puffin file content such as deletion vectors.
+
 ## SYSTEM CLEAR|DROP TEXT INDEX CACHES {#drop-text-index-caches}
 
 Clears the text index's header, dictionary and postings caches.

docs/en/sql-reference/table-functions/iceberg.md

@@ -120,11 +120,20 @@ ClickHouse supports time travel for Iceberg tables, allowing you to query histor
 
 ## Processing of tables with deleted rows {#deleted-rows}
 
-Currently, only Iceberg tables with [position deletes](https://iceberg.apache.org/spec/#position-delete-files) are supported. 
+ClickHouse supports reading Iceberg tables that use the following deletion methods:
 
-The following deletion methods are **not supported**:
-- [Equality deletes](https://iceberg.apache.org/spec/#equality-delete-files)
-- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) (introduced in v3)
+- [Position deletes](https://iceberg.apache.org/spec/#position-delete-files)
+- [Equality deletes](https://iceberg.apache.org/spec/#equality-delete-files) (supported from version 25.8+)
+- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) stored in Puffin files (Iceberg v3, read-only)
+
+The following limitations apply to deletion vectors:
+
+- Only `deletion-vector-v1` Puffin blobs are supported
+- Data files must be in Parquet format
+- Column-scoped deletion vectors (`fields` in the Puffin blob) are not supported
+- Writing deletion vectors is not supported
+
+Parsed deletion vectors can be cached in memory when `use_puffin_files_cache` is enabled and the puffin file has a non-empty `etag`. Empty deletion vectors are cached as well, so repeated reads do not re-fetch the puffin file. The cache can be cleared with `SYSTEM DROP PUFFIN_FILES_CACHE`.
 
 ### Basic usage {#basic-usage}
 

programs/local/LocalServer.cpp

@@ -172,6 +172,10 @@ namespace ServerSetting
     extern const ServerSettingsUInt64 parquet_metadata_cache_size;
     extern const ServerSettingsUInt64 parquet_metadata_cache_max_entries;
     extern const ServerSettingsDouble parquet_metadata_cache_size_ratio;
+    extern const ServerSettingsString puffin_files_cache_policy;
+    extern const ServerSettingsUInt64 puffin_files_cache_size;
+    extern const ServerSettingsUInt64 puffin_files_cache_max_entries;
+    extern const ServerSettingsDouble puffin_files_cache_size_ratio;
     extern const ServerSettingsUInt64 max_active_parts_loading_thread_pool_size;
     extern const ServerSettingsUInt64 max_io_thread_pool_free_size;
     extern const ServerSettingsUInt64 max_io_thread_pool_size;
@@ -1540,6 +1544,17 @@ void LocalServer::processConfig()
     global_context->setParquetMetadataCache(parquet_metadata_cache_policy, parquet_metadata_cache_size, parquet_metadata_cache_max_entries, parquet_metadata_cache_size_ratio);
 #endif
 
+    String puffin_files_cache_policy = server_settings[ServerSetting::puffin_files_cache_policy];
+    size_t puffin_files_cache_size = server_settings[ServerSetting::puffin_files_cache_size];
+    size_t puffin_files_cache_max_entries = server_settings[ServerSetting::puffin_files_cache_max_entries];
+    double puffin_files_cache_size_ratio = server_settings[ServerSetting::puffin_files_cache_size_ratio];
+    if (puffin_files_cache_size > max_cache_size)
+    {
+        puffin_files_cache_size = max_cache_size;
+        LOG_INFO(log, "Lowered Puffin files cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(puffin_files_cache_size));
+    }
+    global_context->setPuffinFilesCache(puffin_files_cache_policy, puffin_files_cache_size, puffin_files_cache_max_entries, puffin_files_cache_size_ratio);
+
     Names allowed_disks_table_engines;
     splitInto<','>(allowed_disks_table_engines, server_settings[ServerSetting::allowed_disks_for_table_engines].value);
     global_context->setAllowedDisksForTableEngines(std::unordered_set<String>(allowed_disks_table_engines.begin(), allowed_disks_table_engines.end()));

programs/server/Server.cpp

@@ -288,6 +288,10 @@ namespace ServerSetting
     extern const ServerSettingsUInt64 parquet_metadata_cache_size;
     extern const ServerSettingsUInt64 parquet_metadata_cache_max_entries;
     extern const ServerSettingsDouble parquet_metadata_cache_size_ratio;
+    extern const ServerSettingsString puffin_files_cache_policy;
+    extern const ServerSettingsUInt64 puffin_files_cache_size;
+    extern const ServerSettingsUInt64 puffin_files_cache_max_entries;
+    extern const ServerSettingsDouble puffin_files_cache_size_ratio;
     extern const ServerSettingsUInt64 io_thread_pool_queue_size;
     extern const ServerSettingsBool jemalloc_enable_global_profiler;
     extern const ServerSettingsBool jemalloc_collect_global_profile_samples_in_trace_log;
@@ -2277,6 +2281,16 @@ try
     }
     global_context->setParquetMetadataCache(parquet_metadata_cache_policy, parquet_metadata_cache_size, parquet_metadata_cache_max_entries, parquet_metadata_cache_size_ratio);
 #endif
+    String puffin_files_cache_policy = server_settings[ServerSetting::puffin_files_cache_policy];
+    size_t puffin_files_cache_size = server_settings[ServerSetting::puffin_files_cache_size];
+    size_t puffin_files_cache_max_entries = server_settings[ServerSetting::puffin_files_cache_max_entries];
+    double puffin_files_cache_size_ratio = server_settings[ServerSetting::puffin_files_cache_size_ratio];
+    if (puffin_files_cache_size > max_cache_size)
+    {
+        puffin_files_cache_size = max_cache_size;
+        LOG_INFO(log, "Lowered Puffin files cache size to {} because the system has limited RAM", formatReadableSizeWithBinarySuffix(puffin_files_cache_size));
+    }
+    global_context->setPuffinFilesCache(puffin_files_cache_policy, puffin_files_cache_size, puffin_files_cache_max_entries, puffin_files_cache_size_ratio);
 
     Names allowed_disks_table_engines;
     splitInto<','>(allowed_disks_table_engines, server_settings[ServerSetting::allowed_disks_for_table_engines].value);
@@ -2706,6 +2720,7 @@ try
 #if USE_PARQUET
                 global_context->updateParquetMetadataCacheConfiguration(config(), max_cache_size_in_bytes);
 #endif
+                global_context->updatePuffinFilesCacheConfiguration(config(), max_cache_size_in_bytes);
             }
 
 #if USE_SSL

src/Access/Common/AccessType.h

@@ -318,6 +318,7 @@ enum class AccessType : uint8_t
     M(SYSTEM_DROP_ICEBERG_METADATA_CACHE, "SYSTEM CLEAR ICEBERG_METADATA_CACHE, SYSTEM DROP ICEBERG_METADATA_CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_AVRO_SCHEMA_CACHE, "SYSTEM CLEAR AVRO SCHEMA CACHE, SYSTEM DROP AVRO SCHEMA CACHE, DROP AVRO SCHEMA CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_PARQUET_METADATA_CACHE, "SYSTEM DROP PARQUET_METADATA_CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
+    M(SYSTEM_DROP_PUFFIN_FILES_CACHE, "SYSTEM DROP PUFFIN_FILES_CACHE", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_PREWARM_PRIMARY_INDEX_CACHE, "SYSTEM PREWARM PRIMARY INDEX, PREWARM PRIMARY INDEX CACHE, PREWARM PRIMARY INDEX", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_PRIMARY_INDEX_CACHE, "SYSTEM CLEAR PRIMARY INDEX CACHE, SYSTEM DROP PRIMARY INDEX, DROP PRIMARY INDEX CACHE, DROP PRIMARY INDEX", GLOBAL, SYSTEM_DROP_CACHE) \
     M(SYSTEM_DROP_UNCOMPRESSED_CACHE, "SYSTEM CLEAR UNCOMPRESSED CACHE, SYSTEM DROP UNCOMPRESSED, DROP UNCOMPRESSED CACHE, DROP UNCOMPRESSED", GLOBAL, SYSTEM_DROP_CACHE) \

src/AggregateFunctions/AggregateFunctionGroupBitmapData.h

@@ -34,6 +34,72 @@ enum BitmapKind
     Bitmap = 1
 };
 
+/// Approximate heap footprint of a 32-bit CRoaring bitmap (index capacity + container capacities).
+/// Unlike `Roaring::getSizeInBytes()`, array/run containers use allocated capacity, not cardinality.
+inline UInt64 estimateRoaring32AllocatedBytes(const roaring::Roaring & bitmap)
+{
+    using namespace roaring::internal;
+    const roaring::roaring_array_t * ra = &bitmap.roaring.high_low_container;
+    UInt64 bytes = sizeof(roaring::Roaring);
+    if (ra->allocation_size > 0)
+    {
+        bytes += static_cast<UInt64>(ra->allocation_size)
+            * (sizeof(uint16_t) + sizeof(uint8_t) + sizeof(container_t *));
+    }
+    for (int32_t i = 0; i < ra->size; ++i)
+    {
+        uint8_t typecode = ra->typecodes[i];
+        const container_t * c = container_unwrap_shared(ra->containers[i], &typecode);
+        switch (typecode)
+        {
+            case ARRAY_CONTAINER_TYPE:
+            {
+                const array_container_t * ac = const_CAST_array(c);
+                bytes += sizeof(array_container_t) + static_cast<UInt64>(ac->capacity) * sizeof(uint16_t);
+                break;
+            }
+            case BITSET_CONTAINER_TYPE:
+            {
+                bytes += sizeof(bitset_container_t) + BITSET_CONTAINER_SIZE_IN_WORDS * sizeof(uint64_t);
+                break;
+            }
+            case RUN_CONTAINER_TYPE:
+            {
+                const run_container_t * rc = const_CAST_run(c);
+                bytes += sizeof(run_container_t) + static_cast<UInt64>(rc->capacity) * sizeof(rle16_t);
+                break;
+            }
+            default:
+                break;
+        }
+    }
+    return bytes;
+}
+
+/// Count distinct high-32 keys in a Roaring64Map. Fast-path when min/max share one key.
+inline UInt64 countRoaring64MapHighKeys(const roaring::Roaring64Map & bitmap)
+{
+    if (bitmap.isEmpty())
+        return 0;
+
+    const UInt64 min_value = bitmap.minimum();
+    const UInt64 max_value = bitmap.maximum();
+    if ((min_value >> 32) == (max_value >> 32))
+        return 1;
+
+    UInt64 keys = 0;
+    UInt64 prev_high = ~UInt64{0};
+    for (auto it = bitmap.begin(); it != bitmap.end(); ++it)
+    {
+        const UInt64 high = static_cast<UInt64>(*it) >> 32;
+        if (high != prev_high)
+        {
+            ++keys;
+            prev_high = high;
+        }
+    }
+    return keys;
+}
 
 /**
   * For a small number of values - an array of fixed size "on the stack".
@@ -93,6 +159,34 @@ class RoaringBitmapWithSmallSet : private boost::noncopyable
         return roaring_bitmap->cardinality();
     }
 
+    UInt64 getAllocatedBytes() const
+    {
+        if (isSmall())
+            return sizeof(small);
+
+        /// Prefer a heap estimate over `getSizeInBytes()` (serialization size). Include a small
+        /// allowance for the shared_ptr control block that owns `roaring_bitmap`.
+        constexpr UInt64 SHARED_PTR_CONTROL_BLOCK = 32;
+
+        if constexpr (sizeof(T) < 8)
+        {
+            return estimateRoaring32AllocatedBytes(*roaring_bitmap) + SHARED_PTR_CONTROL_BLOCK;
+        }
+        else
+        {
+            /// Roaring64Map keeps roarings private; approximate as native serialization size plus
+            /// per-high-key map/Roaring/container overhead (serialization undercounts capacity and
+            /// std::map nodes — important for sparse high keys).
+            constexpr UInt64 PER_HIGH_KEY_OVERHEAD =
+                4 * sizeof(void *) + sizeof(UInt32) + sizeof(roaring::Roaring) + 64;
+
+            const UInt64 serialized = roaring_bitmap->getSizeInBytes(/*portable=*/false);
+            const UInt64 high_keys = countRoaring64MapHighKeys(*roaring_bitmap);
+            return serialized + high_keys * PER_HIGH_KEY_OVERHEAD + sizeof(roaring::Roaring64Map)
+                + SHARED_PTR_CONTROL_BLOCK;
+        }
+    }
+
     void merge(const RoaringBitmapWithSmallSet & r1)
     {
         if (r1.isLarge())
@@ -535,6 +629,51 @@ class RoaringBitmapWithSmallSet : private boost::noncopyable
         return count;
     }
 
+    /**
+     * Count set bits in `[range_start, range_end)` without allocating a result bitmap.
+     * Used by need-only-count DV filtering to avoid an O(N) dense Filter over file rows.
+     * Implemented via roaring `rank` so repeated per-row-group queries stay O(containers),
+     * not O(row_groups × cardinality).
+     */
+    UInt64 rb_range_cardinality(UInt64 range_start, UInt64 range_end) const /// NOLINT
+    {
+        if (range_start >= range_end)
+            return 0;
+
+        if (isSmall())
+        {
+            UInt64 count = 0;
+            for (const auto & x : small)
+            {
+                const UInt64 val = static_cast<UInt64>(x.getValue());
+                if (val >= range_start && val < range_end)
+                    ++count;
+            }
+            return count;
+        }
+
+        /// |bitmap ∩ [start, end)| = rank(end - 1) - rank(start - 1). Same formula as DeleteBitmap.
+        if constexpr (sizeof(T) < 8)
+        {
+            constexpr UInt64 max_row = std::numeric_limits<UInt32>::max();
+            if (range_start > max_row)
+                return 0;
+            const UInt64 hi_inclusive = std::min(range_end - 1, max_row);
+            if (hi_inclusive < range_start)
+                return 0;
+            const UInt64 upper = roaring_bitmap->rank(static_cast<UInt32>(hi_inclusive));
+            const UInt64 lower = (range_start == 0) ? 0 : roaring_bitmap->rank(static_cast<UInt32>(range_start - 1));
+            return upper - lower;
+        }
+        else
+        {
+            const UInt64 hi_inclusive = range_end - 1;
+            const UInt64 upper = roaring_bitmap->rank(hi_inclusive);
+            const UInt64 lower = (range_start == 0) ? 0 : roaring_bitmap->rank(range_start - 1);
+            return upper - lower;
+        }
+    }
+
     /**
      * Return new set of the smallest `limit` values in set which is no less than `range_start`.
      * It's used in subset and currently only support UInt32