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..ec142736f 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(), catalog_options_); + } 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(catalog_options_); } std::string database_path = NewDatabasePath(warehouse_, db_name); std::vector> file_status_list; @@ -261,6 +273,24 @@ Result FileSystemCatalog::TableExistsInFileSystem(const std::string& table Result> FileSystemCatalog::LoadTableSchema( const Identifier& identifier) const { + // Handle sys database global tables + if (IsSystemDatabase(identifier.GetDatabaseName())) { + 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 = 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..79f9fab55 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/catalog/file_system_catalog_test.cpp b/src/paimon/core/catalog/file_system_catalog_test.cpp index 8377493e5..a48ff6ce1 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" @@ -26,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" @@ -42,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); @@ -68,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), @@ -85,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"); @@ -98,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()), @@ -123,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 = { @@ -159,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())}); @@ -217,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 = @@ -288,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 = @@ -433,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()), @@ -489,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()), @@ -514,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()), @@ -537,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 = { @@ -598,16 +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()); - ASSERT_NOK_WITH_MSG(catalog.ListTables("sys"), - "do not support listing tables for system database."); + FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str(), options); + ASSERT_OK_AND_ASSIGN(auto sys_tables, catalog.ListTables("sys")); + ASSERT_FALSE(sys_tables.empty()); + // 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) { @@ -617,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()), @@ -686,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, @@ -742,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 @@ -784,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 @@ -851,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 @@ -913,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 @@ -970,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 @@ -1074,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); @@ -1102,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, "")); @@ -1136,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"), ""), @@ -1194,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 new file mode 100644 index 000000000..48eb9b33a --- /dev/null +++ b/src/paimon/core/table/system/global_system_tables.cpp @@ -0,0 +1,555 @@ +/* + * 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 "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/options_utils.h" +#include "paimon/common/utils/path_util.h" +#include "paimon/common/utils/string_utils.h" +#include "paimon/core/core_options.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/defs.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; + bool requires_catalog; + GlobalSystemTableFactory factory; +}; + +const std::vector& GlobalSystemTableRegistry() { + static const std::vector registry = { + {CatalogOptionsSystemTable::kName, false, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {AllTableOptionsSystemTable::kName, true, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {TablesSystemTable::kName, true, + [](const GlobalSystemTableContext& ctx) -> Result> { + return std::make_shared(ctx); + }}, + {PartitionsSystemTable::kName, true, + [](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::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; + int64_t file_size_in_bytes = 0; + int64_t file_count = 0; + 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. +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)); + + // 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, + core_options.GetCache(), 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(); + + // Convert partition BinaryRow to string representation + std::string partition_key; + if (entry.Partition().GetFieldCount() > 0) { + 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.by_partition[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 + +// ============================================================================= +// GlobalSystemTableLoader +// ============================================================================= + +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 IsEnabled(entry, catalog_options); + } + } + 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) { + 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); +} + +Result> GlobalSystemTableLoader::GetSupportedTableNames( + const std::map& catalog_options) { + std::vector names; + names.reserve(GlobalSystemTableRegistry().size()); + for (const auto& entry : GlobalSystemTableRegistry()) { + PAIMON_ASSIGN_OR_RAISE(bool enabled, IsEnabled(entry, catalog_options)); + if (enabled) { + 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, StringValue(key)); + row.SetField(1, StringValue(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.value(); + 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, StringValue(db)); + row.SetField(1, StringValue(table)); + row.SetField(2, StringValue(key)); + row.SetField(3, StringValue(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::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::int64(), /*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.value(); + auto data_schema = std::dynamic_pointer_cast(schema_ptr); + if (!data_schema) { + continue; + } + + const auto& opts = data_schema->Options(); + 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(); + + 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, !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() && 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; + 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(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(10, NullType()); + row.SetField(11, NullType()); + row.SetField(12, NullType()); + row.SetField(13, 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_file_creation_time", arrow::int64(), /*nullable=*/true), + arrow::field("done", arrow::boolean(), /*nullable=*/false), + }); +} + +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()); + 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().by_partition; + for (const auto& [partition_key, stats] : stats_map) { + if (stats.file_count == 0) { + continue; + } + 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(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(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)); + } + } + } + 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..28160f103 --- /dev/null +++ b/src/paimon/core/table/system/global_system_tables.h @@ -0,0 +1,120 @@ +/* + * 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 { + const Catalog* catalog = nullptr; // 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"; + static constexpr const char* kEnabledOption = "catalog-options-table.enabled"; + + 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 Result IsSupported(const std::string& table_name, + const std::map& catalog_options); + + static Result> Load(const std::string& table_name, + const GlobalSystemTableContext& context); + + 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 f3a0a7f6b..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" @@ -30,6 +31,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 +190,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 == Catalog::SYSTEM_DATABASE_NAME) { + 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 +210,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 +225,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`. 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 5bbede302..ce15d3e84 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 @@ -49,6 +50,7 @@ #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" @@ -3717,4 +3719,310 @@ 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"}, + {CatalogOptionsSystemTable::kEnabledOption, "true"}, + {"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"}, + {"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"); + 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); + + 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=*/{}, no_pk_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); + 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 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; + 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"); + EXPECT_EQ(std::string(type_array->GetString(i)), "TABLE"); + EXPECT_FALSE(part_array->Value(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; + } 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) { + std::map options = {{Options::FILE_SYSTEM, "local"}, + {Options::FILE_FORMAT, "orc"}, + {Options::MANIFEST_FORMAT, "orc"}, + {Options::BUCKET, "1"}, + {Options::BUCKET_KEY, "v"}}; + auto dir = UniqueTestDirectory::Create(); + ASSERT_TRUE(dir); + std::string warehouse = dir->Str(); + + 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)); + + 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)); + 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