CC-265: JDBC Sink: make the Oracle integer type mappings more precise

Author: shikhar

src/main/java/io/confluent/connect/jdbc/sink/dialect/OracleDialect.java

@@ -36,13 +36,13 @@ public OracleDialect() {
 
   private static Map<Schema.Type, String> getSqlTypeMap() {
     Map<Schema.Type, String> map = new HashMap<>();
-    map.put(Schema.Type.INT8, "NUMBER");
-    map.put(Schema.Type.INT16, "NUMBER");
-    map.put(Schema.Type.INT32, "NUMBER");
-    map.put(Schema.Type.INT64, "NUMBER");
+    map.put(Schema.Type.INT8, "NUMBER(3,0)");
+    map.put(Schema.Type.INT16, "NUMBER(5,0)");
+    map.put(Schema.Type.INT32, "NUMBER(10,0)");
+    map.put(Schema.Type.INT64, "NUMBER(19,0)");
     map.put(Schema.Type.FLOAT32, "BINARY_FLOAT");
     map.put(Schema.Type.FLOAT64, "BINARY_DOUBLE");
-    map.put(Schema.Type.BOOLEAN, "NUMBER");
+    map.put(Schema.Type.BOOLEAN, "NUMBER(1,0)");
     map.put(Schema.Type.STRING, "NVARCHAR2(4000)");
     map.put(Schema.Type.BYTES, "BLOB");
     return map;

src/test/java/io/confluent/connect/jdbc/sink/dialect/OracleDialectTest.java

@@ -39,8 +39,8 @@ public void handleCreateTableMultiplePKColumns() {
     ));
 
     String expected = "CREATE TABLE \"tableA\" (" + System.lineSeparator() +
-                      "\"userid\" NUMBER NOT NULL," + System.lineSeparator() +
-                      "\"userdataid\" NUMBER NOT NULL," + System.lineSeparator() +
+                      "\"userid\" NUMBER(10,0) NOT NULL," + System.lineSeparator() +
+                      "\"userdataid\" NUMBER(10,0) NOT NULL," + System.lineSeparator() +
                       "\"info\" NVARCHAR2(4000) NULL," + System.lineSeparator() +
                       "PRIMARY KEY(\"userid\",\"userdataid\"))";
     assertEquals(expected, actual);
@@ -60,14 +60,14 @@ public void handleCreateTableOnePKColumn() {
     ));
 
     String expected = "CREATE TABLE \"tableA\" (" + System.lineSeparator() +
-                      "\"col1\" NUMBER NOT NULL," + System.lineSeparator() +
-                      "\"col2\" NUMBER NULL," + System.lineSeparator() +
+                      "\"col1\" NUMBER(10,0) NOT NULL," + System.lineSeparator() +
+                      "\"col2\" NUMBER(19,0) NULL," + System.lineSeparator() +
                       "\"col3\" NVARCHAR2(4000) NULL," + System.lineSeparator() +
                       "\"col4\" BINARY_FLOAT NULL," + System.lineSeparator() +
                       "\"col5\" BINARY_DOUBLE NULL," + System.lineSeparator() +
-                      "\"col6\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col7\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col8\" NUMBER NULL," + System.lineSeparator() +
+                      "\"col6\" NUMBER(1,0) NULL," + System.lineSeparator() +
+                      "\"col7\" NUMBER(3,0) NULL," + System.lineSeparator() +
+                      "\"col8\" NUMBER(5,0) NULL," + System.lineSeparator() +
                       "PRIMARY KEY(\"col1\"))";
     assertEquals(expected, actual);
   }
@@ -86,14 +86,14 @@ public void handleCreateTableNoPKColumn() {
     ));
 
     String expected = "CREATE TABLE \"tableA\" (" + System.lineSeparator() +
-                      "\"col1\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col2\" NUMBER NULL," + System.lineSeparator() +
+                      "\"col1\" NUMBER(10,0) NULL," + System.lineSeparator() +
+                      "\"col2\" NUMBER(19,0) NULL," + System.lineSeparator() +
                       "\"col3\" NVARCHAR2(4000) NULL," + System.lineSeparator() +
                       "\"col4\" BINARY_FLOAT NULL," + System.lineSeparator() +
                       "\"col5\" BINARY_DOUBLE NULL," + System.lineSeparator() +
-                      "\"col6\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col7\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col8\" NUMBER NULL)";
+                      "\"col6\" NUMBER(1,0) NULL," + System.lineSeparator() +
+                      "\"col7\" NUMBER(3,0) NULL," + System.lineSeparator() +
+                      "\"col8\" NUMBER(5,0) NULL)";
     assertEquals(expected, actual);
   }
 
@@ -113,14 +113,14 @@ public void handleAmendAddColumns() {
     assertEquals(1, actual.size());
 
     String expected = "ALTER TABLE \"tableA\" ADD(" + System.lineSeparator() +
-                      "\"col1\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col2\" NUMBER NULL," + System.lineSeparator() +
+                      "\"col1\" NUMBER(10,0) NULL," + System.lineSeparator() +
+                      "\"col2\" NUMBER(19,0) NULL," + System.lineSeparator() +
                       "\"col3\" NVARCHAR2(4000) NULL," + System.lineSeparator() +
                       "\"col4\" BINARY_FLOAT NULL," + System.lineSeparator() +
                       "\"col5\" BINARY_DOUBLE NULL," + System.lineSeparator() +
-                      "\"col6\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col7\" NUMBER NULL," + System.lineSeparator() +
-                      "\"col8\" NUMBER NULL)";
+                      "\"col6\" NUMBER(1,0) NULL," + System.lineSeparator() +
+                      "\"col7\" NUMBER(3,0) NULL," + System.lineSeparator() +
+                      "\"col8\" NUMBER(5,0) NULL)";
     assertEquals(expected, actual.get(0));
   }