feat(core): support per-column maximum sequence numbers

src/paimon/CMakeLists.txt

@@ -275,6 +275,7 @@ set(PAIMON_CORE_SRCS
     core/io/data_file_meta_first_row_id_legacy_serializer.cpp
     core/io/data_file_meta.cpp
     core/io/data_file_meta_serializer.cpp
+    core/io/data_file_meta_write_cols_legacy_serializer.cpp
     core/io/data_file_path_factory.cpp
     core/io/data_file_index_writer.cpp
     core/io/file_index_options.cpp

src/paimon/core/append/bucketed_append_compact_manager_test.cpp

@@ -115,7 +115,7 @@ class BucketedAppendCompactManagerTest : public testing::Test {
             /*embedded_index=*/nullptr, FileSource::Append(),
             /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
             /*first_row_id=*/std::nullopt,
-            /*write_cols=*/std::nullopt);
+            /*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
     }
 
     std::shared_ptr<DataFileMeta> NewNamedFile(const std::string& file_name, int64_t file_size,
@@ -135,7 +135,7 @@ class BucketedAppendCompactManagerTest : public testing::Test {
             /*embedded_index=*/nullptr, FileSource::Append(),
             /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
             /*first_row_id=*/std::nullopt,
-            /*write_cols=*/std::nullopt);
+            /*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
     }
 
     std::shared_ptr<BucketedDvMaintainer> CreateTestDvMaintainer(

src/paimon/core/deletionvectors/deletion_vector_test.cpp

@@ -54,7 +54,7 @@ std::shared_ptr<DataFileMeta> CreateDataFileMeta(const std::string& file_name) {
         /*min_sequence_number=*/0, /*max_sequence_number=*/0, /*schema_id=*/0,
         DataFileMeta::DUMMY_LEVEL, std::vector<std::optional<std::string>>{}, Timestamp(0, 0),
         std::nullopt, nullptr, FileSource::Append(), std::nullopt, std::nullopt, std::nullopt,
-        std::nullopt);
+        std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 }  // namespace

src/paimon/core/global_index/indexed_split_test.cpp

@@ -56,17 +56,20 @@ TEST(IndexedSplitTest, TestSimple) {
         "file1.orc", 100l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 50l, 249l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
     auto meta2 = std::make_shared<DataFileMeta>(
         "file2.orc", 101l, 100l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 250l, 349l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
     auto meta3 = std::make_shared<DataFileMeta>(
         "file3.orc", 102l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765535214349l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 
     DataSplitImpl::Builder builder(
         /*partition=*/BinaryRow::EmptyRow(),
@@ -82,8 +85,11 @@ TEST(IndexedSplitTest, TestSimple) {
 
     ASSERT_EQ(*result_indexed_split, *expected_indexed_split) << result_indexed_split->ToString();
     ASSERT_OK_AND_ASSIGN(std::string serialize_bytes, Split::Serialize(result_indexed_split, pool));
-    ASSERT_EQ(serialize_bytes,
-              std::string(reinterpret_cast<char*>(split_bytes.data()), split_bytes.size()));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> roundtrip,
+                         Split::Deserialize(serialize_bytes.data(), serialize_bytes.size(), pool));
+    auto roundtrip_indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(roundtrip);
+    ASSERT_EQ(*roundtrip_indexed_split, *expected_indexed_split)
+        << roundtrip_indexed_split->ToString();
 }
 
 TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
@@ -107,17 +113,20 @@ TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
         "file1.orc", 100l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 50l, 249l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765549435648l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 50l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
     auto meta2 = std::make_shared<DataFileMeta>(
         "file2.orc", 101l, 100l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 250l, 349l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765549435649l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 250l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
     auto meta3 = std::make_shared<DataFileMeta>(
         "file3.orc", 102l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(1765549435649l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 
     DataSplitImpl::Builder builder(
         /*partition=*/BinaryRow::EmptyRow(),
@@ -139,16 +148,20 @@ TEST(IndexedSplitTest, TestIndexedSplitWithScore) {
             "rowRanges=[[55, 56],[270, 270],[1001, 1002]], scores=[1.01,2.1,-1.32,4.23,50.74]") !=
         std::string::npos);
     ASSERT_OK_AND_ASSIGN(std::string serialize_bytes, Split::Serialize(result_indexed_split, pool));
-    ASSERT_EQ(serialize_bytes,
-              std::string(reinterpret_cast<char*>(split_bytes.data()), split_bytes.size()));
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Split> roundtrip,
+                         Split::Deserialize(serialize_bytes.data(), serialize_bytes.size(), pool));
+    auto roundtrip_indexed_split = std::dynamic_pointer_cast<IndexedSplitImpl>(roundtrip);
+    ASSERT_EQ(*roundtrip_indexed_split, *expected_indexed_split)
+        << roundtrip_indexed_split->ToString();
 }
 
 TEST(IndexedSplitTest, TestValidate) {
     auto meta = std::make_shared<DataFileMeta>(
         "file.orc", 1l, 200l, BinaryRow::EmptyRow(), BinaryRow::EmptyRow(),
         SimpleStats::EmptyStats(), SimpleStats::EmptyStats(), 1000l, 1199l, 0, 0,
         std::vector<std::optional<std::string>>(), Timestamp(0l, 0), 0, nullptr,
-        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt);
+        FileSource::Append(), std::nullopt, std::nullopt, 1000l, std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 
     DataSplitImpl::Builder builder(
         /*partition=*/BinaryRow::EmptyRow(),

src/paimon/core/index/pksorted/pk_sorted_bucket_index_state_test.cpp

@@ -53,7 +53,7 @@ class PkSortedBucketIndexStateTest : public ::testing::Test {
             /*creation_time=*/Timestamp(0, 0), /*delete_row_count=*/std::nullopt,
             /*embedded_index=*/nullptr, file_source, /*value_stats_cols=*/std::nullopt,
             /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt,
-            /*write_cols=*/std::nullopt);
+            /*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
     }
 
     /// Builds a payload whose source metadata lists the given sources in the given order.

src/paimon/core/io/data_file_meta.cpp

@@ -72,7 +72,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta::ForAppend(
         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);
+        embedded_index, file_source, value_stats_cols, external_path, first_row_id, write_cols,
+        /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level) const {
@@ -84,7 +85,7 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta::Upgrade(int32_t new_level) c
         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);
+        first_row_id, write_cols, column_max_sequence_numbers);
 }
 
 std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
@@ -93,15 +94,24 @@ std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithExtraFiles(
         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);
+        first_row_id, write_cols, column_max_sequence_numbers);
+}
+
+std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithColumnMaxSequenceNumbers(
+    const std::optional<std::vector<int64_t>>& new_column_max_sequence_numbers) 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, extra_files, creation_time,
+        delete_row_count, embedded_index, file_source, value_stats_cols, external_path,
+        first_row_id, write_cols, new_column_max_sequence_numbers);
 }
 
 std::shared_ptr<DataFileMeta> DataFileMeta::CopyWithoutStats() const {
     return std::make_shared<DataFileMeta>(
         file_name, file_size, row_count, min_key, max_key, key_stats, SimpleStats::EmptyStats(),
         min_sequence_number, max_sequence_number, schema_id, level, extra_files, creation_time,
         delete_row_count, embedded_index, file_source, std::vector<std::string>(), external_path,
-        first_row_id, write_cols);
+        first_row_id, write_cols, column_max_sequence_numbers);
 }
 
 DataFileMeta::DataFileMeta(
@@ -113,7 +123,8 @@ DataFileMeta::DataFileMeta(
     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)
+    const std::optional<std::vector<std::string>>& _write_cols,
+    const std::optional<std::vector<int64_t>>& _column_max_sequence_numbers)
     : file_name(_file_name),
       file_size(_file_size),
       row_count(_row_count),
@@ -133,7 +144,8 @@ DataFileMeta::DataFileMeta(
       value_stats_cols(_value_stats_cols),
       external_path(_external_path),
       first_row_id(_first_row_id),
-      write_cols(_write_cols) {}
+      write_cols(_write_cols),
+      column_max_sequence_numbers(_column_max_sequence_numbers) {}
 
 Result<std::string> DataFileMeta::FileFormat() const {
     size_t last_dot_index = file_name.find_last_of(".");
@@ -198,7 +210,8 @@ bool DataFileMeta::operator==(const DataFileMeta& other) const {
            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;
+           write_cols == other.write_cols &&
+           column_max_sequence_numbers == other.column_max_sequence_numbers;
 }
 
 bool DataFileMeta::operator!=(const DataFileMeta& other) const {
@@ -243,7 +256,8 @@ bool DataFileMeta::TEST_Equal(const DataFileMeta& other) const {
            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;
+           first_row_id == other.first_row_id && write_cols == other.write_cols &&
+           column_max_sequence_numbers == other.column_max_sequence_numbers;
 }
 
 std::string DataFileMeta::ToString() const {
@@ -261,7 +275,8 @@ std::string DataFileMeta::ToString() const {
         "{}, "
         "keyStats: {}, valueStats: {}, minSequenceNumber: {}, maxSequenceNumber: {}, schemaId: "
         "{}, level: {}, extraFiles: {}, creationTime: {}, deleteRowCount: {}, fileSource: {}, "
-        "valueStatsCols: {}, externalPath: {}, firstRowId: {}, writeCols: {}}}",
+        "valueStatsCols: {}, externalPath: {}, firstRowId: {}, writeCols: {}, "
+        "columnMaxSequenceNumbers: {}}}",
         file_name, file_size, row_count,
         embedded_index == nullptr ? "null"
                                   : std::string(embedded_index->data(), embedded_index->size()),
@@ -275,7 +290,10 @@ std::string DataFileMeta::ToString() const {
             : 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()));
+        write_cols == std::nullopt ? "null" : fmt::format("{}", write_cols.value()),
+        column_max_sequence_numbers == std::nullopt
+            ? "null"
+            : fmt::format("{}", column_max_sequence_numbers.value()));
 }
 
 int64_t DataFileMeta::GetMaxSequenceNumber(
@@ -315,6 +333,9 @@ const std::shared_ptr<arrow::DataType>& DataFileMeta::DataType() {
          arrow::field("_FIRST_ROW_ID", arrow::int64(), /*nullable=*/true),
          arrow::field("_WRITE_COLS",
                       arrow::list(arrow::field("item", arrow::utf8(), /*nullable=*/false)),
+                      /*nullable=*/true),
+         arrow::field("_WRITE_COLS_SEQUENCES",
+                      arrow::list(arrow::field("item", arrow::int64(), /*nullable=*/false)),
                       /*nullable=*/true)});
     return schema;
 }

src/paimon/core/io/data_file_meta.h

@@ -58,7 +58,8 @@ struct DataFileMeta {
                  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);
+                 const std::optional<std::vector<std::string>>& _write_cols,
+                 const std::optional<std::vector<int64_t>>& _column_max_sequence_numbers);
 
     static Result<std::shared_ptr<DataFileMeta>> ForAppend(
         const std::string& file_name, int64_t file_size, int64_t row_count,
@@ -83,6 +84,9 @@ struct DataFileMeta {
     std::shared_ptr<DataFileMeta> CopyWithExtraFiles(
         const std::vector<std::optional<std::string>>& new_extra_files) const;
 
+    std::shared_ptr<DataFileMeta> CopyWithColumnMaxSequenceNumbers(
+        const std::optional<std::vector<int64_t>>& new_column_max_sequence_numbers) const;
+
     /// Create a copy without value statistics. All other metadata is preserved.
     ///
     /// @return A new metadata object with empty value statistics and value-stat columns.
@@ -167,5 +171,12 @@ struct DataFileMeta {
     std::optional<int64_t> first_row_id;
 
     std::optional<std::vector<std::string>> write_cols;
+
+    /// Maximum sequence number per physical table field after data-evolution compaction.
+    ///
+    /// Values follow the table-field order selected by `write_cols` when it is non-null (system
+    /// fields are ignored), or the file schema field order otherwise. A null value means that only
+    /// the file-level sequence range is available.
+    std::optional<std::vector<int64_t>> column_max_sequence_numbers;
 };
 }  // namespace paimon

src/paimon/core/io/data_file_meta_09_serializer.cpp

@@ -114,7 +114,7 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta09Serializer::FromRow(
         embedded_file_index, file_source,
         /*value_stats_cols=*/std::nullopt, /*external_path=*/std::nullopt,
         /*first_row_id=*/std::nullopt,
-        /*write_cols=*/std::nullopt);
+        /*write_cols=*/std::nullopt, /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 }  // namespace paimon

src/paimon/core/io/data_file_meta_10_serializer.cpp

@@ -123,7 +123,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta10Serializer::FromRow(
         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);
+        /*external_path=*/std::nullopt, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 }  // namespace paimon

src/paimon/core/io/data_file_meta_12_serializer.cpp

@@ -128,7 +128,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMeta12Serializer::FromRow(
         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);
+        external_path, /*first_row_id=*/std::nullopt, /*write_cols=*/std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 }  // namespace paimon

src/paimon/core/io/data_file_meta_first_row_id_legacy_serializer.cpp

@@ -134,7 +134,8 @@ Result<std::shared_ptr<DataFileMeta>> DataFileMetaFirstRowIdLegacySerializer::Fr
         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);
+        external_path, first_row_id, /*write_cols=*/std::nullopt,
+        /*column_max_sequence_numbers=*/std::nullopt);
 }
 
 }  // namespace paimon