feat: Add metadata system tables (alibaba#285)

Author: suxiaogang223

src/paimon/CMakeLists.txt

@@ -201,6 +201,7 @@ set(PAIMON_CORE_SRCS
     core/index/index_file_handler.cpp
     core/index/global_index_meta.cpp
     core/index/index_file_meta_serializer.cpp
+    core/io/generic_row_to_arrow_array_converter.cpp
     core/io/meta_to_arrow_array_converter.cpp
     core/io/async_key_value_producer_and_consumer.cpp
     core/io/data_file_meta_09_serializer.cpp
@@ -319,11 +320,14 @@ set(PAIMON_CORE_SRCS
     core/table/source/data_evolution_batch_scan.cpp
     core/table/system/audit_log_system_table.cpp
     core/table/system/binlog_system_table.cpp
-    core/table/system/options_system_table.cpp
+    core/table/system/in_memory_system_table.cpp
+    core/table/system/metadata_system_tables.cpp
     core/table/system/system_table.cpp
     core/table/system/system_table_scan.cpp
     core/table/system/system_table_schema.cpp
     core/tag/tag.cpp
+    core/utils/branch_manager.cpp
+    core/utils/consumer_manager.cpp
     core/utils/field_mapping.cpp
     core/utils/file_store_path_factory.cpp
     core/utils/file_utils.cpp
@@ -706,6 +710,7 @@ if(PAIMON_BUILD_TESTS)
                     core/table/system/system_table_test.cpp
                     core/tag/tag_test.cpp
                     core/utils/branch_manager_test.cpp
+                    core/utils/consumer_manager_test.cpp
                     core/utils/file_store_path_factory_cache_test.cpp
                     core/utils/field_mapping_test.cpp
                     core/utils/file_store_path_factory_test.cpp

src/paimon/common/reader/prefetch_file_batch_reader_impl_test.cpp

@@ -16,6 +16,7 @@
 
 #include "paimon/common/reader/prefetch_file_batch_reader_impl.h"
 
+#include <atomic>
 #include <set>
 
 #include "arrow/compute/api.h"
@@ -90,7 +91,7 @@ class ControlledMockFormatReaderBuilder : public ReaderBuilder {
 
     Result<std::unique_ptr<FileBatchReader>> Build(
         const std::shared_ptr<InputStream>& path) const override {
-        size_t index = build_count_++;
+        size_t index = build_count_.fetch_add(1);
         Status set_read_ranges_status = index < set_read_ranges_statuses_.size()
                                             ? set_read_ranges_statuses_[index]
                                             : Status::OK();
@@ -109,7 +110,7 @@ class ControlledMockFormatReaderBuilder : public ReaderBuilder {
     std::vector<std::pair<uint64_t, uint64_t>> read_ranges_;
     bool need_prefetch_ = true;
     std::vector<Status> set_read_ranges_statuses_;
-    mutable size_t build_count_ = 0;
+    mutable std::atomic<size_t> build_count_{0};
 };
 
 struct TestParam {

src/paimon/common/utils/date_time_utils.h

@@ -105,15 +105,28 @@ class DateTimeUtils {
         return static_cast<uint64_t>(ts.tv_sec) * 1000000ULL + static_cast<uint64_t>(ts.tv_usec);
     }
 
-    static inline Result<uint64_t> GetCurrentLocalTimeUs() {
-        uint64_t utc_micro = GetCurrentUTCTimeUs();
+    static inline Result<Timestamp> ToLocalTimestamp(const Timestamp& utc_timestamp) {
+        int64_t utc_micro = utc_timestamp.ToMicrosecond();
         auto utc_ts_scalar = std::make_shared<arrow::TimestampScalar>(
-            static_cast<int64_t>(utc_micro), arrow::TimeUnit::MICRO, GetLocalTimezoneName());
+            utc_micro, arrow::TimeUnit::MICRO, GetLocalTimezoneName());
         PAIMON_ASSIGN_OR_RAISE_FROM_ARROW(
             arrow::Datum local_micro, arrow::compute::LocalTimestamp(arrow::Datum(utc_ts_scalar)));
         auto local_ts_scalar =
             std::dynamic_pointer_cast<arrow::TimestampScalar>(local_micro.scalar());
-        return *(static_cast<const int64_t*>(local_ts_scalar->data()));
+        auto [millisecond, nano_of_millisecond] = DateTimeUtils::TimestampConverter(
+            *(static_cast<const int64_t*>(local_ts_scalar->data())),
+            DateTimeUtils::TimeType::MICROSECOND, DateTimeUtils::TimeType::MILLISECOND,
+            DateTimeUtils::TimeType::NANOSECOND);
+        return Timestamp(millisecond, nano_of_millisecond);
+    }
+
+    static inline Result<uint64_t> GetCurrentLocalTimeUs() {
+        auto [millisecond, nano_of_millisecond] = DateTimeUtils::TimestampConverter(
+            GetCurrentUTCTimeUs(), DateTimeUtils::TimeType::MICROSECOND,
+            DateTimeUtils::TimeType::MILLISECOND, DateTimeUtils::TimeType::NANOSECOND);
+        Timestamp utc_timestamp(millisecond, nano_of_millisecond);
+        PAIMON_ASSIGN_OR_RAISE(Timestamp local_timestamp, ToLocalTimestamp(utc_timestamp));
+        return local_timestamp.ToMicrosecond();
     }
 
     static inline Result<int32_t> GetCurrentLocalHour() {

src/paimon/common/utils/date_time_utils_test.cpp

@@ -281,6 +281,21 @@ TEST(DateTimeUtilsTest, TestGetCurrentLocalTimeUs) {
     ASSERT_GE(local_ts - utc_ts, 28800000000l);
 }
 
+TEST(DateTimeUtilsTest, TestToLocalTimestamp) {
+    {
+        TimezoneGuard guard("Asia/Shanghai");
+        ASSERT_OK_AND_ASSIGN(Timestamp timestamp, DateTimeUtils::ToLocalTimestamp(
+                                                      Timestamp::FromEpochMillis(1700000000123L)));
+        ASSERT_EQ(timestamp, Timestamp::FromEpochMillis(1700028800123L));
+    }
+    {
+        TimezoneGuard guard("UTC");
+        ASSERT_OK_AND_ASSIGN(Timestamp timestamp, DateTimeUtils::ToLocalTimestamp(
+                                                      Timestamp::FromEpochMillis(1700000000123L)));
+        ASSERT_EQ(timestamp, Timestamp::FromEpochMillis(1700000000123L));
+    }
+}
+
 TEST(DateTimeUtilsTest, TestGetCurrentLocalHour) {
     int32_t shanghai_hour = 0;
     int32_t utc_hour = 0;

src/paimon/core/catalog/file_system_catalog.cpp

@@ -365,27 +365,11 @@ Result<std::vector<std::string>> FileSystemCatalog::GetSchemaExternalPaths(
 
 Result<std::vector<std::string>> FileSystemCatalog::GetTableBranches(
     const std::string& table_path) const {
-    std::vector<std::string> branches;
-    std::string branch_dir = PathUtil::JoinPath(table_path, "branch");
-    PAIMON_ASSIGN_OR_RAISE(bool branch_dir_exists, fs_->Exists(branch_dir));
-    if (!branch_dir_exists) {
-        return branches;
-    }
-
-    std::vector<std::unique_ptr<BasicFileStatus>> file_status_list;
-    PAIMON_RETURN_NOT_OK(fs_->ListDir(branch_dir, &file_status_list));
-
-    for (const auto& file_status : file_status_list) {
-        if (file_status->IsDir()) {
-            std::string dir_name = PathUtil::GetName(file_status->GetPath());
-            // Branch directory name format: branch-{branch_name}
-            const std::string branch_prefix = BranchManager::BRANCH_PREFIX;
-            if (StringUtils::StartsWith(dir_name, branch_prefix, /*start_pos=*/0)) {
-                std::string branch_name = dir_name.substr(branch_prefix.length());
-                branches.push_back(branch_name);
-            }
-        }
-    }
+    PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> branches,
+                           BranchManager::ListBranches(fs_, table_path));
+    branches.erase(
+        std::remove(branches.begin(), branches.end(), BranchManager::DEFAULT_MAIN_BRANCH),
+        branches.end());
     return branches;
 }
 

src/paimon/core/catalog/file_system_catalog_test.cpp

@@ -90,7 +90,7 @@ TEST(FileSystemCatalogTest, TestCreateSystemDatabaseAndTable) {
                                                    /*ignore_if_exists=*/true),
                             "Cannot create database for system database");
     }
-    // do not support create system table
+    /// Do not support create system table.
     {
         std::map<std::string, std::string> options;
         options[Options::FILE_SYSTEM] = "local";
@@ -281,6 +281,100 @@ TEST(FileSystemCatalogTest, TestAuditLogAndBinlogSystemTableCatalog) {
                         "Cannot rename system table");
 }
 
+TEST(FileSystemCatalogTest, TestMetadataSystemTableCatalog) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str());
+    ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true));
+
+    auto typed_schema =
+        arrow::schema({arrow::field("pk", arrow::utf8()), arrow::field("v", arrow::int32())});
+    ::ArrowSchema schema;
+    ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok());
+    ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema,
+                                  /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, options,
+                                  /*ignore_if_exists=*/false));
+    ArrowSchemaRelease(&schema);
+
+    std::vector<std::string> metadata_tables = {"snapshots", "schemas", "tags", "branches",
+                                                "consumers"};
+    for (const auto& table_name : metadata_tables) {
+        Identifier system_identifier("db1", "tbl1$" + table_name);
+        ASSERT_OK_AND_ASSIGN(bool exists, catalog.TableExists(system_identifier));
+        ASSERT_TRUE(exists) << table_name;
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> system_schema,
+                             catalog.LoadTableSchema(system_identifier));
+        ASSERT_TRUE(std::dynamic_pointer_cast<SystemTableSchema>(system_schema) != nullptr)
+            << table_name;
+        ASSERT_OK_AND_ASSIGN(auto c_schema, system_schema->GetArrowSchema());
+        auto loaded_schema_result = arrow::ImportSchema(c_schema.get());
+        ASSERT_TRUE(loaded_schema_result.ok()) << loaded_schema_result.status().ToString();
+        ASSERT_GT(loaded_schema_result.ValueUnsafe()->num_fields(), 0) << table_name;
+    }
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> snapshots_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1$snapshots")));
+    ASSERT_OK_AND_ASSIGN(auto snapshots_c_schema, snapshots_schema->GetArrowSchema());
+    auto snapshots_arrow_schema = arrow::ImportSchema(snapshots_c_schema.get()).ValueUnsafe();
+    ASSERT_EQ(snapshots_arrow_schema->field_names(),
+              (std::vector<std::string>{
+                  "snapshot_id", "schema_id", "commit_user", "commit_identifier", "commit_kind",
+                  "commit_time", "base_manifest_list", "delta_manifest_list",
+                  "changelog_manifest_list", "total_record_count", "delta_record_count",
+                  "changelog_record_count", "watermark", "next_row_id"}));
+    ASSERT_EQ(snapshots_arrow_schema->field(5)->type()->id(), arrow::Type::TIMESTAMP);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> schemas_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1$schemas")));
+    ASSERT_OK_AND_ASSIGN(auto schemas_c_schema, schemas_schema->GetArrowSchema());
+    auto schemas_arrow_schema = arrow::ImportSchema(schemas_c_schema.get()).ValueUnsafe();
+    ASSERT_EQ(schemas_arrow_schema->field_names(),
+              (std::vector<std::string>{"schema_id", "fields", "partition_keys", "primary_keys",
+                                        "options", "comment", "update_time"}));
+    ASSERT_EQ(schemas_arrow_schema->field(6)->type()->id(), arrow::Type::TIMESTAMP);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> tags_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1$tags")));
+    ASSERT_OK_AND_ASSIGN(auto tags_c_schema, tags_schema->GetArrowSchema());
+    auto tags_arrow_schema = arrow::ImportSchema(tags_c_schema.get()).ValueUnsafe();
+    ASSERT_EQ(tags_arrow_schema->field_names(),
+              (std::vector<std::string>{"tag_name", "snapshot_id", "schema_id", "commit_time",
+                                        "record_count", "create_time", "time_retained"}));
+    ASSERT_EQ(tags_arrow_schema->field(3)->type()->id(), arrow::Type::TIMESTAMP);
+    ASSERT_EQ(tags_arrow_schema->field(5)->type()->id(), arrow::Type::TIMESTAMP);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> branches_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1$branches")));
+    ASSERT_OK_AND_ASSIGN(auto branches_c_schema, branches_schema->GetArrowSchema());
+    auto branches_arrow_schema = arrow::ImportSchema(branches_c_schema.get()).ValueUnsafe();
+    ASSERT_EQ(branches_arrow_schema->field_names(),
+              (std::vector<std::string>{"branch_name", "create_time"}));
+    ASSERT_EQ(branches_arrow_schema->field(1)->type()->id(), arrow::Type::TIMESTAMP);
+
+    ASSERT_OK_AND_ASSIGN(std::shared_ptr<Schema> consumers_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1$consumers")));
+    ASSERT_OK_AND_ASSIGN(auto consumers_c_schema, consumers_schema->GetArrowSchema());
+    auto consumers_arrow_schema = arrow::ImportSchema(consumers_c_schema.get()).ValueUnsafe();
+    ASSERT_EQ(consumers_arrow_schema->field_names(),
+              (std::vector<std::string>{"consumer_id", "next_snapshot_id"}));
+    ASSERT_FALSE(consumers_arrow_schema->field(1)->nullable());
+
+    Identifier snapshots_identifier("db1", "tbl1$snapshots");
+    ::ArrowSchema system_create_schema;
+    ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &system_create_schema).ok());
+    ASSERT_NOK_WITH_MSG(
+        catalog.CreateTable(snapshots_identifier, &system_create_schema, {}, {}, options, false),
+        "Cannot create table for system table");
+    ArrowSchemaRelease(&system_create_schema);
+    ASSERT_NOK_WITH_MSG(catalog.DropTable(snapshots_identifier, false), "Cannot drop system table");
+    ASSERT_NOK_WITH_MSG(catalog.RenameTable(snapshots_identifier, Identifier("db1", "tbl2"), false),
+                        "Cannot rename system table");
+}
+
 TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
     std::map<std::string, std::string> options;
     options[Options::FILE_SYSTEM] = "local";
@@ -625,7 +719,7 @@ TEST(FileSystemCatalogTest, TestDropTable) {
     ASSERT_OK_AND_ASSIGN(bool exist, catalog.TableExists(Identifier("test_db", "tbl1")));
     ASSERT_FALSE(exist);
 
-    // Test 4: Drop system table
+    /// Test 4: Drop system table.
     ASSERT_NOK_WITH_MSG(
         catalog.DropTable(Identifier("test_db", "tbl$system"),
                           /*ignore_if_not_exists=*/false),
@@ -691,7 +785,7 @@ TEST(FileSystemCatalogTest, TestRenameTable) {
                             /*ignore_if_not_exists=*/false),
         "Cannot rename table across databases. Cross-database rename is not supported.");
 
-    // Test 6: Rename system table
+    /// Test 6: Rename system table.
     ASSERT_NOK_WITH_MSG(catalog.RenameTable(Identifier("test_db", "tbl$system"),
                                             Identifier("test_db", "new_system_tbl"),
                                             /*ignore_if_not_exists=*/false),

src/paimon/core/core_options.cpp

@@ -669,7 +669,6 @@ struct CoreOptions::Impl {
         // Parse table-read.sequence-number.enabled - expose sequence number in system tables
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::TABLE_READ_SEQUENCE_NUMBER_ENABLED,
                                                 &table_read_sequence_number_enabled));
-        // Parse key-value.sequence_number.enabled - internal sequence number read switch
         PAIMON_RETURN_NOT_OK(parser.Parse<bool>(Options::KEY_VALUE_SEQUENCE_NUMBER_ENABLED,
                                                 &key_value_sequence_number_enabled));
         // Parse partial-update.remove-record-on-sequence-group

src/paimon/core/io/generic_row_to_arrow_array_converter.cpp

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2026-present Alibaba Inc.
+ *
+ * Licensed 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/generic_row_to_arrow_array_converter.h"
+
+#include <cassert>
+#include <memory>
+#include <utility>
+#include <vector>
+
+#include "arrow/array/builder_nested.h"
+#include "arrow/memory_pool.h"
+#include "arrow/util/checked_cast.h"
+#include "paimon/common/utils/arrow/status_utils.h"
+
+namespace paimon {
+
+Result<std::unique_ptr<GenericRowToArrowArrayConverter>> GenericRowToArrowArrayConverter::Create(
+    const std::shared_ptr<arrow::Schema>& schema, arrow::MemoryPool* pool) {
+    std::unique_ptr<arrow::ArrayBuilder> array_builder;
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::MakeBuilder(
+        pool, std::make_shared<arrow::StructType>(schema->fields()), &array_builder));
+
+    auto struct_builder =
+        arrow::internal::checked_pointer_cast<arrow::StructBuilder>(std::move(array_builder));
+    assert(struct_builder);
+    std::vector<RowToArrowArrayConverter::AppendValueFunc> appenders;
+    appenders.reserve(schema->num_fields());
+    int32_t reserve_count = 1;
+    for (int32_t i = 0; i < schema->num_fields(); ++i) {
+        PAIMON_ASSIGN_OR_RAISE(
+            RowToArrowArrayConverter::AppendValueFunc func,
+            AppendField(/*use_view=*/true, struct_builder->field_builder(i), &reserve_count));
+        appenders.emplace_back(std::move(func));
+    }
+    return std::unique_ptr<GenericRowToArrowArrayConverter>(new GenericRowToArrowArrayConverter(
+        reserve_count, std::move(appenders), std::move(struct_builder), nullptr));
+}
+
+Result<BatchReader::ReadBatch> GenericRowToArrowArrayConverter::NextBatch(
+    const std::vector<GenericRow>& rows) {
+    PAIMON_RETURN_NOT_OK(ResetAndReserve());
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(
+        array_builder_->AppendValues(rows.size(), /*valid_bytes=*/nullptr));
+    for (size_t i = 0; i < appenders_.size(); ++i) {
+        for (const auto& row : rows) {
+            PAIMON_RETURN_NOT_OK_FROM_ARROW(appenders_[i](row, i));
+        }
+    }
+
+    return FinishAndAccumulate();
+}
+
+}  // namespace paimon