[Improve][Catalog] Use default tablepath when can not get the tablepath from source config (#6276)

* Use default tablepath when can not get the tablepath from source config

* revert SqlServerCDCIT run in flink engine

Author: EricJoy2048

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

@@ -215,7 +215,7 @@ public static CatalogTable buildWithConfig(String catalogName, ReadonlyConfig re
         } else {
             Optional<String> resultTableNameOptional =
                     readonlyConfig.getOptional(CommonOptions.RESULT_TABLE_NAME);
-            tablePath = resultTableNameOptional.map(TablePath::of).orElse(TablePath.EMPTY);
+            tablePath = resultTableNameOptional.map(TablePath::of).orElse(TablePath.DEFAULT);
         }
 
         return CatalogTable.of(

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

@@ -34,7 +34,7 @@ public final class TablePath implements Serializable {
     private final String schemaName;
     private final String tableName;
 
-    public static final TablePath EMPTY = TablePath.of(null, null, null);
+    public static final TablePath DEFAULT = TablePath.of("default", "default", "default");
 
     public static TablePath of(String fullName) {
         return of(fullName, false);

seatunnel-api/src/test/java/org/apache/seatunnel/api/table/catalog/CatalogTableUtilTest.java

@@ -138,6 +138,21 @@ public void testCatalogUtilGetCatalogTable() throws FileNotFoundException, URISy
                                 Thread.currentThread().getContextClassLoader()));
     }
 
+    @Test
+    public void testDefaultTablePath() throws FileNotFoundException, URISyntaxException {
+        String path = getTestConfigFile("/conf/default_tablepath.conf");
+        Config config = ConfigFactory.parseFile(new File(path));
+        Config source = config.getConfigList("source").get(0);
+        ReadonlyConfig sourceReadonlyConfig = ReadonlyConfig.fromConfig(source);
+        CatalogTable catalogTable = CatalogTableUtil.buildWithConfig(sourceReadonlyConfig);
+        Assertions.assertEquals(
+                TablePath.DEFAULT.getDatabaseName(), catalogTable.getTablePath().getDatabaseName());
+        Assertions.assertEquals(
+                TablePath.DEFAULT.getSchemaName(), catalogTable.getTablePath().getSchemaName());
+        Assertions.assertEquals(
+                TablePath.DEFAULT.getTableName(), catalogTable.getTablePath().getTableName());
+    }
+
     @Test
     public void testGenericRowSchemaTest() throws FileNotFoundException, URISyntaxException {
         String path = getTestConfigFile("/conf/generic_row.schema.conf");

seatunnel-api/src/test/resources/conf/default_tablepath.conf

@@ -0,0 +1,45 @@
+#
+# 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.
+#
+
+env {
+  job.mode = "BATCH"
+}
+
+source {
+  MongoDB-CDC {
+    hosts = "mongo0:27017"
+    database = ["inventory"]
+    collection = ["inventory.products"]
+    username = superuser
+    password = superpw
+    schema = {
+      fields {
+        "_id": string,
+        "name": string,
+        "description": string,
+        "weight": string
+      }
+    }
+  }
+}
+
+transform {
+}
+
+sink {
+  Console{}
+}

seatunnel-connectors-v2/connector-fake/src/test/java/org/apache/seatunnel/connectors/seatunnel/fake/source/FakeDataGeneratorTest.java

@@ -22,6 +22,7 @@
 
 import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
+import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.type.RowKind;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
@@ -84,15 +85,21 @@ public void testComplexSchemaParse(String conf)
     public void testRowDataParse(String conf) throws FileNotFoundException, URISyntaxException {
         SeaTunnelRow row1 = new SeaTunnelRow(new Object[] {1L, "A", 100});
         row1.setRowKind(RowKind.INSERT);
+        row1.setTableId(TablePath.DEFAULT.getFullName());
         SeaTunnelRow row2 = new SeaTunnelRow(new Object[] {2L, "B", 100});
         row2.setRowKind(RowKind.INSERT);
+        row2.setTableId(TablePath.DEFAULT.getFullName());
         SeaTunnelRow row3 = new SeaTunnelRow(new Object[] {3L, "C", 100});
         row3.setRowKind(RowKind.INSERT);
+        row3.setTableId(TablePath.DEFAULT.getFullName());
         SeaTunnelRow row1UpdateBefore = new SeaTunnelRow(new Object[] {1L, "A", 100});
+        row1UpdateBefore.setTableId(TablePath.DEFAULT.getFullName());
         row1UpdateBefore.setRowKind(RowKind.UPDATE_BEFORE);
         SeaTunnelRow row1UpdateAfter = new SeaTunnelRow(new Object[] {1L, "A_1", 100});
+        row1UpdateAfter.setTableId(TablePath.DEFAULT.getFullName());
         row1UpdateAfter.setRowKind(RowKind.UPDATE_AFTER);
         SeaTunnelRow row2Delete = new SeaTunnelRow(new Object[] {2L, "B", 100});
+        row2Delete.setTableId(TablePath.DEFAULT.getFullName());
         row2Delete.setRowKind(RowKind.DELETE);
         List<SeaTunnelRow> expected =
                 Arrays.asList(row1, row2, row3, row1UpdateBefore, row1UpdateAfter, row2Delete);

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-sqlserver-e2e/src/test/java/org/apache/seatunnel/e2e/connector/cdc/sqlserver/SqlServerCDCIT.java

@@ -65,7 +65,7 @@
 @Slf4j
 @DisabledOnContainer(
         value = {},
-        type = {EngineType.SPARK, EngineType.FLINK},
+        type = {EngineType.SPARK},
         disabledReason = "Currently SPARK do not support cdc")
 public class SqlServerCDCIT extends TestSuiteBase implements TestResource {