You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "shangxinli (via GitHub)" <gi...@apache.org> on 2026/02/28 17:53:03 UTC
[PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
shangxinli opened a new pull request, #582:
URL: https://github.com/apache/iceberg-cpp/pull/582
Implement the PositionDeleteWriter following the same PIMPL pattern as DataWriter. The writer supports both buffered WriteDelete(file_path, pos) calls and direct Write(ArrowArray*) for pre-formed batches. Metadata reports content=kPositionDeletes with sort_order_id=nullopt per spec, and tracks referenced_data_file when all deletes target a single file.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac merged PR #582:
URL: https://github.com/apache/iceberg-cpp/pull/582
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
Posted by "evindj (via GitHub)" <gi...@apache.org>.
evindj commented on code in PR #582:
URL: https://github.com/apache/iceberg-cpp/pull/582#discussion_r2918437396
##########
src/iceberg/test/data_writer_test.cc:
##########
@@ -264,4 +266,134 @@ TEST_F(DataWriterTest, WriteMultipleBatches) {
EXPECT_GT(data_file->file_size_in_bytes, 0);
}
+class PositionDeleteWriterTest : public DataWriterTest {
+ protected:
+ PositionDeleteWriterOptions MakeDeleteOptions() {
+ return PositionDeleteWriterOptions{
+ .path = "test_deletes.parquet",
+ .schema = schema_,
+ .spec = partition_spec_,
+ .partition = PartitionValues{},
+ .format = FileFormatType::kParquet,
+ .io = file_io_,
+ .properties = {{"write.parquet.compression-codec", "uncompressed"}},
+ };
+ }
+
+ std::shared_ptr<::arrow::Array> CreatePositionDeleteData() {
+ auto delete_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+ MetadataColumns::kDeleteFilePath, MetadataColumns::kDeleteFilePos});
+
+ ArrowSchema arrow_c_schema;
+ ICEBERG_THROW_NOT_OK(ToArrowSchema(*delete_schema, &arrow_c_schema));
+ auto arrow_type = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+
+ return ::arrow::json::ArrayFromJSONString(
+ ::arrow::struct_(arrow_type->fields()),
+ R"([["data_file_1.parquet", 0], ["data_file_1.parquet", 5], ["data_file_1.parquet", 10]])")
+ .ValueOrDie();
+ }
+};
+
+TEST_F(PositionDeleteWriterTest, WriteDeleteAndClose) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 5), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 10), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto length_result = writer->Length();
+ ASSERT_THAT(length_result, IsOk());
+ EXPECT_GT(length_result.value(), 0);
+}
+
+TEST_F(PositionDeleteWriterTest, MetadataAfterClose) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 5), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& write_result = metadata_result.value();
+ ASSERT_EQ(write_result.data_files.size(), 1);
+
+ const auto& data_file = write_result.data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_EQ(data_file->file_path, "test_deletes.parquet");
+ EXPECT_EQ(data_file->file_format, FileFormatType::kParquet);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+ EXPECT_FALSE(data_file->sort_order_id.has_value());
+}
+
+TEST_F(PositionDeleteWriterTest, MetadataBeforeCloseReturnsError) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsError(ErrorKind::kValidationFailed));
+ EXPECT_THAT(metadata_result,
+ HasErrorMessage("Cannot get metadata before closing the writer"));
+}
+
+TEST_F(PositionDeleteWriterTest, CloseIsIdempotent) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+}
+
+TEST_F(PositionDeleteWriterTest, WriteMultipleDeletes) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ for (int64_t i = 0; i < 100; ++i) {
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", i), IsOk());
+ }
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& data_file = metadata_result.value().data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+}
+
+TEST_F(PositionDeleteWriterTest, WriteBatchData) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ auto test_data = CreatePositionDeleteData();
+ ArrowArray arrow_array;
+ ASSERT_TRUE(::arrow::ExportArray(*test_data, &arrow_array).ok());
+ ASSERT_THAT(writer->Write(&arrow_array), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& data_file = metadata_result.value().data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+}
Review Comment:
It would be nice if you could test the automatic flush logic.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,203 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= kFlushThreshold) {
Review Comment:
Do we want to make ```kFlushThreshold``` configurable via options?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
Posted by "shangxinli (via GitHub)" <gi...@apache.org>.
shangxinli commented on code in PR #582:
URL: https://github.com/apache/iceberg-cpp/pull/582#discussion_r2934289182
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= options_.flush_threshold) {
+ return FlushBuffer();
+ }
+ return {};
+ }
+
+ Result<int64_t> Length() const {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->length();
+ }
+
+ Status Close() {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ if (closed_) {
+ return {};
+ }
+ if (!buffered_paths_.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(FlushBuffer());
+ }
+ ICEBERG_RETURN_UNEXPECTED(writer_->Close());
+ closed_ = true;
+ return {};
+ }
+
+ Result<FileWriter::WriteResult> Metadata() {
+ ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
+
+ ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics());
+ ICEBERG_ASSIGN_OR_RAISE(auto length, writer_->length());
+ auto split_offsets = writer_->split_offsets();
+
+ // Serialize literal bounds to binary format
+ std::map<int32_t, std::vector<uint8_t>> lower_bounds_map;
+ for (const auto& [col_id, literal] : metrics.lower_bounds) {
+ ICEBERG_ASSIGN_OR_RAISE(auto serialized, literal.Serialize());
+ lower_bounds_map[col_id] = std::move(serialized);
+ }
+ std::map<int32_t, std::vector<uint8_t>> upper_bounds_map;
+ for (const auto& [col_id, literal] : metrics.upper_bounds) {
+ ICEBERG_ASSIGN_OR_RAISE(auto serialized, literal.Serialize());
+ upper_bounds_map[col_id] = std::move(serialized);
+ }
+
+ // Set referenced_data_file if all deletes reference the same data file
+ std::optional<std::string> referenced_data_file;
+ if (referenced_paths_.size() == 1) {
+ referenced_data_file = *referenced_paths_.begin();
+ }
+
+ auto data_file = std::make_shared<DataFile>(DataFile{
+ .content = DataFile::Content::kPositionDeletes,
+ .file_path = options_.path,
+ .file_format = options_.format,
+ .partition = options_.partition,
+ .record_count = metrics.row_count.value_or(-1),
+ .file_size_in_bytes = length,
+ .column_sizes = {metrics.column_sizes.begin(), metrics.column_sizes.end()},
+ .value_counts = {metrics.value_counts.begin(), metrics.value_counts.end()},
+ .null_value_counts = {metrics.null_value_counts.begin(),
+ metrics.null_value_counts.end()},
+ .nan_value_counts = {metrics.nan_value_counts.begin(),
+ metrics.nan_value_counts.end()},
+ .lower_bounds = std::move(lower_bounds_map),
+ .upper_bounds = std::move(upper_bounds_map),
+ .split_offsets = std::move(split_offsets),
+ .sort_order_id = std::nullopt,
+ .referenced_data_file = std::move(referenced_data_file),
+ });
+
+ FileWriter::WriteResult result;
+ result.data_files.push_back(std::move(data_file));
+ return result;
+ }
+
+ private:
+ Impl(PositionDeleteWriterOptions options, std::shared_ptr<Schema> delete_schema,
+ std::unique_ptr<Writer> writer)
+ : options_(std::move(options)),
+ delete_schema_(std::move(delete_schema)),
+ writer_(std::move(writer)) {}
+
+ Status FlushBuffer() {
+ ArrowSchema arrow_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*delete_schema_, &arrow_schema));
+
+ ArrowArray array;
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayInitFromSchema(&array, &arrow_schema, &error), error);
Review Comment:
Done. Added `ArrowSchemaGuard` and `ArrowArrayGuard` in `FlushBuffer` to ensure proper cleanup on early returns. Also fixed the guard destructors to check `release \!= nullptr` before calling `ArrowArrayRelease`/`ArrowSchemaRelease`, since `Write()` transfers ownership of the array via `ImportRecordBatch` (which sets `release = nullptr`).
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= options_.flush_threshold) {
+ return FlushBuffer();
+ }
+ return {};
+ }
+
+ Result<int64_t> Length() const {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->length();
+ }
+
+ Status Close() {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ if (closed_) {
+ return {};
+ }
+ if (!buffered_paths_.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(FlushBuffer());
+ }
+ ICEBERG_RETURN_UNEXPECTED(writer_->Close());
+ closed_ = true;
+ return {};
+ }
+
+ Result<FileWriter::WriteResult> Metadata() {
+ ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
+
+ ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics());
+ ICEBERG_ASSIGN_OR_RAISE(auto length, writer_->length());
+ auto split_offsets = writer_->split_offsets();
+
+ // Serialize literal bounds to binary format
+ std::map<int32_t, std::vector<uint8_t>> lower_bounds_map;
Review Comment:
Done. Added metrics filtering in `Metadata()` to match Java's behavior: always erase `value_counts`, `null_value_counts`, and `nan_value_counts` for `kDeleteFilePathColumnId` and `kDeleteFilePosColumnId`. When `referenced_paths_.size() > 1`, also erase `lower_bounds` and `upper_bounds` for those columns. This mirrors Java's `MetricsUtil.copyWithoutFieldCounts` / `copyWithoutFieldCountsAndBounds` logic.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
Review Comment:
Added a TODO comment. Extracting paths from the raw ArrowArray would require reading the string column from the C Data Interface, which adds complexity. Leaving this for a follow-up since the primary API is `WriteDelete()` which correctly tracks paths.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
Review Comment:
Added a TODO comment. Note that Java has also deprecated the row field in `PositionDelete` as of v1.11.0, so this is low priority. Will add support in a follow-up if needed.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #582:
URL: https://github.com/apache/iceberg-cpp/pull/582#discussion_r2931904994
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= options_.flush_threshold) {
+ return FlushBuffer();
+ }
+ return {};
+ }
+
+ Result<int64_t> Length() const {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->length();
+ }
+
+ Status Close() {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ if (closed_) {
+ return {};
+ }
+ if (!buffered_paths_.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(FlushBuffer());
+ }
+ ICEBERG_RETURN_UNEXPECTED(writer_->Close());
+ closed_ = true;
+ return {};
+ }
+
+ Result<FileWriter::WriteResult> Metadata() {
+ ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
+
+ ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics());
+ ICEBERG_ASSIGN_OR_RAISE(auto length, writer_->length());
+ auto split_offsets = writer_->split_offsets();
+
+ // Serialize literal bounds to binary format
+ std::map<int32_t, std::vector<uint8_t>> lower_bounds_map;
Review Comment:
**Parity Issue (Metrics Bloat):**
Java's `PositionDeleteWriter` explicitly drops field counts (and bounds if referencing multiple files) for the `file_path` and `pos` columns to avoid bloating the manifest. C++ currently copies all metrics blindly.
*Fix:* Add a `// TODO` or implement logic to filter out `lower_bounds`, `upper_bounds`, and counts for `MetadataColumns::kDeleteFilePathColumnId` and `MetadataColumns::kDeleteFilePosColumnId`.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= options_.flush_threshold) {
+ return FlushBuffer();
+ }
+ return {};
+ }
+
+ Result<int64_t> Length() const {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->length();
+ }
+
+ Status Close() {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ if (closed_) {
+ return {};
+ }
+ if (!buffered_paths_.empty()) {
+ ICEBERG_RETURN_UNEXPECTED(FlushBuffer());
+ }
+ ICEBERG_RETURN_UNEXPECTED(writer_->Close());
+ closed_ = true;
+ return {};
+ }
+
+ Result<FileWriter::WriteResult> Metadata() {
+ ICEBERG_CHECK(closed_, "Cannot get metadata before closing the writer");
+
+ ICEBERG_ASSIGN_OR_RAISE(auto metrics, writer_->metrics());
+ ICEBERG_ASSIGN_OR_RAISE(auto length, writer_->length());
+ auto split_offsets = writer_->split_offsets();
+
+ // Serialize literal bounds to binary format
+ std::map<int32_t, std::vector<uint8_t>> lower_bounds_map;
+ for (const auto& [col_id, literal] : metrics.lower_bounds) {
+ ICEBERG_ASSIGN_OR_RAISE(auto serialized, literal.Serialize());
+ lower_bounds_map[col_id] = std::move(serialized);
+ }
+ std::map<int32_t, std::vector<uint8_t>> upper_bounds_map;
+ for (const auto& [col_id, literal] : metrics.upper_bounds) {
+ ICEBERG_ASSIGN_OR_RAISE(auto serialized, literal.Serialize());
+ upper_bounds_map[col_id] = std::move(serialized);
+ }
+
+ // Set referenced_data_file if all deletes reference the same data file
+ std::optional<std::string> referenced_data_file;
+ if (referenced_paths_.size() == 1) {
+ referenced_data_file = *referenced_paths_.begin();
+ }
+
+ auto data_file = std::make_shared<DataFile>(DataFile{
+ .content = DataFile::Content::kPositionDeletes,
+ .file_path = options_.path,
+ .file_format = options_.format,
+ .partition = options_.partition,
+ .record_count = metrics.row_count.value_or(-1),
+ .file_size_in_bytes = length,
+ .column_sizes = {metrics.column_sizes.begin(), metrics.column_sizes.end()},
+ .value_counts = {metrics.value_counts.begin(), metrics.value_counts.end()},
+ .null_value_counts = {metrics.null_value_counts.begin(),
+ metrics.null_value_counts.end()},
+ .nan_value_counts = {metrics.nan_value_counts.begin(),
+ metrics.nan_value_counts.end()},
+ .lower_bounds = std::move(lower_bounds_map),
+ .upper_bounds = std::move(upper_bounds_map),
+ .split_offsets = std::move(split_offsets),
+ .sort_order_id = std::nullopt,
+ .referenced_data_file = std::move(referenced_data_file),
+ });
+
+ FileWriter::WriteResult result;
+ result.data_files.push_back(std::move(data_file));
+ return result;
+ }
+
+ private:
+ Impl(PositionDeleteWriterOptions options, std::shared_ptr<Schema> delete_schema,
+ std::unique_ptr<Writer> writer)
+ : options_(std::move(options)),
+ delete_schema_(std::move(delete_schema)),
+ writer_(std::move(writer)) {}
+
+ Status FlushBuffer() {
+ ArrowSchema arrow_schema;
+ ICEBERG_RETURN_UNEXPECTED(ToArrowSchema(*delete_schema_, &arrow_schema));
+
+ ArrowArray array;
+ ArrowError error;
+ ICEBERG_NANOARROW_RETURN_UNEXPECTED_WITH_ERROR(
+ ArrowArrayInitFromSchema(&array, &arrow_schema, &error), error);
Review Comment:
**Logic Issue (Memory Leak):**
`arrow_schema` and `array` are not protected by RAII wrappers. If any `ICEBERG_NANOARROW_RETURN_UNEXPECTED` macro fails before `arrow_schema.release(&arrow_schema)` or before `writer_->Write(&array)` transfers ownership, the memory will leak.
*Fix:* Use `internal::ArrowSchemaGuard schema_guard(&arrow_schema);` and `internal::ArrowArrayGuard array_guard(&array);` to safely manage their lifecycles.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
Review Comment:
**Parity / Logic Issue:**
If users write batches directly using `Write(ArrowArray* data)`, the `referenced_paths_` set is never updated. As a result, `Metadata()` will fail to correctly populate `referenced_data_file` when the batch contains deletes for a single data file.
*Fix:* Add a `// TODO: Extract paths from ArrowArray to update referenced_paths_` or return `NotImplemented` if batch writing should not be directly used this way.
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,201 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
Review Comment:
**Parity Issue (Missing Implementation):**
`PositionDeleteWriterOptions` accepts a `row_schema`, but `Impl::Make` completely ignores it, creating a schema with only `file_path` and `pos`. The V2 spec allows position deletes to optionally include the deleted row.
*Fix:* Add a `// TODO: Support writing row data if options.row_schema is provided`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] feat: Implement PositionDeleteWriter for position delete files [iceberg-cpp]
Posted by "shangxinli (via GitHub)" <gi...@apache.org>.
shangxinli commented on code in PR #582:
URL: https://github.com/apache/iceberg-cpp/pull/582#discussion_r2920004174
##########
src/iceberg/data/position_delete_writer.cc:
##########
@@ -19,26 +19,203 @@
#include "iceberg/data/position_delete_writer.h"
+#include <map>
+#include <set>
+#include <vector>
+
+#include <nanoarrow/nanoarrow.h>
+
+#include "iceberg/arrow/nanoarrow_status_internal.h"
+#include "iceberg/file_writer.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/util/macros.h"
+
namespace iceberg {
class PositionDeleteWriter::Impl {
public:
+ static Result<std::unique_ptr<Impl>> Make(PositionDeleteWriterOptions options) {
+ // Build the position delete schema with file_path and pos columns
+ std::vector<SchemaField> fields;
+ fields.push_back(MetadataColumns::kDeleteFilePath);
+ fields.push_back(MetadataColumns::kDeleteFilePos);
+
+ auto delete_schema = std::make_shared<Schema>(std::move(fields));
+
+ WriterOptions writer_options{
+ .path = options.path,
+ .schema = delete_schema,
+ .io = options.io,
+ .properties = WriterProperties::FromMap(options.properties),
+ };
+
+ ICEBERG_ASSIGN_OR_RAISE(auto writer,
+ WriterFactoryRegistry::Open(options.format, writer_options));
+
+ return std::unique_ptr<Impl>(
+ new Impl(std::move(options), std::move(delete_schema), std::move(writer)));
+ }
+
+ Status Write(ArrowArray* data) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ return writer_->Write(data);
+ }
+
+ Status WriteDelete(std::string_view file_path, int64_t pos) {
+ ICEBERG_DCHECK(writer_, "Writer not initialized");
+ buffered_paths_.emplace_back(file_path);
+ buffered_positions_.push_back(pos);
+ referenced_paths_.emplace(file_path);
+
+ if (static_cast<int64_t>(buffered_paths_.size()) >= kFlushThreshold) {
Review Comment:
Good idea. Made flush_threshold configurable via PositionDeleteWriterOptions with a default of 1000.
##########
src/iceberg/test/data_writer_test.cc:
##########
@@ -264,4 +266,134 @@ TEST_F(DataWriterTest, WriteMultipleBatches) {
EXPECT_GT(data_file->file_size_in_bytes, 0);
}
+class PositionDeleteWriterTest : public DataWriterTest {
+ protected:
+ PositionDeleteWriterOptions MakeDeleteOptions() {
+ return PositionDeleteWriterOptions{
+ .path = "test_deletes.parquet",
+ .schema = schema_,
+ .spec = partition_spec_,
+ .partition = PartitionValues{},
+ .format = FileFormatType::kParquet,
+ .io = file_io_,
+ .properties = {{"write.parquet.compression-codec", "uncompressed"}},
+ };
+ }
+
+ std::shared_ptr<::arrow::Array> CreatePositionDeleteData() {
+ auto delete_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+ MetadataColumns::kDeleteFilePath, MetadataColumns::kDeleteFilePos});
+
+ ArrowSchema arrow_c_schema;
+ ICEBERG_THROW_NOT_OK(ToArrowSchema(*delete_schema, &arrow_c_schema));
+ auto arrow_type = ::arrow::ImportType(&arrow_c_schema).ValueOrDie();
+
+ return ::arrow::json::ArrayFromJSONString(
+ ::arrow::struct_(arrow_type->fields()),
+ R"([["data_file_1.parquet", 0], ["data_file_1.parquet", 5], ["data_file_1.parquet", 10]])")
+ .ValueOrDie();
+ }
+};
+
+TEST_F(PositionDeleteWriterTest, WriteDeleteAndClose) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 5), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 10), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto length_result = writer->Length();
+ ASSERT_THAT(length_result, IsOk());
+ EXPECT_GT(length_result.value(), 0);
+}
+
+TEST_F(PositionDeleteWriterTest, MetadataAfterClose) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 5), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& write_result = metadata_result.value();
+ ASSERT_EQ(write_result.data_files.size(), 1);
+
+ const auto& data_file = write_result.data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_EQ(data_file->file_path, "test_deletes.parquet");
+ EXPECT_EQ(data_file->file_format, FileFormatType::kParquet);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+ EXPECT_FALSE(data_file->sort_order_id.has_value());
+}
+
+TEST_F(PositionDeleteWriterTest, MetadataBeforeCloseReturnsError) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsError(ErrorKind::kValidationFailed));
+ EXPECT_THAT(metadata_result,
+ HasErrorMessage("Cannot get metadata before closing the writer"));
+}
+
+TEST_F(PositionDeleteWriterTest, CloseIsIdempotent) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", 0), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+ ASSERT_THAT(writer->Close(), IsOk());
+}
+
+TEST_F(PositionDeleteWriterTest, WriteMultipleDeletes) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ for (int64_t i = 0; i < 100; ++i) {
+ ASSERT_THAT(writer->WriteDelete("data_file.parquet", i), IsOk());
+ }
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& data_file = metadata_result.value().data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+}
+
+TEST_F(PositionDeleteWriterTest, WriteBatchData) {
+ auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions());
+ ASSERT_THAT(writer_result, IsOk());
+ auto writer = std::move(writer_result.value());
+
+ auto test_data = CreatePositionDeleteData();
+ ArrowArray arrow_array;
+ ASSERT_TRUE(::arrow::ExportArray(*test_data, &arrow_array).ok());
+ ASSERT_THAT(writer->Write(&arrow_array), IsOk());
+
+ ASSERT_THAT(writer->Close(), IsOk());
+
+ auto metadata_result = writer->Metadata();
+ ASSERT_THAT(metadata_result, IsOk());
+
+ const auto& data_file = metadata_result.value().data_files[0];
+ EXPECT_EQ(data_file->content, DataFile::Content::kPositionDeletes);
+ EXPECT_GT(data_file->file_size_in_bytes, 0);
+}
Review Comment:
Added AutoFlushOnThreshold test that sets threshold to 5, writes 12 deletes, and verifies data was flushed automatically before Close().
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org