From d5ae11095f78c9306fe79552d0a8617ec1319864 Mon Sep 17 00:00:00 2001
From: psainics <pushpender.saini@cloudsufi.com>
Date: Wed, 24 Apr 2024 07:53:44 +0530
Subject: [PATCH] Fix JavaScript Deciaml Values

---
 .../plugin/transform/JavaScriptTransform.java | 18 +++++++++++++
 .../transform/JavaScriptTransformTest.java    | 27 +++++++++++++++++++
 2 files changed, 45 insertions(+)

diff --git a/core-plugins/src/main/java/io/cdap/plugin/transform/JavaScriptTransform.java b/core-plugins/src/main/java/io/cdap/plugin/transform/JavaScriptTransform.java
index e3732d019..efc6ce730 100644
--- a/core-plugins/src/main/java/io/cdap/plugin/transform/JavaScriptTransform.java
+++ b/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,22 @@ 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 Double) {
+            bigDecimal = BigDecimal.valueOf((Double) object).setScale(schema.getScale(), RoundingMode.HALF_EVEN);
+          } else if (object instanceof Integer) {
+            bigDecimal = BigDecimal.valueOf((Integer) object).setScale(schema.getScale(), RoundingMode.HALF_EVEN);
+          }
+          if (bigDecimal != null) {
+            return bigDecimal.unscaledValue().toByteArray();
+          }
+      }
+    }
+
     switch (type) {
       case NULL:
       case BOOLEAN:
diff --git a/core-plugins/src/test/java/io/cdap/plugin/transform/JavaScriptTransformTest.java b/core-plugins/src/test/java/io/cdap/plugin/transform/JavaScriptTransformTest.java
index fba04d046..c10b9a3d6 100644
--- a/core-plugins/src/test/java/io/cdap/plugin/transform/JavaScriptTransformTest.java
+++ b/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,30 @@ 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 inputSchema = Schema.recordOf("test",
+      Schema.Field.of("pie", Schema.decimalOf(3, 2)),
+      Schema.Field.of("int_pie", Schema.decimalOf(1, 0))
+    );
+    StructuredRecord input = StructuredRecord.builder(inputSchema)
+      .setDecimal("pie", new BigDecimal("3.14"))
+      .setDecimal("int_pie", new BigDecimal("3")
+    ).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"));
+  }
 }