[feature](paimon) implement create/drop db, create/drop table for paimon

fe/fe-core/src/main/java/org/apache/doris/catalog/TableIf.java

@@ -488,6 +488,8 @@ public String toEngineName() {
                 case ICEBERG:
                 case ICEBERG_EXTERNAL_TABLE:
                     return "iceberg";
+                case PAIMON_EXTERNAL_TABLE:
+                    return "paimon";
                 case DICTIONARY:
                     return "dictionary";
                 case DORIS_EXTERNAL_TABLE:

fe/fe-core/src/main/java/org/apache/doris/datasource/operations/ExternalMetadataOperations.java

@@ -21,6 +21,7 @@
 import org.apache.doris.datasource.hive.HMSExternalCatalog;
 import org.apache.doris.datasource.hive.HiveMetadataOps;
 import org.apache.doris.datasource.iceberg.IcebergMetadataOps;
+import org.apache.doris.datasource.paimon.PaimonMetadataOps;
 
 import org.apache.hadoop.hive.conf.HiveConf;
 import org.apache.iceberg.catalog.Catalog;
@@ -35,4 +36,9 @@ public static HiveMetadataOps newHiveMetadataOps(HiveConf hiveConf, HMSExternalC
     public static IcebergMetadataOps newIcebergMetadataOps(ExternalCatalog dorisCatalog, Catalog catalog) {
         return new IcebergMetadataOps(dorisCatalog, catalog);
     }
+
+    public static PaimonMetadataOps newPaimonMetaOps(ExternalCatalog dorisCatalog,
+                                                     org.apache.paimon.catalog.Catalog catalog) {
+        return new PaimonMetadataOps(dorisCatalog, catalog);
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/DorisToPaimonTypeVisitor.java

@@ -0,0 +1,109 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+
+package org.apache.doris.datasource.paimon;
+
+import org.apache.doris.catalog.ArrayType;
+import org.apache.doris.catalog.MapType;
+import org.apache.doris.catalog.PrimitiveType;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.catalog.StructField;
+import org.apache.doris.catalog.StructType;
+import org.apache.doris.catalog.Type;
+import org.apache.doris.datasource.DorisTypeVisitor;
+
+import org.apache.paimon.types.BigIntType;
+import org.apache.paimon.types.BooleanType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DateType;
+import org.apache.paimon.types.DecimalType;
+import org.apache.paimon.types.DoubleType;
+import org.apache.paimon.types.FloatType;
+import org.apache.paimon.types.IntType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.types.TimestampType;
+import org.apache.paimon.types.VarBinaryType;
+import org.apache.paimon.types.VarCharType;
+import org.apache.paimon.types.VariantType;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class DorisToPaimonTypeVisitor extends DorisTypeVisitor<DataType> {
+
+    @Override
+    public DataType struct(StructType struct, List<DataType> fieldResults) {
+        List<StructField> fields = struct.getFields();
+        List<DataField> newFields = new ArrayList<>(fields.size());
+        AtomicInteger atomicInteger = new AtomicInteger(-1);
+        for (int i = 0; i < fields.size(); i++) {
+            StructField field = fields.get(i);
+            DataType fieldType = fieldResults.get(i).copy(field.getContainsNull());
+            String comment = field.getComment();
+            DataField dataField = new DataField(atomicInteger.incrementAndGet(), field.getName(), fieldType, comment);
+            newFields.add(dataField);
+        }
+        return new RowType(newFields);
+    }
+
+    @Override
+    public DataType field(StructField field, DataType typeResult) {
+        return typeResult;
+    }
+
+    @Override
+    public DataType array(ArrayType array, DataType elementResult) {
+        return new org.apache.paimon.types.ArrayType(elementResult.copy(array.getContainsNull()));
+    }
+
+    @Override
+    public DataType map(MapType map, DataType keyResult, DataType valueResult) {
+        return new org.apache.paimon.types.MapType(keyResult.copy(false),
+            valueResult.copy(map.getIsValueContainsNull()));
+    }
+
+    @Override
+    public DataType atomic(Type atomic) {
+        PrimitiveType primitiveType = atomic.getPrimitiveType();
+        if (primitiveType.equals(PrimitiveType.BOOLEAN)) {
+            return new BooleanType();
+        } else if (primitiveType.equals(PrimitiveType.INT)) {
+            return new IntType();
+        } else if (primitiveType.equals(PrimitiveType.BIGINT)) {
+            return new BigIntType();
+        } else if (primitiveType.equals(PrimitiveType.FLOAT)) {
+            return new FloatType();
+        } else if (primitiveType.equals(PrimitiveType.DOUBLE)) {
+            return new DoubleType();
+        } else if (primitiveType.isCharFamily()) {
+            return new VarCharType(VarCharType.MAX_LENGTH);
+        } else if (primitiveType.equals(PrimitiveType.DATE) || primitiveType.equals(PrimitiveType.DATEV2)) {
+            return new DateType();
+        } else if (primitiveType.equals(PrimitiveType.DECIMALV2) || primitiveType.isDecimalV3Type()) {
+            return new DecimalType(((ScalarType) atomic).getScalarPrecision(), ((ScalarType) atomic).getScalarScale());
+        } else if (primitiveType.equals(PrimitiveType.DATETIME) || primitiveType.equals(PrimitiveType.DATETIMEV2)) {
+            return new TimestampType();
+        } else if (primitiveType.isVarbinaryType()) {
+            return new VarBinaryType(VarBinaryType.MAX_LENGTH);
+        } else if (primitiveType.isVariantType()) {
+            return new VariantType();
+        }
+        throw new UnsupportedOperationException("Not a supported type: " + primitiveType);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonExternalCatalog.java

@@ -23,13 +23,13 @@
 import org.apache.doris.datasource.InitCatalogLog;
 import org.apache.doris.datasource.NameMapping;
 import org.apache.doris.datasource.SessionContext;
+import org.apache.doris.datasource.operations.ExternalMetadataOperations;
 import org.apache.doris.datasource.property.metastore.AbstractPaimonProperties;
 
 import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.paimon.catalog.Catalog;
-import org.apache.paimon.catalog.Catalog.TableNotExistException;
 import org.apache.paimon.catalog.Identifier;
 import org.apache.paimon.partition.Partition;
 
@@ -62,6 +62,7 @@ protected void initLocalObjectsImpl() {
         catalogType = paimonProperties.getPaimonCatalogType();
         catalog = createCatalog();
         initPreExecutionAuthenticator();
+        metadataOps = ExternalMetadataOperations.newPaimonMetaOps(this, catalog);
     }
 
     @Override
@@ -76,49 +77,16 @@ public String getCatalogType() {
         return catalogType;
     }
 
-    protected List<String> listDatabaseNames() {
-        try {
-            return executionAuthenticator.execute(() -> new ArrayList<>(catalog.listDatabases()));
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to list databases names, catalog name: " + getName(), e);
-        }
-    }
-
     @Override
     public boolean tableExist(SessionContext ctx, String dbName, String tblName) {
         makeSureInitialized();
-        try {
-            return executionAuthenticator.execute(() -> {
-                try {
-                    catalog.getTable(Identifier.create(dbName, tblName));
-                    return true;
-                } catch (TableNotExistException e) {
-                    return false;
-                }
-            });
-
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to check table existence, catalog name: " + getName()
-                    + "error message is:" + ExceptionUtils.getRootCauseMessage(e), e);
-        }
+        return metadataOps.tableExist(dbName, tblName);
     }
 
     @Override
     public List<String> listTableNames(SessionContext ctx, String dbName) {
         makeSureInitialized();
-        try {
-            return executionAuthenticator.execute(() -> {
-                List<String> tableNames = null;
-                try {
-                    tableNames = catalog.listTables(dbName);
-                } catch (Catalog.DatabaseNotExistException e) {
-                    LOG.warn("DatabaseNotExistException", e);
-                }
-                return tableNames;
-            });
-        } catch (Exception e) {
-            throw new RuntimeException("Failed to list table names, catalog name: " + getName(), e);
-        }
+        return metadataOps.listTableNames(dbName);
     }
 
     public List<Partition> getPaimonPartitions(NameMapping nameMapping) {