feat: add FileWriterFactory and writer stub implementations

Add complete API structure for Iceberg data file writers including:

DataWriter:
- Add DataWriterOptions struct with schema, partition spec, format, etc.
- Add DataWriter class extending FileWriter
- Stub implementation returning NotImplemented errors

PositionDeleteWriter:
- Add PositionDeleteWriterOptions struct with optional row schema
- Add PositionDeleteWriter class with WriteDelete method
- Stub implementation returning NotImplemented errors

EqualityDeleteWriter:
- Add EqualityDeleteWriterOptions struct with equality_field_ids
- Add EqualityDeleteWriter class with equality_field_ids accessor
- Stub implementation returning NotImplemented errors

FileWriterFactory:
- Add factory class for creating all three writer types
- Implements NewDataWriter, NewPositionDeleteWriter, NewEqualityDeleteWriter
- Supports configuration via SetEqualityDeleteConfig and SetPositionDeleteRowSchema
- Factory methods construct options and delegate to concrete writer Make methods

All implementations use pimpl idiom for ABI stability. Stub implementations
will be replaced with full functionality in subsequent tasks.

Related to #441 task 5

Author: shangxinli

src/iceberg/data/writer.cc

@@ -23,4 +23,247 @@ namespace iceberg {
 
 FileWriter::~FileWriter() = default;
 
+//=============================================================================
+// DataWriter::Impl
+//=============================================================================
+
+class DataWriter::Impl {
+ public:
+  explicit Impl(const DataWriterOptions& options) : options_(options) {}
+
+  DataWriterOptions options_;
+
+  // TODO: Add the following when implementing:
+  // - FileAppender or equivalent writer for the format
+  // - Metrics collection
+  // - Split offsets tracking
+  // - Encryption key metadata handling
+};
+
+//=============================================================================
+// DataWriter
+//=============================================================================
+
+Result<std::unique_ptr<DataWriter>> DataWriter::Make(const DataWriterOptions& options) {
+  return NotImplemented("DataWriter implementation not yet available");
+}
+
+DataWriter::DataWriter(std::unique_ptr<Impl> impl) : impl_(std::move(impl)) {}
+
+DataWriter::~DataWriter() = default;
+
+Status DataWriter::Write(ArrowArray* data) {
+  return NotImplemented("DataWriter::Write not yet implemented");
+}
+
+Result<int64_t> DataWriter::Length() const {
+  return NotImplemented("DataWriter::Length not yet implemented");
+}
+
+Status DataWriter::Close() {
+  return NotImplemented("DataWriter::Close not yet implemented");
+}
+
+Result<FileWriter::WriteResult> DataWriter::Metadata() {
+  return NotImplemented("DataWriter::Metadata not yet implemented");
+}
+
+//=============================================================================
+// PositionDeleteWriter::Impl
+//=============================================================================
+
+class PositionDeleteWriter::Impl {
+ public:
+  explicit Impl(const PositionDeleteWriterOptions& options) : options_(options) {}
+
+  PositionDeleteWriterOptions options_;
+
+  // TODO: Add the following when implementing:
+  // - FileAppender for position delete format (file path + position columns)
+  // - CharSequenceSet or equivalent for tracking referenced data files
+  // - Metrics collection with field stripping for multi-file deletes
+  // - Split offsets tracking
+  // - Encryption key metadata handling
+};
+
+//=============================================================================
+// PositionDeleteWriter
+//=============================================================================
+
+Result<std::unique_ptr<PositionDeleteWriter>> PositionDeleteWriter::Make(
+    const PositionDeleteWriterOptions& options) {
+  return NotImplemented("PositionDeleteWriter implementation not yet available");
+}
+
+PositionDeleteWriter::PositionDeleteWriter(std::unique_ptr<Impl> impl)
+    : impl_(std::move(impl)) {}
+
+PositionDeleteWriter::~PositionDeleteWriter() = default;
+
+Status PositionDeleteWriter::Write(ArrowArray* data) {
+  return NotImplemented("PositionDeleteWriter::Write not yet implemented");
+}
+
+Status PositionDeleteWriter::WriteDelete(std::string_view file_path, int64_t pos) {
+  return NotImplemented("PositionDeleteWriter::WriteDelete not yet implemented");
+}
+
+Result<int64_t> PositionDeleteWriter::Length() const {
+  return NotImplemented("PositionDeleteWriter::Length not yet implemented");
+}
+
+Status PositionDeleteWriter::Close() {
+  return NotImplemented("PositionDeleteWriter::Close not yet implemented");
+}
+
+Result<FileWriter::WriteResult> PositionDeleteWriter::Metadata() {
+  return NotImplemented("PositionDeleteWriter::Metadata not yet implemented");
+}
+
+//=============================================================================
+// EqualityDeleteWriter::Impl
+//=============================================================================
+
+class EqualityDeleteWriter::Impl {
+ public:
+  explicit Impl(const EqualityDeleteWriterOptions& options) : options_(options) {}
+
+  EqualityDeleteWriterOptions options_;
+
+  // TODO: Add the following when implementing:
+  // - FileAppender for equality delete format
+  // - Metrics collection
+  // - Split offsets tracking
+  // - Encryption key metadata handling
+};
+
+//=============================================================================
+// EqualityDeleteWriter
+//=============================================================================
+
+Result<std::unique_ptr<EqualityDeleteWriter>> EqualityDeleteWriter::Make(
+    const EqualityDeleteWriterOptions& options) {
+  return NotImplemented("EqualityDeleteWriter implementation not yet available");
+}
+
+EqualityDeleteWriter::EqualityDeleteWriter(std::unique_ptr<Impl> impl)
+    : impl_(std::move(impl)) {}
+
+EqualityDeleteWriter::~EqualityDeleteWriter() = default;
+
+Status EqualityDeleteWriter::Write(ArrowArray* data) {
+  return NotImplemented("EqualityDeleteWriter::Write not yet implemented");
+}
+
+Result<int64_t> EqualityDeleteWriter::Length() const {
+  return NotImplemented("EqualityDeleteWriter::Length not yet implemented");
+}
+
+Status EqualityDeleteWriter::Close() {
+  return NotImplemented("EqualityDeleteWriter::Close not yet implemented");
+}
+
+Result<FileWriter::WriteResult> EqualityDeleteWriter::Metadata() {
+  return NotImplemented("EqualityDeleteWriter::Metadata not yet implemented");
+}
+
+const std::vector<int32_t>& EqualityDeleteWriter::equality_field_ids() const {
+  return impl_->options_.equality_field_ids;
+}
+
+//=============================================================================
+// FileWriterFactory::Impl
+//=============================================================================
+
+class FileWriterFactory::Impl {
+ public:
+  Impl(std::shared_ptr<Schema> schema, std::shared_ptr<PartitionSpec> spec,
+       std::shared_ptr<FileIO> io, std::shared_ptr<WriterProperties> properties)
+      : schema_(std::move(schema)),
+        spec_(std::move(spec)),
+        io_(std::move(io)),
+        properties_(std::move(properties)) {}
+
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<PartitionSpec> spec_;
+  std::shared_ptr<FileIO> io_;
+  std::shared_ptr<WriterProperties> properties_;
+
+  std::shared_ptr<Schema> eq_delete_schema_;
+  std::vector<int32_t> equality_field_ids_;
+  std::shared_ptr<Schema> pos_delete_row_schema_;
+};
+
+//=============================================================================
+// FileWriterFactory
+//=============================================================================
+
+FileWriterFactory::FileWriterFactory(std::shared_ptr<Schema> schema,
+                                     std::shared_ptr<PartitionSpec> spec,
+                                     std::shared_ptr<FileIO> io,
+                                     std::shared_ptr<WriterProperties> properties)
+    : impl_(std::make_unique<Impl>(std::move(schema), std::move(spec), std::move(io),
+                                   std::move(properties))) {}
+
+FileWriterFactory::~FileWriterFactory() = default;
+
+void FileWriterFactory::SetEqualityDeleteConfig(std::shared_ptr<Schema> eq_delete_schema,
+                                                std::vector<int32_t> equality_field_ids) {
+  impl_->eq_delete_schema_ = std::move(eq_delete_schema);
+  impl_->equality_field_ids_ = std::move(equality_field_ids);
+}
+
+void FileWriterFactory::SetPositionDeleteRowSchema(
+    std::shared_ptr<Schema> pos_delete_row_schema) {
+  impl_->pos_delete_row_schema_ = std::move(pos_delete_row_schema);
+}
+
+Result<std::unique_ptr<DataWriter>> FileWriterFactory::NewDataWriter(
+    const std::string& path, FileFormatType format, const PartitionValues& partition,
+    std::optional<int32_t> sort_order_id) {
+  DataWriterOptions options;
+  options.path = path;
+  options.schema = impl_->schema_;
+  options.spec = impl_->spec_;
+  options.partition = partition;
+  options.format = format;
+  options.io = impl_->io_;
+  options.sort_order_id = sort_order_id;
+  options.properties = impl_->properties_;
+
+  return DataWriter::Make(options);
+}
+
+Result<std::unique_ptr<PositionDeleteWriter>> FileWriterFactory::NewPositionDeleteWriter(
+    const std::string& path, FileFormatType format, const PartitionValues& partition) {
+  PositionDeleteWriterOptions options;
+  options.path = path;
+  options.schema = impl_->schema_;
+  options.spec = impl_->spec_;
+  options.partition = partition;
+  options.format = format;
+  options.io = impl_->io_;
+  options.row_schema = impl_->pos_delete_row_schema_;
+  options.properties = impl_->properties_;
+
+  return PositionDeleteWriter::Make(options);
+}
+
+Result<std::unique_ptr<EqualityDeleteWriter>> FileWriterFactory::NewEqualityDeleteWriter(
+    const std::string& path, FileFormatType format, const PartitionValues& partition,
+    std::optional<int32_t> sort_order_id) {
+  EqualityDeleteWriterOptions options;
+  options.path = path;
+  options.schema = impl_->eq_delete_schema_ ? impl_->eq_delete_schema_ : impl_->schema_;
+  options.spec = impl_->spec_;
+  options.partition = partition;
+  options.format = format;
+  options.io = impl_->io_;
+  options.equality_field_ids = impl_->equality_field_ids_;
+  options.sort_order_id = sort_order_id;
+  options.properties = impl_->properties_;
+
+  return EqualityDeleteWriter::Make(options);
+}
+
 }  // namespace iceberg