wgtmac commented on code in PR #708:
URL: https://github.com/apache/iceberg-cpp/pull/708#discussion_r3452818505
##########
src/iceberg/update/merging_snapshot_update.cc:
##########
@@ -757,27 +767,137 @@ MergingSnapshotUpdate::MergeDVs() const {
continue;
}
if (dvs.size() > 1) {
- // TODO(Guotao): Merge duplicate DVs for one referenced data file once
C++
- // has DVUtil/Puffin DV rewriting; Java merges them before writing
manifests.
- return NotImplemented(
- "Merging multiple deletion vectors is not supported yet for
referenced "
- "data file: {}",
- referenced_file);
+ if (!writer) {
+ ICEBERG_ASSIGN_OR_RAISE(merged_dv_path,
ctx_->NewDataLocation(std::format(
+ "merged-dvs-{}-{}.puffin",
+ SnapshotId(),
merged_dv_count_++)));
+ ICEBERG_ASSIGN_OR_RAISE(auto output,
+
ctx_->table->io()->NewOutputFile(merged_dv_path));
+ ICEBERG_ASSIGN_OR_RAISE(writer,
puffin::PuffinWriter::Make(std::move(output)));
+ }
+ auto merged_result =
+ MergeDVsForReferencedFile(referenced_file, dvs, *writer,
merged_dv_path);
+ if (!merged_result) {
+ writer.reset();
+ std::ignore = DeleteFile(merged_dv_path);
+ return std::unexpected<Error>(std::move(merged_result.error()));
+ }
+ auto merged = std::move(merged_result.value());
+ merged_files.push_back(merged.file);
+ merged_dvs_.push_back(merged);
+ result.push_back(std::move(merged));
+ continue;
}
result.push_back(dvs.front());
}
+ if (writer) {
+ auto finish = writer->Finish();
+ if (!finish) {
+ writer.reset();
+ std::ignore = DeleteFile(merged_dv_path);
+ return std::unexpected<Error>(std::move(finish.error()));
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto file_size, writer->FileSize());
+ for (const auto& file : merged_files) {
+ file->file_size_in_bytes = file_size;
+ }
+ }
+
return result;
}
+Result<MergingSnapshotUpdate::PendingDeleteFile>
+MergingSnapshotUpdate::MergeDVsForReferencedFile(
+ const std::string& referenced_file, const std::vector<PendingDeleteFile>&
dvs,
+ puffin::PuffinWriter& writer, const std::string& path) {
+ const auto& first_file = dvs.front().file;
+ const auto first_data_sequence_number = dvs.front().data_sequence_number;
+ const auto first_spec_id = first_file->partition_spec_id;
+ RoaringPositionBitmap bitmap;
+
+ for (const auto& dv : dvs) {
+ ICEBERG_PRECHECK(dv.data_sequence_number == first_data_sequence_number,
+ "Cannot merge DVs, mismatched sequence numbers for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->partition_spec_id == first_spec_id,
+ "Cannot merge DVs, mismatched partition specs for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->partition == first_file->partition,
+ "Cannot merge DVs, mismatched partition tuples for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->content_offset.has_value(),
+ "DV must have a content offset: {}", dv.file->file_path);
+ ICEBERG_PRECHECK(dv.file->content_size_in_bytes.has_value(),
+ "DV must have a content size: {}", dv.file->file_path);
+
+ ICEBERG_ASSIGN_OR_RAISE(auto input,
+
ctx_->table->io()->NewInputFile(dv.file->file_path));
+ std::optional<int64_t> file_size;
+ if (dv.file->file_size_in_bytes > 0) {
+ file_size = dv.file->file_size_in_bytes;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, puffin::PuffinReader::Make(
+ std::move(input), std::nullopt,
file_size));
+ ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->ReadFileMetadata());
+ auto blob_it = std::ranges::find_if(metadata.blobs, [&](const auto& blob) {
+ return blob.type == puffin::StandardBlobTypes::kDeletionVectorV1 &&
+ blob.offset == dv.file->content_offset.value() &&
+ blob.length == dv.file->content_size_in_bytes.value();
+ });
+ ICEBERG_PRECHECK(blob_it != metadata.blobs.end(),
+ "Cannot find DV blob at offset {} with length {} in {}",
+ dv.file->content_offset.value(),
+ dv.file->content_size_in_bytes.value(),
dv.file->file_path);
+ ICEBERG_ASSIGN_OR_RAISE(auto blob, reader->ReadBlob(*blob_it));
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto dv_bitmap,
+ RoaringPositionBitmap::Deserialize(std::string_view(
Review Comment:
Java/spec-compliant deletion-vector-v1 blobs are not raw Roaring payloads:
they include the 4-byte length, magic bytes, bitmap, and CRC, and Java
validates CRC/cardinality during deserialize. `PuffinReader::ReadBlob` returns
the stored blob bytes unchanged, so passing them directly to
`RoaringPositionBitmap::Deserialize()` will reject or misread DVs written by
Java. The writer below has the symmetric issue because `bitmap.Serialize()`
writes only the raw bitmap. Please add DV-level framing/parsing here.
##########
src/iceberg/update/merging_snapshot_update.cc:
##########
@@ -757,27 +767,137 @@ MergingSnapshotUpdate::MergeDVs() const {
continue;
}
if (dvs.size() > 1) {
- // TODO(Guotao): Merge duplicate DVs for one referenced data file once
C++
- // has DVUtil/Puffin DV rewriting; Java merges them before writing
manifests.
- return NotImplemented(
- "Merging multiple deletion vectors is not supported yet for
referenced "
- "data file: {}",
- referenced_file);
+ if (!writer) {
+ ICEBERG_ASSIGN_OR_RAISE(merged_dv_path,
ctx_->NewDataLocation(std::format(
+ "merged-dvs-{}-{}.puffin",
+ SnapshotId(),
merged_dv_count_++)));
+ ICEBERG_ASSIGN_OR_RAISE(auto output,
+
ctx_->table->io()->NewOutputFile(merged_dv_path));
+ ICEBERG_ASSIGN_OR_RAISE(writer,
puffin::PuffinWriter::Make(std::move(output)));
+ }
+ auto merged_result =
+ MergeDVsForReferencedFile(referenced_file, dvs, *writer,
merged_dv_path);
+ if (!merged_result) {
+ writer.reset();
+ std::ignore = DeleteFile(merged_dv_path);
+ return std::unexpected<Error>(std::move(merged_result.error()));
+ }
+ auto merged = std::move(merged_result.value());
+ merged_files.push_back(merged.file);
+ merged_dvs_.push_back(merged);
+ result.push_back(std::move(merged));
+ continue;
}
result.push_back(dvs.front());
}
+ if (writer) {
+ auto finish = writer->Finish();
+ if (!finish) {
+ writer.reset();
+ std::ignore = DeleteFile(merged_dv_path);
+ return std::unexpected<Error>(std::move(finish.error()));
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto file_size, writer->FileSize());
+ for (const auto& file : merged_files) {
+ file->file_size_in_bytes = file_size;
+ }
+ }
+
return result;
}
+Result<MergingSnapshotUpdate::PendingDeleteFile>
+MergingSnapshotUpdate::MergeDVsForReferencedFile(
+ const std::string& referenced_file, const std::vector<PendingDeleteFile>&
dvs,
+ puffin::PuffinWriter& writer, const std::string& path) {
+ const auto& first_file = dvs.front().file;
+ const auto first_data_sequence_number = dvs.front().data_sequence_number;
+ const auto first_spec_id = first_file->partition_spec_id;
+ RoaringPositionBitmap bitmap;
+
+ for (const auto& dv : dvs) {
+ ICEBERG_PRECHECK(dv.data_sequence_number == first_data_sequence_number,
+ "Cannot merge DVs, mismatched sequence numbers for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->partition_spec_id == first_spec_id,
+ "Cannot merge DVs, mismatched partition specs for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->partition == first_file->partition,
+ "Cannot merge DVs, mismatched partition tuples for {}",
+ referenced_file);
+ ICEBERG_PRECHECK(dv.file->content_offset.has_value(),
+ "DV must have a content offset: {}", dv.file->file_path);
+ ICEBERG_PRECHECK(dv.file->content_size_in_bytes.has_value(),
+ "DV must have a content size: {}", dv.file->file_path);
+
+ ICEBERG_ASSIGN_OR_RAISE(auto input,
+
ctx_->table->io()->NewInputFile(dv.file->file_path));
+ std::optional<int64_t> file_size;
+ if (dv.file->file_size_in_bytes > 0) {
+ file_size = dv.file->file_size_in_bytes;
+ }
+ ICEBERG_ASSIGN_OR_RAISE(auto reader, puffin::PuffinReader::Make(
+ std::move(input), std::nullopt,
file_size));
+ ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->ReadFileMetadata());
+ auto blob_it = std::ranges::find_if(metadata.blobs, [&](const auto& blob) {
+ return blob.type == puffin::StandardBlobTypes::kDeletionVectorV1 &&
+ blob.offset == dv.file->content_offset.value() &&
+ blob.length == dv.file->content_size_in_bytes.value();
+ });
+ ICEBERG_PRECHECK(blob_it != metadata.blobs.end(),
+ "Cannot find DV blob at offset {} with length {} in {}",
+ dv.file->content_offset.value(),
+ dv.file->content_size_in_bytes.value(),
dv.file->file_path);
+ ICEBERG_ASSIGN_OR_RAISE(auto blob, reader->ReadBlob(*blob_it));
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto dv_bitmap,
+ RoaringPositionBitmap::Deserialize(std::string_view(
+ reinterpret_cast<const char*>(blob.second.data()),
blob.second.size())));
+ bitmap.Or(dv_bitmap);
+ }
+
+ bitmap.Optimize();
+ ICEBERG_ASSIGN_OR_RAISE(auto serialized, bitmap.Serialize());
+
+ ICEBERG_ASSIGN_OR_RAISE(
+ auto blob_metadata,
+ writer.Write(puffin::Blob{
Review Comment:
This Puffin blob metadata does not match Java or the Puffin DV spec. Java
`BaseDVFileWriter` writes `fields=[ROW_POSITION]`, `snapshot-id=-1`,
`sequence-number=-1`, and `referenced-data-file`/`cardinality` properties.
Writing empty fields, concrete snapshot/sequence values, and no properties
produces invalid DV footer metadata for other implementations. Please mirror
the Java DV writer metadata.
--
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]
