Skip to content

Commit ef13433

Browse files
shangxinlishangxinli
committed
feat: add FileWriterFactory with separated writer files
Reorganize writer files and add comprehensive improvements per review feedback. File Organization: - Separate PositionDeleteWriter into its own .h/.cc files - Separate EqualityDeleteWriter into its own .h/.cc files - Keep DataWriter in writer.h/.cc as the main writer - Use internal factory functions to handle cross-file construction High Priority Improvements: - Fix parameter passing: Use pass-by-value + std::move for sink parameters - Add input validation: Validate schema, spec, path, and equality_field_ids - Add thread safety documentation: Document that factory and writers are NOT thread-safe Medium Priority Improvements: - Add state management to stubs: Track is_closed state, validate operations - Improve WriteResult documentation: Document when multiple files are produced Implementation Details: - FileWriterFactory directly creates writers (true factory pattern) - Writers use friend pattern - only factory can construct them - Internal MakeXxxInternal functions handle cross-file construction - Stub implementations validate inputs and state before returning NotImplemented - All 23 tests passing - All pre-commit checks passing (clang-format, cmake-format) Related to #441
1 parent cf0fd37 commit ef13433

8 files changed

Lines changed: 857 additions & 4 deletions

File tree

‎src/iceberg/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@ set(ICEBERG_INCLUDES "$<BUILD_INTERFACE:${PROJECT_BINARY_DIR}/src>"
2020
set(ICEBERG_SOURCES
2121
arrow_c_data_guard_internal.cc
2222
catalog/memory/in_memory_catalog.cc
23+
data/equality_delete_writer.cc
24+
data/position_delete_writer.cc
2325
data/writer.cc
2426
delete_file_index.cc
2527
expression/aggregate.cc

‎src/iceberg/data/equality_delete_writer.cc‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#include "iceberg/data/equality_delete_writer.h"
21+
22+
namespace iceberg {
23+
24+
//=============================================================================
25+
// EqualityDeleteWriter - stub implementation (to be completed in separate PR per #441)
26+
//=============================================================================
27+
28+
class EqualityDeleteWriter::Impl {
29+
public:
30+
explicit Impl(const EqualityDeleteWriterOptions& options) : options_(options) {}
31+
EqualityDeleteWriterOptions options_;
32+
bool is_closed_ = false;
33+
};
34+
35+
EqualityDeleteWriter::EqualityDeleteWriter(std::unique_ptr<Impl> impl)
36+
: impl_(std::move(impl)) {}
37+
EqualityDeleteWriter::~EqualityDeleteWriter() = default;
38+
39+
Status EqualityDeleteWriter::Write(ArrowArray* data) {
40+
if (!data) {
41+
return InvalidArgument("Cannot write null data");
42+
}
43+
if (impl_->is_closed_) {
44+
return Invalid("Writer is already closed");
45+
}
46+
return NotImplemented("EqualityDeleteWriter not yet implemented - see #441");
47+
}
48+
49+
Result<int64_t> EqualityDeleteWriter::Length() const {
50+
return NotImplemented("EqualityDeleteWriter not yet implemented - see #441");
51+
}
52+
53+
Status EqualityDeleteWriter::Close() {
54+
if (impl_->is_closed_) {
55+
return {}; // Close is idempotent
56+
}
57+
impl_->is_closed_ = true;
58+
return NotImplemented("EqualityDeleteWriter not yet implemented - see #441");
59+
}
60+
61+
Result<FileWriter::WriteResult> EqualityDeleteWriter::Metadata() {
62+
if (!impl_->is_closed_) {
63+
return Invalid("Writer must be closed before getting metadata");
64+
}
65+
return NotImplemented("EqualityDeleteWriter not yet implemented - see #441");
66+
}
67+
68+
const std::vector<int32_t>& EqualityDeleteWriter::equality_field_ids() const {
69+
return impl_->options_.equality_field_ids;
70+
}
71+
72+
// Internal factory function for FileWriterFactory
73+
std::unique_ptr<EqualityDeleteWriter> MakeEqualityDeleteWriterInternal(
74+
const EqualityDeleteWriterOptions& options) {
75+
auto impl = std::make_unique<EqualityDeleteWriter::Impl>(options);
76+
return std::unique_ptr<EqualityDeleteWriter>(new EqualityDeleteWriter(std::move(impl)));
77+
}
78+
79+
} // namespace iceberg

‎src/iceberg/data/equality_delete_writer.h‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#pragma once
21+
22+
/// \file iceberg/data/equality_delete_writer.h
23+
/// Equality delete writer for Iceberg tables.
24+
25+
#include <cstdint>
26+
#include <memory>
27+
#include <optional>
28+
#include <string>
29+
#include <vector>
30+
31+
#include "iceberg/arrow_c_data.h"
32+
#include "iceberg/data/writer.h"
33+
#include "iceberg/file_format.h"
34+
#include "iceberg/iceberg_export.h"
35+
#include "iceberg/result.h"
36+
#include "iceberg/row/partition_values.h"
37+
#include "iceberg/type_fwd.h"
38+
39+
namespace iceberg {
40+
41+
/// \brief Options for creating an EqualityDeleteWriter.
42+
///
43+
/// \note The following features from Java EqualityDeleteWriter are not yet supported:
44+
/// - Encryption key metadata
45+
/// - Metrics collection and reporting
46+
/// - Split offsets tracking
47+
struct ICEBERG_EXPORT EqualityDeleteWriterOptions {
48+
std::string path;
49+
std::shared_ptr<Schema> schema;
50+
std::shared_ptr<PartitionSpec> spec;
51+
PartitionValues partition;
52+
FileFormatType format = FileFormatType::kParquet;
53+
std::shared_ptr<FileIO> io;
54+
std::vector<int32_t> equality_field_ids;
55+
std::optional<int32_t> sort_order_id;
56+
std::shared_ptr<class WriterProperties> properties;
57+
};
58+
59+
/// \brief Writer for Iceberg equality delete files.
60+
///
61+
/// \warning Thread Safety: Writer instances are NOT thread-safe. Each writer should only
62+
/// be used by a single thread. Do not call Write(), Close(), or Metadata() concurrently.
63+
class ICEBERG_EXPORT EqualityDeleteWriter : public FileWriter {
64+
public:
65+
~EqualityDeleteWriter() override;
66+
67+
Status Write(ArrowArray* data) override;
68+
Result<int64_t> Length() const override;
69+
Status Close() override;
70+
Result<WriteResult> Metadata() override;
71+
72+
const std::vector<int32_t>& equality_field_ids() const;
73+
74+
private:
75+
friend class FileWriterFactory;
76+
friend std::unique_ptr<EqualityDeleteWriter> MakeEqualityDeleteWriterInternal(
77+
const EqualityDeleteWriterOptions&);
78+
class Impl;
79+
std::unique_ptr<Impl> impl_;
80+
explicit EqualityDeleteWriter(std::unique_ptr<Impl> impl);
81+
};
82+
83+
} // namespace iceberg

‎src/iceberg/data/position_delete_writer.cc‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#include "iceberg/data/position_delete_writer.h"
21+
22+
namespace iceberg {
23+
24+
//=============================================================================
25+
// PositionDeleteWriter - stub implementation (to be completed in separate PR per #441)
26+
//=============================================================================
27+
28+
class PositionDeleteWriter::Impl {
29+
public:
30+
explicit Impl(const PositionDeleteWriterOptions& options) : options_(options) {}
31+
PositionDeleteWriterOptions options_;
32+
bool is_closed_ = false;
33+
};
34+
35+
PositionDeleteWriter::PositionDeleteWriter(std::unique_ptr<Impl> impl)
36+
: impl_(std::move(impl)) {}
37+
PositionDeleteWriter::~PositionDeleteWriter() = default;
38+
39+
Status PositionDeleteWriter::Write(ArrowArray* data) {
40+
if (!data) {
41+
return InvalidArgument("Cannot write null data");
42+
}
43+
if (impl_->is_closed_) {
44+
return Invalid("Writer is already closed");
45+
}
46+
return NotImplemented("PositionDeleteWriter not yet implemented - see #441");
47+
}
48+
49+
Status PositionDeleteWriter::WriteDelete(std::string_view file_path, int64_t pos) {
50+
if (file_path.empty()) {
51+
return InvalidArgument("File path cannot be empty");
52+
}
53+
if (impl_->is_closed_) {
54+
return Invalid("Writer is already closed");
55+
}
56+
return NotImplemented("PositionDeleteWriter not yet implemented - see #441");
57+
}
58+
59+
Result<int64_t> PositionDeleteWriter::Length() const {
60+
return NotImplemented("PositionDeleteWriter not yet implemented - see #441");
61+
}
62+
63+
Status PositionDeleteWriter::Close() {
64+
if (impl_->is_closed_) {
65+
return {}; // Close is idempotent
66+
}
67+
impl_->is_closed_ = true;
68+
return NotImplemented("PositionDeleteWriter not yet implemented - see #441");
69+
}
70+
71+
Result<FileWriter::WriteResult> PositionDeleteWriter::Metadata() {
72+
if (!impl_->is_closed_) {
73+
return Invalid("Writer must be closed before getting metadata");
74+
}
75+
return NotImplemented("PositionDeleteWriter not yet implemented - see #441");
76+
}
77+
78+
// Internal factory function for FileWriterFactory
79+
std::unique_ptr<PositionDeleteWriter> MakePositionDeleteWriterInternal(
80+
const PositionDeleteWriterOptions& options) {
81+
auto impl = std::make_unique<PositionDeleteWriter::Impl>(options);
82+
return std::unique_ptr<PositionDeleteWriter>(new PositionDeleteWriter(std::move(impl)));
83+
}
84+
85+
} // namespace iceberg

‎src/iceberg/data/position_delete_writer.h‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,81 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
#pragma once
21+
22+
/// \file iceberg/data/position_delete_writer.h
23+
/// Position delete writer for Iceberg tables.
24+
25+
#include <cstdint>
26+
#include <memory>
27+
#include <string>
28+
#include <string_view>
29+
30+
#include "iceberg/arrow_c_data.h"
31+
#include "iceberg/data/writer.h"
32+
#include "iceberg/file_format.h"
33+
#include "iceberg/iceberg_export.h"
34+
#include "iceberg/result.h"
35+
#include "iceberg/row/partition_values.h"
36+
#include "iceberg/type_fwd.h"
37+
38+
namespace iceberg {
39+
40+
/// \brief Options for creating a PositionDeleteWriter.
41+
///
42+
/// \note The following features from Java PositionDeleteWriter are not yet supported:
43+
/// - Encryption key metadata
44+
/// - Referenced data files tracking (CharSequenceSet referencedDataFiles)
45+
/// - Metrics stripping for multi-file deletes
46+
/// - Split offsets tracking
47+
struct ICEBERG_EXPORT PositionDeleteWriterOptions {
48+
std::string path;
49+
std::shared_ptr<Schema> schema;
50+
std::shared_ptr<PartitionSpec> spec;
51+
PartitionValues partition;
52+
FileFormatType format = FileFormatType::kParquet;
53+
std::shared_ptr<FileIO> io;
54+
std::shared_ptr<Schema> row_schema; // Optional row data schema
55+
std::shared_ptr<class WriterProperties> properties;
56+
};
57+
58+
/// \brief Writer for Iceberg position delete files.
59+
///
60+
/// \warning Thread Safety: Writer instances are NOT thread-safe. Each writer should only
61+
/// be used by a single thread. Do not call Write(), Close(), or Metadata() concurrently.
62+
class ICEBERG_EXPORT PositionDeleteWriter : public FileWriter {
63+
public:
64+
~PositionDeleteWriter() override;
65+
66+
Status Write(ArrowArray* data) override;
67+
Status WriteDelete(std::string_view file_path, int64_t pos);
68+
Result<int64_t> Length() const override;
69+
Status Close() override;
70+
Result<WriteResult> Metadata() override;
71+
72+
private:
73+
friend class FileWriterFactory;
74+
friend std::unique_ptr<PositionDeleteWriter> MakePositionDeleteWriterInternal(
75+
const PositionDeleteWriterOptions&);
76+
class Impl;
77+
std::unique_ptr<Impl> impl_;
78+
explicit PositionDeleteWriter(std::unique_ptr<Impl> impl);
79+
};
80+
81+
} // namespace iceberg

0 commit comments

Comments
 (0)