address review: make flush threshold configurable and add test

Make kFlushThreshold configurable via PositionDeleteWriterOptions with
a default of 1000. Add AutoFlushOnThreshold test that uses a small
threshold to verify the automatic flush logic.

Author: shangxinli

src/iceberg/data/position_delete_writer.cc

@@ -70,7 +70,7 @@ class PositionDeleteWriter::Impl {
     buffered_positions_.push_back(pos);
     referenced_paths_.emplace(file_path);
 
-    if (static_cast<int64_t>(buffered_paths_.size()) >= kFlushThreshold) {
+    if (static_cast<int64_t>(buffered_paths_.size()) >= options_.flush_threshold) {
       return FlushBuffer();
     }
     return {};
@@ -145,8 +145,6 @@ class PositionDeleteWriter::Impl {
   }
 
  private:
-  static constexpr int64_t kFlushThreshold = 1000;
-
   Impl(PositionDeleteWriterOptions options, std::shared_ptr<Schema> delete_schema,
        std::unique_ptr<Writer> writer)
       : options_(std::move(options)),

src/iceberg/data/position_delete_writer.h

@@ -47,6 +47,7 @@ struct ICEBERG_EXPORT PositionDeleteWriterOptions {
   FileFormatType format = FileFormatType::kParquet;
   std::shared_ptr<FileIO> io;
   std::shared_ptr<Schema> row_schema;  // Optional row data schema
+  int64_t flush_threshold = 1000;      // Number of buffered deletes before auto-flush
   std::unordered_map<std::string, std::string> properties;
 };
 

src/iceberg/test/data_writer_test.cc

@@ -268,14 +268,15 @@ TEST_F(DataWriterTest, WriteMultipleBatches) {
 
 class PositionDeleteWriterTest : public DataWriterTest {
  protected:
-  PositionDeleteWriterOptions MakeDeleteOptions() {
+  PositionDeleteWriterOptions MakeDeleteOptions(int64_t flush_threshold = 1000) {
     return PositionDeleteWriterOptions{
         .path = "test_deletes.parquet",
         .schema = schema_,
         .spec = partition_spec_,
         .partition = PartitionValues{},
         .format = FileFormatType::kParquet,
         .io = file_io_,
+        .flush_threshold = flush_threshold,
         .properties = {{"write.parquet.compression-codec", "uncompressed"}},
     };
   }
@@ -396,4 +397,30 @@ TEST_F(PositionDeleteWriterTest, WriteBatchData) {
   EXPECT_GT(data_file->file_size_in_bytes, 0);
 }
 
+TEST_F(PositionDeleteWriterTest, AutoFlushOnThreshold) {
+  // Use a small flush threshold to trigger automatic flush
+  const int64_t flush_threshold = 5;
+  auto writer_result = PositionDeleteWriter::Make(MakeDeleteOptions(flush_threshold));
+  ASSERT_THAT(writer_result, IsOk());
+  auto writer = std::move(writer_result.value());
+
+  // Write more deletes than the threshold to trigger auto-flush
+  for (int64_t i = 0; i < 12; ++i) {
+    ASSERT_THAT(writer->WriteDelete("data_file.parquet", i), IsOk());
+  }
+
+  // Length should be > 0 since auto-flush should have written data
+  auto length_result = writer->Length();
+  ASSERT_THAT(length_result, IsOk());
+  EXPECT_GT(length_result.value(), 0);
+
+  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);
+}
+
 }  // namespace iceberg