[Catalog][DM][Improve] dm catalog and dm auto create table and auto create table add constraintKey

seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/Column.java

@@ -55,7 +55,10 @@ public abstract class Column implements Serializable {
 
     protected final String comment;
 
-    /** Field type in the database * */
+    /**
+     * Field type in the database For example : varchar is varchar(50),DECIMAL is DECIMAL(20,5) ,
+     * int is int Each database can customize the sourceType according to its own characteristics*
+     */
     protected final String sourceType;
 
     /** Unsigned bit * */
@@ -64,7 +67,13 @@ public abstract class Column implements Serializable {
     /** Whether to use the 0 bit * */
     protected final boolean isZeroFill;
 
-    /** Bit length * */
+    /**
+     * Bit length For different database byte types, it is possible to define the number of bits is
+     * not the same, so the byte types parameter is converted to bits according to the database at
+     * the time of construction, and then the bitLen is converted to the child end when the
+     * automatic table is builtinteger may be cross the border For example, Mysql has 8 bits of
+     * bytes: bitLen = bytesLen << 3
+     */
     protected final Long bitLen;
 
     /** integer may be cross the border * */

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/AbstractJdbcCreateTableSqlBuilder.java

@@ -0,0 +1,43 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.jdbc.catalog;
+
+import org.apache.seatunnel.api.table.catalog.ConstraintKey;
+import org.apache.seatunnel.api.table.catalog.PrimaryKey;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+// todo Abstract the concatenation common content of jdbc auto-build statements
+public abstract class AbstractJdbcCreateTableSqlBuilder {
+
+    protected boolean primaryCompareToConstrainKey(
+            PrimaryKey primaryKey, ConstraintKey constraintKey) {
+        List<String> columnNames = primaryKey.getColumnNames();
+        List<ConstraintKey.ConstraintKeyColumn> constraintKeyColumnNames =
+                constraintKey.getColumnNames();
+        return columnNames.stream()
+                .map(Object::toString)
+                .collect(Collectors.toList())
+                .containsAll(
+                        constraintKeyColumnNames.stream()
+                                .map(ConstraintKey.ConstraintKeyColumn::getColumnName)
+                                .collect(Collectors.toList()));
+    }
+}

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/catalog/dm/DamengCatalog.java

@@ -53,14 +53,19 @@ public class DamengCatalog extends AbstractJdbcCatalog {
 
     private static final String SELECT_COLUMNS_SQL =
             "SELECT COLUMNS.COLUMN_NAME, COLUMNS.DATA_TYPE, COLUMNS.DATA_LENGTH, COLUMNS.DATA_PRECISION, COLUMNS.DATA_SCALE "
-                    + ", COLUMNS.NULLABLE, COLUMNS.DATA_DEFAULT, COMMENTS.COMMENTS "
+                    + ", COLUMNS.NULLABLE, COLUMNS.DATA_DEFAULT, COMMENTS.COMMENTS ,"
+                    + "CASE \n"
+                    + "        WHEN COLUMNS.DATA_TYPE IN ('CHAR', 'CHARACTER', 'VARCHAR', 'VARCHAR2', 'VARBINARY', 'BINARY') THEN COLUMNS.DATA_TYPE || '(' || COLUMNS.DATA_LENGTH || ')'\n"
+                    + "        WHEN COLUMNS.DATA_TYPE IN ('NUMERIC', 'DECIMAL', 'NUMBER') AND COLUMNS.DATA_PRECISION IS NOT NULL AND COLUMNS.DATA_SCALE IS NOT NULL AND COLUMNS.DATA_PRECISION != 0 AND COLUMNS.DATA_SCALE != 0 THEN COLUMNS.DATA_TYPE || '(' || COLUMNS.DATA_PRECISION || ', ' || COLUMNS.DATA_SCALE || ')'\n"
+                    + "        ELSE COLUMNS.DATA_TYPE\n"
+                    + "    END AS SOURCE_TYPE \n"
                     + "FROM ALL_TAB_COLUMNS COLUMNS "
                     + "LEFT JOIN ALL_COL_COMMENTS COMMENTS "
                     + "ON COLUMNS.OWNER = COMMENTS.SCHEMA_NAME "
                     + "AND COLUMNS.TABLE_NAME = COMMENTS.TABLE_NAME "
                     + "AND COLUMNS.COLUMN_NAME = COMMENTS.COLUMN_NAME "
-                    + "WHERE COLUMNS.OWNER = ? "
-                    + "AND COLUMNS.TABLE_NAME = ? "
+                    + "WHERE COLUMNS.OWNER = '%s' "
+                    + "AND COLUMNS.TABLE_NAME = '%s' "
                     + "ORDER BY COLUMNS.COLUMN_ID ASC";
 
     public DamengCatalog(
@@ -79,17 +84,17 @@ protected String getListDatabaseSql() {
 
     @Override
     protected String getCreateTableSql(TablePath tablePath, CatalogTable table) {
-        throw new UnsupportedOperationException();
+        return new DamengCreateTableSqlBuilder(table).build(tablePath);
     }
 
     @Override
     protected String getDropTableSql(TablePath tablePath) {
-        return String.format("DROP TABLE %s", getTableName(tablePath));
+        return String.format("DROP TABLE %s", tablePath.getSchemaAndTableName("\""));
     }
 
     @Override
     protected String getTableName(TablePath tablePath) {
-        return tablePath.getSchemaAndTableName().toUpperCase();
+        return tablePath.getSchemaAndTableName();
     }
 
     @Override
@@ -115,6 +120,7 @@ protected String getSelectColumnsSql(TablePath tablePath) {
     protected Column buildColumn(ResultSet resultSet) throws SQLException {
         String columnName = resultSet.getString("COLUMN_NAME");
         String typeName = resultSet.getString("DATA_TYPE");
+        String sourceTypeName = resultSet.getString("SOURCE_TYPE");
         long columnLength = resultSet.getLong("DATA_LENGTH");
         long columnPrecision = resultSet.getLong("DATA_PRECISION");
         long columnScale = resultSet.getLong("DATA_SCALE");
@@ -123,20 +129,52 @@ protected Column buildColumn(ResultSet resultSet) throws SQLException {
         boolean isNullable = resultSet.getString("NULLABLE").equals("Y");
 
         SeaTunnelDataType<?> type = fromJdbcType(typeName, columnPrecision, columnScale);
-
+        long bitLen = 0;
+        long longColumnLength = 0;
+        switch (typeName) {
+            case DamengDataTypeConvertor.DM_BIT:
+                bitLen = columnLength;
+                break;
+            case DamengDataTypeConvertor.DM_DECIMAL:
+            case DamengDataTypeConvertor.DM_TIMESTAMP:
+            case DamengDataTypeConvertor.DM_DATETIME:
+            case DamengDataTypeConvertor.DM_TIME:
+                columnLength = columnScale;
+                break;
+            case DamengDataTypeConvertor.DM_CHAR:
+            case DamengDataTypeConvertor.DM_CHARACTER:
+            case DamengDataTypeConvertor.DM_VARCHAR:
+            case DamengDataTypeConvertor.DM_VARCHAR2:
+            case DamengDataTypeConvertor.DM_LONGVARCHAR:
+            case DamengDataTypeConvertor.DM_CLOB:
+            case DamengDataTypeConvertor.DM_TEXT:
+            case DamengDataTypeConvertor.DM_LONG:
+                longColumnLength = columnLength;
+                break;
+            case DamengDataTypeConvertor.DM_BINARY:
+            case DamengDataTypeConvertor.DM_VARBINARY:
+            case DamengDataTypeConvertor.DM_BLOB:
+            case DamengDataTypeConvertor.DM_BFILE:
+            case DamengDataTypeConvertor.DM_IMAGE:
+            case DamengDataTypeConvertor.DM_LONGVARBINARY:
+                bitLen = columnLength * 8;
+                break;
+            default:
+                break;
+        }
         return PhysicalColumn.of(
                 columnName,
                 type,
-                0,
+                ((int) columnLength),
                 isNullable,
                 defaultValue,
                 columnComment,
-                typeName,
+                sourceTypeName,
                 false,
                 false,
-                0L,
+                bitLen,
                 null,
-                columnLength);
+                longColumnLength);
     }
 
     private SeaTunnelDataType<?> fromJdbcType(String typeName, long precision, long scale) {