Preserve BigDecimal precision during JSON ingestion

pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java

@@ -33,8 +33,10 @@
 public class JSONMessageDecoder implements StreamMessageDecoder<byte[]> {
   private static final String JSON_RECORD_EXTRACTOR_CLASS =
       "org.apache.pinot.plugin.inputformat.json.JSONRecordExtractor";
+  private static final String PRESERVE_DECIMAL_PRECISION_CONFIG_KEY = "preserveDecimalPrecision";
 
   private RecordExtractor<Map<String, Object>> _jsonRecordExtractor;
+  private boolean _preserveDecimalPrecision;
 
   @Override
   public void init(Map<String, String> props, Set<String> fieldsToRead, String topicName)
@@ -46,6 +48,14 @@ public void init(Map<String, String> props, Set<String> fieldsToRead, String top
     if (recordExtractorClass == null) {
       recordExtractorClass = JSON_RECORD_EXTRACTOR_CLASS;
     }
+    String preserveDecimalPrecision = null;
+    if (props != null) {
+      preserveDecimalPrecision = props.get(PRESERVE_DECIMAL_PRECISION_CONFIG_KEY);
+    }
+    _preserveDecimalPrecision = preserveDecimalPrecision != null
+        ? Boolean.parseBoolean(preserveDecimalPrecision)
+        : JSON_RECORD_EXTRACTOR_CLASS.equals(recordExtractorClass);
+
     _jsonRecordExtractor = PluginManager.get().createInstance(recordExtractorClass);
     _jsonRecordExtractor.init(fieldsToRead, null);
   }
@@ -59,7 +69,9 @@ public GenericRow decode(byte[] payload, GenericRow destination) {
   public GenericRow decode(byte[] payload, int offset, int length, GenericRow destination) {
     try {
       // Parse directly to Map, avoiding intermediate JsonNode representation for better performance
-      Map<String, Object> jsonMap = JsonUtils.bytesToMap(payload, offset, length);
+      Map<String, Object> jsonMap = _preserveDecimalPrecision
+          ? JsonUtils.bytesToMapWithBigDecimal(payload, offset, length)
+          : JsonUtils.bytesToMap(payload, offset, length);
       return _jsonRecordExtractor.extract(jsonMap, destination);
     } catch (Exception e) {
       throw new RuntimeException(

pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java

@@ -28,14 +28,15 @@
 
 
 /// Extracts Pinot [GenericRow] from a parsed JSON `Map<String, Object>` (Jackson representation). JSON has
-/// no native bytes / float / big-decimal type.
+/// no native bytes / float type.
 ///
 /// **JSON source type → Java input → Java output type:**
 /// - `true` / `false` → `Boolean` → `Boolean`
 /// - int that fits in 32 bits → `Integer` → `Integer`
 /// - int that overflows 32 bits but fits in 64 → `Long` → `Long`
 /// - int that overflows 64 bits → `BigInteger` → `BigDecimal` (Pinot has no `BigInteger` data type)
-/// - decimal → `Double` → `Double` (never `Float` or `BigDecimal` with default Jackson config)
+/// - decimal → `BigDecimal` → `BigDecimal` when parsed with Pinot's BigDecimal-aware JSON readers
+/// - decimal → `Double` → `Double` for callers that still provide default Jackson-decoded maps
 /// - string → `String` → `String`
 /// - `null` → `null` → `null`
 /// - array → `List` → `Object[]` (each element recursively converted)
@@ -60,7 +61,7 @@ public GenericRow extract(Map<String, Object> from, GenericRow to) {
 
   /// Walks a non-null Jackson-parsed value and produces the contract shape: `BigDecimal` for `BigInteger`
   /// (oversized ints), `Object[]` for JSON arrays, `Map<String, Object>` for JSON objects, pass-through for
-  /// the other Jackson scalar types (`Boolean`, `Integer`, `Long`, `Double`, `String`).
+  /// the other Jackson scalar types (`Boolean`, `Integer`, `Long`, `Double`, `BigDecimal`, `String`).
   private static Object convert(Object value) {
     // BigInteger widens (Pinot has no BigInteger type)
     if (value instanceof BigInteger) {
@@ -76,7 +77,7 @@ private static Object convert(Object value) {
       //noinspection unchecked
       return convertMap((Map<String, Object>) value);
     }
-    // Single value pass-through (Boolean / Integer / Long / Double / String)
+    // Single value pass-through (Boolean / Integer / Long / Double / BigDecimal / String)
     return value;
   }
 

pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReader.java

@@ -51,7 +51,7 @@ private void init()
       throws IOException {
     _inputStream = RecordReaderUtils.getBufferedInputStream(_dataFile);
     try {
-      _iterator = JsonUtils.DEFAULT_READER.forType(new TypeReference<Map<String, Object>>() {
+      _iterator = JsonUtils.READER_WITH_BIG_DECIMAL.forType(new TypeReference<Map<String, Object>>() {
       }).readValues(_inputStream);
     } catch (Exception e) {
       if (_iterator != null) {

pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java

@@ -22,13 +22,19 @@
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
+import java.math.BigDecimal;
 import java.net.URL;
+import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import org.apache.pinot.spi.data.FieldSpec;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
 import org.apache.pinot.spi.data.Schema;
 import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordExtractor;
+import org.apache.pinot.spi.data.readers.RecordExtractorConfig;
+import org.apache.pinot.spi.stream.StreamMessageDecoder;
 import org.apache.pinot.spi.utils.JsonUtils;
 import org.testng.annotations.Test;
 
@@ -38,6 +44,7 @@
 
 
 public class JSONMessageDecoderTest {
+  private static final String PRECISE_DECIMAL = "12345678901234567890.12345678901234567890";
 
   @Test
   public void testJsonDecoderWithoutOutgoingTimeSpec()
@@ -74,6 +81,62 @@ public void testJsonDecoderNoTimeSpec()
     testJsonDecoder(sourceFields);
   }
 
+  @Test
+  public void testJsonDecoderPreservesBigDecimalPrecision()
+      throws Exception {
+    JSONMessageDecoder decoder = new JSONMessageDecoder();
+    decoder.init(Map.of(), Set.of("decimalMetric", "doubleMetric"), "testTopic");
+
+    byte[] payload = ("{\"decimalMetric\":" + PRECISE_DECIMAL + ",\"doubleMetric\":1.25}").getBytes(
+        StandardCharsets.UTF_8);
+    GenericRow row = decoder.decode(payload, new GenericRow());
+
+    assertEquals(row.getValue("decimalMetric"), new BigDecimal(PRECISE_DECIMAL));
+    assertEquals(((BigDecimal) row.getValue("doubleMetric")).doubleValue(), 1.25d);
+  }
+
+  @Test
+  public void testCustomRecordExtractorKeepsLegacyDoubleParsingByDefault()
+      throws Exception {
+    JSONMessageDecoder decoder = new JSONMessageDecoder();
+    decoder.init(Map.of(StreamMessageDecoder.RECORD_EXTRACTOR_CONFIG_KEY, DecimalTypeRecordExtractor.class.getName()),
+        Set.of("decimalMetric"), "testTopic");
+
+    GenericRow row = decoder.decode("{\"decimalMetric\":1.25}".getBytes(StandardCharsets.UTF_8), new GenericRow());
+
+    assertEquals(row.getValue("decimalMetricType"), Double.class.getName());
+    assertEquals(row.getValue("decimalMetric"), 1.25d);
+  }
+
+  @Test
+  public void testCustomRecordExtractorCanOptIntoBigDecimalParsing()
+      throws Exception {
+    JSONMessageDecoder decoder = new JSONMessageDecoder();
+    decoder.init(Map.of(
+            StreamMessageDecoder.RECORD_EXTRACTOR_CONFIG_KEY, DecimalTypeRecordExtractor.class.getName(),
+            "preserveDecimalPrecision", "true"),
+        Set.of("decimalMetric"), "testTopic");
+
+    GenericRow row = decoder.decode("{\"decimalMetric\":1.25}".getBytes(StandardCharsets.UTF_8), new GenericRow());
+
+    assertEquals(row.getValue("decimalMetricType"), BigDecimal.class.getName());
+    assertEquals(row.getValue("decimalMetric"), new BigDecimal("1.25"));
+  }
+
+  public static class DecimalTypeRecordExtractor implements RecordExtractor<Map<String, Object>> {
+    @Override
+    public void init(Set<String> fields, RecordExtractorConfig config) {
+    }
+
+    @Override
+    public GenericRow extract(Map<String, Object> from, GenericRow to) {
+      Object value = from.get("decimalMetric");
+      to.putValue("decimalMetric", value);
+      to.putValue("decimalMetricType", value.getClass().getName());
+      return to;
+    }
+  }
+
   private Schema loadSchema(String resourcePath)
       throws Exception {
     URL resource = getClass().getClassLoader().getResource(resourcePath);
@@ -107,10 +170,10 @@ private void testJsonDecoder(Map<String, DataType> sourceFields)
               assertEquals(actualValue, expectedValue.asLong());
               break;
             case FLOAT:
-              assertEquals(actualValue, (float) expectedValue.asDouble());
+              assertEquals(((Number) actualValue).floatValue(), (float) expectedValue.asDouble());
               break;
             case DOUBLE:
-              assertEquals(actualValue, expectedValue.asDouble());
+              assertEquals(((Number) actualValue).doubleValue(), expectedValue.asDouble());
               break;
             default:
               fail("Shouldn't arrive here.");

pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractorTest.java

@@ -74,6 +74,13 @@ public void testDoublePreserved() {
     assertEquals(result, 1.5d);
   }
 
+  @Test
+  public void testBigDecimalPreserved() {
+    BigDecimal value = new BigDecimal("12345678901234567890.12345678901234567890");
+    Object result = extract(value);
+    assertEquals(result, value);
+  }
+
   @Test
   public void testStringPreserved() {
     Object result = extract("hello");

pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReaderTest.java

@@ -21,18 +21,23 @@
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import java.io.File;
 import java.io.FileWriter;
+import java.math.BigDecimal;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import org.apache.pinot.spi.data.FieldSpec;
 import org.apache.pinot.spi.data.readers.AbstractRecordReaderTest;
 import org.apache.pinot.spi.data.readers.GenericRow;
 import org.apache.pinot.spi.data.readers.PrimaryKey;
 import org.apache.pinot.spi.data.readers.RecordReader;
 import org.apache.pinot.spi.utils.JsonUtils;
 import org.testng.Assert;
+import org.testng.annotations.Test;
 
 
 public class JSONRecordReaderTest extends AbstractRecordReaderTest {
+  private static final String PRECISE_DECIMAL = "12345678901234567890.12345678901234567890";
+  private static final double EPSILON = 1e-6d;
 
   @Override
   protected RecordReader createRecordReader(File file)
@@ -61,6 +66,24 @@ protected String getDataFileName() {
     return "data.json";
   }
 
+  @Test
+  public void testRecordReaderPreservesBigDecimalPrecision()
+      throws Exception {
+    File dataFile = new File(_tempDir, "big_decimal_data.json");
+    try (FileWriter fileWriter = new FileWriter(dataFile)) {
+      fileWriter.write("{\"decimalMetric\":" + PRECISE_DECIMAL + ",\"doubleMetric\":1.25}");
+    }
+
+    try (JSONRecordReader recordReader = new JSONRecordReader()) {
+      recordReader.init(dataFile, Set.of("decimalMetric", "doubleMetric"), null);
+      GenericRow actualRecord = recordReader.next();
+
+      Assert.assertEquals(actualRecord.getValue("decimalMetric"), new BigDecimal(PRECISE_DECIMAL));
+      Assert.assertEquals(((BigDecimal) actualRecord.getValue("doubleMetric")).doubleValue(), 1.25d);
+      Assert.assertFalse(recordReader.hasNext());
+    }
+  }
+
   @Override
   protected void checkValue(RecordReader recordReader, List<Map<String, Object>> expectedRecordsMap,
       List<Object[]> expectedPrimaryKeys)
@@ -71,14 +94,13 @@ protected void checkValue(RecordReader recordReader, List<Map<String, Object>> e
       for (FieldSpec fieldSpec : _pinotSchema.getAllFieldSpecs()) {
         String fieldSpecName = fieldSpec.getName();
         if (fieldSpec.isSingleValueField()) {
-          Assert.assertEquals(actualRecord.getValue(fieldSpecName).toString(),
-              expectedRecord.get(fieldSpecName).toString());
+          assertRecordValueEquals(actualRecord.getValue(fieldSpecName), expectedRecord.get(fieldSpecName));
         } else {
           Object[] actualRecords = (Object[]) actualRecord.getValue(fieldSpecName);
           List expectedRecords = (List) expectedRecord.get(fieldSpecName);
           Assert.assertEquals(actualRecords.length, expectedRecords.size());
           for (int j = 0; j < actualRecords.length; j++) {
-            Assert.assertEquals(actualRecords[j].toString(), expectedRecords.get(j).toString());
+            assertRecordValueEquals(actualRecords[j], expectedRecords.get(j));
           }
         }
       }
@@ -87,4 +109,15 @@ protected void checkValue(RecordReader recordReader, List<Map<String, Object>> e
     }
     Assert.assertFalse(recordReader.hasNext());
   }
+
+  private static void assertRecordValueEquals(Object actualValue, Object expectedValue) {
+    boolean isFloating = expectedValue instanceof Float || expectedValue instanceof Double
+        || actualValue instanceof Float || actualValue instanceof Double
+        || (actualValue instanceof BigDecimal && (expectedValue instanceof Float || expectedValue instanceof Double));
+    if (isFloating) {
+      Assert.assertEquals(((Number) actualValue).doubleValue(), ((Number) expectedValue).doubleValue(), EPSILON);
+    } else {
+      Assert.assertEquals(actualValue.toString(), expectedValue.toString());
+    }
+  }
 }

pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java

@@ -277,16 +277,31 @@ public static Map<String, Object> stringToMap(String jsonString)
     return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonString);
   }
 
+  public static Map<String, Object> stringToMapWithBigDecimal(String jsonString)
+      throws JsonProcessingException {
+    return READER_WITH_BIG_DECIMAL.forType(MAP_TYPE_REFERENCE).readValue(jsonString);
+  }
+
   public static Map<String, Object> bytesToMap(byte[] jsonBytes)
       throws IOException {
     return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes);
   }
 
+  public static Map<String, Object> bytesToMapWithBigDecimal(byte[] jsonBytes)
+      throws IOException {
+    return READER_WITH_BIG_DECIMAL.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes);
+  }
+
   public static Map<String, Object> bytesToMap(byte[] jsonBytes, int offset, int length)
       throws IOException {
     return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes, offset, length);
   }
 
+  public static Map<String, Object> bytesToMapWithBigDecimal(byte[] jsonBytes, int offset, int length)
+      throws IOException {
+    return READER_WITH_BIG_DECIMAL.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes, offset, length);
+  }
+
   public static String objectToString(Object object)
       throws JsonProcessingException {
     return DEFAULT_WRITER.writeValueAsString(object);

pinot-spi/src/test/java/org/apache/pinot/spi/utils/JsonUtilsTest.java

@@ -22,6 +22,7 @@
 import com.google.common.collect.ImmutableSet;
 import java.io.File;
 import java.io.IOException;
+import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
 import java.sql.Timestamp;
 import java.time.LocalDate;
@@ -49,6 +50,7 @@
 
 public class JsonUtilsTest {
   private static final String JSON_FILE = "json_util_test.json";
+  private static final String PRECISE_DECIMAL = "12345678901234567890.12345678901234567890";
 
   @Test
   public void testFlatten()
@@ -293,6 +295,25 @@ public void testFlatten()
     }
   }
 
+  @Test
+  public void testMapParsingWithBigDecimalPreservesDecimalPrecision()
+      throws Exception {
+    String json = "{\"decimalMetric\":" + PRECISE_DECIMAL + ",\"nested\":{\"value\":" + PRECISE_DECIMAL + "}}";
+
+    byte[] jsonBytes = json.getBytes(StandardCharsets.UTF_8);
+    Map<String, Object> stringMap = JsonUtils.stringToMapWithBigDecimal(json);
+    Map<String, Object> bytesMap = JsonUtils.bytesToMapWithBigDecimal(jsonBytes);
+    Map<String, Object> offsetBytesMap = JsonUtils.bytesToMapWithBigDecimal(jsonBytes, 0, jsonBytes.length);
+
+    BigDecimal expectedValue = new BigDecimal(PRECISE_DECIMAL);
+    assertEquals(stringMap.get("decimalMetric"), expectedValue);
+    assertEquals(((Map<?, ?>) stringMap.get("nested")).get("value"), expectedValue);
+    assertEquals(bytesMap.get("decimalMetric"), expectedValue);
+    assertEquals(((Map<?, ?>) bytesMap.get("nested")).get("value"), expectedValue);
+    assertEquals(offsetBytesMap.get("decimalMetric"), expectedValue);
+    assertEquals(((Map<?, ?>) offsetBytesMap.get("nested")).get("value"), expectedValue);
+  }
+
   @Test
   public void testFlattenWithMaxLevels()
       throws IOException {