From dd3384a365a5ab12518afc33f17c1401b6c19272 Mon Sep 17 00:00:00 2001 From: xuanyili Date: Sat, 5 Sep 2026 17:19:33 +0000 Subject: [PATCH 1/2] test: add cross-language deletion vector fixtures --- dev/dv-fixtures/generate_go_fixtures.go | 152 +++++++++++ src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/puffin_dv_interop_test.cc | 256 ++++++++++++++++++ .../test/resources/deletion_vectors/README.md | 44 +++ .../go/all-container-types-dv.puffin | Bin 0 -> 16781 bytes .../deletion_vectors/go/multi-blob-dv.puffin | Bin 0 -> 650 bytes .../deletion_vectors/go/single-blob-dv.puffin | Bin 0 -> 314 bytes .../java/multi-blob-dv.puffin | Bin 0 -> 639 bytes .../java/single-blob-dv.puffin | Bin 0 -> 384 bytes 9 files changed, 453 insertions(+) create mode 100644 dev/dv-fixtures/generate_go_fixtures.go create mode 100644 src/iceberg/test/puffin_dv_interop_test.cc create mode 100644 src/iceberg/test/resources/deletion_vectors/README.md create mode 100644 src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin create mode 100644 src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin create mode 100644 src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin create mode 100644 src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin create mode 100644 src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin diff --git a/dev/dv-fixtures/generate_go_fixtures.go b/dev/dv-fixtures/generate_go_fixtures.go new file mode 100644 index 000000000..67d26d6c0 --- /dev/null +++ b/dev/dv-fixtures/generate_go_fixtures.go @@ -0,0 +1,152 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//go:build ignore + +package main + +import ( + "bytes" + "fmt" + "os" + "path/filepath" + "strconv" + + "github.com/apache/iceberg-go/puffin" + "github.com/apache/iceberg-go/table/dv" +) + +type fixtureBlob struct { + referencedDataFile string + positions []uint64 + ranges []positionRange +} + +type positionRange struct { + start uint64 + end uint64 +} + +func writeFixture(outputDir, fileName, createdBy string, blobs []fixtureBlob) error { + var output bytes.Buffer + writer, err := puffin.NewWriter(&output) + if err != nil { + return err + } + if err := writer.SetCreatedBy(createdBy); err != nil { + return err + } + + for _, blob := range blobs { + bitmap := dv.NewRoaringPositionBitmap() + for _, position := range blob.positions { + bitmap.Set(position) + } + for _, positionRange := range blob.ranges { + bitmap.SetRange(positionRange.start, positionRange.end) + } + payload, err := dv.SerializeDV(bitmap) + if err != nil { + return err + } + _, err = writer.AddBlob(puffin.BlobMetadataInput{ + Type: puffin.BlobTypeDeletionVector, + SnapshotID: -1, + SequenceNumber: -1, + Fields: []int32{}, + Properties: map[string]string{ + "referenced-data-file": blob.referencedDataFile, + "cardinality": strconv.FormatInt(bitmap.Cardinality(), 10), + }, + }, payload) + if err != nil { + return err + } + } + + if err := writer.Finish(); err != nil { + return err + } + return os.WriteFile(filepath.Join(outputDir, fileName), output.Bytes(), 0o644) +} + +func main() { + if len(os.Args) != 2 { + fmt.Fprintln(os.Stderr, "usage: go run generate_go_fixtures.go OUTPUT_DIR") + os.Exit(2) + } + outputDir := os.Args[1] + if err := os.MkdirAll(outputDir, 0o755); err != nil { + panic(err) + } + + err := writeFixture(outputDir, "single-blob-dv.puffin", + "iceberg-go test fixture", []fixtureBlob{{ + referencedDataFile: "data/test.parquet", + positions: []uint64{1, 3, 5, 7, 9}, + }}) + if err != nil { + panic(err) + } + + err = writeFixture(outputDir, "multi-blob-dv.puffin", + "iceberg-go cross-language fixture", []fixtureBlob{ + { + referencedDataFile: "s3://warehouse/db/table/data/go-file-001.parquet", + positions: []uint64{ + 0, 100, 200, (uint64(1) << 32) + 7, + }, + }, + { + referencedDataFile: "s3://warehouse/db/table/data/go-file-002.parquet", + positions: []uint64{ + 50, 150, (uint64(2) << 32) + 9, + }, + }, + }) + if err != nil { + panic(err) + } + + position := func(bucket, container, value uint64) uint64 { + return (bucket << 32) + (container << 16) + value + } + allContainerPositions := []uint64{ + position(0, 0, 5), + position(0, 0, 7), + position(1, 0, 10), + position(1, 0, 20), + } + for bucket := uint64(0); bucket < 2; bucket++ { + for value := uint64(0); value < 10000; value += 2 { + allContainerPositions = + append(allContainerPositions, position(bucket, 2, value)) + } + } + err = writeFixture(outputDir, "all-container-types-dv.puffin", + "iceberg-go cross-language fixture", []fixtureBlob{{ + referencedDataFile: "s3://warehouse/db/table/data/all-containers.parquet", + positions: allContainerPositions, + ranges: []positionRange{ + {start: position(0, 1, 1), end: position(0, 1, 1000)}, + {start: position(1, 1, 10), end: position(1, 1, 500)}, + }, + }}) + if err != nil { + panic(err) + } +} diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index 5ca9fd915..f5efed591 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -158,6 +158,7 @@ add_iceberg_test(util_test add_iceberg_test(puffin_test USE_DATA SOURCES + puffin_dv_interop_test.cc puffin_format_test.cc puffin_json_test.cc puffin_reader_writer_test.cc) diff --git a/src/iceberg/test/puffin_dv_interop_test.cc b/src/iceberg/test/puffin_dv_interop_test.cc new file mode 100644 index 000000000..b81dcdd51 --- /dev/null +++ b/src/iceberg/test/puffin_dv_interop_test.cc @@ -0,0 +1,256 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/puffin_reader.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/std_io.h" +#include "iceberg/test/test_resource.h" + +namespace iceberg::puffin { + +namespace { + +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +int64_t Position(int64_t bucket, int64_t container, int64_t value) { + return (bucket << 32) + (container << 16) + value; +} + +std::vector AllContainerPositions() { + std::vector positions = { + Position(0, 0, 5), + Position(0, 0, 7), + Position(1, 0, 10), + Position(1, 0, 20), + }; + positions.reserve(10004); + for (int64_t bucket = 0; bucket < 2; ++bucket) { + for (int64_t value = 0; value < 10000; value += 2) { + positions.push_back(Position(bucket, 2, value)); + } + } + return positions; +} + +struct PositionRange { + int64_t begin; + int64_t end; +}; + +struct ExpectedBlob { + std::string referenced_data_file; + std::vector input_fields; + int64_t offset; + int64_t length; + int64_t cardinality; + std::vector positions; + std::vector ranges; +}; + +std::shared_ptr MakeDeleteFile(const ExpectedBlob& expected, + const std::string& fixture_path, + int64_t file_size) { + return std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = fixture_path, + .file_format = FileFormatType::kPuffin, + .record_count = expected.cardinality, + .file_size_in_bytes = file_size, + .referenced_data_file = expected.referenced_data_file, + .content_offset = expected.offset, + .content_size_in_bytes = expected.length, + }); +} + +void AssertPositions(const ExpectedBlob& expected, + const std::shared_ptr& delete_file, + const std::shared_ptr& io) { + ASSERT_TRUE(delete_file->content_offset.has_value()); + ASSERT_TRUE(delete_file->content_size_in_bytes.has_value()); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(delete_file->file_path)); + ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); + std::vector data(static_cast(*delete_file->content_size_in_bytes)); + ASSERT_THAT(stream->ReadFully(*delete_file->content_offset, data), IsOk()); + + std::span blob(reinterpret_cast(data.data()), + data.size()); + ICEBERG_UNWRAP_OR_FAIL(auto positions, + PositionDeleteIndex::Deserialize(blob, delete_file)); + EXPECT_EQ(positions.Cardinality(), expected.cardinality); + int64_t expected_cardinality = static_cast(expected.positions.size()); + for (const auto& range : expected.ranges) { + expected_cardinality += range.end - range.begin; + } + ASSERT_EQ(expected_cardinality, expected.cardinality); + for (int64_t position : expected.positions) { + EXPECT_TRUE(positions.IsDeleted(position)) << "Missing position " << position; + } + for (const auto& range : expected.ranges) { + for (int64_t position = range.begin; position < range.end; ++position) { + ASSERT_TRUE(positions.IsDeleted(position)) << "Missing position " << position; + } + } +} + +void AssertFixture(const std::string& resource_name, + const std::vector& expected_blobs) { + const std::string fixture_path = GetResourcePath(resource_name); + const auto file_size = static_cast(std::filesystem::file_size(fixture_path)); + auto io = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(fixture_path)); + ICEBERG_UNWRAP_OR_FAIL(auto reader, PuffinReader::Make(std::move(input_file))); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, reader->ReadFileMetadata()); + + ASSERT_EQ(metadata.blobs.size(), expected_blobs.size()); + std::unordered_set seen; + for (const auto& blob : metadata.blobs) { + const auto& referenced_data_file = + blob.properties.at(std::string(kReferencedDataFileProperty)); + auto expected = std::ranges::find_if(expected_blobs, [&](const auto& candidate) { + return candidate.referenced_data_file == referenced_data_file; + }); + ASSERT_NE(expected, expected_blobs.end()) + << "Unexpected referenced data file " << referenced_data_file; + ASSERT_TRUE(seen.insert(referenced_data_file).second) + << "Duplicate referenced data file " << referenced_data_file; + + EXPECT_EQ(blob.type, StandardBlobTypes::kDeletionVectorV1); + EXPECT_EQ(blob.input_fields, expected->input_fields); + EXPECT_EQ(blob.snapshot_id, -1); + EXPECT_EQ(blob.sequence_number, -1); + EXPECT_EQ(blob.offset, expected->offset); + EXPECT_EQ(blob.length, expected->length); + EXPECT_TRUE(blob.compression_codec.empty()); + EXPECT_EQ(blob.properties.at(std::string(kCardinalityProperty)), + std::to_string(expected->cardinality)); + + AssertPositions(*expected, MakeDeleteFile(*expected, fixture_path, file_size), io); + } + EXPECT_EQ(seen.size(), expected_blobs.size()); +} + +} // namespace + +TEST(PuffinDVInteropTest, ReadsJavaSingleBlobFixture) { + AssertFixture( + "deletion_vectors/java/single-blob-dv.puffin", + {{ + .referenced_data_file = "s3://warehouse/db/table/data/00000-0-abc.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 4, + .length = 50, + .cardinality = 5, + .positions = {1, 3, 5, 7, 9}, + }}); +} + +TEST(PuffinDVInteropTest, ReadsJavaMultiBlobFixture) { + AssertFixture( + "deletion_vectors/java/multi-blob-dv.puffin", + { + { + .referenced_data_file = "s3://warehouse/db/table/data/file-001.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 4, + .length = 46, + .cardinality = 3, + .positions = {0, 100, 200}, + }, + { + .referenced_data_file = "s3://warehouse/db/table/data/file-002.parquet", + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .offset = 50, + .length = 44, + .cardinality = 2, + .positions = {50, 150}, + }, + }); +} + +TEST(PuffinDVInteropTest, ReadsGoSingleBlobFixture) { + AssertFixture("deletion_vectors/go/single-blob-dv.puffin", + {{ + .referenced_data_file = "data/test.parquet", + .input_fields = {}, + .offset = 4, + .length = 50, + .cardinality = 5, + .positions = {1, 3, 5, 7, 9}, + }}); +} + +TEST(PuffinDVInteropTest, ReadsGoMultiBlobFixture) { + AssertFixture( + "deletion_vectors/go/multi-blob-dv.puffin", + { + { + .referenced_data_file = "s3://warehouse/db/table/data/go-file-001.parquet", + .input_fields = {}, + .offset = 4, + .length = 68, + .cardinality = 4, + .positions = {0, 100, 200, (int64_t{1} << 32) + 7}, + }, + { + .referenced_data_file = "s3://warehouse/db/table/data/go-file-002.parquet", + .input_fields = {}, + .offset = 72, + .length = 66, + .cardinality = 3, + .positions = {50, 150, (int64_t{2} << 32) + 9}, + }, + }); +} + +TEST(PuffinDVInteropTest, ReadsGoAllContainerTypesFixture) { + AssertFixture( + "deletion_vectors/go/all-container-types-dv.puffin", + {{ + .referenced_data_file = "s3://warehouse/db/table/data/all-containers.parquet", + .input_fields = {}, + .offset = 4, + .length = 16466, + .cardinality = 11493, + .positions = AllContainerPositions(), + .ranges = + { + {Position(0, 1, 1), Position(0, 1, 1000)}, + {Position(1, 1, 10), Position(1, 1, 500)}, + }, + }}); +} + +} // namespace iceberg::puffin diff --git a/src/iceberg/test/resources/deletion_vectors/README.md b/src/iceberg/test/resources/deletion_vectors/README.md new file mode 100644 index 000000000..983cbfe3e --- /dev/null +++ b/src/iceberg/test/resources/deletion_vectors/README.md @@ -0,0 +1,44 @@ +# Deletion vector interoperability fixtures + +These fixtures verify deletion-vector compatibility against bytes produced by +implementations other than Iceberg C++. + +## Java fixtures + +The files under `java/` were copied byte-for-byte from Apache Iceberg Go commit +`3020adbbc3faff047da6f483f739f1b5e1de611b`, where they are consumed by +`table/dv/dv_cross_client_test.go`. + +- `single-blob-dv.puffin` and `multi-blob-dv.puffin` were produced by Apache + Iceberg Java using + `iceberg-go/dev/dv-fixtures/GenerateDVFixtures.java`. + +The committed Puffin bytes were reproduced against Apache Iceberg Java commit +`76d35b1e40f77edcad19646bb6afdd9f05249964`. To regenerate them, follow +`iceberg-go/dev/dv-fixtures/README.md` at the Iceberg Go commit above using +that Java revision, then copy the generated files into `java/`. + +## Go fixtures + +The files under `go/` are produced with the Iceberg Go Puffin and deletion +vector writers. From an Iceberg Go checkout at commit +`3020adbbc3faff047da6f483f739f1b5e1de611b`, run: + +```bash +go run /path/to/iceberg-cpp/dev/dv-fixtures/generate_go_fixtures.go \ + /path/to/iceberg-cpp/src/iceberg/test/resources/deletion_vectors/go +``` + +The multi-blob fixture includes positions in three distinct high-32-bit +buckets. `all-container-types-dv.puffin` covers array, run, and bitmap +containers inside a complete Puffin file. + +## Checksums + +```text +ab8309671c0c5ef1956f4f1d7b907f4a69ffeb154ab77d5a4855d7dd4779108c java/single-blob-dv.puffin +fed7edbb5a343a6c6fc4706ff4c213ab3f0a50916baeb228619f1f2c956f3f27 java/multi-blob-dv.puffin +dd293e827439cd22053a9356ae6d56f0d5d69e648b1850d39663cdc5c7ec5a77 go/single-blob-dv.puffin +000a77697d5ddce01e0242786b91fc15482a8b8f4fa3c8e214a936d7c2fc978b go/multi-blob-dv.puffin +d9f374706891b4780f8a59f155a7cc4cae4161b6415578abb4790f6eed793a52 go/all-container-types-dv.puffin +``` diff --git a/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin new file mode 100644 index 0000000000000000000000000000000000000000..5582a008e1adcba5868684e7169fffc95a351171 GIT binary patch literal 16781 zcmeI)u};G<5C-712qCe;1ArHnnh;S{A}lb#z{JuCF*wO3wPfsIJ3v)+=8<>~#2YXm z^)0vvq#_=n`kf`Vth@6!c*!ZZ*Oyt6T->~WoK?eQJh18B|0=K$g>gj?`Q1Xqc}pS}6&Z@!S@}d|bY4DXVkBx! zrs}udk3_Jl4Ko*|u0$>~UZBU0Y)R5~4+Z(3MOW8>qR7W1VaPUdCh}}Do=irf^{yo! zHF3O!D3!0YRYu1-C(p9+*(vAqRFm&xR9g`Qb6udcD$=M5!!Nui zRmRBD*{HN7AC6n)xvVIbORo7}PERUw=%P=lOX8v%*|>w!yAY&Nw&_$u2ep2QohOcP L)VEXc=*hkTRU)zo literal 0 HcmV?d00001 diff --git a/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin new file mode 100644 index 0000000000000000000000000000000000000000..4404137af3286dfc595d7dc47568d7d06cf902af GIT binary patch literal 650 zcmcIiJx{|h5OupCHhw~N6GKdzwiK{fSeUyb1}DCxmW&;2r%+Y>8_fI_Hde&I-(Y6q zk_u5(f(7Bl_wn@Z-m_&JIfbXMCbV;g9J*AMzyf( z1cH@yn7hce0>LvX;BAApKyI6x9Q~%~$}(UCF-gFnt>PTS^fZCmyBdAeNMtK0q*vNX zqvMh$PiTa;y6q1sMcojAl91fQA zWPkGaW3M0X8vwZa_`I_VY5q$u0S@yiTne~=6F7r&xPZ6s>U}M&8QYNB5%uGYyc-BL zS_;HIMYW}lQ&!ssYjmE1Eo)CTj?oO`Cb9AswQ2;Qmx*FWDo)QWPSi-%FGu gwM>SE*cLJ0k!0yrpQ^&EoPXshlkmR&-c#=X0SLfd%>V!Z literal 0 HcmV?d00001 diff --git a/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin new file mode 100644 index 0000000000000000000000000000000000000000..b9037900c2cab7f0d8b869f477a81440816a5df8 GIT binary patch literal 639 zcmcJNy-ve05Xap<0lYzE0Rkx|A59BnLqbSwomdcqV_y3la?7z|-C7e;+^JP0vpgLdf3R`-vQ&@%{3ckhMP~TdSZXujHco*hOb&vwE;n zr{s}5JiRmdz3x(o%tzy(~?#qwIJz$;Q_q zbq#M_Z9G$w<}ASg+;-3iU`=;ZfL#YoSvv4E&qAu9seFy*B&4l1Em*Gr%Py#e5^SF; zndF|cQt5seH_D^vj$5cr=OB_rm=ig0ss&SmPWMT0!H3Qr-K`f(!O} T%S8>*m-YaQZV%T-)*r_w%Cx|g literal 0 HcmV?d00001 diff --git a/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin new file mode 100644 index 0000000000000000000000000000000000000000..5c0007e217711b51d5ee7be20b589f4d0aa0a39c GIT binary patch literal 384 zcmY*V%TB{E5Of6yapM!BJpqbILQ@{TAtBB^aX}m$d(&7lcCej7RrRa5^ACv6>4lA2 z4(w@nG&?hz`Qv@X7`yoXxziI`{zjCH?e0T%O5lVYup@TNPT1!*`y_>YMs@dR@<6a)?g$vGa)=4T4t$;M;&sA$P-T zgJB2ywvCuT)Hzsmi`0Rbl{xgm_ZX5vl3hc<7Q^_ext2-t*4QyjoQk4&mjOFJL@abu zB-vP6h*Ol~@RG|$UG_4Nb)u+B25p?QCap+d2Ad5f{colM%0y1h$S?{i>VhwP_VARc h;A7;<8gwaZ%|{vn#_UY?QgvADx#uJ*_C=2^{sKt2a)JN= literal 0 HcmV?d00001 From 0ff15f6474babd225fd3f365212e938b41fc5c16 Mon Sep 17 00:00:00 2001 From: xuanyili Date: Sat, 5 Sep 2026 18:24:25 +0000 Subject: [PATCH 2/2] fix: validate deletion vectors against Puffin metadata --- src/iceberg/deletes/dv_util.cc | 83 ++++- .../deletes/roaring_position_bitmap.cc | 1 + src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/dv_util_test.cc | 303 ++++++++++++++++++ src/iceberg/test/puffin_dv_interop_test.cc | 13 +- 5 files changed, 380 insertions(+), 21 deletions(-) create mode 100644 src/iceberg/test/dv_util_test.cc diff --git a/src/iceberg/deletes/dv_util.cc b/src/iceberg/deletes/dv_util.cc index 670ccfedb..e9fb6449d 100644 --- a/src/iceberg/deletes/dv_util.cc +++ b/src/iceberg/deletes/dv_util.cc @@ -17,6 +17,7 @@ * under the License. */ +#include #include #include #include @@ -36,13 +37,63 @@ #include "iceberg/metadata_columns.h" #include "iceberg/partition_spec.h" #include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/puffin_reader.h" #include "iceberg/result.h" #include "iceberg/util/content_file_util.h" #include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" #include "iceberg/version.h" namespace iceberg { +namespace { + +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +Status ValidateDVBlobMetadata(const puffin::BlobMetadata& blob, + const DataFile& delete_file) { + ICEBERG_PRECHECK(blob.type == puffin::StandardBlobTypes::kDeletionVectorV1, + "Invalid deletion vector blob type '{}', expected '{}'", blob.type, + puffin::StandardBlobTypes::kDeletionVectorV1); + ICEBERG_PRECHECK(blob.snapshot_id == -1, + "Deletion vector requires snapshot-id -1, got {}", blob.snapshot_id); + ICEBERG_PRECHECK(blob.sequence_number == -1, + "Deletion vector requires sequence-number -1, got {}", + blob.sequence_number); + ICEBERG_PRECHECK(blob.compression_codec.empty(), + "Deletion vector must not be compressed, got '{}'", + blob.compression_codec); + + auto referenced_data_file = + blob.properties.find(std::string(kReferencedDataFileProperty)); + ICEBERG_PRECHECK(referenced_data_file != blob.properties.end() && + !referenced_data_file->second.empty(), + "Deletion vector blob requires non-empty '{}' property", + kReferencedDataFileProperty); + ICEBERG_PRECHECK( + referenced_data_file->second == *delete_file.referenced_data_file, + "Manifest referenced_data_file '{}' does not match Puffin '{}' property '{}'", + *delete_file.referenced_data_file, kReferencedDataFileProperty, + referenced_data_file->second); + + auto cardinality = blob.properties.find(std::string(kCardinalityProperty)); + ICEBERG_PRECHECK(cardinality != blob.properties.end(), + "Deletion vector blob requires '{}' property", kCardinalityProperty); + ICEBERG_ASSIGN_OR_RAISE(auto parsed_cardinality, + StringUtils::ParseNumber(cardinality->second)); + ICEBERG_PRECHECK(parsed_cardinality >= 0, + "Deletion vector cardinality must be non-negative, got {}", + parsed_cardinality); + ICEBERG_PRECHECK( + parsed_cardinality == delete_file.record_count, + "Manifest record_count {} does not match Puffin cardinality property {}", + delete_file.record_count, parsed_cardinality); + return {}; +} + +} // namespace + Result>> DVUtil::MergeAndWriteDVs( std::span groups, std::string_view output_path, const std::shared_ptr& io) { @@ -84,6 +135,10 @@ Result DVUtil::ReadDV(const std::shared_ptr& dele delete_file->content_size_in_bytes.has_value(), "Deletion vector requires content_offset and content_size_in_bytes: {}", delete_file->file_path); + ICEBERG_PRECHECK(delete_file->referenced_data_file.has_value() && + !delete_file->referenced_data_file->empty(), + "Deletion vector requires referenced_data_file: {}", + delete_file->file_path); const int64_t offset = delete_file->content_offset.value(); const int64_t length = delete_file->content_size_in_bytes.value(); @@ -94,14 +149,21 @@ Result DVUtil::ReadDV(const std::shared_ptr& dele "Cannot read deletion vector larger than 2GB: {}", length); ICEBERG_ASSIGN_OR_RAISE(auto input_file, io->NewInputFile(delete_file->file_path)); - ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open()); - - std::vector bytes(static_cast(length)); - ICEBERG_RETURN_UNEXPECTED(stream->ReadFully(offset, bytes)); - ICEBERG_RETURN_UNEXPECTED(stream->Close()); - - std::span blob(reinterpret_cast(bytes.data()), - bytes.size()); + ICEBERG_ASSIGN_OR_RAISE(auto reader, puffin::PuffinReader::Make(std::move(input_file))); + ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->ReadFileMetadata()); + auto blob_metadata = std::ranges::find_if( + metadata.blobs, [offset](const auto& blob) { return blob.offset == offset; }); + ICEBERG_PRECHECK(blob_metadata != metadata.blobs.end(), + "No Puffin blob starts at manifest content_offset {}", offset); + ICEBERG_PRECHECK( + blob_metadata->length == length, + "Puffin blob at offset {} has length {}, manifest content_size_in_bytes is {}", + offset, blob_metadata->length, length); + ICEBERG_RETURN_UNEXPECTED(ValidateDVBlobMetadata(*blob_metadata, *delete_file)); + + ICEBERG_ASSIGN_OR_RAISE(auto blob_data, reader->ReadBlob(*blob_metadata)); + std::span blob(reinterpret_cast(blob_data.second.data()), + blob_data.second.size()); return PositionDeleteIndex::Deserialize(blob, delete_file); } @@ -118,8 +180,9 @@ Result DVUtil::WriteDVBlob(puffin::PuffinWriter& writer, .data = std::move(data), .requested_compression = puffin::PuffinCompressionCodec::kNone, }; - blob.properties.emplace("referenced-data-file", std::string(referenced_data_file)); - blob.properties.emplace("cardinality", std::format("{}", positions.Cardinality())); + blob.properties.emplace(kReferencedDataFileProperty, std::string(referenced_data_file)); + blob.properties.emplace(kCardinalityProperty, + std::format("{}", positions.Cardinality())); return writer.Write(blob); } diff --git a/src/iceberg/deletes/roaring_position_bitmap.cc b/src/iceberg/deletes/roaring_position_bitmap.cc index a2827d4bb..2020160f1 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.cc +++ b/src/iceberg/deletes/roaring_position_bitmap.cc @@ -271,6 +271,7 @@ Result RoaringPositionBitmap::Deserialize(std::string_vie --remaining_count; } + ICEBERG_PRECHECK(remaining == 0, "Trailing data after bitmaps: {} bytes", remaining); return RoaringPositionBitmap(std::move(impl)); } diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index f5efed591..3238e9c2d 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -158,6 +158,7 @@ add_iceberg_test(util_test add_iceberg_test(puffin_test USE_DATA SOURCES + dv_util_test.cc puffin_dv_interop_test.cc puffin_format_test.cc puffin_json_test.cc diff --git a/src/iceberg/test/dv_util_test.cc b/src/iceberg/test/dv_util_test.cc new file mode 100644 index 000000000..d84eb0784 --- /dev/null +++ b/src/iceberg/test/dv_util_test.cc @@ -0,0 +1,303 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/deletes/dv_util_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/file_format.h" +#include "iceberg/manifest/manifest_entry.h" +#include "iceberg/metadata_columns.h" +#include "iceberg/puffin/file_metadata.h" +#include "iceberg/puffin/json_serde_internal.h" +#include "iceberg/puffin/puffin_format.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/endian.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +namespace { + +constexpr std::string_view kReferencedDataFile = "data.parquet"; +constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file"; +constexpr std::string_view kCardinalityProperty = "cardinality"; + +struct MetadataTestCase { + std::string name; + std::function&)> mutate_data; + std::function mutate_blob; + std::function mutate_manifest; + std::string expected_error; +}; + +void AddTrailingBitmapData(std::vector& data) { + constexpr size_t kLengthPrefixBytes = 4; + constexpr size_t kCrcBytes = 4; + + data.insert(data.end() - kCrcBytes, 0); + const auto length = static_cast(data.size() - kLengthPrefixBytes - kCrcBytes); + WriteBigEndian(length, data.data()); + + uLong crc = crc32(0L, Z_NULL, 0); + crc = crc32(crc, reinterpret_cast(data.data() + kLengthPrefixBytes), + static_cast(length)); + WriteBigEndian(static_cast(crc), data.data() + data.size() - kCrcBytes); +} + +Result> WriteDVFixture(const std::shared_ptr& io, + const MetadataTestCase& test_case) { + PositionDeleteIndex positions; + positions.Delete(1); + positions.Delete(3); + positions.Delete(5); + ICEBERG_ASSIGN_OR_RAISE(auto data, positions.Serialize()); + if (test_case.mutate_data) { + test_case.mutate_data(data); + } + + puffin::Blob blob{ + .type = std::string(puffin::StandardBlobTypes::kDeletionVectorV1), + .input_fields = {MetadataColumns::kFilePositionColumnId}, + .snapshot_id = -1, + .sequence_number = -1, + .data = std::move(data), + .requested_compression = puffin::PuffinCompressionCodec::kNone, + .properties = + { + {std::string(kReferencedDataFileProperty), + std::string(kReferencedDataFile)}, + {std::string(kCardinalityProperty), "3"}, + }, + }; + if (test_case.mutate_blob) { + test_case.mutate_blob(blob); + } + + const std::string path = "memory://" + test_case.name + ".puffin"; + const auto codec = + blob.requested_compression.value_or(puffin::PuffinCompressionCodec::kNone); + puffin::BlobMetadata blob_metadata{ + .type = blob.type, + .input_fields = blob.input_fields, + .snapshot_id = blob.snapshot_id, + .sequence_number = blob.sequence_number, + .offset = puffin::PuffinFormat::kMagicLength, + .length = static_cast(blob.data.size()), + .compression_codec = std::string(puffin::CodecName(codec)), + .properties = blob.properties, + }; + const std::string footer = + puffin::ToJsonString(puffin::FileMetadata{.blobs = {blob_metadata}}); + + std::vector file_data; + auto append = [&file_data](const void* data, size_t size) { + auto bytes = std::span(reinterpret_cast(data), size); + file_data.insert(file_data.end(), bytes.begin(), bytes.end()); + }; + append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size()); + append(blob.data.data(), blob.data.size()); + append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size()); + append(footer.data(), footer.size()); + + std::array footer_struct{}; + WriteLittleEndian(static_cast(footer.size()), footer_struct.data()); + std::memcpy(footer_struct.data() + puffin::PuffinFormat::kFooterStructMagicOffset, + puffin::PuffinFormat::kMagicV1.data(), + puffin::PuffinFormat::kMagicV1.size()); + append(footer_struct.data(), footer_struct.size()); + io->AddFile(path, file_data); + + auto delete_file = std::make_shared(DataFile{ + .content = DataFile::Content::kPositionDeletes, + .file_path = path, + .file_format = FileFormatType::kPuffin, + .record_count = 3, + .file_size_in_bytes = static_cast(file_data.size()), + .referenced_data_file = std::string(kReferencedDataFile), + .content_offset = blob_metadata.offset, + .content_size_in_bytes = blob_metadata.length, + }); + if (test_case.mutate_manifest) { + test_case.mutate_manifest(*delete_file); + } + return delete_file; +} + +} // namespace + +TEST(DVUtilTest, RejectsMalformedPuffinMetadata) { + const std::vector cases = { + { + .name = "wrong-offset", + .mutate_manifest = [](DataFile& file) { ++*file.content_offset; }, + .expected_error = "No Puffin blob starts", + }, + { + .name = "wrong-length", + .mutate_manifest = [](DataFile& file) { --*file.content_size_in_bytes; }, + .expected_error = "manifest content_size_in_bytes", + }, + { + .name = "missing-manifest-reference", + .mutate_manifest = [](DataFile& file) { file.referenced_data_file.reset(); }, + .expected_error = "requires referenced_data_file", + }, + { + .name = "empty-manifest-reference", + .mutate_manifest = [](DataFile& file) { file.referenced_data_file = ""; }, + .expected_error = "requires referenced_data_file", + }, + { + .name = "wrong-type", + .mutate_blob = [](puffin::Blob& blob) { blob.type = "test-blob"; }, + .expected_error = "Invalid deletion vector blob type", + }, + { + .name = "wrong-snapshot", + .mutate_blob = [](puffin::Blob& blob) { blob.snapshot_id = 0; }, + .expected_error = "snapshot-id -1", + }, + { + .name = "wrong-sequence", + .mutate_blob = [](puffin::Blob& blob) { blob.sequence_number = 0; }, + .expected_error = "sequence-number -1", + }, + { + .name = "compressed", + .mutate_blob = + [](puffin::Blob& blob) { + blob.requested_compression = puffin::PuffinCompressionCodec::kZstd; + }, + .expected_error = "must not be compressed", + }, + { + .name = "missing-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties.erase(std::string(kReferencedDataFileProperty)); + }, + .expected_error = "requires non-empty 'referenced-data-file'", + }, + { + .name = "empty-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kReferencedDataFileProperty)] = ""; + }, + .expected_error = "requires non-empty 'referenced-data-file'", + }, + { + .name = "mismatched-reference", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kReferencedDataFileProperty)] = + "other.parquet"; + }, + .expected_error = "does not match Puffin", + }, + { + .name = "missing-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties.erase(std::string(kCardinalityProperty)); + }, + .expected_error = "requires 'cardinality'", + }, + { + .name = "invalid-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "three"; + }, + .expected_error = "Failed to parse", + }, + { + .name = "negative-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "-1"; + }, + .expected_error = "must be non-negative", + }, + { + .name = "mismatched-cardinality", + .mutate_blob = + [](puffin::Blob& blob) { + blob.properties[std::string(kCardinalityProperty)] = "4"; + }, + .expected_error = "Manifest record_count 3", + }, + { + .name = "trailing-bitmap-data", + .mutate_data = AddTrailingBitmapData, + .expected_error = "Trailing data after bitmaps", + }, + }; + + for (const auto& test_case : cases) { + SCOPED_TRACE(test_case.name); + auto io = std::make_shared(); + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + EXPECT_THAT(DVUtil::ReadDV(delete_file, io), + HasErrorMessage(test_case.expected_error)); + } +} + +TEST(DVUtilTest, ReadsValidPuffinMetadata) { + auto io = std::make_shared(); + MetadataTestCase test_case{.name = "valid"}; + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); + + EXPECT_EQ(positions.Cardinality(), 3); + EXPECT_TRUE(positions.IsDeleted(1)); + EXPECT_TRUE(positions.IsDeleted(3)); + EXPECT_TRUE(positions.IsDeleted(5)); +} + +TEST(DVUtilTest, ReadsUsingActualFileSize) { + auto io = std::make_shared(); + MetadataTestCase test_case{ + .name = "stale-manifest-file-size", + .mutate_manifest = [](DataFile& file) { file.file_size_in_bytes = 0; }, + }; + ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case)); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); + + EXPECT_EQ(positions.Cardinality(), 3); +} + +} // namespace iceberg diff --git a/src/iceberg/test/puffin_dv_interop_test.cc b/src/iceberg/test/puffin_dv_interop_test.cc index b81dcdd51..b322fe0ae 100644 --- a/src/iceberg/test/puffin_dv_interop_test.cc +++ b/src/iceberg/test/puffin_dv_interop_test.cc @@ -17,17 +17,16 @@ * under the License. */ -#include #include #include #include -#include #include #include #include #include +#include "iceberg/deletes/dv_util_internal.h" #include "iceberg/deletes/position_delete_index.h" #include "iceberg/file_format.h" #include "iceberg/manifest/manifest_entry.h" @@ -100,15 +99,7 @@ void AssertPositions(const ExpectedBlob& expected, const std::shared_ptr& io) { ASSERT_TRUE(delete_file->content_offset.has_value()); ASSERT_TRUE(delete_file->content_size_in_bytes.has_value()); - ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(delete_file->file_path)); - ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open()); - std::vector data(static_cast(*delete_file->content_size_in_bytes)); - ASSERT_THAT(stream->ReadFully(*delete_file->content_offset, data), IsOk()); - - std::span blob(reinterpret_cast(data.data()), - data.size()); - ICEBERG_UNWRAP_OR_FAIL(auto positions, - PositionDeleteIndex::Deserialize(blob, delete_file)); + ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io)); EXPECT_EQ(positions.Cardinality(), expected.cardinality); int64_t expected_cardinality = static_cast(expected.positions.size()); for (const auto& range : expected.ranges) {