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 a22663a feat(manifest): add manifest entry, file metadata and
serialization utilities (#86)
a22663a is described below
commit a22663ac5ddd9c7bfc1883f62ef184c34a517d24
Author: Yonghao Fang <[email protected]>
AuthorDate: Thu Jun 18 10:16:42 2026 +0800
feat(manifest): add manifest entry, file metadata and serialization
utilities (#86)
---
src/paimon/core/manifest/file_entry.h | 160 +++++++++++++++++
src/paimon/core/manifest/file_entry_test.cpp | 193 +++++++++++++++++++++
src/paimon/core/manifest/file_kind.cpp | 47 +++++
src/paimon/core/manifest/file_kind.h | 46 +++++
src/paimon/core/manifest/file_kind_test.cpp | 40 +++++
src/paimon/core/manifest/file_source.cpp | 32 ++++
src/paimon/core/manifest/file_source.h | 80 +++++++++
src/paimon/core/manifest/file_source_test.cpp | 35 ++++
src/paimon/core/manifest/manifest_entry.cpp | 57 ++++++
src/paimon/core/manifest/manifest_entry.h | 160 +++++++++++++++++
.../core/manifest/manifest_entry_serializer.cpp | 79 +++++++++
.../core/manifest/manifest_entry_serializer.h | 63 +++++++
.../manifest/manifest_entry_serializer_test.cpp | 63 +++++++
src/paimon/core/manifest/manifest_entry_writer.cpp | 80 +++++++++
src/paimon/core/manifest/manifest_entry_writer.h | 81 +++++++++
.../core/manifest/manifest_entry_writer_test.cpp | 192 ++++++++++++++++++++
src/paimon/core/manifest/manifest_file_meta.cpp | 101 +++++++++++
src/paimon/core/manifest/manifest_file_meta.h | 111 ++++++++++++
.../manifest/manifest_file_meta_serializer.cpp | 141 +++++++++++++++
.../core/manifest/manifest_file_meta_serializer.h | 62 +++++++
.../manifest_file_meta_serializer_test.cpp | 107 ++++++++++++
21 files changed, 1930 insertions(+)
diff --git a/src/paimon/core/manifest/file_entry.h
b/src/paimon/core/manifest/file_entry.h
new file mode 100644
index 0000000..30705a8
--- /dev/null
+++ b/src/paimon/core/manifest/file_entry.h
@@ -0,0 +1,160 @@
+/*
+ * 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 <cstddef>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <variant>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/utils/linked_hash_map.h"
+#include "paimon/common/utils/murmurhash_utils.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+/// Entry representing a file.
+class FileEntry {
+ public:
+ /// The same `Identifier` indicates that the `ManifestEntry` refers to the
same data
+ /// file.
+ struct Identifier {
+ Identifier(const BinaryRow& _partition, int32_t _bucket, int32_t
_level,
+ const std::string& _file_name, const
std::optional<std::string>& _external_path)
+ : partition(_partition),
+ bucket(_bucket),
+ level(_level),
+ file_name(_file_name),
+ external_path(_external_path) {}
+
+ bool operator==(const Identifier& other) const {
+ return partition == other.partition && bucket == other.bucket &&
level == other.level &&
+ file_name == other.file_name && external_path ==
other.external_path;
+ }
+
+ size_t HashCode() const {
+ if (hash_ == static_cast<size_t>(-1)) {
+ hash_ = partition.HashCode();
+ hash_ =
+ MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const
void*>(&bucket),
+ /*offset=*/0,
sizeof(bucket), /*seed=*/hash_);
+ hash_ =
MurmurHashUtils::HashUnsafeBytes(reinterpret_cast<const void*>(&level),
+ /*offset=*/0,
sizeof(level),
+ /*seed=*/hash_);
+ hash_ = MurmurHashUtils::HashUnsafeBytes(
+ reinterpret_cast<const void*>(file_name.data()),
+ /*offset=*/0, file_name.size(),
+ /*seed=*/hash_);
+ if (external_path) {
+ hash_ = MurmurHashUtils::HashUnsafeBytes(
+ reinterpret_cast<const
void*>(external_path.value().data()),
+ /*offset=*/0, external_path.value().size(),
+ /*seed=*/hash_);
+ }
+ }
+ return hash_;
+ }
+
+ std::string ToString() const {
+ std::string external_path_str =
+ external_path == std::nullopt ? "null" : external_path.value();
+ return fmt::format("{{{}, {}, {}, {}, {}}}", partition.ToString(),
bucket, level,
+ file_name, external_path_str);
+ }
+
+ BinaryRow partition;
+ int32_t bucket;
+ int32_t level;
+ std::string file_name;
+ std::optional<std::string> external_path;
+
+ private:
+ mutable size_t hash_ = -1;
+ };
+
+ public:
+ virtual ~FileEntry() = default;
+ virtual const FileKind& Kind() const = 0;
+ virtual const BinaryRow& Partition() const = 0;
+ virtual int32_t Bucket() const = 0;
+ virtual int32_t Level() const = 0;
+ virtual const std::string& FileName() const = 0;
+ virtual const std::optional<std::string>& ExternalPath() const = 0;
+ virtual Identifier CreateIdentifier() const = 0;
+ virtual const BinaryRow& MinKey() const = 0;
+ virtual const BinaryRow& MaxKey() const = 0;
+
+ template <typename T>
+ static Status MergeEntries(const std::vector<T>& unmerged_entries,
+ std::vector<T>* merged_entries) {
+ LinkedHashMap<Identifier, T> merged_map;
+ PAIMON_RETURN_NOT_OK(MergeEntries(unmerged_entries, &merged_map));
+ for (const auto& [identifier, entry] : merged_map) {
+ merged_entries->emplace_back(entry);
+ }
+ return Status::OK();
+ }
+
+ template <typename T>
+ static Status MergeEntries(const std::vector<T>& unmerged_entries,
+ LinkedHashMap<Identifier, T>* merged_map_ptr) {
+ auto& merged_map = *merged_map_ptr;
+ for (const auto& entry : unmerged_entries) {
+ Identifier identifier = entry.CreateIdentifier();
+ const auto& kind = entry.Kind();
+ auto iter = merged_map.find(identifier);
+ if (kind == FileKind::Add()) {
+ if (iter != merged_map.end()) {
+ return Status::Invalid(fmt::format(
+ "Trying to add file {} which is already added.",
identifier.ToString()));
+ }
+ merged_map.insert(identifier, entry);
+ } else if (kind == FileKind::Delete()) {
+ // each dataFile will only be added once and deleted once,
+ // if we know that it is added before then both add and delete
entry can be
+ // removed because there won't be further operations on this
file,
+ // otherwise we have to keep the delete entry because the add
entry must be
+ // in the previous manifest files
+ if (iter != merged_map.end()) {
+ merged_map.erase(identifier);
+ } else {
+ merged_map.insert(identifier, entry);
+ }
+ } else {
+ return Status::Invalid("Unknown value kind ",
+
std::to_string(static_cast<int32_t>(kind.ToByteValue())));
+ }
+ }
+ return Status::OK();
+ }
+};
+} // namespace paimon
+
+namespace std {
+template <>
+struct hash<paimon::FileEntry::Identifier> {
+ size_t operator()(const paimon::FileEntry::Identifier& identifier) const {
+ return identifier.HashCode();
+ }
+};
+} // namespace std
diff --git a/src/paimon/core/manifest/file_entry_test.cpp
b/src/paimon/core/manifest/file_entry_test.cpp
new file mode 100644
index 0000000..83f5929
--- /dev/null
+++ b/src/paimon/core/manifest/file_entry_test.cpp
@@ -0,0 +1,193 @@
+/*
+ * 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/manifest/file_entry.h"
+
+#include <cassert>
+#include <list>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class MockFileEntry : public FileEntry {
+ public:
+ MockFileEntry(FileKind kind, BinaryRow partition, int32_t bucket, int32_t
level,
+ const std::string& file_name,
+ const std::optional<std::string>& external_path =
std::nullopt)
+ : kind_(kind),
+ partition_(partition),
+ bucket_(bucket),
+ level_(level),
+ file_name_(file_name),
+ external_path_(external_path),
+ min_key_(BinaryRow::EmptyRow()),
+ max_key_(BinaryRow::EmptyRow()) {}
+
+ const FileKind& Kind() const override {
+ return kind_;
+ }
+ const BinaryRow& Partition() const override {
+ return partition_;
+ }
+ int32_t Bucket() const override {
+ return bucket_;
+ }
+ int32_t Level() const override {
+ return level_;
+ }
+ const std::string& FileName() const override {
+ return file_name_;
+ }
+ const std::optional<std::string>& ExternalPath() const override {
+ return external_path_;
+ }
+ Identifier CreateIdentifier() const override {
+ return Identifier(partition_, bucket_, level_, file_name_,
external_path_);
+ }
+ const BinaryRow& MinKey() const override {
+ assert(false);
+ return min_key_;
+ }
+ const BinaryRow& MaxKey() const override {
+ assert(false);
+ return max_key_;
+ }
+
+ private:
+ FileKind kind_;
+ BinaryRow partition_;
+ int32_t bucket_;
+ int32_t level_;
+ std::string file_name_;
+ std::optional<std::string> external_path_;
+ BinaryRow min_key_;
+ BinaryRow max_key_;
+};
+
+class FileEntryTest : public testing::Test {
+ public:
+ void SetUp() override {
+ pool_ = GetDefaultPool();
+ }
+
+ BinaryRow GetPartition(const std::string& part_str) {
+ BinaryRow part(/*arity=*/1);
+ BinaryRowWriter writer(&part, /*initial_size=*/20, pool_.get());
+ writer.WriteString(0, BinaryString::FromString(part_str, pool_.get()));
+ return part;
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_;
+};
+
+TEST_F(FileEntryTest, TestMergeEntriesSimple) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file2");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file3");
+ entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(2u, merged_entries.size());
+ ASSERT_EQ("file2", merged_entries[0].FileName());
+ ASSERT_EQ("file3", merged_entries[1].FileName());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddSameFile) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_NOK_WITH_MSG(FileEntry::MergeEntries(entries, &merged_entries),
+ "which is already added.");
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffPart) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("1"), /*bucket=*/0,
/*level=*/0, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(2u, merged_entries.size());
+ ASSERT_EQ("file1", merged_entries[0].FileName());
+ ASSERT_EQ("file1", merged_entries[1].FileName());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffBucket) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/1,
/*level=*/0, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(2u, merged_entries.size());
+ ASSERT_EQ("file1", merged_entries[0].FileName());
+ ASSERT_EQ("file1", merged_entries[1].FileName());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffLevel) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/1, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(2u, merged_entries.size());
+ ASSERT_EQ("file1", merged_entries[0].FileName());
+ ASSERT_EQ("file1", merged_entries[1].FileName());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddSameFileWithDiffExternalPath) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1",
+ "/tmp/external_path1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1",
+ "/tmp/external_path2");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(2u, merged_entries.size());
+ ASSERT_EQ("file1", merged_entries[0].FileName());
+ ASSERT_EQ("file1", merged_entries[1].FileName());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesAddAndDelete) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_OK(FileEntry::MergeEntries(entries, &merged_entries));
+ ASSERT_EQ(0u, merged_entries.size());
+}
+
+TEST_F(FileEntryTest, TestMergeEntriesDeleteAndAdd) {
+ std::vector<MockFileEntry> entries;
+ entries.emplace_back(FileKind::Delete(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ entries.emplace_back(FileKind::Add(), GetPartition("0"), /*bucket=*/0,
/*level=*/0, "file1");
+ std::vector<MockFileEntry> merged_entries;
+ ASSERT_NOK(FileEntry::MergeEntries(entries, &merged_entries));
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/manifest/file_kind.cpp
b/src/paimon/core/manifest/file_kind.cpp
new file mode 100644
index 0000000..82cfc18
--- /dev/null
+++ b/src/paimon/core/manifest/file_kind.cpp
@@ -0,0 +1,47 @@
+/*
+ * 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/manifest/file_kind.h"
+
+#include "fmt/format.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+const FileKind& FileKind::Add() {
+ static const FileKind file_kind = FileKind(static_cast<int8_t>(0));
+ return file_kind;
+}
+
+const FileKind& FileKind::Delete() {
+ static const FileKind file_kind = FileKind(static_cast<int8_t>(1));
+ return file_kind;
+}
+
+Result<FileKind> FileKind::FromByteValue(int8_t value) {
+ switch (value) {
+ case 0:
+ return Add();
+ case 1:
+ return Delete();
+ default:
+ return Status::Invalid(fmt::format("Unsupported byte value {} for
file kind.",
+ static_cast<int32_t>(value)));
+ }
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/file_kind.h
b/src/paimon/core/manifest/file_kind.h
new file mode 100644
index 0000000..7532893
--- /dev/null
+++ b/src/paimon/core/manifest/file_kind.h
@@ -0,0 +1,46 @@
+/*
+ * 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 "paimon/result.h"
+
+namespace paimon {
+/// Kind of a file.
+class FileKind {
+ public:
+ FileKind() = default;
+ explicit FileKind(int8_t value) : value_(value) {}
+ int8_t ToByteValue() const {
+ return value_;
+ }
+
+ static const FileKind& Add();
+ static const FileKind& Delete();
+
+ static Result<FileKind> FromByteValue(int8_t value);
+
+ bool operator==(const FileKind& other) const {
+ return value_ == other.value_;
+ }
+
+ private:
+ int8_t value_{-1};
+};
+} // namespace paimon
diff --git a/src/paimon/core/manifest/file_kind_test.cpp
b/src/paimon/core/manifest/file_kind_test.cpp
new file mode 100644
index 0000000..190efec
--- /dev/null
+++ b/src/paimon/core/manifest/file_kind_test.cpp
@@ -0,0 +1,40 @@
+/*
+ * 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/manifest/file_kind.h"
+
+#include <memory>
+
+#include "gtest/gtest.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(FileKindTest, FromByteValueValidCases) {
+ ASSERT_OK_AND_ASSIGN(auto add_result, FileKind::FromByteValue(0));
+ ASSERT_EQ(add_result, FileKind::Add());
+
+ ASSERT_OK_AND_ASSIGN(auto delete_result, FileKind::FromByteValue(1));
+ ASSERT_EQ(delete_result, FileKind::Delete());
+}
+
+TEST(FileKindTest, FromByteValueInvalidCase) {
+ ASSERT_NOK_WITH_MSG(FileKind::FromByteValue(2), "Unsupported byte value");
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/manifest/file_source.cpp
b/src/paimon/core/manifest/file_source.cpp
new file mode 100644
index 0000000..1712dd8
--- /dev/null
+++ b/src/paimon/core/manifest/file_source.cpp
@@ -0,0 +1,32 @@
+/*
+ * 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/manifest/file_source.h"
+
+namespace paimon {
+
+const FileSource& FileSource::Append() {
+ static const FileSource file_source = FileSource(static_cast<int8_t>(0));
+ return file_source;
+}
+
+const FileSource& FileSource::Compact() {
+ static const FileSource file_source = FileSource(static_cast<int8_t>(1));
+ return file_source;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/file_source.h
b/src/paimon/core/manifest/file_source.h
new file mode 100644
index 0000000..262b0b8
--- /dev/null
+++ b/src/paimon/core/manifest/file_source.h
@@ -0,0 +1,80 @@
+/*
+ * 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 <string>
+
+#include "fmt/format.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+/// The Source of a file.
+class FileSource {
+ public:
+ /// The file from new input.
+ static const FileSource& Append();
+ /// The file from compaction.
+ static const FileSource& Compact();
+
+ int8_t ToByteValue() const {
+ return value_;
+ }
+
+ static Result<FileSource> FromByteValue(int8_t value) {
+ switch (value) {
+ case 0:
+ return Append();
+ case 1:
+ return Compact();
+ default:
+ return Status::Invalid(
+ fmt::format("Unsupported byte value {} for value kind.",
value));
+ }
+ }
+
+ bool operator==(const FileSource& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return value_ == other.value_;
+ }
+
+ bool operator!=(const FileSource& other) const {
+ return !(*this == other);
+ }
+
+ std::string ToString() const {
+ switch (value_) {
+ case 0:
+ return "APPEND";
+ case 1:
+ return "COMPACT";
+ default:
+ return "UNKNOWN";
+ }
+ }
+
+ private:
+ explicit FileSource(int8_t value) : value_(value) {}
+
+ private:
+ int8_t value_;
+};
+} // namespace paimon
diff --git a/src/paimon/core/manifest/file_source_test.cpp
b/src/paimon/core/manifest/file_source_test.cpp
new file mode 100644
index 0000000..c9011b3
--- /dev/null
+++ b/src/paimon/core/manifest/file_source_test.cpp
@@ -0,0 +1,35 @@
+/*
+ * 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/manifest/file_source.h"
+
+#include "gtest/gtest.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+TEST(FileSourceTest, TestSimple) {
+ ASSERT_OK_AND_ASSIGN(FileSource append, FileSource::FromByteValue(0));
+ ASSERT_OK_AND_ASSIGN(FileSource compact, FileSource::FromByteValue(1));
+ ASSERT_NOK(FileSource::FromByteValue(2));
+ ASSERT_EQ(append.ToString(), "APPEND");
+ ASSERT_EQ(compact.ToString(), "COMPACT");
+ ASSERT_NE(append, compact);
+ ASSERT_EQ(append, append);
+}
+
+} // namespace paimon::test
diff --git a/src/paimon/core/manifest/manifest_entry.cpp
b/src/paimon/core/manifest/manifest_entry.cpp
new file mode 100644
index 0000000..c19e32a
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry.cpp
@@ -0,0 +1,57 @@
+/*
+ * 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/manifest/manifest_entry.h"
+
+#include "arrow/type_fwd.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+const std::shared_ptr<arrow::DataType>& ManifestEntry::DataType() {
+ static std::shared_ptr<arrow::DataType> data_type =
+ arrow::struct_({arrow::field("_KIND", arrow::int8(),
/*nullable=*/false),
+ arrow::field("_PARTITION", arrow::binary(),
/*nullable=*/false),
+ arrow::field("_BUCKET", arrow::int32(),
/*nullable=*/false),
+ arrow::field("_TOTAL_BUCKETS", arrow::int32(),
/*nullable=*/false),
+ arrow::field("_FILE", DataFileMeta::DataType(),
/*nullable=*/false)});
+ return data_type;
+}
+
+int64_t ManifestEntry::RecordCountAdd(const std::vector<ManifestEntry>&
entries) {
+ int64_t record_count = 0;
+ for (const auto& entry : entries) {
+ if (entry.Kind() == FileKind::Add()) {
+ record_count += entry.File()->row_count;
+ }
+ }
+ return record_count;
+}
+
+int64_t ManifestEntry::RecordCountDelete(const std::vector<ManifestEntry>&
entries) {
+ int64_t record_count = 0;
+ for (const auto& entry : entries) {
+ if (entry.Kind() == FileKind::Delete()) {
+ record_count += entry.File()->row_count;
+ }
+ }
+ return record_count;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry.h
b/src/paimon/core/manifest/manifest_entry.h
new file mode 100644
index 0000000..85e387a
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry.h
@@ -0,0 +1,160 @@
+/*
+ * 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 <optional>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_entry.h"
+#include "paimon/core/manifest/file_kind.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+/// Entry of a manifest file, representing an addition / deletion of a data
file.
+class ManifestEntry : public FileEntry {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+ static int64_t RecordCountAdd(const std::vector<ManifestEntry>&
manifest_entries);
+ static int64_t RecordCountDelete(const std::vector<ManifestEntry>&
manifest_entries);
+
+ ManifestEntry(const FileKind& kind, const BinaryRow& partition, int32_t
bucket,
+ int32_t total_buckets, const std::shared_ptr<DataFileMeta>&
file)
+ : kind_(kind),
+ partition_(partition),
+ bucket_(bucket),
+ total_buckets_(total_buckets),
+ file_(file) {}
+
+ ManifestEntry(const ManifestEntry& other) noexcept {
+ *this = other;
+ }
+
+ ManifestEntry& operator=(const ManifestEntry& other) noexcept {
+ if (&other == this) {
+ return *this;
+ }
+ kind_ = other.kind_;
+ partition_ = other.partition_;
+ bucket_ = other.bucket_;
+ total_buckets_ = other.total_buckets_;
+ file_ = other.file_;
+ return *this;
+ }
+
+ ManifestEntry(ManifestEntry&& other) noexcept {
+ *this = std::move(other);
+ }
+
+ ManifestEntry& operator=(ManifestEntry&& other) noexcept {
+ if (&other == this) {
+ return *this;
+ }
+ kind_ = other.kind_;
+ partition_ = other.partition_;
+ bucket_ = other.bucket_;
+ total_buckets_ = other.total_buckets_;
+ file_ = std::move(other.file_);
+ return *this;
+ }
+
+ const FileKind& Kind() const override {
+ return kind_;
+ }
+
+ const BinaryRow& Partition() const override {
+ return partition_;
+ }
+
+ int32_t Bucket() const override {
+ return bucket_;
+ }
+
+ int32_t Level() const override {
+ return file_->level;
+ }
+
+ const std::string& FileName() const override {
+ return file_->file_name;
+ }
+
+ const std::optional<std::string>& ExternalPath() const override {
+ return file_->external_path;
+ }
+
+ const BinaryRow& MinKey() const override {
+ return file_->min_key;
+ }
+
+ const BinaryRow& MaxKey() const override {
+ return file_->max_key;
+ }
+
+ FileEntry::Identifier CreateIdentifier() const override {
+ return FileEntry::Identifier(partition_, bucket_, file_->level,
file_->file_name,
+ file_->external_path);
+ }
+
+ int32_t TotalBuckets() const {
+ return total_buckets_;
+ }
+
+ const std::shared_ptr<DataFileMeta>& File() const {
+ return file_;
+ }
+
+ bool operator==(const ManifestEntry& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return kind_ == other.kind_ && partition_ == other.partition_ &&
bucket_ == other.bucket_ &&
+ total_buckets_ == other.total_buckets_ && *file_ ==
*(other.file_);
+ }
+
+ std::string ToString() const {
+ return fmt::format("{{{}, {}, {}, {}, {}}}",
static_cast<int32_t>(kind_.ToByteValue()),
+ partition_.ToString(), bucket_, total_buckets_,
file_->ToString());
+ }
+
+ void AssignSequenceNumber(int64_t min_sequence_number, int64_t
max_sequence_number) {
+ file_->AssignSequenceNumber(min_sequence_number, max_sequence_number);
+ }
+
+ void AssignFirstRowId(int64_t first_row_id) {
+ file_->AssignFirstRowId(first_row_id);
+ }
+
+ private:
+ FileKind kind_;
+ // for tables without partition this field should be a row with 0 columns
(not null)
+ BinaryRow partition_;
+ int32_t bucket_;
+ int32_t total_buckets_;
+ std::shared_ptr<DataFileMeta> file_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.cpp
b/src/paimon/core/manifest/manifest_entry_serializer.cpp
new file mode 100644
index 0000000..b8176d8
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_serializer.cpp
@@ -0,0 +1,79 @@
+/*
+ * 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/manifest/manifest_entry_serializer.h"
+
+#include <string>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/status.h"
+
+namespace paimon {
+class MemoryPool;
+struct DataFileMeta;
+
+Result<ManifestEntry> ManifestEntrySerializer::ConvertFrom(int32_t version,
+ const InternalRow&
row) const {
+ if (version != VERSION_2) {
+ if (version == VERSION_1) {
+ return Status::Invalid(
+ fmt::format("The current version {} is not compatible with the
version {}, "
+ "please recreate the table.",
+ GetVersion(), version));
+ }
+ return Status::Invalid("Unsupported version", std::to_string(version));
+ }
+ auto kind = row.GetByte(0);
+ PAIMON_ASSIGN_OR_RAISE(FileKind file_kind, FileKind::FromByteValue(kind));
+ auto partition_bytes = row.GetBinary(1);
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow partition,
+
SerializationUtils::DeserializeBinaryRow(partition_bytes));
+ auto bucket = row.GetInt(2);
+ auto total_buckets = row.GetInt(3);
+ auto file = row.GetRow(4, data_file_meta_serializer_.NumFields());
+ if (!file) {
+ return Status::Invalid("ManifestEntry convert from row failed, with
null DataFileMeta");
+ }
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<DataFileMeta> meta,
+ data_file_meta_serializer_.FromRow(*file))
+ return ManifestEntry(file_kind, partition, bucket, total_buckets, meta);
+}
+
+Result<BinaryRow> ManifestEntrySerializer::ToRow(const ManifestEntry& record)
const {
+ BinaryRow row(GetDataType()->num_fields());
+ BinaryRowWriter writer(&row, 32 * 1024, pool_.get());
+
+ writer.WriteInt(0, GetVersion());
+ writer.WriteByte(1, record.Kind().ToByteValue());
+ auto partition_bytes =
SerializationUtils::SerializeBinaryRow(record.Partition(), pool_.get());
+ assert(partition_bytes);
+ writer.WriteBinary(2, *partition_bytes);
+ writer.WriteInt(3, record.Bucket());
+ writer.WriteInt(4, record.TotalBuckets());
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow data_file_meta_row,
+ data_file_meta_serializer_.ToRow(record.File()));
+ writer.WriteRow(5, data_file_meta_row);
+ writer.Complete();
+ return row;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry_serializer.h
b/src/paimon/core/manifest/manifest_entry_serializer.h
new file mode 100644
index 0000000..7438895
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_serializer.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 <cstdint>
+#include <memory>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/core/io/data_file_meta_serializer.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/utils/versioned_object_serializer.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class ArrayBuilder;
+} // namespace arrow
+struct ArrowArray;
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `ManifestEntry`.
+class ManifestEntrySerializer : public
VersionedObjectSerializer<ManifestEntry> {
+ public:
+ explicit ManifestEntrySerializer(const std::shared_ptr<MemoryPool>& pool)
+ : VersionedObjectSerializer<ManifestEntry>(pool),
+ arrow_pool_(GetArrowPool(pool_)),
+ data_file_meta_serializer_(pool) {}
+
+ int32_t GetVersion() const override {
+ return VERSION_2;
+ }
+
+ Result<ManifestEntry> ConvertFrom(int32_t version, const InternalRow& row)
const override;
+
+ Result<BinaryRow> ToRow(const ManifestEntry& record) const override;
+
+ private:
+ static constexpr int32_t VERSION_1 = 1;
+ static constexpr int32_t VERSION_2 = 2;
+ std::unique_ptr<arrow::MemoryPool> arrow_pool_;
+ DataFileMetaSerializer data_file_meta_serializer_;
+};
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
new file mode 100644
index 0000000..3fa2346
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_serializer_test.cpp
@@ -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.
+ */
+#include "paimon/core/manifest/manifest_entry_serializer.h"
+
+#include <optional>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/core/manifest/manifest_entry.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 ManifestEntrySerializerTest : public testing::Test {
+ private:
+ std::shared_ptr<DataFileMeta> GetDataFileMeta() {
+ return std::make_shared<DataFileMeta>(
+ "some_file_name", 1024, 8, DataFileMeta::EmptyMinKey(),
DataFileMeta::EmptyMaxKey(),
+ SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
/*min_seq_no=*/16,
+ /*max_seq_no=*/32,
+ /*schema_id=*/1, /*level=*/2,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/3,
+ /*embedded_index=*/nullptr, /*file_source=*/std::nullopt,
+ /*value_stats_cols=*/std::nullopt,
/*external_path=*/std::optional<std::string>(),
+ /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt);
+ }
+};
+TEST_F(ManifestEntrySerializerTest, TestToFromRow) {
+ auto pool = GetDefaultPool();
+ std::vector<ManifestEntry> entries = {
+ ManifestEntry(FileKind::Add(), BinaryRow::EmptyRow(), 0, 2,
GetDataFileMeta()),
+ ManifestEntry(FileKind::Add(), BinaryRowGenerator::GenerateRow({10},
pool.get()), 1, 2,
+ GetDataFileMeta())};
+ ManifestEntrySerializer serializer(pool);
+ for (const auto& entry : entries) {
+ ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(entry));
+ ASSERT_OK_AND_ASSIGN(auto result_entry, serializer.FromRow(row));
+ ASSERT_EQ(entry, result_entry);
+ ASSERT_EQ(entry.ToString(), result_entry.ToString());
+ }
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/manifest/manifest_entry_writer.cpp
b/src/paimon/core/manifest/manifest_entry_writer.cpp
new file mode 100644
index 0000000..aa25ada
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_writer.cpp
@@ -0,0 +1,80 @@
+/*
+ * 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/manifest/manifest_entry_writer.h"
+
+#include <algorithm>
+#include <cassert>
+#include <utility>
+#include <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/core/stats/simple_stats_converter.h"
+
+namespace paimon {
+class ColumnStats;
+
+Status ManifestEntryWriter::Write(const ManifestEntry& entry) {
+ PAIMON_RETURN_NOT_OK(SingleFileWriter::Write(entry));
+ if (entry.Kind() == FileKind::Add()) {
+ num_added_files_++;
+ } else if (entry.Kind() == FileKind::Delete()) {
+ num_deleted_files_++;
+ } else {
+ return Status::NotImplemented(fmt::format(
+ "Unknown entry kind: {}",
static_cast<int32_t>(entry.Kind().ToByteValue())));
+ }
+ schema_id_ = std::max(schema_id_, entry.File()->schema_id);
+ min_bucket_ = std::min(min_bucket_, entry.Bucket());
+ max_bucket_ = std::max(max_bucket_, entry.Bucket());
+ min_level_ = std::min(min_level_, entry.Level());
+ max_level_ = std::max(max_level_, entry.Level());
+
+ if (row_id_stats_) {
+ std::optional<int64_t> first_row_id = entry.File()->first_row_id;
+ if (first_row_id) {
+ row_id_stats_.value().Collect(first_row_id.value(),
entry.File()->row_count);
+ } else {
+ row_id_stats_ = std::nullopt;
+ }
+ }
+
+ assert(partition_stats_collector_);
+
PAIMON_RETURN_NOT_OK(partition_stats_collector_->Collect(entry.Partition()));
+ return Status::OK();
+}
+
+Result<ManifestFileMeta> ManifestEntryWriter::GetResult() {
+ PAIMON_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<ColumnStats>> col_stats,
+ partition_stats_collector_->GetResult());
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats stats,
+ SimpleStatsConverter::ToBinary(col_stats,
pool_.get()));
+ return ManifestFileMeta(
+ PathUtil::GetName(path_), output_bytes_, num_added_files_,
num_deleted_files_, stats,
+ schema_id_, min_bucket_, max_bucket_, min_level_, max_level_,
+ (row_id_stats_ == std::nullopt ? std::nullopt
+ :
std::optional<int64_t>(row_id_stats_.value().min_row_id)),
+ (row_id_stats_ == std::nullopt ? std::nullopt
+ :
std::optional<int64_t>(row_id_stats_.value().max_row_id)));
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry_writer.h
b/src/paimon/core/manifest/manifest_entry_writer.h
new file mode 100644
index 0000000..04e7090
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_writer.h
@@ -0,0 +1,81 @@
+/*
+ * 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 <algorithm>
+#include <cstdint>
+#include <functional>
+#include <limits>
+#include <memory>
+#include <optional>
+#include <string>
+
+#include "paimon/core/io/single_file_writer.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/manifest/manifest_file_meta.h"
+#include "paimon/core/stats/simple_stats_collector.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class Schema;
+} // namespace arrow
+struct ArrowArray;
+
+namespace paimon {
+class ManifestEntry;
+class MemoryPool;
+
+class ManifestEntryWriter : public SingleFileWriter<const ManifestEntry&,
ManifestFileMeta> {
+ public:
+ ManifestEntryWriter(const std::string& compression,
+ std::function<Status(const ManifestEntry&,
ArrowArray*)> converter,
+ const std::shared_ptr<MemoryPool>& pool,
+ const std::shared_ptr<arrow::Schema>& partition_type)
+ : SingleFileWriter<const ManifestEntry&,
ManifestFileMeta>(compression, converter),
+ pool_(pool),
+
partition_stats_collector_(std::make_shared<SimpleStatsCollector>(partition_type))
{}
+
+ Status Write(const ManifestEntry& entry) override;
+ Result<ManifestFileMeta> GetResult() override;
+
+ private:
+ struct RowIdStats {
+ void Collect(int64_t first_row_id, int64_t row_count) {
+ min_row_id = std::min(min_row_id, first_row_id);
+ max_row_id = std::max(max_row_id, first_row_id + row_count - 1);
+ }
+
+ int64_t min_row_id = std::numeric_limits<int64_t>::max();
+ int64_t max_row_id = std::numeric_limits<int64_t>::min();
+ };
+
+ std::shared_ptr<MemoryPool> pool_;
+ std::shared_ptr<SimpleStatsCollector> partition_stats_collector_;
+
+ int64_t num_added_files_ = 0;
+ int64_t num_deleted_files_ = 0;
+ int64_t schema_id_ = 0;
+ int32_t min_bucket_ = std::numeric_limits<int32_t>::max();
+ int32_t max_bucket_ = std::numeric_limits<int32_t>::min();
+ int32_t min_level_ = std::numeric_limits<int32_t>::max();
+ int32_t max_level_ = std::numeric_limits<int32_t>::min();
+ std::optional<RowIdStats> row_id_stats_ = RowIdStats();
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_entry_writer_test.cpp
b/src/paimon/core/manifest/manifest_entry_writer_test.cpp
new file mode 100644
index 0000000..3c3d00d
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_entry_writer_test.cpp
@@ -0,0 +1,192 @@
+/*
+ * 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/manifest/manifest_entry_writer.h"
+
+#include <map>
+#include <optional>
+#include <utility>
+#include <variant>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/abi.h"
+#include "arrow/c/bridge.h"
+#include "gtest/gtest.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+#include "paimon/core/core_options.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/io/meta_to_arrow_array_converter.h"
+#include "paimon/core/manifest/file_kind.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/manifest/manifest_entry_serializer.h"
+#include "paimon/core/utils/versioned_object_serializer.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/defs.h"
+#include "paimon/format/file_format.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon {
+class WriterBuilder;
+} // namespace paimon
+
+namespace paimon::test {
+class ManifestEntryWriterTest : public ::testing::Test {
+ public:
+ ManifestEntry CreateEntry(const std::optional<int64_t>& first_row_id,
int64_t row_count) const {
+ auto meta = std::make_shared<DataFileMeta>(
+ "data-d7725088-6bd4-4e70-9ce6-714ae93b47cc-0.orc",
/*file_size=*/863, row_count,
+ /*min_key=*/BinaryRow::EmptyRow(),
+ /*max_key=*/BinaryRow::EmptyRow(),
+ /*key_stats=*/
+ SimpleStats::EmptyStats(),
+ /*value_stats=*/SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/row_count,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1743525392885ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, first_row_id,
+ /*write_cols=*/std::nullopt);
+ return {FileKind::Add(), BinaryRowGenerator::GenerateRow({10},
pool_.get()), /*bucket=*/0,
+ /*total_buckets=*/-1, meta};
+ }
+
+ std::unique_ptr<ManifestEntryWriter> CreateEntryWriter(
+ const std::string& entry_file_name) const {
+ std::shared_ptr<arrow::Schema> part_type =
+ arrow::schema(arrow::FieldVector({arrow::field("f1",
arrow::int32())}));
+ auto serializer = std::make_shared<ManifestEntrySerializer>(pool_);
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<MetaToArrowArrayConverter>
to_array_converter,
+
MetaToArrowArrayConverter::Create(serializer->GetDataType(), pool_));
+
+ auto converter = [serializer, to_array_converter](ManifestEntry entry,
+ ::ArrowArray* dest)
-> Status {
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow entry_row,
serializer->ToRow(entry));
+ PAIMON_ASSIGN_OR_RAISE(std::shared_ptr<arrow::Array> array,
+ to_array_converter->NextBatch({entry_row}));
+ PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportArray(*array, dest));
+ return Status::OK();
+ };
+ auto writer = std::make_unique<ManifestEntryWriter>("zstd", converter,
pool_, part_type);
+
+ EXPECT_OK_AND_ASSIGN(CoreOptions options,
+ CoreOptions::FromMap({{Options::FILE_FORMAT,
"orc"}}));
+ auto file_format = options.GetWriteFileFormat(/*level=*/0);
+ std::shared_ptr<arrow::DataType> data_type =
+
VersionedObjectSerializer<ManifestEntry>::VersionType(ManifestEntry::DataType());
+ ArrowSchema arrow_schema;
+ EXPECT_TRUE(arrow::ExportType(*data_type, &arrow_schema).ok());
+ EXPECT_OK_AND_ASSIGN(std::shared_ptr<WriterBuilder> writer_builder,
+ file_format->CreateWriterBuilder(&arrow_schema,
/*batch_size=*/100));
+ EXPECT_OK(writer->Init(options.GetFileSystem(), entry_file_name,
writer_builder));
+ return writer;
+ }
+
+ private:
+ std::shared_ptr<MemoryPool> pool_ = GetDefaultPool();
+};
+
+TEST_F(ManifestEntryWriterTest, TestSimple) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::string entry_file_name = dir->Str() +
"/manifest/my-manifest-file-name";
+ auto writer = CreateEntryWriter(entry_file_name);
+
+ auto meta1 = std::make_shared<DataFileMeta>(
+ "data-d7725088-6bd4-4e70-9ce6-714ae93b47cc-0.orc", /*file_size=*/863,
/*row_count=*/1,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Alice"), 1},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Alice"), 1},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alice"), 1},
{std::string("Alice"), 1},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alice"), 10, 1, 11.1},
+ {std::string("Alice"), 10, 1, 11.1},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/0,
+ /*level=*/4, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1743525392885ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+
+ auto meta2 = std::make_shared<DataFileMeta>(
+ "data-5858a84b-7081-4618-b828-ae3918c5e1f6-0.orc", /*file_size=*/943,
/*row_count=*/4,
+ /*min_key=*/BinaryRowGenerator::GenerateRow({std::string("Alex"), 0},
pool_.get()),
+ /*max_key=*/BinaryRowGenerator::GenerateRow({std::string("Tony"), 0},
pool_.get()),
+ /*key_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 0},
{std::string("Tony"), 0},
+ {0, 0}, pool_.get()),
+ /*value_stats=*/
+ BinaryRowGenerator::GenerateStats({std::string("Alex"), 20, 0, 12.1},
+ {std::string("Tony"), 20, 0, 16.1},
{0, 0, 0, 0},
+ pool_.get()),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/3, /*schema_id=*/0,
+ /*level=*/5, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1743525392921ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+
+ auto entry1 = ManifestEntry(FileKind::Add(),
BinaryRowGenerator::GenerateRow({10}, pool_.get()),
+ 0, 2, meta1);
+ auto entry2 = ManifestEntry(FileKind::Add(),
BinaryRowGenerator::GenerateRow({20}, pool_.get()),
+ 1, 2, meta2);
+ ASSERT_OK(writer->Write(entry1));
+ ASSERT_OK(writer->Write(entry2));
+ ASSERT_EQ(writer->min_bucket_, 0);
+ ASSERT_EQ(writer->max_bucket_, 1);
+ ASSERT_EQ(writer->min_level_, 4);
+ ASSERT_EQ(writer->max_level_, 5);
+
+ // check partition stats
+ ASSERT_OK_AND_ASSIGN(ManifestFileMeta meta, writer->GetResult());
+ auto partition_stats = meta.PartitionStats();
+ ASSERT_EQ(partition_stats, BinaryRowGenerator::GenerateStats({10}, {20},
{0}, pool_.get()));
+}
+
+TEST_F(ManifestEntryWriterTest, TestWithRowIds) {
+ auto dir = UniqueTestDirectory::Create();
+ ASSERT_TRUE(dir);
+ std::string entry_file_name = dir->Str() +
"/manifest/my-manifest-file-name";
+ auto writer = CreateEntryWriter(entry_file_name);
+ // [10, 30)
+ auto entry1 = CreateEntry(/*first_row_id=*/10, /*row_count=*/20);
+ ASSERT_OK(writer->Write(entry1));
+ // [50, 80)
+ auto entry2 = CreateEntry(/*first_row_id=*/50, /*row_count=*/30);
+ ASSERT_OK(writer->Write(entry2));
+ ASSERT_TRUE(writer->row_id_stats_);
+ ASSERT_EQ(writer->row_id_stats_.value().min_row_id, 10);
+ ASSERT_EQ(writer->row_id_stats_.value().max_row_id, 79);
+ // null first row id forces row_id_stats_ to be always null
+ auto entry3 = CreateEntry(/*first_row_id=*/std::nullopt, /*row_count=*/30);
+ ASSERT_OK(writer->Write(entry3));
+ // [100, 110)
+ auto entry4 = CreateEntry(/*first_row_id=*/100, /*row_count=*/10);
+ ASSERT_OK(writer->Write(entry4));
+ ASSERT_FALSE(writer->row_id_stats_);
+
+ ASSERT_OK_AND_ASSIGN(ManifestFileMeta meta, writer->GetResult());
+ ASSERT_FALSE(meta.MinRowId());
+ ASSERT_FALSE(meta.MaxRowId());
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/manifest/manifest_file_meta.cpp
b/src/paimon/core/manifest/manifest_file_meta.cpp
new file mode 100644
index 0000000..4174f24
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_file_meta.cpp
@@ -0,0 +1,101 @@
+/*
+ * 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/manifest/manifest_file_meta.h"
+
+#include "arrow/type_fwd.h"
+#include "fmt/format.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+
+ManifestFileMeta::ManifestFileMeta(
+ const std::string& file_name, int64_t file_size, int64_t num_added_files,
+ int64_t num_deleted_files, const SimpleStats& partition_stats, int64_t
schema_id,
+ const std::optional<int32_t>& min_bucket, const std::optional<int32_t>&
max_bucket,
+ const std::optional<int32_t>& min_level, const std::optional<int32_t>&
max_level,
+ const std::optional<int64_t>& min_row_id, const std::optional<int64_t>&
max_row_id)
+ : file_name_(file_name),
+ file_size_(file_size),
+ num_added_files_(num_added_files),
+ num_deleted_files_(num_deleted_files),
+ partition_stats_(partition_stats),
+ schema_id_(schema_id),
+ min_bucket_(min_bucket),
+ max_bucket_(max_bucket),
+ min_level_(min_level),
+ max_level_(max_level),
+ min_row_id_(min_row_id),
+ max_row_id_(max_row_id) {}
+
+bool ManifestFileMeta::operator==(const ManifestFileMeta& other) const {
+ if (this == &other) {
+ return true;
+ }
+ return file_name_ == other.file_name_ && file_size_ == other.file_size_ &&
+ num_added_files_ == other.num_added_files_ &&
+ num_deleted_files_ == other.num_deleted_files_ &&
+ partition_stats_ == other.partition_stats_ && schema_id_ ==
other.schema_id_ &&
+ min_bucket_ == other.min_bucket_ && max_bucket_ ==
other.max_bucket_ &&
+ min_level_ == other.min_level_ && max_level_ == other.max_level_ &&
+ min_row_id_ == other.min_row_id_ && max_row_id_ ==
other.max_row_id_;
+}
+
+const std::shared_ptr<arrow::DataType>& ManifestFileMeta::DataType() {
+ static std::shared_ptr<arrow::DataType> data_type = arrow::struct_({
+ arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_NUM_ADDED_FILES", arrow::int64(), /*nullable=*/false),
+ arrow::field("_NUM_DELETED_FILES", arrow::int64(), /*nullable=*/false),
+ arrow::field("_PARTITION_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_BUCKET", arrow::int32(), /*nullable=*/true),
+ arrow::field("_MAX_BUCKET", arrow::int32(), /*nullable=*/true),
+ arrow::field("_MIN_LEVEL", arrow::int32(), /*nullable=*/true),
+ arrow::field("_MAX_LEVEL", arrow::int32(), /*nullable=*/true),
+ arrow::field("_MIN_ROW_ID", arrow::int64(), /*nullable=*/true),
+ arrow::field("_MAX_ROW_ID", arrow::int64(), /*nullable=*/true),
+ });
+ return data_type;
+}
+
+std::string ManifestFileMeta::ToString() const {
+ std::string min_bucket_str =
+ min_bucket_ != std::nullopt ? std::to_string(min_bucket_.value()) :
"null";
+ std::string max_bucket_str =
+ max_bucket_ != std::nullopt ? std::to_string(max_bucket_.value()) :
"null";
+
+ std::string min_level_str =
+ min_level_ != std::nullopt ? std::to_string(min_level_.value()) :
"null";
+ std::string max_level_str =
+ max_level_ != std::nullopt ? std::to_string(max_level_.value()) :
"null";
+
+ std::string min_row_id_str =
+ min_row_id_ != std::nullopt ? std::to_string(min_row_id_.value()) :
"null";
+ std::string max_row_id_str =
+ max_row_id_ != std::nullopt ? std::to_string(max_row_id_.value()) :
"null";
+
+ return fmt::format("{{{}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}}}",
file_name_, file_size_,
+ num_added_files_, num_deleted_files_,
partition_stats_.ToString(),
+ schema_id_, min_bucket_str, max_bucket_str,
min_level_str, max_level_str,
+ min_row_id_str, max_row_id_str);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_file_meta.h
b/src/paimon/core/manifest/manifest_file_meta.h
new file mode 100644
index 0000000..0cc0478
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_file_meta.h
@@ -0,0 +1,111 @@
+/*
+ * 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 <optional>
+#include <string>
+
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/visibility.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+
+/// Metadata of a manifest file.
+class PAIMON_EXPORT ManifestFileMeta {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ ManifestFileMeta(const std::string& file_name, int64_t file_size, int64_t
num_added_files,
+ int64_t num_deleted_files, const SimpleStats&
partition_stats,
+ int64_t schema_id, const std::optional<int32_t>&
min_bucket,
+ const std::optional<int32_t>& max_bucket,
+ const std::optional<int32_t>& min_level,
+ const std::optional<int32_t>& max_level,
+ const std::optional<int64_t>& min_row_id,
+ const std::optional<int64_t>& max_row_id);
+
+ const std::string& FileName() const {
+ return file_name_;
+ }
+
+ int64_t FileSize() const {
+ return file_size_;
+ }
+
+ int64_t NumAddedFiles() const {
+ return num_added_files_;
+ }
+
+ int64_t NumDeletedFiles() const {
+ return num_deleted_files_;
+ }
+
+ SimpleStats PartitionStats() const {
+ return partition_stats_;
+ }
+
+ int64_t SchemaId() const {
+ return schema_id_;
+ }
+
+ const std::optional<int32_t>& MinBucket() const {
+ return min_bucket_;
+ }
+ const std::optional<int32_t>& MaxBucket() const {
+ return max_bucket_;
+ }
+ const std::optional<int32_t>& MinLevel() const {
+ return min_level_;
+ }
+ const std::optional<int32_t>& MaxLevel() const {
+ return max_level_;
+ }
+
+ const std::optional<int64_t>& MinRowId() const {
+ return min_row_id_;
+ }
+ const std::optional<int64_t>& MaxRowId() const {
+ return max_row_id_;
+ }
+
+ std::string ToString() const;
+
+ bool operator==(const ManifestFileMeta& other) const;
+
+ private:
+ std::string file_name_;
+ int64_t file_size_;
+ int64_t num_added_files_;
+ int64_t num_deleted_files_;
+ SimpleStats partition_stats_;
+ int64_t schema_id_;
+ std::optional<int32_t> min_bucket_;
+ std::optional<int32_t> max_bucket_;
+ std::optional<int32_t> min_level_;
+ std::optional<int32_t> max_level_;
+ std::optional<int64_t> min_row_id_;
+ std::optional<int64_t> max_row_id_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_file_meta_serializer.cpp
b/src/paimon/core/manifest/manifest_file_meta_serializer.cpp
new file mode 100644
index 0000000..5018aea
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_file_meta_serializer.cpp
@@ -0,0 +1,141 @@
+/*
+ * 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/manifest/manifest_file_meta_serializer.h"
+
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+Result<BinaryRow> ManifestFileMetaSerializer::ToRow(const ManifestFileMeta&
record) const {
+ BinaryRow row(GetDataType()->num_fields());
+ BinaryRowWriter writer(&row, 32 * 1024, pool_.get());
+
+ writer.WriteInt(0, GetVersion());
+ writer.WriteString(1, BinaryString::FromString(record.FileName(),
pool_.get()));
+ writer.WriteLong(2, record.FileSize());
+ writer.WriteLong(3, record.NumAddedFiles());
+ writer.WriteLong(4, record.NumDeletedFiles());
+ writer.WriteRow(5, record.PartitionStats().ToRow());
+ writer.WriteLong(6, record.SchemaId());
+
+ auto min_bucket = record.MinBucket();
+ if (!min_bucket) {
+ writer.SetNullAt(7);
+ } else {
+ writer.WriteInt(7, min_bucket.value());
+ }
+
+ auto max_bucket = record.MaxBucket();
+ if (!max_bucket) {
+ writer.SetNullAt(8);
+ } else {
+ writer.WriteInt(8, max_bucket.value());
+ }
+
+ auto min_level = record.MinLevel();
+ if (!min_level) {
+ writer.SetNullAt(9);
+ } else {
+ writer.WriteInt(9, min_level.value());
+ }
+
+ auto max_level = record.MaxLevel();
+ if (!max_level) {
+ writer.SetNullAt(10);
+ } else {
+ writer.WriteInt(10, max_level.value());
+ }
+ auto min_row_id = record.MinRowId();
+ if (!min_row_id) {
+ writer.SetNullAt(11);
+ } else {
+ writer.WriteInt(11, min_row_id.value());
+ }
+
+ auto max_row_id = record.MaxRowId();
+ if (!max_row_id) {
+ writer.SetNullAt(12);
+ } else {
+ writer.WriteInt(12, max_row_id.value());
+ }
+ writer.Complete();
+ return row;
+}
+
+Result<ManifestFileMeta> ManifestFileMetaSerializer::ConvertFrom(int32_t
version,
+ const
InternalRow& row) const {
+ if (version != VERSION_2) {
+ if (version == VERSION_1) {
+ return Status::Invalid(
+ fmt::format("The current version {} is not compatible with the
"
+ "version {}, please recreate the table.",
+ GetVersion(), version));
+ }
+ return Status::Invalid(fmt::format("Unsupported version: {}",
version));
+ }
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto num_added_files = row.GetLong(2);
+ auto num_deleted_files = row.GetLong(3);
+ auto partition_stats_row = row.GetRow(4, 3);
+ if (partition_stats_row == nullptr) {
+ return Status::Invalid(
+ "ManifestFileMeta convert from row failed, with null partition
stats");
+ }
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats partition_stats,
+ SimpleStats::FromRow(partition_stats_row.get(),
pool_.get()));
+
+ auto schema_id = row.GetLong(5);
+ std::optional<int32_t> min_bucket;
+ if (!row.IsNullAt(6)) {
+ min_bucket = row.GetInt(6);
+ }
+ std::optional<int32_t> max_bucket;
+ if (!row.IsNullAt(7)) {
+ max_bucket = row.GetInt(7);
+ }
+ std::optional<int32_t> min_level;
+ if (!row.IsNullAt(8)) {
+ min_level = row.GetInt(8);
+ }
+ std::optional<int32_t> max_level;
+ if (!row.IsNullAt(9)) {
+ max_level = row.GetInt(9);
+ }
+ std::optional<int64_t> min_row_id;
+ if (!row.IsNullAt(10)) {
+ min_row_id = row.GetLong(10);
+ }
+ std::optional<int64_t> max_row_id;
+ if (!row.IsNullAt(11)) {
+ max_row_id = row.GetLong(11);
+ }
+ return ManifestFileMeta(file_name.ToString(), file_size, num_added_files,
num_deleted_files,
+ partition_stats, schema_id, min_bucket,
max_bucket, min_level,
+ max_level, min_row_id, max_row_id);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_file_meta_serializer.h
b/src/paimon/core/manifest/manifest_file_meta_serializer.h
new file mode 100644
index 0000000..b726065
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_file_meta_serializer.h
@@ -0,0 +1,62 @@
+/*
+ * 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 <vector>
+
+#include "arrow/api.h"
+#include "paimon/common/utils/arrow/mem_utils.h"
+#include "paimon/core/manifest/manifest_entry.h"
+#include "paimon/core/manifest/manifest_file_meta.h"
+#include "paimon/core/utils/versioned_object_serializer.h"
+#include "paimon/result.h"
+
+struct ArrowArray;
+
+namespace arrow {
+class ArrayBuilder;
+} // namespace arrow
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `ManifestFileMeta`.
+class ManifestFileMetaSerializer : public
VersionedObjectSerializer<ManifestFileMeta> {
+ public:
+ explicit ManifestFileMetaSerializer(const std::shared_ptr<MemoryPool>&
pool)
+ : VersionedObjectSerializer<ManifestFileMeta>(pool),
arrow_pool_(GetArrowPool(pool_)) {}
+
+ int32_t GetVersion() const override {
+ return VERSION_2;
+ }
+
+ Result<ManifestFileMeta> ConvertFrom(int32_t version, const InternalRow&
row) const override;
+
+ Result<BinaryRow> ToRow(const ManifestFileMeta& record) const override;
+
+ private:
+ static constexpr int32_t VERSION_2 = 2;
+ static constexpr int32_t VERSION_1 = 1;
+
+ std::unique_ptr<arrow::MemoryPool> arrow_pool_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/manifest/manifest_file_meta_serializer_test.cpp
b/src/paimon/core/manifest/manifest_file_meta_serializer_test.cpp
new file mode 100644
index 0000000..98e79c1
--- /dev/null
+++ b/src/paimon/core/manifest/manifest_file_meta_serializer_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/manifest/manifest_file_meta_serializer.h"
+
+#include <optional>
+#include <string>
+#include <variant>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/data_define.h"
+#include "paimon/common/data/generic_row.h"
+#include "paimon/core/manifest/manifest_file_meta.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/binary_row_generator.h"
+#include "paimon/testing/utils/testharness.h"
+namespace paimon::test {
+TEST(ManifestFileMetaSerializerTest, TestToFromRow) {
+ ManifestFileMeta meta1(/*file_name=*/"meta1", /*file_size=*/10,
/*num_added_files=*/15,
+ /*num_deleted_files=*/20,
SimpleStats::EmptyStats(), /*schema_id=*/0,
+ /*min_bucket=*/std::nullopt,
/*max_bucket=*/std::nullopt,
+ /*min_level=*/std::nullopt,
/*max_level=*/std::nullopt,
+ /*min_row_id=*/std::nullopt,
/*max_row_id=*/std::nullopt);
+ ManifestFileMeta meta2(/*file_name=*/"meta2", /*file_size=*/25,
/*num_added_files=*/30,
+ /*num_deleted_files=*/35,
SimpleStats::EmptyStats(), /*schema_id=*/1,
+ /*min_bucket=*/std::nullopt,
/*max_bucket=*/std::nullopt,
+ /*min_level=*/std::nullopt,
/*max_level=*/std::nullopt,
+ /*min_row_id=*/200, /*max_row_id=*/233);
+
+ std::vector<ManifestFileMeta> metas = {meta1, meta2};
+ ManifestFileMetaSerializer serializer(GetDefaultPool());
+ for (const auto& meta : metas) {
+ ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(meta));
+ ASSERT_OK_AND_ASSIGN(auto result_meta, serializer.FromRow(row));
+ ASSERT_EQ(result_meta, meta);
+ ASSERT_EQ(result_meta.ToString(), meta.ToString());
+ }
+}
+
+TEST(ManifestFileMetaSerializerTest, TestInvalidCase) {
+ auto pool = GetDefaultPool();
+
+ {
+ GenericRow row(12);
+ row.SetField(0, BinaryString::FromString("meta1", pool.get()));
+ row.SetField(1, 10L);
+ row.SetField(2, 20L);
+ row.SetField(3, 30L);
+ auto simple_stats = BinaryRowGenerator::GenerateStats(
+ {"Alex", 10, 0, 12.1}, {"Emily", 10, 0, 16.1}, {0, 0, 0, 0},
pool.get());
+ auto simple_stats_row =
std::make_shared<BinaryRow>(simple_stats.ToRow());
+ row.SetField(4, simple_stats_row);
+ row.SetField(5, 2L);
+ row.SetField(6, NullType());
+ row.SetField(7, NullType());
+ row.SetField(8, NullType());
+ row.SetField(9, NullType());
+ row.SetField(10, NullType());
+ row.SetField(11, NullType());
+
+ ManifestFileMetaSerializer serializer(pool);
+ ASSERT_NOK_WITH_MSG(serializer.ConvertFrom(/*version=*/1, row),
+ "The current version 2 is not compatible with the
version 1, please "
+ "recreate the table.");
+ ASSERT_NOK_WITH_MSG(serializer.ConvertFrom(/*version=*/3, row),
"Unsupported version: 3");
+ ASSERT_OK(serializer.ConvertFrom(/*version=*/2, row));
+ }
+ {
+ GenericRow row(12);
+ row.SetField(0, BinaryString::FromString("meta1", pool.get()));
+ row.SetField(1, 10L);
+ row.SetField(2, 20L);
+ row.SetField(3, 30L);
+ row.SetField(4, std::shared_ptr<BinaryRow>());
+ row.SetField(5, 2L);
+ row.SetField(6, NullType());
+ row.SetField(7, NullType());
+ row.SetField(8, NullType());
+ row.SetField(9, NullType());
+ row.SetField(10, NullType());
+ row.SetField(11, NullType());
+
+ ManifestFileMetaSerializer serializer(pool);
+ ASSERT_NOK_WITH_MSG(serializer.ConvertFrom(/*version=*/2, row),
+ "ManifestFileMeta convert from row failed, with
null partition stats");
+ }
+}
+
+} // namespace paimon::test
