feat(catalog, schema): Add existence check and schema improvements

include/paimon/catalog/catalog.h

@@ -95,6 +95,18 @@ class PAIMON_EXPORT Catalog {
     /// status.
     virtual Result<std::vector<std::string>> ListTables(const std::string& db_name) 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.
+    /// @return A result containing true if the database exists, false otherwise, or an error status.
+    virtual Result<bool> DataBaseExists(const std::string& db_name) const = 0;
+
+    /// Checks whether a table with the specified identifier exists in the catalog.
+    ///
+    /// @param identifier The identifier of the table to check for existence.
+    /// @return A result containing true if the table exists, false otherwise, or an error status.
+    virtual Result<bool> TableExists(const Identifier& identifier) const = 0;
+
     /// Loads the latest schema of a specified table.
     ///
     /// @note System tables will not be supported.

include/paimon/schema/schema.h

@@ -39,6 +39,13 @@ class PAIMON_EXPORT Schema {
     /// @return A result containing an ArrowSchema, or an error status if conversion fails.
     virtual Result<std::unique_ptr<::ArrowSchema>> GetArrowSchema() const = 0;
 
+    /// Get the JSON schema representation of this table schema.
+    ///
+    /// This method provides a JSON string that represents the complete schema information.
+    ///
+    /// @return A string containing the JSON schema, or an error status on failure.
+    virtual Result<std::string> GetJsonSchema() const = 0;
+
     /// Get the names of all fields in the table schema.
     /// @return A vector of field names.
     virtual std::vector<std::string> FieldNames() const = 0;

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_impl.h"
 #include "paimon/core/schema/schema_manager.h"
 #include "paimon/fs/file_system.h"
 #include "paimon/logging.h"
@@ -87,6 +86,12 @@ Result<bool> FileSystemCatalog::DataBaseExists(const std::string& db_name) const
     return fs_->Exists(NewDatabasePath(warehouse_, db_name));
 }
 
+Result<bool> FileSystemCatalog::TableExists(const Identifier& identifier) const {
+    PAIMON_ASSIGN_OR_RAISE(std::optional<std::shared_ptr<TableSchema>> latest_schema,
+                       TableSchemaExists(identifier));
+    return latest_schema != std::nullopt;
+}
+
 Status FileSystemCatalog::CreateTable(const Identifier& identifier, ArrowSchema* c_schema,
                                       const std::vector<std::string>& partition_keys,
                                       const std::vector<std::string>& primary_keys,
@@ -216,7 +221,7 @@ Result<std::shared_ptr<Schema>> FileSystemCatalog::LoadTableSchema(
     if (!latest_schema) {
         return Status::NotExist(fmt::format("{} not exist", identifier.ToString()));
     }
-    return std::make_shared<SchemaImpl>(*latest_schema);
+    return std::static_pointer_cast<Schema>(*latest_schema);
 }
 
 }  // namespace paimon

src/paimon/core/catalog/file_system_catalog.h

@@ -49,6 +49,8 @@ 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<bool> DataBaseExists(const std::string& db_name) const override;
+    Result<bool> TableExists(const Identifier& identifier) const override;
     Result<std::shared_ptr<Schema>> LoadTableSchema(const Identifier& identifier) const override;
 
  private:
@@ -57,7 +59,6 @@ class FileSystemCatalog : public Catalog {
     static bool IsSystemDatabase(const std::string& db_name);
     static bool IsSpecifiedSystemTable(const Identifier& identifier);
     static bool IsSystemTable(const Identifier& identifier);
-    Result<bool> DataBaseExists(const std::string& db_name) const;
     Result<std::optional<std::shared_ptr<TableSchema>>> TableSchemaExists(
         const Identifier& identifier) const;
 

src/paimon/core/schema/table_schema.cpp

@@ -27,6 +27,7 @@
 #include "fmt/format.h"
 #include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/date_time_utils.h"
+#include "paimon/common/utils/field_type_utils.h"
 #include "paimon/common/utils/object_utils.h"
 #include "paimon/common/utils/options_utils.h"
 #include "paimon/common/utils/rapidjson_util.h"
@@ -169,6 +170,14 @@ bool TableSchema::operator==(const TableSchema& other) const {
            options_ == other.options_ && comment_ == other.comment_ &&
            time_millis_ == other.time_millis_;
 }
+
+Result<std::unique_ptr<::ArrowSchema>> TableSchema::GetArrowSchema() const {
+    auto schema = DataField::ConvertDataFieldsToArrowSchema(fields_);
+    auto c_schema = std::make_unique<::ArrowSchema>();
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(arrow::ExportSchema(*schema, c_schema.get()));
+    return c_schema;
+}
+
 std::vector<std::string> TableSchema::FieldNames() const {
     std::vector<std::string> field_names;
     field_names.reserve(fields_.size());

src/paimon/core/schema/table_schema.h

@@ -27,6 +27,7 @@
 #include "paimon/common/types/data_field.h"
 #include "paimon/common/utils/jsonizable.h"
 #include "paimon/result.h"
+#include "paimon/schema/schema.h"
 #include "rapidjson/allocators.h"
 #include "rapidjson/document.h"
 #include "rapidjson/rapidjson.h"
@@ -35,7 +36,7 @@ struct ArrowSchema;
 
 namespace paimon {
 /// Schema of a table, including schemaId and fieldId.
-class TableSchema : public Jsonizable<TableSchema> {
+class TableSchema : public Schema, public Jsonizable<TableSchema> {
  public:
     static constexpr int64_t FIRST_SCHEMA_ID = 0;
     static constexpr int32_t PAIMON_07_VERSION = 1;
@@ -57,28 +58,35 @@ class TableSchema : public Jsonizable<TableSchema> {
 
     bool operator==(const TableSchema& other) const;
 
-    std::vector<std::string> FieldNames() const;
-    int64_t Id() const {
+    Result<std::unique_ptr<::ArrowSchema>> GetArrowSchema() const override;
+
+    Result<std::string> GetJsonSchema() const override {
+        return ToJsonString();
+    }
+
+    std::vector<std::string> FieldNames() const override;
+
+    int64_t Id() const override {
         return id_;
     }
-    const std::vector<std::string>& PrimaryKeys() const {
+    const std::vector<std::string>& PrimaryKeys() const override {
         return primary_keys_;
     }
-    const std::vector<std::string>& PartitionKeys() const {
+    const std::vector<std::string>& PartitionKeys() const override {
         return partition_keys_;
     }
 
-    const std::vector<std::string>& BucketKeys() const {
+    const std::vector<std::string>& BucketKeys() const override {
         return bucket_keys_;
     }
 
-    int32_t NumBuckets() const {
+    int32_t NumBuckets() const override {
         return num_bucket_;
     }
-    int32_t HighestFieldId() const {
+    int32_t HighestFieldId() const override {
         return highest_field_id_;
     }
-    const std::map<std::string, std::string>& Options() const {
+    const std::map<std::string, std::string>& Options() const override {
         return options_;
     }
     const std::vector<DataField>& Fields() const {
@@ -92,7 +100,7 @@ class TableSchema : public Jsonizable<TableSchema> {
     Result<std::vector<DataField>> GetFields(const std::vector<std::string>& field_names) const;
     Result<std::vector<std::string>> TrimmedPrimaryKeys() const;
 
-    std::optional<std::string> Comment() const {
+    std::optional<std::string> Comment() const override {
         return comment_;
     }
 
@@ -130,7 +138,7 @@ class TableSchema : public Jsonizable<TableSchema> {
     std::vector<std::string> partition_keys_;
     std::vector<std::string> primary_keys_;
     std::vector<std::string> bucket_keys_;
-    int32_t num_bucket_;
+    int32_t num_bucket_ = -1;
     std::map<std::string, std::string> options_;
     std::optional<std::string> comment_;
     int64_t time_millis_ = -1;