Merge pull request #457 from data-integrations/fix/numeric_precision_fix_1.10

[PLUGIN-1680] numeric_precision_fix_1.10

Author: Shubhangi-cs

database-commons/src/main/java/io/cdap/plugin/db/DBRecord.java

@@ -219,7 +219,7 @@ public void write(PreparedStatement stmt) throws SQLException {
     }
   }
 
-  private Schema getNonNullableSchema(Schema.Field field) {
+  protected Schema getNonNullableSchema(Schema.Field field) {
     Schema schema = field.getSchema();
     if (field.getSchema().isNullable()) {
       schema = field.getSchema().getNonNullable();

postgresql-plugin/src/main/java/io/cdap/plugin/postgres/PostgresDBRecord.java

@@ -106,6 +106,7 @@ private Object createPGobject(String type, String value, ClassLoader classLoader
   protected void writeToDB(PreparedStatement stmt, Schema.Field field, int fieldIndex) throws SQLException {
     int sqlIndex = fieldIndex + 1;
     ColumnType columnType = columnTypes.get(fieldIndex);
+    Schema fieldSchema = getNonNullableSchema(field);
     if (PostgresSchemaReader.STRING_MAPPED_POSTGRES_TYPES_NAMES.contains(columnType.getTypeName()) ||
       PostgresSchemaReader.STRING_MAPPED_POSTGRES_TYPES.contains(columnType.getType())) {
       stmt.setObject(sqlIndex, createPGobject(columnType.getTypeName(),
@@ -114,7 +115,7 @@ protected void writeToDB(PreparedStatement stmt, Schema.Field field, int fieldIn
       return;
     }
     if (columnType.getType() == Types.NUMERIC && record.get(field.getName()) != null &&
-      field.getSchema().getType() == Schema.Type.STRING) {
+      fieldSchema.getType() == Schema.Type.STRING) {
       stmt.setBigDecimal(sqlIndex, new BigDecimal((String) record.get(field.getName())));
       return;
     }