This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a948c40 feat: add file writer infrastructure and data/compact
increment models (#81)
a948c40 is described below
commit a948c40095f5a2b084b2854956479138a7c30872
Author: Yonghao Fang <[email protected]>
AuthorDate: Tue Jun 16 09:11:54 2026 +0800
feat: add file writer infrastructure and data/compact increment models (#81)
---
src/paimon/core/io/compact_increment.h | 156 +++++++++++++
src/paimon/core/io/compact_increment_test.cpp | 132 +++++++++++
src/paimon/core/io/data_file_writer.cpp | 83 +++++++
src/paimon/core/io/data_file_writer.h | 70 ++++++
src/paimon/core/io/data_increment.h | 158 ++++++++++++++
src/paimon/core/io/data_increment_test.cpp | 107 +++++++++
src/paimon/core/io/file_writer.h | 63 ++++++
src/paimon/core/io/key_value_data_file_writer.cpp | 203 +++++++++++++++++
src/paimon/core/io/key_value_data_file_writer.h | 89 ++++++++
src/paimon/core/io/rolling_blob_file_writer.cpp | 198 +++++++++++++++++
src/paimon/core/io/rolling_blob_file_writer.h | 94 ++++++++
.../core/io/rolling_blob_file_writer_test.cpp | 94 ++++++++
src/paimon/core/io/rolling_file_writer.h | 199 +++++++++++++++++
src/paimon/core/io/single_file_writer.h | 241 +++++++++++++++++++++
src/paimon/core/io/single_file_writer_test.cpp | 109 ++++++++++
15 files changed, 1996 insertions(+)
diff --git a/src/paimon/core/io/compact_increment.h
b/src/paimon/core/io/compact_increment.h
new file mode 100644
index 0000000..8304d7a
--- /dev/null
+++ b/src/paimon/core/io/compact_increment.h
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <iterator>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/core/index/index_file_meta.h"
+#include "paimon/core/io/data_file_meta.h"
+
+namespace paimon {
+// Files changed before and after compaction, with changelog produced during
compaction.
+class CompactIncrement {
+ public:
+ CompactIncrement(std::vector<std::shared_ptr<DataFileMeta>>&&
compact_before,
+ std::vector<std::shared_ptr<DataFileMeta>>&&
compact_after,
+ std::vector<std::shared_ptr<DataFileMeta>>&&
changelog_files)
+ : CompactIncrement(std::move(compact_before), std::move(compact_after),
+ std::move(changelog_files), {}, {}) {}
+
+ CompactIncrement(std::vector<std::shared_ptr<DataFileMeta>>&&
compact_before,
+ std::vector<std::shared_ptr<DataFileMeta>>&&
compact_after,
+ std::vector<std::shared_ptr<DataFileMeta>>&&
changelog_files,
+ std::vector<std::shared_ptr<IndexFileMeta>>&&
new_index_files,
+ std::vector<std::shared_ptr<IndexFileMeta>>&&
deleted_index_files)
+ : compact_before_(std::move(compact_before)),
+ compact_after_(std::move(compact_after)),
+ changelog_files_(std::move(changelog_files)),
+ new_index_files_(std::move(new_index_files)),
+ deleted_index_files_(std::move(deleted_index_files)) {}
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& CompactBefore() const {
+ return compact_before_;
+ }
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& CompactAfter() const {
+ return compact_after_;
+ }
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& ChangelogFiles() const {
+ return changelog_files_;
+ }
+
+ const std::vector<std::shared_ptr<IndexFileMeta>>& NewIndexFiles() const {
+ return new_index_files_;
+ }
+
+ const std::vector<std::shared_ptr<IndexFileMeta>>& DeletedIndexFiles()
const {
+ return deleted_index_files_;
+ }
+
+ void AddNewIndexFiles(std::vector<std::shared_ptr<IndexFileMeta>>&&
new_index_files) {
+ new_index_files_.insert(new_index_files_.end(),
+
std::make_move_iterator(new_index_files.begin()),
+
std::make_move_iterator(new_index_files.end()));
+ }
+
+ void AddDeletedIndexFiles(std::vector<std::shared_ptr<IndexFileMeta>>&&
deleted_index_files) {
+ deleted_index_files_.insert(deleted_index_files_.end(),
+
std::make_move_iterator(deleted_index_files.begin()),
+
std::make_move_iterator(deleted_index_files.end()));
+ }
+
+ bool IsEmpty() const {
+ return compact_before_.empty() && compact_after_.empty() &&
changelog_files_.empty() &&
+ new_index_files_.empty() && deleted_index_files_.empty();
+ }
+
+ bool operator==(const CompactIncrement& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return ObjectUtils::Equal(compact_before_, other.compact_before_) &&
+ ObjectUtils::Equal(compact_after_, other.compact_after_) &&
+ ObjectUtils::Equal(changelog_files_, other.changelog_files_) &&
+ ObjectUtils::Equal(new_index_files_, other.new_index_files_) &&
+ ObjectUtils::Equal(deleted_index_files_,
other.deleted_index_files_);
+ }
+
+ bool TEST_Equal(const CompactIncrement& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return ObjectUtils::TEST_Equal(compact_before_, other.compact_before_)
&&
+ ObjectUtils::TEST_Equal(compact_after_, other.compact_after_) &&
+ ObjectUtils::TEST_Equal(changelog_files_,
other.changelog_files_) &&
+ ObjectUtils::TEST_Equal(new_index_files_,
other.new_index_files_) &&
+ ObjectUtils::TEST_Equal(deleted_index_files_,
other.deleted_index_files_);
+ }
+
+ std::string ToString() const {
+ std::vector<std::string> compact_before_names;
+ compact_before_names.reserve(compact_before_.size());
+ for (const auto& file : compact_before_) {
+ compact_before_names.emplace_back(file->file_name);
+ }
+ std::vector<std::string> compact_after_names;
+ compact_after_names.reserve(compact_after_.size());
+ for (const auto& file : compact_after_) {
+ compact_after_names.emplace_back(file->file_name);
+ }
+ std::vector<std::string> changelog_files_names;
+ changelog_files_names.reserve(changelog_files_.size());
+ for (const auto& file : changelog_files_) {
+ changelog_files_names.emplace_back(file->file_name);
+ }
+ std::vector<std::string> new_index_names;
+ new_index_names.reserve(new_index_files_.size());
+ for (const auto& new_index_file : new_index_files_) {
+ new_index_names.emplace_back(new_index_file->FileName());
+ }
+ std::vector<std::string> deleted_index_names;
+ deleted_index_names.reserve(deleted_index_files_.size());
+ for (const auto& deleted_index_file : deleted_index_files_) {
+ deleted_index_names.emplace_back(deleted_index_file->FileName());
+ }
+
+ return fmt::format(
+ "CompactIncrement {{compactBefore = {}, compactAfter = {},
changelogFiles = {}, "
+ "newIndexFiles = {}, deletedIndexFiles = {}}}",
+ fmt::join(compact_before_names, ", "),
fmt::join(compact_after_names, ", "),
+ fmt::join(changelog_files_names, ", "), fmt::join(new_index_names,
", "),
+ fmt::join(deleted_index_names, ", "));
+ }
+
+ private:
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before_;
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after_;
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files_;
+ std::vector<std::shared_ptr<IndexFileMeta>> new_index_files_;
+ std::vector<std::shared_ptr<IndexFileMeta>> deleted_index_files_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/compact_increment_test.cpp
b/src/paimon/core/io/compact_increment_test.cpp
new file mode 100644
index 0000000..8deae09
--- /dev/null
+++ b/src/paimon/core/io/compact_increment_test.cpp
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/compact_increment.h"
+
+#include <optional>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/result.h"
+
+namespace paimon::test {
+
+class CompactIncrementTest : public ::testing::Test {
+ public:
+ std::shared_ptr<DataFileMeta> CreateDataFileMeta(const std::string&
file_name) {
+ return DataFileMeta::ForAppend(file_name, 100, 100,
SimpleStats::EmptyStats(), 0, 100, 0,
+ FileSource::Append(), std::nullopt,
std::nullopt,
+ std::nullopt, std::nullopt)
+ .value();
+ }
+};
+
+TEST_F(CompactIncrementTest, TestCompactBefore) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before = {file1, file2};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after = {file3};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files = {file4};
+
+ CompactIncrement increment(std::move(compact_before),
std::move(compact_after),
+ std::move(changelog_files));
+
+ ASSERT_TRUE(ObjectUtils::Equal(increment.CompactBefore(), {file1, file2}));
+}
+
+TEST_F(CompactIncrementTest, TestCompactAfter) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before = {file1};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after = {file2, file3};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files = {file4};
+
+ CompactIncrement increment(std::move(compact_before),
std::move(compact_after),
+ std::move(changelog_files));
+ ASSERT_TRUE(ObjectUtils::Equal(increment.CompactAfter(), {file2, file3}));
+}
+
+TEST_F(CompactIncrementTest, TestChangelogFiles) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before = {file1};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after = {file2};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files = {file3,
file4};
+
+ CompactIncrement increment(std::move(compact_before),
std::move(compact_after),
+ std::move(changelog_files));
+ ASSERT_TRUE(ObjectUtils::Equal(increment.ChangelogFiles(), {file3,
file4}));
+}
+
+TEST_F(CompactIncrementTest, TestIsEmpty) {
+ CompactIncrement increment({}, {}, {});
+ ASSERT_TRUE(increment.IsEmpty());
+}
+
+TEST_F(CompactIncrementTest, TestEqualityOperator) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before1 = {file1};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after1 = {file2};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files1 = {file3};
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before2 = {file1};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after2 = {file2};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files2 = {file3};
+
+ CompactIncrement increment1(std::move(compact_before1),
std::move(compact_after1),
+ std::move(changelog_files1));
+ CompactIncrement increment2(std::move(compact_before2),
std::move(compact_after2),
+ std::move(changelog_files2));
+ ASSERT_EQ(increment1, increment2);
+}
+
+TEST_F(CompactIncrementTest, TestToString) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ std::vector<std::shared_ptr<DataFileMeta>> compact_before = {file1, file2};
+ std::vector<std::shared_ptr<DataFileMeta>> compact_after = {file3};
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files = {file4};
+
+ CompactIncrement increment(std::move(compact_before),
std::move(compact_after),
+ std::move(changelog_files));
+ std::string expected =
+ "CompactIncrement {compactBefore = file1, file2, compactAfter = file3,
changelogFiles = "
+ "file4, newIndexFiles = , deletedIndexFiles = }";
+ ASSERT_EQ(increment.ToString(), expected);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/data_file_writer.cpp
b/src/paimon/core/io/data_file_writer.cpp
new file mode 100644
index 0000000..6f8bb3b
--- /dev/null
+++ b/src/paimon/core/io/data_file_writer.cpp
@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/data_file_writer.h"
+
+#include <cassert>
+
+#include "arrow/c/abi.h"
+#include "paimon/common/utils/long_counter.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/core/stats/simple_stats_converter.h"
+#include "paimon/format/format_stats_extractor.h"
+
+namespace paimon {
+class MemoryPool;
+
+DataFileWriter::DataFileWriter(
+ const std::string& compression, std::function<Status(::ArrowArray*,
::ArrowArray*)> converter,
+ int64_t schema_id, const std::shared_ptr<LongCounter>& seq_num_counter,
FileSource file_source,
+ const std::shared_ptr<FormatStatsExtractor>& stats_extractor, bool
is_external_path,
+ const std::optional<std::vector<std::string>>& write_cols,
+ const std::shared_ptr<MemoryPool>& pool)
+ : SingleFileWriter(compression, converter),
+ pool_(pool),
+ schema_id_(schema_id),
+ is_external_path_(is_external_path),
+ seq_num_counter_(seq_num_counter),
+ file_source_(file_source),
+ stats_extractor_(stats_extractor),
+ write_cols_(write_cols) {}
+
+Status DataFileWriter::Write(ArrowArray* batch) {
+ int64_t record_count = batch->length;
+ PAIMON_RETURN_NOT_OK(SingleFileWriter::Write(batch));
+ seq_num_counter_->Add(record_count);
+ return Status::OK();
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileWriter::GetResult() {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<ColumnStats>>
field_stats, GetFieldStats());
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats stats,
+ SimpleStatsConverter::ToBinary(field_stats,
pool_.get()));
+ // TODO(xinyu.lxy): do not support write value stats cols for now
+ std::optional<std::string> final_path;
+ if (is_external_path_) {
+ PAIMON_ASSIGN_OR_RAISE(Path external_path, PathUtil::ToPath(path_));
+ final_path = external_path.ToString();
+ }
+ return DataFileMeta::ForAppend(
+ PathUtil::GetName(path_), output_bytes_, RecordCount(), stats,
+ seq_num_counter_->GetValue() - RecordCount(),
seq_num_counter_->GetValue() - 1, schema_id_,
+ {}, /*embedded_index=*/nullptr, file_source_,
/*value_stats_cols=*/std::nullopt, final_path,
+ /*first_row_id=*/std::nullopt, write_cols_);
+}
+
+Result<std::vector<std::shared_ptr<ColumnStats>>>
DataFileWriter::GetFieldStats() {
+ if (!closed_) {
+ return Status::Invalid("Cannot access metric unless the writer is
closed.");
+ }
+ if (stats_extractor_ == nullptr) {
+ assert(false);
+ return Status::Invalid("simple stats extractor is null pointer.");
+ }
+ return stats_extractor_->Extract(fs_, path_, pool_);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_writer.h
b/src/paimon/core/io/data_file_writer.h
new file mode 100644
index 0000000..097c9b9
--- /dev/null
+++ b/src/paimon/core/io/data_file_writer.h
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <optional>
+#include <string>
+#include <vector>
+
+#include "arrow/c/abi.h"
+#include "paimon/common/utils/long_counter.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class ColumnStats;
+class FormatStatsExtractor;
+class LongCounter;
+class MemoryPool;
+
+class DataFileWriter : public SingleFileWriter<::ArrowArray*,
std::shared_ptr<DataFileMeta>> {
+ public:
+ DataFileWriter(const std::string& compression,
+ std::function<Status(::ArrowArray*, ::ArrowArray*)>
converter, int64_t schema_id,
+ const std::shared_ptr<LongCounter>& seq_num_counter,
FileSource file_source,
+ const std::shared_ptr<FormatStatsExtractor>&
stats_extractor,
+ bool is_external_path, const
std::optional<std::vector<std::string>>& write_cols,
+ const std::shared_ptr<MemoryPool>& pool);
+
+ Status Write(::ArrowArray* batch) override;
+
+ Result<std::shared_ptr<DataFileMeta>> GetResult() override;
+
+ private:
+ Result<std::vector<std::shared_ptr<ColumnStats>>> GetFieldStats();
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ int64_t schema_id_;
+ bool is_external_path_;
+
+ std::shared_ptr<LongCounter> seq_num_counter_;
+ FileSource file_source_;
+ std::shared_ptr<FormatStatsExtractor> stats_extractor_;
+ std::optional<std::vector<std::string>> write_cols_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_increment.h
b/src/paimon/core/io/data_increment.h
new file mode 100644
index 0000000..602eb37
--- /dev/null
+++ b/src/paimon/core/io/data_increment.h
@@ -0,0 +1,158 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <iterator>
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/utils/object_utils.h"
+#include "paimon/core/index/index_file_meta.h"
+#include "paimon/core/io/data_file_meta.h"
+
+namespace paimon {
+// Increment of data files, changelog files and index files.
+class DataIncrement {
+ public:
+ explicit DataIncrement(std::vector<std::shared_ptr<IndexFileMeta>>&&
new_index_files)
+ : DataIncrement({}, {}, {}, std::move(new_index_files), {}) {}
+
+ DataIncrement(std::vector<std::shared_ptr<DataFileMeta>>&& new_files,
+ std::vector<std::shared_ptr<DataFileMeta>>&& deleted_files,
+ std::vector<std::shared_ptr<DataFileMeta>>&& changelog_files)
+ : DataIncrement(std::move(new_files), std::move(deleted_files),
std::move(changelog_files),
+ {}, {}) {}
+
+ DataIncrement(std::vector<std::shared_ptr<DataFileMeta>>&& new_files,
+ std::vector<std::shared_ptr<DataFileMeta>>&& deleted_files,
+ std::vector<std::shared_ptr<DataFileMeta>>&& changelog_files,
+ std::vector<std::shared_ptr<IndexFileMeta>>&&
new_index_files,
+ std::vector<std::shared_ptr<IndexFileMeta>>&&
deleted_index_files)
+ : new_files_(std::move(new_files)),
+ deleted_files_(std::move(deleted_files)),
+ changelog_files_(std::move(changelog_files)),
+ new_index_files_(std::move(new_index_files)),
+ deleted_index_files_(std::move(deleted_index_files)) {}
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& NewFiles() const {
+ return new_files_;
+ }
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& DeletedFiles() const {
+ return deleted_files_;
+ }
+
+ const std::vector<std::shared_ptr<DataFileMeta>>& ChangelogFiles() const {
+ return changelog_files_;
+ }
+
+ const std::vector<std::shared_ptr<IndexFileMeta>>& NewIndexFiles() const {
+ return new_index_files_;
+ }
+
+ const std::vector<std::shared_ptr<IndexFileMeta>>& DeletedIndexFiles()
const {
+ return deleted_index_files_;
+ }
+
+ void AddNewIndexFiles(std::vector<std::shared_ptr<IndexFileMeta>>&&
new_index_files) {
+ new_index_files_.insert(new_index_files_.end(),
+
std::make_move_iterator(new_index_files.begin()),
+
std::make_move_iterator(new_index_files.end()));
+ }
+
+ void AddDeletedIndexFiles(std::vector<std::shared_ptr<IndexFileMeta>>&&
deleted_index_files) {
+ deleted_index_files_.insert(deleted_index_files_.end(),
+
std::make_move_iterator(deleted_index_files.begin()),
+
std::make_move_iterator(deleted_index_files.end()));
+ }
+
+ bool IsEmpty() const {
+ return new_files_.empty() && deleted_files_.empty() &&
changelog_files_.empty() &&
+ new_index_files_.empty() && deleted_index_files_.empty();
+ }
+
+ bool operator==(const DataIncrement& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return ObjectUtils::Equal(new_files_, other.new_files_) &&
+ ObjectUtils::Equal(deleted_files_, other.deleted_files_) &&
+ ObjectUtils::Equal(changelog_files_, other.changelog_files_) &&
+ ObjectUtils::Equal(new_index_files_, other.new_index_files_) &&
+ ObjectUtils::Equal(deleted_index_files_,
other.deleted_index_files_);
+ }
+
+ bool TEST_Equal(const DataIncrement& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return ObjectUtils::TEST_Equal(new_files_, other.new_files_) &&
+ ObjectUtils::TEST_Equal(deleted_files_, other.deleted_files_) &&
+ ObjectUtils::TEST_Equal(changelog_files_,
other.changelog_files_) &&
+ ObjectUtils::TEST_Equal(new_index_files_,
other.new_index_files_) &&
+ ObjectUtils::TEST_Equal(deleted_index_files_,
other.deleted_index_files_);
+ }
+
+ std::string ToString() const {
+ std::vector<std::string> new_files_names;
+ new_files_names.reserve(new_files_.size());
+ for (const auto& new_file : new_files_) {
+ new_files_names.emplace_back(new_file->file_name);
+ }
+ std::vector<std::string> deleted_files_names;
+ deleted_files_names.reserve(deleted_files_.size());
+ for (const auto& deleted_file : deleted_files_) {
+ deleted_files_names.emplace_back(deleted_file->file_name);
+ }
+ std::vector<std::string> changelog_files_names;
+ changelog_files_names.reserve(changelog_files_.size());
+ for (const auto& changelog_file : changelog_files_) {
+ changelog_files_names.emplace_back(changelog_file->file_name);
+ }
+ std::vector<std::string> new_index_names;
+ new_index_names.reserve(new_index_files_.size());
+ for (const auto& new_index_file : new_index_files_) {
+ new_index_names.emplace_back(new_index_file->FileName());
+ }
+ std::vector<std::string> deleted_index_names;
+ deleted_index_names.reserve(deleted_index_files_.size());
+ for (const auto& deleted_index_file : deleted_index_files_) {
+ deleted_index_names.emplace_back(deleted_index_file->FileName());
+ }
+
+ return fmt::format(
+ "DataIncrement {{newFiles = {}, deletedFiles = {}, changelogFiles
= {}, newIndexFiles "
+ "= {}, deletedIndexFiles = {}}}",
+ fmt::join(new_files_names, ", "), fmt::join(deleted_files_names,
", "),
+ fmt::join(changelog_files_names, ", "), fmt::join(new_index_names,
", "),
+ fmt::join(deleted_index_names, ", "));
+ }
+
+ private:
+ std::vector<std::shared_ptr<DataFileMeta>> new_files_;
+ std::vector<std::shared_ptr<DataFileMeta>> deleted_files_;
+ std::vector<std::shared_ptr<DataFileMeta>> changelog_files_;
+ std::vector<std::shared_ptr<IndexFileMeta>> new_index_files_;
+ std::vector<std::shared_ptr<IndexFileMeta>> deleted_index_files_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_increment_test.cpp
b/src/paimon/core/io/data_increment_test.cpp
new file mode 100644
index 0000000..21c8ea8
--- /dev/null
+++ b/src/paimon/core/io/data_increment_test.cpp
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/data_increment.h"
+
+#include <optional>
+
+#include "gtest/gtest.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/result.h"
+
+namespace paimon::test {
+
+class DataIncrementTest : public ::testing::Test {
+ public:
+ std::shared_ptr<DataFileMeta> CreateDataFileMeta(const std::string&
file_name) {
+ return DataFileMeta::ForAppend(file_name, 100, 100,
SimpleStats::EmptyStats(), 0, 100, 0,
+ FileSource::Append(), std::nullopt,
std::nullopt,
+ std::nullopt, std::nullopt)
+ .value();
+ }
+};
+
+TEST_F(DataIncrementTest, TestNewFiles) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+
+ DataIncrement increment({file1, file2}, {}, {});
+ ASSERT_EQ(increment.NewFiles().size(), 2);
+ ASSERT_EQ(increment.NewFiles()[0]->file_name, "file1");
+ ASSERT_EQ(increment.NewFiles()[1]->file_name, "file2");
+}
+
+TEST_F(DataIncrementTest, TestDeletedFiles) {
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+
+ DataIncrement increment({}, {file3}, {});
+ ASSERT_EQ(increment.DeletedFiles().size(), 1);
+ ASSERT_EQ(increment.DeletedFiles()[0]->file_name, "file3");
+}
+
+TEST_F(DataIncrementTest, TestChangelogFiles) {
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ DataIncrement increment({}, {}, {file4});
+ ASSERT_EQ(increment.ChangelogFiles().size(), 1);
+ ASSERT_EQ(increment.ChangelogFiles()[0]->file_name, "file4");
+}
+
+TEST_F(DataIncrementTest, TestIsEmpty) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ DataIncrement increment1({}, {}, {});
+ ASSERT_TRUE(increment1.IsEmpty());
+
+ DataIncrement increment2({file1}, {}, {});
+ ASSERT_FALSE(increment2.IsEmpty());
+
+ DataIncrement increment3({}, {}, {file4});
+ ASSERT_FALSE(increment3.IsEmpty());
+}
+
+TEST_F(DataIncrementTest, TestEqualityOperator) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ DataIncrement increment1({file1}, {file3}, {file4});
+ DataIncrement increment2({file1}, {file3}, {file4});
+ DataIncrement increment3({file2}, {file3}, {file4});
+
+ ASSERT_TRUE(increment1 == increment2);
+ ASSERT_FALSE(increment1 == increment3);
+}
+
+TEST_F(DataIncrementTest, TestToString) {
+ std::shared_ptr<DataFileMeta> file1 = CreateDataFileMeta("file1");
+ std::shared_ptr<DataFileMeta> file2 = CreateDataFileMeta("file2");
+ std::shared_ptr<DataFileMeta> file3 = CreateDataFileMeta("file3");
+ std::shared_ptr<DataFileMeta> file4 = CreateDataFileMeta("file4");
+
+ DataIncrement increment({file1, file2}, {file3}, {file4});
+ std::string expected =
+ "DataIncrement {newFiles = file1, file2, deletedFiles = file3,
changelogFiles = file4, "
+ "newIndexFiles = , deletedIndexFiles = }";
+ ASSERT_EQ(increment.ToString(), expected);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/file_writer.h b/src/paimon/core/io/file_writer.h
new file mode 100644
index 0000000..b03fbd2
--- /dev/null
+++ b/src/paimon/core/io/file_writer.h
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+
+#include "paimon/type_fwd.h"
+
+namespace paimon {
+
+/// File writer to accept one record or a branch of records and generate
metadata after closing it.
+/// <T> record type.
+/// <R> file result to collect.
+template <typename T, typename R>
+class FileWriter {
+ public:
+ FileWriter() = default;
+ virtual ~FileWriter() = default;
+
+ /// Add one record to this file writer.
+ ///
+ /// @note If any error occurs during writing, the writer should clean up
useless files for
+ /// the user.
+ ///
+ /// @param record to write.
+ /// @return Status if encounter any IO error.
+ virtual Status Write(T record) = 0;
+
+ /// The total written record count.
+ ///
+ /// @return record count.
+ virtual int64_t RecordCount() const = 0;
+
+ /// Abort to clear orphan file(s) if encounter any error.
+ ///
+ /// @note This implementation must be reentrant.
+ virtual void Abort() = 0;
+
+ virtual Status Close() = 0;
+
+ /// @return the result for this closed file writer.
+ virtual Result<R> GetResult() = 0;
+
+ virtual std::shared_ptr<Metrics> GetMetrics() const = 0;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/key_value_data_file_writer.cpp
b/src/paimon/core/io/key_value_data_file_writer.cpp
new file mode 100644
index 0000000..82bbba9
--- /dev/null
+++ b/src/paimon/core/io/key_value_data_file_writer.cpp
@@ -0,0 +1,203 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/key_value_data_file_writer.h"
+
+#include <algorithm>
+#include <cassert>
+#include <cstddef>
+#include <optional>
+#include <utility>
+#include <variant>
+
+#include "arrow/type.h"
+#include "fmt/format.h"
+#include "paimon/common/data/binary_array.h"
+#include "paimon/common/data/binary_array_writer.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/data_define.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/table/special_fields.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/core/stats/simple_stats_converter.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/format/format_stats_extractor.h"
+
+struct ArrowArray;
+
+namespace paimon {
+class MemoryPool;
+
+KeyValueDataFileWriter::KeyValueDataFileWriter(
+ const std::string& compression, std::function<Status(KeyValueBatch&&,
::ArrowArray*)> converter,
+ int64_t schema_id, int32_t level, FileSource file_source,
+ const std::vector<std::string>& primary_keys,
+ const std::shared_ptr<FormatStatsExtractor>& stats_extractor,
+ const std::shared_ptr<arrow::Schema>& write_schema, bool is_external_path,
+ const std::shared_ptr<MemoryPool>& pool)
+ : SingleFileWriter(compression, converter),
+ pool_(pool),
+ schema_id_(schema_id),
+ level_(level),
+ file_source_(file_source),
+ primary_keys_(primary_keys),
+ stats_extractor_(stats_extractor),
+ write_schema_(write_schema),
+ is_external_path_(is_external_path),
+ disable_stats_(stats_extractor == nullptr) {}
+
+Status KeyValueDataFileWriter::Write(KeyValueBatch batch) {
+ // update min and max key
+ if (!min_key_) {
+ min_key_ = batch.min_key;
+ }
+ max_key_ = batch.max_key;
+ // update min/max sequence number
+ min_sequence_number_ = std::min(min_sequence_number_,
batch.min_sequence_number);
+ max_sequence_number_ = std::max(max_sequence_number_,
batch.max_sequence_number);
+ // update delete row count
+ delete_row_count_ += batch.delete_row_count;
+ PAIMON_RETURN_NOT_OK(SingleFileWriter::Write(std::move(batch)));
+ return Status::OK();
+}
+
+Result<std::shared_ptr<DataFileMeta>> KeyValueDataFileWriter::GetResult() {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<ColumnStats>>
field_stats, GetFieldStats());
+ if (!disable_stats_ && field_stats.size() !=
static_cast<size_t>(write_schema_->num_fields())) {
+ return Status::Invalid("invalid field stats, mismatch with write
schema");
+ }
+ // min/max key
+ BinaryRow min_key(primary_keys_.size());
+ BinaryRow max_key(primary_keys_.size());
+ PAIMON_RETURN_NOT_OK(GenerateMinMaxKey(&min_key, &max_key));
+
+ // key value stats
+ SimpleStats key_stats = SimpleStats::EmptyStats();
+ SimpleStats value_stats = SimpleStats::EmptyStats();
+ if (!disable_stats_) {
+ PAIMON_RETURN_NOT_OK(GenerateKeyValueStats(field_stats, &key_stats,
&value_stats));
+ } else {
+ PAIMON_RETURN_NOT_OK(GenerateKeyStatsWithAllNull(&key_stats));
+ }
+ // TODO(xinyu.lxy): do not support write value stats cols for now
+ std::optional<std::string> final_path;
+ if (is_external_path_) {
+ PAIMON_ASSIGN_OR_RAISE(Path external_path, PathUtil::ToPath(path_));
+ final_path = external_path.ToString();
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t local_micro,
DateTimeUtils::GetCurrentLocalTimeUs());
+ return std::make_shared<DataFileMeta>(
+ PathUtil::GetName(path_), output_bytes_, RecordCount(), min_key,
max_key, key_stats,
+ value_stats, min_sequence_number_, max_sequence_number_, schema_id_,
level_,
+ /*extra_files=*/std::vector<std::optional<std::string>>(),
+ Timestamp(/*millisecond=*/local_micro / 1000,
/*nano_of_millisecond=*/0), delete_row_count_,
+ /*embedded_index=*/nullptr, file_source_,
+ /*value_stats_cols=*/std::nullopt, final_path,
/*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+}
+
+Status KeyValueDataFileWriter::GenerateMinMaxKey(BinaryRow* min_key,
BinaryRow* max_key) const {
+ BinaryRowWriter min_writer(min_key, /*initial_size=*/1024, pool_.get());
+ BinaryRowWriter max_writer(max_key, /*initial_size=*/1024, pool_.get());
+ min_writer.Reset();
+ max_writer.Reset();
+ for (size_t i = 0; i < primary_keys_.size(); ++i) {
+ auto data_type =
write_schema_->GetFieldByName(primary_keys_[i])->type();
+ InternalRow::FieldGetterFunc getter;
+ PAIMON_ASSIGN_OR_RAISE(getter,
+ InternalRow::CreateFieldGetter(i, data_type,
/*use_view=*/true));
+ BinaryRowWriter::FieldSetterFunc setter;
+ PAIMON_ASSIGN_OR_RAISE(setter, BinaryRowWriter::CreateFieldSetter(i,
data_type));
+ setter(getter(*min_key_), &min_writer);
+ setter(getter(*max_key_), &max_writer);
+ }
+ min_writer.Complete();
+ max_writer.Complete();
+ return Status::OK();
+}
+
+Status KeyValueDataFileWriter::GenerateKeyValueStats(
+ const std::vector<std::shared_ptr<ColumnStats>>& field_stats, SimpleStats*
key_stats,
+ SimpleStats* value_stats) const {
+ // key stats
+ std::vector<std::shared_ptr<ColumnStats>> key_column_stats;
+ key_column_stats.reserve(primary_keys_.size());
+ for (const auto& key : primary_keys_) {
+ int32_t idx = write_schema_->GetFieldIndex(key);
+ if (idx == -1) {
+ return Status::Invalid(
+ fmt::format("cannot find primary key field {} in write
schema", key));
+ }
+ key_column_stats.push_back(field_stats[idx]);
+ }
+ PAIMON_ASSIGN_OR_RAISE(*key_stats,
+ SimpleStatsConverter::ToBinary(key_column_stats,
pool_.get()));
+ // value stats
+ std::vector<std::shared_ptr<ColumnStats>> value_column_stats(
+ field_stats.begin() + SpecialFields::KEY_VALUE_SPECIAL_FIELD_COUNT,
field_stats.end());
+ PAIMON_ASSIGN_OR_RAISE(*value_stats,
+ SimpleStatsConverter::ToBinary(value_column_stats,
pool_.get()));
+ return Status::OK();
+}
+
+Status KeyValueDataFileWriter::GenerateKeyStatsWithAllNull(SimpleStats*
key_stats) const {
+ BinaryRow min_values(primary_keys_.size());
+ BinaryRow max_values(primary_keys_.size());
+ BinaryArray null_counts;
+
+ BinaryRowWriter min_writer(&min_values, /*initial_size=*/0, pool_.get());
+ BinaryRowWriter max_writer(&max_values, /*initial_size=*/0, pool_.get());
+ BinaryArrayWriter null_counts_writer(&null_counts, primary_keys_.size(),
sizeof(int64_t),
+ pool_.get());
+ min_writer.Reset();
+ max_writer.Reset();
+ null_counts_writer.Reset();
+
+ for (size_t i = 0; i < primary_keys_.size(); ++i) {
+ auto data_type =
write_schema_->GetFieldByName(primary_keys_[i])->type();
+ BinaryRowWriter::FieldSetterFunc setter;
+ PAIMON_ASSIGN_OR_RAISE(setter, BinaryRowWriter::CreateFieldSetter(i,
data_type));
+ setter(NullType(), &min_writer);
+ setter(NullType(), &max_writer);
+ null_counts_writer.SetNullAt(i);
+ }
+ min_writer.Complete();
+ max_writer.Complete();
+ null_counts_writer.Complete();
+ *key_stats = SimpleStats(min_values, max_values, null_counts);
+ return Status::OK();
+}
+
+Result<std::vector<std::shared_ptr<ColumnStats>>>
KeyValueDataFileWriter::GetFieldStats() {
+ if (!closed_) {
+ return Status::Invalid("Cannot access metric unless the writer is
closed.");
+ }
+ if (disable_stats_) {
+ return std::vector<std::shared_ptr<ColumnStats>>();
+ }
+ if (stats_extractor_ == nullptr) {
+ assert(false);
+ return Status::Invalid("simple stats extractor is null pointer.");
+ }
+ return stats_extractor_->Extract(fs_, path_, pool_);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/key_value_data_file_writer.h
b/src/paimon/core/io/key_value_data_file_writer.h
new file mode 100644
index 0000000..f62bf64
--- /dev/null
+++ b/src/paimon/core/io/key_value_data_file_writer.h
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <cstdint>
+#include <functional>
+#include <limits>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/key_value.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class Schema;
+} // namespace arrow
+struct ArrowArray;
+
+namespace paimon {
+class ColumnStats;
+class FormatStatsExtractor;
+class BinaryRow;
+class InternalRow;
+class MemoryPool;
+class SimpleStats;
+
+class KeyValueDataFileWriter
+ : public SingleFileWriter<KeyValueBatch, std::shared_ptr<DataFileMeta>> {
+ public:
+ KeyValueDataFileWriter(const std::string& compression,
+ std::function<Status(KeyValueBatch&&,
::ArrowArray*)> converter,
+ int64_t schema_id, int32_t level, FileSource
file_source,
+ const std::vector<std::string>& primary_keys,
+ const std::shared_ptr<FormatStatsExtractor>&
stats_extractor,
+ const std::shared_ptr<arrow::Schema>& write_schema,
+ bool is_external_path, const
std::shared_ptr<MemoryPool>& pool);
+
+ Status Write(KeyValueBatch batch) override;
+
+ Result<std::shared_ptr<DataFileMeta>> GetResult() override;
+
+ private:
+ Result<std::vector<std::shared_ptr<ColumnStats>>> GetFieldStats();
+
+ Status GenerateMinMaxKey(BinaryRow* min_key, BinaryRow* max_key) const;
+
+ Status GenerateKeyValueStats(const
std::vector<std::shared_ptr<ColumnStats>>& field_stats,
+ SimpleStats* key_stats, SimpleStats*
value_stats) const;
+ Status GenerateKeyStatsWithAllNull(SimpleStats* key_stats) const;
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+ int64_t schema_id_;
+ int32_t level_;
+ FileSource file_source_;
+ std::vector<std::string> primary_keys_;
+ std::shared_ptr<FormatStatsExtractor> stats_extractor_;
+ std::shared_ptr<arrow::Schema> write_schema_;
+ bool is_external_path_;
+ bool disable_stats_;
+
+ int64_t delete_row_count_ = 0;
+ int64_t min_sequence_number_ = std::numeric_limits<int64_t>::max();
+ int64_t max_sequence_number_ = std::numeric_limits<int64_t>::min();
+ std::shared_ptr<InternalRow> min_key_;
+ std::shared_ptr<InternalRow> max_key_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/rolling_blob_file_writer.cpp
b/src/paimon/core/io/rolling_blob_file_writer.cpp
new file mode 100644
index 0000000..ccd6ca4
--- /dev/null
+++ b/src/paimon/core/io/rolling_blob_file_writer.cpp
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/rolling_blob_file_writer.h"
+
+#include <memory>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "arrow/array/array_base.h"
+#include "arrow/array/array_nested.h"
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "arrow/c/helpers.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/data/blob_utils.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/macros.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+
+RollingBlobFileWriter::RollingBlobFileWriter(
+ int64_t target_file_size,
+ std::function<Result<std::unique_ptr<MainWriter>>()> create_file_writer,
+ const std::shared_ptr<arrow::Schema>& blob_schema,
+ MultipleBlobFileWriter::BlobWriterCreator blob_writer_creator,
+ const std::shared_ptr<arrow::DataType>& data_type)
+ : RollingFileWriter<::ArrowArray*,
std::shared_ptr<DataFileMeta>>(target_file_size,
+
create_file_writer),
+ blob_schema_(blob_schema),
+ blob_writer_creator_(std::move(blob_writer_creator)),
+ data_type_(data_type),
+ logger_(Logger::GetLogger("RollingBlobFileWriter")) {}
+
+Status RollingBlobFileWriter::Write(::ArrowArray* record) {
+ ScopeGuard guard([this]() -> void { this->Abort(); });
+ // Open the current writer if write the first record or roll over happen
before.
+ if (PAIMON_UNLIKELY(current_writer_ == nullptr)) {
+ PAIMON_RETURN_NOT_OK(OpenCurrentWriter());
+ }
+ if (PAIMON_UNLIKELY(blob_writer_ == nullptr)) {
+ blob_writer_ = std::make_unique<MultipleBlobFileWriter>(blob_schema_,
blob_writer_creator_);
+ }
+ int64_t record_count = record->length;
+ PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(std::shared_ptr<arrow::Array>
arrow_array,
+ arrow::ImportArray(record, data_type_));
+ auto struct_array =
std::dynamic_pointer_cast<arrow::StructArray>(arrow_array);
+
+ PAIMON_ASSIGN_OR_RAISE(BlobUtils::SeparatedStructArrays separated_arrays,
+ BlobUtils::SeparateBlobArray(struct_array));
+ // Write main (non-blob) data
+ ::ArrowArray c_main_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportArray(*separated_arrays.main_array, &c_main_array));
+ ScopeGuard array_lifecycle_guard(
+ [&c_main_array]() -> void { ArrowArrayRelease(&c_main_array); });
+ PAIMON_RETURN_NOT_OK(current_writer_->Write(&c_main_array));
+
+ // Write blob data via MultipleBlobFileWriter (each blob field
independently)
+ ::ArrowArray c_blob_array;
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(
+ arrow::ExportArray(*separated_arrays.blob_array, &c_blob_array));
+ ScopeGuard blob_array_guard([&c_blob_array]() -> void {
ArrowArrayRelease(&c_blob_array); });
+ PAIMON_RETURN_NOT_OK(blob_writer_->Write(&c_blob_array));
+
+ record_count_ += record_count;
+ PAIMON_ASSIGN_OR_RAISE(bool need_rolling_file, NeedRollingFile());
+ if (need_rolling_file) {
+ PAIMON_RETURN_NOT_OK(CloseCurrentWriter());
+ }
+ guard.Release();
+ return Status::OK();
+}
+
+Status RollingBlobFileWriter::CloseCurrentWriter() {
+ if (current_writer_ == nullptr) {
+ return Status::OK();
+ }
+ if (blob_writer_ == nullptr) {
+ return Status::OK();
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> main_data_file_meta,
CloseMainWriter());
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>>
blob_metas,
+ CloseBlobWriter());
+ PAIMON_RETURN_NOT_OK(
+ ValidateFileConsistency(main_data_file_meta, blob_metas,
blob_schema_->num_fields()));
+
+ results_.push_back(main_data_file_meta);
+ results_.insert(results_.end(), blob_metas.begin(), blob_metas.end());
+
+ current_writer_.reset();
+ return Status::OK();
+}
+
+Result<std::vector<std::shared_ptr<DataFileMeta>>>
RollingBlobFileWriter::GetResult() {
+ if (!closed_) {
+ return Status::Invalid("Cannot access the results unless close all
writers.");
+ }
+ return results_;
+}
+
+Result<std::shared_ptr<DataFileMeta>> RollingBlobFileWriter::CloseMainWriter()
{
+ PAIMON_RETURN_NOT_OK(current_writer_->Close());
+ PAIMON_ASSIGN_OR_RAISE(auto abort_executor,
current_writer_->GetAbortExecutor());
+ closed_writers_.push_back(abort_executor);
+ return current_writer_->GetResult();
+}
+
+Result<std::vector<std::shared_ptr<DataFileMeta>>>
RollingBlobFileWriter::CloseBlobWriter() {
+ PAIMON_RETURN_NOT_OK(blob_writer_->Close());
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<DataFileMeta>> results,
+ blob_writer_->GetResult());
+ blob_writer_.reset();
+ return results;
+}
+
+Status RollingBlobFileWriter::ValidateFileConsistency(
+ const std::shared_ptr<DataFileMeta>& main_data_file_meta,
+ const std::vector<std::shared_ptr<DataFileMeta>>& blob_tagged_metas,
int32_t blob_field_count) {
+ if (blob_tagged_metas.empty()) {
+ return Status::OK();
+ }
+ // With multiple blob fields, each blob field produces its own set of
files.
+ // total_blob_row_count should be exactly main_row_count *
blob_field_count.
+ int64_t main_row_count = main_data_file_meta->row_count;
+ int64_t expected_blob_row_count = main_row_count * blob_field_count;
+ int64_t total_blob_row_count = 0;
+ for (const auto& blob_tagged_meta : blob_tagged_metas) {
+ total_blob_row_count += blob_tagged_meta->row_count;
+ }
+ if (total_blob_row_count != expected_blob_row_count) {
+ std::vector<std::string> blob_file_names;
+ for (const auto& blob_tagged_meta : blob_tagged_metas) {
+ blob_file_names.push_back(blob_tagged_meta->file_name);
+ }
+ return Status::Invalid(fmt::format(
+ "This is a bug: The row count of main file and blob files does not
match. "
+ "Main file: {} (row count: {}), blob field count: {}, "
+ "expected blob row count: {}, blob files: {} (actual total row
count: {})",
+ main_data_file_meta->file_name, main_row_count, blob_field_count,
+ expected_blob_row_count, fmt::join(blob_file_names, ", "),
total_blob_row_count));
+ }
+ return Status::OK();
+}
+
+Status RollingBlobFileWriter::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ auto s = CloseCurrentWriter();
+ if (!s.ok()) {
+ if (current_writer_) {
+ PAIMON_LOG_WARN(logger_, "Exception occurs when writing file %s.
Cleaning up: %s",
+ current_writer_->GetPath().c_str(),
s.ToString().c_str());
+ }
+ Abort();
+ }
+ closed_ = true;
+ return s;
+}
+
+void RollingBlobFileWriter::Abort() {
+ if (current_writer_ != nullptr) {
+ current_writer_->Abort();
+ current_writer_.reset();
+ }
+ for (auto& abort_executor : closed_writers_) {
+ abort_executor.Abort();
+ }
+ if (blob_writer_ != nullptr) {
+ blob_writer_->Abort();
+ blob_writer_.reset();
+ }
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/rolling_blob_file_writer.h
b/src/paimon/core/io/rolling_blob_file_writer.h
new file mode 100644
index 0000000..f35f696
--- /dev/null
+++ b/src/paimon/core/io/rolling_blob_file_writer.h
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <set>
+#include <vector>
+
+#include "arrow/array/array_nested.h"
+#include "arrow/c/bridge.h"
+#include "arrow/result.h"
+#include "paimon/common/metrics/metrics_impl.h"
+#include "paimon/core/io/multiple_blob_file_writer.h"
+#include "paimon/core/io/rolling_file_writer.h"
+#include "paimon/metrics.h"
+#include "paimon/record_batch.h"
+
+namespace paimon {
+
+/// A rolling file writer that handles both normal data and blob data. This
writer creates separate
+/// files for normal columns and blob columns, managing their lifecycle and
ensuring consistency
+/// between them.
+///
+/// Multiple blob fields are supported. Each blob field is written to its own
set of blob files
+/// independently via MultipleBlobFileWriter.
+///
+/// <pre>
+/// For example,
+/// given a table schema with normal columns (id INT, name STRING) and blob
columns (data1 BLOB,
+/// data2 BLOB), this writer will create separate files for (id, name),
(data1), and (data2).
+/// It will roll files based on the specified target file size, ensuring that
both normal and blob
+/// files are rolled simultaneously.
+///
+/// Every time a file is rolled, the writer will close the current normal data
file and blob data
+/// files, so one normal data file may correspond to multiple blob data files.
+///
+/// Normal file1: f1.parquet may include (blob1_1.blob, blob1_2.blob,
blob2_1.blob)
+/// Normal file2: f2.parquet may include (blob1_3.blob, blob2_2.blob)
+///
+/// </pre>
+class RollingBlobFileWriter
+ : public RollingFileWriter<::ArrowArray*, std::shared_ptr<DataFileMeta>> {
+ public:
+ using MainWriter = SingleFileWriter<::ArrowArray*,
std::shared_ptr<DataFileMeta>>;
+
+ RollingBlobFileWriter(int64_t target_file_size,
+ std::function<Result<std::unique_ptr<MainWriter>>()>
create_file_writer,
+ const std::shared_ptr<arrow::Schema>& blob_schema,
+ MultipleBlobFileWriter::BlobWriterCreator
blob_writer_creator,
+ const std::shared_ptr<arrow::DataType>& data_type);
+ ~RollingBlobFileWriter() override = default;
+
+ Status Write(::ArrowArray* record) override;
+ void Abort() override;
+ Status Close() override;
+ Result<std::vector<std::shared_ptr<DataFileMeta>>> GetResult() override;
+
+ private:
+ static Status ValidateFileConsistency(
+ const std::shared_ptr<DataFileMeta>& main_data_file_meta,
+ const std::vector<std::shared_ptr<DataFileMeta>>& blob_tagged_metas,
+ int32_t blob_field_count);
+
+ Status CloseCurrentWriter();
+
+ Result<std::shared_ptr<DataFileMeta>> CloseMainWriter();
+ Result<std::vector<std::shared_ptr<DataFileMeta>>> CloseBlobWriter();
+
+ std::shared_ptr<arrow::Schema> blob_schema_;
+ MultipleBlobFileWriter::BlobWriterCreator blob_writer_creator_;
+ std::unique_ptr<MultipleBlobFileWriter> blob_writer_;
+ std::shared_ptr<arrow::DataType> data_type_;
+
+ std::unique_ptr<Logger> logger_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/rolling_blob_file_writer_test.cpp
b/src/paimon/core/io/rolling_blob_file_writer_test.cpp
new file mode 100644
index 0000000..9e2a4be
--- /dev/null
+++ b/src/paimon/core/io/rolling_blob_file_writer_test.cpp
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/rolling_blob_file_writer.h"
+
+#include <optional>
+#include <string>
+#include <variant>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/data/data_define.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class RollingBlobFileWriterTest : public testing::Test {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+};
+
+TEST_F(RollingBlobFileWriterTest, ValidateFileConsistency) {
+ auto file_meta1 = std::make_shared<DataFileMeta>(
+ "data-xxx.xxx", /*file_size=*/405, /*row_count=*/4,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
+ /*key_stats=*/SimpleStats::EmptyStats(),
+ BinaryRowGenerator::GenerateStats({std::string("str_0"), 1},
{std::string("str_3"), 2},
+ std::vector<int64_t>({0, 2}),
pool_.get()),
+ /*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1724090888706ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
/*first_row_id=*/0,
+ /*write_cols=*/std::vector<std::string>({"f0", "f1"}));
+
+ auto file_meta2 = std::make_shared<DataFileMeta>(
+ "data-xxx.blob", /*file_size=*/764, /*row_count=*/3,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
+ /*key_stats=*/SimpleStats::EmptyStats(),
+ BinaryRowGenerator::GenerateStats({NullType()}, {NullType()},
std::vector<int64_t>({-1}),
+ pool_.get()),
+ /*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1724090888706ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
/*first_row_id=*/0,
+ /*write_cols=*/std::vector<std::string>({"blob"}));
+
+ auto file_meta3 = std::make_shared<DataFileMeta>(
+ "data-xxx.blob", /*file_size=*/3023, /*row_count=*/1,
+ /*min_key=*/BinaryRow::EmptyRow(), /*max_key=*/BinaryRow::EmptyRow(),
+ /*key_stats=*/SimpleStats::EmptyStats(),
+ BinaryRowGenerator::GenerateStats({NullType()}, {NullType()},
std::vector<int64_t>({-1}),
+ pool_.get()),
+ /*min_sequence_number=*/1, /*max_sequence_number=*/1, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1724090888706ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
/*first_row_id=*/3,
+ /*write_cols=*/std::vector<std::string>({"blob"}));
+ ASSERT_OK(RollingBlobFileWriter::ValidateFileConsistency(file_meta1,
{file_meta2, file_meta3},
+
/*blob_field_count=*/1));
+
ASSERT_NOK_WITH_MSG(RollingBlobFileWriter::ValidateFileConsistency(file_meta1,
{file_meta2},
+
/*blob_field_count=*/2),
+ "This is a bug: The row count of main file and blob
files does not match.");
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/io/rolling_file_writer.h
b/src/paimon/core/io/rolling_file_writer.h
new file mode 100644
index 0000000..fa8a51d
--- /dev/null
+++ b/src/paimon/core/io/rolling_file_writer.h
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "arrow/c/bridge.h"
+#include "paimon/common/metrics/metrics_impl.h"
+#include "paimon/core/io/file_writer.h"
+#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/key_value.h"
+#include "paimon/metrics.h"
+#include "paimon/record_batch.h"
+
+namespace paimon {
+
+// Writer to roll over to a new file if the current size exceed the target
file size.
+template <typename T, typename R>
+class RollingFileWriter : public FileWriter<T, std::vector<R>> {
+ public:
+ RollingFileWriter(
+ int64_t target_file_size,
+ std::function<Result<std::unique_ptr<SingleFileWriter<T, R>>>()>
create_file_writer)
+ : target_file_size_(target_file_size),
+ create_file_writer(create_file_writer),
+ metrics_(std::make_shared<MetricsImpl>()),
+ logger_(Logger::GetLogger("RollingFileWriter")) {}
+
+ ~RollingFileWriter() = default;
+
+ Status Write(T record) override;
+ void Abort() override;
+ Status Close() override;
+ Result<std::vector<R>> GetResult() override;
+
+ int64_t RecordCount() const override {
+ return record_count_;
+ }
+
+ std::shared_ptr<Metrics> GetMetrics() const override {
+ return metrics_;
+ }
+
+ int64_t TargetFileSize() const {
+ return target_file_size_;
+ }
+
+ protected:
+ static constexpr int32_t CHECK_ROLLING_RECORD_CNT = 1000;
+
+ bool SuggestCheck();
+ Result<bool> NeedRollingFile();
+ Result<std::unique_ptr<SingleFileWriter<T, R>>> NewWriter();
+ Status OpenCurrentWriter();
+
+ int64_t target_file_size_ = 0;
+ std::function<Result<std::unique_ptr<SingleFileWriter<T, R>>>()>
create_file_writer;
+ std::shared_ptr<Metrics> metrics_;
+
+ int64_t record_count_ = 0;
+ int64_t last_need_rolling_record_count_ = 0;
+ bool closed_ = false;
+
+ std::vector<typename SingleFileWriter<T, R>::AbortExecutor>
closed_writers_;
+ std::vector<R> results_;
+ std::unique_ptr<SingleFileWriter<T, R>> current_writer_;
+
+ private:
+ Status CloseCurrentWriter();
+
+ std::unique_ptr<Logger> logger_;
+};
+
+template <typename T, typename R>
+bool RollingFileWriter<T, R>::SuggestCheck() {
+ bool suggest_check = false;
+ if (record_count_ - last_need_rolling_record_count_ >=
CHECK_ROLLING_RECORD_CNT) {
+ suggest_check = true;
+ last_need_rolling_record_count_ = record_count_;
+ }
+ return suggest_check;
+}
+
+template <typename T, typename R>
+Result<bool> RollingFileWriter<T, R>::NeedRollingFile() {
+ return current_writer_->ReachTargetSize(SuggestCheck(), target_file_size_);
+}
+
+template <typename T, typename R>
+Status RollingFileWriter<T, R>::Write(T record) {
+ ScopeGuard guard([this]() -> void { this->Abort(); });
+ // Open the current writer if write the first record or roll over happen
before.
+ if (PAIMON_UNLIKELY(current_writer_ == nullptr)) {
+ PAIMON_RETURN_NOT_OK(OpenCurrentWriter());
+ }
+ int64_t record_count = 0;
+ if constexpr (std::is_same_v<T, ::ArrowArray*>) {
+ record_count = record->length;
+ } else if constexpr (std::is_same_v<T, KeyValueBatch>) {
+ record_count = record.batch->length;
+ } else {
+ record_count = 1;
+ }
+ PAIMON_RETURN_NOT_OK(current_writer_->Write(std::move(record)));
+ record_count_ += record_count;
+ PAIMON_ASSIGN_OR_RAISE(bool need_rolling_file, NeedRollingFile());
+ if (need_rolling_file) {
+ PAIMON_RETURN_NOT_OK(CloseCurrentWriter());
+ }
+ guard.Release();
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Result<std::vector<R>> RollingFileWriter<T, R>::GetResult() {
+ if (!closed_) {
+ return Status::Invalid("Cannot access the results unless close all
writers.");
+ }
+ return results_;
+}
+
+template <typename T, typename R>
+Result<std::unique_ptr<SingleFileWriter<T, R>>> RollingFileWriter<T,
R>::NewWriter() {
+ return create_file_writer();
+}
+
+template <typename T, typename R>
+Status RollingFileWriter<T, R>::OpenCurrentWriter() {
+ PAIMON_ASSIGN_OR_RAISE(current_writer_, NewWriter());
+ if (metrics_) {
+ metrics_->Merge(current_writer_->GetMetrics());
+ }
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Status RollingFileWriter<T, R>::CloseCurrentWriter() {
+ if (current_writer_ == nullptr) {
+ return Status::OK();
+ }
+ std::shared_ptr<Metrics> current_metrics = current_writer_->GetMetrics();
+ PAIMON_RETURN_NOT_OK(current_writer_->Close());
+ PAIMON_ASSIGN_OR_RAISE(auto abort_executor,
current_writer_->GetAbortExecutor());
+ closed_writers_.push_back(abort_executor);
+ PAIMON_ASSIGN_OR_RAISE(R result, current_writer_->GetResult());
+ results_.push_back(result);
+ current_writer_.reset();
+ if (metrics_) {
+ metrics_->Merge(current_metrics);
+ }
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Status RollingFileWriter<T, R>::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ auto s = CloseCurrentWriter();
+ if (!s.ok()) {
+ if (current_writer_) {
+ PAIMON_LOG_WARN(logger_, "Exception occurs when writing file %s.
Cleaning up: %s",
+ current_writer_->GetPath().c_str(),
s.ToString().c_str());
+ }
+ Abort();
+ }
+ closed_ = true;
+ return s;
+}
+
+template <typename T, typename R>
+void RollingFileWriter<T, R>::Abort() {
+ if (current_writer_ != nullptr) {
+ current_writer_->Abort();
+ }
+ for (auto& abort_executor : closed_writers_) {
+ abort_executor.Abort();
+ }
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/single_file_writer.h
b/src/paimon/core/io/single_file_writer.h
new file mode 100644
index 0000000..2ead8ae
--- /dev/null
+++ b/src/paimon/core/io/single_file_writer.h
@@ -0,0 +1,241 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <cassert>
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <string>
+#include <utility>
+
+#include "arrow/c/abi.h"
+#include "arrow/c/helpers.h"
+#include "fmt/format.h"
+#include "paimon/common/utils/arrow/arrow_utils.h"
+#include "paimon/common/utils/scope_guard.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/file_writer.h"
+#include "paimon/format/format_writer.h"
+#include "paimon/format/writer_builder.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/logging.h"
+#include "paimon/macros.h"
+#include "paimon/record_batch.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class RecordBatch;
+class Metrics;
+
+/// A `FileWriter` to produce a single file.
+///
+/// <T> type of records to write.
+/// <R> is the type of result to produce after writing a file.
+template <typename T, typename R>
+class SingleFileWriter : public FileWriter<T, R> {
+ public:
+ /// Abort executor to just have reference of path instead of whole writer.
+ class AbortExecutor {
+ public:
+ AbortExecutor(const std::shared_ptr<FileSystem>& fs, const
std::string& path)
+ : fs_(fs), path_(path),
logger_(Logger::GetLogger("AbortExecutor")) {}
+
+ void Abort() {
+ if (fs_) {
+ auto status = fs_->Delete(path_);
+ if (!status.ok()) {
+ PAIMON_LOG_WARN(logger_, "Exception occurs when deleting
%s: %s", path_.c_str(),
+ status.ToString().c_str());
+ }
+ }
+ }
+
+ private:
+ std::shared_ptr<FileSystem> fs_;
+ std::string path_;
+ std::shared_ptr<Logger> logger_;
+ };
+
+ SingleFileWriter(const std::string& compression,
+ std::function<Status(T, ::ArrowArray*)> converter)
+ : compression_(compression),
+ converter_(converter),
+ logger_(Logger::GetLogger("SingleFileWriter")) {}
+
+ virtual Status Init(const std::shared_ptr<FileSystem>& fs, const
std::string& path,
+ const std::shared_ptr<WriterBuilder>& writer_builder);
+
+ Status Write(T record) override;
+
+ int64_t RecordCount() const override {
+ return record_count_;
+ }
+ void Abort() override;
+ Status Close() override;
+
+ std::shared_ptr<Metrics> GetMetrics() const override {
+ if (writer_) {
+ return writer_->GetWriterMetrics();
+ }
+ return nullptr;
+ }
+
+ Result<bool> ReachTargetSize(bool suggested_check, int64_t target_size);
+
+ Result<AbortExecutor> GetAbortExecutor() const {
+ if (closed_ == false) {
+ return Status::Invalid("Writer should be closed!");
+ }
+ return AbortExecutor(fs_, path_);
+ }
+
+ std::string GetPath() const {
+ return path_;
+ }
+
+ protected:
+ int64_t output_bytes_ = -1;
+ std::string compression_;
+ std::function<Status(T, ArrowArray*)> converter_;
+ std::shared_ptr<FileSystem> fs_;
+ std::shared_ptr<OutputStream> out_; // nullptr for DirectWriterBuilder
+ bool closed_ = false;
+ std::string path_;
+
+ private:
+ int64_t record_count_ = 0;
+ std::unique_ptr<FormatWriter> writer_;
+
+ std::unique_ptr<Logger> logger_;
+};
+
+template <typename T, typename R>
+Status SingleFileWriter<T, R>::Init(const std::shared_ptr<FileSystem>& fs,
const std::string& path,
+ const std::shared_ptr<WriterBuilder>&
writer_builder) {
+ ScopeGuard guard([this]() -> void {
+ this->Abort();
+ PAIMON_LOG_WARN(logger_,
+ "Exception occurs when initializing single file writer
%s. Cleaning up.",
+ path_.c_str());
+ });
+ path_ = path;
+ fs_ = fs;
+
+ if (auto specific_fs_writer_builder =
+
std::dynamic_pointer_cast<SpecificFSWriterBuilder>(writer_builder)) {
+ specific_fs_writer_builder->WithFileSystem(fs);
+ }
+ if (auto direct_writer_builder =
+ std::dynamic_pointer_cast<DirectWriterBuilder>(writer_builder)) {
+ PAIMON_ASSIGN_OR_RAISE(writer_,
direct_writer_builder->BuildFromPath(path));
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(out_, fs_->Create(path, /*overwrite=*/false));
+ PAIMON_ASSIGN_OR_RAISE(writer_, writer_builder->Build(out_,
compression_));
+ assert(out_);
+ }
+ assert(writer_);
+ record_count_ = 0;
+ closed_ = false;
+ guard.Release();
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Status SingleFileWriter<T, R>::Write(T record) {
+ if (PAIMON_UNLIKELY(closed_)) {
+ return Status::Invalid("Writer has already closed!");
+ }
+ ScopeGuard guard([this]() -> void { this->Abort(); });
+ int64_t record_count = 0;
+ if (!converter_) {
+ if constexpr (std::is_same_v<T, ::ArrowArray*>) {
+ record_count = record->length;
+ ScopeGuard inner_guard([&record]() { ArrowArrayRelease(record); });
+ PAIMON_RETURN_NOT_OK(writer_->AddBatch(record));
+ inner_guard.Release();
+ } else {
+ return Status::Invalid("converter is not set");
+ }
+ } else {
+ ArrowArray array;
+ ArrowArrayMarkReleased(&array); // reset array
+ ScopeGuard inner_guard([&array]() { ArrowArrayRelease(&array); });
+ PAIMON_RETURN_NOT_OK(converter_(std::move(record), &array));
+ record_count = array.length;
+ PAIMON_RETURN_NOT_OK(writer_->AddBatch(&array));
+ inner_guard.Release();
+ }
+ record_count_ += record_count;
+ guard.Release();
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Status SingleFileWriter<T, R>::Close() {
+ if (closed_) {
+ return Status::OK();
+ }
+ PAIMON_LOG_DEBUG(logger_, "Closing file %s", path_.c_str());
+ ScopeGuard guard([this]() -> void {
+ this->Abort();
+ PAIMON_LOG_WARN(logger_, "Exception occurs when closing file %s.
Cleaning up.",
+ path_.c_str());
+ });
+ PAIMON_RETURN_NOT_OK(writer_->Flush());
+ PAIMON_RETURN_NOT_OK(writer_->Finish());
+ if (out_) {
+ PAIMON_RETURN_NOT_OK(out_->Flush());
+ PAIMON_ASSIGN_OR_RAISE(output_bytes_, out_->GetPos());
+ PAIMON_RETURN_NOT_OK(out_->Close());
+ } else {
+ PAIMON_ASSIGN_OR_RAISE(std::unique_ptr<FileStatus> file_status,
fs_->GetFileStatus(path_));
+ output_bytes_ = file_status->GetLen();
+ }
+ closed_ = true;
+ guard.Release();
+ return Status::OK();
+}
+
+template <typename T, typename R>
+Result<bool> SingleFileWriter<T, R>::ReachTargetSize(bool suggested_check,
int64_t target_size) {
+ return writer_->ReachTargetSize(suggested_check, target_size);
+}
+
+template <typename T, typename R>
+void SingleFileWriter<T, R>::Abort() {
+ if (out_) {
+ auto status = out_->Close();
+ if (!status.ok()) {
+ PAIMON_LOG_WARN(logger_, "Exception occurs when closing %s: %s",
path_.c_str(),
+ status.ToString().c_str());
+ }
+ }
+ if (fs_) {
+ auto status = fs_->Delete(path_);
+ if (!status.ok()) {
+ PAIMON_LOG_WARN(logger_, "Exception occurs when closing %s: %s",
path_.c_str(),
+ status.ToString().c_str());
+ }
+ }
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/single_file_writer_test.cpp
b/src/paimon/core/io/single_file_writer_test.cpp
new file mode 100644
index 0000000..c3ef9eb
--- /dev/null
+++ b/src/paimon/core/io/single_file_writer_test.cpp
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "paimon/core/io/single_file_writer.h"
+
+#include <map>
+#include <utility>
+
+#include "arrow/api.h"
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "arrow/ipc/json_simple.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class SimpleSingleFileWriter : public SingleFileWriter<int32_t, bool> {
+ public:
+ SimpleSingleFileWriter(const std::string& compression,
+ std::function<Status(int32_t, ArrowArray*)>
converter)
+ : SingleFileWriter<int32_t, bool>(compression, converter) {}
+
+ Result<bool> GetResult() override {
+ return true;
+ }
+};
+
+TEST(SingleFileWriterTest, TestSimple) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::string file_path = dir->Str() + "/single-file";
+ auto data_type = arrow::struct_({arrow::field("col", arrow::int32())});
+ auto converter = [&](int32_t value, ::ArrowArray* dest) -> Status {
+ std::string value_str = "[[" + std::to_string(value) + "]]";
+ auto array =
+ arrow::ipc::internal::json::ArrayFromJSON(data_type,
value_str.c_str()).ValueOrDie();
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, dest));
+ return Status::OK();
+ };
+ SimpleSingleFileWriter writer("zstd", converter);
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::MANIFEST_FORMAT, "orc"},
{Options::FILE_FORMAT, "orc"}}));
+ auto file_format = options.GetWriteFileFormat(/*level=*/0);
+ auto file_system = options.GetFileSystem();
+ ArrowSchema arrow_schema;
+ ASSERT_TRUE(arrow::ExportType(*data_type, &arrow_schema).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> writer_builder,
+ file_format->CreateWriterBuilder(&arrow_schema,
/*batch_size=*/100));
+ ASSERT_OK(writer.Init(file_system, file_path, writer_builder));
+ ASSERT_EQ(file_path, writer.GetPath());
+ ASSERT_OK(writer.Write(100));
+ ASSERT_EQ(1, writer.RecordCount());
+ ASSERT_NOK_WITH_MSG(writer.GetAbortExecutor(), "Writer should be closed");
+ ASSERT_OK(writer.Close());
+ ASSERT_OK_AND_ASSIGN(auto file_status,
file_system->GetFileStatus(file_path));
+ ASSERT_GT(file_status->GetLen(), 0);
+ ASSERT_OK_AND_ASSIGN(auto abort_executor, writer.GetAbortExecutor());
+ abort_executor.Abort();
+ ASSERT_OK_AND_ASSIGN(auto exist, file_system->Exists(file_path));
+ ASSERT_FALSE(exist);
+}
+
+TEST(SingleFileWriterTest, TestInvalidConvert) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::string file_path = dir->Str() + "/single-file";
+ auto data_type = arrow::struct_({arrow::field("col", arrow::int32())});
+ auto converter = [&](int32_t value, ::ArrowArray* dest) -> Status {
+ return Status::Invalid("");
+ };
+ SimpleSingleFileWriter writer("zstd", converter);
+ ASSERT_OK_AND_ASSIGN(
+ CoreOptions options,
+ CoreOptions::FromMap({{Options::MANIFEST_FORMAT, "orc"},
{Options::FILE_FORMAT, "orc"}}));
+ auto file_format = options.GetWriteFileFormat(/*level=*/0);
+ auto file_system = options.GetFileSystem();
+ ArrowSchema arrow_schema;
+ ASSERT_TRUE(arrow::ExportType(*data_type, &arrow_schema).ok());
+ ASSERT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> writer_builder,
+ file_format->CreateWriterBuilder(&arrow_schema,
/*batch_size=*/100));
+ ASSERT_OK(writer.Init(file_system, file_path, writer_builder));
+ ASSERT_NOK(writer.Write(100));
+ writer.Abort();
+ ASSERT_OK_AND_ASSIGN(auto exist, file_system->Exists(file_path));
+ ASSERT_FALSE(exist);
+}
+
+} // namespace paimon::test
