[Feature][Catalog] Add JDBC Catalog auto create table

.github/workflows/backend.yml

@@ -736,6 +736,30 @@ jobs:
         env:
           MAVEN_OPTS: -Xmx4096m
 
+  jdbc-connectors-it-part-4:
+    needs: [ changes, sanity-check ]
+    if: needs.changes.outputs.api == 'true'
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        java: [ '8', '11' ]
+        os: [ 'ubuntu-latest' ]
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up JDK ${{ matrix.java }}
+        uses: actions/setup-java@v3
+        with:
+          java-version: ${{ matrix.java }}
+          distribution: 'temurin'
+          cache: 'maven'
+      - name: run jdbc connectors integration test (part-4)
+        if: needs.changes.outputs.api == 'true'
+        run: |
+          ./mvnw -B -T 1C verify -DskipUT=true -DskipIT=false -D"license.skipAddThirdParty"=true --no-snapshot-updates -pl :connector-jdbc-e2e-part-4 -am -Pci
+        env:
+          MAVEN_OPTS: -Xmx4096m
+
   kafka-connector-it:
     needs: [ changes, sanity-check ]
     if: needs.changes.outputs.api == 'true'

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

@@ -38,6 +38,8 @@ public final class CatalogTable implements Serializable {
 
     private final String comment;
 
+    private final String catalogName;
+
     public static CatalogTable of(
             TableIdentifier tableId,
             TableSchema tableSchema,
@@ -47,17 +49,38 @@ public static CatalogTable of(
         return new CatalogTable(tableId, tableSchema, options, partitionKeys, comment);
     }
 
+    public static CatalogTable of(
+            TableIdentifier tableId,
+            TableSchema tableSchema,
+            Map<String, String> options,
+            List<String> partitionKeys,
+            String comment,
+            String catalogName) {
+        return new CatalogTable(tableId, tableSchema, options, partitionKeys, comment, catalogName);
+    }
+
     private CatalogTable(
             TableIdentifier tableId,
             TableSchema tableSchema,
             Map<String, String> options,
             List<String> partitionKeys,
             String comment) {
+        this(tableId, tableSchema, options, partitionKeys, comment, "");
+    }
+
+    private CatalogTable(
+            TableIdentifier tableId,
+            TableSchema tableSchema,
+            Map<String, String> options,
+            List<String> partitionKeys,
+            String comment,
+            String catalogName) {
         this.tableId = tableId;
         this.tableSchema = tableSchema;
         this.options = options;
         this.partitionKeys = partitionKeys;
         this.comment = comment;
+        this.catalogName = catalogName;
     }
 
     public TableIdentifier getTableId() {
@@ -80,6 +103,10 @@ public String getComment() {
         return comment;
     }
 
+    public String getCatalogName() {
+        return catalogName;
+    }
+
     @Override
     public String toString() {
         return "CatalogTable{"

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

@@ -1,13 +1,12 @@
 /*
- * 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
+ * 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
+ *    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,
@@ -23,6 +22,7 @@
 import lombok.Data;
 
 import java.io.Serializable;
+import java.util.Map;
 
 /**
  * Represent the column of {@link TableSchema}.
@@ -54,19 +54,71 @@ public abstract class Column implements Serializable {
 
     protected final String comment;
 
+    /** Field type in the database * */
+    protected final String sourceType;
+
+    /** Unsigned bit * */
+    protected final boolean isUnsigned;
+
+    /** Whether to use the 0 bit * */
+    protected final boolean isZeroFill;
+
+    /** Bit length * */
+    protected final Long bitLen;
+
+    /** integer may be cross the border * */
+    protected final Long longColumnLength;
+
+    /** your options * */
+    protected final Map<String, Object> options;
+
     protected Column(
             String name,
             SeaTunnelDataType<?> dataType,
             Integer columnLength,
             boolean nullable,
             Object defaultValue,
             String comment) {
+        this(
+                name,
+                dataType,
+                columnLength,
+                nullable,
+                defaultValue,
+                comment,
+                null,
+                false,
+                false,
+                null,
+                0L,
+                null);
+    }
+
+    protected Column(
+            String name,
+            SeaTunnelDataType<?> dataType,
+            Integer columnLength,
+            boolean nullable,
+            Object defaultValue,
+            String comment,
+            String sourceType,
+            boolean isUnsigned,
+            boolean isZeroFill,
+            Long bitLen,
+            Long longColumnLength,
+            Map<String, Object> options) {
         this.name = name;
         this.dataType = dataType;
         this.columnLength = columnLength;
         this.nullable = nullable;
         this.defaultValue = defaultValue;
         this.comment = comment;
+        this.sourceType = sourceType;
+        this.isUnsigned = isUnsigned;
+        this.isZeroFill = isZeroFill;
+        this.bitLen = bitLen;
+        this.longColumnLength = longColumnLength;
+        this.options = options;
     }
 
     /**

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

@@ -1,13 +1,12 @@
 /*
- * 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
+ * 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
+ *    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,
@@ -23,6 +22,8 @@
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
 
+import java.util.Map;
+
 /** Representation of a physical column. */
 @EqualsAndHashCode(callSuper = true)
 @ToString(callSuper = true)
@@ -38,6 +39,34 @@ protected PhysicalColumn(
         super(name, dataType, columnLength, nullable, defaultValue, comment);
     }
 
+    protected PhysicalColumn(
+            String name,
+            SeaTunnelDataType<?> dataType,
+            Integer columnLength,
+            boolean nullable,
+            Object defaultValue,
+            String comment,
+            String sourceType,
+            boolean isUnsigned,
+            boolean isZeroFill,
+            Long bitLen,
+            Long longColumnLength,
+            Map<String, Object> options) {
+        super(
+                name,
+                dataType,
+                columnLength,
+                nullable,
+                defaultValue,
+                comment,
+                sourceType,
+                isUnsigned,
+                isZeroFill,
+                bitLen,
+                longColumnLength,
+                options);
+    }
+
     public static PhysicalColumn of(
             String name,
             SeaTunnelDataType<?> dataType,
@@ -48,18 +77,70 @@ public static PhysicalColumn of(
         return new PhysicalColumn(name, dataType, columnLength, nullable, defaultValue, comment);
     }
 
+    public static PhysicalColumn of(
+            String name,
+            SeaTunnelDataType<?> dataType,
+            Integer columnLength,
+            boolean nullable,
+            Object defaultValue,
+            String comment,
+            String sourceType,
+            boolean isUnsigned,
+            boolean isZeroFill,
+            Long bitLen,
+            Map<String, Object> options,
+            Long longColumnLength) {
+        return new PhysicalColumn(
+                name,
+                dataType,
+                columnLength,
+                nullable,
+                defaultValue,
+                comment,
+                sourceType,
+                isUnsigned,
+                isZeroFill,
+                bitLen,
+                longColumnLength,
+                options);
+    }
+
     @Override
     public boolean isPhysical() {
         return true;
     }
 
     @Override
     public Column copy(SeaTunnelDataType<?> newType) {
-        return PhysicalColumn.of(name, newType, columnLength, nullable, defaultValue, comment);
+        return PhysicalColumn.of(
+                name,
+                newType,
+                columnLength,
+                nullable,
+                defaultValue,
+                comment,
+                sourceType,
+                isUnsigned,
+                isZeroFill,
+                bitLen,
+                options,
+                longColumnLength);
     }
 
     @Override
     public Column copy() {
-        return PhysicalColumn.of(name, dataType, columnLength, nullable, defaultValue, comment);
+        return PhysicalColumn.of(
+                name,
+                dataType,
+                columnLength,
+                nullable,
+                defaultValue,
+                comment,
+                sourceType,
+                isUnsigned,
+                isZeroFill,
+                bitLen,
+                options,
+                longColumnLength);
     }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/config/MongodbSourceOptions.java

@@ -27,8 +27,6 @@
 import org.bson.json.JsonMode;
 import org.bson.json.JsonWriterSettings;
 
-import java.util.Arrays;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -234,20 +232,39 @@ public class MongodbSourceOptions extends SourceOptions {
                     .withDescription(
                             "Decides if the table options contains Debezium client properties that start with prefix 'debezium'.");
 
+    //    public static final Option<StartupMode> STARTUP_MODE =
+    //            Options.key(SourceOptions.STARTUP_MODE_KEY)
+    //                    .singleChoice(
+    //                            StartupMode.class,
+    //                            Arrays.asList(
+    //                                    StartupMode.INITIAL, StartupMode.EARLIEST,
+    // StartupMode.LATEST))
+    //                    .defaultValue(StartupMode.INITIAL)
+    //                    .withDescription(
+    //                            "Optional startup mode for CDC source, valid enumerations are "
+    //                                    + "\"initial\", \"earliest\", \"latest\", \"timestamp\"\n
+    // or \"specific\"");
+
     public static final Option<StartupMode> STARTUP_MODE =
             Options.key(SourceOptions.STARTUP_MODE_KEY)
-                    .singleChoice(
-                            StartupMode.class,
-                            Arrays.asList(
-                                    StartupMode.INITIAL, StartupMode.EARLIEST, StartupMode.LATEST))
+                    .enumType(StartupMode.class)
                     .defaultValue(StartupMode.INITIAL)
                     .withDescription(
                             "Optional startup mode for CDC source, valid enumerations are "
                                     + "\"initial\", \"earliest\", \"latest\", \"timestamp\"\n or \"specific\"");
 
+    //    public static final Option<StopMode> STOP_MODE =
+    //            Options.key(SourceOptions.STOP_MODE_KEY)
+    //                    .singleChoice(StopMode.class, Collections.singletonList(StopMode.NEVER))
+    //                    .defaultValue(StopMode.NEVER)
+    //                    .withDescription(
+    //                            "Optional stop mode for CDC source, valid enumerations are "
+    //                                    + "\"never\", \"latest\", \"timestamp\"\n or
+    // \"specific\"");
+
     public static final Option<StopMode> STOP_MODE =
             Options.key(SourceOptions.STOP_MODE_KEY)
-                    .singleChoice(StopMode.class, Collections.singletonList(StopMode.NEVER))
+                    .enumType(StopMode.class)
                     .defaultValue(StopMode.NEVER)
                     .withDescription(
                             "Optional stop mode for CDC source, valid enumerations are "