Merge pull request #312 from data-integrations/feature/PLUGIN-1433-OracleNumericScaleChanges

* Using the scale from logical schema instead of the number value.

* Added Warning logs when detecting precision less Oracle numbers.

* Added Unit Tests for precision less Number.

Author: MrRahulSharma

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceDBRecord.java

@@ -214,19 +214,23 @@ private void handleOracleSpecificType(ResultSet resultSet, StructuredRecord.Buil
         if (Double.class.getTypeName().equals(resultSet.getMetaData().getColumnClassName(columnIndex))) {
           recordBuilder.set(field.getName(), resultSet.getDouble(columnIndex));
         } else {
-          // For a Number type without specified precision and scale, precision will be 0 and scale will be -127
-          if (precision == 0) {
-            // reference : https://docs.oracle.com/cd/B28359_01/server.111/b28318/datatype.htm#CNCPT1832
-            scale = 0;
-          }
           // It's required to pass 'scale' parameter since in the case of Oracle, scale of 'BigDecimal' depends on the
-          // scale of actual value. For example for value '77.12' scale will be '2' even if sql scale is '6'
-          BigDecimal decimal = resultSet.getBigDecimal(columnIndex, scale);
+          // scale set in the logical schema. For example for value '77.12' if the scale set in the logical schema is
+          // set to 4 then the number will change to '77.1200'. Also if the value is '22.1274' and the logical schema
+          // scale is set to 2 then the decimal value used will be '22.13' after rounding.
+          BigDecimal decimal = resultSet.getBigDecimal(columnIndex, getScale(field.getSchema()));
           recordBuilder.setDecimal(field.getName(), decimal);
         }
     }
   }
 
+  /**
+   * Get the scale set in Non-nullable schema associated with the schema
+   * */
+  private int getScale(Schema schema) {
+    return schema.isNullable() ? schema.getNonNullable().getScale() : schema.getScale();
+  }
+
   private boolean isLongOrLongRaw(int columnType) {
     return columnType == OracleSourceSchemaReader.LONG || columnType == OracleSourceSchemaReader.LONG_RAW;
   }

oracle-plugin/src/main/java/io/cdap/plugin/oracle/OracleSourceSchemaReader.java

@@ -19,6 +19,8 @@
 import com.google.common.collect.ImmutableSet;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.plugin.db.CommonSchemaReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
@@ -42,6 +44,11 @@ public class OracleSourceSchemaReader extends CommonSchemaReader {
   public static final int LONG = -1;
   public static final int LONG_RAW = -4;
 
+  /**
+   * Logger instance for Oracle Schema reader.
+   */
+  private static final Logger LOG = LoggerFactory.getLogger(OracleSourceSchemaReader.class);
+
   public static final Set<Integer> ORACLE_TYPES = ImmutableSet.of(
     INTERVAL_DS,
     INTERVAL_YM,
@@ -100,6 +107,11 @@ public Schema getSchema(ResultSetMetaData metadata, int index) throws SQLExcepti
             // reference : https://docs.oracle.com/cd/B28359_01/server.111/b28318/datatype.htm#CNCPT1832
             precision = 38;
             scale = 0;
+            LOG.warn(String.format("%s type with undefined precision and scale is detected, "
+                    + "there may be a precision loss while running the pipeline. "
+                    + "Please define an output precision and scale for field '%s' to avoid precision loss.",
+                metadata.getColumnTypeName(index),
+                metadata.getColumnName(index)));
           }
           return Schema.decimalOf(precision, scale);
         }

oracle-plugin/src/test/java/io/cdap/plugin/oracle/OracleSourceDBRecordUnitTest.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright © 2019 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.oracle;
+
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.math.BigDecimal;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit Test class for the OracleSourceDBRecord
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class OracleSourceDBRecordUnitTest {
+
+  private static final int DEFAULT_PRECISION = 38;
+
+  @Mock
+  ResultSet resultSet;
+
+  @Mock
+  ResultSetMetaData resultSetMetaData;
+
+  /**
+   * Validate the precision less Numbers handling against following use cases.
+   * 1. Ensure that for Number(0,-127) non nullable type a Number(38,0) is returned by default.
+   * 2. Ensure that for Number(0,-127) non nullable type a Number(38,4) is returned,
+   *    if schema defined this as Number(38,4).
+   * 3. Ensure that for Number(0,-127) nullable type a Number(38,0) is returned by default.
+   * 4. Ensure that for Number(0,-127) nullable type a Number(38,4) is returned,
+   *    if schema defined this as Number(38,4).
+   * @throws Exception
+   */
+  @Test
+  public void validatePrecisionLessDecimalParsing() throws Exception {
+    Schema.Field field1 = Schema.Field.of("ID1", Schema.decimalOf(DEFAULT_PRECISION));
+    Schema.Field field2 = Schema.Field.of("ID2", Schema.decimalOf(DEFAULT_PRECISION, 4));
+    Schema.Field field3 = Schema.Field.of("ID3", Schema.nullableOf(Schema.decimalOf(DEFAULT_PRECISION)));
+    Schema.Field field4 = Schema.Field.of("ID4", Schema.nullableOf(Schema.decimalOf(DEFAULT_PRECISION, 4)));
+
+    Schema schema = Schema.recordOf(
+        "dbRecord",
+        field1,
+        field2,
+        field3,
+        field4
+    );
+
+    when(resultSet.getMetaData()).thenReturn(resultSetMetaData);
+    when(resultSet.getBigDecimal(eq(1), eq(0))).thenReturn(new BigDecimal("123"));
+    when(resultSet.getBigDecimal(eq(2), eq(4))).thenReturn(new BigDecimal("123.4568"));
+    when(resultSet.getBigDecimal(eq(3), eq(0))).thenReturn(new BigDecimal("123"));
+    when(resultSet.getBigDecimal(eq(4), eq(4))).thenReturn(new BigDecimal("123.4568"));
+
+    StructuredRecord.Builder builder = StructuredRecord.builder(schema);
+    OracleSourceDBRecord dbRecord = new OracleSourceDBRecord(null, null);
+    dbRecord.handleField(resultSet, builder, field1, 1, Types.NUMERIC, 0, -127);
+    dbRecord.handleField(resultSet, builder, field2, 2, Types.NUMERIC, 0, -127);
+    dbRecord.handleField(resultSet, builder, field3, 3, Types.NUMERIC, 0, -127);
+    dbRecord.handleField(resultSet, builder, field4, 4, Types.NUMERIC, 0, -127);
+
+    StructuredRecord record = builder.build();
+    Assert.assertEquals(record.getDecimal("ID1").toPlainString(), "123");
+    Assert.assertEquals(record.getDecimal("ID2").toPlainString(), "123.4568");
+    Assert.assertEquals(record.getDecimal("ID3").toPlainString(), "123");
+    Assert.assertEquals(record.getDecimal("ID4").toPlainString(), "123.4568");
+  }
+}