Handling mysql decimal data types with precision 19 or higher (opensearch-project#6369)

Signed-off-by: Divyansh Bokadia <dbokadia@amazon.com>

Author: divbok

data-prepper-plugins/parquet-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/parquet/GenericRecordJsonEncoder.java

@@ -21,6 +21,7 @@
 import org.apache.avro.generic.GenericContainer;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericEnumSymbol;
+import org.apache.avro.generic.GenericFixed;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.commons.text.StringEscapeUtils;
@@ -84,8 +85,8 @@ private void serialize(final Object datum, final StringBuilder buffer,
                 if (fieldSchema.getType() == Schema.Type.UNION) {
                     for (Schema s : fieldSchema.getTypes()) {
                         if (s.getType() != Schema.Type.NULL) {
-                            if (s.getType() == Schema.Type.BYTES &&
-                                    s.getLogicalType() instanceof LogicalTypes.Decimal) {
+                            if ((s.getType() == Schema.Type.BYTES || s.getType() == Schema.Type.FIXED)
+                                    && s.getLogicalType() instanceof LogicalTypes.Decimal) {
                                 serialize(logicalTypeConverter.apply(getField(datum, f.name(), f.pos())), buffer, seenObjects, ((LogicalTypes.Decimal) s.getLogicalType()).getScale());
                                 serializedDecimal = true;
                                 break;
@@ -172,7 +173,22 @@ private void serialize(final Object datum, final StringBuilder buffer,
             buffer.append("\"");
             buffer.append(datum);
             buffer.append("\"");
-        } else if (datum instanceof GenericData) {
+        } else if (datum instanceof GenericFixed ) {
+            GenericFixed fixed = (GenericFixed) datum;
+            byte[] bytes = fixed.bytes();
+            if (decimalScale != null) {
+                BigInteger unscaledValue = new BigInteger(bytes);
+                BigDecimal decimal = new BigDecimal(unscaledValue, decimalScale);
+                buffer.append(decimal.toString());
+            }
+            //Fallback mechanism
+            else {
+                buffer.append("{\"bytes\": \"");
+                writeEscapedString(new String(bytes, StandardCharsets.ISO_8859_1), buffer);
+                buffer.append("\"}");
+            }
+        }
+        else if (datum instanceof GenericData) {
             if (seenObjects.containsKey(datum)) {
                 buffer.append(TOSTRING_CIRCULAR_REFERENCE_ERROR_TEXT);
                 return;

data-prepper-plugins/parquet-codecs/src/test/java/org/opensearch/dataprepper/plugins/codec/parquet/GenericRecordJsonEncoderTest.java

@@ -316,4 +316,69 @@ void serialize_WithNullDecimalLogicalType_ReturnsNull() {
         assertEquals("{\"amount\": null}", json);
     }
 
+    @Test
+    void serialize_WithFixedDecimalLogicalType_UsesScaleFromSchema() {
+        BigDecimal value = new BigDecimal("12.34").setScale(2);
+        byte[] decimalBytes = value.unscaledValue().toByteArray();
+        
+        Schema decimalSchema = new Schema.Parser().parse(
+                "{ \"type\": \"record\", \"name\": \"DecimalRecord\", \"fields\": [" +
+                        "{\"name\": \"amount\", \"type\": [\"null\", {\"type\":\"fixed\",\"size\":" + decimalBytes.length +
+                        ",\"name\":\"DecimalFixed\",\"logicalType\":\"decimal\",\"precision\":4,\"scale\":2}]}" +
+                        "] }"
+        );
+
+        GenericRecord record = new GenericData.Record(decimalSchema);
+        
+        Schema fixedSchema = decimalSchema.getField("amount").schema().getTypes().get(1);
+        GenericData.Fixed fixedValue = new GenericData.Fixed(fixedSchema, decimalBytes);
+        record.put("amount", fixedValue);
+
+        String json = encoder.serialize(record);
+
+        assertEquals("{\"amount\": 12.34}", json);
+    }
+
+    @Test
+    void serialize_WithNonNullableFixedDecimalLogicalType_UsesScaleFromSchema() {
+        BigDecimal value = new BigDecimal("12.3456").setScale(4);
+        byte[] decimalBytes = value.unscaledValue().toByteArray();
+        
+        Schema decimalSchema = new Schema.Parser().parse(
+                "{ \"type\": \"record\", \"name\": \"DecimalRecord\", \"fields\": [" +
+                        "{\"name\": \"amount\", \"type\": {\"type\":\"fixed\",\"size\":" + decimalBytes.length +
+                        ",\"name\":\"DecimalFixed\",\"logicalType\":\"decimal\",\"precision\":6,\"scale\":4}}" +
+                        "] }"
+        );
+
+        GenericRecord record = new GenericData.Record(decimalSchema);
+        
+        Schema fixedSchema = decimalSchema.getField("amount").schema();
+        GenericData.Fixed fixedValue = new GenericData.Fixed(fixedSchema, decimalBytes);
+        record.put("amount", fixedValue);
+
+        String json = encoder.serialize(record);
+
+        assertEquals("{\"amount\": 12.3456}", json);
+    }
+
+    @Test
+    void serialize_WithNonDecimalFixedType_ReturnsEscapedString() {
+        byte[] tokenBytes = new byte[] { 34, 92, 13, 10, 9};
+        Schema tokenSchema = new Schema.Parser().parse(
+                "{ \"type\": \"record\", \"name\": \"MyRecord\", \"fields\": [" +
+                        "{\"name\": \"token\", \"type\": {\"type\":\"fixed\",\"name\":\"TokenFixed\",\"size\":5}}" +
+                        "] }"
+        );
+
+        GenericRecord record = new GenericData.Record(tokenSchema);
+
+        Schema fixedSchema = tokenSchema.getField("token").schema();
+        GenericData.Fixed fixedValue = new GenericData.Fixed(fixedSchema, tokenBytes);
+        record.put("token", fixedValue);
+
+        String json = encoder.serialize(record);
+
+        assertEquals("{\"token\": {\"bytes\": \"\\\"\\\\\\r\\n\\t\"}}", json);
+    }
 }