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 5c494b640dce8552a44e198787c054c9c5faa6a4 Mon Sep 17 00:00:00 2001 From: xuanyili Date: Sat, 5 Sep 2026 21:15:39 +0000 Subject: [PATCH 2/2] fix: support all non-negative int64 deletion vector positions --- src/iceberg/deletes/dv_writer.cc | 27 +++- src/iceberg/deletes/dv_writer.h | 6 + src/iceberg/deletes/dv_writer_internal.h | 40 +++++ src/iceberg/deletes/position_delete_index.cc | 21 ++- src/iceberg/deletes/position_delete_index.h | 8 + .../deletes/position_delete_range_consumer.cc | 10 +- .../deletes/roaring_position_bitmap.cc | 70 +++----- src/iceberg/deletes/roaring_position_bitmap.h | 23 +-- src/iceberg/test/CMakeLists.txt | 1 + src/iceberg/test/dv_writer_preflight_test.cc | 85 ++++++++++ src/iceberg/test/dv_writer_test.cc | 10 +- .../test/position_delete_index_test.cc | 21 +++ .../position_delete_range_consumer_test.cc | 15 +- .../test/roaring_position_bitmap_test.cc | 151 +++++++++++------- 14 files changed, 342 insertions(+), 146 deletions(-) create mode 100644 src/iceberg/deletes/dv_writer_internal.h create mode 100644 src/iceberg/test/dv_writer_preflight_test.cc diff --git a/src/iceberg/deletes/dv_writer.cc b/src/iceberg/deletes/dv_writer.cc index a51407fd3..5525ae927 100644 --- a/src/iceberg/deletes/dv_writer.cc +++ b/src/iceberg/deletes/dv_writer.cc @@ -19,7 +19,9 @@ #include "iceberg/deletes/dv_writer.h" +#include #include +#include #include #include #include @@ -30,8 +32,8 @@ #include #include "iceberg/deletes/dv_util_internal.h" +#include "iceberg/deletes/dv_writer_internal.h" #include "iceberg/deletes/position_delete_index.h" -#include "iceberg/deletes/roaring_position_bitmap.h" #include "iceberg/file_format.h" #include "iceberg/file_io.h" // IWYU pragma: keep #include "iceberg/manifest/manifest_entry.h" @@ -49,7 +51,8 @@ namespace iceberg { class DVWriter::Impl { public: - explicit Impl(DVWriterOptions options) : options_(std::move(options)) {} + Impl(DVWriterOptions options, size_t max_serialized_length) + : options_(std::move(options)), max_serialized_length_(max_serialized_length) {} // Accumulated positions and metadata for a single referenced data file. struct Deletes { @@ -79,9 +82,7 @@ class DVWriter::Impl { ICEBERG_PRECHECK(!referenced_data_file.empty(), "Deletion vector requires a non-empty referenced data file"); ICEBERG_PRECHECK(spec != nullptr, "Deletion vector requires a partition spec"); - ICEBERG_PRECHECK(pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition, - "Deletion vector position out of range [0, {}]: {}", - RoaringPositionBitmap::kMaxPosition, pos); + ICEBERG_PRECHECK(pos >= 0, "Deletion vector position must be non-negative: {}", pos); DeletesFor(referenced_data_file, spec, partition).positions.Delete(pos); return {}; } @@ -112,6 +113,11 @@ class DVWriter::Impl { ICEBERG_RETURN_UNEXPECTED(LoadPreviousDeletes(path, deletes)); } + for (auto& [_, deletes] : deletes_by_path_) { + ICEBERG_RETURN_UNEXPECTED( + deletes.positions.ValidateSerializedSize(max_serialized_length_)); + } + ICEBERG_ASSIGN_OR_RAISE(auto output_file, options_.io->NewOutputFile(options_.path)); const std::string output_path(options_.path); ICEBERG_ASSIGN_OR_RAISE( @@ -198,6 +204,7 @@ class DVWriter::Impl { std::map blobs_by_path_; DeleteWriteResult result_; bool closed_ = false; + size_t max_serialized_length_; }; DVWriter::DVWriter(std::unique_ptr impl) : impl_(std::move(impl)) {} @@ -205,12 +212,18 @@ DVWriter::DVWriter(std::unique_ptr impl) : impl_(std::move(impl)) {} DVWriter::~DVWriter() = default; Result> DVWriter::Make(DVWriterOptions options) { + return internal::DVWriterFactory::Make( + std::move(options), static_cast(std::numeric_limits::max())); +} + +Result> internal::DVWriterFactory::Make( + DVWriterOptions options, size_t max_serialized_length) { ICEBERG_PRECHECK(!options.path.empty(), "DVWriter requires an output path"); ICEBERG_PRECHECK(options.io != nullptr, "DVWriter requires a FileIO"); ICEBERG_PRECHECK(options.load_previous_deletes != nullptr, "DVWriter requires a load_previous_deletes callback"); - return std::unique_ptr( - new DVWriter(std::make_unique(std::move(options)))); + return std::unique_ptr(new DVWriter( + std::make_unique(std::move(options), max_serialized_length))); } Status DVWriter::Delete(std::string_view referenced_data_file, int64_t pos, diff --git a/src/iceberg/deletes/dv_writer.h b/src/iceberg/deletes/dv_writer.h index b2725996b..c08230d68 100644 --- a/src/iceberg/deletes/dv_writer.h +++ b/src/iceberg/deletes/dv_writer.h @@ -38,6 +38,10 @@ namespace iceberg { +namespace internal { +class DVWriterFactory; +} + /// \brief File metadata for deletion vectors produced by DVWriter. struct ICEBERG_EXPORT DeleteWriteResult { /// Deletion vector files produced by the writer. @@ -87,6 +91,8 @@ class ICEBERG_EXPORT DVWriter { std::unique_ptr impl_; explicit DVWriter(std::unique_ptr impl); + + friend class internal::DVWriterFactory; }; } // namespace iceberg diff --git a/src/iceberg/deletes/dv_writer_internal.h b/src/iceberg/deletes/dv_writer_internal.h new file mode 100644 index 000000000..b0bd87f70 --- /dev/null +++ b/src/iceberg/deletes/dv_writer_internal.h @@ -0,0 +1,40 @@ +/* + * 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. + */ + +#pragma once + +/// \file iceberg/deletes/dv_writer_internal.h +/// Internal deletion vector writer helpers. + +#include +#include + +#include "iceberg/deletes/dv_writer.h" +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" + +namespace iceberg::internal { + +class ICEBERG_EXPORT DVWriterFactory { + public: + static Result> Make(DVWriterOptions options, + size_t max_serialized_length); +}; + +} // namespace iceberg::internal diff --git a/src/iceberg/deletes/position_delete_index.cc b/src/iceberg/deletes/position_delete_index.cc index 53e33e635..8c5099c27 100644 --- a/src/iceberg/deletes/position_delete_index.cc +++ b/src/iceberg/deletes/position_delete_index.cc @@ -49,6 +49,7 @@ constexpr std::array kMagic = {0xD1, 0xD3, 0x39, 0x64}; constexpr int32_t kLengthPrefixBytes = 4; constexpr int32_t kMagicBytes = 4; constexpr int32_t kCrcBytes = 4; +constexpr size_t kMaxSerializedLength = std::numeric_limits::max(); uint32_t ComputeCrc32(std::span bytes) { uLong crc = crc32(0L, Z_NULL, 0); @@ -142,16 +143,24 @@ void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) { other.delete_files_.end()); } -Result> PositionDeleteIndex::Serialize() { +Status PositionDeleteIndex::ValidateSerializedSize(size_t max_length) { bitmap_.Optimize(); // run-length encode before serializing - std::vector blob(kLengthPrefixBytes); - blob.insert(blob.end(), kMagic.begin(), kMagic.end()); - ICEBERG_ASSIGN_OR_RAISE(const auto vector_size, bitmap_.SerializeTo(blob)); - + const size_t vector_size = bitmap_.SerializedSizeInBytes(); const size_t magic_and_vector_size = kMagicBytes + vector_size; - ICEBERG_PRECHECK(magic_and_vector_size <= std::numeric_limits::max(), + ICEBERG_PRECHECK(magic_and_vector_size <= max_length, "Deletion vector is too large to serialize: {} bytes", magic_and_vector_size); + return {}; +} + +Result> PositionDeleteIndex::Serialize() { + ICEBERG_RETURN_UNEXPECTED(ValidateSerializedSize(kMaxSerializedLength)); + const size_t vector_size = bitmap_.SerializedSizeInBytes(); + const size_t magic_and_vector_size = kMagicBytes + vector_size; + + std::vector blob(kLengthPrefixBytes); + blob.insert(blob.end(), kMagic.begin(), kMagic.end()); + ICEBERG_RETURN_UNEXPECTED(bitmap_.SerializeTo(blob)); WriteBigEndian(static_cast(magic_and_vector_size), blob.data()); const auto crc_offset = blob.size(); diff --git a/src/iceberg/deletes/position_delete_index.h b/src/iceberg/deletes/position_delete_index.h index 6f301210e..cfb01b4c4 100644 --- a/src/iceberg/deletes/position_delete_index.h +++ b/src/iceberg/deletes/position_delete_index.h @@ -22,6 +22,7 @@ /// \file iceberg/deletes/position_delete_index.h /// Index of deleted row positions for a data file. +#include #include #include #include @@ -34,6 +35,8 @@ namespace iceberg { +class DVWriter; + /// \brief Tracks deleted row positions using a bitmap. /// /// This class provides a domain-specific API for position deletes @@ -53,6 +56,8 @@ class ICEBERG_EXPORT PositionDeleteIndex { /// \brief Mark a range of positions as deleted [pos_start, pos_end). /// \param pos_start Start position (inclusive) /// \param pos_end End position (exclusive) + /// \note Because pos_end is an int64_t exclusive endpoint, this method cannot + /// include INT64_MAX. Call Delete(INT64_MAX) separately. void Delete(int64_t pos_start, int64_t pos_end); /// \brief Check if a position is deleted. @@ -97,6 +102,8 @@ class ICEBERG_EXPORT PositionDeleteIndex { private: explicit PositionDeleteIndex(RoaringPositionBitmap bitmap); + Status ValidateSerializedSize(size_t max_length); + // Bulk-add positions sharing high-32-bit `key`. Private hook for // `ForEachPositionDelete`'s bulk path; keeps `Delete` the sole public // mutation surface. @@ -105,6 +112,7 @@ class ICEBERG_EXPORT PositionDeleteIndex { friend void ICEBERG_EXPORT ForEachPositionDelete(std::span positions, PositionDeleteIndex& target, std::vector& scratch); + friend class DVWriter; RoaringPositionBitmap bitmap_; std::vector> delete_files_; diff --git a/src/iceberg/deletes/position_delete_range_consumer.cc b/src/iceberg/deletes/position_delete_range_consumer.cc index f7cf258c3..6362a4371 100644 --- a/src/iceberg/deletes/position_delete_range_consumer.cc +++ b/src/iceberg/deletes/position_delete_range_consumer.cc @@ -31,9 +31,7 @@ namespace iceberg { namespace { -bool IsValidPosition(int64_t pos) { - return pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition; -} +bool IsValidPosition(int64_t pos) { return pos >= 0; } // Unsigned subtraction so negative or wrap-around input can't // false-positive via signed overflow. @@ -45,11 +43,13 @@ bool IsAdjacent(int64_t prev, int64_t next) { // bulk path groups by this key before flushing via `BulkAddForKey`. int32_t HighKeyFromPosition(int64_t pos) { return static_cast(pos >> 32); } -// Emit `[range_start, last_position]`, collapsing singletons. Callers -// pre-filter via `IsValidPosition`, so `last_position + 1` cannot overflow. +// Emit `[range_start, last_position]`, collapsing singletons. void EmitRange(PositionDeleteIndex& target, int64_t range_start, int64_t last_position) { if (range_start == last_position) { target.Delete(range_start); + } else if (last_position == RoaringPositionBitmap::kMaxPosition) { + target.Delete(range_start, last_position); + target.Delete(last_position); } else { target.Delete(range_start, last_position + 1); } diff --git a/src/iceberg/deletes/roaring_position_bitmap.cc b/src/iceberg/deletes/roaring_position_bitmap.cc index a2827d4bb..bbe0b027a 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.cc +++ b/src/iceberg/deletes/roaring_position_bitmap.cc @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -51,34 +52,20 @@ int64_t ToPosition(int32_t key, uint32_t pos32) { return (int64_t{key} << 32) | int64_t{pos32}; } -Status ValidatePosition(int64_t pos) { - if (pos < 0 || pos > RoaringPositionBitmap::kMaxPosition) { - return InvalidArgument("Bitmap supports positions that are >= 0 and <= {}: {}", - RoaringPositionBitmap::kMaxPosition, pos); - } - return {}; -} - -void WriteBitmaps(const std::vector& bitmaps, uint8_t* buf) { +void WriteBitmaps(const std::map& bitmaps, uint8_t* buf) { WriteLittleEndian(static_cast(bitmaps.size()), buf); buf += kBitmapCountSizeBytes; - for (int32_t key = 0; std::cmp_less(key, bitmaps.size()); ++key) { + for (const auto& [key, bitmap] : bitmaps) { WriteLittleEndian(key, buf); buf += kBitmapKeySizeBytes; - buf += bitmaps[key].write(reinterpret_cast(buf), /*portable=*/true); + buf += bitmap.write(reinterpret_cast(buf), /*portable=*/true); } } } // namespace struct RoaringPositionBitmap::Impl { - std::vector bitmaps; - - void AllocateBitmapsIfNeeded(int32_t required_length) { - if (std::cmp_less(bitmaps.size(), required_length)) { - bitmaps.resize(static_cast(required_length)); - } - } + std::map bitmaps; }; RoaringPositionBitmap::RoaringPositionBitmap() : impl_(std::make_unique()) {} @@ -108,24 +95,24 @@ RoaringPositionBitmap::RoaringPositionBitmap(std::unique_ptr impl) : impl_(std::move(impl)) {} void RoaringPositionBitmap::Add(int64_t pos) { - if (pos < 0 || pos > kMaxPosition) { + if (pos < 0) { return; // Silently ignore invalid positions } int32_t key = Key(pos); uint32_t pos32 = Pos32Bits(pos); - impl_->AllocateBitmapsIfNeeded(key + 1); impl_->bitmaps[key].add(pos32); } void RoaringPositionBitmap::AddManyForKey(int32_t key, std::span positions) { - impl_->AllocateBitmapsIfNeeded(key + 1); + if (key < 0 || positions.empty()) { + return; + } impl_->bitmaps[key].addMany(positions.size(), positions.data()); } void RoaringPositionBitmap::AddRange(int64_t pos_start, int64_t pos_end) { pos_start = std::max(pos_start, int64_t{0}); - pos_end = std::min(pos_end, kMaxPosition + 1); if (pos_start >= pos_end) { return; } @@ -133,61 +120,60 @@ void RoaringPositionBitmap::AddRange(int64_t pos_start, int64_t pos_end) { int64_t pos_last = pos_end - 1; int32_t start_key = Key(pos_start); int32_t end_key = Key(pos_last); - impl_->AllocateBitmapsIfNeeded(end_key + 1); - for (int32_t key = start_key; key <= end_key; ++key) { + for (int64_t key = start_key; key <= end_key; ++key) { uint64_t low_start = (key == start_key) ? Pos32Bits(pos_start) : uint64_t{0}; uint64_t low_end = (key == end_key) ? static_cast(Pos32Bits(pos_last)) + 1 : (uint64_t{1} << 32); - impl_->bitmaps[key].addRange(low_start, low_end); + impl_->bitmaps[static_cast(key)].addRange(low_start, low_end); } } bool RoaringPositionBitmap::Contains(int64_t pos) const { - if (pos < 0 || pos > kMaxPosition) { + if (pos < 0) { return false; // Invalid positions are not contained } int32_t key = Key(pos); uint32_t pos32 = Pos32Bits(pos); - return std::cmp_less(key, impl_->bitmaps.size()) && impl_->bitmaps[key].contains(pos32); + auto it = impl_->bitmaps.find(key); + return it != impl_->bitmaps.end() && it->second.contains(pos32); } bool RoaringPositionBitmap::IsEmpty() const { return Cardinality() == 0; } size_t RoaringPositionBitmap::Cardinality() const { size_t total = 0; - for (const auto& bitmap : impl_->bitmaps) { + for (const auto& [_, bitmap] : impl_->bitmaps) { total += bitmap.cardinality(); } return total; } void RoaringPositionBitmap::Or(const RoaringPositionBitmap& other) { - impl_->AllocateBitmapsIfNeeded(static_cast(other.impl_->bitmaps.size())); - for (size_t key = 0; key < other.impl_->bitmaps.size(); ++key) { - impl_->bitmaps[key] |= other.impl_->bitmaps[key]; + for (const auto& [key, bitmap] : other.impl_->bitmaps) { + impl_->bitmaps[key] |= bitmap; } } bool RoaringPositionBitmap::Optimize() { bool changed = false; - for (auto& bitmap : impl_->bitmaps) { + for (auto& [_, bitmap] : impl_->bitmaps) { changed |= bitmap.runOptimize(); } return changed; } void RoaringPositionBitmap::ForEach(const std::function& fn) const { - for (size_t key = 0; key < impl_->bitmaps.size(); ++key) { - for (uint32_t pos32 : impl_->bitmaps[key]) { - fn(ToPosition(static_cast(key), pos32)); + for (const auto& [key, bitmap] : impl_->bitmaps) { + for (uint32_t pos32 : bitmap) { + fn(ToPosition(key, pos32)); } } } size_t RoaringPositionBitmap::SerializedSizeInBytes() const { size_t size = kBitmapCountSizeBytes; - for (const auto& bitmap : impl_->bitmaps) { + for (const auto& [_, bitmap] : impl_->bitmaps) { size += kBitmapKeySizeBytes + bitmap.getSizeInBytes(/*portable=*/true); } return size; @@ -238,18 +224,10 @@ Result RoaringPositionBitmap::Deserialize(std::string_vie remaining -= kBitmapKeySizeBytes; ICEBERG_PRECHECK(key >= 0, "Invalid unsigned key: {}", key); - ICEBERG_PRECHECK(key < std::numeric_limits::max(), "Key is too large: {}", - key); ICEBERG_PRECHECK(key > last_key, "Keys must be sorted in ascending order, got key {} after {}", key, last_key); - // Fill gaps with empty bitmaps - while (last_key < key - 1) { - impl->bitmaps.emplace_back(); - ++last_key; - } - // Read bitmap using portable safe deserialization. // CRoaring's readSafe may throw on corrupted data. roaring::Roaring bitmap; @@ -266,7 +244,9 @@ Result RoaringPositionBitmap::Deserialize(std::string_vie buf += bitmap_size; remaining -= bitmap_size; - impl->bitmaps.emplace_back(std::move(bitmap)); + if (!bitmap.isEmpty()) { + impl->bitmaps.emplace(key, std::move(bitmap)); + } last_key = key; --remaining_count; } diff --git a/src/iceberg/deletes/roaring_position_bitmap.h b/src/iceberg/deletes/roaring_position_bitmap.h index 119387661..e3aee0b35 100644 --- a/src/iceberg/deletes/roaring_position_bitmap.h +++ b/src/iceberg/deletes/roaring_position_bitmap.h @@ -20,10 +20,11 @@ #pragma once /// \file iceberg/deletes/roaring_position_bitmap.h -/// A 64-bit position bitmap using an array of 32-bit Roaring bitmaps. +/// A 64-bit position bitmap using sparse 32-bit Roaring bitmaps. #include #include +#include #include #include #include @@ -37,7 +38,7 @@ namespace iceberg { class PositionDeleteIndex; -/// \brief A bitmap that supports positive 64-bit positions, optimized +/// \brief A bitmap that supports non-negative 64-bit positions, optimized /// for cases where most positions fit in 32 bits. /// /// Incoming 64-bit positions are divided into a 32-bit "key" using the @@ -51,8 +52,8 @@ class PositionDeleteIndex; /// for `deletion-vector-v1` persistence. class ICEBERG_EXPORT RoaringPositionBitmap { public: - /// \brief Maximum supported position (aligned with the Java implementation). - static constexpr int64_t kMaxPosition = 0x7FFFFFFE80000000LL; + /// \brief Maximum supported position. + static constexpr int64_t kMaxPosition = std::numeric_limits::max(); RoaringPositionBitmap(); ~RoaringPositionBitmap(); @@ -64,21 +65,21 @@ class ICEBERG_EXPORT RoaringPositionBitmap { RoaringPositionBitmap& operator=(const RoaringPositionBitmap& other); /// \brief Sets a position in the bitmap. - /// \param pos the position (must be >= 0 and <= kMaxPosition) - /// \note Invalid positions are silently ignored + /// \param pos the position (must be non-negative) + /// \note Negative positions are silently ignored. void Add(int64_t pos); /// \brief Sets a range of positions [pos_start, pos_end). /// \param pos_start the start of the range (inclusive), clamped to 0 - /// \param pos_end the end of the range (exclusive), clamped to kMaxPosition + 1 - /// \note If pos_start > pos_end, the call is silently ignored. - /// If pos_start == pos_end, this method does nothing. - /// Positions outside [0, kMaxPosition] are silently ignored. + /// \param pos_end the end of the range (exclusive) + /// \note Empty and reversed ranges are silently ignored. + /// \note Because pos_end is an int64_t exclusive endpoint, this method cannot + /// include kMaxPosition. Call Add(kMaxPosition) separately. void AddRange(int64_t pos_start, int64_t pos_end); /// \brief Checks if a position is set in the bitmap. /// \param pos the position to check - /// \return true if the position is set, false otherwise (including invalid positions) + /// \return true if the position is set, false otherwise (including negative positions) bool Contains(int64_t pos) const; /// \brief Returns true if the bitmap has no positions set. diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index f5efed591..2f5ff52a6 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -132,6 +132,7 @@ add_iceberg_test(util_test content_file_util_test.cc data_file_set_test.cc decimal_test.cc + dv_writer_preflight_test.cc endian_test.cc file_io_test.cc formatter_test.cc diff --git a/src/iceberg/test/dv_writer_preflight_test.cc b/src/iceberg/test/dv_writer_preflight_test.cc new file mode 100644 index 000000000..1025710d4 --- /dev/null +++ b/src/iceberg/test/dv_writer_preflight_test.cc @@ -0,0 +1,85 @@ +/* + * 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 "iceberg/deletes/dv_writer.h" + +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/deletes/dv_writer_internal.h" +#include "iceberg/deletes/position_delete_index.h" +#include "iceberg/partition_spec.h" +#include "iceberg/result.h" +#include "iceberg/row/partition_values.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/mock_io.h" +#include "iceberg/util/endian.h" + +namespace iceberg { + +namespace { + +Result> NoPreviousDeletes(std::string_view) { + return std::nullopt; +} + +} // namespace + +TEST(DVWriterPreflightTest, RejectsLaterOversizedVectorBeforeCreatingOutput) { + auto io = std::make_shared(); + auto spec = PartitionSpec::Unpartitioned(); + const std::string output_path = "memory://oversized.puffin"; + + PositionDeleteIndex small; + small.Delete(0); + ICEBERG_UNWRAP_OR_FAIL(auto small_blob, small.Serialize()); + const auto small_length = ReadBigEndian(small_blob.data()); + ASSERT_GT(small_length, 0); + const size_t max_serialized_length = static_cast(small_length); + + PositionDeleteIndex large; + for (int64_t pos = 0; pos < 100; ++pos) { + large.Delete(pos * 2); + } + ICEBERG_UNWRAP_OR_FAIL(auto large_blob, large.Serialize()); + const auto large_length = ReadBigEndian(large_blob.data()); + ASSERT_GT(large_length, 0); + ASSERT_GT(static_cast(large_length), max_serialized_length); + + ICEBERG_UNWRAP_OR_FAIL( + auto writer, + internal::DVWriterFactory::Make( + DVWriterOptions{.path = output_path, + .io = io, + .load_previous_deletes = NoPreviousDeletes}, + max_serialized_length)); + ASSERT_THAT(writer->Delete("a.parquet", small, spec, PartitionValues{}), IsOk()); + ASSERT_THAT(writer->Delete("b.parquet", large, spec, PartitionValues{}), IsOk()); + + EXPECT_THAT(writer->Close(), IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(io->NewInputFile(output_path), IsError(ErrorKind::kNotFound)); +} + +} // namespace iceberg diff --git a/src/iceberg/test/dv_writer_test.cc b/src/iceberg/test/dv_writer_test.cc index 8eb4b2444..ac23c2973 100644 --- a/src/iceberg/test/dv_writer_test.cc +++ b/src/iceberg/test/dv_writer_test.cc @@ -332,18 +332,16 @@ TEST(DVWriterTest, DeleteRejectsEmptyReferencedFile) { IsError(ErrorKind::kInvalidArgument)); } -TEST(DVWriterTest, DeleteRejectsOutOfRangePosition) { +TEST(DVWriterTest, DeleteRejectsNegativeAndAcceptsMaximumPosition) { auto io = std::make_shared(); auto spec = UnpartitionedSpec(); ICEBERG_UNWRAP_OR_FAIL( auto writer, DVWriter::Make(MakeDVWriterOptions(io, "memory://invalid.puffin"))); - // Negative and out-of-range positions are rejected rather than silently - // dropped by the underlying bitmap. EXPECT_THAT(writer->Delete("data.parquet", -1, spec, PartitionValues{}), IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(writer->Delete("data.parquet", RoaringPositionBitmap::kMaxPosition + 1, - spec, PartitionValues{}), - IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(writer->Delete("data.parquet", RoaringPositionBitmap::kMaxPosition, spec, + PartitionValues{}), + IsOk()); } // Close propagates a load_previous_deletes failure and returns no metadata. diff --git a/src/iceberg/test/position_delete_index_test.cc b/src/iceberg/test/position_delete_index_test.cc index 1075ee6fa..e3dc29307 100644 --- a/src/iceberg/test/position_delete_index_test.cc +++ b/src/iceberg/test/position_delete_index_test.cc @@ -20,6 +20,7 @@ #include "iceberg/deletes/position_delete_index.h" #include +#include #include #include @@ -185,6 +186,26 @@ TEST(PositionDeleteIndexTest, TestLargePositions) { ASSERT_FALSE(index.IsDeleted(large_pos + 1)); } +TEST(PositionDeleteIndexTest, TestSparseDistantPositionsRoundTrip) { + PositionDeleteIndex index; + const std::vector positions = { + 1, + (int64_t{1} << 32) + 2, + std::numeric_limits::max(), + }; + for (auto pos : positions) { + index.Delete(pos); + } + + ICEBERG_UNWRAP_OR_FAIL(auto blob, index.Serialize()); + EXPECT_LT(blob.size(), 1024); + ICEBERG_UNWRAP_OR_FAIL(auto restored, PositionDeleteIndex::Deserialize( + blob, DeleteFileFor(blob, positions.size()))); + for (auto pos : positions) { + EXPECT_TRUE(restored.IsDeleted(pos)); + } +} + TEST(PositionDeleteIndexTest, TestOverlappingRanges) { PositionDeleteIndex index; diff --git a/src/iceberg/test/position_delete_range_consumer_test.cc b/src/iceberg/test/position_delete_range_consumer_test.cc index 5a58fa5a2..59dccf22d 100644 --- a/src/iceberg/test/position_delete_range_consumer_test.cc +++ b/src/iceberg/test/position_delete_range_consumer_test.cc @@ -28,7 +28,6 @@ #include #include "iceberg/deletes/position_delete_index.h" -#include "iceberg/deletes/roaring_position_bitmap.h" namespace iceberg { @@ -38,7 +37,7 @@ namespace { std::set ExpectedValidSet(const std::vector& positions) { std::set expected; for (int64_t pos : positions) { - if (pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition) { + if (pos >= 0) { expected.insert(pos); } } @@ -95,12 +94,9 @@ TEST(PositionDeleteRangeConsumerTest, DuplicatesAreIdempotent) { TEST(PositionDeleteRangeConsumerTest, InvalidPositionsSilentlySkipped) { // Invalids at the edges, mid-run, and mixed with valid contiguous runs - // must all be dropped without breaking coalescing around them. We stay - // well below `kMaxPosition` to avoid forcing the bitmap to resize its - // backing vector to ~2^31 empty containers. + // must all be dropped without breaking coalescing around them. AssertMatchesBaseline({std::numeric_limits::min(), -5, -4, 10, 11, -999, 12, - 13, RoaringPositionBitmap::kMaxPosition + 1, - std::numeric_limits::max()}); + 13, std::numeric_limits::max()}); } TEST(PositionDeleteRangeConsumerTest, ContiguousRunAcrossKeyBoundary) { @@ -114,6 +110,11 @@ TEST(PositionDeleteRangeConsumerTest, ContiguousRunAcrossKeyBoundary) { AssertMatchesBaseline(positions); } +TEST(PositionDeleteRangeConsumerTest, ContiguousRunEndingAtMaximumPosition) { + const int64_t max = std::numeric_limits::max(); + AssertMatchesBaseline({max - 2, max - 1, max}); +} + TEST(PositionDeleteRangeConsumerTest, DispatcherAgreesAtBothDensities) { // Above the sniff threshold at densities below and above the 10% // cutoff. We can't observe the choice directly; agreement with the diff --git a/src/iceberg/test/roaring_position_bitmap_test.cc b/src/iceberg/test/roaring_position_bitmap_test.cc index 51f401b18..bf5b057ff 100644 --- a/src/iceberg/test/roaring_position_bitmap_test.cc +++ b/src/iceberg/test/roaring_position_bitmap_test.cc @@ -23,20 +23,25 @@ #include #include #include +#include #include #include #include #include #include +#include #include "iceberg/test/matchers.h" #include "iceberg/test/test_config.h" +#include "iceberg/util/endian.h" namespace iceberg { namespace { +constexpr size_t kBitmapCountSizeBytes = 8; +constexpr size_t kBitmapKeySizeBytes = 4; constexpr int64_t kBitmapSize = 0xFFFFFFFFL; constexpr int64_t kBitmapOffset = kBitmapSize + 1L; constexpr int64_t kContainerSize = 0xFFFF; // Character.MAX_VALUE @@ -174,13 +179,12 @@ TEST(RoaringPositionBitmapTest, TestAddRangeClampNegativeStart) { ASSERT_FALSE(bitmap.Contains(-1)); } -TEST(RoaringPositionBitmapTest, TestAddRangeClampBeyondMaxPosition) { +TEST(RoaringPositionBitmapTest, TestAddRangeWithMaximumExclusiveEnd) { RoaringPositionBitmap bitmap; - // Range entirely beyond kMaxPosition: after clamping both endpoints the range - // becomes empty, so no allocation or insertion happens. - bitmap.AddRange(RoaringPositionBitmap::kMaxPosition + 1, - RoaringPositionBitmap::kMaxPosition + 10); - ASSERT_TRUE(bitmap.IsEmpty()); + bitmap.AddRange(RoaringPositionBitmap::kMaxPosition - 1, + RoaringPositionBitmap::kMaxPosition); + ASSERT_TRUE(bitmap.Contains(RoaringPositionBitmap::kMaxPosition - 1)); + ASSERT_FALSE(bitmap.Contains(RoaringPositionBitmap::kMaxPosition)); } struct AddRangeNoOpParams { @@ -271,45 +275,21 @@ INSTANTIATE_TEST_SUITE_P( }), [](const ::testing::TestParamInfo& info) { return info.param.name; }); -enum class InteropBitmapShape { - kEmpty, - kOnly32BitPositions, - kSpreadAcrossKeys, -}; - struct InteropCase { const char* file_name; - InteropBitmapShape expected_shape; + size_t expected_cardinality; + std::vector expected_positions; + std::vector absent_positions; }; -void AssertInteropBitmapShape(const RoaringPositionBitmap& bitmap, - InteropBitmapShape expected_shape) { - bool saw_pos_lt_32_bit = false; - bool saw_pos_ge_32_bit = false; - - bitmap.ForEach([&](int64_t pos) { - if (pos < (int64_t{1} << 32)) { - saw_pos_lt_32_bit = true; - } else { - saw_pos_ge_32_bit = true; - } - }); - - switch (expected_shape) { - case InteropBitmapShape::kEmpty: - ASSERT_TRUE(bitmap.IsEmpty()); - ASSERT_EQ(bitmap.Cardinality(), 0u); - break; - case InteropBitmapShape::kOnly32BitPositions: - ASSERT_GT(bitmap.Cardinality(), 0u); - ASSERT_TRUE(saw_pos_lt_32_bit); - ASSERT_FALSE(saw_pos_ge_32_bit); - break; - case InteropBitmapShape::kSpreadAcrossKeys: - ASSERT_GT(bitmap.Cardinality(), 0u); - ASSERT_TRUE(saw_pos_lt_32_bit); - ASSERT_TRUE(saw_pos_ge_32_bit); - break; +void AssertInteropCase(const RoaringPositionBitmap& bitmap, + const InteropCase& test_case) { + ASSERT_EQ(bitmap.Cardinality(), test_case.expected_cardinality); + for (int64_t pos : test_case.expected_positions) { + ASSERT_TRUE(bitmap.Contains(pos)) << "Missing position: " << pos; + } + for (int64_t pos : test_case.absent_positions) { + ASSERT_FALSE(bitmap.Contains(pos)) << "Unexpected position: " << pos; } } @@ -342,7 +322,7 @@ TEST(RoaringPositionBitmapTest, TestAddPositionsRequiringMultipleBitmaps) { bitmap.Add(pos4); AssertEqualContent(bitmap, {pos1, pos2, pos3, pos4}); - ASSERT_EQ(bitmap.SerializedSizeInBytes(), 1260); + ASSERT_LT(bitmap.SerializedSizeInBytes(), 128); } TEST(RoaringPositionBitmapTest, TestAddEmptyRange) { @@ -424,6 +404,59 @@ TEST(RoaringPositionBitmapTest, TestSerializeDeserializeEmpty) { ASSERT_EQ(copy.Cardinality(), 0); } +TEST(RoaringPositionBitmapTest, TestSerializeOmitsEmptyBuckets) { + roaring::Roaring empty_bucket; + std::string bytes(kBitmapCountSizeBytes + sizeof(int32_t) + + empty_bucket.getSizeInBytes(/*portable=*/true), + '\0'); + WriteLittleEndian(int64_t{1}, bytes.data()); + WriteLittleEndian(int32_t{7}, bytes.data() + kBitmapCountSizeBytes); + empty_bucket.write(bytes.data() + kBitmapCountSizeBytes + sizeof(int32_t), + /*portable=*/true); + + ICEBERG_UNWRAP_OR_FAIL(auto bitmap, RoaringPositionBitmap::Deserialize(bytes)); + ICEBERG_UNWRAP_OR_FAIL(auto serialized, bitmap.Serialize()); + ASSERT_EQ(serialized.size(), kBitmapCountSizeBytes); + ASSERT_EQ(ReadLittleEndian(serialized.data()), 0); +} + +TEST(RoaringPositionBitmapTest, TestSerializeOrdersKeysAscending) { + RoaringPositionBitmap bitmap; + bitmap.Add(std::numeric_limits::max()); + bitmap.Add((int64_t{100} << 32) | 9); + bitmap.Add(std::numeric_limits::max()); + bitmap.Add((int64_t{1} << 32) | 7); + bitmap.Add(3); + + ICEBERG_UNWRAP_OR_FAIL(auto serialized, bitmap.Serialize()); + const char* cursor = serialized.data(); + size_t remaining = serialized.size(); + ASSERT_GE(remaining, kBitmapCountSizeBytes); + ASSERT_EQ(ReadLittleEndian(cursor), 4); + cursor += kBitmapCountSizeBytes; + remaining -= kBitmapCountSizeBytes; + + std::vector keys; + const std::vector expected_cardinalities = {2, 1, 1, 1}; + while (keys.size() < 4) { + ASSERT_GE(remaining, kBitmapKeySizeBytes); + keys.push_back(ReadLittleEndian(cursor)); + cursor += kBitmapKeySizeBytes; + remaining -= kBitmapKeySizeBytes; + + auto bucket = roaring::Roaring::readSafe(cursor, remaining); + ASSERT_EQ(bucket.cardinality(), expected_cardinalities[keys.size() - 1]); + const size_t bucket_size = bucket.getSizeInBytes(/*portable=*/true); + ASSERT_LE(bucket_size, remaining); + cursor += bucket_size; + remaining -= bucket_size; + } + + EXPECT_EQ(keys, (std::vector{0, 1, 100, std::numeric_limits::max()})); + EXPECT_EQ(keys.back(), std::numeric_limits::max()); + EXPECT_EQ(remaining, 0u); +} + TEST(RoaringPositionBitmapTest, TestSerializeDeserializeAllContainerBitmap) { RoaringPositionBitmap bitmap; @@ -509,21 +542,18 @@ TEST(RoaringPositionBitmapTest, TestOptimize) { AssertEqualContent(copy, expected_positions); } -TEST(RoaringPositionBitmapTest, TestUnsupportedPositions) { +TEST(RoaringPositionBitmapTest, TestPositionBounds) { RoaringPositionBitmap bitmap; // Negative position bitmap.Add(-1L); ASSERT_FALSE(bitmap.Contains(-1L)); - // Contains with negative position - - // Position exceeding MAX_POSITION - should be silently ignored - bitmap.Add(RoaringPositionBitmap::kMaxPosition + 1L); - ASSERT_FALSE(bitmap.Contains(RoaringPositionBitmap::kMaxPosition + 1L)); + bitmap.Add(RoaringPositionBitmap::kMaxPosition); + ASSERT_TRUE(bitmap.Contains(RoaringPositionBitmap::kMaxPosition)); - // Contains with position exceeding MAX_POSITION - should return false - ASSERT_FALSE(bitmap.Contains(RoaringPositionBitmap::kMaxPosition + 1L)); + auto copy = RoundTripSerialize(bitmap); + ASSERT_TRUE(copy.Contains(RoaringPositionBitmap::kMaxPosition)); } TEST(RoaringPositionBitmapTest, TestRandomSparseBitmap) { @@ -612,10 +642,17 @@ TEST(RoaringPositionBitmapInteropTest, TestDeserializeSupportedRoaringExamples) // roaring position bitmap interoperability test resources. static const std::vector kCases = { {.file_name = "64map32bitvals.bin", - .expected_shape = InteropBitmapShape::kOnly32BitPositions}, - {.file_name = "64mapempty.bin", .expected_shape = InteropBitmapShape::kEmpty}, + .expected_cardinality = 10, + .expected_positions = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9}, + .absent_positions = {10}}, + {.file_name = "64mapempty.bin", + .expected_cardinality = 0, + .expected_positions = {}, + .absent_positions = {0}}, {.file_name = "64mapspreadvals.bin", - .expected_shape = InteropBitmapShape::kSpreadAcrossKeys}, + .expected_cardinality = 100, + .expected_positions = {0, (int64_t{3} << 32) | 7, (int64_t{9} << 32) | 9}, + .absent_positions = {int64_t{10} << 32}}, }; for (const auto& test_case : kCases) { @@ -625,14 +662,10 @@ TEST(RoaringPositionBitmapInteropTest, TestDeserializeSupportedRoaringExamples) ASSERT_THAT(result, IsOk()); const auto& bitmap = result.value(); - AssertInteropBitmapShape(bitmap, test_case.expected_shape); - - std::set positions; - bitmap.ForEach([&](int64_t pos) { positions.insert(pos); }); - AssertEqualContent(bitmap, positions); + AssertInteropCase(bitmap, test_case); auto copy = RoundTripSerialize(bitmap); - AssertEqualContent(copy, positions); + AssertInteropCase(copy, test_case); } }