[Fix] Fix Oracle type converter handle negative scale in number type (apache#6758)

Author: Hisoka-X

Diff for: seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeConverter.java

@@ -117,21 +117,22 @@ public Column convert(BasicTypeDefine typeDefine) {
                     scale = 127;
                 }
 
-                if (scale == 0) {
-                    if (precision == 1) {
+                if (scale <= 0) {
+                    int newPrecision = (int) (precision - scale);
+                    if (newPrecision == 1) {
                         builder.dataType(BasicType.BOOLEAN_TYPE);
-                    } else if (precision <= 9) {
+                    } else if (newPrecision <= 9) {
                         builder.dataType(BasicType.INT_TYPE);
-                    } else if (precision <= 18) {
+                    } else if (newPrecision <= 18) {
                         builder.dataType(BasicType.LONG_TYPE);
-                    } else if (precision < 38) {
-                        builder.dataType(new DecimalType(precision.intValue(), 0));
-                        builder.columnLength(precision);
+                    } else if (newPrecision < 38) {
+                        builder.dataType(new DecimalType(newPrecision, 0));
+                        builder.columnLength((long) newPrecision);
                     } else {
                         builder.dataType(new DecimalType(DEFAULT_PRECISION, 0));
                         builder.columnLength((long) DEFAULT_PRECISION);
                     }
-                } else if (scale > 0 && scale <= DEFAULT_SCALE) {
+                } else if (scale <= DEFAULT_SCALE) {
                     builder.dataType(new DecimalType(precision.intValue(), scale));
                     builder.columnLength(precision);
                     builder.scale(scale);

Diff for: seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleTypeConverterTest.java

@@ -754,4 +754,59 @@ public void testReconvertDatetime() {
                 typeDefine.getDataType());
         Assertions.assertEquals(column.getScale(), typeDefine.getScale());
     }
+
+    @Test
+    public void testNumberWithNegativeScale() {
+        BasicTypeDefine<Object> typeDefine =
+                BasicTypeDefine.builder()
+                        .name("test")
+                        .columnType("number(38,-1)")
+                        .dataType("number")
+                        .precision(38L)
+                        .scale(-1)
+                        .build();
+        Column column = OracleTypeConverter.INSTANCE.convert(typeDefine);
+        Assertions.assertEquals(typeDefine.getName(), column.getName());
+        Assertions.assertEquals(new DecimalType(38, 0), column.getDataType());
+        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
+
+        typeDefine =
+                BasicTypeDefine.builder()
+                        .name("test")
+                        .columnType("number(5,-2)")
+                        .dataType("number")
+                        .precision(5L)
+                        .scale(-2)
+                        .build();
+        column = OracleTypeConverter.INSTANCE.convert(typeDefine);
+        Assertions.assertEquals(typeDefine.getName(), column.getName());
+        Assertions.assertEquals(BasicType.INT_TYPE, column.getDataType());
+        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
+
+        typeDefine =
+                BasicTypeDefine.builder()
+                        .name("test")
+                        .columnType("number(9,-2)")
+                        .dataType("number")
+                        .precision(9L)
+                        .scale(-2)
+                        .build();
+        column = OracleTypeConverter.INSTANCE.convert(typeDefine);
+        Assertions.assertEquals(typeDefine.getName(), column.getName());
+        Assertions.assertEquals(BasicType.LONG_TYPE, column.getDataType());
+        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
+
+        typeDefine =
+                BasicTypeDefine.builder()
+                        .name("test")
+                        .columnType("number(14,-11)")
+                        .dataType("number")
+                        .precision(14L)
+                        .scale(-11)
+                        .build();
+        column = OracleTypeConverter.INSTANCE.convert(typeDefine);
+        Assertions.assertEquals(typeDefine.getName(), column.getName());
+        Assertions.assertEquals(new DecimalType(25, 0), column.getDataType());
+        Assertions.assertEquals(typeDefine.getColumnType(), column.getSourceType());
+    }
 }

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcOracleIT.java

@@ -73,6 +73,7 @@ public class JdbcOracleIT extends AbstractJdbcIT {
                     + "    CHAR_10_COL                   char(10),\n"
                     + "    CLOB_COL                      clob,\n"
                     + "    NUMBER_3_SF_2_DP              number(3, 2),\n"
+                    + "    NUMBER_7_SF_N2_DP             number(7, -2),\n"
                     + "    INTEGER_COL                   integer,\n"
                     + "    FLOAT_COL                     float(10),\n"
                     + "    REAL_COL                      real,\n"
@@ -90,6 +91,7 @@ public class JdbcOracleIT extends AbstractJdbcIT {
                 "CHAR_10_COL",
                 "CLOB_COL",
                 "NUMBER_3_SF_2_DP",
+                "NUMBER_7_SF_N2_DP",
                 "INTEGER_COL",
                 "FLOAT_COL",
                 "REAL_COL",
@@ -170,6 +172,7 @@ Pair<String[], List<SeaTunnelRow>> initTestData() {
                                 String.format("f%s", i),
                                 String.format("f%s", i),
                                 BigDecimal.valueOf(1.1),
+                                BigDecimal.valueOf(2400),
                                 i,
                                 Float.parseFloat("2.2"),
                                 Float.parseFloat("2.2"),

Diff for: seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_oracle_source_to_sink.conf

@@ -30,7 +30,7 @@ source {
     url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
     user = testUser
     password = testPassword
-    query = "SELECT VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL FROM E2E_TABLE_SOURCE"
+    query = "SELECT VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL FROM E2E_TABLE_SOURCE"
     properties {
        database.oracle.jdbc.timezoneAsRegion = "false"
     }
@@ -46,7 +46,7 @@ sink {
     url = "jdbc:oracle:thin:@e2e_oracleDb:1521/TESTUSER"
     user = testUser
     password = testPassword
-    query = "INSERT INTO E2E_TABLE_SINK (VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)"
+    query = "INSERT INTO E2E_TABLE_SINK (VARCHAR_10_COL,CHAR_10_COL,CLOB_COL,NUMBER_3_SF_2_DP,NUMBER_7_SF_N2_DP,INTEGER_COL,FLOAT_COL,REAL_COL,BINARY_FLOAT_COL,BINARY_DOUBLE_COL,DATE_COL,TIMESTAMP_WITH_3_FRAC_SEC_COL,TIMESTAMP_WITH_LOCAL_TZ,XML_TYPE_COL) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
     properties {
        database.oracle.jdbc.timezoneAsRegion = "false"
     }