CDAP-15551 DB2 db plugin enhancements: all data types support + proper test coverage - map DECFLOAT to string.

vhlinskyi · vhlinskyi · commit a72dd1c46acc · 2019-07-12T12:37:01.000+03:00
diff --git a/db2-plugin/docs/Db2-batchsource.md b/db2-plugin/docs/Db2-batchsource.md
@@ -78,3 +78,29 @@ non-nullable varchars, output records will have this schema:
 | name           | string              |
 | email          | string              |
 | phone          | string              |
+
+Data Types Mapping
+------
+| DB2 Data Type                | CDAP Schema Data Type | Comment |
+|------------------------------|-----------------------|---------|
+| SMALLINT                     | int                   |         |
+| INTEGER                      | int                   |         |
+| BIGINT                       | long                  |         |
+| DECIMAL(p,s) or NUMERIC(p,s) | decimal               |         |
+| DECFLOAT                     | string                |         |
+| REAL                         | float                 |         |
+| DOUBLE                       | double                |         |
+| CHAR                         | string                |         |
+| VARCHAR                      | string                |         |
+| CHAR(n) FOR BIT DATA         | bytes                 |         |
+| VARCHAR(n) FOR BIT DATA      | bytes                 |         |
+| BINARY                       | bytes                 |         |
+| VARBINARY                    | bytes                 |         |
+| GRAPHIC                      | string                |         |
+| VARGRAPHIC                   | string                |         |
+| CLOB                         | string                |         |
+| BLOB                         | bytes                 |         |
+| DBCLOB                       | string                |         |
+| DATE                         | date                  |         |
+| TIME                         | time                  |         |
+| TIMESTAMP                    | timestamp             |         |
diff --git a/db2-plugin/src/main/java/io/cdap/plugin/db2/DB2Record.java b/db2-plugin/src/main/java/io/cdap/plugin/db2/DB2Record.java
@@ -65,7 +65,7 @@ private void handleSpecificType(ResultSet resultSet,
     String columnTypeName = metaData.getColumnTypeName(columnIndex);
 
     if (DB2SchemaReader.DB2_DECFLOAT.equals(columnTypeName)) {
-      recordBuilder.set(field.getName(), resultSet.getBigDecimal(columnIndex).doubleValue());
+      recordBuilder.set(field.getName(), resultSet.getString(columnIndex));
     }
   }
 }
diff --git a/db2-plugin/src/main/java/io/cdap/plugin/db2/DB2SchemaReader.java b/db2-plugin/src/main/java/io/cdap/plugin/db2/DB2SchemaReader.java
@@ -37,7 +37,7 @@ public class DB2SchemaReader extends CommonSchemaReader {
   @Override
   public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLException {
     if (DB2_TYPES.contains(metadata.getColumnType(index)) && DB2_DECFLOAT.equals(metadata.getColumnTypeName(index))) {
-      return Schema.of(Schema.Type.DOUBLE);
+      return Schema.of(Schema.Type.STRING);
     } else {
       return super.getSchema(metadata, index);
     }
diff --git a/db2-plugin/src/main/java/io/cdap/plugin/db2/Db2Sink.java b/db2-plugin/src/main/java/io/cdap/plugin/db2/Db2Sink.java
@@ -20,11 +20,18 @@
 import io.cdap.cdap.api.annotation.Name;
 import io.cdap.cdap.api.annotation.Plugin;
 import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.cdap.etl.api.batch.BatchSink;
 import io.cdap.plugin.db.DBRecord;
 import io.cdap.plugin.db.SchemaReader;
 import io.cdap.plugin.db.batch.config.DBSpecificSinkConfig;
 import io.cdap.plugin.db.batch.sink.AbstractDBSink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.Objects;
 
 
 /**
@@ -34,6 +41,7 @@
 @Name(Db2Constants.PLUGIN_NAME)
 @Description("Writes records to a DB2 table. Each record will be written in a row in the table.")
 public class Db2Sink extends AbstractDBSink {
+  private static final Logger LOG = LoggerFactory.getLogger(Db2Sink.class);
 
   private final Db2SinkConfig db2SinkConfig;
 
@@ -61,4 +69,24 @@ protected DBRecord getDBRecord(StructuredRecord.Builder output) {
   protected SchemaReader getSchemaReader() {
     return new DB2SchemaReader();
   }
+
+  @Override
+  protected boolean isFieldCompatible(Schema.Field field, ResultSetMetaData metadata, int index) throws SQLException {
+    Schema.Type fieldType = field.getSchema().isNullable() ? field.getSchema().getNonNullable().getType()
+      : field.getSchema().getType();
+
+    //DECFLOAT is mapped to string
+    String colTypeName = metadata.getColumnTypeName(index);
+    if (DB2SchemaReader.DB2_DECFLOAT.equals(colTypeName)) {
+      if (Objects.equals(fieldType, Schema.Type.STRING)) {
+        return true;
+      } else {
+        LOG.error("Field '{}' was given as type '{}' but must be of type 'string' for the DB2 column of " +
+                    "DECFLOAT type.", field.getName(), fieldType);
+        return false;
+      }
+    }
+
+    return super.isFieldCompatible(field, metadata, index);
+  }
 }
diff --git a/db2-plugin/src/test/java/io/cdap/plugin/db2/Db2SinkTestRun.java b/db2-plugin/src/test/java/io/cdap/plugin/db2/Db2SinkTestRun.java
@@ -53,29 +53,29 @@
  */
 public class Db2SinkTestRun extends Db2PluginTestBase {
   private static final Schema SCHEMA = Schema.recordOf(
-    "dbRecord",
-    Schema.Field.of("SMALLINT_COL", Schema.of(Schema.Type.INT)),
-    Schema.Field.of("INTEGER_COL", Schema.of(Schema.Type.INT)),
-    Schema.Field.of("BIGINT_COL", Schema.of(Schema.Type.LONG)),
-    Schema.Field.of("DECIMAL_COL", Schema.decimalOf(10, 6)),
-    Schema.Field.of("NUMERIC_COL", Schema.decimalOf(10, 6)),
-    Schema.Field.of("DECFLOAT_COL", Schema.of(Schema.Type.DOUBLE)),
-    Schema.Field.of("REAL_COL", Schema.of(Schema.Type.FLOAT)),
-    Schema.Field.of("DOUBLE_COL", Schema.of(Schema.Type.DOUBLE)),
-    Schema.Field.of("CHAR_COL", Schema.of(Schema.Type.STRING)),
-    Schema.Field.of("VARCHAR_COL", Schema.of(Schema.Type.STRING)),
-    Schema.Field.of("CHAR_BIT_COL", Schema.of(Schema.Type.BYTES)),
-    Schema.Field.of("VARCHAR_BIT_COL", Schema.of(Schema.Type.BYTES)),
-    Schema.Field.of("GRAPHIC_COL", Schema.of(Schema.Type.STRING)),
-    Schema.Field.of("CLOB_COL", Schema.of(Schema.Type.STRING)),
-    Schema.Field.of("BLOB_COL", Schema.of(Schema.Type.BYTES)),
-    Schema.Field.of("DATE_COL", Schema.of(Schema.LogicalType.DATE)),
-    Schema.Field.of("TIME_COL", Schema.of(Schema.LogicalType.TIME_MICROS)),
-    Schema.Field.of("TIMESTAMP_COL", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)),
-    Schema.Field.of("BINARY_COL", Schema.of(Schema.Type.BYTES)),
-    Schema.Field.of("VARBINARY_COL", Schema.of(Schema.Type.BYTES)),
-    Schema.Field.of("VARGRAPHIC_COL", Schema.of(Schema.Type.STRING)),
-    Schema.Field.of("DBCLOB_COL", Schema.of(Schema.Type.STRING))
+      "dbRecord",
+      Schema.Field.of("SMALLINT_COL", Schema.of(Schema.Type.INT)),
+      Schema.Field.of("INTEGER_COL", Schema.of(Schema.Type.INT)),
+      Schema.Field.of("BIGINT_COL", Schema.of(Schema.Type.LONG)),
+      Schema.Field.of("DECIMAL_COL", Schema.decimalOf(10, 6)),
+      Schema.Field.of("NUMERIC_COL", Schema.decimalOf(10, 6)),
+      Schema.Field.of("DECFLOAT_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("REAL_COL", Schema.of(Schema.Type.FLOAT)),
+      Schema.Field.of("DOUBLE_COL", Schema.of(Schema.Type.DOUBLE)),
+      Schema.Field.of("CHAR_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("VARCHAR_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("CHAR_BIT_COL", Schema.of(Schema.Type.BYTES)),
+      Schema.Field.of("VARCHAR_BIT_COL", Schema.of(Schema.Type.BYTES)),
+      Schema.Field.of("GRAPHIC_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("CLOB_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("BLOB_COL", Schema.of(Schema.Type.BYTES)),
+      Schema.Field.of("DATE_COL", Schema.of(Schema.LogicalType.DATE)),
+      Schema.Field.of("TIME_COL", Schema.of(Schema.LogicalType.TIME_MICROS)),
+      Schema.Field.of("TIMESTAMP_COL", Schema.of(Schema.LogicalType.TIMESTAMP_MICROS)),
+      Schema.Field.of("BINARY_COL", Schema.of(Schema.Type.BYTES)),
+      Schema.Field.of("VARBINARY_COL", Schema.of(Schema.Type.BYTES)),
+      Schema.Field.of("VARGRAPHIC_COL", Schema.of(Schema.Type.STRING)),
+      Schema.Field.of("DBCLOB_COL", Schema.of(Schema.Type.STRING))
   );
 
   @Before
@@ -107,6 +107,16 @@ public void testDBSinkWithDBSchemaAndInvalidData() throws Exception {
     }
   }
 
+  @Test
+  public void testDBSinkWithExplicitInputSchema() throws Exception {
+    testDBSink("testDBSinkWithExplicitInputSchema", "input-dbsinktest-explicit", true);
+  }
+
+  @Test
+  public void testDBSinkWithInferredInputSchema() throws Exception {
+    testDBSink("testDBSinkWithInferredInputSchema", "input-dbsinktest-inferred", false);
+  }
+
   public void testDBSink(String appName, String inputDatasetName, boolean setInputSchema) throws Exception {
     ETLPlugin sourceConfig = (setInputSchema)
       ? MockSource.getPlugin(inputDatasetName, SCHEMA)
@@ -199,7 +209,7 @@ private void createInputData(String inputDatasetName) throws Exception {
                          .set("BIGINT_COL", (long) i)
                          .setDecimal("NUMERIC_COL", new BigDecimal(3.458d, new MathContext(PRECISION)).setScale(SCALE))
                          .setDecimal("DECIMAL_COL", new BigDecimal(3.459d, new MathContext(PRECISION)).setScale(SCALE))
-                         .set("DECFLOAT_COL", .42 + i)
+                         .set("DECFLOAT_COL", Double.toString(.42 + i))
                          .set("REAL_COL", 24.123f + i)
                          .set("DOUBLE_COL", 3.456)
                          .set("CHAR_COL", name)
diff --git a/db2-plugin/src/test/java/io/cdap/plugin/db2/Db2SourceTestRun.java b/db2-plugin/src/test/java/io/cdap/plugin/db2/Db2SourceTestRun.java
@@ -86,8 +86,8 @@ public void testDBMacroSupport() throws Exception {
   public void testDBSource() throws Exception {
     String importQuery = "SELECT SMALLINT_COL, INTEGER_COL, BIGINT_COL, DECIMAL_COL, NUMERIC_COL, " +
       " REAL_COL, DOUBLE_COL, CHAR_COL, DECFLOAT_COL, VARCHAR_COL, CHAR_BIT_COL, VARCHAR_BIT_COL, GRAPHIC_COL, " +
-      " CLOB_COL, BLOB_COL, DATE_COL, TIME_COL, TIMESTAMP_COL, BINARY_COL, VARBINARY_COL, VARGRAPHIC_COL, DBCLOB_COL " +
-      "FROM my_table WHERE SMALLINT_COL < 3 AND $CONDITIONS";
+      " CLOB_COL, BLOB_COL, DATE_COL, TIME_COL, TIMESTAMP_COL, BINARY_COL, VARBINARY_COL, VARGRAPHIC_COL, DBCLOB_COL" +
+      " FROM my_table WHERE SMALLINT_COL < 3 AND $CONDITIONS";
 
 
     String boundingQuery = "SELECT MIN(SMALLINT_COL),MAX(SMALLINT_COL) from my_table";
@@ -135,8 +135,8 @@ public void testDBSource() throws Exception {
                         row1.getDecimal("NUMERIC_COL"));
     Assert.assertEquals(new BigDecimal(5.14, new MathContext(PRECISION)).setScale(SCALE),
                         row2.getDecimal("NUMERIC_COL"));
-    Assert.assertEquals(4.14, row1.get("DECFLOAT_COL"), 0.00001);
-    Assert.assertEquals(5.14, row2.get("DECFLOAT_COL"), 0.00001);
+    Assert.assertTrue(((String) row1.get("DECFLOAT_COL")).startsWith("4.14"));
+    Assert.assertTrue(((String) row2.get("DECFLOAT_COL")).startsWith("5.14"));
     Assert.assertEquals(4.14f, row1.get("REAL_COL"), 0.00001f);
     Assert.assertEquals(5.14f, row2.get("REAL_COL"), 0.00001f);
     Assert.assertEquals(4.14, row1.get("DOUBLE_COL"), 0.00001);

Original file line number	Diff line number	Diff line change
`@@ -65,7 +65,7 @@ private void handleSpecificType(ResultSet resultSet,`
`65`	`65`	`String columnTypeName = metaData.getColumnTypeName(columnIndex);`
`66`	`66`
`67`	`67`	`if (DB2SchemaReader.DB2_DECFLOAT.equals(columnTypeName)) {`
`68`		`- recordBuilder.set(field.getName(), resultSet.getBigDecimal(columnIndex).doubleValue());`
	`68`	`+ recordBuilder.set(field.getName(), resultSet.getString(columnIndex));`
`69`	`69`	`}`
`70`	`70`	`}`
`71`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ public class DB2SchemaReader extends CommonSchemaReader {`
`37`	`37`	`@Override`
`38`	`38`	`public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLException {`
`39`	`39`	`if (DB2_TYPES.contains(metadata.getColumnType(index)) && DB2_DECFLOAT.equals(metadata.getColumnTypeName(index))) {`
`40`		`- return Schema.of(Schema.Type.DOUBLE);`
	`40`	`+ return Schema.of(Schema.Type.STRING);`
`41`	`41`	`} else {`
`42`	`42`	`return super.getSchema(metadata, index);`
`43`	`43`	`}`