Support for Databricks deletion vectors

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

@@ -131,19 +131,19 @@ 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)
+- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) (Iceberg v3, read-only), stored either in Puffin files or in Delta-style `deletion_vector_*.bin` files using the same `deletion-vector-v1` envelope at the manifest `content_offset` / `content_size_in_bytes`
 
 The following limitations apply to deletion vectors:
 
-- Only `deletion-vector-v1` Puffin blobs are supported
+- Only `deletion-vector-v1` blobs are supported (Puffin container or Delta `.bin` slice)
 - Data files must be in Parquet format
 - Column-scoped deletion vectors (user column ids in puffin `fields`) are not supported. Writers may set `fields` to `[]` or to the Iceberg reserved `_pos` id (`2147483645`) for file-scoped deletion vectors.
 - Writing deletion vectors is not supported
 - `DELETE` / `UPDATE` mutations on Iceberg format version 3+ tables are rejected (writers must not add position-delete files)
 
-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. Parsed footers for coalesced multi-DV Puffin files are memoized with that cache (same identity: storage, path, `etag`) so slices share one footer parse; the memo shares `puffin_files_cache_size` / max-entry limits and is dropped when the cache is disabled (`puffin_files_cache_size=0`) or cleared. The cache can be cleared with `SYSTEM DROP PUFFIN FILES CACHE`.
+Parsed deletion vectors can be cached in memory when `use_puffin_files_cache` is enabled and the deletion-vector object has a non-empty `etag`. Empty deletion vectors are cached as well, so repeated reads do not re-fetch the object. Parsed footers for coalesced multi-DV Puffin files are memoized with that cache (same identity: storage, path, `etag`) so slices share one footer parse; Delta `.bin` files have no Puffin footer and skip that memo. The memo shares `puffin_files_cache_size` / max-entry limits and is dropped when the cache is disabled (`puffin_files_cache_size=0`) or cleared. The cache can be cleared with `SYSTEM DROP PUFFIN FILES CACHE`.
 
-For [`icebergCluster`](/sql-reference/table-functions/icebergCluster.md) (and `object_storage_cluster`), the initiator loads and materializes each data file's deletion vector while distributing tasks, then sends the resulting row bitmap to workers with the task. Workers apply the bitmap; they do not re-read the Puffin blob for that path. On wide v3 tables this can make the initiator a serialization point for deletion-vector I/O and decode.
+For [`icebergCluster`](/sql-reference/table-functions/icebergCluster.md) (and `object_storage_cluster`), the initiator loads and materializes each data file's deletion vector while distributing tasks, then sends the resulting row bitmap to workers with the task. Workers apply the bitmap; they do not re-read the Puffin or `.bin` object for that path. On wide v3 tables this can make the initiator a serialization point for deletion-vector I/O and decode.
 
 ### Basic usage {#basic-usage}
  ```sql

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

@@ -124,19 +124,19 @@ 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)
+- [Deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) (Iceberg v3, read-only), stored either in Puffin files or in Delta-style `deletion_vector_*.bin` files using the same `deletion-vector-v1` envelope at the manifest `content_offset` / `content_size_in_bytes`
 
 The following limitations apply to deletion vectors:
 
-- Only `deletion-vector-v1` Puffin blobs are supported
+- Only `deletion-vector-v1` blobs are supported (Puffin container or Delta `.bin` slice)
 - Data files must be in Parquet format
 - Column-scoped deletion vectors (user column ids in puffin `fields`) are not supported. Writers may set `fields` to `[]` or to the Iceberg reserved `_pos` id (`2147483645`) for file-scoped deletion vectors.
 - Writing deletion vectors is not supported
 - `DELETE` / `UPDATE` mutations on Iceberg format version 3+ tables are rejected (writers must not add position-delete files)
 
-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. Parsed footers for coalesced multi-DV Puffin files are memoized with that cache (same identity: storage, path, `etag`) so slices share one footer parse; the memo shares `puffin_files_cache_size` / max-entry limits and is dropped when the cache is disabled (`puffin_files_cache_size=0`) or cleared. The cache can be cleared with `SYSTEM DROP PUFFIN FILES CACHE`.
+Parsed deletion vectors can be cached in memory when `use_puffin_files_cache` is enabled and the deletion-vector object has a non-empty `etag`. Empty deletion vectors are cached as well, so repeated reads do not re-fetch the object. Parsed footers for coalesced multi-DV Puffin files are memoized with that cache (same identity: storage, path, `etag`) so slices share one footer parse; Delta `.bin` files have no Puffin footer and skip that memo. The memo shares `puffin_files_cache_size` / max-entry limits and is dropped when the cache is disabled (`puffin_files_cache_size=0`) or cleared. The cache can be cleared with `SYSTEM DROP PUFFIN FILES CACHE`.
 
-For [`icebergCluster`](/sql-reference/table-functions/icebergCluster.md) (and `object_storage_cluster`), the initiator loads and materializes each data file's deletion vector while distributing tasks, then sends the resulting row bitmap to workers with the task. Workers apply the bitmap; they do not re-read the Puffin blob for that path. On wide v3 tables this can make the initiator a serialization point for deletion-vector I/O and decode.
+For [`icebergCluster`](/sql-reference/table-functions/icebergCluster.md) (and `object_storage_cluster`), the initiator loads and materializes each data file's deletion vector while distributing tasks, then sends the resulting row bitmap to workers with the task. Workers apply the bitmap; they do not re-read the Puffin or `.bin` object for that path. On wide v3 tables this can make the initiator a serialization point for deletion-vector I/O and decode.
 
 ### Basic usage {#basic-usage}
 

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

@@ -14,7 +14,7 @@ Allows processing files from Apache [Iceberg](https://iceberg.apache.org/) in pa
 
 ## Deletion vectors on cluster reads {#deletion-vectors-cluster}
 
-Iceberg v3 [deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) are loaded on the **initiator** while it distributes tasks: for each data file the initiator reads the Puffin blob, validates it, materializes deleted row positions, and attaches the bitmap to the task sent to workers. Workers apply that bitmap when reading Parquet; they do not fetch or parse the Puffin file again for that path. See [Processing of tables with deleted rows](/sql-reference/table-functions/iceberg.md#deleted-rows) for format limits and caching.
+Iceberg v3 [deletion vectors](https://iceberg.apache.org/spec/#deletion-vectors) are loaded on the **initiator** while it distributes tasks: for each data file the initiator reads the Puffin or Delta `.bin` object, validates the `deletion-vector-v1` envelope, materializes deleted row positions, and attaches the bitmap to the task sent to workers. Workers apply that bitmap when reading Parquet; they do not fetch or parse the deletion-vector file again for that path. See [Processing of tables with deleted rows](/sql-reference/table-functions/iceberg.md#deleted-rows) for format limits and caching.
 
 On wide tables with many deletion vectors, initiator-side decode and per-task bitmap serialization can become a bottleneck even when Parquet reads are well parallelized across the cluster.
 

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

@@ -294,6 +294,7 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
         case FileContentType::POSITION_DELETE: {
             /// reference_file_path can be absent in schema for some reason, though it is present in specification: https://iceberg.apache.org/spec/#manifests
             const bool is_puffin = Poco::toLower(file_format) == "puffin";
+            const bool has_dv_offsets = content_offset.has_value() && content_size_in_bytes.has_value();
             std::optional<Iceberg::IcebergPathFromMetadata> lower_reference_data_file_path;
             std::optional<Iceberg::IcebergPathFromMetadata> upper_reference_data_file_path;
             bool bounds_set_by_referenced_data_file = false;
@@ -309,9 +310,9 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
                     bounds_set_by_referenced_data_file = true;
                 }
             }
-            /// Parquet position deletes may fall back to file-path column bounds. Puffin deletion
+            /// Parquet position deletes may fall back to file-path column bounds. Deletion
             /// vectors must use the dedicated referenced_data_file field only.
-            if (!bounds_set_by_referenced_data_file && !is_puffin)
+            if (!bounds_set_by_referenced_data_file && !is_puffin && !has_dv_offsets)
             {
                 if (auto it = value_for_bounds.find(IcebergPositionDeleteTransform::data_file_path_column_field_id);
                     it != value_for_bounds.end())
@@ -326,13 +327,17 @@ ParsedManifestFileEntryPtr AvroForIcebergDeserializer::createParsedManifestFileE
 
             if (is_puffin)
             {
-                if (!content_offset.has_value() || !content_size_in_bytes.has_value())
+                if (!has_dv_offsets)
                 {
                     throw Exception(
                         DB::ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION,
                         "Puffin deletion vector entry in manifest file '{}' is missing content_offset or content_size_in_bytes",
                         manifest_file_path);
                 }
+            }
+
+            if (is_puffin || has_dv_offsets)
+            {
                 requireDirectReferencedDataFileForPuffinDeletionVector(
                     bounds_set_by_referenced_data_file, lower_reference_data_file_path, manifest_file_path);
             }

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp

@@ -18,6 +18,7 @@
 #include <IO/WithFileSize.h>
 
 #include <atomic>
+#include <optional>
 
 namespace DB
 {
@@ -131,7 +132,9 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
     ContextPtr context,
     LoggerPtr log,
     bool disable_filesystem_cache,
-    FooterBlobsPtr preloaded_footer)
+    FooterBlobsPtr preloaded_footer,
+    PuffinFilesCache * footer_cache = nullptr,
+    const std::optional<PuffinFooterCacheKey> & footer_key = {})
 {
     RelativePathWithMetadata puffin_object{puffin_path};
     auto read_settings = context->getReadSettings();
@@ -148,16 +151,35 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
     if (!file_size)
         throw Exception(ErrorCodes::BAD_ARGUMENTS, "Cannot determine Puffin file size for '{}'", puffin_path);
 
-    FooterBlobsPtr footer_owner = preloaded_footer;
-    if (!footer_owner)
-        footer_owner = std::make_shared<const std::vector<PuffinBlob>>(readPuffinFooterBlobsFromSeekable(*seekable, *file_size));
+    const auto container = detectIcebergDeletionVectorContainer(
+        *seekable, *file_size, content_offset, content_size_in_bytes, puffin_path);
 
-    bindDeletionVectorBlob(
-        *footer_owner,
-        content_offset,
-        content_size_in_bytes,
-        expected_data_file.serialize(),
-        expected_cardinality);
+    if (container == IcebergDeletionVectorContainer::Puffin)
+    {
+        FooterBlobsPtr footer_owner = preloaded_footer;
+        if (!footer_owner)
+        {
+            if (footer_cache && footer_key.has_value())
+            {
+                footer_owner = footer_cache->getOrSetFooter(*footer_key, [&]()
+                {
+                    return readFooterBlobs(object_storage, puffin_path, context, log, /*disable_filesystem_cache=*/ true);
+                });
+            }
+            else
+            {
+                footer_owner = std::make_shared<const std::vector<PuffinBlob>>(
+                    readPuffinFooterBlobsFromSeekable(*seekable, *file_size));
+            }
+        }
+
+        bindDeletionVectorBlob(
+            *footer_owner,
+            content_offset,
+            content_size_in_bytes,
+            expected_data_file.serialize(),
+            expected_cardinality);
+    }
 
     auto deleted_positions = readDeletionVectorFromPuffin(
         *read_buffer, content_offset, content_size_in_bytes, expected_cardinality);
@@ -173,8 +195,9 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVectorUncached(
 
     LOG_DEBUG(
         log,
-        "Loaded deletion vector from puffin file '{}' for data file '{}': {} deleted rows",
+        "Loaded deletion vector from file '{}' ({}) for data file '{}': {} deleted rows",
         puffin_path,
+        container == IcebergDeletionVectorContainer::Puffin ? "Puffin" : "Delta .bin",
         expected_data_file.serialize(),
         deleted_positions.size());
 
@@ -367,14 +390,10 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
     }
 
     /// Footer is keyed by file identity only, so N DV slices in one coalesced Puffin share one parse.
-    /// Resolve the footer only on a deletion-vector cache miss (nested memo lookup).
+    /// Resolve the footer only on a Puffin deletion-vector cache miss (nested memo lookup).
+    /// Delta `.bin` objects have no Puffin footer; detection inside the uncached loader skips memo.
     return cache->getOrSetDeletionVector(*cache_key, [&]()
     {
-        auto footer = cache->getOrSetFooter(*footer_key, [&]()
-        {
-            return readFooterBlobs(object_storage, puffin_path, context, log, /*disable_filesystem_cache=*/ true);
-        });
-
         return loadDeletionVectorUncached(
             object_storage,
             puffin_path,
@@ -386,7 +405,9 @@ DataLakeObjectMetadata::ExcludedRowsPtr loadDeletionVector(
             context,
             log,
             true,
-            footer);
+            nullptr,
+            cache.get(),
+            footer_key);
     });
 }
 

src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp

@@ -398,7 +398,8 @@ ObjectInfoPtr IcebergIterator::next(size_t)
 
             /// For icebergCluster, next() runs on the initiator's task-distribution path: DV
             /// I/O / CRC / roaring materialization happen here, then excluded_rows is sent on
-            /// the wire per task. Workers apply the bitmap and do not re-read the Puffin blob.
+            /// the wire per task. Workers apply the bitmap and do not re-read the deletion-vector
+            /// object (Puffin or Delta `.bin`).
             auto excluded_rows = Iceberg::loadDeletionVector(
                 puffin_storage,
                 puffin_key,
@@ -417,7 +418,7 @@ ObjectInfoPtr IcebergIterator::next(size_t)
             has_deletion_vector = true;
             LOG_DEBUG(
                 logger,
-                "Attached deletion vector from puffin file `{}` to data file `{}`",
+                "Attached deletion vector from file `{}` to data file `{}`",
                 parsed_entry->file_path_key,
                 data_file_path);
         }

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp

@@ -92,7 +92,7 @@ void requireDirectReferencedDataFileForPuffinDeletionVector(
     {
         throw DB::Exception(
             DB::ErrorCodes::ICEBERG_SPECIFICATION_VIOLATION,
-            "Puffin deletion vector entry in manifest file '{}' is missing referenced_data_file",
+            "Deletion vector entry in manifest file '{}' is missing referenced_data_file",
             manifest_file_path);
     }
 }

src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h

@@ -30,7 +30,6 @@ struct ColumnInfo
 #include <vector>
 
 #include <boost/noncopyable.hpp>
-#include <Poco/String.h>
 
 namespace DB::Iceberg
 {
@@ -103,15 +102,14 @@ struct ParsedManifestFileEntry : boost::noncopyable
     Int64 record_count;
     Int64 file_size_in_bytes;
 
-    /// Iceberg v3 deletion vector metadata (position delete entries with puffin format)
+    /// Iceberg v3 deletion vector metadata (`content_offset` / `content_size_in_bytes`).
+    /// Present for Puffin containers and for Delta-style `.bin` files that store the same envelope.
     std::optional<Int64> content_offset;
     std::optional<Int64> content_size_in_bytes;
 
     bool isDeletionVector() const
     {
-        return Poco::toLower(file_format) == "puffin"
-            && content_offset.has_value()
-            && content_size_in_bytes.has_value();
+        return content_offset.has_value() && content_size_in_bytes.has_value();
     }
 
     ParsedManifestFileEntry(
@@ -182,8 +180,9 @@ std::optional<Int64> getRecordCountInAllFilesExcludingDeleted(
 std::optional<Int64> getBytesSizeInAllDataFilesExcludingDeleted(
     const std::vector<ProcessedManifestFileEntryPtr> & files);
 
-/// Puffin deletion vectors must identify the data file via the dedicated `referenced_data_file`
-/// manifest field (non-empty). Position-delete lower/upper bounds must not be used as a fallback.
+/// Deletion vectors (Puffin or Delta `.bin`) must identify the data file via the dedicated
+/// `referenced_data_file` manifest field (non-empty). Position-delete lower/upper bounds
+/// must not be used as a fallback.
 void requireDirectReferencedDataFileForPuffinDeletionVector(
     bool set_from_referenced_data_file_field,
     const std::optional<IcebergPathFromMetadata> & referenced_path,