wgtmac commented on code in PR #777:
URL: https://github.com/apache/iceberg-cpp/pull/777#discussion_r3537703282
##########
src/iceberg/data/delete_loader.cc:
##########
@@ -180,21 +237,23 @@ Result<PositionDeleteIndex>
DeleteLoader::LoadPositionDeletes(
PositionDeleteIndex index;
for (const auto& file : delete_files) {
+ ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null");
+
if (file->referenced_data_file.has_value() &&
Review Comment:
Please don't skip a mismatched DV here. Java validates the DV against the
requested data file and fails when it references another file. Returning an
empty index hides bad metadata.
##########
src/iceberg/data/delete_loader.cc:
##########
@@ -180,21 +237,23 @@ Result<PositionDeleteIndex>
DeleteLoader::LoadPositionDeletes(
PositionDeleteIndex index;
for (const auto& file : delete_files) {
+ ICEBERG_PRECHECK(file != nullptr, "Delete file must not be null");
+
if (file->referenced_data_file.has_value() &&
file->referenced_data_file.value() != data_file_path) {
continue;
}
if (file->IsDeletionVector()) {
Review Comment:
Please don't merge DVs with regular position delete files in this loader
path. Java only reads a DV through the single-DV path, and Iceberg readers
should ignore matching position deletes once a DV applies.
##########
src/iceberg/deletes/position_delete_index.cc:
##########
@@ -37,6 +84,106 @@ int64_t PositionDeleteIndex::Cardinality() const {
void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) {
bitmap_.Or(other.bitmap_);
+ delete_files_.insert(delete_files_.end(), other.delete_files_.begin(),
+ other.delete_files_.end());
+}
+
+void PositionDeleteIndex::AddDeleteFile(std::shared_ptr<DataFile> delete_file)
{
+ delete_files_.push_back(std::move(delete_file));
+}
+
+Result<std::vector<uint8_t>> PositionDeleteIndex::Serialize() {
+ bitmap_.Optimize(); // run-length encode before serializing
+ ICEBERG_ASSIGN_OR_RAISE(auto vector, bitmap_.Serialize());
+
+ // The length prefix and CRC both cover the magic sequence plus the vector.
+ const size_t magic_and_vector_size = static_cast<size_t>(kMagicBytes) +
vector.size();
+ ICEBERG_PRECHECK(
+ magic_and_vector_size <=
static_cast<size_t>(std::numeric_limits<int32_t>::max()),
+ "Deletion vector is too large to serialize: {} bytes",
magic_and_vector_size);
+
+ std::vector<uint8_t> blob(static_cast<size_t>(kLengthPrefixBytes) +
+ magic_and_vector_size +
static_cast<size_t>(kCrcBytes));
+ uint8_t* buf = blob.data();
+
+ WriteBigEndian(static_cast<int32_t>(magic_and_vector_size), buf);
+ buf += kLengthPrefixBytes;
+
+ uint8_t* checksum_begin = buf;
+ std::memcpy(buf, kMagic.data(), kMagicBytes);
+ buf += kMagicBytes;
+ std::memcpy(buf, vector.data(), vector.size());
+ buf += vector.size();
+
+ WriteBigEndian(
+ ComputeCrc32(std::span<const uint8_t>(checksum_begin,
magic_and_vector_size)), buf);
+ return blob;
+}
+
+Result<PositionDeleteIndex> PositionDeleteIndex::Deserialize(
+ std::span<const uint8_t> blob, std::shared_ptr<DataFile> delete_file) {
+ ICEBERG_PRECHECK(delete_file != nullptr,
+ "Deletion vector requires a source delete file");
+ // The blob bytes must match the length recorded in the delete file metadata.
+ if (delete_file->content_size_in_bytes.has_value()) {
Review Comment:
Please require content_size_in_bytes here instead of treating it as
optional. DV metadata requires this field, and deserialization should not allow
callers to bypass the size check.
##########
src/iceberg/data/writer.h:
##########
@@ -33,6 +34,20 @@
namespace iceberg {
+/// \brief The result of writing delete files.
+///
+/// Shared by the position, equality, and deletion-vector delete writers: the
Review Comment:
Please either wire this result through the other delete writers or keep the
comment scoped to the DV writer. Position and equality delete writers still
return FileWriter::WriteResult today.
##########
src/iceberg/test/deletion_vector_writer_test.cc:
##########
@@ -0,0 +1,381 @@
+/*
+ * 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/data/deletion_vector_writer.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/data/delete_loader.h"
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/deletes/roaring_position_bitmap.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_io.h"
+
+namespace iceberg {
+
+namespace {
+
+std::shared_ptr<DataFile> FindByReferencedFile(
+ const std::vector<std::shared_ptr<DataFile>>& files, const std::string&
ref) {
+ for (const auto& file : files) {
+ if (file->referenced_data_file == ref) {
+ return file;
+ }
+ }
+ return nullptr;
+}
+
+std::shared_ptr<PartitionSpec> UnpartitionedSpec() {
+ return PartitionSpec::Unpartitioned();
+}
+
+// A load_previous_deletes callback for data files that have no existing
deletes.
+Result<std::shared_ptr<PositionDeleteIndex>>
NoPreviousDeletes(std::string_view) {
+ return nullptr;
+}
+
+} // namespace
+
+TEST(DeletionVectorWriterTest, WriteThenLoadEndToEnd) {
+ auto io = std::make_shared<MockFileIO>();
+ auto spec = UnpartitionedSpec();
+
+ std::vector<std::shared_ptr<DataFile>> delete_files;
+ {
+ ICEBERG_UNWRAP_OR_FAIL(auto writer,
+
DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+ .path = "memory://deletes.puffin",
+ .io = io,
+ .properties = {{"created-by",
"iceberg-cpp-test"}},
+ .load_previous_deletes = NoPreviousDeletes,
+ }));
+
+ ASSERT_THAT(writer->Delete("data-a.parquet", 0, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-a.parquet", 5, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-a.parquet", 10, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-b.parquet", 1, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-b.parquet", 2, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata());
+ delete_files = result.delete_files;
+ // Each referenced data file is reported once.
+ EXPECT_EQ(result.referenced_data_files.size(), 2u);
+ // No previous deletes were loaded, so nothing was rewritten.
+ EXPECT_TRUE(result.rewritten_delete_files.empty());
+ }
+
+ // One DataFile per referenced data file.
+ ASSERT_EQ(delete_files.size(), 2u);
+
+ auto dv_a = FindByReferencedFile(delete_files, "data-a.parquet");
+ auto dv_b = FindByReferencedFile(delete_files, "data-b.parquet");
+ ASSERT_NE(dv_a, nullptr);
+ ASSERT_NE(dv_b, nullptr);
+
+ // Metadata is spec-compliant for a deletion vector.
+ EXPECT_EQ(dv_a->content, DataFile::Content::kPositionDeletes);
+ EXPECT_EQ(dv_a->file_format, FileFormatType::kPuffin);
+ EXPECT_TRUE(dv_a->IsDeletionVector());
+ EXPECT_EQ(dv_a->file_path, "memory://deletes.puffin");
+ EXPECT_EQ(dv_a->record_count, 3);
+ EXPECT_TRUE(dv_a->content_offset.has_value());
+ EXPECT_TRUE(dv_a->content_size_in_bytes.has_value());
+ EXPECT_GT(dv_a->file_size_in_bytes, 0);
+ EXPECT_EQ(dv_a->partition_spec_id, spec->spec_id());
+ EXPECT_EQ(dv_b->record_count, 2);
+
+ // Both blobs live in the same Puffin file but at different offsets.
+ EXPECT_EQ(dv_a->file_path, dv_b->file_path);
+ EXPECT_NE(dv_a->content_offset.value(), dv_b->content_offset.value());
+
+ // Read back through DeleteLoader for data-a.parquet.
+ DeleteLoader loader(io);
+ {
+ auto result = loader.LoadPositionDeletes(delete_files, "data-a.parquet");
+ ASSERT_THAT(result, IsOk());
+ auto& index = result.value();
+ EXPECT_EQ(index.Cardinality(), 3);
+ EXPECT_TRUE(index.IsDeleted(0));
+ EXPECT_TRUE(index.IsDeleted(5));
+ EXPECT_TRUE(index.IsDeleted(10));
+ EXPECT_FALSE(index.IsDeleted(1));
+ }
+
+ // And for data-b.parquet (the loader filters by referenced_data_file).
+ {
+ auto result = loader.LoadPositionDeletes(delete_files, "data-b.parquet");
+ ASSERT_THAT(result, IsOk());
+ auto& index = result.value();
+ EXPECT_EQ(index.Cardinality(), 2);
+ EXPECT_TRUE(index.IsDeleted(1));
+ EXPECT_TRUE(index.IsDeleted(2));
+ EXPECT_FALSE(index.IsDeleted(0));
+ }
+}
+
+// The PositionDeleteIndex overload bulk-adds positions for a data file.
+TEST(DeletionVectorWriterTest, DeleteFromIndex) {
+ auto io = std::make_shared<MockFileIO>();
+ auto spec = UnpartitionedSpec();
+
+ PositionDeleteIndex positions;
+ positions.Delete(0);
+ positions.Delete(3, 6); // [3, 6) -> 3, 4, 5
+
+ ICEBERG_UNWRAP_OR_FAIL(auto writer,
+
DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+ .path = "memory://from-index.puffin",
+ .io = io,
+ .load_previous_deletes = NoPreviousDeletes}));
+ ASSERT_THAT(writer->Delete("data.parquet", positions, spec,
PartitionValues{}), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata());
+ ASSERT_EQ(result.delete_files.size(), 1u);
+ EXPECT_EQ(result.delete_files[0]->record_count, 4);
+
+ DeleteLoader loader(io);
+ auto loaded = loader.LoadPositionDeletes(result.delete_files,
"data.parquet");
+ ASSERT_THAT(loaded, IsOk());
+ EXPECT_EQ(loaded.value().Cardinality(), 4);
+ EXPECT_TRUE(loaded.value().IsDeleted(0));
+ EXPECT_TRUE(loaded.value().IsDeleted(5));
+ EXPECT_FALSE(loaded.value().IsDeleted(6));
+}
+
+// Previously written deletes are merged into the new vector, and the
file-scoped
+// delete files they came from are reported as rewritten.
+TEST(DeletionVectorWriterTest, LoadPreviousDeletesMergesAndReportsRewritten) {
+ auto io = std::make_shared<MockFileIO>();
+ auto spec = UnpartitionedSpec();
+
+ // Build a previous DV index carrying its source (file-scoped) delete file,
+ // the same way DeleteLoader would produce it.
+ PositionDeleteIndex previous_positions;
+ previous_positions.Delete(100);
+ previous_positions.Delete(200);
+ ICEBERG_UNWRAP_OR_FAIL(auto previous_blob, previous_positions.Serialize());
+ auto previous_dv = std::make_shared<DataFile>(DataFile{
+ .content = DataFile::Content::kPositionDeletes,
+ .file_path = "memory://old.puffin",
+ .file_format = FileFormatType::kPuffin,
+ .record_count = 2,
+ .referenced_data_file = "data.parquet",
+ .content_offset = 0,
+ .content_size_in_bytes = static_cast<int64_t>(previous_blob.size()),
+ });
+
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto writer,
+ DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+ .path = "memory://merged.puffin",
+ .io = io,
+ .load_previous_deletes =
+ [&](std::string_view path) ->
Result<std::shared_ptr<PositionDeleteIndex>> {
+ if (path != "data.parquet") {
+ return nullptr;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto index, PositionDeleteIndex::Deserialize(previous_blob,
previous_dv));
+ return std::make_shared<PositionDeleteIndex>(std::move(index));
+ },
+ }));
+
+ ASSERT_THAT(writer->Delete("data.parquet", 0, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata());
+ ASSERT_EQ(result.delete_files.size(), 1u);
+ // New position plus the two previous positions.
+ EXPECT_EQ(result.delete_files[0]->record_count, 3);
+ // The previous DV is file-scoped, so it is reported for removal.
+ ASSERT_EQ(result.rewritten_delete_files.size(), 1u);
+ EXPECT_EQ(result.rewritten_delete_files[0]->file_path,
"memory://old.puffin");
+
+ DeleteLoader loader(io);
+ auto loaded = loader.LoadPositionDeletes(result.delete_files,
"data.parquet");
+ ASSERT_THAT(loaded, IsOk());
+ EXPECT_EQ(loaded.value().Cardinality(), 3);
+ EXPECT_TRUE(loaded.value().IsDeleted(0));
+ EXPECT_TRUE(loaded.value().IsDeleted(100));
+ EXPECT_TRUE(loaded.value().IsDeleted(200));
+}
+
+// A previous delete that is not file-scoped (e.g. a partition-scoped position
+// delete) is merged into the new vector but is NOT reported as rewritten.
+TEST(DeletionVectorWriterTest,
PartitionScopedPreviousDeleteMergesButNotRewritten) {
+ auto io = std::make_shared<MockFileIO>();
+ auto spec = UnpartitionedSpec();
+
+ // No referenced_data_file and no equal file_path bounds -> not file-scoped.
+ auto previous_position_delete = std::make_shared<DataFile>(DataFile{
+ .content = DataFile::Content::kPositionDeletes,
+ .file_path = "memory://partition-deletes.parquet",
+ .file_format = FileFormatType::kParquet,
+ .record_count = 1,
+ .content_offset = 0,
+ .content_size_in_bytes = 0, // set below to match the serialized blob
+ });
+
+ PositionDeleteIndex previous_positions;
+ previous_positions.Delete(50);
+ ICEBERG_UNWRAP_OR_FAIL(auto previous_blob, previous_positions.Serialize());
+ previous_position_delete->content_size_in_bytes =
+ static_cast<int64_t>(previous_blob.size());
+
+ ICEBERG_UNWRAP_OR_FAIL(
+ auto writer,
+ DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+ .path = "memory://merged-partition.puffin",
+ .io = io,
+ .load_previous_deletes =
+ [&](std::string_view path) ->
Result<std::shared_ptr<PositionDeleteIndex>> {
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto index, PositionDeleteIndex::Deserialize(previous_blob,
Review Comment:
Please don't create this previous position-delete index by deserializing a
DV blob. This test is meant to model a Parquet position delete, so it should
get the index through DeleteLoader or add the source file directly.
##########
src/iceberg/test/delete_loader_test.cc:
##########
@@ -239,15 +302,90 @@ TEST_F(DeleteLoaderTest,
LoadPositionDeletesFastPathHonorsReferencedDataFile) {
ASSERT_FALSE(index.IsDeleted(kRowCount));
}
-TEST_F(DeleteLoaderTest, LoadPositionDeletesRejectsDV) {
+TEST_F(DeleteLoaderTest, LoadDeletionVector) {
+ auto dv_file =
+ WriteDeletionVector("dv-a.puffin", "data.parquet", {0, 5, 10,
4'000'000'000LL});
+
+ std::vector<std::shared_ptr<DataFile>> files = {dv_file};
+ auto result = loader_->LoadPositionDeletes(files, "data.parquet");
+ ASSERT_THAT(result, IsOk());
+
+ auto& index = result.value();
+ ASSERT_EQ(index.Cardinality(), 4);
+ ASSERT_TRUE(index.IsDeleted(0));
+ ASSERT_TRUE(index.IsDeleted(5));
+ ASSERT_TRUE(index.IsDeleted(10));
+ ASSERT_TRUE(index.IsDeleted(4'000'000'000LL));
+ ASSERT_FALSE(index.IsDeleted(1));
+}
+
+TEST_F(DeleteLoaderTest, LoadDeletionVectorSkipsMismatchedReferencedDataFile) {
Review Comment:
This test should expect an error for a mismatched DV, not an empty index.
Java validates referencedDataFile against the requested path before reading the
DV.
##########
src/iceberg/test/deletion_vector_writer_test.cc:
##########
@@ -0,0 +1,381 @@
+/*
+ * 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/data/deletion_vector_writer.h"
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/data/delete_loader.h"
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/deletes/roaring_position_bitmap.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/row/partition_values.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_io.h"
+
+namespace iceberg {
+
+namespace {
+
+std::shared_ptr<DataFile> FindByReferencedFile(
+ const std::vector<std::shared_ptr<DataFile>>& files, const std::string&
ref) {
+ for (const auto& file : files) {
+ if (file->referenced_data_file == ref) {
+ return file;
+ }
+ }
+ return nullptr;
+}
+
+std::shared_ptr<PartitionSpec> UnpartitionedSpec() {
+ return PartitionSpec::Unpartitioned();
+}
+
+// A load_previous_deletes callback for data files that have no existing
deletes.
+Result<std::shared_ptr<PositionDeleteIndex>>
NoPreviousDeletes(std::string_view) {
+ return nullptr;
+}
+
+} // namespace
+
+TEST(DeletionVectorWriterTest, WriteThenLoadEndToEnd) {
+ auto io = std::make_shared<MockFileIO>();
+ auto spec = UnpartitionedSpec();
+
+ std::vector<std::shared_ptr<DataFile>> delete_files;
+ {
+ ICEBERG_UNWRAP_OR_FAIL(auto writer,
+
DeletionVectorWriter::Make(DeletionVectorWriterOptions{
+ .path = "memory://deletes.puffin",
+ .io = io,
+ .properties = {{"created-by",
"iceberg-cpp-test"}},
+ .load_previous_deletes = NoPreviousDeletes,
+ }));
+
+ ASSERT_THAT(writer->Delete("data-a.parquet", 0, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-a.parquet", 5, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-a.parquet", 10, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-b.parquet", 1, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Delete("data-b.parquet", 2, spec, PartitionValues{}),
IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ ICEBERG_UNWRAP_OR_FAIL(auto result, writer->Metadata());
+ delete_files = result.delete_files;
+ // Each referenced data file is reported once.
+ EXPECT_EQ(result.referenced_data_files.size(), 2u);
+ // No previous deletes were loaded, so nothing was rewritten.
+ EXPECT_TRUE(result.rewritten_delete_files.empty());
+ }
+
+ // One DataFile per referenced data file.
+ ASSERT_EQ(delete_files.size(), 2u);
+
+ auto dv_a = FindByReferencedFile(delete_files, "data-a.parquet");
+ auto dv_b = FindByReferencedFile(delete_files, "data-b.parquet");
+ ASSERT_NE(dv_a, nullptr);
+ ASSERT_NE(dv_b, nullptr);
+
+ // Metadata is spec-compliant for a deletion vector.
Review Comment:
Please remove this narration. The assertions already show the DV metadata
being checked, and the comment adds noise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
