[Improve][JdbcSource] Optimize catalog-table metadata merge logic (apache#5828)

Author: hailin0

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtils.java

@@ -173,64 +173,58 @@ private static CatalogTable getCatalogTable(
         return jdbcCatalog.getTable(tableConfig.getQuery());
     }
 
-    private static CatalogTable mergeCatalogTable(
-            CatalogTable tableOfPath, CatalogTable tableOfQuery) {
-        String catalogName =
-                tableOfQuery.getTableId() == null
-                        ? DEFAULT_CATALOG_NAME
-                        : tableOfQuery.getTableId().getCatalogName();
-        TableIdentifier tableIdentifier =
-                TableIdentifier.of(
-                        catalogName,
-                        tableOfPath.getTableId().getDatabaseName(),
-                        tableOfPath.getTableId().getSchemaName(),
-                        tableOfPath.getTableId().getTableName());
-
+    static CatalogTable mergeCatalogTable(CatalogTable tableOfPath, CatalogTable tableOfQuery) {
         TableSchema tableSchemaOfPath = tableOfPath.getTableSchema();
         Map<String, Column> columnsOfPath =
                 tableSchemaOfPath.getColumns().stream()
-                        .collect(Collectors.toMap(Column::getName, Function.identity()));
-        Set<String> columnKeysOfPath = columnsOfPath.keySet();
+                        .collect(
+                                Collectors.toMap(
+                                        Column::getName,
+                                        Function.identity(),
+                                        (o1, o2) -> o1,
+                                        LinkedHashMap::new));
         TableSchema tableSchemaOfQuery = tableOfQuery.getTableSchema();
         Map<String, Column> columnsOfQuery =
                 tableSchemaOfQuery.getColumns().stream()
-                        .collect(Collectors.toMap(Column::getName, Function.identity()));
+                        .collect(
+                                Collectors.toMap(
+                                        Column::getName,
+                                        Function.identity(),
+                                        (o1, o2) -> o1,
+                                        LinkedHashMap::new));
         Set<String> columnKeysOfQuery = columnsOfQuery.keySet();
 
-        if (columnKeysOfPath.equals(columnKeysOfQuery)) {
-            boolean schemaEquals =
-                    columnKeysOfPath.stream()
-                            .allMatch(
-                                    key ->
-                                            columnsOfPath
-                                                    .get(key)
-                                                    .getDataType()
-                                                    .equals(columnsOfQuery.get(key).getDataType()));
-            if (schemaEquals) {
-                return CatalogTable.of(
-                        tableIdentifier,
-                        TableSchema.builder()
-                                .primaryKey(tableSchemaOfPath.getPrimaryKey())
-                                .constraintKey(tableSchemaOfPath.getConstraintKeys())
-                                .columns(tableSchemaOfQuery.getColumns())
-                                .build(),
-                        tableOfPath.getOptions(),
-                        tableOfPath.getPartitionKeys(),
-                        tableOfPath.getComment(),
-                        tableIdentifier.getCatalogName());
-            }
+        List<Column> columnsOfMerge =
+                tableSchemaOfQuery.getColumns().stream()
+                        .filter(
+                                column ->
+                                        columnsOfPath.containsKey(column.getName())
+                                                && columnsOfPath
+                                                        .get(column.getName())
+                                                        .getDataType()
+                                                        .equals(
+                                                                columnsOfQuery
+                                                                        .get(column.getName())
+                                                                        .getDataType()))
+                        .map(column -> columnsOfPath.get(column.getName()))
+                        .collect(Collectors.toList());
+        boolean schemaIncludeAllColumns = columnsOfMerge.size() == columnKeysOfQuery.size();
+        boolean schemaEquals =
+                schemaIncludeAllColumns && columnsOfMerge.size() == columnsOfPath.size();
+        if (schemaEquals) {
+            return tableOfPath;
         }
 
         PrimaryKey primaryKeyOfPath = tableSchemaOfPath.getPrimaryKey();
         List<ConstraintKey> constraintKeysOfPath = tableSchemaOfPath.getConstraintKeys();
         List<String> partitionKeysOfPath = tableOfPath.getPartitionKeys();
-        PrimaryKey primaryKeyOfQuery = null;
-        List<ConstraintKey> constraintKeysOfQuery = new ArrayList<>();
-        List<String> partitionKeysOfQuery = new ArrayList<>();
+        PrimaryKey primaryKeyOfMerge = null;
+        List<ConstraintKey> constraintKeysOfMerge = new ArrayList<>();
+        List<String> partitionKeysOfMerge = new ArrayList<>();
 
         if (primaryKeyOfPath != null
                 && columnKeysOfQuery.containsAll(primaryKeyOfPath.getColumnNames())) {
-            primaryKeyOfQuery = primaryKeyOfPath;
+            primaryKeyOfMerge = primaryKeyOfPath;
         }
         if (constraintKeysOfPath != null) {
             for (ConstraintKey constraintKey : constraintKeysOfPath) {
@@ -239,26 +233,47 @@ private static CatalogTable mergeCatalogTable(
                                 .map(e -> e.getColumnName())
                                 .collect(Collectors.toSet());
                 if (columnKeysOfQuery.containsAll(constraintKeyFields)) {
-                    constraintKeysOfQuery.add(constraintKey);
+                    constraintKeysOfMerge.add(constraintKey);
                 }
             }
         }
         if (partitionKeysOfPath != null && columnKeysOfQuery.containsAll(partitionKeysOfPath)) {
-            partitionKeysOfQuery = partitionKeysOfPath;
+            partitionKeysOfMerge = partitionKeysOfPath;
+        }
+        if (schemaIncludeAllColumns) {
+            return CatalogTable.of(
+                    tableOfPath.getTableId(),
+                    TableSchema.builder()
+                            .primaryKey(primaryKeyOfMerge)
+                            .constraintKey(constraintKeysOfMerge)
+                            .columns(columnsOfMerge)
+                            .build(),
+                    tableOfPath.getOptions(),
+                    partitionKeysOfMerge,
+                    tableOfPath.getComment());
         }
 
+        String catalogName =
+                tableOfQuery.getTableId() == null
+                        ? DEFAULT_CATALOG_NAME
+                        : tableOfQuery.getTableId().getCatalogName();
+        TableIdentifier tableIdentifier =
+                TableIdentifier.of(
+                        catalogName,
+                        tableOfPath.getTableId().getDatabaseName(),
+                        tableOfPath.getTableId().getSchemaName(),
+                        tableOfPath.getTableId().getTableName());
         CatalogTable mergedCatalogTable =
                 CatalogTable.of(
                         tableIdentifier,
                         TableSchema.builder()
-                                .primaryKey(primaryKeyOfQuery)
-                                .constraintKey(constraintKeysOfQuery)
+                                .primaryKey(primaryKeyOfMerge)
+                                .constraintKey(constraintKeysOfMerge)
                                 .columns(tableSchemaOfQuery.getColumns())
                                 .build(),
                         tableOfPath.getOptions(),
-                        partitionKeysOfQuery,
-                        tableOfPath.getComment(),
-                        tableIdentifier.getCatalogName());
+                        partitionKeysOfMerge,
+                        tableOfPath.getComment());
 
         log.info("Merged catalog table of path {}", tableOfPath.getTableId().toTablePath());
         return mergedCatalogTable;