diff --git a/docs/en/engines/table-engines/integrations/iceberg.md b/docs/en/engines/table-engines/integrations/iceberg.md index 19b87ea5c1a9..f67379d3a369 100644 --- a/docs/en/engines/table-engines/integrations/iceberg.md +++ b/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 diff --git a/docs/en/sql-reference/table-functions/iceberg.md b/docs/en/sql-reference/table-functions/iceberg.md index 1abff945e6bb..6dd8b8dd3a79 100644 --- a/docs/en/sql-reference/table-functions/iceberg.md +++ b/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} diff --git a/docs/en/sql-reference/table-functions/icebergCluster.md b/docs/en/sql-reference/table-functions/icebergCluster.md index 312e5b30ec0b..72577f140e71 100644 --- a/docs/en/sql-reference/table-functions/icebergCluster.md +++ b/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. diff --git a/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp b/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp index 277d04c3a77a..8aca0793296f 100644 --- a/src/Storages/ObjectStorage/DataLakes/Common/AvroForIcebergDeserializer.cpp +++ b/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 lower_reference_data_file_path; std::optional 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); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp index 91d40f6df66f..7f77e3f40e3c 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergDeletionVector.cpp @@ -18,6 +18,7 @@ #include #include +#include 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 & 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>(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>( + 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); }); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp index b843d87e2868..b8a20b9b46bc 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/IcebergIterator.cpp +++ b/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); } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp index bffd91213282..201fdcd3a7eb 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.cpp +++ b/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); } } diff --git a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h index ff5fd50c0884..acf7a39b17e9 100644 --- a/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h +++ b/src/Storages/ObjectStorage/DataLakes/Iceberg/ManifestFile.h @@ -30,7 +30,6 @@ struct ColumnInfo #include #include -#include 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 content_offset; std::optional 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 getRecordCountInAllFilesExcludingDeleted( std::optional getBytesSizeInAllDataFilesExcludingDeleted( const std::vector & 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 & referenced_path, diff --git a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.cpp b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.cpp index b91a87fc6269..92a11a6457d0 100644 --- a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.cpp +++ b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.cpp @@ -8,6 +8,8 @@ #include #include #include +#include +#include #include #include @@ -107,25 +109,126 @@ void checkDeletionVectorBlobReadLimits(Int64 length, std::optional expec throw Exception(ErrorCodes::BAD_ARGUMENTS, "Deletion vector blob is too small"); } -void validateDeletionVectorEnvelope(const UInt8 * header, Int64 length) +bool isPuffinFileMagic(const UInt8 * header) +{ + return std::memcmp(header, PUFFIN_FILE_MAGIC, sizeof(PUFFIN_FILE_MAGIC)) == 0; +} + +bool isDeletionVectorV1Envelope(const UInt8 * header, Int64 length) { + if (length < 12) + return false; + const UInt32 combined_length = readBigEndianUInt32(header); if (std::memcmp(header + sizeof(UInt32), DELETION_VECTOR_MAGIC, sizeof(DELETION_VECTOR_MAGIC)) != 0) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector magic"); + return false; if (combined_length < sizeof(DELETION_VECTOR_MAGIC)) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector combined length: {}", combined_length); + return false; UInt64 expected_blob_size = 0; if (common::addOverflow(static_cast(combined_length), UInt64{8}, expected_blob_size)) - throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector combined length: {}", combined_length); + return false; + + return static_cast(length) == expected_blob_size; +} + +void validateDeletionVectorEnvelope(const UInt8 * header, Int64 length) +{ + if (!isDeletionVectorV1Envelope(header, length)) + { + const UInt32 combined_length = readBigEndianUInt32(header); + if (std::memcmp(header + sizeof(UInt32), DELETION_VECTOR_MAGIC, sizeof(DELETION_VECTOR_MAGIC)) != 0) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector magic"); + + if (combined_length < sizeof(DELETION_VECTOR_MAGIC)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector combined length: {}", combined_length); + + UInt64 expected_blob_size = 0; + if (common::addOverflow(static_cast(combined_length), UInt64{8}, expected_blob_size)) + throw Exception(ErrorCodes::BAD_ARGUMENTS, "Invalid deletion vector combined length: {}", combined_length); - if (static_cast(length) != expected_blob_size) throw Exception( ErrorCodes::BAD_ARGUMENTS, "Deletion vector blob size {} does not match combined length {}", length, combined_length); + } +} + +namespace +{ + +String formatFourBytesHex(const UInt8 * bytes) +{ + return fmt::format( + "{:02X} {:02X} {:02X} {:02X}", + static_cast(bytes[0]), + static_cast(bytes[1]), + static_cast(bytes[2]), + static_cast(bytes[3])); +} + +[[noreturn]] void throwUnknownDeletionVectorContainer( + std::string_view path, const UInt8 * header, Int64 content_offset) +{ + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Deletion vector file '{}' is neither a Puffin container (header magic {}, expected PFA1) " + "nor a deletion-vector-v1 / Delta .bin envelope at offset {}", + path, + formatFourBytesHex(header), + content_offset); +} + +} + +IcebergDeletionVectorContainer detectIcebergDeletionVectorContainer( + SeekableReadBuffer & file, + size_t file_size, + Int64 content_offset, + Int64 content_size_in_bytes, + std::string_view path) +{ + checkDeletionVectorBlobReadLimits(content_size_in_bytes, /*expected_cardinality=*/std::nullopt); + validatePuffinBlobBounds(content_offset, content_size_in_bytes, file_size, "Deletion vector"); + + if (file_size < PUFFIN_MAGIC_SIZE) + { + throw Exception( + ErrorCodes::BAD_ARGUMENTS, + "Deletion vector file '{}' is neither a Puffin container (file size {} is smaller than PFA1) " + "nor a deletion-vector-v1 / Delta .bin envelope at offset {}", + path, + file_size, + content_offset); + } + + file.seek(0, SEEK_SET); + UInt8 header[PUFFIN_MAGIC_SIZE]; + file.readStrict(reinterpret_cast(header), sizeof(header)); + + if (isPuffinFileMagic(header)) + return IcebergDeletionVectorContainer::Puffin; + + Int64 envelope_end = 0; + const bool envelope_header_in_bounds + = !common::addOverflow(content_offset, static_cast(8), envelope_end) + && envelope_end >= 0 + && static_cast(envelope_end) <= file_size + && content_offset >= 0; + + if (envelope_header_in_bounds) + { + file.seek(content_offset, SEEK_SET); + UInt8 envelope[8]; + file.readStrict(reinterpret_cast(envelope), sizeof(envelope)); + if (isDeletionVectorV1Envelope(envelope, content_size_in_bytes)) + return IcebergDeletionVectorContainer::SliceOnly; + } + + throwUnknownDeletionVectorContainer(path, header, content_offset); + UNREACHABLE(); } std::vector deserializeDeletionVectorV1Blob(std::string_view blob_bytes, std::optional expected_cardinality) diff --git a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h index b78f57041d02..4412015160d1 100644 --- a/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h +++ b/src/Storages/ObjectStorage/DataLakes/PuffinDeletionVectorReader.h @@ -4,10 +4,11 @@ #include #include +#include #include +#include #include #include -#include namespace DB { @@ -44,6 +45,34 @@ void validatePuffinBlobBounds(Int64 offset, Int64 length, size_t file_size, std: /// Iceberg deletion-vector-v1 envelope magic (`0xD1D33964`), shared with SQL `Puffin` decode. inline constexpr UInt8 DELETION_VECTOR_MAGIC[4] = {0xD1, 0xD3, 0x39, 0x64}; +/// Puffin file magic (`PFA1`), shared with SQL `Puffin` decode. +inline constexpr UInt8 PUFFIN_FILE_MAGIC[4] = {0x50, 0x46, 0x41, 0x31}; + +enum class IcebergDeletionVectorContainer : uint8_t +{ + /// Classic Iceberg v3 Puffin (`PFA1` + JSON footer). Footer identity bind is required. + Puffin, + /// Spec-aligned `deletion-vector-v1` envelope at `content_offset` without a Puffin container + /// (Databricks / Delta `deletion_vector_*.bin`). Identity comes from the Iceberg manifest. + SliceOnly, +}; + +/// True when `header` is the 4-byte Puffin file magic `PFA1`. +bool isPuffinFileMagic(const UInt8 * header); + +/// True when `header` (first 8 bytes of a blob) is a `deletion-vector-v1` envelope for `length`. +/// Does not throw; used to distinguish Delta `.bin` slices from unknown files without catch-as-control-flow. +bool isDeletionVectorV1Envelope(const UInt8 * header, Int64 length); + +/// Peek the object start: `PFA1` => Puffin; otherwise require a valid `deletion-vector-v1` envelope +/// at (`content_offset`, `content_size_in_bytes`). Throws `BAD_ARGUMENTS` if neither matches. +IcebergDeletionVectorContainer detectIcebergDeletionVectorContainer( + SeekableReadBuffer & file, + size_t file_size, + Int64 content_offset, + Int64 content_size_in_bytes, + std::string_view path); + /// Fail closed before envelope peek / full allocate. Shared by SQL `Puffin` and Iceberg loaders. /// Order: cardinality ceiling, then length bounds (`length < 0`, absolute blob cap, min envelope). void checkDeletionVectorBlobReadLimits(Int64 length, std::optional expected_cardinality); diff --git a/src/Storages/ObjectStorage/DataLakes/tests/gtest_delta_bin_deletion_vector.cpp b/src/Storages/ObjectStorage/DataLakes/tests/gtest_delta_bin_deletion_vector.cpp new file mode 100644 index 000000000000..62f3db885bb4 --- /dev/null +++ b/src/Storages/ObjectStorage/DataLakes/tests/gtest_delta_bin_deletion_vector.cpp @@ -0,0 +1,186 @@ +#include + +#include +#include +#include + +#include +#include +#include +#include + +using namespace DB; + +namespace DB +{ +namespace ErrorCodes +{ +extern const int BAD_ARGUMENTS; +} +} + +namespace +{ + +/// deletion-vector-v1 blob for positions {2, 5} (cardinality 2). Same bytes as +/// gtest_puffin_deletion_vector_envelope.cpp. +constexpr UInt8 two_position_dv_blob[] = { + 0x00, 0x00, 0x00, 0x24, 0xD1, 0xD3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x3A, 0x30, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, + 0x10, 0x00, 0x00, 0x00, 0x02, 0x00, 0x05, 0x00, 0x2C, 0xDB, 0x9F, 0xC1, +}; + +/// deletion-vector-v1 blob for positions {7, 9} (cardinality 2). +constexpr UInt8 second_two_position_dv_blob[] = { + 0x00, 0x00, 0x00, 0x24, 0xD1, 0xD3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x3A, 0x30, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, + 0x10, 0x00, 0x00, 0x00, 0x07, 0x00, 0x09, 0x00, 0xB7, 0xB0, 0x20, 0xFF, +}; + +/// Coalesced Puffin with two equal-cardinality DVs (same fixture as footer-bind tests). +constexpr UInt8 two_equal_cardinality_dvs_puffin[] = { + 0x50, 0x46, 0x41, 0x31, 0x00, 0x00, 0x00, 0x24, 0xD1, 0xD3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x3A, 0x30, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x01, 0x00, 0x10, 0x00, 0x00, 0x00, 0x02, 0x00, 0x05, 0x00, 0x2C, 0xDB, 0x9F, 0xC1, + 0x00, 0x00, 0x00, 0x24, 0xD1, 0xD3, 0x39, 0x64, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x3A, 0x30, 0x00, 0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, + 0x10, 0x00, 0x00, 0x00, 0x07, 0x00, 0x09, 0x00, 0xB7, 0xB0, 0x20, 0xFF, 0x50, 0x46, 0x41, 0x31, + 0x7B, 0x22, 0x62, 0x6C, 0x6F, 0x62, 0x73, 0x22, 0x3A, 0x20, 0x5B, 0x7B, 0x22, 0x74, 0x79, 0x70, + 0x65, 0x22, 0x3A, 0x20, 0x22, 0x64, 0x65, 0x6C, 0x65, 0x74, 0x69, 0x6F, 0x6E, 0x2D, 0x76, 0x65, + 0x63, 0x74, 0x6F, 0x72, 0x2D, 0x76, 0x31, 0x22, 0x2C, 0x20, 0x22, 0x66, 0x69, 0x65, 0x6C, 0x64, + 0x73, 0x22, 0x3A, 0x20, 0x5B, 0x5D, 0x2C, 0x20, 0x22, 0x73, 0x6E, 0x61, 0x70, 0x73, 0x68, 0x6F, + 0x74, 0x2D, 0x69, 0x64, 0x22, 0x3A, 0x20, 0x2D, 0x31, 0x2C, 0x20, 0x22, 0x73, 0x65, 0x71, 0x75, + 0x65, 0x6E, 0x63, 0x65, 0x2D, 0x6E, 0x75, 0x6D, 0x62, 0x65, 0x72, 0x22, 0x3A, 0x20, 0x2D, 0x31, + 0x2C, 0x20, 0x22, 0x6F, 0x66, 0x66, 0x73, 0x65, 0x74, 0x22, 0x3A, 0x20, 0x34, 0x2C, 0x20, 0x22, + 0x6C, 0x65, 0x6E, 0x67, 0x74, 0x68, 0x22, 0x3A, 0x20, 0x34, 0x34, 0x2C, 0x20, 0x22, 0x70, 0x72, + 0x6F, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, 0x22, 0x3A, 0x20, 0x7B, 0x22, 0x72, 0x65, 0x66, + 0x65, 0x72, 0x65, 0x6E, 0x63, 0x65, 0x64, 0x2D, 0x64, 0x61, 0x74, 0x61, 0x2D, 0x66, 0x69, 0x6C, + 0x65, 0x22, 0x3A, 0x20, 0x22, 0x2F, 0x64, 0x61, 0x74, 0x61, 0x2F, 0x66, 0x69, 0x6C, 0x65, 0x5F, + 0x61, 0x2E, 0x70, 0x61, 0x72, 0x71, 0x75, 0x65, 0x74, 0x22, 0x2C, 0x20, 0x22, 0x63, 0x61, 0x72, + 0x64, 0x69, 0x6E, 0x61, 0x6C, 0x69, 0x74, 0x79, 0x22, 0x3A, 0x20, 0x22, 0x32, 0x22, 0x7D, 0x7D, + 0x2C, 0x20, 0x7B, 0x22, 0x74, 0x79, 0x70, 0x65, 0x22, 0x3A, 0x20, 0x22, 0x64, 0x65, 0x6C, 0x65, + 0x74, 0x69, 0x6F, 0x6E, 0x2D, 0x76, 0x65, 0x63, 0x74, 0x6F, 0x72, 0x2D, 0x76, 0x31, 0x22, 0x2C, + 0x20, 0x22, 0x66, 0x69, 0x65, 0x6C, 0x64, 0x73, 0x22, 0x3A, 0x20, 0x5B, 0x5D, 0x2C, 0x20, 0x22, + 0x73, 0x6E, 0x61, 0x70, 0x73, 0x68, 0x6F, 0x74, 0x2D, 0x69, 0x64, 0x22, 0x3A, 0x20, 0x2D, 0x31, + 0x2C, 0x20, 0x22, 0x73, 0x65, 0x71, 0x75, 0x65, 0x6E, 0x63, 0x65, 0x2D, 0x6E, 0x75, 0x6D, 0x62, + 0x65, 0x72, 0x22, 0x3A, 0x20, 0x2D, 0x31, 0x2C, 0x20, 0x22, 0x6F, 0x66, 0x66, 0x73, 0x65, 0x74, + 0x22, 0x3A, 0x20, 0x34, 0x38, 0x2C, 0x20, 0x22, 0x6C, 0x65, 0x6E, 0x67, 0x74, 0x68, 0x22, 0x3A, + 0x20, 0x34, 0x34, 0x2C, 0x20, 0x22, 0x70, 0x72, 0x6F, 0x70, 0x65, 0x72, 0x74, 0x69, 0x65, 0x73, + 0x22, 0x3A, 0x20, 0x7B, 0x22, 0x72, 0x65, 0x66, 0x65, 0x72, 0x65, 0x6E, 0x63, 0x65, 0x64, 0x2D, + 0x64, 0x61, 0x74, 0x61, 0x2D, 0x66, 0x69, 0x6C, 0x65, 0x22, 0x3A, 0x20, 0x22, 0x2F, 0x64, 0x61, + 0x74, 0x61, 0x2F, 0x66, 0x69, 0x6C, 0x65, 0x5F, 0x62, 0x2E, 0x70, 0x61, 0x72, 0x71, 0x75, 0x65, + 0x74, 0x22, 0x2C, 0x20, 0x22, 0x63, 0x61, 0x72, 0x64, 0x69, 0x6E, 0x61, 0x6C, 0x69, 0x74, 0x79, + 0x22, 0x3A, 0x20, 0x22, 0x32, 0x22, 0x7D, 0x7D, 0x5D, 0x7D, 0x9A, 0x01, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x50, 0x46, 0x41, 0x31, +}; + +std::vector makeDeltaBin(std::initializer_list blobs) +{ + std::vector file; + file.push_back(0x01); + for (std::string_view blob : blobs) + file.insert(file.end(), blob.begin(), blob.end()); + return file; +} + +} + +TEST(DeltaBinDeletionVector, DetectsSliceOnlyAndDecodesSingleBlob) +{ + const auto file_bytes = makeDeltaBin({std::string_view( + reinterpret_cast(two_position_dv_blob), sizeof(two_position_dv_blob))}); + ReadBufferFromMemory file(file_bytes.data(), file_bytes.size()); + + EXPECT_EQ( + detectIcebergDeletionVectorContainer( + file, file_bytes.size(), /*content_offset=*/1, static_cast(sizeof(two_position_dv_blob)), "deletion_vector.bin"), + IcebergDeletionVectorContainer::SliceOnly); + + const auto positions = readDeletionVectorFromPuffin( + file, /*offset=*/1, static_cast(sizeof(two_position_dv_blob)), /*expected_cardinality=*/2); + ASSERT_EQ(positions.size(), 2u); + EXPECT_EQ(positions[0], 2u); + EXPECT_EQ(positions[1], 5u); +} + +TEST(DeltaBinDeletionVector, DecodesTwoSlicesInOneFile) +{ + const auto file_bytes = makeDeltaBin( + {std::string_view(reinterpret_cast(two_position_dv_blob), sizeof(two_position_dv_blob)), + std::string_view(reinterpret_cast(second_two_position_dv_blob), sizeof(second_two_position_dv_blob))}); + ReadBufferFromMemory file(file_bytes.data(), file_bytes.size()); + + const Int64 first_offset = 1; + const Int64 second_offset = 1 + static_cast(sizeof(two_position_dv_blob)); + + EXPECT_EQ( + detectIcebergDeletionVectorContainer( + file, file_bytes.size(), second_offset, static_cast(sizeof(second_two_position_dv_blob)), "deletion_vector.bin"), + IcebergDeletionVectorContainer::SliceOnly); + + const auto first = readDeletionVectorFromPuffin( + file, first_offset, static_cast(sizeof(two_position_dv_blob)), /*expected_cardinality=*/2); + ASSERT_EQ(first.size(), 2u); + EXPECT_EQ(first[0], 2u); + EXPECT_EQ(first[1], 5u); + + const auto second = readDeletionVectorFromPuffin( + file, second_offset, static_cast(sizeof(second_two_position_dv_blob)), /*expected_cardinality=*/2); + ASSERT_EQ(second.size(), 2u); + EXPECT_EQ(second[0], 7u); + EXPECT_EQ(second[1], 9u); +} + +TEST(DeltaBinDeletionVector, RejectsGarbageHeaderAndInvalidEnvelope) +{ + /// Parquet-like magic; slice at offset 1 is not a deletion-vector-v1 envelope. + /// File must be large enough that offset+length is in bounds so detection reaches magic checks. + constexpr UInt8 garbage[16] = {0x50, 0x41, 0x52, 0x31, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}; + ReadBufferFromMemory file(garbage, sizeof(garbage)); + + try + { + detectIcebergDeletionVectorContainer( + file, sizeof(garbage), /*content_offset=*/1, /*content_size_in_bytes=*/12, "not-a-dv.bin"); + FAIL() << "Expected exception"; + } + catch (const Exception & e) + { + EXPECT_EQ(e.code(), ErrorCodes::BAD_ARGUMENTS); + EXPECT_NE(e.message().find("neither a Puffin container"), std::string::npos); + EXPECT_NE(e.message().find("50 41 52 31"), std::string::npos); + EXPECT_NE(e.message().find("not-a-dv.bin"), std::string::npos); + EXPECT_NE(e.message().find("offset 1"), std::string::npos); + } +} + +TEST(DeltaBinDeletionVector, PuffinHeaderStillTakesFooterBindPath) +{ + ReadBufferFromMemory file(two_equal_cardinality_dvs_puffin, sizeof(two_equal_cardinality_dvs_puffin)); + EXPECT_EQ( + detectIcebergDeletionVectorContainer( + file, sizeof(two_equal_cardinality_dvs_puffin), /*content_offset=*/4, /*content_size_in_bytes=*/44, "dv.puffin"), + IcebergDeletionVectorContainer::Puffin); + + const auto blobs = readPuffinFooterBlobsFromSeekable(file, sizeof(two_equal_cardinality_dvs_puffin)); + const auto & blob = bindDeletionVectorBlob(blobs, /*content_offset=*/4, /*content_size_in_bytes=*/44, "/data/file_a.parquet", 2); + EXPECT_EQ(blob.type, "deletion-vector-v1"); + + const auto positions = readDeletionVectorFromPuffin(file, 4, 44, /*expected_cardinality=*/2); + ASSERT_EQ(positions.size(), 2u); + EXPECT_EQ(positions[0], 2u); + EXPECT_EQ(positions[1], 5u); +} + +TEST(DeltaBinDeletionVector, AcceptsBareEnvelopeWithoutVersionByte) +{ + ReadBufferFromMemory file(two_position_dv_blob, sizeof(two_position_dv_blob)); + EXPECT_EQ( + detectIcebergDeletionVectorContainer( + file, + sizeof(two_position_dv_blob), + /*content_offset=*/0, + static_cast(sizeof(two_position_dv_blob)), + "bare-envelope.bin"), + IcebergDeletionVectorContainer::SliceOnly); +} diff --git a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_dv_referenced_data_file.cpp b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_dv_referenced_data_file.cpp index dba4b03f5244..568ca89a0797 100644 --- a/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_dv_referenced_data_file.cpp +++ b/src/Storages/ObjectStorage/DataLakes/tests/gtest_puffin_dv_referenced_data_file.cpp @@ -8,6 +8,11 @@ #include #include +#include +#include +#include +#include + using namespace DB; using namespace DB::Iceberg; @@ -68,4 +73,43 @@ TEST(PuffinDeletionVectorReferencedDataFile, RejectsEmptyPath) } } +namespace +{ + +std::shared_ptr makePositionDeleteEntry( + String file_format, std::optional content_offset, std::optional content_size_in_bytes) +{ + return std::make_shared( + FileContentType::POSITION_DELETE, + IcebergPathFromMetadata::deserialize("s3://bucket/deletes/dv.bin"), + /*row_number=*/0, + ManifestEntryStatus::ADDED, + /*written_sequence_number=*/std::nullopt, + /*written_snapshot_id=*/std::nullopt, + DB::Row{}, + std::unordered_map{}, + std::unordered_map>{}, + std::move(file_format), + IcebergPathFromMetadata::deserialize("s3://bucket/data/file.parquet"), + IcebergPathFromMetadata::deserialize("s3://bucket/data/file.parquet"), + /*equality_ids=*/std::nullopt, + /*sort_order_id=*/std::nullopt, + /*record_count=*/2, + /*file_size_in_bytes=*/45, + content_offset, + content_size_in_bytes); +} + +} + +TEST(IcebergDeletionVectorClassification, OffsetsClassifyAsDeletionVectorRegardlessOfFileFormat) +{ + EXPECT_TRUE(makePositionDeleteEntry("puffin", 1, 44)->isDeletionVector()); + EXPECT_TRUE(makePositionDeleteEntry("PUFFIN", 4, 44)->isDeletionVector()); + EXPECT_TRUE(makePositionDeleteEntry("parquet", 1, 44)->isDeletionVector()); + EXPECT_FALSE(makePositionDeleteEntry("puffin", std::nullopt, std::nullopt)->isDeletionVector()); + EXPECT_FALSE(makePositionDeleteEntry("parquet", std::nullopt, std::nullopt)->isDeletionVector()); + EXPECT_FALSE(makePositionDeleteEntry("parquet", 1, std::nullopt)->isDeletionVector()); +} + #endif diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/.gitignore b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/.gitignore new file mode 100644 index 000000000000..cea42dcdcd26 --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/.gitignore @@ -0,0 +1,2 @@ +# Root .gitignore ignores `*.bin`. Keep Databricks UniForm deletion-vector objects. +!*.bin diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/README.md b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/README.md new file mode 100644 index 000000000000..356bdfea265d --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/README.md @@ -0,0 +1,13 @@ +# Databricks UniForm Iceberg v3 + Delta `.bin` deletion vectors + +Customer-provided table (`altinity_dv_puffin_repro_v3.zip`). Databricks UniForm writes Iceberg +metadata under `_iceberg/metadata/` and data / deletion vectors at the table root. + +- Schema: `Id String`, `Name String`, `UpdatedAt timestamptz` +- 10000 rows (`Id` `"1"` .. `"10000"`) +- `DELETE WHERE cast(Id as int) <= 1000` → 1000 positions in `deletion_vector_*.bin` +- Delete-manifest entry: `file_format = PUFFIN`, `content_offset = 1`, `content_size_in_bytes = 251` +- Object header is Delta version `0x01`, not Puffin `PFA1` + +The integration test copies this tree, flattens `_iceberg/metadata` to `metadata/`, and rewrites +Databricks `s3://` URIs onto the ClickHouse test warehouse path. diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00000-102ad1cf-eb84-47b8-ae93-31b30e5d8359.gz.metadata.json b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00000-102ad1cf-eb84-47b8-ae93-31b30e5d8359.gz.metadata.json new file mode 100644 index 000000000000..287f3d218749 Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00000-102ad1cf-eb84-47b8-ae93-31b30e5d8359.gz.metadata.json differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00001-364c19a9-7e98-46cf-9c64-8d0cdf419c13.gz.metadata.json b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00001-364c19a9-7e98-46cf-9c64-8d0cdf419c13.gz.metadata.json new file mode 100644 index 000000000000..c52c0f805bea Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00001-364c19a9-7e98-46cf-9c64-8d0cdf419c13.gz.metadata.json differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00002-8de40d94-9846-4a90-bf5a-80cf5e93be12.gz.metadata.json b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00002-8de40d94-9846-4a90-bf5a-80cf5e93be12.gz.metadata.json new file mode 100644 index 000000000000..2b1be6fa2ffa Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/00002-8de40d94-9846-4a90-bf5a-80cf5e93be12.gz.metadata.json differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/7fc1a95b-41b1-4c5c-a585-cd0480cc6739-m0.avro b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/7fc1a95b-41b1-4c5c-a585-cd0480cc6739-m0.avro new file mode 100644 index 000000000000..edf39d414bbd Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/7fc1a95b-41b1-4c5c-a585-cd0480cc6739-m0.avro differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/e5531858-556b-45f9-a989-a7ba0b0ae967-m0.avro b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/e5531858-556b-45f9-a989-a7ba0b0ae967-m0.avro new file mode 100644 index 000000000000..c49e7340201c Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/e5531858-556b-45f9-a989-a7ba0b0ae967-m0.avro differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-4017675268503509610-1-7fc1a95b-41b1-4c5c-a585-cd0480cc6739.avro b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-4017675268503509610-1-7fc1a95b-41b1-4c5c-a585-cd0480cc6739.avro new file mode 100644 index 000000000000..dc9dbead027c Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-4017675268503509610-1-7fc1a95b-41b1-4c5c-a585-cd0480cc6739.avro differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-6006002030350834282-1-e5531858-556b-45f9-a989-a7ba0b0ae967.avro b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-6006002030350834282-1-e5531858-556b-45f9-a989-a7ba0b0ae967.avro new file mode 100644 index 000000000000..4cd9d9d485fe Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/_iceberg/metadata/snap-6006002030350834282-1-e5531858-556b-45f9-a989-a7ba0b0ae967.avro differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/deletion_vector_3d169d0e-b938-4854-8cca-d9d38949f1b0.bin b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/deletion_vector_3d169d0e-b938-4854-8cca-d9d38949f1b0.bin new file mode 100644 index 000000000000..5234bf016097 Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/deletion_vector_3d169d0e-b938-4854-8cca-d9d38949f1b0.bin differ diff --git a/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/jk/part-00000-c0585bde-5e01-4bc7-b94d-2e5639a05e68.c000.zstd.parquet b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/jk/part-00000-c0585bde-5e01-4bc7-b94d-2e5639a05e68.c000.zstd.parquet new file mode 100644 index 000000000000..f041bcd533a5 Binary files /dev/null and b/tests/integration/test_storage_iceberg_with_spark/data/altinity_dv_puffin_repro_v3/jk/part-00000-c0585bde-5e01-4bc7-b94d-2e5639a05e68.c000.zstd.parquet differ diff --git a/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors.py b/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors.py index b6b42786c07a..1d5fd3e64d63 100644 --- a/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors.py +++ b/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors.py @@ -1,3 +1,5 @@ +import json +import os import uuid import pytest @@ -40,6 +42,223 @@ def _strip_file_uri_scheme(path): return path +PUFFIN_MAGIC = b"PFA1" +PUFFIN_FOOTER_TRAILER_SIZE = 12 +ICEBERG_WAREHOUSE = "/var/lib/clickhouse/user_files/iceberg_data" + + +def _avro_long(value): + if value is None: + return None + if isinstance(value, dict): + if "long" in value: + return int(value["long"]) + if not value: + return None + return int(next(iter(value.values()))) + return int(value) + + +def _read_avro_file(path): + import avro.datafile + import avro.io + + with open(path, "rb") as f: + reader = avro.datafile.DataFileReader(f, avro.io.DatumReader()) + schema = reader.datum_reader.writers_schema + meta = dict(reader.meta) + codec = reader.codec + records = list(reader) + reader.close() + return records, schema, meta, codec + + +def _write_avro_file(path, records, schema, meta, codec): + import avro.datafile + import avro.io + + with open(path, "wb") as f: + writer = avro.datafile.DataFileWriter(f, avro.io.DatumWriter(), schema, codec=codec) + for key, value in meta.items(): + if key not in ("avro.schema", "avro.codec"): + writer.set_meta(key, value) + for record in records: + writer.append(record) + writer.flush() + writer.close() + + +def _resolve_under_table(table_dir, path): + path = _strip_file_uri_scheme(path) + if os.path.isfile(path): + return path + joined = os.path.join(table_dir, path.lstrip("/")) + if os.path.isfile(joined): + return joined + raise RuntimeError(f"Cannot resolve Iceberg path '{path}' under '{table_dir}'") + + +def _current_metadata(table_dir): + metadata_dir = os.path.join(table_dir, "metadata") + hint_path = os.path.join(metadata_dir, "version-hint.text") + if os.path.isfile(hint_path): + with open(hint_path, encoding="utf-8") as f: + version = f.read().strip() + if version.startswith("v"): + version = version[1:] + path = os.path.join(metadata_dir, f"v{version}.metadata.json") + if os.path.isfile(path): + with open(path, encoding="utf-8") as f: + return json.load(f) + + candidates = [ + os.path.join(metadata_dir, name) + for name in os.listdir(metadata_dir) + if name.endswith(".metadata.json") + ] + if not candidates: + raise RuntimeError(f"No metadata JSON under {metadata_dir}") + + best = None + best_ts = -1 + for path in candidates: + with open(path, encoding="utf-8") as f: + meta = json.load(f) + ts = int(meta.get("last-updated-ms") or 0) + if ts >= best_ts: + best = meta + best_ts = ts + return best + + +def _puffin_to_delta_bin(data): + """Extract deletion-vector-v1 blobs from a Puffin file into a Delta `.bin` envelope. + + Databricks UniForm / Delta DV objects are `version=1` plus the same + `deletion-vector-v1` bytes Iceberg stores inside Puffin. Manifest + `content_offset` for a single blob becomes 1. + """ + if len(data) < 16 or data[:4] != PUFFIN_MAGIC or data[-4:] != PUFFIN_MAGIC: + raise ValueError("not a Puffin file") + + footer_length = int.from_bytes(data[-12:-8], "little", signed=True) + flags = data[-8:-4] + if flags[0] & 0x01: + raise ValueError("compressed Puffin footers are not supported by this test helper") + if footer_length <= 0: + raise ValueError(f"invalid Puffin footer length {footer_length}") + + payload_start = len(data) - PUFFIN_FOOTER_TRAILER_SIZE - footer_length + if payload_start < 8 or data[payload_start - 4 : payload_start] != PUFFIN_MAGIC: + raise ValueError("invalid Puffin footer magic") + + footer = json.loads(data[payload_start : payload_start + footer_length]) + out = bytearray([0x01]) + offset_map = {} + for blob in footer.get("blobs", []): + if blob.get("type") != "deletion-vector-v1": + continue + old_offset = int(blob["offset"]) + length = int(blob["length"]) + slice_bytes = data[old_offset : old_offset + length] + if len(slice_bytes) != length: + raise ValueError(f"truncated Puffin blob at offset {old_offset}") + offset_map[old_offset] = len(out) + out.extend(slice_bytes) + + if not offset_map: + raise ValueError("Puffin file has no deletion-vector-v1 blobs") + return bytes(out), offset_map + + +def convert_spark_puffin_dvs_to_delta_bin(table_name): + """Rewrite Spark Iceberg v3 Puffin DVs as Databricks-style `.bin` files. + + Spark `DELETE` on format-version 3 writes `PFA1` Puffin. Databricks UniForm + writes `deletion_vector_.bin` (version byte + envelope) and still + stores `content_offset` / `content_size_in_bytes` on the Iceberg delete + manifest. Overwrite each referenced Puffin in place and retarget offsets + so ClickHouse must take the slice-only path (skip the Puffin footer). + """ + table_dir = os.path.join(ICEBERG_WAREHOUSE, "default", table_name) + metadata = _current_metadata(table_dir) + snapshot_id = metadata.get("current-snapshot-id") + snapshot = next((s for s in metadata.get("snapshots", []) if s.get("snapshot-id") == snapshot_id), None) + if not snapshot: + raise RuntimeError(f"Snapshot {snapshot_id} not found for {table_name}") + + manifest_list_path = _resolve_under_table(table_dir, snapshot["manifest-list"]) + manifest_list_records, _, _, _ = _read_avro_file(manifest_list_path) + manifest_paths = [ + _resolve_under_table(table_dir, record["manifest_path"]) + for record in manifest_list_records + if "manifest_path" in record + ] + + dv_paths = set() + for manifest_path in manifest_paths: + records, _, _, _ = _read_avro_file(manifest_path) + for record in records: + data_file = record.get("data_file") or {} + if _avro_long(data_file.get("content_offset")) is None: + continue + if _avro_long(data_file.get("content_size_in_bytes")) is None: + continue + dv_paths.add(_resolve_under_table(table_dir, data_file["file_path"])) + + if not dv_paths: + raise RuntimeError(f"No Iceberg deletion-vector entries found under {table_dir}") + + conversions = {} + for path in sorted(dv_paths): + with open(path, "rb") as f: + original = f.read() + if original[:4] != PUFFIN_MAGIC: + raise RuntimeError(f"Expected Puffin magic in '{path}', got {original[:4]!r}") + new_bytes, offset_map = _puffin_to_delta_bin(original) + if new_bytes[:4] == PUFFIN_MAGIC: + raise RuntimeError(f"Converted DV '{path}' still starts with Puffin magic") + with open(path, "wb") as f: + f.write(new_bytes) + conversions[path] = (offset_map, len(new_bytes)) + + patched_entries = 0 + for manifest_path in manifest_paths: + records, schema, meta, codec = _read_avro_file(manifest_path) + changed = False + for record in records: + data_file = record.get("data_file") + if not data_file: + continue + content_offset = _avro_long(data_file.get("content_offset")) + content_size = _avro_long(data_file.get("content_size_in_bytes")) + if content_offset is None or content_size is None: + continue + local_path = _resolve_under_table(table_dir, data_file["file_path"]) + if local_path not in conversions: + continue + offset_map, new_size = conversions[local_path] + if content_offset not in offset_map: + raise RuntimeError( + f"content_offset {content_offset} missing from converted Puffin '{local_path}'" + ) + new_offset = offset_map[content_offset] + if new_offset + content_size > new_size: + raise RuntimeError( + f"Delta .bin slice [{new_offset}, {new_offset + content_size}) exceeds file size {new_size}" + ) + data_file["content_offset"] = new_offset + data_file["file_size_in_bytes"] = new_size + changed = True + patched_entries += 1 + if changed: + _write_avro_file(manifest_path, records, schema, meta, codec) + + if patched_entries == 0: + raise RuntimeError(f"Failed to patch deletion-vector manifests for {table_name}") + return len(conversions) + + def add_equality_deletes_by_id(spark, table_name, ids): """Commit an Iceberg equality-delete file for the given `id` values. @@ -143,6 +362,68 @@ def test_deletion_vectors(started_cluster_iceberg_with_spark, storage_type, run_ ] +@pytest.mark.parametrize("run_on_cluster", [False, True]) +@pytest.mark.parametrize("storage_type", ["s3", "azure", "local"]) +def test_deletion_vectors_delta_bin(started_cluster_iceberg_with_spark, storage_type, run_on_cluster): + """Iceberg v3 DVs stored as Delta `.bin` files (Databricks UniForm), not Puffin. + + Spark Iceberg DELETE writes `PFA1` Puffin. Convert those objects to the + Databricks layout (`0x01` + deletion-vector-v1 envelope, `content_offset=1`) + so ClickHouse must skip the Puffin footer and decode the slice in place. + """ + if storage_type == "local" and run_on_cluster: + pytest.skip("Local storage with cluster execution is not supported") + + instance = started_cluster_iceberg_with_spark.instances["node1"] + spark = started_cluster_iceberg_with_spark.spark_session + table_name = "test_deletion_vectors_delta_bin_" + storage_type + "_" + get_uuid_str() + deleted_ids = [2, 5, 7, 100] + + spark.sql( + f""" + CREATE TABLE {table_name} (id bigint) USING iceberg + TBLPROPERTIES ( + 'format-version' = '3', + 'write.delete.mode' = 'merge-on-read', + 'write.update.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read' + ) + """ + ) + spark.sql(f"INSERT INTO {table_name} SELECT id FROM range(0, 200)") + spark.sql( + f"DELETE FROM {table_name} WHERE id IN ({', '.join(str(x) for x in deleted_ids)})" + ) + + converted = convert_spark_puffin_dvs_to_delta_bin(table_name) + assert converted >= 1 + + upload_table(started_cluster_iceberg_with_spark, storage_type, table_name) + + expression = get_creation_expression( + storage_type, + table_name, + started_cluster_iceberg_with_spark, + run_on_cluster=run_on_cluster, + table_function=True, + ) + + expected = [x for x in range(200) if x not in deleted_ids] + settings = {"use_iceberg_metadata_files_cache": 0, "use_puffin_files_cache": 0} + + assert int(instance.query(f"SELECT count() FROM {expression}", settings=settings)) == len(expected) + assert get_array(instance.query(f"SELECT id FROM {expression}", settings=settings)) == expected + assert ( + int( + instance.query( + f"SELECT count() FROM {expression}", + settings={**settings, "optimize_trivial_count_query": 1}, + ) + ) + == len(expected) + ) + + @pytest.mark.parametrize("run_on_cluster", [False, True]) @pytest.mark.parametrize("storage_type", ["s3", "azure", "local"]) def test_deletion_vectors_aggregates(started_cluster_iceberg_with_spark, storage_type, run_on_cluster): diff --git a/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors_databricks_uniform.py b/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors_databricks_uniform.py new file mode 100644 index 000000000000..76baa2d11826 --- /dev/null +++ b/tests/integration/test_storage_iceberg_with_spark/test_deletion_vectors_databricks_uniform.py @@ -0,0 +1,180 @@ +import gzip +import json +import os +import shutil + +import pytest + +from helpers.iceberg_utils import ( + default_upload_directory, + get_creation_expression, + get_uuid_str, +) + + +DATABRICKS_TABLE_PREFIX = ( + "s3://ttd-unity-useast/env=test/__unitystorage/catalogs/" + "c137d337-b630-4388-b072-9dc88ee567d6/tables/c771d2d4-eac8-401b-b26d-232b2b8fae5a" +) + +FIXTURE_DIR = os.path.join( + os.path.dirname(__file__), "data", "altinity_dv_puffin_repro_v3" +) + +DV_BIN_NAME = "deletion_vector_3d169d0e-b938-4854-8cca-d9d38949f1b0.bin" +EXPECTED_LIVE_IDS = list(range(1001, 10001)) + + +def get_int_array(query_result: str): + return [int(x) for x in query_result.strip().split("\n") if x] + + +def upload_table(cluster, storage_type, table_name): + default_upload_directory( + cluster, + storage_type, + f"/iceberg_data/default/{table_name}/", + f"/iceberg_data/default/{table_name}/", + ) + + +def _replace_strings(obj, old, new): + if isinstance(obj, str): + value = obj.replace(old, new) + value = value.replace("/_iceberg/metadata/", "/metadata/") + if value.endswith("/_iceberg"): + return value[: -len("/_iceberg")] + return value + if isinstance(obj, list): + return [_replace_strings(item, old, new) for item in obj] + if isinstance(obj, dict): + return {key: _replace_strings(value, old, new) for key, value in obj.items()} + return obj + + +def _rewrite_avro_paths(path, old, new): + import avro.datafile + import avro.io + + with open(path, "rb") as handle: + reader = avro.datafile.DataFileReader(handle, avro.io.DatumReader()) + schema = reader.datum_reader.writers_schema + meta = dict(reader.meta) + codec = reader.codec + records = [_replace_strings(record, old, new) for record in reader] + reader.close() + + with open(path, "wb") as handle: + writer = avro.datafile.DataFileWriter(handle, avro.io.DatumWriter(), schema, codec=codec) + for key, value in meta.items(): + if key not in ("avro.schema", "avro.codec"): + writer.set_meta(key, value) + for record in records: + writer.append(record) + writer.flush() + writer.close() + + +def prepare_databricks_uniform_table(dest_dir): + """Copy the UniForm fixture and rewrite Databricks URIs onto `dest_dir`. + + Iceberg `location` is `_iceberg` while data files live on the parent table root. + Flatten metadata to `metadata/` and set `location` to the parent so + `IcebergPathResolver` can map both data files and the `.bin` DV. + """ + if os.path.exists(dest_dir): + shutil.rmtree(dest_dir) + os.makedirs(dest_dir) + + shutil.copy2(os.path.join(FIXTURE_DIR, DV_BIN_NAME), os.path.join(dest_dir, DV_BIN_NAME)) + shutil.copytree(os.path.join(FIXTURE_DIR, "jk"), os.path.join(dest_dir, "jk")) + shutil.copytree( + os.path.join(FIXTURE_DIR, "_iceberg", "metadata"), + os.path.join(dest_dir, "metadata"), + ) + + with open(os.path.join(dest_dir, DV_BIN_NAME), "rb") as handle: + header = handle.read(4) + assert header[:1] == b"\x01", header + assert header != b"PFA1", header + + new_prefix = dest_dir.rstrip("/") + metadata_dir = os.path.join(dest_dir, "metadata") + for name in os.listdir(metadata_dir): + path = os.path.join(metadata_dir, name) + if name.endswith(".gz.metadata.json"): + with gzip.open(path, "rt", encoding="utf-8") as handle: + text = handle.read() + text = text.replace(DATABRICKS_TABLE_PREFIX, new_prefix) + text = text.replace("/_iceberg/metadata/", "/metadata/") + metadata = json.loads(text) + location = metadata.get("location", "") + if location.endswith("/_iceberg"): + metadata["location"] = location[: -len("/_iceberg")] + with gzip.open(path, "wt", encoding="utf-8") as handle: + json.dump(metadata, handle, separators=(",", ":")) + elif name.endswith(".avro"): + _rewrite_avro_paths(path, DATABRICKS_TABLE_PREFIX, new_prefix) + + return dest_dir + + +@pytest.mark.parametrize("run_on_cluster", [False, True]) +@pytest.mark.parametrize("storage_type", ["s3", "azure", "local"]) +def test_deletion_vectors_databricks_uniform_bin( + started_cluster_iceberg_with_spark, storage_type, run_on_cluster +): + """Read a Databricks UniForm Iceberg v3 table whose DVs are Delta `.bin` files. + + Spark Iceberg writes Puffin; this fixture is the customer layout: `file_format=PUFFIN`, + `content_offset=1`, object bytes `0x01` + deletion-vector-v1 envelope. + """ + if storage_type == "local" and run_on_cluster: + pytest.skip("Local storage with cluster execution is not supported") + + instance = started_cluster_iceberg_with_spark.instances["node1"] + table_name = "altinity_dv_puffin_repro_v3_" + storage_type + "_" + get_uuid_str() + dest_dir = f"/var/lib/clickhouse/user_files/iceberg_data/default/{table_name}" + prepare_databricks_uniform_table(dest_dir) + upload_table(started_cluster_iceberg_with_spark, storage_type, table_name) + + expression = get_creation_expression( + storage_type, + table_name, + started_cluster_iceberg_with_spark, + run_on_cluster=run_on_cluster, + table_function=True, + format_version=3, + ) + settings = {"use_iceberg_metadata_files_cache": 0, "use_puffin_files_cache": 0} + + assert int(instance.query(f"SELECT count() FROM {expression}", settings=settings)) == 9000 + assert ( + int(instance.query(f"SELECT min(toInt64(Id)) FROM {expression}", settings=settings)) + == 1001 + ) + assert ( + int(instance.query(f"SELECT max(toInt64(Id)) FROM {expression}", settings=settings)) + == 10000 + ) + assert ( + int( + instance.query( + f"SELECT countIf(toInt64(Id) <= 1000) FROM {expression}", + settings=settings, + ) + ) + == 0 + ) + assert get_int_array( + instance.query(f"SELECT toInt64(Id) FROM {expression} ORDER BY toInt64(Id)", settings=settings) + ) == EXPECTED_LIVE_IDS + assert ( + int( + instance.query( + f"SELECT count() FROM {expression}", + settings={**settings, "optimize_trivial_count_query": 1}, + ) + ) + == 9000 + )