[Bug][connector-jdbc] Fix integer overflow when merging table schema with large column length (#10387)

Co-authored-by: zengyi <zengyi@chinatelecom.cn>

Author: yzeng1618

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

@@ -332,23 +332,22 @@ static CatalogTable mergeCatalogTable(CatalogTable tableOfPath, CatalogTable tab
                                                                             .get(column.getName())
                                                                             .getDataType()
                                                                             .getSqlType())
-                                            ? PhysicalColumn.of(
+                                            ? new PhysicalColumn(
                                                     column.getName(),
                                                     column.getDataType(),
-                                                    column.getColumnLength() == null
-                                                            ? null
-                                                            : Math.toIntExact(
-                                                                    column.getColumnLength()),
+                                                    column.getColumnLength(),
+                                                    column.getScale(),
                                                     column.isNullable(),
                                                     column.getDefaultValue(),
                                                     columnsOfPath
                                                             .get(column.getName())
                                                             .getComment(),
                                                     column.getSourceType(),
+                                                    column.getSinkType(),
+                                                    column.getOptions(),
                                                     column.isUnsigned(),
                                                     column.isZeroFill(),
                                                     column.getBitLen(),
-                                                    column.getOptions(),
                                                     column.getLongColumnLength())
                                             : column;
                                 })

seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/utils/JdbcCatalogUtilsTest.java

@@ -335,6 +335,94 @@ public void testColumnNotIncludeMerge() {
                 mergeTable.getTableSchema().getColumns());
     }
 
+    @Test
+    public void testColumnNotIncludeMergeWithLargeColumnLength() {
+        long largeLength = 4294967295L;
+
+        CatalogTable tableOfPath =
+                CatalogTable.of(
+                        TableIdentifier.of("mysql-1", "database-x", null, "table-x"),
+                        TableSchema.builder()
+                                .column(
+                                        PhysicalColumn.of(
+                                                "id",
+                                                BasicType.LONG_TYPE,
+                                                (Long) null,
+                                                false,
+                                                null,
+                                                "id comment"))
+                                .column(
+                                        PhysicalColumn.of(
+                                                "config",
+                                                BasicType.STRING_TYPE,
+                                                largeLength,
+                                                false,
+                                                null,
+                                                "config comment"))
+                                .build(),
+                        Collections.emptyMap(),
+                        Collections.emptyList(),
+                        null);
+
+        CatalogTable tableOfQuery =
+                CatalogTable.of(
+                        TableIdentifier.of("default", null, null, "default"),
+                        TableSchema.builder()
+                                .column(
+                                        PhysicalColumn.of(
+                                                "id",
+                                                BasicType.LONG_TYPE,
+                                                (Long) null,
+                                                true,
+                                                null,
+                                                null))
+                                .column(
+                                        PhysicalColumn.of(
+                                                "config",
+                                                BasicType.STRING_TYPE,
+                                                largeLength,
+                                                true,
+                                                null,
+                                                null))
+                                .column(
+                                        PhysicalColumn.of(
+                                                "dummy",
+                                                BasicType.INT_TYPE,
+                                                (Long) null,
+                                                true,
+                                                null,
+                                                null))
+                                .build(),
+                        Collections.emptyMap(),
+                        Collections.emptyList(),
+                        null);
+
+        CatalogTable mergeTable = JdbcCatalogUtils.mergeCatalogTable(tableOfPath, tableOfQuery);
+
+        Assertions.assertEquals(
+                tableOfPath.getTableId().toTablePath(), mergeTable.getTableId().toTablePath());
+        Assertions.assertEquals(
+                tableOfQuery.getTableId().getCatalogName(),
+                mergeTable.getTableId().getCatalogName());
+
+        Map<String, Column> mergedColumns =
+                mergeTable.getTableSchema().getColumns().stream()
+                        .collect(Collectors.toMap(e -> e.getName(), e -> e));
+
+        Column mergedId = mergedColumns.get("id");
+        Column mergedConfig = mergedColumns.get("config");
+
+        Assertions.assertNotNull(mergedId);
+        Assertions.assertNotNull(mergedConfig);
+
+        // The merge should use the query column as base, and fill comment from the table_path.
+        Assertions.assertTrue(mergedId.isNullable());
+        Assertions.assertEquals("id comment", mergedId.getComment());
+
+        Assertions.assertEquals(Long.valueOf(largeLength), mergedConfig.getColumnLength());
+        Assertions.assertEquals("config comment", mergedConfig.getComment());
+    }
+
     @Test
     public void testDecimalColumnMerge() {
         CatalogTable tableOfQuery =