feat(catalog): enrich catalog interface with more methods

include/paimon/catalog/catalog.h

@@ -16,23 +16,33 @@
 
 #pragma once
 
+#include <chrono>
 #include <map>
 #include <memory>
 #include <string>
+#include <variant>
 #include <vector>
 
 #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"
 
 struct ArrowSchema;
 
 namespace paimon {
+// Tag name or snapshot id
+using Instant = std::variant<std::string, int64_t>;
+
+class Database;
+class Table;
+class View;
+class Schema;
+class Snapshot;
+class PartitionStatistics;
+class Tag;
 class Identifier;
-
 /// This interface is responsible for reading and writing metadata such as database/table from a
 /// paimon catalog.
 class PAIMON_EXPORT Catalog {
@@ -99,6 +109,38 @@ class PAIMON_EXPORT Catalog {
     /// status.
     virtual Result<std::vector<std::string>> ListTables(const std::string& db_name) const = 0;
 
+    /// Drops a database.
+    ///
+    /// @param name Name of the database to be dropped.
+    /// @param ignore_if_not_exists If true, no action is taken if the database does not exist.
+    /// @param cascade If true, drops all tables and functions in the database before dropping the
+    /// database.
+    /// @return A status indicating success or failure.
+    virtual Status DropDatabase(const std::string& name, bool ignore_if_not_exists,
+                                bool cascade) = 0;
+
+    /// Drops a table.
+    ///
+    /// @param identifier Identifier of the table to drop.
+    /// @param ignore_if_not_exists If true, no action is taken if the table does not exist.
+    /// @return A status indicating success or failure.
+    virtual Status DropTable(const Identifier& identifier, bool ignore_if_not_exists) = 0;
+
+    /// Renames a table.
+    ///
+    /// @param from_table Current identifier of the table.
+    /// @param to_table New identifier for the table.
+    /// @param ignore_if_not_exists If true, no action is taken if the table does not exist.
+    /// @return A status indicating success or failure.
+    virtual Status RenameTable(const Identifier& from_table, const Identifier& to_table,
+                               bool ignore_if_not_exists) = 0;
+
+    /// Gets a table.
+    ///
+    /// @param identifier Identifier of the table to get.
+    /// @return A result containing the table, or an error status.
+    virtual Result<std::shared_ptr<Table>> GetTable(const Identifier& identifier) const = 0;
+
     /// Checks whether a database with the specified name exists in the catalog.
     ///
     /// @param db_name The name of the database to check for existence.

include/paimon/catalog/database.h

@@ -0,0 +1,50 @@
+/*
+ * 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 <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "paimon/type_fwd.h"
+#include "paimon/visibility.h"
+
+struct ArrowSchema;
+
+namespace paimon {
+
+/// Interface of a database in a catalog.
+class PAIMON_EXPORT Database {
+ public:
+    /// ================== Table Metadata =====================
+
+    /// A name to identify this database.
+    virtual std::string Name() const = 0;
+
+    /// Get the table-level options associated with this schema.
+    /// @return Options
+    virtual const std::map<std::string, std::string>& Options() const = 0;
+
+    /// Get an optional comment describing the table.
+    /// @return The table comment if set, or std::nullopt otherwise.
+    virtual std::optional<std::string> Comment() const = 0;
+};
+
+}  // namespace paimon

src/paimon/core/catalog/file_system_catalog.cpp

@@ -27,7 +27,6 @@
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/path_util.h"
 #include "paimon/common/utils/string_utils.h"
-#include "paimon/core/schema/schema_manager.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/logging.h"
 #include "paimon/result.h"
@@ -240,4 +239,110 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
     return std::static_pointer_cast<Schema>(*latest_schema);
 }
 
+Result<std::shared_ptr<Table>> FileSystemCatalog::GetTable(const Identifier& identifier) const {
+    std::string table_path = GetTableLocation(identifier);
+    PAIMON_ASSIGN_OR_RAISE(bool exist, fs_->Exists(table_path));
+    if (!exist) {
+        return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
+    }
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
+                           TableSchemaExists(identifier));
+    if (!latest_schema) {
+        return Status::NotExist(
+            fmt::format("load table schema for {} failed", identifier.ToString()));
+    }
+    auto schema = std::static_pointer_cast<Schema>(*latest_schema);
+    return std::make_shared<Table>(schema, identifier.GetDatabaseName(), identifier.GetTableName());
+}
+
+Status FileSystemCatalog::DropDatabase(const std::string& name, bool ignore_if_not_exists,
+                                       bool cascade) {
+    if (IsSystemDatabase(name)) {
+        return Status::Invalid(fmt::format("Cannot drop system database {}.", name));
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(bool exist, DatabaseExists(name));
+    if (!exist) {
+        if (ignore_if_not_exists) {
+            return Status::OK();
+        } else {
+            return Status::NotExist(fmt::format("database {} does not exist", name));
+        }
+    }
+
+    std::string db_path = NewDatabasePath(warehouse_, name);
+
+    if (cascade) {
+        // List all tables in the database and drop them
+        PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> tables, ListTables(name));
+        for (const std::string& table_name : tables) {
+            Identifier table_id(name, table_name);
+            PAIMON_RETURN_NOT_OK(DropTable(table_id, false));
+        }
+    } else {
+        // Check if database is empty
+        PAIMON_ASSIGN_OR_RAISE(std::vector<std::string> tables, ListTables(name));
+        if (!tables.empty()) {
+            return Status::Invalid(
+                fmt::format("Cannot drop non-empty database {}. Use cascade=true to force.", name));
+        }
+    }
+
+    // Delete the database directory
+    PAIMON_RETURN_NOT_OK(fs_->Delete(db_path));
+    return Status::OK();
+}
+
+Status FileSystemCatalog::DropTable(const Identifier& identifier, bool ignore_if_not_exists) {
+    if (IsSystemTable(identifier)) {
+        return Status::Invalid(fmt::format("Cannot drop system table {}.", identifier.ToString()));
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(bool exist, TableExists(identifier));
+    if (!exist) {
+        if (ignore_if_not_exists) {
+            return Status::OK();
+        } else {
+            return Status::NotExist(fmt::format("table {} does not exist", identifier.ToString()));
+        }
+    }
+
+    std::string table_path = GetTableLocation(identifier);
+    PAIMON_RETURN_NOT_OK(fs_->Delete(table_path));
+    return Status::OK();
+}
+
+Status FileSystemCatalog::RenameTable(const Identifier& from_table, const Identifier& to_table,
+                                      bool ignore_if_not_exists) {
+    if (IsSystemTable(from_table) || IsSystemTable(to_table)) {
+        return Status::Invalid(fmt::format("Cannot rename system table {} or {}.",
+                                           from_table.ToString(), to_table.ToString()));
+    }
+
+    if (from_table.GetDatabaseName() != to_table.GetDatabaseName()) {
+        return Status::Invalid(
+            "Cannot rename table across databases. Cross-database rename is not supported.");
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(bool from_exist, TableExists(from_table));
+    if (!from_exist) {
+        if (ignore_if_not_exists) {
+            return Status::OK();
+        } else {
+            return Status::NotExist(
+                fmt::format("source table {} does not exist", from_table.ToString()));
+        }
+    }
+
+    PAIMON_ASSIGN_OR_RAISE(bool to_exist, TableExists(to_table));
+    if (to_exist) {
+        return Status::Invalid(fmt::format("target table {} already exists", to_table.ToString()));
+    }
+
+    std::string from_path = GetTableLocation(from_table);
+    std::string to_path = GetTableLocation(to_table);
+    PAIMON_RETURN_NOT_OK(fs_->Rename(from_path, to_path));
+    return Status::OK();
+}
+
 }  // namespace paimon

src/paimon/core/catalog/file_system_catalog.h

@@ -22,6 +22,8 @@
 #include <vector>
 
 #include "paimon/catalog/catalog.h"
+#include "paimon/core/schema/schema_manager.h"
+#include "paimon/core/table/table.h"
 #include "paimon/logging.h"
 #include "paimon/result.h"
 #include "paimon/status.h"
@@ -46,7 +48,10 @@ class FileSystemCatalog : public Catalog {
                        const std::vector<std::string>& primary_keys,
                        const std::map<std::string, std::string>& options,
                        bool ignore_if_exists) override;
-
+    Status DropDatabase(const std::string& name, bool ignore_if_not_exists, bool cascade) override;
+    Status DropTable(const Identifier& identifier, bool ignore_if_not_exists) override;
+    Status RenameTable(const Identifier& from_table, const Identifier& to_table,
+                       bool ignore_if_not_exists) override;
     Result<std::vector<std::string>> ListDatabases() const override;
     Result<std::vector<std::string>> ListTables(const std::string& db_name) const override;
     Result<bool> DatabaseExists(const std::string& db_name) const override;
@@ -56,6 +61,7 @@ class FileSystemCatalog : public Catalog {
     Result<std::shared_ptr<Schema>> LoadTableSchema(const Identifier& identifier) const override;
     std::string GetRootPath() const override;
     std::shared_ptr<FileSystem> GetFileSystem() const override;
+    Result<std::shared_ptr<Table>> GetTable(const Identifier& identifier) const override;
 
  private:
     static std::string NewDatabasePath(const std::string& warehouse, const std::string& db_name);