From 0ee3048c98976e39f9f1c9a86e9fa977e9a3f4d4 Mon Sep 17 00:00:00 2001 From: Socrates Date: Sat, 13 Jun 2026 18:18:20 +0800 Subject: [PATCH 01/13] feat: add global system tables framework under sys database - Add GetOptions() to Catalog interface for catalog-level config access - FileSystemCatalog stores and exposes catalog_options - New GlobalSystemTableLoader with independent registry for sys tables - Implement sys.catalog_options, sys.all_table_options, sys.tables - Stub for sys.partitions (manifest aggregation to follow) - Extend SystemTablePath with is_global flag - TryParsePath detects sys/ paths for TableScan/TableRead routing - FileSystemCatalog handles sys database in ListTables, DatabaseExists, TableExists, LoadTableSchema Co-Authored-By: Claude Fable 5 --- include/paimon/catalog/catalog.h | 5 + src/paimon/CMakeLists.txt | 1 + src/paimon/core/catalog/catalog.cpp | 2 +- .../core/catalog/file_system_catalog.cpp | 39 +- src/paimon/core/catalog/file_system_catalog.h | 5 +- .../table/system/global_system_tables.cpp | 362 ++++++++++++++++++ .../core/table/system/global_system_tables.h | 117 ++++++ src/paimon/core/table/system/system_table.cpp | 28 +- src/paimon/core/table/system/system_table.h | 2 + 9 files changed, 553 insertions(+), 8 deletions(-) create mode 100644 src/paimon/core/table/system/global_system_tables.cpp create mode 100644 src/paimon/core/table/system/global_system_tables.h diff --git a/include/paimon/catalog/catalog.h b/include/paimon/catalog/catalog.h index 0ff9349bd..745756b71 100644 --- a/include/paimon/catalog/catalog.h +++ b/include/paimon/catalog/catalog.h @@ -183,6 +183,11 @@ class PAIMON_EXPORT Catalog { /// @return A shared pointer to the file system instance. virtual std::shared_ptr GetFileSystem() const = 0; + /// Returns the catalog-level options that were passed during catalog creation. + /// + /// @return A const reference to the map of catalog options (key-value pairs). + virtual const std::map& GetOptions() const = 0; + /// Loads the latest schema of a specified table. /// /// @note System tables will not be supported. diff --git a/src/paimon/CMakeLists.txt b/src/paimon/CMakeLists.txt index 46cc13854..03900085f 100644 --- a/src/paimon/CMakeLists.txt +++ b/src/paimon/CMakeLists.txt @@ -354,6 +354,7 @@ 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/global_system_tables.cpp core/table/system/in_memory_system_table.cpp core/table/system/metadata_system_tables.cpp core/table/system/read_optimized_system_table.cpp diff --git a/src/paimon/core/catalog/catalog.cpp b/src/paimon/core/catalog/catalog.cpp index 08b879c7a..80f93cb29 100644 --- a/src/paimon/core/catalog/catalog.cpp +++ b/src/paimon/core/catalog/catalog.cpp @@ -32,7 +32,7 @@ Result> Catalog::Create(const std::string& root_path, const std::map& options, const std::shared_ptr& file_system) { PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options, file_system)); - return std::make_unique(core_options.GetFileSystem(), root_path); + return std::make_unique(core_options.GetFileSystem(), root_path, options); } } // namespace paimon diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp index db9c2494e..eab4ae666 100644 --- a/src/paimon/core/catalog/file_system_catalog.cpp +++ b/src/paimon/core/catalog/file_system_catalog.cpp @@ -31,6 +31,7 @@ #include "paimon/common/utils/string_utils.h" #include "paimon/core/core_options.h" #include "paimon/core/snapshot.h" +#include "paimon/core/table/system/global_system_tables.h" #include "paimon/core/table/system/system_table.h" #include "paimon/core/table/system/system_table_schema.h" #include "paimon/core/utils/branch_manager.h" @@ -47,8 +48,12 @@ struct ArrowSchema; namespace paimon { FileSystemCatalog::FileSystemCatalog(const std::shared_ptr& fs, - const std::string& warehouse) - : fs_(fs), warehouse_(warehouse), logger_(Logger::GetLogger("FileSystemCatalog")) {} + const std::string& warehouse, + const std::map& catalog_options) + : fs_(fs), + warehouse_(warehouse), + catalog_options_(catalog_options), + logger_(Logger::GetLogger("FileSystemCatalog")) {} Status FileSystemCatalog::CreateDatabase(const std::string& db_name, const std::map& options, @@ -88,13 +93,16 @@ Status FileSystemCatalog::CreateDatabaseImpl(const std::string& db_name, Result FileSystemCatalog::DatabaseExists(const std::string& db_name) const { if (IsSystemDatabase(db_name)) { - return Status::NotImplemented( - "do not support checking DatabaseExists for system database."); + return true; } return fs_->Exists(NewDatabasePath(warehouse_, db_name)); } Result FileSystemCatalog::TableExists(const Identifier& identifier) const { + // Handle sys database global tables + if (IsSystemDatabase(identifier.GetDatabaseName())) { + return GlobalSystemTableLoader::IsSupported(identifier.GetTableName()); + } PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); if (is_system_table) { PAIMON_ASSIGN_OR_RAISE(std::optional system_table_name, @@ -184,6 +192,10 @@ std::shared_ptr FileSystemCatalog::GetFileSystem() const { return fs_; } +const std::map& FileSystemCatalog::GetOptions() const { + return catalog_options_; +} + bool FileSystemCatalog::IsSystemDatabase(const std::string& db_name) { return db_name == SYSTEM_DATABASE_NAME; } @@ -228,7 +240,7 @@ Result> FileSystemCatalog::ListDatabases() const { Result> FileSystemCatalog::ListTables(const std::string& db_name) const { if (IsSystemDatabase(db_name)) { - return Status::NotImplemented("do not support listing tables for system database."); + return GlobalSystemTableLoader::GetSupportedTableNames(); } std::string database_path = NewDatabasePath(warehouse_, db_name); std::vector> file_status_list; @@ -261,6 +273,23 @@ Result FileSystemCatalog::TableExistsInFileSystem(const std::string& table Result> FileSystemCatalog::LoadTableSchema( const Identifier& identifier) const { + // Handle sys database global tables + if (IsSystemDatabase(identifier.GetDatabaseName())) { + if (!GlobalSystemTableLoader::IsSupported(identifier.GetTableName())) { + return Status::NotExist(fmt::format("{} not exist", identifier.ToString())); + } + GlobalSystemTableContext context; + context.catalog = const_cast(this); + context.fs = fs_; + context.warehouse = warehouse_; + context.catalog_options = catalog_options_; + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr system_table, + GlobalSystemTableLoader::Load(identifier.GetTableName(), context)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, + system_table->ArrowSchema()); + return std::make_shared(std::move(arrow_schema)); + } PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); if (is_system_table) { PAIMON_ASSIGN_OR_RAISE(std::optional system_table_name, diff --git a/src/paimon/core/catalog/file_system_catalog.h b/src/paimon/core/catalog/file_system_catalog.h index b9fbc2b47..3b93dc650 100644 --- a/src/paimon/core/catalog/file_system_catalog.h +++ b/src/paimon/core/catalog/file_system_catalog.h @@ -38,7 +38,8 @@ class Logger; class FileSystemCatalog : public Catalog { public: - FileSystemCatalog(const std::shared_ptr& fs, const std::string& warehouse); + FileSystemCatalog(const std::shared_ptr& fs, const std::string& warehouse, + const std::map& catalog_options); Status CreateDatabase(const std::string& db_name, const std::map& options, @@ -61,6 +62,7 @@ class FileSystemCatalog : public Catalog { Result> LoadTableSchema(const Identifier& identifier) const override; std::string GetRootPath() const override; std::shared_ptr GetFileSystem() const override; + const std::map& GetOptions() const override; Result> GetTable(const Identifier& identifier) const override; Result> ListSnapshots(const Identifier& identifier, const std::string& branch) const override; @@ -92,6 +94,7 @@ class FileSystemCatalog : public Catalog { std::shared_ptr fs_; std::string warehouse_; + std::map catalog_options_; std::shared_ptr logger_; }; diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp new file mode 100644 index 000000000..4e91928bb --- /dev/null +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -0,0 +1,362 @@ +/* + * 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/table/system/global_system_tables.h" + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "arrow/api.h" +#include "paimon/catalog/catalog.h" +#include "paimon/catalog/identifier.h" +#include "paimon/common/data/binary_string.h" +#include "paimon/common/data/generic_row.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/core/core_options.h" +#include "paimon/core/schema/schema_manager.h" +#include "paimon/core/schema/table_schema.h" +#include "paimon/core/snapshot.h" +#include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/snapshot_manager.h" +#include "paimon/memory/memory_pool.h" +#include "paimon/status.h" + +namespace paimon { +namespace { + +// ============================================================================= +// Registry +// ============================================================================= + +using GlobalSystemTableFactory = + std::function>(const GlobalSystemTableContext&)>; + +struct GlobalSystemTableRegistryEntry { + std::string name; + GlobalSystemTableFactory factory; +}; + +const std::vector& GlobalSystemTableRegistry() { + static const std::vector registry = { + {CatalogOptionsSystemTable::kName, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {AllTableOptionsSystemTable::kName, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {TablesSystemTable::kName, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {PartitionsSystemTable::kName, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + }; + return registry; +} + +// ============================================================================= +// Helpers for sys.tables and sys.partitions +// ============================================================================= + +VariantType StringValue(const std::string& value) { + return BinaryString::FromString(value, GetDefaultPool().get()); +} + +VariantType OptionalStringValue(const std::optional& value) { + if (!value) { + return NullType(); + } + return StringValue(value.value()); +} + +VariantType OptionalInt64Value(const std::optional& value) { + if (!value) { + return NullType(); + } + return value.value(); +} + +} // namespace + +// ============================================================================= +// GlobalSystemTableLoader +// ============================================================================= + +bool GlobalSystemTableLoader::IsSupported(const std::string& table_name) { + std::string normalized = StringUtils::ToLowerCase(table_name); + for (const auto& entry : GlobalSystemTableRegistry()) { + if (entry.name == normalized) { + return true; + } + } + return false; +} + +Result> GlobalSystemTableLoader::Load( + const std::string& table_name, const GlobalSystemTableContext& context) { + std::string normalized = StringUtils::ToLowerCase(table_name); + for (const auto& entry : GlobalSystemTableRegistry()) { + if (entry.name == normalized) { + return entry.factory(context); + } + } + return Status::NotImplemented("unsupported global system table: ", table_name); +} + +std::vector GlobalSystemTableLoader::GetSupportedTableNames() { + std::vector names; + names.reserve(GlobalSystemTableRegistry().size()); + for (const auto& entry : GlobalSystemTableRegistry()) { + names.push_back(entry.name); + } + return names; +} + +// ============================================================================= +// sys.catalog_options +// ============================================================================= + +CatalogOptionsSystemTable::CatalogOptionsSystemTable(GlobalSystemTableContext context) + : InMemorySystemTable("sys/catalog_options"), context_(std::move(context)) {} + +std::string CatalogOptionsSystemTable::Name() const { + return kName; +} + +Result> CatalogOptionsSystemTable::ArrowSchema() const { + return arrow::schema({ + arrow::field("key", arrow::utf8(), /*nullable=*/false), + arrow::field("value", arrow::utf8(), /*nullable=*/false), + }); +} + +Result> CatalogOptionsSystemTable::BuildRows() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); + std::vector rows; + rows.reserve(context_.catalog_options.size()); + for (const auto& [key, value] : context_.catalog_options) { + GenericRow row(schema->num_fields()); + row.SetField(0, std::string_view(key)); + row.SetField(1, std::string_view(value)); + rows.push_back(std::move(row)); + } + return rows; +} + +// ============================================================================= +// sys.all_table_options +// ============================================================================= + +AllTableOptionsSystemTable::AllTableOptionsSystemTable(GlobalSystemTableContext context) + : InMemorySystemTable("sys/all_table_options"), context_(std::move(context)) {} + +std::string AllTableOptionsSystemTable::Name() const { + return kName; +} + +Result> AllTableOptionsSystemTable::ArrowSchema() const { + return arrow::schema({ + arrow::field("database_name", arrow::utf8(), /*nullable=*/false), + arrow::field("table_name", arrow::utf8(), /*nullable=*/false), + arrow::field("key", arrow::utf8(), /*nullable=*/false), + arrow::field("value", arrow::utf8(), /*nullable=*/false), + }); +} + +Result> AllTableOptionsSystemTable::BuildRows() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); + std::vector rows; + + PAIMON_ASSIGN_OR_RAISE(std::vector databases, + context_.catalog->ListDatabases()); + for (const auto& db : databases) { + PAIMON_ASSIGN_OR_RAISE(std::vector tables, + context_.catalog->ListTables(db)); + for (const auto& table : tables) { + Identifier id(db, table); + auto schema_result = context_.catalog->LoadTableSchema(id); + if (!schema_result.ok()) { + continue; // skip tables with errors (e.g. dropped concurrently) + } + auto schema_ptr = schema_result.ValueUnsafe(); + auto data_schema = std::dynamic_pointer_cast(schema_ptr); + if (!data_schema) { + continue; + } + for (const auto& [key, value] : data_schema->Options()) { + GenericRow row(schema->num_fields()); + row.SetField(0, std::string_view(db)); + row.SetField(1, std::string_view(table)); + row.SetField(2, std::string_view(key)); + row.SetField(3, std::string_view(value)); + rows.push_back(std::move(row)); + } + } + } + return rows; +} + +// ============================================================================= +// sys.tables +// ============================================================================= + +TablesSystemTable::TablesSystemTable(GlobalSystemTableContext context) + : InMemorySystemTable("sys/tables"), context_(std::move(context)) {} + +std::string TablesSystemTable::Name() const { + return kName; +} + +Result> TablesSystemTable::ArrowSchema() const { + return arrow::schema({ + arrow::field("database_name", arrow::utf8(), /*nullable=*/false), + arrow::field("table_name", arrow::utf8(), /*nullable=*/false), + arrow::field("table_type", arrow::utf8(), /*nullable=*/false), + arrow::field("partitioned", arrow::boolean(), /*nullable=*/false), + arrow::field("primary_key", arrow::utf8(), /*nullable=*/false), + arrow::field("record_count", arrow::int64(), /*nullable=*/true), + arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), + arrow::field("file_count", arrow::int64(), /*nullable=*/true), + arrow::field("last_file_creation_time", + arrow::timestamp(arrow::TimeUnit::MILLI), /*nullable=*/true), + }); +} + +Result> TablesSystemTable::BuildRows() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); + std::vector rows; + + PAIMON_ASSIGN_OR_RAISE(std::vector databases, + context_.catalog->ListDatabases()); + for (const auto& db : databases) { + PAIMON_ASSIGN_OR_RAISE(std::vector tables, + context_.catalog->ListTables(db)); + for (const auto& table : tables) { + Identifier id(db, table); + auto schema_result = context_.catalog->LoadTableSchema(id); + if (!schema_result.ok()) { + continue; + } + auto schema_ptr = schema_result.ValueUnsafe(); + auto data_schema = std::dynamic_pointer_cast(schema_ptr); + if (!data_schema) { + continue; + } + + // Determine table type + std::string table_type_str = "MANAGED"; + // Check if table has external path in options + const auto& options = data_schema->Options(); + // (simplified: could check for external path options) + + bool partitioned = !data_schema->PartitionKeys().empty(); + std::string primary_keys_str; + const auto& pks = data_schema->PrimaryKeys(); + for (size_t i = 0; i < pks.size(); ++i) { + if (i > 0) primary_keys_str += ","; + primary_keys_str += pks[i]; + } + + GenericRow row(schema->num_fields()); + row.SetField(0, std::string_view(db)); + row.SetField(1, std::string_view(table)); + row.SetField(2, StringValue(table_type_str)); + row.SetField(3, partitioned); + row.SetField(4, primary_keys_str.empty() + ? VariantType(NullType()) + : VariantType(StringValue(primary_keys_str))); + + // Try to get stats from latest snapshot + PAIMON_ASSIGN_OR_RAISE(std::string table_path, + context_.catalog->GetTableLocation(id)); + SnapshotManager snapshot_manager(context_.fs, table_path, + BranchManager::DEFAULT_MAIN_BRANCH); + auto snapshot_result = snapshot_manager.LatestSnapshot(); + if (snapshot_result.ok() && snapshot_result.ValueUnsafe()) { + const auto& snapshot = *snapshot_result.ValueUnsafe(); + row.SetField(5, OptionalInt64Value(snapshot.TotalRecordCount())); + // file_size and file_count not available from Snapshot alone; + // leave as null for now + row.SetField(6, NullType()); + row.SetField(7, NullType()); + row.SetField(8, NullType()); + } else { + row.SetField(5, NullType()); + row.SetField(6, NullType()); + row.SetField(7, NullType()); + row.SetField(8, NullType()); + } + + rows.push_back(std::move(row)); + } + } + return rows; +} + +// ============================================================================= +// sys.partitions +// ============================================================================= + +PartitionsSystemTable::PartitionsSystemTable(GlobalSystemTableContext context) + : InMemorySystemTable("sys/partitions"), context_(std::move(context)) {} + +std::string PartitionsSystemTable::Name() const { + return kName; +} + +Result> PartitionsSystemTable::ArrowSchema() const { + return arrow::schema({ + arrow::field("database_name", arrow::utf8(), /*nullable=*/false), + arrow::field("table_name", arrow::utf8(), /*nullable=*/false), + arrow::field("partition_name", arrow::utf8(), /*nullable=*/true), + arrow::field("record_count", arrow::int64(), /*nullable=*/true), + arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), + arrow::field("file_count", arrow::int64(), /*nullable=*/true), + arrow::field("last_update_time", + arrow::timestamp(arrow::TimeUnit::MILLI), /*nullable=*/true), + }); +} + +Result> PartitionsSystemTable::BuildRows() const { + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); + std::vector rows; + + // TODO(suxiaogang223): Implement partition-level aggregation using + // manifest entry reading (similar to FilesSystemTable::BuildRows() + // but grouped by partition). For now, return empty result set. + // + // The implementation should: + // 1. Enumerate all databases and tables + // 2. For each partitioned table, read latest snapshot's manifest entries + // 3. Group DataFileMeta entries by entry.Partition() + // 4. Aggregate: sum(file_size), sum(record_count), count files, + // max(creation_time) + + return rows; +} + +} // namespace paimon diff --git a/src/paimon/core/table/system/global_system_tables.h b/src/paimon/core/table/system/global_system_tables.h new file mode 100644 index 000000000..b3dc37570 --- /dev/null +++ b/src/paimon/core/table/system/global_system_tables.h @@ -0,0 +1,117 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include + +#include "paimon/core/table/system/in_memory_system_table.h" + +namespace paimon { +class Catalog; +class FileSystem; + +/// Context passed to global system table constructors, providing catalog-level +/// access for enumerating databases, tables, and reading metadata. +struct GlobalSystemTableContext { + Catalog* catalog; // non-owning pointer + std::shared_ptr fs; + std::string warehouse; + std::map catalog_options; +}; + +/// System table for `sys.catalog_options`, exposing catalog-level configuration +/// as key/value rows. +class CatalogOptionsSystemTable : public InMemorySystemTable { + public: + static constexpr const char* kName = "catalog_options"; + + explicit CatalogOptionsSystemTable(GlobalSystemTableContext context); + + std::string Name() const override; + Result> ArrowSchema() const override; + Result> BuildRows() const override; + + private: + GlobalSystemTableContext context_; +}; + +/// System table for `sys.all_table_options`, exposing all table options across +/// all databases as (database_name, table_name, key, value) rows. +class AllTableOptionsSystemTable : public InMemorySystemTable { + public: + static constexpr const char* kName = "all_table_options"; + + explicit AllTableOptionsSystemTable(GlobalSystemTableContext context); + + std::string Name() const override; + Result> ArrowSchema() const override; + Result> BuildRows() const override; + + private: + GlobalSystemTableContext context_; +}; + +/// System table for `sys.tables`, exposing metadata for all tables across all +/// databases including record counts and file statistics. +class TablesSystemTable : public InMemorySystemTable { + public: + static constexpr const char* kName = "tables"; + + explicit TablesSystemTable(GlobalSystemTableContext context); + + std::string Name() const override; + Result> ArrowSchema() const override; + Result> BuildRows() const override; + + private: + GlobalSystemTableContext context_; +}; + +/// System table for `sys.partitions`, exposing partition-level file statistics +/// for all tables across all databases. +class PartitionsSystemTable : public InMemorySystemTable { + public: + static constexpr const char* kName = "partitions"; + + explicit PartitionsSystemTable(GlobalSystemTableContext context); + + std::string Name() const override; + Result> ArrowSchema() const override; + Result> BuildRows() const override; + + private: + GlobalSystemTableContext context_; +}; + +/// Loader for global system tables under the `sys` database. +/// +/// Maintains its own registry with a factory signature that receives a +/// GlobalSystemTableContext instead of a per-table TableSchema. +class GlobalSystemTableLoader { + public: + static bool IsSupported(const std::string& table_name); + + static Result> Load( + const std::string& table_name, const GlobalSystemTableContext& context); + + static std::vector GetSupportedTableNames(); +}; + +} // namespace paimon diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp index f3a0a7f6b..ef4142c22 100644 --- a/src/paimon/core/table/system/system_table.cpp +++ b/src/paimon/core/table/system/system_table.cpp @@ -30,6 +30,7 @@ #include "paimon/core/schema/table_schema.h" #include "paimon/core/table/system/audit_log_system_table.h" #include "paimon/core/table/system/binlog_system_table.h" +#include "paimon/core/table/system/global_system_tables.h" #include "paimon/core/table/system/metadata_system_tables.h" #include "paimon/core/table/system/read_optimized_system_table.h" #include "paimon/core/utils/branch_manager.h" @@ -188,6 +189,17 @@ Result> SystemTableLoader::Load( Result> SystemTableLoader::TryParsePath(const std::string& path) { std::string table_name = PathUtil::GetName(path); + std::string parent = PathUtil::GetParentDirPath(path); + std::string parent_name = PathUtil::GetName(parent); + + // Detect global system table paths: /sys/ + if (parent_name == "sys") { + SystemTablePath system_table_path; + system_table_path.is_global = true; + system_table_path.system_table_name = table_name; + return std::optional(std::move(system_table_path)); + } + Identifier identifier(table_name); PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); if (!is_system_table) { @@ -197,7 +209,6 @@ Result> SystemTableLoader::TryParsePath(const std PAIMON_ASSIGN_OR_RAISE(std::optional branch, identifier.GetBranchName()); PAIMON_ASSIGN_OR_RAISE(std::optional system_table_name, identifier.GetSystemTableName()); - std::string parent = PathUtil::GetParentDirPath(path); SystemTablePath system_table_path; system_table_path.table_path = PathUtil::JoinPath(parent, data_table_name); system_table_path.branch = std::move(branch); @@ -213,6 +224,21 @@ Result> SystemTableLoader::LoadFromPath( return Status::Invalid("path is not a system table path: ", path); } const auto& parsed = system_table_path.value(); + + // Handle global system tables (under sys/ directory) + if (parsed.is_global) { + GlobalSystemTableContext context; + context.fs = fs; + // The warehouse is the grandparent of the sys/ path + context.warehouse = PathUtil::GetParentDirPath(PathUtil::GetParentDirPath(path)); + context.catalog_options = dynamic_options; + // Note: context.catalog is intentionally left as nullptr here. + // Global tables loaded from path do not have a Catalog reference and + // cannot enumerate databases/tables. Only tables that don't require + // catalog enumeration (e.g. catalog_options) will work in this path. + return GlobalSystemTableLoader::Load(parsed.system_table_name, context); + } + SchemaManager schema_manager(fs, parsed.table_path, parsed.branch.value_or(BranchManager::DEFAULT_MAIN_BRANCH)); PAIMON_ASSIGN_OR_RAISE(std::optional> latest_schema, diff --git a/src/paimon/core/table/system/system_table.h b/src/paimon/core/table/system/system_table.h index 5db20789e..3f897460c 100644 --- a/src/paimon/core/table/system/system_table.h +++ b/src/paimon/core/table/system/system_table.h @@ -42,6 +42,8 @@ struct SystemTablePath { std::optional branch; /// System table name, for example `options` or `snapshots`. std::string system_table_name; + /// Whether this is a global system table under the `sys` database. + bool is_global = false; }; /// Base interface for table-scoped system tables such as `T$options` and `T$snapshots`. From dd0dab9669244bbc69d0013b1d206678ac7db8f8 Mon Sep 17 00:00:00 2001 From: Socrates Date: Sun, 14 Jun 2026 00:03:56 +0800 Subject: [PATCH 02/13] fix: remove unused helper functions in global system tables --- .../core/table/system/global_system_tables.cpp | 18 +++--------------- 1 file changed, 3 insertions(+), 15 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 4e91928bb..d49e49936 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -86,20 +86,6 @@ VariantType StringValue(const std::string& value) { return BinaryString::FromString(value, GetDefaultPool().get()); } -VariantType OptionalStringValue(const std::optional& value) { - if (!value) { - return NullType(); - } - return StringValue(value.value()); -} - -VariantType OptionalInt64Value(const std::optional& value) { - if (!value) { - return NullType(); - } - return value.value(); -} - } // namespace // ============================================================================= @@ -298,7 +284,9 @@ Result> TablesSystemTable::BuildRows() const { auto snapshot_result = snapshot_manager.LatestSnapshot(); if (snapshot_result.ok() && snapshot_result.ValueUnsafe()) { const auto& snapshot = *snapshot_result.ValueUnsafe(); - row.SetField(5, OptionalInt64Value(snapshot.TotalRecordCount())); + auto total_count = snapshot.TotalRecordCount(); + row.SetField(5, total_count ? VariantType(total_count.value()) + : VariantType(NullType())); // file_size and file_count not available from Snapshot alone; // leave as null for now row.SetField(6, NullType()); From bac294dfd9fcf30ff2001634d3177f35abf4b59d Mon Sep 17 00:00:00 2001 From: Socrates Date: Sun, 14 Jun 2026 00:04:49 +0800 Subject: [PATCH 03/13] fix: replace ValueUnsafe() with value() in global system tables --- src/paimon/core/table/system/global_system_tables.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index d49e49936..2b158c1b1 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -188,7 +188,7 @@ Result> AllTableOptionsSystemTable::BuildRows() const { if (!schema_result.ok()) { continue; // skip tables with errors (e.g. dropped concurrently) } - auto schema_ptr = schema_result.ValueUnsafe(); + auto schema_ptr = schema_result.value(); auto data_schema = std::dynamic_pointer_cast(schema_ptr); if (!data_schema) { continue; @@ -247,7 +247,7 @@ Result> TablesSystemTable::BuildRows() const { if (!schema_result.ok()) { continue; } - auto schema_ptr = schema_result.ValueUnsafe(); + auto schema_ptr = schema_result.value(); auto data_schema = std::dynamic_pointer_cast(schema_ptr); if (!data_schema) { continue; @@ -282,8 +282,8 @@ Result> TablesSystemTable::BuildRows() const { SnapshotManager snapshot_manager(context_.fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH); auto snapshot_result = snapshot_manager.LatestSnapshot(); - if (snapshot_result.ok() && snapshot_result.ValueUnsafe()) { - const auto& snapshot = *snapshot_result.ValueUnsafe(); + if (snapshot_result.ok() && snapshot_result.value()) { + const auto& snapshot = *snapshot_result.value(); auto total_count = snapshot.TotalRecordCount(); row.SetField(5, total_count ? VariantType(total_count.value()) : VariantType(NullType())); From 143c550319fb8e57fec387144e6a9f5cdc5a71cc Mon Sep 17 00:00:00 2001 From: Socrates Date: Sun, 14 Jun 2026 00:06:05 +0800 Subject: [PATCH 04/13] fix: add default value for catalog_options in FileSystemCatalog constructor --- src/paimon/core/catalog/file_system_catalog.h | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/paimon/core/catalog/file_system_catalog.h b/src/paimon/core/catalog/file_system_catalog.h index 3b93dc650..2a4656cc1 100644 --- a/src/paimon/core/catalog/file_system_catalog.h +++ b/src/paimon/core/catalog/file_system_catalog.h @@ -39,7 +39,7 @@ class Logger; class FileSystemCatalog : public Catalog { public: FileSystemCatalog(const std::shared_ptr& fs, const std::string& warehouse, - const std::map& catalog_options); + const std::map& catalog_options = {}); Status CreateDatabase(const std::string& db_name, const std::map& options, From ac0bce036a6acba6c70c50b6cad017a56a760f85 Mon Sep 17 00:00:00 2001 From: Socrates Date: Sun, 14 Jun 2026 00:08:25 +0800 Subject: [PATCH 05/13] fix: update TestInvalidList to expect global system table names --- .../core/catalog/file_system_catalog_test.cpp | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp index 8377493e5..a84296221 100644 --- a/src/paimon/core/catalog/file_system_catalog_test.cpp +++ b/src/paimon/core/catalog/file_system_catalog_test.cpp @@ -16,6 +16,8 @@ #include "paimon/core/catalog/file_system_catalog.h" +#include + #include "arrow/api.h" #include "arrow/c/abi.h" #include "arrow/c/bridge.h" @@ -606,8 +608,16 @@ TEST(FileSystemCatalogTest, TestInvalidList) { auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); - ASSERT_NOK_WITH_MSG(catalog.ListTables("sys"), - "do not support listing tables for system database."); + ASSERT_OK_AND_ASSIGN(auto sys_tables, catalog.ListTables("sys")); + ASSERT_FALSE(sys_tables.empty()); + // Verify expected global system table names are present + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "catalog_options") != + sys_tables.end()); + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "all_table_options") != + sys_tables.end()); + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "tables") != sys_tables.end()); + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "partitions") != + sys_tables.end()); } TEST(FileSystemCatalogTest, TestValidateTableSchema) { From c1c03a9c73632509ff4574ba7222c913f285c634 Mon Sep 17 00:00:00 2001 From: Socrates Date: Sun, 14 Jun 2026 00:42:16 +0800 Subject: [PATCH 06/13] improve: add table_type detection and clearer TODO for manifest-dependent fields --- .../core/table/system/global_system_tables.cpp | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 2b158c1b1..4da161a30 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -33,6 +33,7 @@ #include "paimon/common/utils/string_utils.h" #include "paimon/common/utils/path_util.h" #include "paimon/core/core_options.h" +#include "paimon/defs.h" #include "paimon/core/schema/schema_manager.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/snapshot.h" @@ -253,11 +254,12 @@ Result> TablesSystemTable::BuildRows() const { continue; } - // Determine table type + // Determine table type: EXTERNAL if data-file.external-paths is set std::string table_type_str = "MANAGED"; - // Check if table has external path in options - const auto& options = data_schema->Options(); - // (simplified: could check for external path options) + const auto& opts = data_schema->Options(); + if (opts.find(Options::DATA_FILE_EXTERNAL_PATHS) != opts.end()) { + table_type_str = "EXTERNAL"; + } bool partitioned = !data_schema->PartitionKeys().empty(); std::string primary_keys_str; @@ -287,8 +289,10 @@ Result> TablesSystemTable::BuildRows() const { auto total_count = snapshot.TotalRecordCount(); row.SetField(5, total_count ? VariantType(total_count.value()) : VariantType(NullType())); - // file_size and file_count not available from Snapshot alone; - // leave as null for now + // TODO(suxiaogang223): Populate file_size_in_bytes, file_count, and + // last_file_creation_time by reading manifest entries. This requires + // the manifest reading infrastructure from the files/manifests system + // tables PR (codex/system-table-files-manifests-pr4). row.SetField(6, NullType()); row.SetField(7, NullType()); row.SetField(8, NullType()); From 918c23169deece49bf1c8e75afca63c60b2d5d01 Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 2 Jul 2026 16:58:15 +0800 Subject: [PATCH 07/13] feat: populate sys.tables file stats via manifest reading Replace the TODO stubs in TablesSystemTable::BuildRows() with actual manifest entry aggregation. The new AggregateFileStats() helper reads the latest snapshot data files and computes record_count, file_size, file_count, and last_file_creation_time. Co-Authored-By: Claude Fable 5 --- .../table/system/global_system_tables.cpp | 153 ++++++++++++++++-- 1 file changed, 137 insertions(+), 16 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 4da161a30..1121a6820 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -34,11 +34,21 @@ #include "paimon/common/utils/path_util.h" #include "paimon/core/core_options.h" #include "paimon/defs.h" +#include "paimon/core/io/data_file_meta.h" +#include "paimon/core/manifest/file_entry.h" +#include "paimon/core/manifest/file_kind.h" +#include "paimon/core/manifest/manifest_entry.h" +#include "paimon/core/manifest/manifest_file.h" +#include "paimon/core/manifest/manifest_file_meta.h" +#include "paimon/core/manifest/manifest_list.h" #include "paimon/core/schema/schema_manager.h" #include "paimon/core/schema/table_schema.h" #include "paimon/core/snapshot.h" #include "paimon/core/utils/branch_manager.h" +#include "paimon/core/utils/field_mapping.h" +#include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" +#include "paimon/data/timestamp.h" #include "paimon/memory/memory_pool.h" #include "paimon/status.h" @@ -87,6 +97,111 @@ VariantType StringValue(const std::string& value) { return BinaryString::FromString(value, GetDefaultPool().get()); } +// Aggregated file-level statistics for a table or partition. +struct FileStats { + int64_t record_count = 0; + int64_t file_size_in_bytes = 0; + int64_t file_count = 0; + int64_t last_file_creation_time_millis = 0; +}; + +// Read the latest snapshot's data files and aggregate statistics. +// Returns an empty map if no snapshot or no data files exist. +Result> AggregateFileStats( + const std::shared_ptr& fs, const std::string& table_path, + const std::map& options) { + std::map result; + + SnapshotManager snapshot_manager(fs, table_path, + BranchManager::DEFAULT_MAIN_BRANCH); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, + snapshot_manager.LatestSnapshot()); + if (!snapshot) { + return result; + } + + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, + CoreOptions::FromMap(options)); + + // Use SchemaManager to load the latest schema for field/partition info + SchemaManager schema_mgr(fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH); + auto latest_schema_result = schema_mgr.Latest(); + if (!latest_schema_result.ok() || !latest_schema_result.value()) { + return result; + } + auto table_schema = *latest_schema_result.value(); + + auto pool = GetDefaultPool(); + + std::shared_ptr arrow_schema = + DataField::ConvertDataFieldsToArrowSchema(table_schema->Fields()); + PAIMON_ASSIGN_OR_RAISE(std::vector external_paths, + core_options.CreateExternalPaths()); + PAIMON_ASSIGN_OR_RAISE(std::optional global_index_external_path, + core_options.CreateGlobalIndexExternalPath()); + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr path_factory, + FileStorePathFactory::Create( + table_path, arrow_schema, table_schema->PartitionKeys(), + core_options.GetPartitionDefaultName(), + core_options.GetFileFormat()->Identifier(), + core_options.DataFilePrefix(), + core_options.LegacyPartitionNameEnabled(), external_paths, + global_index_external_path, core_options.IndexFileInDataFileDir(), pool)); + + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr manifest_list, + ManifestList::Create(fs, core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, pool)); + + std::vector manifests; + PAIMON_RETURN_NOT_OK( + manifest_list->ReadDataManifests(*snapshot, &manifests)); + + PAIMON_ASSIGN_OR_RAISE( + std::shared_ptr partition_schema, + FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); + PAIMON_ASSIGN_OR_RAISE( + std::unique_ptr manifest_file, + ManifestFile::Create(fs, core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetManifestTargetFileSize(), pool, + core_options, partition_schema)); + + std::vector entries; + for (const auto& manifest : manifests) { + PAIMON_RETURN_NOT_OK( + manifest_file->Read(manifest.FileName(), /*filter=*/nullptr, &entries)); + } + + std::vector merged_entries; + PAIMON_RETURN_NOT_OK(FileEntry::MergeEntries(entries, &merged_entries)); + + for (const auto& entry : merged_entries) { + if (!(entry.Kind() == FileKind::Add())) { + continue; + } + const auto& file = entry.File(); + + // Use empty string key for unpartitioned tables + std::string partition_key; + if (entry.Partition().GetFieldCount() > 0) { + partition_key = "partitioned"; + } + + auto& stats = result[partition_key]; + stats.record_count += file->row_count; + stats.file_size_in_bytes += file->file_size; + stats.file_count++; + int64_t creation_millis = file->creation_time.GetMillisecond(); + if (creation_millis > stats.last_file_creation_time_millis) { + stats.last_file_creation_time_millis = creation_millis; + } + } + + return result; +} + } // namespace // ============================================================================= @@ -278,24 +393,30 @@ Result> TablesSystemTable::BuildRows() const { ? VariantType(NullType()) : VariantType(StringValue(primary_keys_str))); - // Try to get stats from latest snapshot + // Get table path and aggregate file stats from manifest entries PAIMON_ASSIGN_OR_RAISE(std::string table_path, context_.catalog->GetTableLocation(id)); - SnapshotManager snapshot_manager(context_.fs, table_path, - BranchManager::DEFAULT_MAIN_BRANCH); - auto snapshot_result = snapshot_manager.LatestSnapshot(); - if (snapshot_result.ok() && snapshot_result.value()) { - const auto& snapshot = *snapshot_result.value(); - auto total_count = snapshot.TotalRecordCount(); - row.SetField(5, total_count ? VariantType(total_count.value()) - : VariantType(NullType())); - // TODO(suxiaogang223): Populate file_size_in_bytes, file_count, and - // last_file_creation_time by reading manifest entries. This requires - // the manifest reading infrastructure from the files/manifests system - // tables PR (codex/system-table-files-manifests-pr4). - row.SetField(6, NullType()); - row.SetField(7, NullType()); - row.SetField(8, NullType()); + + auto file_stats_result = + AggregateFileStats(context_.fs, table_path, data_schema->Options()); + if (file_stats_result.ok()) { + auto& all_stats = file_stats_result.value(); + int64_t total_record = 0, total_size = 0, total_files = 0, + max_creation = 0; + for (const auto& [key, stats] : all_stats) { + total_record += stats.record_count; + total_size += stats.file_size_in_bytes; + total_files += stats.file_count; + if (stats.last_file_creation_time_millis > max_creation) { + max_creation = stats.last_file_creation_time_millis; + } + } + row.SetField(5, VariantType(total_record)); + row.SetField(6, VariantType(total_size)); + row.SetField(7, VariantType(total_files)); + row.SetField(8, max_creation > 0 + ? VariantType(Timestamp::FromEpochMillis(max_creation)) + : VariantType(NullType())); } else { row.SetField(5, NullType()); row.SetField(6, NullType()); From daffd92ba960df1a3e757d71fb582f9d9fef0b50 Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 2 Jul 2026 17:36:58 +0800 Subject: [PATCH 08/13] feat: implement sys.partitions with partition-level aggregation Replace the stub with actual partition-level file statistics using the AggregateFileStats helper. For each partitioned table, read manifest entries and emit one row per partition with record_count, file_size, file_count, and last_update_time. Co-Authored-By: Claude Fable 5 --- .../table/system/global_system_tables.cpp | 80 ++++++++++++++++--- 1 file changed, 67 insertions(+), 13 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 1121a6820..55bc196df 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -30,6 +30,7 @@ #include "paimon/catalog/identifier.h" #include "paimon/common/data/binary_string.h" #include "paimon/common/data/generic_row.h" +#include "paimon/common/utils/binary_row_partition_computer.h" #include "paimon/common/utils/string_utils.h" #include "paimon/common/utils/path_util.h" #include "paimon/core/core_options.h" @@ -152,7 +153,8 @@ Result> AggregateFileStats( PAIMON_ASSIGN_OR_RAISE( std::unique_ptr manifest_list, ManifestList::Create(fs, core_options.GetManifestFormat(), - core_options.GetManifestCompression(), path_factory, pool)); + core_options.GetManifestCompression(), path_factory, + core_options.GetCache(), pool)); std::vector manifests; PAIMON_RETURN_NOT_OK( @@ -183,10 +185,16 @@ Result> AggregateFileStats( } const auto& file = entry.File(); - // Use empty string key for unpartitioned tables + // Convert partition BinaryRow to string representation std::string partition_key; if (entry.Partition().GetFieldCount() > 0) { - partition_key = "partitioned"; + PAIMON_ASSIGN_OR_RAISE( + partition_key, + BinaryRowPartitionComputer::PartToSimpleString( + partition_schema, entry.Partition(), ",", + /*max_length=*/255, + /*legacy_partition_name_enabled=*/false)); + partition_key = "{" + partition_key + "}"; } auto& stats = result[partition_key]; @@ -458,17 +466,63 @@ Result> PartitionsSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); std::vector rows; - // TODO(suxiaogang223): Implement partition-level aggregation using - // manifest entry reading (similar to FilesSystemTable::BuildRows() - // but grouped by partition). For now, return empty result set. - // - // The implementation should: - // 1. Enumerate all databases and tables - // 2. For each partitioned table, read latest snapshot's manifest entries - // 3. Group DataFileMeta entries by entry.Partition() - // 4. Aggregate: sum(file_size), sum(record_count), count files, - // max(creation_time) + PAIMON_ASSIGN_OR_RAISE(std::vector databases, + context_.catalog->ListDatabases()); + for (const auto& db : databases) { + PAIMON_ASSIGN_OR_RAISE(std::vector tables, + context_.catalog->ListTables(db)); + for (const auto& table : tables) { + Identifier id(db, table); + auto schema_result = context_.catalog->LoadTableSchema(id); + if (!schema_result.ok()) { + continue; + } + auto schema_ptr = schema_result.value(); + auto data_schema = std::dynamic_pointer_cast(schema_ptr); + if (!data_schema) { + continue; + } + // Only emit rows for partitioned tables + if (data_schema->PartitionKeys().empty()) { + continue; + } + + // Get table path and aggregate file stats by partition + auto table_path_result = context_.catalog->GetTableLocation(id); + if (!table_path_result.ok()) { + continue; + } + std::string table_path = table_path_result.value(); + + auto file_stats_result = + AggregateFileStats(context_.fs, table_path, data_schema->Options()); + if (!file_stats_result.ok()) { + continue; + } + + auto& stats_map = file_stats_result.value(); + for (const auto& [partition_key, stats] : stats_map) { + if (stats.file_count == 0) { + continue; + } + GenericRow row(schema->num_fields()); + row.SetField(0, std::string_view(db)); + row.SetField(1, std::string_view(table)); + row.SetField(2, partition_key.empty() + ? VariantType(NullType()) + : VariantType(StringValue(partition_key))); + row.SetField(3, VariantType(stats.record_count)); + row.SetField(4, VariantType(stats.file_size_in_bytes)); + row.SetField(5, VariantType(stats.file_count)); + row.SetField(6, stats.last_file_creation_time_millis > 0 + ? VariantType(Timestamp::FromEpochMillis( + stats.last_file_creation_time_millis)) + : VariantType(NullType())); + rows.push_back(std::move(row)); + } + } + } return rows; } From 00ee7dd14d482a16d277571d5f03d65dddcd3c1f Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 2 Jul 2026 20:59:58 +0800 Subject: [PATCH 09/13] test: add integration tests for global system tables Add 4 tests to SystemTableReadInteTest: - TestReadGlobalCatalogOptions: verifies sys.catalog_options schema and content - TestReadGlobalAllTableOptions: verifies sys.all_table_options with table options - TestReadGlobalTables: verifies sys.tables schema, table_type, partitioned, pk - TestReadGlobalPartitions: verifies sys.partitions returns empty for unpartitioned Tests use a ReadGlobalSystemTable helper that creates the GlobalSystemTableContext with a proper Catalog pointer. Co-Authored-By: Claude Fable 5 --- .../table/system/global_system_tables.cpp | 16 +- test/inte/read_inte_test.cpp | 237 ++++++++++++++++++ 2 files changed, 245 insertions(+), 8 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 55bc196df..e0c0a592a 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -319,10 +319,10 @@ Result> AllTableOptionsSystemTable::BuildRows() const { } for (const auto& [key, value] : data_schema->Options()) { GenericRow row(schema->num_fields()); - row.SetField(0, std::string_view(db)); - row.SetField(1, std::string_view(table)); - row.SetField(2, std::string_view(key)); - row.SetField(3, std::string_view(value)); + row.SetField(0, StringValue(db)); + row.SetField(1, StringValue(table)); + row.SetField(2, StringValue(key)); + row.SetField(3, StringValue(value)); rows.push_back(std::move(row)); } } @@ -393,8 +393,8 @@ Result> TablesSystemTable::BuildRows() const { } GenericRow row(schema->num_fields()); - row.SetField(0, std::string_view(db)); - row.SetField(1, std::string_view(table)); + row.SetField(0, StringValue(db)); + row.SetField(1, StringValue(table)); row.SetField(2, StringValue(table_type_str)); row.SetField(3, partitioned); row.SetField(4, primary_keys_str.empty() @@ -507,8 +507,8 @@ Result> PartitionsSystemTable::BuildRows() const { continue; } GenericRow row(schema->num_fields()); - row.SetField(0, std::string_view(db)); - row.SetField(1, std::string_view(table)); + row.SetField(0, StringValue(db)); + row.SetField(1, StringValue(table)); row.SetField(2, partition_key.empty() ? VariantType(NullType()) : VariantType(StringValue(partition_key))); diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 5bbede302..fb3aa96ef 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -14,6 +14,7 @@ * limitations under the License. */ +#include #include #include #include @@ -53,6 +54,7 @@ #include "paimon/data/decimal.h" #include "paimon/data/timestamp.h" #include "paimon/defs.h" +#include "paimon/core/table/system/global_system_tables.h" #include "paimon/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" @@ -3717,4 +3719,239 @@ TEST_P(ReadInteTest, TestSpecificFs) { ASSERT_GT(io_count, 0); } +// ============================================================================= +// Global System Table Tests +// ============================================================================= + +namespace { + +Result ReadGlobalSystemTable( + const std::string& table_name, Catalog* catalog, + const std::shared_ptr& fs, const std::string& warehouse, + const std::map& options) { + GlobalSystemTableContext ctx; + ctx.catalog = catalog; + ctx.fs = fs; + ctx.warehouse = warehouse; + ctx.catalog_options = options; + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr system_table, + GlobalSystemTableLoader::Load(table_name, ctx)); + + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, + system_table->ArrowSchema()); + + std::string sys_path = PathUtil::JoinPath(PathUtil::JoinPath(warehouse, "sys"), table_name); + + ScanContextBuilder scan_context_builder(sys_path); + scan_context_builder.SetOptions(options); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr scan_context, + scan_context_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_scan, + system_table->NewScan(scan_context)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr plan, table_scan->CreatePlan()); + + ReadContextBuilder read_context_builder(sys_path); + read_context_builder.SetOptions(options); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr read_context, + read_context_builder.Finish()); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr table_read, + system_table->NewRead(read_context)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr batch_reader, + table_read->CreateReader(plan->Splits())); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr result, + ReadResultCollector::CollectResult(batch_reader.get())); + return SystemTableReadResult(std::move(batch_reader), result); +} + +} // namespace + +TEST(SystemTableReadInteTest, TestReadGlobalCatalogOptions) { + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {"custom.catalog.option", "test-value"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); + ASSERT_OK_AND_ASSIGN(auto result, + ReadGlobalSystemTable("catalog_options", catalog.get(), + catalog->GetFileSystem(), warehouse, options)); + auto struct_array = SingleStructChunk(result); + ASSERT_TRUE(struct_array); + auto key_array = std::dynamic_pointer_cast(struct_array->field(0)); + auto value_array = std::dynamic_pointer_cast(struct_array->field(1)); + ASSERT_TRUE(key_array); + ASSERT_TRUE(value_array); + + // Build a map from the result + std::map result_map; + for (int64_t i = 0; i < struct_array->length(); ++i) { + result_map[key_array->GetString(i)] = value_array->GetString(i); + } + ASSERT_EQ(result_map["file-system"], "local"); + ASSERT_EQ(result_map["file.format"], "orc"); +} + +TEST(SystemTableReadInteTest, TestReadGlobalAllTableOptions) { + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {Options::MANIFEST_FORMAT, "orc"}, + {"table.option.custom", "my-value"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); + // Create a database and table + ASSERT_OK(catalog->CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())}); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok()); + ASSERT_OK(catalog->CreateTable(Identifier("test_db", "test_tbl"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&schema); + + // Verify basic enumeration works + ASSERT_OK_AND_ASSIGN(auto dbs, catalog->ListDatabases()); + ASSERT_TRUE(std::find(dbs.begin(), dbs.end(), "test_db") != dbs.end()); + ASSERT_OK_AND_ASSIGN(auto tbls, catalog->ListTables("test_db")); + ASSERT_TRUE(std::find(tbls.begin(), tbls.end(), "test_tbl") != tbls.end()); + + // Verify schema loads and has Options + ASSERT_OK_AND_ASSIGN(auto loaded_schema, + catalog->LoadTableSchema(Identifier("test_db", "test_tbl"))); + auto ds = std::dynamic_pointer_cast(loaded_schema); + ASSERT_TRUE(ds != nullptr) << "LoadTableSchema did not return DataSchema"; + ASSERT_FALSE(ds->Options().empty()) << "Table schema has no options"; + + // Directly test BuildRows + { + GlobalSystemTableContext ctx; + ctx.catalog = catalog.get(); + ctx.fs = catalog->GetFileSystem(); + ctx.warehouse = warehouse; + ctx.catalog_options = options; + AllTableOptionsSystemTable table(ctx); + ASSERT_OK_AND_ASSIGN(auto rows, table.BuildRows()); + ASSERT_GT(rows.size(), 0) << "BuildRows returned empty, expected at least 1 row"; + } + + ASSERT_OK_AND_ASSIGN(auto result, + ReadGlobalSystemTable("all_table_options", catalog.get(), + catalog->GetFileSystem(), warehouse, options)); + auto struct_array = SingleStructChunk(result); + ASSERT_TRUE(struct_array); + ASSERT_GE(struct_array->length(), 1) << "result has " << struct_array->length() << " rows"; + + auto db_array = std::dynamic_pointer_cast(struct_array->field(0)); + auto tbl_array = std::dynamic_pointer_cast(struct_array->field(1)); + auto key_array = std::dynamic_pointer_cast(struct_array->field(2)); + auto val_array = std::dynamic_pointer_cast(struct_array->field(3)); + ASSERT_TRUE(db_array); + ASSERT_TRUE(tbl_array); + ASSERT_TRUE(key_array); + ASSERT_TRUE(val_array); + + // Verify that our table's options appear in the result + bool found_db = false; + bool found_format = false; + for (int64_t i = 0; i < struct_array->length(); ++i) { + auto db_name = std::string(db_array->GetString(i)); + auto tbl_name = std::string(tbl_array->GetString(i)); + if (db_name == "test_db" && tbl_name == "test_tbl") { + found_db = true; + auto key_str = std::string(key_array->GetString(i)); + if (key_str == "file.format") { + EXPECT_EQ(std::string(val_array->GetString(i)), "orc"); + found_format = true; + } + } + } + ASSERT_TRUE(found_db) << "test_db.test_tbl not found in sys.all_table_options"; + ASSERT_TRUE(found_format) << "file.format option not found in sys.all_table_options"; +} + +TEST(SystemTableReadInteTest, TestReadGlobalTables) { + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {Options::MANIFEST_FORMAT, "orc"}, + {Options::BUCKET, "1"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); + auto fs = catalog->GetFileSystem(); + + // Create a database and a PK table + ASSERT_OK(catalog->CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + 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("test_db", "test_tbl"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{"pk"}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&schema); + + ASSERT_OK_AND_ASSIGN(auto result, + ReadGlobalSystemTable("tables", catalog.get(), fs, warehouse, options)); + auto struct_array = SingleStructChunk(result); + ASSERT_TRUE(struct_array); + ASSERT_GE(struct_array->length(), 1); + + auto db_array = std::dynamic_pointer_cast(struct_array->field(0)); + auto tbl_array = std::dynamic_pointer_cast(struct_array->field(1)); + auto type_array = std::dynamic_pointer_cast(struct_array->field(2)); + auto part_array = std::dynamic_pointer_cast(struct_array->field(3)); + auto pk_array = std::dynamic_pointer_cast(struct_array->field(4)); + ASSERT_TRUE(db_array); + ASSERT_TRUE(tbl_array); + ASSERT_TRUE(type_array); + ASSERT_TRUE(part_array); + ASSERT_TRUE(pk_array); + + // Find our table by table name + bool found = false; + for (int64_t i = 0; i < struct_array->length(); ++i) { + if (std::string(tbl_array->GetString(i)) == "test_tbl") { + EXPECT_EQ(std::string(db_array->GetString(i)), "test_db"); + EXPECT_EQ(std::string(type_array->GetString(i)), "MANAGED"); + EXPECT_FALSE(part_array->Value(i)); + // pk is stored as BinaryString; check not null + EXPECT_FALSE(pk_array->IsNull(i)); + found = true; + } + } + ASSERT_TRUE(found) << "table not found in sys.tables"; +} + +TEST(SystemTableReadInteTest, TestReadGlobalPartitions) { + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {Options::MANIFEST_FORMAT, "orc"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); + auto fs = catalog->GetFileSystem(); + + // Create a database and an unpartitioned table (no partitions → empty result) + ASSERT_OK(catalog->CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); + auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())}); + ::ArrowSchema schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok()); + ASSERT_OK(catalog->CreateTable(Identifier("test_db", "test_tbl"), &schema, + /*partition_keys=*/{}, /*primary_keys=*/{}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&schema); + + // Unpartitioned tables are skipped by sys.partitions → empty result. + // CollectResult returns null shared_ptr when result is empty. + ASSERT_OK_AND_ASSIGN(auto part_result, + ReadGlobalSystemTable("partitions", catalog.get(), fs, warehouse, options)); + ASSERT_FALSE(part_result.array) << "expected null array for empty partitions result"; +} + } // namespace paimon::test From 31f52b54350ce12a29bf912ef86776c295b57e5a Mon Sep 17 00:00:00 2001 From: Socrates Date: Thu, 2 Jul 2026 21:06:19 +0800 Subject: [PATCH 10/13] style: apply clang-format fixes from pre-commit Co-Authored-By: Claude Fable 5 --- .../core/catalog/file_system_catalog.cpp | 5 +- src/paimon/core/catalog/file_system_catalog.h | 2 +- .../core/catalog/file_system_catalog_test.cpp | 3 +- .../table/system/global_system_tables.cpp | 97 ++++++++----------- .../core/table/system/global_system_tables.h | 4 +- test/inte/read_inte_test.cpp | 11 +-- 6 files changed, 50 insertions(+), 72 deletions(-) diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp index eab4ae666..a1e9a9f65 100644 --- a/src/paimon/core/catalog/file_system_catalog.cpp +++ b/src/paimon/core/catalog/file_system_catalog.cpp @@ -283,9 +283,8 @@ Result> FileSystemCatalog::LoadTableSchema( context.fs = fs_; context.warehouse = warehouse_; context.catalog_options = catalog_options_; - PAIMON_ASSIGN_OR_RAISE( - std::shared_ptr system_table, - GlobalSystemTableLoader::Load(identifier.GetTableName(), context)); + PAIMON_ASSIGN_OR_RAISE(std::shared_ptr system_table, + GlobalSystemTableLoader::Load(identifier.GetTableName(), context)); PAIMON_ASSIGN_OR_RAISE(std::shared_ptr arrow_schema, system_table->ArrowSchema()); return std::make_shared(std::move(arrow_schema)); diff --git a/src/paimon/core/catalog/file_system_catalog.h b/src/paimon/core/catalog/file_system_catalog.h index 2a4656cc1..7a2ad4a17 100644 --- a/src/paimon/core/catalog/file_system_catalog.h +++ b/src/paimon/core/catalog/file_system_catalog.h @@ -39,7 +39,7 @@ class Logger; class FileSystemCatalog : public Catalog { public: FileSystemCatalog(const std::shared_ptr& fs, const std::string& warehouse, - const std::map& catalog_options = {}); + const std::map& catalog_options = {}); Status CreateDatabase(const std::string& db_name, const std::map& options, diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp index a84296221..f970fddb6 100644 --- a/src/paimon/core/catalog/file_system_catalog_test.cpp +++ b/src/paimon/core/catalog/file_system_catalog_test.cpp @@ -616,8 +616,7 @@ TEST(FileSystemCatalogTest, TestInvalidList) { ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "all_table_options") != sys_tables.end()); ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "tables") != sys_tables.end()); - ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "partitions") != - sys_tables.end()); + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "partitions") != sys_tables.end()); } TEST(FileSystemCatalogTest, TestValidateTableSchema) { diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index e0c0a592a..3e0a3e5df 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -31,10 +31,9 @@ #include "paimon/common/data/binary_string.h" #include "paimon/common/data/generic_row.h" #include "paimon/common/utils/binary_row_partition_computer.h" -#include "paimon/common/utils/string_utils.h" #include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" #include "paimon/core/core_options.h" -#include "paimon/defs.h" #include "paimon/core/io/data_file_meta.h" #include "paimon/core/manifest/file_entry.h" #include "paimon/core/manifest/file_kind.h" @@ -50,6 +49,7 @@ #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" #include "paimon/data/timestamp.h" +#include "paimon/defs.h" #include "paimon/memory/memory_pool.h" #include "paimon/status.h" @@ -113,16 +113,13 @@ Result> AggregateFileStats( const std::map& options) { std::map result; - SnapshotManager snapshot_manager(fs, table_path, - BranchManager::DEFAULT_MAIN_BRANCH); - PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, - snapshot_manager.LatestSnapshot()); + SnapshotManager snapshot_manager(fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH); + PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, snapshot_manager.LatestSnapshot()); if (!snapshot) { return result; } - PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, - CoreOptions::FromMap(options)); + PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options)); // Use SchemaManager to load the latest schema for field/partition info SchemaManager schema_mgr(fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH); @@ -144,31 +141,27 @@ Result> AggregateFileStats( std::shared_ptr path_factory, FileStorePathFactory::Create( table_path, arrow_schema, table_schema->PartitionKeys(), - core_options.GetPartitionDefaultName(), - core_options.GetFileFormat()->Identifier(), - core_options.DataFilePrefix(), - core_options.LegacyPartitionNameEnabled(), external_paths, - global_index_external_path, core_options.IndexFileInDataFileDir(), pool)); + core_options.GetPartitionDefaultName(), core_options.GetFileFormat()->Identifier(), + core_options.DataFilePrefix(), core_options.LegacyPartitionNameEnabled(), + external_paths, global_index_external_path, core_options.IndexFileInDataFileDir(), + pool)); - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr manifest_list, - ManifestList::Create(fs, core_options.GetManifestFormat(), - core_options.GetManifestCompression(), path_factory, - core_options.GetCache(), pool)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr manifest_list, + ManifestList::Create(fs, core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetCache(), pool)); std::vector manifests; - PAIMON_RETURN_NOT_OK( - manifest_list->ReadDataManifests(*snapshot, &manifests)); + PAIMON_RETURN_NOT_OK(manifest_list->ReadDataManifests(*snapshot, &manifests)); PAIMON_ASSIGN_OR_RAISE( std::shared_ptr partition_schema, FieldMapping::GetPartitionSchema(arrow_schema, table_schema->PartitionKeys())); - PAIMON_ASSIGN_OR_RAISE( - std::unique_ptr manifest_file, - ManifestFile::Create(fs, core_options.GetManifestFormat(), - core_options.GetManifestCompression(), path_factory, - core_options.GetManifestTargetFileSize(), pool, - core_options, partition_schema)); + PAIMON_ASSIGN_OR_RAISE(std::unique_ptr manifest_file, + ManifestFile::Create(fs, core_options.GetManifestFormat(), + core_options.GetManifestCompression(), path_factory, + core_options.GetManifestTargetFileSize(), pool, + core_options, partition_schema)); std::vector entries; for (const auto& manifest : manifests) { @@ -188,12 +181,10 @@ Result> AggregateFileStats( // Convert partition BinaryRow to string representation std::string partition_key; if (entry.Partition().GetFieldCount() > 0) { - PAIMON_ASSIGN_OR_RAISE( - partition_key, - BinaryRowPartitionComputer::PartToSimpleString( - partition_schema, entry.Partition(), ",", - /*max_length=*/255, - /*legacy_partition_name_enabled=*/false)); + PAIMON_ASSIGN_OR_RAISE(partition_key, BinaryRowPartitionComputer::PartToSimpleString( + partition_schema, entry.Partition(), ",", + /*max_length=*/255, + /*legacy_partition_name_enabled=*/false)); partition_key = "{" + partition_key + "}"; } @@ -301,11 +292,9 @@ Result> AllTableOptionsSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); std::vector rows; - PAIMON_ASSIGN_OR_RAISE(std::vector databases, - context_.catalog->ListDatabases()); + PAIMON_ASSIGN_OR_RAISE(std::vector databases, context_.catalog->ListDatabases()); for (const auto& db : databases) { - PAIMON_ASSIGN_OR_RAISE(std::vector tables, - context_.catalog->ListTables(db)); + PAIMON_ASSIGN_OR_RAISE(std::vector tables, context_.catalog->ListTables(db)); for (const auto& table : tables) { Identifier id(db, table); auto schema_result = context_.catalog->LoadTableSchema(id); @@ -351,8 +340,8 @@ Result> TablesSystemTable::ArrowSchema() const { arrow::field("record_count", arrow::int64(), /*nullable=*/true), arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), arrow::field("file_count", arrow::int64(), /*nullable=*/true), - arrow::field("last_file_creation_time", - arrow::timestamp(arrow::TimeUnit::MILLI), /*nullable=*/true), + arrow::field("last_file_creation_time", arrow::timestamp(arrow::TimeUnit::MILLI), + /*nullable=*/true), }); } @@ -360,11 +349,9 @@ Result> TablesSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); std::vector rows; - PAIMON_ASSIGN_OR_RAISE(std::vector databases, - context_.catalog->ListDatabases()); + PAIMON_ASSIGN_OR_RAISE(std::vector databases, context_.catalog->ListDatabases()); for (const auto& db : databases) { - PAIMON_ASSIGN_OR_RAISE(std::vector tables, - context_.catalog->ListTables(db)); + PAIMON_ASSIGN_OR_RAISE(std::vector tables, context_.catalog->ListTables(db)); for (const auto& table : tables) { Identifier id(db, table); auto schema_result = context_.catalog->LoadTableSchema(id); @@ -397,20 +384,17 @@ Result> TablesSystemTable::BuildRows() const { row.SetField(1, StringValue(table)); row.SetField(2, StringValue(table_type_str)); row.SetField(3, partitioned); - row.SetField(4, primary_keys_str.empty() - ? VariantType(NullType()) - : VariantType(StringValue(primary_keys_str))); + row.SetField(4, primary_keys_str.empty() ? VariantType(NullType()) + : VariantType(StringValue(primary_keys_str))); // Get table path and aggregate file stats from manifest entries - PAIMON_ASSIGN_OR_RAISE(std::string table_path, - context_.catalog->GetTableLocation(id)); + PAIMON_ASSIGN_OR_RAISE(std::string table_path, context_.catalog->GetTableLocation(id)); auto file_stats_result = AggregateFileStats(context_.fs, table_path, data_schema->Options()); if (file_stats_result.ok()) { auto& all_stats = file_stats_result.value(); - int64_t total_record = 0, total_size = 0, total_files = 0, - max_creation = 0; + int64_t total_record = 0, total_size = 0, total_files = 0, max_creation = 0; for (const auto& [key, stats] : all_stats) { total_record += stats.record_count; total_size += stats.file_size_in_bytes; @@ -457,8 +441,8 @@ Result> PartitionsSystemTable::ArrowSchema() cons arrow::field("record_count", arrow::int64(), /*nullable=*/true), arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), arrow::field("file_count", arrow::int64(), /*nullable=*/true), - arrow::field("last_update_time", - arrow::timestamp(arrow::TimeUnit::MILLI), /*nullable=*/true), + arrow::field("last_update_time", arrow::timestamp(arrow::TimeUnit::MILLI), + /*nullable=*/true), }); } @@ -466,11 +450,9 @@ Result> PartitionsSystemTable::BuildRows() const { PAIMON_ASSIGN_OR_RAISE(std::shared_ptr schema, ArrowSchema()); std::vector rows; - PAIMON_ASSIGN_OR_RAISE(std::vector databases, - context_.catalog->ListDatabases()); + PAIMON_ASSIGN_OR_RAISE(std::vector databases, context_.catalog->ListDatabases()); for (const auto& db : databases) { - PAIMON_ASSIGN_OR_RAISE(std::vector tables, - context_.catalog->ListTables(db)); + PAIMON_ASSIGN_OR_RAISE(std::vector tables, context_.catalog->ListTables(db)); for (const auto& table : tables) { Identifier id(db, table); auto schema_result = context_.catalog->LoadTableSchema(id); @@ -509,9 +491,8 @@ Result> PartitionsSystemTable::BuildRows() const { GenericRow row(schema->num_fields()); row.SetField(0, StringValue(db)); row.SetField(1, StringValue(table)); - row.SetField(2, partition_key.empty() - ? VariantType(NullType()) - : VariantType(StringValue(partition_key))); + row.SetField(2, partition_key.empty() ? VariantType(NullType()) + : VariantType(StringValue(partition_key))); row.SetField(3, VariantType(stats.record_count)); row.SetField(4, VariantType(stats.file_size_in_bytes)); row.SetField(5, VariantType(stats.file_count)); diff --git a/src/paimon/core/table/system/global_system_tables.h b/src/paimon/core/table/system/global_system_tables.h index b3dc37570..5615abf84 100644 --- a/src/paimon/core/table/system/global_system_tables.h +++ b/src/paimon/core/table/system/global_system_tables.h @@ -108,8 +108,8 @@ class GlobalSystemTableLoader { public: static bool IsSupported(const std::string& table_name); - static Result> Load( - const std::string& table_name, const GlobalSystemTableContext& context); + static Result> Load(const std::string& table_name, + const GlobalSystemTableContext& context); static std::vector GetSupportedTableNames(); }; diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index fb3aa96ef..8c80ade27 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -50,11 +50,11 @@ #include "paimon/core/table/source/data_split_impl.h" #include "paimon/core/table/source/deletion_file.h" #include "paimon/core/table/source/fallback_data_split.h" +#include "paimon/core/table/system/global_system_tables.h" #include "paimon/core/tag/tag.h" #include "paimon/data/decimal.h" #include "paimon/data/timestamp.h" #include "paimon/defs.h" -#include "paimon/core/table/system/global_system_tables.h" #include "paimon/fs/file_system.h" #include "paimon/fs/local/local_file_system.h" #include "paimon/memory/memory_pool.h" @@ -3726,9 +3726,8 @@ TEST_P(ReadInteTest, TestSpecificFs) { namespace { Result ReadGlobalSystemTable( - const std::string& table_name, Catalog* catalog, - const std::shared_ptr& fs, const std::string& warehouse, - const std::map& options) { + const std::string& table_name, Catalog* catalog, const std::shared_ptr& fs, + const std::string& warehouse, const std::map& options) { GlobalSystemTableContext ctx; ctx.catalog = catalog; ctx.fs = fs; @@ -3949,8 +3948,8 @@ TEST(SystemTableReadInteTest, TestReadGlobalPartitions) { // Unpartitioned tables are skipped by sys.partitions → empty result. // CollectResult returns null shared_ptr when result is empty. - ASSERT_OK_AND_ASSIGN(auto part_result, - ReadGlobalSystemTable("partitions", catalog.get(), fs, warehouse, options)); + ASSERT_OK_AND_ASSIGN(auto part_result, ReadGlobalSystemTable("partitions", catalog.get(), fs, + warehouse, options)); ASSERT_FALSE(part_result.array) << "expected null array for empty partitions result"; } From d7192b2135bacc32a705bb959f1ff9289ccff57b Mon Sep 17 00:00:00 2001 From: Socrates Date: Mon, 13 Jul 2026 19:35:37 +0800 Subject: [PATCH 11/13] fix: address global system table review feedback --- .../core/catalog/file_system_catalog.cpp | 10 +- src/paimon/core/catalog/file_system_catalog.h | 2 +- .../core/catalog/file_system_catalog_test.cpp | 61 ++++--- .../table/system/global_system_tables.cpp | 161 +++++++++++------- .../core/table/system/global_system_tables.h | 9 +- src/paimon/core/table/system/system_table.cpp | 3 +- .../core/table/system/system_table_test.cpp | 8 + test/inte/read_inte_test.cpp | 106 +++++++++--- 8 files changed, 242 insertions(+), 118 deletions(-) diff --git a/src/paimon/core/catalog/file_system_catalog.cpp b/src/paimon/core/catalog/file_system_catalog.cpp index a1e9a9f65..ec142736f 100644 --- a/src/paimon/core/catalog/file_system_catalog.cpp +++ b/src/paimon/core/catalog/file_system_catalog.cpp @@ -101,7 +101,7 @@ Result FileSystemCatalog::DatabaseExists(const std::string& db_name) const Result FileSystemCatalog::TableExists(const Identifier& identifier) const { // Handle sys database global tables if (IsSystemDatabase(identifier.GetDatabaseName())) { - return GlobalSystemTableLoader::IsSupported(identifier.GetTableName()); + return GlobalSystemTableLoader::IsSupported(identifier.GetTableName(), catalog_options_); } PAIMON_ASSIGN_OR_RAISE(bool is_system_table, identifier.IsSystemTable()); if (is_system_table) { @@ -240,7 +240,7 @@ Result> FileSystemCatalog::ListDatabases() const { Result> FileSystemCatalog::ListTables(const std::string& db_name) const { if (IsSystemDatabase(db_name)) { - return GlobalSystemTableLoader::GetSupportedTableNames(); + return GlobalSystemTableLoader::GetSupportedTableNames(catalog_options_); } std::string database_path = NewDatabasePath(warehouse_, db_name); std::vector> file_status_list; @@ -275,11 +275,13 @@ Result> FileSystemCatalog::LoadTableSchema( const Identifier& identifier) const { // Handle sys database global tables if (IsSystemDatabase(identifier.GetDatabaseName())) { - if (!GlobalSystemTableLoader::IsSupported(identifier.GetTableName())) { + PAIMON_ASSIGN_OR_RAISE(bool supported, GlobalSystemTableLoader::IsSupported( + identifier.GetTableName(), catalog_options_)); + if (!supported) { return Status::NotExist(fmt::format("{} not exist", identifier.ToString())); } GlobalSystemTableContext context; - context.catalog = const_cast(this); + context.catalog = this; context.fs = fs_; context.warehouse = warehouse_; context.catalog_options = catalog_options_; diff --git a/src/paimon/core/catalog/file_system_catalog.h b/src/paimon/core/catalog/file_system_catalog.h index 7a2ad4a17..79f9fab55 100644 --- a/src/paimon/core/catalog/file_system_catalog.h +++ b/src/paimon/core/catalog/file_system_catalog.h @@ -39,7 +39,7 @@ class Logger; class FileSystemCatalog : public Catalog { public: FileSystemCatalog(const std::shared_ptr& fs, const std::string& warehouse, - const std::map& catalog_options = {}); + const std::map& catalog_options); Status CreateDatabase(const std::string& db_name, const std::map& options, diff --git a/src/paimon/core/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp index f970fddb6..a48ff6ce1 100644 --- a/src/paimon/core/catalog/file_system_catalog_test.cpp +++ b/src/paimon/core/catalog/file_system_catalog_test.cpp @@ -28,6 +28,7 @@ #include "paimon/common/utils/path_util.h" #include "paimon/core/core_options.h" #include "paimon/core/schema/table_schema.h" +#include "paimon/core/table/system/global_system_tables.h" #include "paimon/core/table/system/system_table_schema.h" #include "paimon/defs.h" #include "paimon/fs/file_system.h" @@ -44,7 +45,7 @@ TEST(FileSystemCatalogTest, TestDatabaseExists) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK_AND_ASSIGN(auto exist, catalog.DatabaseExists("db1")); ASSERT_FALSE(exist); @@ -70,7 +71,7 @@ TEST(FileSystemCatalogTest, TestInvalidCreateDatabase) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_NOK_WITH_MSG( catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true), @@ -87,7 +88,7 @@ TEST(FileSystemCatalogTest, TestCreateSystemDatabaseAndTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_NOK_WITH_MSG(catalog.CreateDatabase(Catalog::SYSTEM_DATABASE_NAME, options, /*ignore_if_exists=*/true), "Cannot create database for system database"); @@ -100,7 +101,7 @@ TEST(FileSystemCatalogTest, TestCreateSystemDatabaseAndTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), @@ -125,7 +126,7 @@ TEST(FileSystemCatalogTest, TestCreateTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = { @@ -161,7 +162,7 @@ TEST(FileSystemCatalogTest, TestOptionsSystemTableCatalog) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())}); @@ -219,7 +220,7 @@ TEST(FileSystemCatalogTest, TestAuditLogAndBinlogSystemTableCatalog) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); auto typed_schema = @@ -290,7 +291,7 @@ TEST(FileSystemCatalogTest, TestMetadataSystemTableCatalog) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); auto typed_schema = @@ -435,7 +436,7 @@ TEST(FileSystemCatalogTest, TestCreateTableWithBlob) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = {arrow::field("f0", arrow::boolean()), @@ -491,7 +492,7 @@ TEST(FileSystemCatalogTest, TestInvalidCreateTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), @@ -516,7 +517,7 @@ TEST(FileSystemCatalogTest, TestCreateTableWhileDbNotExist) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); arrow::FieldVector fields = { arrow::field("f0", arrow::boolean()), arrow::field("f1", arrow::int8()), @@ -539,7 +540,7 @@ TEST(FileSystemCatalogTest, TestCreateTableWhileTableExist) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); { ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = { @@ -600,23 +601,29 @@ TEST(FileSystemCatalogTest, TestCreateTableWhileTableExist) { } } -TEST(FileSystemCatalogTest, TestInvalidList) { +TEST(FileSystemCatalogTest, TestSystemList) { std::map 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()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK_AND_ASSIGN(auto sys_tables, catalog.ListTables("sys")); ASSERT_FALSE(sys_tables.empty()); - // Verify expected global system table names are present - ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "catalog_options") != + // catalog_options is disabled by default, matching Java Paimon. + ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "catalog_options") == sys_tables.end()); ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "all_table_options") != sys_tables.end()); ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "tables") != sys_tables.end()); ASSERT_TRUE(std::find(sys_tables.begin(), sys_tables.end(), "partitions") != sys_tables.end()); + + options[CatalogOptionsSystemTable::kEnabledOption] = "true"; + FileSystemCatalog enabled_catalog(core_options.GetFileSystem(), dir->Str(), options); + ASSERT_OK_AND_ASSIGN(auto enabled_sys_tables, enabled_catalog.ListTables("sys")); + ASSERT_TRUE(std::find(enabled_sys_tables.begin(), enabled_sys_tables.end(), + "catalog_options") != enabled_sys_tables.end()); } TEST(FileSystemCatalogTest, TestValidateTableSchema) { @@ -626,7 +633,7 @@ TEST(FileSystemCatalogTest, TestValidateTableSchema) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true)); arrow::FieldVector fields = { arrow::field("f0", arrow::utf8()), @@ -695,7 +702,7 @@ TEST(FileSystemCatalogTest, TestDropDatabase) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); // Test 1: Drop non-existent database with ignore_if_not_exists=true ASSERT_OK(catalog.DropDatabase("non_existent_db", /*ignore_if_not_exists=*/true, @@ -751,7 +758,7 @@ TEST(FileSystemCatalogTest, TestDropTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); // Test 1: Drop non-existent table with ignore_if_not_exists=true @@ -793,7 +800,7 @@ TEST(FileSystemCatalogTest, TestRenameTable) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); // Test 1: Rename non-existent table with ignore_if_not_exists=true @@ -860,7 +867,7 @@ TEST(FileSystemCatalogTest, TestDropTableWithExternalPath) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); // Create external path directory @@ -922,7 +929,7 @@ TEST(FileSystemCatalogTest, TestDropTableWithMultipleExternalPaths) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); // Create multiple external path directories @@ -979,7 +986,7 @@ TEST(FileSystemCatalogTest, TestDropTableWithGlobalIndexExternalPathOnMainBranch ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_OK(catalog.CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); // Create external path directory for global index @@ -1083,7 +1090,7 @@ TEST(FileSystemCatalogTest, TestDropTableWithGlobalIndexExternalPathOnBranch) { ASSERT_TRUE(external_exists); // Drop the table via catalog - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); Identifier identifier("test_db", "append_table_with_rt_branch"); ASSERT_OK_AND_ASSIGN(bool table_exists, catalog.TableExists(identifier)); ASSERT_TRUE(table_exists); @@ -1111,7 +1118,7 @@ TEST(FileSystemCatalogTest, TestListSnapshots) { std::string db_path = dir->Str() + "/test_db.db"; ASSERT_TRUE(TestUtil::CopyDirectory(test_data_path, db_path)); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); Identifier id("test_db", "append_table_with_multiple_file_format"); ASSERT_OK_AND_ASSIGN(std::vector snapshots, catalog.ListSnapshots(id, "")); @@ -1145,7 +1152,7 @@ TEST(FileSystemCatalogTest, TestListSnapshotsTableNotExist) { ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options)); auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); ASSERT_NOK_WITH_MSG( catalog.ListSnapshots(Identifier("non_existent_db", "non_existent_table"), ""), @@ -1203,7 +1210,7 @@ TEST(FileSystemCatalogTest, TestDropTableWithBranchExternalPaths) { ASSERT_TRUE(external_exists); // Drop the table via catalog - FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str()); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); Identifier identifier("test_db", "append_table_with_rt_branch"); ASSERT_OK_AND_ASSIGN(bool table_exists, catalog.TableExists(identifier)); ASSERT_TRUE(table_exists); diff --git a/src/paimon/core/table/system/global_system_tables.cpp b/src/paimon/core/table/system/global_system_tables.cpp index 3e0a3e5df..48eb9b33a 100644 --- a/src/paimon/core/table/system/global_system_tables.cpp +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -16,12 +16,10 @@ #include "paimon/core/table/system/global_system_tables.h" -#include #include #include #include #include -#include #include #include @@ -30,7 +28,7 @@ #include "paimon/catalog/identifier.h" #include "paimon/common/data/binary_string.h" #include "paimon/common/data/generic_row.h" -#include "paimon/common/utils/binary_row_partition_computer.h" +#include "paimon/common/utils/options_utils.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/string_utils.h" #include "paimon/core/core_options.h" @@ -48,7 +46,6 @@ #include "paimon/core/utils/field_mapping.h" #include "paimon/core/utils/file_store_path_factory.h" #include "paimon/core/utils/snapshot_manager.h" -#include "paimon/data/timestamp.h" #include "paimon/defs.h" #include "paimon/memory/memory_pool.h" #include "paimon/status.h" @@ -65,24 +62,25 @@ using GlobalSystemTableFactory = struct GlobalSystemTableRegistryEntry { std::string name; + bool requires_catalog; GlobalSystemTableFactory factory; }; const std::vector& GlobalSystemTableRegistry() { static const std::vector registry = { - {CatalogOptionsSystemTable::kName, + {CatalogOptionsSystemTable::kName, false, [](const GlobalSystemTableContext& ctx) -> Result> { return std::make_shared(ctx); }}, - {AllTableOptionsSystemTable::kName, + {AllTableOptionsSystemTable::kName, true, [](const GlobalSystemTableContext& ctx) -> Result> { return std::make_shared(ctx); }}, - {TablesSystemTable::kName, + {TablesSystemTable::kName, true, [](const GlobalSystemTableContext& ctx) -> Result> { return std::make_shared(ctx); }}, - {PartitionsSystemTable::kName, + {PartitionsSystemTable::kName, true, [](const GlobalSystemTableContext& ctx) -> Result> { return std::make_shared(ctx); }}, @@ -98,6 +96,30 @@ VariantType StringValue(const std::string& value) { return BinaryString::FromString(value, GetDefaultPool().get()); } +VariantType OptionalStringValue(const std::map& options, + const std::string& key) { + auto it = options.find(key); + return it == options.end() ? VariantType(NullType()) : VariantType(StringValue(it->second)); +} + +Result OptionalLongValue(const std::map& options, + const std::string& key) { + if (options.find(key) == options.end()) { + return VariantType(NullType()); + } + PAIMON_ASSIGN_OR_RAISE(int64_t value, OptionsUtils::GetValueFromMap(options, key)); + return VariantType(value); +} + +Result IsEnabled(const GlobalSystemTableRegistryEntry& entry, + const std::map& catalog_options) { + if (entry.name != CatalogOptionsSystemTable::kName) { + return true; + } + return OptionsUtils::GetValueFromMap(catalog_options, + CatalogOptionsSystemTable::kEnabledOption, false); +} + // Aggregated file-level statistics for a table or partition. struct FileStats { int64_t record_count = 0; @@ -106,18 +128,23 @@ struct FileStats { int64_t last_file_creation_time_millis = 0; }; +struct AggregatedFileStats { + bool has_snapshot = false; + std::map by_partition; +}; + // Read the latest snapshot's data files and aggregate statistics. -// Returns an empty map if no snapshot or no data files exist. -Result> AggregateFileStats( - const std::shared_ptr& fs, const std::string& table_path, - const std::map& options) { - std::map result; +Result AggregateFileStats(const std::shared_ptr& fs, + const std::string& table_path, + const std::map& options) { + AggregatedFileStats result; SnapshotManager snapshot_manager(fs, table_path, BranchManager::DEFAULT_MAIN_BRANCH); PAIMON_ASSIGN_OR_RAISE(std::optional snapshot, snapshot_manager.LatestSnapshot()); if (!snapshot) { return result; } + result.has_snapshot = true; PAIMON_ASSIGN_OR_RAISE(CoreOptions core_options, CoreOptions::FromMap(options)); @@ -181,14 +208,17 @@ Result> AggregateFileStats( // Convert partition BinaryRow to string representation std::string partition_key; if (entry.Partition().GetFieldCount() > 0) { - PAIMON_ASSIGN_OR_RAISE(partition_key, BinaryRowPartitionComputer::PartToSimpleString( - partition_schema, entry.Partition(), ",", - /*max_length=*/255, - /*legacy_partition_name_enabled=*/false)); - partition_key = "{" + partition_key + "}"; + PAIMON_ASSIGN_OR_RAISE(auto partition_values, + path_factory->GeneratePartitionVector(entry.Partition())); + for (const auto& [key, value] : partition_values) { + if (!partition_key.empty()) { + partition_key += "/"; + } + partition_key += key + "=" + value; + } } - auto& stats = result[partition_key]; + auto& stats = result.by_partition[partition_key]; stats.record_count += file->row_count; stats.file_size_in_bytes += file->file_size; stats.file_count++; @@ -207,11 +237,12 @@ Result> AggregateFileStats( // GlobalSystemTableLoader // ============================================================================= -bool GlobalSystemTableLoader::IsSupported(const std::string& table_name) { +Result GlobalSystemTableLoader::IsSupported( + const std::string& table_name, const std::map& catalog_options) { std::string normalized = StringUtils::ToLowerCase(table_name); for (const auto& entry : GlobalSystemTableRegistry()) { if (entry.name == normalized) { - return true; + return IsEnabled(entry, catalog_options); } } return false; @@ -222,17 +253,29 @@ Result> GlobalSystemTableLoader::Load( std::string normalized = StringUtils::ToLowerCase(table_name); for (const auto& entry : GlobalSystemTableRegistry()) { if (entry.name == normalized) { + PAIMON_ASSIGN_OR_RAISE(bool enabled, IsEnabled(entry, context.catalog_options)); + if (!enabled) { + return Status::NotExist("global system table is disabled: ", table_name); + } + if (entry.requires_catalog && context.catalog == nullptr) { + return Status::NotImplemented("global system table requires catalog context: ", + table_name); + } return entry.factory(context); } } return Status::NotImplemented("unsupported global system table: ", table_name); } -std::vector GlobalSystemTableLoader::GetSupportedTableNames() { +Result> GlobalSystemTableLoader::GetSupportedTableNames( + const std::map& catalog_options) { std::vector names; names.reserve(GlobalSystemTableRegistry().size()); for (const auto& entry : GlobalSystemTableRegistry()) { - names.push_back(entry.name); + PAIMON_ASSIGN_OR_RAISE(bool enabled, IsEnabled(entry, catalog_options)); + if (enabled) { + names.push_back(entry.name); + } } return names; } @@ -261,8 +304,8 @@ Result> CatalogOptionsSystemTable::BuildRows() const { rows.reserve(context_.catalog_options.size()); for (const auto& [key, value] : context_.catalog_options) { GenericRow row(schema->num_fields()); - row.SetField(0, std::string_view(key)); - row.SetField(1, std::string_view(value)); + row.SetField(0, StringValue(key)); + row.SetField(1, StringValue(value)); rows.push_back(std::move(row)); } return rows; @@ -336,12 +379,16 @@ Result> TablesSystemTable::ArrowSchema() const { arrow::field("table_name", arrow::utf8(), /*nullable=*/false), arrow::field("table_type", arrow::utf8(), /*nullable=*/false), arrow::field("partitioned", arrow::boolean(), /*nullable=*/false), - arrow::field("primary_key", arrow::utf8(), /*nullable=*/false), + arrow::field("primary_key", arrow::boolean(), /*nullable=*/false), + arrow::field("owner", arrow::utf8(), /*nullable=*/true), + arrow::field("created_at", arrow::int64(), /*nullable=*/true), + arrow::field("created_by", arrow::utf8(), /*nullable=*/true), + arrow::field("updated_at", arrow::int64(), /*nullable=*/true), + arrow::field("updated_by", arrow::utf8(), /*nullable=*/true), arrow::field("record_count", arrow::int64(), /*nullable=*/true), arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), arrow::field("file_count", arrow::int64(), /*nullable=*/true), - arrow::field("last_file_creation_time", arrow::timestamp(arrow::TimeUnit::MILLI), - /*nullable=*/true), + arrow::field("last_file_creation_time", arrow::int64(), /*nullable=*/true), }); } @@ -364,36 +411,34 @@ Result> TablesSystemTable::BuildRows() const { continue; } - // Determine table type: EXTERNAL if data-file.external-paths is set - std::string table_type_str = "MANAGED"; const auto& opts = data_schema->Options(); - if (opts.find(Options::DATA_FILE_EXTERNAL_PATHS) != opts.end()) { - table_type_str = "EXTERNAL"; - } + auto table_type = opts.find("type"); + const std::string table_type_str = + table_type == opts.end() ? "TABLE" : table_type->second; bool partitioned = !data_schema->PartitionKeys().empty(); - std::string primary_keys_str; - const auto& pks = data_schema->PrimaryKeys(); - for (size_t i = 0; i < pks.size(); ++i) { - if (i > 0) primary_keys_str += ","; - primary_keys_str += pks[i]; - } GenericRow row(schema->num_fields()); row.SetField(0, StringValue(db)); row.SetField(1, StringValue(table)); row.SetField(2, StringValue(table_type_str)); row.SetField(3, partitioned); - row.SetField(4, primary_keys_str.empty() ? VariantType(NullType()) - : VariantType(StringValue(primary_keys_str))); + row.SetField(4, !data_schema->PrimaryKeys().empty()); + row.SetField(5, OptionalStringValue(opts, "owner")); + PAIMON_ASSIGN_OR_RAISE(VariantType created_at, OptionalLongValue(opts, "createdAt")); + row.SetField(6, std::move(created_at)); + row.SetField(7, OptionalStringValue(opts, "createdBy")); + PAIMON_ASSIGN_OR_RAISE(VariantType updated_at, OptionalLongValue(opts, "updatedAt")); + row.SetField(8, std::move(updated_at)); + row.SetField(9, OptionalStringValue(opts, "updatedBy")); // Get table path and aggregate file stats from manifest entries PAIMON_ASSIGN_OR_RAISE(std::string table_path, context_.catalog->GetTableLocation(id)); auto file_stats_result = AggregateFileStats(context_.fs, table_path, data_schema->Options()); - if (file_stats_result.ok()) { - auto& all_stats = file_stats_result.value(); + if (file_stats_result.ok() && file_stats_result.value().has_snapshot) { + auto& all_stats = file_stats_result.value().by_partition; int64_t total_record = 0, total_size = 0, total_files = 0, max_creation = 0; for (const auto& [key, stats] : all_stats) { total_record += stats.record_count; @@ -403,17 +448,16 @@ Result> TablesSystemTable::BuildRows() const { max_creation = stats.last_file_creation_time_millis; } } - row.SetField(5, VariantType(total_record)); - row.SetField(6, VariantType(total_size)); - row.SetField(7, VariantType(total_files)); - row.SetField(8, max_creation > 0 - ? VariantType(Timestamp::FromEpochMillis(max_creation)) - : VariantType(NullType())); + row.SetField(10, VariantType(total_record)); + row.SetField(11, VariantType(total_size)); + row.SetField(12, VariantType(total_files)); + row.SetField( + 13, max_creation > 0 ? VariantType(max_creation) : VariantType(NullType())); } else { - row.SetField(5, NullType()); - row.SetField(6, NullType()); - row.SetField(7, NullType()); - row.SetField(8, NullType()); + row.SetField(10, NullType()); + row.SetField(11, NullType()); + row.SetField(12, NullType()); + row.SetField(13, NullType()); } rows.push_back(std::move(row)); @@ -441,8 +485,8 @@ Result> PartitionsSystemTable::ArrowSchema() cons arrow::field("record_count", arrow::int64(), /*nullable=*/true), arrow::field("file_size_in_bytes", arrow::int64(), /*nullable=*/true), arrow::field("file_count", arrow::int64(), /*nullable=*/true), - arrow::field("last_update_time", arrow::timestamp(arrow::TimeUnit::MILLI), - /*nullable=*/true), + arrow::field("last_file_creation_time", arrow::int64(), /*nullable=*/true), + arrow::field("done", arrow::boolean(), /*nullable=*/false), }); } @@ -483,7 +527,7 @@ Result> PartitionsSystemTable::BuildRows() const { continue; } - auto& stats_map = file_stats_result.value(); + auto& stats_map = file_stats_result.value().by_partition; for (const auto& [partition_key, stats] : stats_map) { if (stats.file_count == 0) { continue; @@ -497,9 +541,10 @@ Result> PartitionsSystemTable::BuildRows() const { row.SetField(4, VariantType(stats.file_size_in_bytes)); row.SetField(5, VariantType(stats.file_count)); row.SetField(6, stats.last_file_creation_time_millis > 0 - ? VariantType(Timestamp::FromEpochMillis( - stats.last_file_creation_time_millis)) + ? VariantType(stats.last_file_creation_time_millis) : VariantType(NullType())); + // File-system catalog partitions are not explicitly marked as done. + row.SetField(7, false); rows.push_back(std::move(row)); } } diff --git a/src/paimon/core/table/system/global_system_tables.h b/src/paimon/core/table/system/global_system_tables.h index 5615abf84..81b8ce516 100644 --- a/src/paimon/core/table/system/global_system_tables.h +++ b/src/paimon/core/table/system/global_system_tables.h @@ -30,7 +30,7 @@ class FileSystem; /// Context passed to global system table constructors, providing catalog-level /// access for enumerating databases, tables, and reading metadata. struct GlobalSystemTableContext { - Catalog* catalog; // non-owning pointer + const Catalog* catalog = nullptr; // non-owning pointer std::shared_ptr fs; std::string warehouse; std::map catalog_options; @@ -41,6 +41,7 @@ struct GlobalSystemTableContext { class CatalogOptionsSystemTable : public InMemorySystemTable { public: static constexpr const char* kName = "catalog_options"; + static constexpr const char* kEnabledOption = "catalog-options-table.enabled"; explicit CatalogOptionsSystemTable(GlobalSystemTableContext context); @@ -106,12 +107,14 @@ class PartitionsSystemTable : public InMemorySystemTable { /// GlobalSystemTableContext instead of a per-table TableSchema. class GlobalSystemTableLoader { public: - static bool IsSupported(const std::string& table_name); + static Result IsSupported(const std::string& table_name, + const std::map& catalog_options = {}); static Result> Load(const std::string& table_name, const GlobalSystemTableContext& context); - static std::vector GetSupportedTableNames(); + static Result> GetSupportedTableNames( + const std::map& catalog_options = {}); }; } // namespace paimon diff --git a/src/paimon/core/table/system/system_table.cpp b/src/paimon/core/table/system/system_table.cpp index ef4142c22..ba3703a70 100644 --- a/src/paimon/core/table/system/system_table.cpp +++ b/src/paimon/core/table/system/system_table.cpp @@ -23,6 +23,7 @@ #include #include +#include "paimon/catalog/catalog.h" #include "paimon/catalog/identifier.h" #include "paimon/common/utils/path_util.h" #include "paimon/common/utils/string_utils.h" @@ -193,7 +194,7 @@ Result> SystemTableLoader::TryParsePath(const std std::string parent_name = PathUtil::GetName(parent); // Detect global system table paths: /sys/ - if (parent_name == "sys") { + if (parent_name == Catalog::SYSTEM_DATABASE_NAME) { SystemTablePath system_table_path; system_table_path.is_global = true; system_table_path.system_table_name = table_name; diff --git a/src/paimon/core/table/system/system_table_test.cpp b/src/paimon/core/table/system/system_table_test.cpp index ec61439e6..c610b8121 100644 --- a/src/paimon/core/table/system/system_table_test.cpp +++ b/src/paimon/core/table/system/system_table_test.cpp @@ -30,6 +30,7 @@ #include "paimon/core/table/system/read_optimized_system_table.h" #include "paimon/defs.h" #include "paimon/fs/file_system.h" +#include "paimon/fs/file_system_factory.h" #include "paimon/result.h" #include "paimon/status.h" #include "paimon/testing/utils/testharness.h" @@ -96,4 +97,11 @@ TEST(SystemTableTest, TestReadOptimizedSystemTablePathParsing) { ASSERT_EQ(parsed->system_table_name, ReadOptimizedSystemTable::kName); } +TEST(SystemTableTest, TestGlobalSystemTableWithoutCatalogReturnsNotImplemented) { + ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", "/tmp", {})); + std::shared_ptr shared_fs(std::move(fs)); + ASSERT_NOK_WITH_MSG(SystemTableLoader::LoadFromPath(shared_fs, "/tmp/warehouse/sys/tables", {}), + "global system table requires catalog context: tables"); +} + } // namespace paimon::test diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index 8c80ade27..c6eeea43a 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -3765,9 +3765,11 @@ Result ReadGlobalSystemTable( } // namespace TEST(SystemTableReadInteTest, TestReadGlobalCatalogOptions) { - std::map options = {{Options::FILE_SYSTEM, "local"}, - {Options::FILE_FORMAT, "orc"}, - {"custom.catalog.option", "test-value"}}; + std::map options = { + {Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {CatalogOptionsSystemTable::kEnabledOption, "true"}, + {"custom.catalog.option", "test-value"}}; auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); @@ -3874,7 +3876,12 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { std::map options = {{Options::FILE_SYSTEM, "local"}, {Options::FILE_FORMAT, "orc"}, {Options::MANIFEST_FORMAT, "orc"}, - {Options::BUCKET, "1"}}; + {Options::BUCKET, "1"}, + {"owner", "alice"}, + {"createdAt", "1000"}, + {"createdBy", "creator"}, + {"updatedAt", "2000"}, + {"updatedBy", "updater"}}; auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); @@ -3899,27 +3906,49 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { auto struct_array = SingleStructChunk(result); ASSERT_TRUE(struct_array); ASSERT_GE(struct_array->length(), 1); + ASSERT_EQ(StructFieldNames(struct_array), + (std::vector{ + "database_name", "table_name", "table_type", "partitioned", "primary_key", + "owner", "created_at", "created_by", "updated_at", "updated_by", "record_count", + "file_size_in_bytes", "file_count", "last_file_creation_time"})); auto db_array = std::dynamic_pointer_cast(struct_array->field(0)); auto tbl_array = std::dynamic_pointer_cast(struct_array->field(1)); auto type_array = std::dynamic_pointer_cast(struct_array->field(2)); auto part_array = std::dynamic_pointer_cast(struct_array->field(3)); - auto pk_array = std::dynamic_pointer_cast(struct_array->field(4)); + auto pk_array = std::dynamic_pointer_cast(struct_array->field(4)); + auto owner_array = std::dynamic_pointer_cast(struct_array->field(5)); + auto created_at_array = std::dynamic_pointer_cast(struct_array->field(6)); + auto created_by_array = std::dynamic_pointer_cast(struct_array->field(7)); + auto updated_at_array = std::dynamic_pointer_cast(struct_array->field(8)); + auto updated_by_array = std::dynamic_pointer_cast(struct_array->field(9)); + auto record_count_array = std::dynamic_pointer_cast(struct_array->field(10)); ASSERT_TRUE(db_array); ASSERT_TRUE(tbl_array); ASSERT_TRUE(type_array); ASSERT_TRUE(part_array); ASSERT_TRUE(pk_array); + ASSERT_TRUE(owner_array); + ASSERT_TRUE(created_at_array); + ASSERT_TRUE(created_by_array); + ASSERT_TRUE(updated_at_array); + ASSERT_TRUE(updated_by_array); + ASSERT_TRUE(record_count_array); // Find our table by table name bool found = false; for (int64_t i = 0; i < struct_array->length(); ++i) { if (std::string(tbl_array->GetString(i)) == "test_tbl") { EXPECT_EQ(std::string(db_array->GetString(i)), "test_db"); - EXPECT_EQ(std::string(type_array->GetString(i)), "MANAGED"); + EXPECT_EQ(std::string(type_array->GetString(i)), "TABLE"); EXPECT_FALSE(part_array->Value(i)); - // pk is stored as BinaryString; check not null - EXPECT_FALSE(pk_array->IsNull(i)); + EXPECT_TRUE(pk_array->Value(i)); + EXPECT_EQ(owner_array->GetString(i), "alice"); + EXPECT_EQ(created_at_array->Value(i), 1000); + EXPECT_EQ(created_by_array->GetString(i), "creator"); + EXPECT_EQ(updated_at_array->Value(i), 2000); + EXPECT_EQ(updated_by_array->GetString(i), "updater"); + EXPECT_TRUE(record_count_array->IsNull(i)); found = true; } } @@ -3929,28 +3958,57 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { TEST(SystemTableReadInteTest, TestReadGlobalPartitions) { std::map options = {{Options::FILE_SYSTEM, "local"}, {Options::FILE_FORMAT, "orc"}, - {Options::MANIFEST_FORMAT, "orc"}}; + {Options::MANIFEST_FORMAT, "orc"}, + {Options::BUCKET, "1"}, + {Options::BUCKET_KEY, "v"}}; auto dir = UniqueTestDirectory::Create(); ASSERT_TRUE(dir); - std::string warehouse = PathUtil::JoinPath(dir->Str(), "warehouse"); - ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); - auto fs = catalog->GetFileSystem(); + std::string warehouse = dir->Str(); - // Create a database and an unpartitioned table (no partitions → empty result) - ASSERT_OK(catalog->CreateDatabase("test_db", options, /*ignore_if_exists=*/false)); - auto typed_schema = arrow::schema({arrow::field("f0", arrow::int32())}); - ::ArrowSchema schema; - ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &schema).ok()); - ASSERT_OK(catalog->CreateTable(Identifier("test_db", "test_tbl"), &schema, - /*partition_keys=*/{}, /*primary_keys=*/{}, options, - /*ignore_if_exists=*/false)); - ArrowSchemaRelease(&schema); + arrow::FieldVector fields = { + arrow::field("dt", arrow::utf8()), + arrow::field("region", arrow::utf8()), + arrow::field("v", arrow::int32()), + }; + auto schema = arrow::schema(fields); + ASSERT_OK_AND_ASSIGN(auto helper, TestHelper::Create(warehouse, schema, + /*partition_keys=*/{"dt", "region"}, + /*primary_keys=*/{}, options, + /*is_streaming_mode=*/true)); + ASSERT_OK_AND_ASSIGN( + std::unique_ptr batch, + TestHelper::MakeRecordBatch(arrow::struct_(fields), R"([["2026-07-13", "cn", 1]])", + /*partition_map=*/{{"dt", "2026-07-13"}, {"region", "cn"}}, + /*bucket=*/0, {})); + ASSERT_OK(helper->WriteAndCommit(std::move(batch), /*commit_identifier=*/0, + /*expected_commit_messages=*/std::nullopt)); - // Unpartitioned tables are skipped by sys.partitions → empty result. - // CollectResult returns null shared_ptr when result is empty. + ASSERT_OK_AND_ASSIGN(auto catalog, Catalog::Create(warehouse, options)); + auto fs = catalog->GetFileSystem(); ASSERT_OK_AND_ASSIGN(auto part_result, ReadGlobalSystemTable("partitions", catalog.get(), fs, warehouse, options)); - ASSERT_FALSE(part_result.array) << "expected null array for empty partitions result"; + auto struct_array = SingleStructChunk(part_result); + ASSERT_TRUE(struct_array); + ASSERT_EQ(StructFieldNames(struct_array), + (std::vector{"database_name", "table_name", "partition_name", + "record_count", "file_size_in_bytes", "file_count", + "last_file_creation_time", "done"})); + ASSERT_EQ(struct_array->length(), 1); + auto db_array = std::dynamic_pointer_cast(struct_array->field(0)); + auto table_array = std::dynamic_pointer_cast(struct_array->field(1)); + auto partition_array = std::dynamic_pointer_cast(struct_array->field(2)); + auto creation_time_array = std::dynamic_pointer_cast(struct_array->field(6)); + auto done_array = std::dynamic_pointer_cast(struct_array->field(7)); + ASSERT_TRUE(db_array); + ASSERT_TRUE(table_array); + ASSERT_TRUE(partition_array); + ASSERT_TRUE(creation_time_array); + ASSERT_TRUE(done_array); + EXPECT_EQ(db_array->GetString(0), "foo"); + EXPECT_EQ(table_array->GetString(0), "bar"); + EXPECT_EQ(partition_array->GetString(0), "dt=2026-07-13/region=cn"); + EXPECT_FALSE(creation_time_array->IsNull(0)); + EXPECT_FALSE(done_array->Value(0)); } } // namespace paimon::test From 0402ff699c48512372e2cedfb8c6ba93c0bfb767 Mon Sep 17 00:00:00 2001 From: Socrates Date: Wed, 15 Jul 2026 17:32:55 +0800 Subject: [PATCH 12/13] fix: complete global system table review feedback --- src/paimon/core/table/system/global_system_tables.h | 4 ++-- test/inte/read_inte_test.cpp | 12 ++++++++++++ 2 files changed, 14 insertions(+), 2 deletions(-) diff --git a/src/paimon/core/table/system/global_system_tables.h b/src/paimon/core/table/system/global_system_tables.h index 81b8ce516..28160f103 100644 --- a/src/paimon/core/table/system/global_system_tables.h +++ b/src/paimon/core/table/system/global_system_tables.h @@ -108,13 +108,13 @@ class PartitionsSystemTable : public InMemorySystemTable { class GlobalSystemTableLoader { public: static Result IsSupported(const std::string& table_name, - const std::map& catalog_options = {}); + const std::map& catalog_options); static Result> Load(const std::string& table_name, const GlobalSystemTableContext& context); static Result> GetSupportedTableNames( - const std::map& catalog_options = {}); + const std::map& catalog_options); }; } // namespace paimon diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index c6eeea43a..f90e05a3a 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -3901,6 +3901,13 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { /*ignore_if_exists=*/false)); ArrowSchemaRelease(&schema); + ::ArrowSchema no_pk_schema; + ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &no_pk_schema).ok()); + ASSERT_OK(catalog->CreateTable(Identifier("test_db", "test_no_pk_tbl"), &no_pk_schema, + /*partition_keys=*/{}, /*primary_keys=*/{}, options, + /*ignore_if_exists=*/false)); + ArrowSchemaRelease(&no_pk_schema); + ASSERT_OK_AND_ASSIGN(auto result, ReadGlobalSystemTable("tables", catalog.get(), fs, warehouse, options)); auto struct_array = SingleStructChunk(result); @@ -3937,6 +3944,7 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { // Find our table by table name bool found = false; + bool found_no_pk = false; for (int64_t i = 0; i < struct_array->length(); ++i) { if (std::string(tbl_array->GetString(i)) == "test_tbl") { EXPECT_EQ(std::string(db_array->GetString(i)), "test_db"); @@ -3950,9 +3958,13 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { EXPECT_EQ(updated_by_array->GetString(i), "updater"); EXPECT_TRUE(record_count_array->IsNull(i)); found = true; + } else if (std::string(tbl_array->GetString(i)) == "test_no_pk_tbl") { + EXPECT_FALSE(pk_array->Value(i)); + found_no_pk = true; } } ASSERT_TRUE(found) << "table not found in sys.tables"; + ASSERT_TRUE(found_no_pk) << "no-PK table not found in sys.tables"; } TEST(SystemTableReadInteTest, TestReadGlobalPartitions) { From fc812bb89f5c526c7c331ff4af6eb53a51e9dc33 Mon Sep 17 00:00:00 2001 From: Socrates Date: Wed, 15 Jul 2026 19:23:06 +0800 Subject: [PATCH 13/13] test: configure bucket key for append-only table --- test/inte/read_inte_test.cpp | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/test/inte/read_inte_test.cpp b/test/inte/read_inte_test.cpp index f90e05a3a..ce15d3e84 100644 --- a/test/inte/read_inte_test.cpp +++ b/test/inte/read_inte_test.cpp @@ -3901,10 +3901,12 @@ TEST(SystemTableReadInteTest, TestReadGlobalTables) { /*ignore_if_exists=*/false)); ArrowSchemaRelease(&schema); + std::map no_pk_options = options; + no_pk_options[Options::BUCKET_KEY] = "pk"; ::ArrowSchema no_pk_schema; ASSERT_TRUE(arrow::ExportSchema(*typed_schema, &no_pk_schema).ok()); ASSERT_OK(catalog->CreateTable(Identifier("test_db", "test_no_pk_tbl"), &no_pk_schema, - /*partition_keys=*/{}, /*primary_keys=*/{}, options, + /*partition_keys=*/{}, /*primary_keys=*/{}, no_pk_options, /*ignore_if_exists=*/false)); ArrowSchemaRelease(&no_pk_schema);