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 6bf64d6 feat(core): add data file meta with versioned serializers and
path factory (#80)
6bf64d6 is described below
commit 6bf64d601769e090fbb73e5bfe90a7bb023283b2
Author: Yonghao Fang <[email protected]>
AuthorDate: Tue Jun 16 09:11:43 2026 +0800
feat(core): add data file meta with versioned serializers and path factory
(#80)
---
src/paimon/core/io/data_file_meta.cpp | 314 +++++++++++++++++++++
src/paimon/core/io/data_file_meta.h | 166 +++++++++++
.../core/io/data_file_meta_09_serializer.cpp | 120 ++++++++
src/paimon/core/io/data_file_meta_09_serializer.h | 45 +++
.../core/io/data_file_meta_10_serializer.cpp | 129 +++++++++
src/paimon/core/io/data_file_meta_10_serializer.h | 49 ++++
.../core/io/data_file_meta_12_serializer.cpp | 134 +++++++++
src/paimon/core/io/data_file_meta_12_serializer.h | 49 ++++
...ta_file_meta_first_row_id_legacy_serializer.cpp | 140 +++++++++
...data_file_meta_first_row_id_legacy_serializer.h | 50 ++++
src/paimon/core/io/data_file_meta_serializer.cpp | 177 ++++++++++++
src/paimon/core/io/data_file_meta_serializer.h | 45 +++
.../core/io/data_file_meta_serializer_test.cpp | 93 ++++++
src/paimon/core/io/data_file_meta_test.cpp | 203 +++++++++++++
src/paimon/core/io/data_file_path_factory.cpp | 86 ++++++
src/paimon/core/io/data_file_path_factory.h | 112 ++++++++
src/paimon/core/io/data_file_path_factory_test.cpp | 145 ++++++++++
17 files changed, 2057 insertions(+)
diff --git a/src/paimon/core/io/data_file_meta.cpp
b/src/paimon/core/io/data_file_meta.cpp
new file mode 100644
index 0000000..86b4d62
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta.cpp
@@ -0,0 +1,314 @@
+/*
+ * 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_meta.h"
+
+#include <algorithm>
+#include <cstddef>
+#include <utility>
+
+#include "arrow/type_fwd.h"
+#include "fmt/format.h"
+#include "fmt/ranges.h"
+#include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/status.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+
+const BinaryRow& DataFileMeta::EmptyMinKey() {
+ static const BinaryRow empty_row = BinaryRow::EmptyRow();
+ return empty_row;
+}
+
+const BinaryRow& DataFileMeta::EmptyMaxKey() {
+ static const BinaryRow empty_row = BinaryRow::EmptyRow();
+ return empty_row;
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta::ForAppend(
+ const std::string& file_name, int64_t file_size, int64_t row_count,
+ const SimpleStats& row_stats, int64_t min_sequence_number, int64_t
max_sequence_number,
+ int64_t schema_id, const std::optional<FileSource>& file_source,
+ const std::optional<std::vector<std::string>>& value_stats_cols,
+ const std::optional<std::string>& external_path, const
std::optional<int64_t>& first_row_id,
+ const std::optional<std::vector<std::string>>& write_cols) {
+ return ForAppend(file_name, file_size, row_count, row_stats,
min_sequence_number,
+ max_sequence_number, schema_id,
std::vector<std::optional<std::string>>(),
+ nullptr, file_source, value_stats_cols, external_path,
first_row_id,
+ write_cols);
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta::ForAppend(
+ const std::string& file_name, int64_t file_size, int64_t row_count,
+ const SimpleStats& row_stats, int64_t min_sequence_number, int64_t
max_sequence_number,
+ int64_t schema_id, const std::vector<std::optional<std::string>>&
extra_files,
+ const std::shared_ptr<Bytes>& embedded_index, const
std::optional<FileSource>& file_source,
+ const std::optional<std::vector<std::string>>& value_stats_cols,
+ const std::optional<std::string>& external_path, const
std::optional<int64_t>& first_row_id,
+ const std::optional<std::vector<std::string>>& write_cols) {
+ PAIMON_ASSIGN_OR_RAISE(int64_t local_micro,
DateTimeUtils::GetCurrentLocalTimeUs());
+ return std::make_shared<DataFileMeta>(
+ file_name, file_size, row_count, EmptyMinKey(), EmptyMaxKey(),
SimpleStats::EmptyStats(),
+ row_stats, min_sequence_number, max_sequence_number, schema_id,
DUMMY_LEVEL, extra_files,
+ Timestamp(/*millisecond=*/local_micro / 1000,
/*nano_of_millisecond=*/0), 0ll,
+ embedded_index, file_source, value_stats_cols, external_path,
first_row_id, write_cols);
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level)
const {
+ if (new_level <= level) {
+ return Status::Invalid(
+ fmt::format("new level {} should be greater than current level
{}", new_level, level));
+ }
+ return std::make_shared<DataFileMeta>(
+ file_name, file_size, row_count, min_key, max_key, key_stats,
value_stats,
+ min_sequence_number, max_sequence_number, schema_id, new_level,
extra_files, creation_time,
+ delete_row_count, embedded_index, file_source, value_stats_cols,
external_path,
+ first_row_id, write_cols);
+}
+
+std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
+ const std::vector<std::optional<std::string>>& new_extra_files) const {
+ return std::make_shared<DataFileMeta>(
+ file_name, file_size, row_count, min_key, max_key, key_stats,
value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
new_extra_files, creation_time,
+ delete_row_count, embedded_index, file_source, value_stats_cols,
external_path,
+ first_row_id, write_cols);
+}
+
+DataFileMeta::DataFileMeta(
+ const std::string& _file_name, int64_t _file_size, int64_t _row_count,
+ const BinaryRow& _min_key, const BinaryRow& _max_key, const SimpleStats&
_key_stats,
+ const SimpleStats& _value_stats, int64_t _min_sequence_number, int64_t
_max_sequence_number,
+ int64_t _schema_id, int32_t _level, const
std::vector<std::optional<std::string>>& _extra_files,
+ const Timestamp& _creation_time, const std::optional<int64_t>&
_delete_row_count,
+ const std::shared_ptr<Bytes>& _embedded_index, const
std::optional<FileSource>& _file_source,
+ const std::optional<std::vector<std::string>>& _value_stats_cols,
+ const std::optional<std::string>& _external_path, const
std::optional<int64_t>& _first_row_id,
+ const std::optional<std::vector<std::string>>& _write_cols)
+ : file_name(_file_name),
+ file_size(_file_size),
+ row_count(_row_count),
+ min_key(_min_key),
+ max_key(_max_key),
+ key_stats(_key_stats),
+ value_stats(_value_stats),
+ min_sequence_number(_min_sequence_number),
+ max_sequence_number(_max_sequence_number),
+ schema_id(_schema_id),
+ level(_level),
+ extra_files(_extra_files),
+ creation_time(_creation_time),
+ delete_row_count(_delete_row_count),
+ embedded_index(_embedded_index),
+ file_source(_file_source),
+ value_stats_cols(_value_stats_cols),
+ external_path(_external_path),
+ first_row_id(_first_row_id),
+ write_cols(_write_cols) {}
+
+Result<std::string> DataFileMeta::FileFormat() const {
+ size_t last_dot_index = file_name.find_last_of(".");
+ if (last_dot_index == std::string::npos || last_dot_index ==
file_name.length() - 1) {
+ return Status::Invalid("cannot find format from file ", file_name);
+ }
+ return file_name.substr(last_dot_index + 1);
+}
+
+std::optional<std::string> DataFileMeta::ExternalPathDir() const {
+ if (!external_path) {
+ return std::nullopt;
+ }
+ return PathUtil::GetParentDirPath(external_path.value());
+}
+
+Result<int64_t> DataFileMeta::CreationTimeEpochMillis() const {
+ PAIMON_ASSIGN_OR_RAISE(Timestamp utc_ts,
DateTimeUtils::ToUTCTimestamp(creation_time));
+ return utc_ts.GetMillisecond();
+}
+
+Result<std::optional<RoaringBitmap32>> DataFileMeta::ToFileSelection(
+ const std::optional<std::vector<Range>>& row_ranges) const {
+ if (!row_ranges) {
+ return std::optional<RoaringBitmap32>();
+ }
+ PAIMON_ASSIGN_OR_RAISE(int64_t start, NonNullFirstRowId());
+ int64_t end = start + row_count - 1;
+ Range file_range(start, end);
+
+ RoaringBitmap32 selection;
+ for (const auto& row_range : row_ranges.value()) {
+ auto intersect_result = Range::Intersection(file_range, row_range);
+ if (intersect_result) {
+
selection.AddRange(static_cast<int32_t>(intersect_result.value().from - start),
+
static_cast<int32_t>(intersect_result.value().to - start + 1));
+ }
+ }
+ if (selection.Cardinality() == row_count) {
+ // If all rows are selected, do not push down selection bitmap.
+ return std::optional<RoaringBitmap32>();
+ }
+ return std::optional<RoaringBitmap32>(selection);
+}
+
+bool DataFileMeta::operator==(const DataFileMeta& other) const {
+ if (this == &other) {
+ return true;
+ }
+ if ((embedded_index && !other.embedded_index) || (!embedded_index &&
other.embedded_index)) {
+ return false;
+ }
+ if (embedded_index && other.embedded_index && !(*embedded_index ==
*other.embedded_index)) {
+ return false;
+ }
+ return file_name == other.file_name && file_size == other.file_size &&
+ row_count == other.row_count && min_key == other.min_key && max_key
== other.max_key &&
+ key_stats == other.key_stats && value_stats == other.value_stats &&
+ min_sequence_number == other.min_sequence_number &&
+ max_sequence_number == other.max_sequence_number && schema_id ==
other.schema_id &&
+ level == other.level && extra_files == other.extra_files &&
+ creation_time == other.creation_time && delete_row_count ==
other.delete_row_count &&
+ file_source == other.file_source && value_stats_cols ==
other.value_stats_cols &&
+ external_path == other.external_path && first_row_id ==
other.first_row_id &&
+ write_cols == other.write_cols;
+}
+
+bool DataFileMeta::operator!=(const DataFileMeta& other) const {
+ return !(*this == other);
+}
+
+bool DataFileMeta::TEST_Equal(const DataFileMeta& other) const {
+ if (this == &other) {
+ return true;
+ }
+ auto compare_optional_ignore_name = [](const std::optional<std::string>&
lhs,
+ const std::optional<std::string>&
rhs) -> bool {
+ if (lhs != rhs) {
+ if (lhs == std::nullopt || rhs == std::nullopt) {
+ return false;
+ }
+ }
+ return true;
+ };
+
+ if (extra_files.size() != other.extra_files.size()) {
+ return false;
+ } else {
+ for (size_t i = 0; i < extra_files.size(); ++i) {
+ if (!compare_optional_ignore_name(extra_files[i],
other.extra_files[i])) {
+ return false;
+ }
+ }
+ }
+
+ if ((embedded_index && !other.embedded_index) || (!embedded_index &&
other.embedded_index)) {
+ return false;
+ }
+ if (embedded_index && other.embedded_index && !(*embedded_index ==
*other.embedded_index)) {
+ return false;
+ }
+ // ignore file_name, file_size, extra_files, creation_time and external
path
+ return row_count == other.row_count && min_key == other.min_key && max_key
== other.max_key &&
+ key_stats == other.key_stats && value_stats == other.value_stats &&
+ min_sequence_number == other.min_sequence_number &&
+ max_sequence_number == other.max_sequence_number && schema_id ==
other.schema_id &&
+ level == other.level && delete_row_count == other.delete_row_count
&&
+ file_source == other.file_source && value_stats_cols ==
other.value_stats_cols &&
+ compare_optional_ignore_name(external_path, other.external_path) &&
+ first_row_id == other.first_row_id && write_cols ==
other.write_cols;
+}
+
+std::string DataFileMeta::ToString() const {
+ std::vector<std::string> extra_files_str;
+ for (const auto& file : extra_files) {
+ if (file == std::nullopt) {
+ extra_files_str.emplace_back("null");
+ } else {
+ extra_files_str.emplace_back(file.value());
+ }
+ }
+
+ return fmt::format(
+ "{{fileName: {}, fileSize: {}, rowCount: {}, embeddedIndex: {},
minKey: {}, maxKey: "
+ "{}, "
+ "keyStats: {}, valueStats: {}, minSequenceNumber: {},
maxSequenceNumber: {}, schemaId: "
+ "{}, level: {}, extraFiles: {}, creationTime: {}, deleteRowCount: {},
fileSource: {}, "
+ "valueStatsCols: {}, externalPath: {}, firstRowId: {}, writeCols:
{}}}",
+ file_name, file_size, row_count,
+ embedded_index == nullptr ? "null"
+ : std::string(embedded_index->data(),
embedded_index->size()),
+ min_key.ToString(), max_key.ToString(), key_stats.ToString(),
value_stats.ToString(),
+ min_sequence_number, max_sequence_number, schema_id, level,
extra_files_str,
+ creation_time.ToString(),
+ delete_row_count == std::nullopt ? "null" :
std::to_string(delete_row_count.value()),
+ file_source == std::nullopt ? "null" : file_source.value().ToString(),
+ value_stats_cols == std::nullopt
+ ? "null"
+ : fmt::format("{}", fmt::join(value_stats_cols.value(), ", ")),
+ external_path == std::nullopt ? "null" : external_path.value(),
+ first_row_id == std::nullopt ? "null" :
std::to_string(first_row_id.value()),
+ write_cols == std::nullopt ? "null" : fmt::format("{}",
write_cols.value()));
+}
+
+int64_t DataFileMeta::GetMaxSequenceNumber(
+ const std::vector<std::shared_ptr<DataFileMeta>>& file_metas) {
+ int64_t ret = -1;
+ for (const auto& meta : file_metas) {
+ ret = std::max(ret, meta->max_sequence_number);
+ }
+ return ret;
+}
+
+const std::shared_ptr<arrow::DataType>& DataFileMeta::DataType() {
+ static std::shared_ptr<arrow::DataType> schema = arrow::struct_(
+ {arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_ROW_COUNT", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_MAX_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_KEY_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_VALUE_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_MIN_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_MAX_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_LEVEL", arrow::int32(), /*nullable=*/false),
+ arrow::field("_EXTRA_FILES",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/false),
+ arrow::field("_CREATION_TIME",
arrow::timestamp(arrow::TimeUnit::MILLI),
+ /*nullable=*/true),
+ arrow::field("_DELETE_ROW_COUNT", arrow::int64(), /*nullable=*/true),
+ arrow::field("_EMBEDDED_FILE_INDEX", arrow::binary(),
/*nullable=*/true),
+ arrow::field("_FILE_SOURCE", arrow::int8(), /*nullable=*/true),
+ arrow::field("_VALUE_STATS_COLS",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/true),
+ arrow::field("_EXTERNAL_PATH", arrow::utf8(), /*nullable=*/true),
+ arrow::field("_FIRST_ROW_ID", arrow::int64(), /*nullable=*/true),
+ arrow::field("_WRITE_COLS",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/true)});
+ return schema;
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta.h
b/src/paimon/core/io/data_file_meta.h
new file mode 100644
index 0000000..443e9c4
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta.h
@@ -0,0 +1,166 @@
+/*
+ * 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 <vector>
+
+#include "fmt/format.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/utils/range.h"
+#include "paimon/utils/roaring_bitmap32.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+class Bytes;
+
+/// Metadata of a data file.
+struct DataFileMeta {
+ static const BinaryRow& EmptyMinKey();
+ static const BinaryRow& EmptyMaxKey();
+ static constexpr int32_t DUMMY_LEVEL = 0;
+
+ DataFileMeta(const std::string& _file_name, int64_t _file_size, int64_t
_row_count,
+ const BinaryRow& _min_key, const BinaryRow& _max_key,
+ const SimpleStats& _key_stats, const SimpleStats&
_value_stats,
+ int64_t _min_sequence_number, int64_t _max_sequence_number,
int64_t _schema_id,
+ int32_t _level, const
std::vector<std::optional<std::string>>& _extra_files,
+ const Timestamp& _creation_time, const
std::optional<int64_t>& _delete_row_count,
+ const std::shared_ptr<Bytes>& _embedded_index,
+ const std::optional<FileSource>& _file_source,
+ const std::optional<std::vector<std::string>>&
_value_stats_cols,
+ const std::optional<std::string>& _external_path,
+ const std::optional<int64_t>& _first_row_id,
+ const std::optional<std::vector<std::string>>& _write_cols);
+
+ static Result<std::shared_ptr<DataFileMeta>> ForAppend(
+ const std::string& file_name, int64_t file_size, int64_t row_count,
+ const SimpleStats& row_stats, int64_t min_sequence_number, int64_t
max_sequence_number,
+ int64_t schema_id, const std::optional<FileSource>& file_source,
+ const std::optional<std::vector<std::string>>& value_stats_cols,
+ const std::optional<std::string>& external_path, const
std::optional<int64_t>& first_row_id,
+ const std::optional<std::vector<std::string>>& write_cols);
+
+ static Result<std::shared_ptr<DataFileMeta>> ForAppend(
+ const std::string& file_name, int64_t file_size, int64_t row_count,
+ const SimpleStats& row_stats, int64_t min_sequence_number, int64_t
max_sequence_number,
+ int64_t schema_id, const std::vector<std::optional<std::string>>&
extra_files,
+ const std::shared_ptr<Bytes>& embedded_index, const
std::optional<FileSource>& file_source,
+ const std::optional<std::vector<std::string>>& value_stats_cols,
+ const std::optional<std::string>& external_path, const
std::optional<int64_t>& first_row_id,
+ const std::optional<std::vector<std::string>>& write_cols);
+
+ Result<std::shared_ptr<DataFileMeta>> Upgrade(int32_t new_level) const;
+
+ /// Create a copy of this DataFileMeta with the given extra files.
+ std::shared_ptr<DataFileMeta> CopyWithExtraFiles(
+ const std::vector<std::optional<std::string>>& new_extra_files) const;
+
+ std::optional<int64_t> AddRowCount() const {
+ return delete_row_count == std::nullopt ? std::optional<int64_t>()
+ : row_count -
delete_row_count.value();
+ }
+
+ Result<int64_t> CreationTimeEpochMillis() const;
+
+ Result<std::string> FileFormat() const;
+ std::optional<std::string> ExternalPathDir() const;
+
+ bool operator==(const DataFileMeta& other) const;
+ bool operator!=(const DataFileMeta& other) const;
+ bool TEST_Equal(const DataFileMeta& other) const;
+ std::string ToString() const;
+
+ void AssignSequenceNumber(int64_t _min_sequence_number, int64_t
_max_sequence_number) {
+ min_sequence_number = _min_sequence_number;
+ max_sequence_number = _max_sequence_number;
+ }
+
+ void AssignFirstRowId(int64_t _first_row_id) {
+ first_row_id = _first_row_id;
+ }
+
+ Result<int64_t> NonNullFirstRowId() const {
+ if (first_row_id) {
+ return first_row_id.value();
+ }
+ return Status::Invalid(fmt::format("First row id of {} should not be
null.", file_name));
+ }
+
+ // empty row_ranges indicates all rows in the file are needed, return null
bitmap
+ Result<std::optional<RoaringBitmap32>> ToFileSelection(
+ const std::optional<std::vector<Range>>& row_ranges) const;
+
+ static int64_t GetMaxSequenceNumber(
+ const std::vector<std::shared_ptr<DataFileMeta>>& file_metas);
+
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ std::string file_name;
+ int64_t file_size;
+ // total number of rows (including add & delete) in this file
+ int64_t row_count;
+ BinaryRow min_key;
+ BinaryRow max_key;
+ SimpleStats key_stats;
+ SimpleStats value_stats;
+
+ int64_t min_sequence_number;
+ int64_t max_sequence_number;
+ int64_t schema_id;
+ int32_t level;
+
+ /// Usage:
+ ///
+ /// Paimon 0.2
+ /// Stores changelog files for `CoreOptions.ChangelogProducer#INPUT`.
Changelog
+ /// files are moved to `DataIncrement` since Paimon 0.3.
+ std::vector<std::optional<std::string>> extra_files;
+ Timestamp creation_time;
+
+ // row_count = add_row_count + delete_row_count
+ // Why don't we keep add_row_count and delete_row_count?
+ // Because in previous versions of DataFileMeta, we only keep row_count.
+ // We have to keep the compatibility.
+ std::optional<int64_t> delete_row_count;
+
+ // file index filter bytes, if it is small, store in data file meta
+ std::shared_ptr<Bytes> embedded_index;
+ std::optional<FileSource> file_source;
+ std::optional<std::vector<std::string>> value_stats_cols;
+
+ // external path of file, if it is null, it is in the default warehouse
path.
+ std::optional<std::string> external_path;
+
+ std::optional<int64_t> first_row_id;
+
+ std::optional<std::vector<std::string>> write_cols;
+};
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_09_serializer.cpp
b/src/paimon/core/io/data_file_meta_09_serializer.cpp
new file mode 100644
index 0000000..7eaffc9
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_09_serializer.cpp
@@ -0,0 +1,120 @@
+/*
+ * 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_meta_09_serializer.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "arrow/api.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/internal_row_utils.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/status.h"
+
+namespace paimon {
+class Bytes;
+class InternalArray;
+
+const std::shared_ptr<arrow::DataType>& DataFileMeta09Serializer::DataType() {
+ static std::shared_ptr<arrow::DataType> schema = arrow::struct_(
+ {arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_ROW_COUNT", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_MAX_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_KEY_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_VALUE_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_MIN_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_MAX_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_LEVEL", arrow::int32(), /*nullable=*/false),
+ arrow::field("_EXTRA_FILES",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/false),
+ arrow::field("_CREATION_TIME",
arrow::timestamp(arrow::TimeUnit::MILLI),
+ /*nullable=*/true),
+ arrow::field("_DELETE_ROW_COUNT", arrow::int64(), /*nullable=*/true),
+ arrow::field("_EMBEDDED_FILE_INDEX", arrow::binary(),
/*nullable=*/true),
+ arrow::field("_FILE_SOURCE", arrow::int8(), /*nullable=*/true)});
+ return schema;
+}
+
+Result<BinaryRow> DataFileMeta09Serializer::ToRow(const
std::shared_ptr<DataFileMeta>& meta) const {
+ assert(false);
+ return Status::Invalid("DataFileMeta09Serializer to row is not valid");
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta09Serializer::FromRow(
+ const InternalRow& row) const {
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto row_count = row.GetLong(2);
+ auto min_key = row.GetBinary(3);
+ auto max_key = row.GetBinary(4);
+ auto key_stats_row = row.GetRow(5, 3);
+ auto value_stats_row = row.GetRow(6, 3);
+ auto min_sequence_number = row.GetLong(7);
+ auto max_sequence_number = row.GetLong(8);
+ auto schema_id = row.GetLong(9);
+ auto level = row.GetInt(10);
+ std::shared_ptr<InternalArray> extra_files = row.GetArray(11);
+ auto creation_time = row.GetTimestamp(12, 3);
+
+ assert(min_key && max_key && key_stats_row && value_stats_row);
+ if (extra_files == nullptr) {
+ return Status::Invalid("extra files is empty");
+ }
+
+ std::optional<int64_t> delete_row_count;
+ if (!row.IsNullAt(13)) {
+ delete_row_count = row.GetLong(13);
+ }
+ std::shared_ptr<Bytes> embedded_file_index;
+ if (!row.IsNullAt(14)) {
+ embedded_file_index = row.GetBinary(14);
+ }
+
+ std::optional<FileSource> file_source;
+ if (!row.IsNullAt(15)) {
+ PAIMON_ASSIGN_OR_RAISE(file_source,
FileSource::FromByteValue(row.GetByte(15)));
+ }
+
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow min_values,
SerializationUtils::DeserializeBinaryRow(min_key));
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow max_values,
SerializationUtils::DeserializeBinaryRow(max_key));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats key_stats,
+ SimpleStats::FromRow(key_stats_row.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats value_stats,
+ SimpleStats::FromRow(value_stats_row.get(),
pool_.get()));
+ return std::make_shared<DataFileMeta>(
+ file_name.ToString(), file_size, row_count, min_values, max_values,
key_stats, value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
+ InternalRowUtils::FromStringArrayData(extra_files.get()),
creation_time, delete_row_count,
+ embedded_file_index, file_source,
+ /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
+ /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_09_serializer.h
b/src/paimon/core/io/data_file_meta_09_serializer.h
new file mode 100644
index 0000000..b80b895
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_09_serializer.h
@@ -0,0 +1,45 @@
+/*
+ * 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 <vector>
+
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/object_serializer.h"
+#include "paimon/result.h"
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `DataFileMeta` with 0.9 version.
+class DataFileMeta09Serializer : public
ObjectSerializer<std::shared_ptr<DataFileMeta>> {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ explicit DataFileMeta09Serializer(const std::shared_ptr<MemoryPool>& pool)
+ : ObjectSerializer<std::shared_ptr<DataFileMeta>>(DataType(), pool) {}
+
+ Result<BinaryRow> ToRow(const std::shared_ptr<DataFileMeta>& meta) const
override;
+ Result<std::shared_ptr<DataFileMeta>> FromRow(const InternalRow& row)
const override;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_10_serializer.cpp
b/src/paimon/core/io/data_file_meta_10_serializer.cpp
new file mode 100644
index 0000000..fea60ee
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_10_serializer.cpp
@@ -0,0 +1,129 @@
+/*
+ * 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_meta_10_serializer.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "arrow/api.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/internal_row_utils.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/status.h"
+
+namespace paimon {
+class Bytes;
+class InternalArray;
+
+const std::shared_ptr<arrow::DataType>& DataFileMeta10Serializer::DataType() {
+ static std::shared_ptr<arrow::DataType> schema = arrow::struct_(
+ {arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_ROW_COUNT", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_MAX_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_KEY_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_VALUE_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_MIN_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_MAX_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_LEVEL", arrow::int32(), /*nullable=*/false),
+ arrow::field("_EXTRA_FILES",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/false),
+ arrow::field("_CREATION_TIME",
arrow::timestamp(arrow::TimeUnit::MILLI),
+ /*nullable=*/true),
+ arrow::field("_DELETE_ROW_COUNT", arrow::int64(), /*nullable=*/true),
+ arrow::field("_EMBEDDED_FILE_INDEX", arrow::binary(),
/*nullable=*/true),
+ arrow::field("_FILE_SOURCE", arrow::int8(), /*nullable=*/true),
+ arrow::field("_VALUE_STATS_COLS",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/true)});
+ return schema;
+}
+
+Result<BinaryRow> DataFileMeta10Serializer::ToRow(const
std::shared_ptr<DataFileMeta>& meta) const {
+ assert(false);
+ return Status::Invalid("to row for data file meta 10 serializer is
invalid");
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta10Serializer::FromRow(
+ const InternalRow& row) const {
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto row_count = row.GetLong(2);
+ auto min_key = row.GetBinary(3);
+ auto max_key = row.GetBinary(4);
+ auto key_stats_row = row.GetRow(5, 3);
+ auto value_stats_row = row.GetRow(6, 3);
+ auto min_sequence_number = row.GetLong(7);
+ auto max_sequence_number = row.GetLong(8);
+ auto schema_id = row.GetLong(9);
+ auto level = row.GetInt(10);
+ std::shared_ptr<InternalArray> extra_files = row.GetArray(11);
+ auto creation_time = row.GetTimestamp(12, 3);
+
+ assert(min_key && max_key && key_stats_row && value_stats_row);
+ if (extra_files == nullptr) {
+ return Status::Invalid("extra files is empty");
+ }
+
+ std::optional<int64_t> delete_row_count;
+ if (!row.IsNullAt(13)) {
+ delete_row_count = row.GetLong(13);
+ }
+ std::shared_ptr<Bytes> embedded_file_index;
+ if (!row.IsNullAt(14)) {
+ embedded_file_index = row.GetBinary(14);
+ }
+
+ std::optional<FileSource> file_source;
+ if (!row.IsNullAt(15)) {
+ PAIMON_ASSIGN_OR_RAISE(file_source,
FileSource::FromByteValue(row.GetByte(15)));
+ }
+
+ std::optional<std::vector<std::string>> value_stats_cols;
+ if (!row.IsNullAt(16)) {
+ std::shared_ptr<InternalArray> array = row.GetArray(16);
+ if (array == nullptr) {
+ return Status::Invalid("invalid value stats cols");
+ }
+ value_stats_cols =
InternalRowUtils::FromNotNullStringArrayData(array.get());
+ }
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow min_values,
SerializationUtils::DeserializeBinaryRow(min_key));
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow max_values,
SerializationUtils::DeserializeBinaryRow(max_key));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats key_stats,
+ SimpleStats::FromRow(key_stats_row.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats value_stats,
+ SimpleStats::FromRow(value_stats_row.get(),
pool_.get()));
+ return std::make_shared<DataFileMeta>(
+ file_name.ToString(), file_size, row_count, min_values, max_values,
key_stats, value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
+ InternalRowUtils::FromStringArrayData(extra_files.get()),
creation_time, delete_row_count,
+ embedded_file_index, file_source,
std::optional<std::vector<std::string>>(value_stats_cols),
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_10_serializer.h
b/src/paimon/core/io/data_file_meta_10_serializer.h
new file mode 100644
index 0000000..236df31
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_10_serializer.h
@@ -0,0 +1,49 @@
+/*
+ * 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 <vector>
+
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/object_serializer.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `DataFileMeta` with 1.0 snapshot version.
+class DataFileMeta10Serializer : public
ObjectSerializer<std::shared_ptr<DataFileMeta>> {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ explicit DataFileMeta10Serializer(const std::shared_ptr<MemoryPool>& pool)
+ : ObjectSerializer<std::shared_ptr<DataFileMeta>>(DataType(), pool) {}
+
+ Result<BinaryRow> ToRow(const std::shared_ptr<DataFileMeta>& meta) const
override;
+ Result<std::shared_ptr<DataFileMeta>> FromRow(const InternalRow& row)
const override;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_12_serializer.cpp
b/src/paimon/core/io/data_file_meta_12_serializer.cpp
new file mode 100644
index 0000000..fe46a85
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_12_serializer.cpp
@@ -0,0 +1,134 @@
+/*
+ * 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_meta_12_serializer.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "arrow/api.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/internal_row_utils.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/status.h"
+namespace paimon {
+class Bytes;
+class InternalArray;
+
+const std::shared_ptr<arrow::DataType>& DataFileMeta12Serializer::DataType() {
+ static std::shared_ptr<arrow::DataType> schema = arrow::struct_(
+ {arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_ROW_COUNT", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_MAX_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_KEY_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_VALUE_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_MIN_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_MAX_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_LEVEL", arrow::int32(), /*nullable=*/false),
+
+ arrow::field("_EXTRA_FILES", arrow::list(arrow::field("item",
arrow::utf8(), false)),
+ /*nullable=*/false),
+ arrow::field("_CREATION_TIME",
arrow::timestamp(arrow::TimeUnit::MILLI),
+ /*nullable=*/true),
+ arrow::field("_DELETE_ROW_COUNT", arrow::int64(), /*nullable=*/true),
+ arrow::field("_EMBEDDED_FILE_INDEX", arrow::binary(),
/*nullable=*/true),
+ arrow::field("_FILE_SOURCE", arrow::int8(), /*nullable=*/true),
+ arrow::field("_VALUE_STATS_COLS",
+ arrow::list(arrow::field("f0", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/true),
+ arrow::field("_EXTERNAL_PATH", arrow::utf8(), /*nullable=*/true)});
+ return schema;
+}
+
+Result<BinaryRow> DataFileMeta12Serializer::ToRow(const
std::shared_ptr<DataFileMeta>& meta) const {
+ assert(false);
+ return Status::Invalid("to row for data file meta 12 serializer is
invalid");
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMeta12Serializer::FromRow(
+ const InternalRow& row) const {
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto row_count = row.GetLong(2);
+ auto min_key = row.GetBinary(3);
+ auto max_key = row.GetBinary(4);
+ auto key_stats_row = row.GetRow(5, 3);
+ auto value_stats_row = row.GetRow(6, 3);
+ auto min_sequence_number = row.GetLong(7);
+ auto max_sequence_number = row.GetLong(8);
+ auto schema_id = row.GetLong(9);
+ auto level = row.GetInt(10);
+ std::shared_ptr<InternalArray> extra_files = row.GetArray(11);
+ auto creation_time = row.GetTimestamp(12, 3);
+
+ assert(min_key && max_key && key_stats_row && value_stats_row);
+ if (extra_files == nullptr) {
+ return Status::Invalid("extra files is empty");
+ }
+
+ std::optional<int64_t> delete_row_count;
+ if (!row.IsNullAt(13)) {
+ delete_row_count = row.GetLong(13);
+ }
+ std::shared_ptr<Bytes> embedded_file_index;
+ if (!row.IsNullAt(14)) {
+ embedded_file_index = row.GetBinary(14);
+ }
+
+ std::optional<FileSource> file_source;
+ if (!row.IsNullAt(15)) {
+ PAIMON_ASSIGN_OR_RAISE(file_source,
FileSource::FromByteValue(row.GetByte(15)));
+ }
+
+ std::optional<std::vector<std::string>> value_stats_cols;
+ if (!row.IsNullAt(16)) {
+ std::shared_ptr<InternalArray> array = row.GetArray(16);
+ if (array == nullptr) {
+ return Status::Invalid("invalid value stats cols");
+ }
+ value_stats_cols =
InternalRowUtils::FromNotNullStringArrayData(array.get());
+ }
+
+ std::optional<std::string> external_path;
+ if (!row.IsNullAt(17)) {
+ external_path = row.GetString(17).ToString();
+ }
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow min_values,
SerializationUtils::DeserializeBinaryRow(min_key));
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow max_values,
SerializationUtils::DeserializeBinaryRow(max_key));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats key_stats,
+ SimpleStats::FromRow(key_stats_row.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats value_stats,
+ SimpleStats::FromRow(value_stats_row.get(),
pool_.get()));
+ return std::make_shared<DataFileMeta>(
+ file_name.ToString(), file_size, row_count, min_values, max_values,
key_stats, value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
+ InternalRowUtils::FromStringArrayData(extra_files.get()),
creation_time, delete_row_count,
+ embedded_file_index, file_source,
std::optional<std::vector<std::string>>(value_stats_cols),
+ external_path, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_12_serializer.h
b/src/paimon/core/io/data_file_meta_12_serializer.h
new file mode 100644
index 0000000..74c29fd
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_12_serializer.h
@@ -0,0 +1,49 @@
+/*
+ * 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 <vector>
+
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/object_serializer.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `DataFileMeta` with 1.2 snapshot version.
+class DataFileMeta12Serializer : public
ObjectSerializer<std::shared_ptr<DataFileMeta>> {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ explicit DataFileMeta12Serializer(const std::shared_ptr<MemoryPool>& pool)
+ : ObjectSerializer<std::shared_ptr<DataFileMeta>>(DataType(), pool) {}
+
+ Result<BinaryRow> ToRow(const std::shared_ptr<DataFileMeta>& meta) const
override;
+ Result<std::shared_ptr<DataFileMeta>> FromRow(const InternalRow& row)
const override;
+};
+
+} // namespace paimon
diff --git
a/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.cpp
b/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.cpp
new file mode 100644
index 0000000..a834e47
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.cpp
@@ -0,0 +1,140 @@
+/*
+ * 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_meta_first_row_id_legacy_serializer.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "arrow/api.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/internal_row_utils.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/status.h"
+
+namespace paimon {
+class Bytes;
+class InternalArray;
+
+const std::shared_ptr<arrow::DataType>&
DataFileMetaFirstRowIdLegacySerializer::DataType() {
+ static std::shared_ptr<arrow::DataType> schema = arrow::struct_(
+ {arrow::field("_FILE_NAME", arrow::utf8(), /*nullable=*/false),
+ arrow::field("_FILE_SIZE", arrow::int64(), /*nullable=*/false),
+ arrow::field("_ROW_COUNT", arrow::int64(), /*nullable=*/false),
+ arrow::field("_MIN_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_MAX_KEY", arrow::binary(), /*nullable=*/false),
+ arrow::field("_KEY_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_VALUE_STATS", SimpleStats::DataType(),
/*nullable=*/false),
+ arrow::field("_MIN_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_MAX_SEQUENCE_NUMBER", arrow::int64(),
/*nullable=*/false),
+ arrow::field("_SCHEMA_ID", arrow::int64(), /*nullable=*/false),
+ arrow::field("_LEVEL", arrow::int32(), /*nullable=*/false),
+ arrow::field("_EXTRA_FILES",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/false),
+ arrow::field("_CREATION_TIME",
arrow::timestamp(arrow::TimeUnit::NANO), /*nullable=*/true),
+ arrow::field("_DELETE_ROW_COUNT", arrow::int64(), /*nullable=*/true),
+ arrow::field("_EMBEDDED_FILE_INDEX", arrow::binary(),
/*nullable=*/true),
+ arrow::field("_FILE_SOURCE", arrow::int8(), /*nullable=*/true),
+ arrow::field("_VALUE_STATS_COLS",
+ arrow::list(arrow::field("item", arrow::utf8(),
/*nullable=*/false)),
+ /*nullable=*/true),
+ arrow::field("_EXTERNAL_PATH", arrow::utf8(), /*nullable=*/true),
+ arrow::field("_FIRST_ROW_ID", arrow::int64(), /*nullable=*/true)});
+ return schema;
+}
+
+Result<BinaryRow> DataFileMetaFirstRowIdLegacySerializer::ToRow(
+ const std::shared_ptr<DataFileMeta>& meta) const {
+ assert(false);
+ return Status::Invalid("to row for data file meta first row id legacy
serializer is invalid");
+}
+
+Result<std::shared_ptr<DataFileMeta>>
DataFileMetaFirstRowIdLegacySerializer::FromRow(
+ const InternalRow& row) const {
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto row_count = row.GetLong(2);
+ auto min_key = row.GetBinary(3);
+ auto max_key = row.GetBinary(4);
+ auto key_stats_row = row.GetRow(5, 3);
+ auto value_stats_row = row.GetRow(6, 3);
+ auto min_sequence_number = row.GetLong(7);
+ auto max_sequence_number = row.GetLong(8);
+ auto schema_id = row.GetLong(9);
+ auto level = row.GetInt(10);
+ std::shared_ptr<InternalArray> extra_files = row.GetArray(11);
+ auto creation_time = row.GetTimestamp(12, 3);
+
+ assert(min_key && max_key && key_stats_row && value_stats_row);
+ if (extra_files == nullptr) {
+ return Status::Invalid("extra files is empty");
+ }
+
+ std::optional<int64_t> delete_row_count;
+ if (!row.IsNullAt(13)) {
+ delete_row_count = row.GetLong(13);
+ }
+ std::shared_ptr<Bytes> embedded_file_index;
+ if (!row.IsNullAt(14)) {
+ embedded_file_index = row.GetBinary(14);
+ }
+
+ std::optional<FileSource> file_source;
+ if (!row.IsNullAt(15)) {
+ PAIMON_ASSIGN_OR_RAISE(file_source,
FileSource::FromByteValue(row.GetByte(15)));
+ }
+
+ std::optional<std::vector<std::string>> value_stats_cols;
+ if (!row.IsNullAt(16)) {
+ std::shared_ptr<InternalArray> array = row.GetArray(16);
+ if (array == nullptr) {
+ return Status::Invalid("invalid value stats cols");
+ }
+ value_stats_cols =
InternalRowUtils::FromNotNullStringArrayData(array.get());
+ }
+
+ std::optional<std::string> external_path;
+ if (!row.IsNullAt(17)) {
+ external_path = row.GetString(17).ToString();
+ }
+ std::optional<int64_t> first_row_id;
+ if (!row.IsNullAt(18)) {
+ first_row_id = row.GetLong(18);
+ }
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow min_values,
SerializationUtils::DeserializeBinaryRow(min_key));
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow max_values,
SerializationUtils::DeserializeBinaryRow(max_key));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats key_stats,
+ SimpleStats::FromRow(key_stats_row.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats value_stats,
+ SimpleStats::FromRow(value_stats_row.get(),
pool_.get()));
+ return std::make_shared<DataFileMeta>(
+ file_name.ToString(), file_size, row_count, min_values, max_values,
key_stats, value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
+ InternalRowUtils::FromStringArrayData(extra_files.get()),
creation_time, delete_row_count,
+ embedded_file_index, file_source,
std::optional<std::vector<std::string>>(value_stats_cols),
+ external_path, first_row_id, /*write_cols=*/std::nullopt);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h
b/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h
new file mode 100644
index 0000000..e5a447c
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.h
@@ -0,0 +1,50 @@
+/*
+ * 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 <vector>
+
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/object_serializer.h"
+#include "paimon/result.h"
+
+namespace arrow {
+class DataType;
+} // namespace arrow
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `DataFileMeta` with first row id as its last field
+class DataFileMetaFirstRowIdLegacySerializer
+ : public ObjectSerializer<std::shared_ptr<DataFileMeta>> {
+ public:
+ static const std::shared_ptr<arrow::DataType>& DataType();
+
+ explicit DataFileMetaFirstRowIdLegacySerializer(const
std::shared_ptr<MemoryPool>& pool)
+ : ObjectSerializer<std::shared_ptr<DataFileMeta>>(DataType(), pool) {}
+
+ Result<BinaryRow> ToRow(const std::shared_ptr<DataFileMeta>& meta) const
override;
+ Result<std::shared_ptr<DataFileMeta>> FromRow(const InternalRow& row)
const override;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_serializer.cpp
b/src/paimon/core/io/data_file_meta_serializer.cpp
new file mode 100644
index 0000000..5602996
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_serializer.cpp
@@ -0,0 +1,177 @@
+/*
+ * 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_meta_serializer.h"
+
+#include <cassert>
+#include <cstdint>
+#include <optional>
+#include <string>
+#include <utility>
+
+#include "paimon/common/data/binary_row_writer.h"
+#include "paimon/common/data/binary_string.h"
+#include "paimon/common/data/internal_row.h"
+#include "paimon/common/utils/internal_row_utils.h"
+#include "paimon/common/utils/serialization_utils.h"
+#include "paimon/core/manifest/file_source.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+class Bytes;
+class InternalArray;
+class MemoryPool;
+
+Result<BinaryRow> DataFileMetaSerializer::ToRow(const
std::shared_ptr<DataFileMeta>& meta) const {
+ BinaryRow row(20);
+ BinaryRowWriter writer(&row, 32 * 1024, pool_.get());
+ writer.WriteString(0, BinaryString::FromString(meta->file_name,
pool_.get()));
+ writer.WriteLong(1, meta->file_size);
+ writer.WriteLong(2, meta->row_count);
+ auto min_key_bytes = SerializationUtils::SerializeBinaryRow(meta->min_key,
pool_.get());
+ writer.WriteBinary(3, *min_key_bytes);
+ auto max_key_bytes = SerializationUtils::SerializeBinaryRow(meta->max_key,
pool_.get());
+ writer.WriteBinary(4, *max_key_bytes);
+ writer.WriteRow(5, meta->key_stats.ToRow());
+ writer.WriteRow(6, meta->value_stats.ToRow());
+ writer.WriteLong(7, meta->min_sequence_number);
+ writer.WriteLong(8, meta->max_sequence_number);
+ writer.WriteLong(9, meta->schema_id);
+ writer.WriteInt(10, meta->level);
+ writer.WriteArray(11,
InternalRowUtils::ToStringArrayData(meta->extra_files, pool_));
+ writer.WriteTimestamp(12, meta->creation_time, 3);
+ if (meta->delete_row_count == std::nullopt) {
+ writer.SetNullAt(13);
+ } else {
+ writer.WriteLong(13, meta->delete_row_count.value());
+ }
+ if (meta->embedded_index == nullptr) {
+ writer.SetNullAt(14);
+ } else {
+ writer.WriteBinary(14, *meta->embedded_index);
+ }
+ if (meta->file_source == std::nullopt) {
+ writer.SetNullAt(15);
+ } else {
+ writer.WriteByte(15, meta->file_source.value().ToByteValue());
+ }
+ if (meta->value_stats_cols == std::nullopt) {
+ writer.SetNullAt(16);
+ } else {
+ writer.WriteArray(
+ 16,
InternalRowUtils::ToNotNullStringArrayData(meta->value_stats_cols.value(),
pool_));
+ }
+ if (meta->external_path == std::nullopt) {
+ writer.SetNullAt(17);
+ } else {
+ writer.WriteString(17,
BinaryString::FromString(meta->external_path.value(), pool_.get()));
+ }
+ if (meta->first_row_id == std::nullopt) {
+ writer.SetNullAt(18);
+ } else {
+ writer.WriteLong(18, meta->first_row_id.value());
+ }
+ if (meta->write_cols == std::nullopt) {
+ writer.SetNullAt(19);
+ } else {
+ writer.WriteArray(
+ 19,
InternalRowUtils::ToNotNullStringArrayData(meta->write_cols.value(), pool_));
+ }
+ writer.Complete();
+ return row;
+}
+
+Result<std::shared_ptr<DataFileMeta>> DataFileMetaSerializer::FromRow(
+ const InternalRow& row) const {
+ auto file_name = row.GetString(0);
+ auto file_size = row.GetLong(1);
+ auto row_count = row.GetLong(2);
+ auto min_key = row.GetBinary(3);
+ auto max_key = row.GetBinary(4);
+ auto key_stats_row = row.GetRow(5, 3);
+ auto value_stats_row = row.GetRow(6, 3);
+ auto min_sequence_number = row.GetLong(7);
+ auto max_sequence_number = row.GetLong(8);
+ auto schema_id = row.GetLong(9);
+ auto level = row.GetInt(10);
+ std::shared_ptr<InternalArray> extra_files = row.GetArray(11);
+ auto creation_time = row.GetTimestamp(12, 3);
+
+ assert(min_key && max_key && key_stats_row && value_stats_row);
+ if (extra_files == nullptr) {
+ return Status::Invalid("extra files is empty");
+ }
+
+ std::optional<int64_t> delete_row_count;
+ if (!row.IsNullAt(13)) {
+ delete_row_count = row.GetLong(13);
+ }
+ std::shared_ptr<Bytes> embedded_file_index;
+ if (!row.IsNullAt(14)) {
+ embedded_file_index = row.GetBinary(14);
+ }
+
+ std::optional<FileSource> file_source;
+ if (!row.IsNullAt(15)) {
+ PAIMON_ASSIGN_OR_RAISE(file_source,
FileSource::FromByteValue(row.GetByte(15)));
+ }
+
+ std::optional<std::vector<std::string>> value_stats_cols;
+ if (!row.IsNullAt(16)) {
+ std::shared_ptr<InternalArray> array = row.GetArray(16);
+ if (array == nullptr) {
+ return Status::Invalid("invalid value stats cols");
+ }
+ value_stats_cols =
InternalRowUtils::FromNotNullStringArrayData(array.get());
+ }
+
+ std::optional<std::string> external_path;
+ if (!row.IsNullAt(17)) {
+ external_path = row.GetString(17).ToString();
+ }
+ std::optional<int64_t> first_row_id;
+ if (!row.IsNullAt(18)) {
+ first_row_id = row.GetLong(18);
+ }
+
+ std::optional<std::vector<std::string>> write_cols;
+ if (!row.IsNullAt(19)) {
+ std::shared_ptr<InternalArray> array = row.GetArray(19);
+ if (array == nullptr) {
+ return Status::Invalid("invalid write cols");
+ }
+ write_cols = InternalRowUtils::FromNotNullStringArrayData(array.get());
+ }
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow min_values,
SerializationUtils::DeserializeBinaryRow(min_key));
+ PAIMON_ASSIGN_OR_RAISE(BinaryRow max_values,
SerializationUtils::DeserializeBinaryRow(max_key));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats key_stats,
+ SimpleStats::FromRow(key_stats_row.get(),
pool_.get()));
+ PAIMON_ASSIGN_OR_RAISE(SimpleStats value_stats,
+ SimpleStats::FromRow(value_stats_row.get(),
pool_.get()));
+ return std::make_shared<DataFileMeta>(
+ file_name.ToString(), file_size, row_count, min_values, max_values,
key_stats, value_stats,
+ min_sequence_number, max_sequence_number, schema_id, level,
+ InternalRowUtils::FromStringArrayData(extra_files.get()),
creation_time, delete_row_count,
+ embedded_file_index, file_source,
std::optional<std::vector<std::string>>(value_stats_cols),
+ external_path, first_row_id, write_cols);
+}
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_serializer.h
b/src/paimon/core/io/data_file_meta_serializer.h
new file mode 100644
index 0000000..5713df1
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_serializer.h
@@ -0,0 +1,45 @@
+/*
+ * 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 <vector>
+
+#include "arrow/status.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/object_serializer.h"
+#include "paimon/result.h"
+
+namespace paimon {
+class InternalRow;
+class MemoryPool;
+
+/// Serializer for `DataFileMeta`.
+class DataFileMetaSerializer : public
ObjectSerializer<std::shared_ptr<DataFileMeta>> {
+ public:
+ explicit DataFileMetaSerializer(const std::shared_ptr<MemoryPool>& pool)
+ :
ObjectSerializer<std::shared_ptr<DataFileMeta>>(DataFileMeta::DataType(), pool)
{}
+
+ Result<BinaryRow> ToRow(const std::shared_ptr<DataFileMeta>& meta) const
override;
+
+ Result<std::shared_ptr<DataFileMeta>> FromRow(const InternalRow& row)
const override;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_meta_serializer_test.cpp
b/src/paimon/core/io/data_file_meta_serializer_test.cpp
new file mode 100644
index 0000000..18ba257
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_serializer_test.cpp
@@ -0,0 +1,93 @@
+/*
+ * 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_meta_serializer.h"
+
+#include <cstdint>
+#include <optional>
+#include <string>
+
+#include "arrow/api.h"
+#include "arrow/array/builder_base.h"
+#include "gtest/gtest.h"
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/stats/simple_stats.h"
+#include "paimon/data/timestamp.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace arrow {
+class Array;
+} // namespace arrow
+
+namespace paimon::test {
+class DataFileMetaSerializerTest : public testing::Test {
+ public:
+ void SetUp() override {
+ memory_pool_ = GetDefaultPool();
+ }
+
+ 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);
+ }
+
+ const int32_t TRIES = 100;
+
+ std::shared_ptr<MemoryPool> memory_pool_;
+};
+
+TEST_F(DataFileMetaSerializerTest, TestToFromRow) {
+ DataFileMetaSerializer serializer(memory_pool_);
+ auto expected = GetDataFileMeta();
+ for (int32_t i = 0; i < TRIES; i++) {
+ ASSERT_OK_AND_ASSIGN(auto row, serializer.ToRow(expected));
+ ASSERT_OK_AND_ASSIGN(auto actual, serializer.FromRow(row));
+ ASSERT_EQ(expected->ToString(), actual->ToString());
+ }
+}
+
+TEST_F(DataFileMetaSerializerTest, TestSerialize) {
+ DataFileMetaSerializer serializer(memory_pool_);
+ auto expected = GetDataFileMeta();
+ for (int32_t i = 0; i < TRIES; i++) {
+ MemorySegmentOutputStream out(1024, memory_pool_);
+ ASSERT_OK(serializer.Serialize(expected, &out));
+ PAIMON_UNIQUE_PTR<Bytes> bytes = MemorySegmentUtils::CopyToBytes(
+ out.Segments(), 0, out.CurrentSize(), memory_pool_.get());
+ auto input_stream =
std::make_shared<ByteArrayInputStream>(bytes->data(), bytes->size());
+ DataInputStream in(input_stream);
+ ASSERT_OK_AND_ASSIGN(auto actual, serializer.Deserialize(&in));
+ ASSERT_EQ(expected->ToString(), actual->ToString());
+ }
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/io/data_file_meta_test.cpp
b/src/paimon/core/io/data_file_meta_test.cpp
new file mode 100644
index 0000000..78b27b8
--- /dev/null
+++ b/src/paimon/core/io/data_file_meta_test.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/data_file_meta.h"
+
+#include "gtest/gtest.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+TEST(DataFileMetaTest, TestAddRowCount) {
+ DataFileMeta file_meta("data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc",
/*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(),
BinaryRow::EmptyRow(),
+ SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0,
/*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/{},
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt,
/*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ ASSERT_EQ(3, file_meta.AddRowCount().value());
+ // test null delete row count
+ file_meta.delete_row_count = std::nullopt;
+ ASSERT_EQ(std::nullopt, file_meta.AddRowCount());
+}
+
+TEST(DataFileMetaTest, TestFileFormat) {
+ DataFileMeta file_meta("data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc",
/*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(),
BinaryRow::EmptyRow(),
+ SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0,
/*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/{},
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt,
/*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ ASSERT_OK_AND_ASSIGN(auto file_format, file_meta.FileFormat());
+ ASSERT_EQ("orc", file_format);
+ file_meta.file_name =
"data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.parquet";
+ ASSERT_OK_AND_ASSIGN(file_format, file_meta.FileFormat());
+ ASSERT_EQ("parquet", file_format);
+ // test invalid data file name
+ file_meta.file_name = "data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0";
+ ASSERT_NOK_WITH_MSG(file_meta.FileFormat(),
+ "cannot find format from file
data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0");
+}
+
+TEST(DataFileMetaTest, TestExternalPathDir) {
+ DataFileMeta file_meta(
+ "data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc", /*file_size=*/645,
/*row_count=*/5,
+ BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
SimpleStats::EmptyStats(),
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/{},
/*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+
/*external_path=*/"file:/tmp/bucket-0/data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc",
+ /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt);
+ ASSERT_EQ("file:/tmp/bucket-0", file_meta.ExternalPathDir().value());
+ file_meta.external_path = std::nullopt;
+ ASSERT_EQ(std::nullopt, file_meta.ExternalPathDir());
+}
+
+TEST(DataFileMetaTest, TestGetMaxSequenceNumber) {
+ auto file_meta1 = std::make_shared<DataFileMeta>(
+ "data-80110e15-97b5-4bcf-ac09-6ca2659a4950-0.orc", /*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
SimpleStats::EmptyStats(),
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+ auto file_meta2 = std::make_shared<DataFileMeta>(
+ "data-80110e15-97b5-4bcf-ac09-6ca2659a4950-1.orc", /*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
SimpleStats::EmptyStats(),
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/5, /*max_sequence_number=*/10, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
/*write_cols=*/std::nullopt);
+ ASSERT_EQ(4, DataFileMeta::GetMaxSequenceNumber({file_meta1}));
+ ASSERT_EQ(10, DataFileMeta::GetMaxSequenceNumber({file_meta1,
file_meta2}));
+ ASSERT_EQ(-1, DataFileMeta::GetMaxSequenceNumber({}));
+ ASSERT_EQ(file_meta1, file_meta1);
+ ASSERT_NE(file_meta1, file_meta2);
+}
+
+TEST(DataFileMetaTest, TestNonNullFirstRowId) {
+ {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-0.orc", /*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
+ SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/100,
/*write_cols=*/std::nullopt);
+ ASSERT_OK_AND_ASSIGN(int64_t first_row_id,
file_meta->NonNullFirstRowId());
+ ASSERT_EQ(100, first_row_id);
+ }
+ {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-1.orc", /*file_size=*/645,
+ /*row_count=*/5, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
+ SimpleStats::EmptyStats(), SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4,
/*schema_id=*/0,
+ /*level=*/0,
/*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/2, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ ASSERT_NOK_WITH_MSG(file_meta->NonNullFirstRowId(),
+ "First row id of data-1.orc should not be null.");
+ }
+}
+
+TEST(DataFileMetaTest, TestToFileSelection) {
+ // row id range [100, 110)
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-0.orc", /*file_size=*/645,
+ /*row_count=*/10, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
SimpleStats::EmptyStats(),
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/100, /*max_sequence_number=*/109,
/*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/100,
/*write_cols=*/std::nullopt);
+
+ {
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap32> result,
+ file_meta->ToFileSelection(std::nullopt));
+ ASSERT_FALSE(result);
+ }
+ {
+ std::vector<Range> ranges = {Range(10l, 20l), Range(105l, 107l),
Range(500l, 520l)};
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap32> result,
+ file_meta->ToFileSelection(ranges));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result.value().ToString(), "{5,6,7}");
+ }
+ {
+ std::vector<Range> ranges = {};
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap32> result,
+ file_meta->ToFileSelection(ranges));
+ ASSERT_TRUE(result);
+ ASSERT_EQ(result.value().ToString(), "{}");
+ }
+ {
+ std::vector<Range> ranges = {Range(100l, 109l)};
+ ASSERT_OK_AND_ASSIGN(std::optional<RoaringBitmap32> result,
+ file_meta->ToFileSelection(ranges));
+ ASSERT_FALSE(result);
+ }
+}
+
+TEST(DataFileMetaTest, TestUpgrade) {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-0.orc", /*file_size=*/645,
+ /*row_count=*/10, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
SimpleStats::EmptyStats(),
+ SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/100, /*max_sequence_number=*/109,
/*schema_id=*/0,
+ /*level=*/5, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/std::nullopt, /*first_row_id=*/100,
/*write_cols=*/std::nullopt);
+ // test normal upgrade
+ ASSERT_OK_AND_ASSIGN(auto new_file_meta, file_meta->Upgrade(10));
+ ASSERT_EQ(new_file_meta->level, 10);
+ // check other members
+ file_meta->level = 10;
+ ASSERT_EQ(*new_file_meta, *file_meta);
+
+ // test invalid upgrade
+ ASSERT_NOK_WITH_MSG(file_meta->Upgrade(1),
+ "new level 1 should be greater than current level 10");
+}
+} // namespace paimon::test
diff --git a/src/paimon/core/io/data_file_path_factory.cpp
b/src/paimon/core/io/data_file_path_factory.cpp
new file mode 100644
index 0000000..1721436
--- /dev/null
+++ b/src/paimon/core/io/data_file_path_factory.cpp
@@ -0,0 +1,86 @@
+/*
+ * 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_path_factory.h"
+
+#include <optional>
+#include <utility>
+
+#include "paimon/common/utils/path_util.h"
+#include "paimon/common/utils/uuid.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/macros.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+const char DataFilePathFactory::CHANGELOG_FILE_PREFIX[] = "changelog-";
+const char DataFilePathFactory::INDEX_PATH_SUFFIX[] = ".index";
+
+Status DataFilePathFactory::Init(const std::string& parent, const std::string&
format_identifier,
+ const std::string& data_file_prefix,
+ std::unique_ptr<ExternalPathProvider>&&
external_path_provider) {
+ if (PAIMON_UNLIKELY(!UUID::Generate(&uuid_))) {
+ return Status::Invalid("fail to generate uuid for data file path
factory");
+ }
+ parent_ = parent;
+ path_count_.store(0);
+ format_identifier_ = format_identifier;
+ data_file_prefix_ = data_file_prefix;
+ external_path_provider_ = std::move(external_path_provider);
+ return Status::OK();
+}
+
+std::string DataFilePathFactory::NewPath(const std::string& prefix) const {
+ return NewPathFromName(NewFileName(prefix));
+}
+
+std::string DataFilePathFactory::ToPath(const std::string& file_name) const {
+ return PathUtil::JoinPath(parent_, file_name);
+}
+
+std::string DataFilePathFactory::ToPath(const std::shared_ptr<DataFileMeta>&
file_meta) const {
+ if (file_meta->external_path) {
+ return file_meta->external_path.value();
+ }
+ return PathUtil::JoinPath(parent_, file_meta->file_name);
+}
+
+std::string DataFilePathFactory::ToFileIndexPath(const std::string& file_path)
const {
+ std::string parent = PathUtil::GetParentDirPath(file_path);
+ return PathUtil::JoinPath(parent, PathUtil::GetName(file_path) +
INDEX_PATH_SUFFIX);
+}
+
+std::string DataFilePathFactory::ToAlignedPath(const std::string& file_name,
+ const
std::shared_ptr<DataFileMeta>& aligned) const {
+ auto external_path = aligned->ExternalPathDir();
+ return PathUtil::JoinPath(external_path ? external_path.value() : parent_,
file_name);
+}
+
+std::vector<std::string> DataFilePathFactory::CollectFiles(
+ const std::shared_ptr<DataFileMeta>& file_meta) const {
+ std::vector<std::string> paths;
+ paths.push_back(ToPath(file_meta));
+ for (const auto& extra_file : file_meta->extra_files) {
+ if (extra_file) {
+ paths.push_back(ToAlignedPath(extra_file.value(), file_meta));
+ }
+ }
+ return paths;
+}
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_path_factory.h
b/src/paimon/core/io/data_file_path_factory.h
new file mode 100644
index 0000000..b49154f
--- /dev/null
+++ b/src/paimon/core/io/data_file_path_factory.h
@@ -0,0 +1,112 @@
+/*
+ * 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 <atomic>
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/common/fs/external_path_provider.h"
+#include "paimon/common/utils/path_util.h"
+#include "paimon/core/io/data_file_meta.h"
+#include "paimon/core/utils/path_factory.h"
+
+namespace paimon {
+
+class Status;
+struct DataFileMeta;
+
+/// %Factory which produces new paths and converts paths for data files.
+class DataFilePathFactory : public PathFactory {
+ public:
+ static const char CHANGELOG_FILE_PREFIX[];
+ static const char INDEX_PATH_SUFFIX[];
+
+ Status Init(const std::string& parent, const std::string&
format_identifier,
+ const std::string& data_file_prefix,
+ std::unique_ptr<ExternalPathProvider>&&
external_path_provider);
+
+ const std::string& Parent() const {
+ return parent_;
+ }
+
+ const std::string& DataFilePrefix() const {
+ return data_file_prefix_;
+ }
+
+ std::string NewPath() const override {
+ return NewPath(data_file_prefix_);
+ }
+
+ std::string NewChangelogPath() const {
+ return NewPath(std::string(CHANGELOG_FILE_PREFIX));
+ }
+
+ std::string NewBlobPath() const {
+ return NewPathFromName(NewFileName(data_file_prefix_, ".blob"));
+ }
+
+ std::string NewPathFromName(const std::string& file_name) const {
+ if (external_path_provider_ != nullptr) {
+ return external_path_provider_->GetNextExternalDataPath(file_name);
+ }
+ return PathUtil::JoinPath(parent_, file_name);
+ }
+
+ std::string ToPath(const std::string& file_name) const override;
+ std::string ToPath(const std::shared_ptr<DataFileMeta>& file_meta) const;
+
+ const std::string& GetUUID() const {
+ return uuid_;
+ }
+
+ std::string ToFileIndexPath(const std::string& file_path) const;
+ std::string ToAlignedPath(const std::string& file_name,
+ const std::shared_ptr<DataFileMeta>& aligned)
const;
+
+ std::vector<std::string> CollectFiles(const std::shared_ptr<DataFileMeta>&
file_meta) const;
+ bool IsExternalPath() const {
+ return external_path_provider_ != nullptr;
+ }
+
+ private:
+ std::string NewPath(const std::string& prefix) const;
+
+ std::string NewFileName(const std::string& prefix) const {
+ // TODO(yonghao.fyh): add compress extension as java paimon if needed
+ std::string extension = "." + format_identifier_;
+ return NewFileName(prefix, extension);
+ }
+
+ std::string NewFileName(const std::string& prefix, const std::string&
extension) const {
+ return prefix + uuid_ + "-" + std::to_string(path_count_.fetch_add(1))
+ extension;
+ }
+
+ private:
+ std::string parent_;
+ std::string uuid_;
+ mutable std::atomic<int32_t> path_count_;
+ std::string format_identifier_;
+ std::string data_file_prefix_;
+ std::unique_ptr<ExternalPathProvider> external_path_provider_;
+};
+
+} // namespace paimon
diff --git a/src/paimon/core/io/data_file_path_factory_test.cpp
b/src/paimon/core/io/data_file_path_factory_test.cpp
new file mode 100644
index 0000000..293f14a
--- /dev/null
+++ b/src/paimon/core/io/data_file_path_factory_test.cpp
@@ -0,0 +1,145 @@
+/*
+ * 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_path_factory.h"
+
+#include <optional>
+#include <utility>
+
+#include "gtest/gtest.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/common/fs/external_path_provider.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/result.h"
+#include "paimon/status.h"
+#include "paimon/testing/utils/testharness.h"
+
+namespace paimon::test {
+
+class DataFilePathFactoryTest : public ::testing::Test {
+ protected:
+ void SetUp() override {
+ // Initialize the DataFilePathFactory with a parent directory and
format identifier
+ ASSERT_OK(factory_.Init(/*parent=*/"/tmp", /*format_identifier=*/"txt",
+ /*data_file_prefix=*/"data-",
/*external_path_provider=*/nullptr));
+ }
+
+ DataFilePathFactory factory_;
+};
+
+TEST_F(DataFilePathFactoryTest, TestNewPath) {
+ std::string path1 = factory_.NewPath();
+ std::string path2 = factory_.NewPath();
+
+ // Ensure that the paths are unique
+ ASSERT_NE(path1, path2);
+ ASSERT_TRUE(path1.find("/tmp/data-") != std::string::npos);
+ ASSERT_TRUE(path2.find("/tmp/data-") != std::string::npos);
+ // test Parent() and NewPathFromName()
+ ASSERT_EQ(factory_.Parent(), "/tmp");
+ ASSERT_EQ(factory_.NewPathFromName("index-file"), "/tmp/index-file");
+}
+
+TEST_F(DataFilePathFactoryTest, TestNewPathWithDataFilePrefixAndExternalPath) {
+ DataFilePathFactory factory;
+ ASSERT_OK_AND_ASSIGN(
+ std::unique_ptr<ExternalPathProvider> external_path_provider,
+ ExternalPathProvider::Create({"/tmp/external_path/"},
"p0=1/p1=0/bucket-0"));
+
+ ASSERT_OK(factory_.Init(/*parent=*/"/tmp/p0=1/p1=0/bucket-0/",
/*format_identifier=*/"txt",
+ /*data_file_prefix=*/"test-data-",
std::move(external_path_provider)));
+ std::string path1 = factory_.NewPath();
+ std::string path2 = factory_.NewPath();
+
+ // Ensure that the paths are unique
+ ASSERT_NE(path1, path2);
+ ASSERT_TRUE(path1.find("/tmp/external_path/p0=1/p1=0/bucket-0/test-data-")
!=
+ std::string::npos);
+ ASSERT_TRUE(path2.find("/tmp/external_path/p0=1/p1=0/bucket-0/test-data-")
!=
+ std::string::npos);
+
+ // test Parent() and NewPathFromName()
+ ASSERT_EQ(factory_.Parent(), "/tmp/p0=1/p1=0/bucket-0/");
+ ASSERT_EQ(factory_.NewPathFromName("index-file"),
+ "/tmp/external_path/p0=1/p1=0/bucket-0/index-file");
+}
+
+TEST_F(DataFilePathFactoryTest, TestToPath) {
+ std::string file_name = "example.txt";
+
+ ASSERT_EQ(factory_.ToPath(file_name), "/tmp/example.txt");
+
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "example.txt", /*file_size=*/645, /*row_count=*/5,
BinaryRow::EmptyRow(),
+ BinaryRow::EmptyRow(), SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/"file:/test/bucket-0/example.txt",
/*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+ ASSERT_EQ(factory_.ToPath(file_meta), "file:/test/bucket-0/example.txt");
+}
+
+TEST_F(DataFilePathFactoryTest, TestToFileIndexPath) {
+ std::string file_path = "/tmp/example.txt";
+ std::string index_path = factory_.ToFileIndexPath(file_path);
+
+ ASSERT_EQ(index_path, "/tmp/example.txt.index");
+}
+
+TEST_F(DataFilePathFactoryTest, TestToAlignedPath) {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-0.txt", /*file_size=*/645, /*row_count=*/5,
BinaryRow::EmptyRow(),
+ BinaryRow::EmptyRow(), SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 0),
+ /*delete_row_count=*/0, /*embedded_index=*/nullptr,
FileSource::Append(),
+ /*value_stats_cols=*/std::nullopt,
+ /*external_path=*/"file:/test/bucket-0/data-0.txt",
/*first_row_id=*/std::nullopt,
+ /*write_cols=*/std::nullopt);
+
+ ASSERT_EQ(factory_.ToAlignedPath("index-0", file_meta),
"file:/test/bucket-0/index-0");
+
+ file_meta->external_path = std::nullopt;
+ ASSERT_EQ(factory_.ToAlignedPath("index-0", file_meta), "/tmp/index-0");
+}
+
+TEST_F(DataFilePathFactoryTest, TestCollectFiles) {
+ auto file_meta = std::make_shared<DataFileMeta>(
+ "data-0.txt", /*file_size=*/645, /*row_count=*/5,
BinaryRow::EmptyRow(),
+ BinaryRow::EmptyRow(), SimpleStats::EmptyStats(),
SimpleStats::EmptyStats(),
+ /*min_sequence_number=*/0, /*max_sequence_number=*/4, /*schema_id=*/0,
+ /*level=*/0, /*extra_files=*/std::vector<std::optional<std::string>>(),
+ /*creation_time=*/Timestamp(1737111915429ll, 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);
+ ASSERT_EQ(factory_.CollectFiles(file_meta),
std::vector<std::string>({"/tmp/data-0.txt"}));
+
+ file_meta->extra_files = {"data-0.index", "data-1.index"};
+ ASSERT_EQ(
+ factory_.CollectFiles(file_meta),
+ std::vector<std::string>({"/tmp/data-0.txt", "/tmp/data-0.index",
"/tmp/data-1.index"}));
+}
+} // namespace paimon::test
