feat(catalog): add LoadTableSchema interface

Author: dalingmeng

include/paimon/catalog/catalog.h

@@ -23,6 +23,7 @@
 
 #include "paimon/catalog/identifier.h"
 #include "paimon/result.h"
+#include "paimon/schema/schema.h"
 #include "paimon/status.h"
 #include "paimon/type_fwd.h"
 #include "paimon/visibility.h"
@@ -93,6 +94,16 @@ class PAIMON_EXPORT Catalog {
     /// @return A result containing a vector of table names in the specified database, or an error
     /// status.
     virtual Result<std::vector<std::string>> ListTables(const std::string& db_name) const = 0;
+
+    /// Loads the latest schema of a specified table.
+    ///
+    /// @note System tables will not be supported.
+    ///
+    /// @param identifier The identifier (database and table name) of the table to load.
+    /// @return A result containing table schema if the table exists, or std::nullopt if it
+    /// doesn't, or an error status on failure.
+    virtual Result<std::optional<std::shared_ptr<Schema>>> LoadTableSchema(
+        const Identifier& identifier) const = 0;
 };
 
 }  // namespace paimon

include/paimon/schema/schema.h

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2025-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 <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "arrow/api.h"
+#include "arrow/c/bridge.h"
+#include "paimon/result.h"
+#include "paimon/visibility.h"
+struct ArrowSchema;
+
+namespace paimon {
+class Catalog;
+class Identifier;
+/// This interface provides access to TableSchema-related information.
+class PAIMON_EXPORT Schema {
+ public:
+    Schema();
+    ~Schema();
+
+ public:
+    /// Get the Arrow C schema representation of this table schema.
+    /// @return A result containing an ArrowSchema, or an error status if conversion fails.
+    Result<std::unique_ptr<::ArrowSchema>> GetArrowSchema() const;
+
+    /// Get the names of all fields in the table schema.
+    /// @return A vector of field names.
+    std::vector<std::string> FieldNames() const;
+
+    /// Get the unique identifier of this table schema.
+    /// @return The schema ID
+    int64_t Id() const;
+
+    /// Get the list of primary key field names.
+    /// @return A reference to the vector of primary key names; empty if no primary keys are
+    /// defined.
+    const std::vector<std::string>& PrimaryKeys() const;
+
+    /// Get the list of partition key field names.
+    /// @return A reference to the vector of partition key names; empty if the table is not
+    /// partitioned.
+    const std::vector<std::string>& PartitionKeys() const;
+
+    /// Get the list of bucket key field names used for bucketing.
+    /// @return A reference to the vector of bucket key names.
+    const std::vector<std::string>& BucketKeys() const;
+
+    /// Get the number of buckets configured for this table.
+    /// @return The number of buckets;
+    int32_t NumBuckets() const;
+
+    /// Get the highest field ID assigned in this schema.
+    /// @return The maximum field ID.
+    int32_t HighestFieldId() const;
+
+    /// Get the table-level options associated with this schema.
+    /// @return A reference to the map of option key-value pairs (e.g., file format, filesystem).
+    const std::map<std::string, std::string>& Options();
+
+    /// Get an optional comment describing the table.
+    /// @return The table comment if set, or std::nullopt otherwise.
+    std::optional<std::string> Comment();
+
+ public:
+    /// Initializes the internal implementation of the Schema
+    /// @param catalog Pointer to the Catalog instance.
+    /// @param identifier The identifier of the table whose schema is to be loaded.
+    /// @return A result indicating success (true) or failure (false).
+
+    Result<bool> MakeImpl(const Catalog* catalog, const Identifier& identifier);
+
+    class Impl;
+
+ private:
+    std::unique_ptr<Impl> impl_;
+};
+
+}  // namespace paimon

src/paimon/CMakeLists.txt

@@ -218,6 +218,7 @@ set(PAIMON_CORE_SRCS
     core/schema/schema_manager.cpp
     core/schema/schema_validation.cpp
     core/schema/table_schema.cpp
+    core/schema/schema.cpp
     core/snapshot.cpp
     core/stats/simple_stats_collector.cpp
     core/stats/simple_stats_converter.cpp

src/paimon/core/catalog/file_system_catalog.cpp

@@ -211,4 +211,17 @@ Result<bool> FileSystemCatalog::TableExistsInFileSystem(const std::string& table
     }
 }
 
+Result<std::optional<std::shared_ptr<Schema>>> FileSystemCatalog::LoadTableSchema(
+    const Identifier& identifier) const {
+    if (IsSystemTable(identifier)) {
+        return Status::NotImplemented("do not support loading schema for system table.");
+    }
+    std::shared_ptr<Schema> schema = std::make_shared<Schema>();
+    PAIMON_ASSIGN_OR_RAISE(bool impl_status, schema->MakeImpl(this, identifier));
+    if (impl_status) {
+        return std::optional<std::shared_ptr<Schema>>(schema);
+    }
+    return std::optional<std::shared_ptr<Schema>>();
+}
+
 }  // namespace paimon

src/paimon/core/catalog/file_system_catalog.h

@@ -49,6 +49,14 @@ class FileSystemCatalog : public Catalog {
 
     Result<std::vector<std::string>> ListDatabases() const override;
     Result<std::vector<std::string>> ListTables(const std::string& database_names) const override;
+    Result<std::optional<std::shared_ptr<Schema>>> LoadTableSchema(
+        const Identifier& identifier) const override;
+    std::shared_ptr<FileSystem> GetFileSystem() const {
+        return fs_;
+    }
+    const std::string& Warehouse() const {
+        return warehouse_;
+    }
 
  private:
     static std::string NewDatabasePath(const std::string& warehouse, const std::string& db_name);

src/paimon/core/catalog/file_system_catalog_test.cpp

@@ -180,6 +180,12 @@ TEST(FileSystemCatalogTest, TestCreateTableWithBlob) {
     ASSERT_OK_AND_ASSIGN(std::vector<std::string> table_names, catalog.ListTables("db1"));
     ASSERT_EQ(1, table_names.size());
     ASSERT_EQ(table_names[0], "tbl1");
+    ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<Schema>> table_schema,
+                         catalog.LoadTableSchema(Identifier("db1", "tbl1")));
+    ASSERT_TRUE(table_schema.has_value());
+    ASSERT_OK_AND_ASSIGN(auto arrow_schema, (*table_schema)->GetArrowSchema());
+    auto loaded_schema = arrow::ImportSchema(arrow_schema.get()).ValueOrDie();
+    ASSERT_TRUE(typed_schema.Equals(loaded_schema));
     ArrowSchemaRelease(&schema);
 }
 
@@ -309,4 +315,56 @@ TEST(FileSystemCatalogTest, TestInvalidList) {
                         "do not support listing tables for system database.");
 }
 
+TEST(FileSystemCatalogTest, TestValidateTableSchema) {
+    std::map<std::string, std::string> options;
+    options[Options::FILE_SYSTEM] = "local";
+    options[Options::FILE_FORMAT] = "orc";
+    ASSERT_OK_AND_ASSIGN(auto core_options, CoreOptions::FromMap(options));
+    auto dir = UniqueTestDirectory::Create();
+    ASSERT_TRUE(dir);
+    FileSystemCatalog catalog(core_options.GetFileSystem(), dir->Str());
+    {
+        ASSERT_OK(catalog.CreateDatabase("db1", options, /*ignore_if_exists=*/true));
+        arrow::FieldVector fields = {
+            arrow::field("f0", arrow::utf8()),
+            arrow::field("f1", arrow::int32()),
+            arrow::field("f2", arrow::int32()),
+            arrow::field("f3", arrow::float64()),
+        };
+        arrow::Schema typed_schema(fields);
+        ::ArrowSchema schema;
+        ASSERT_TRUE(arrow::ExportSchema(typed_schema, &schema).ok());
+        ASSERT_OK(catalog.CreateTable(Identifier("db1", "tbl1"), &schema, {"f1"}, {}, options,
+                                      /*ignore_if_exists=*/false));
+
+        ASSERT_OK_AND_ASSIGN(std::optional<std::shared_ptr<Schema>> table_schema,
+                             catalog.LoadTableSchema(Identifier("db0", "tbl0")));
+        ASSERT_FALSE(table_schema.has_value());
+        ASSERT_OK_AND_ASSIGN(table_schema, catalog.LoadTableSchema(Identifier("db1", "tbl1")));
+        ASSERT_TRUE(table_schema.has_value());
+        ASSERT_EQ(0, (*table_schema)->Id());
+        ASSERT_EQ(3, (*table_schema)->HighestFieldId());
+        ASSERT_EQ(1, (*table_schema)->PartitionKeys().size());
+        ASSERT_EQ(0, (*table_schema)->PrimaryKeys().size());
+        ASSERT_EQ(-1, (*table_schema)->NumBuckets());
+        ASSERT_FALSE((*table_schema)->Comment().has_value());
+        std::vector<std::string> fieldNames = (*table_schema)->FieldNames();
+        std::vector<std::string> expectedFieldNames = {"f0", "f1", "f2", "f3"};
+        ASSERT_EQ(fieldNames, expectedFieldNames);
+
+        ASSERT_OK_AND_ASSIGN(auto arrow_schema, (*table_schema)->GetArrowSchema());
+        auto loaded_schema = arrow::ImportSchema(arrow_schema.get()).ValueOrDie();
+        ASSERT_TRUE(typed_schema.Equals(loaded_schema));
+
+        ASSERT_OK_AND_ASSIGN(auto fs, FileSystemFactory::Get("local", dir->Str(), {}));
+        ASSERT_OK(fs->Delete(PathUtil::JoinPath(dir->Str(), "db1.db/tbl1/schema/schema-0")));
+        ASSERT_OK_AND_ASSIGN(table_schema, catalog.LoadTableSchema(Identifier("db1", "tbl1")));
+        ASSERT_FALSE(table_schema.has_value());
+
+        ASSERT_NOK_WITH_MSG(catalog.LoadTableSchema(Identifier("db1", "tbl$11")),
+                            "do not support loading schema for system table.");
+        ArrowSchemaRelease(&schema);
+    }
+}
+
 }  // namespace paimon::test

src/paimon/core/schema/schema.cpp

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2025-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/schema/schema.h"
+
+#include "paimon/catalog/identifier.h"
+#include "paimon/core/catalog/file_system_catalog.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/schema/table_schema.h"
+
+namespace paimon {
+
+class Schema::Impl {
+ public:
+    std::shared_ptr<TableSchema> table_schema_;
+};
+
+Schema::Schema() = default;
+
+Schema::~Schema() = default;
+
+Result<bool> Schema::MakeImpl(const Catalog* catalog, const Identifier& identifier) {
+    auto fs_catalog = dynamic_cast<const FileSystemCatalog*>(catalog);
+    if (fs_catalog == nullptr) {
+        return false;
+    }
+    impl_ = std::make_unique<Impl>();
+    SchemaManager schema_manager(fs_catalog->GetFileSystem(),
+                                 fs_catalog->NewDataTablePath(fs_catalog->Warehouse(), identifier));
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
+                           schema_manager.Latest());
+    if (latest_schema.has_value()) {
+        impl_->table_schema_ = latest_schema.value();
+        return true;
+    }
+    return false;
+}
+
+Result<std::unique_ptr<::ArrowSchema>> Schema::GetArrowSchema() const {
+    return impl_->table_schema_->GetArrowSchema();
+}
+
+std::vector<std::string> Schema::FieldNames() const {
+    return impl_->table_schema_->FieldNames();
+}
+
+int64_t Schema::Id() const {
+    return impl_->table_schema_->Id();
+}
+
+const std::vector<std::string>& Schema::PrimaryKeys() const {
+    return impl_->table_schema_->PrimaryKeys();
+}
+
+const std::vector<std::string>& Schema::PartitionKeys() const {
+    return impl_->table_schema_->PartitionKeys();
+}
+
+const std::vector<std::string>& Schema::BucketKeys() const {
+    return impl_->table_schema_->BucketKeys();
+}
+
+int32_t Schema::NumBuckets() const {
+    return impl_->table_schema_->NumBuckets();
+}
+
+int32_t Schema::HighestFieldId() const {
+    return impl_->table_schema_->HighestFieldId();
+}
+
+const std::map<std::string, std::string>& Schema::Options() {
+    return impl_->table_schema_->Options();
+}
+
+std::optional<std::string> Schema::Comment() {
+    return impl_->table_schema_->Commit();
+}
+
+}  // namespace paimon

src/paimon/core/schema/table_schema.cpp

@@ -22,6 +22,7 @@
 #include <utility>
 
 #include "arrow/api.h"
+#include "arrow/c/bridge.h"
 #include "arrow/util/checked_cast.h"
 #include "fmt/format.h"
 #include "paimon/common/utils/arrow/status_utils.h"
@@ -333,4 +334,11 @@ bool TableSchema::CrossPartitionUpdate() const {
     return !ObjectUtils::ContainsAll(primary_keys_, partition_keys_);
 }
 
+Result<std::unique_ptr<::ArrowSchema>> TableSchema::GetArrowSchema() const {
+    std::shared_ptr<arrow::Schema> schema = DataField::ConvertDataFieldsToArrowSchema(fields_);
+    auto arrow_schema = std::make_unique<::ArrowSchema>();
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, arrow_schema.get()));
+    return arrow_schema;
+}
+
 }  // namespace paimon

src/paimon/core/schema/table_schema.h

@@ -98,6 +98,8 @@ class TableSchema : public Jsonizable<TableSchema> {
 
     bool CrossPartitionUpdate() const;
 
+    Result<std::unique_ptr<::ArrowSchema>> GetArrowSchema() const;
+
  private:
     JSONIZABLE_FRIEND_AND_DEFAULT_CTOR(TableSchema);