[PLUGIN-1723] Fix JavaScript Deciaml Values

core-plugins/docs/JavaScript-transform.md

@@ -63,6 +63,9 @@ or else scale the ``count`` field by 1024.
 **schema:** The schema of output objects. If no schema is given, it is assumed that the output
 schema is the same as the input schema.
 
+> for `Decimal` type the value is rounded using the `RoundingMode.HALF_EVEN` method if it does not fit within 
+the schema. This ensures that the value adheres to the precision and scale defined in the schema.
+
 **lookup:** The configuration of the lookup tables to be used in your script.
 For example, if lookup table "purchases" is configured, then you will be able to perform
 operations with that lookup table in your script: ``context.getLookup('purchases').lookup('key')``

core-plugins/src/main/java/io/cdap/plugin/transform/JavaScriptTransform.java

@@ -53,6 +53,8 @@
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -271,6 +273,21 @@ private InvalidEntry<StructuredRecord> getErrorObject(Map result, StructuredReco
   private Object decode(Object object, Schema schema) {
     Schema.Type type = schema.getType();
 
+    Schema.LogicalType logicalType = schema.getLogicalType();
+    if (logicalType != null) {
+      switch (logicalType) {
+        case DECIMAL:
+          BigDecimal bigDecimal = null;
+          if (object instanceof Number) {
+            double doubleValue = ((Number) object).doubleValue();
+            bigDecimal = BigDecimal.valueOf(doubleValue).setScale(schema.getScale(), RoundingMode.HALF_EVEN);
+          }
+          if (bigDecimal != null) {
+            return bigDecimal.unscaledValue().toByteArray();
+          }
+      }
+    }
+
     switch (type) {
       case NULL:
       case BOOLEAN:

core-plugins/src/test/java/io/cdap/plugin/transform/JavaScriptTransformTest.java

@@ -34,6 +34,7 @@
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.util.Collections;
 import java.util.HashMap;
@@ -404,4 +405,34 @@ public void testComplex() throws Exception {
     Assert.assertEquals(1, mockContext.getMockMetrics().getCount("script.transform.count"));
     Assert.assertEquals(1, mockContext.getMockMetrics().getPipelineCount("transform.1.script.transform.count"));
   }
+
+  @Test
+  public void testDecimalTransform() throws Exception {
+    Schema outputSchema = Schema.recordOf("test",
+      Schema.Field.of("pie", Schema.decimalOf(3, 2)),
+      Schema.Field.of("int_pie", Schema.decimalOf(1, 0)),
+      Schema.Field.of("long_pie", Schema.decimalOf(10, 0))
+    );
+    Schema inputSchema = Schema.recordOf("test",
+      Schema.Field.of("pie", Schema.decimalOf(3, 2)),
+      Schema.Field.of("int_pie", Schema.decimalOf(1, 0)),
+      Schema.Field.of("long_pie", Schema.decimalOf(10, 0))
+    );
+    StructuredRecord input = StructuredRecord.builder(inputSchema)
+      .setDecimal("pie", new BigDecimal("3.14"))
+      .setDecimal("int_pie", new BigDecimal("3"))
+      .setDecimal("long_pie", new BigDecimal("3147483647")
+    ).build();
+    JavaScriptTransform.Config config = new JavaScriptTransform.Config(
+      "function transform(input, emitter, context) { emitter.emit(input); }", outputSchema.toString(), null);
+    Transform<StructuredRecord, StructuredRecord> transform = new JavaScriptTransform(config);
+    transform.initialize(new MockTransformContext());
+    MockEmitter<StructuredRecord> emitter = new MockEmitter<>();
+    transform.transform(input, emitter);
+    StructuredRecord output = emitter.getEmitted().get(0);
+    Assert.assertEquals(outputSchema, output.getSchema());
+    Assert.assertEquals(input.getDecimal("pie"), output.getDecimal("pie"));
+    Assert.assertEquals(input.getDecimal("int_pie"), output.getDecimal("int_pie"));
+    Assert.assertEquals(input.getDecimal("long_pie"), output.getDecimal("long_pie"));
+  }
 }