Iceberg deletion vector support (attempt #2)

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/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}
 

docs/reference/formats/Puffin/Puffin.mdx

@@ -1,6 +1,10 @@
 ---
 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'

docs/reference/formats/Puffin/PuffinMetadata.mdx

@@ -1,6 +1,10 @@
 ---
 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'

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

src/Client/BuzzHouse/Generator/SessionSettings.cpp

@@ -1645,6 +1645,7 @@ static std::unordered_map<String, CHSetting> serverSettings2 = {
     {"use_page_cache_for_local_disks", trueOrFalseSetting},
     {"use_page_cache_for_object_storage", trueOrFalseSetting},
     {"use_parquet_metadata_cache", trueOrFalseSetting},
+    {"use_puffin_files_cache", trueOrFalseSetting},
     {"use_query_cache", trueOrFalseSetting},
     {"use_roaring_bitmap_iceberg_positional_deletes", trueOrFalseSetting},
     {"use_skip_indexes_if_final_exact_mode", CHSetting(trueOrFalse, {"0", "1"}, true)},