From b44a6c9ea91a68e644697c8b50455d215818cde4 Mon Sep 17 00:00:00 2001 From: Vamsi-klu Date: Sat, 11 Jul 2026 21:37:38 +0000 Subject: [PATCH] Preserve BigDecimal precision during JSON ingestion --- .../inputformat/json/JSONMessageDecoder.java | 14 +++- .../inputformat/json/JSONRecordExtractor.java | 9 +-- .../inputformat/json/JSONRecordReader.java | 2 +- .../json/JSONMessageDecoderTest.java | 67 ++++++++++++++++++- .../json/JSONRecordExtractorTest.java | 7 ++ .../json/JSONRecordReaderTest.java | 39 ++++++++++- .../org/apache/pinot/spi/utils/JsonUtils.java | 15 +++++ .../apache/pinot/spi/utils/JsonUtilsTest.java | 21 ++++++ 8 files changed, 163 insertions(+), 11 deletions(-) diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java index 08a1122e6c4e..6f04fb3de575 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoder.java +++ b/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 { 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> _jsonRecordExtractor; + private boolean _preserveDecimalPrecision; @Override public void init(Map props, Set fieldsToRead, String topicName) @@ -46,6 +48,14 @@ public void init(Map props, Set 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 jsonMap = JsonUtils.bytesToMap(payload, offset, length); + Map jsonMap = _preserveDecimalPrecision + ? JsonUtils.bytesToMapWithBigDecimal(payload, offset, length) + : JsonUtils.bytesToMap(payload, offset, length); return _jsonRecordExtractor.extract(jsonMap, destination); } catch (Exception e) { throw new RuntimeException( diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java index d2fbe09da64f..072e4687857f 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractor.java +++ b/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` (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 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` 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) value); } - // Single value pass-through (Boolean / Integer / Long / Double / String) + // Single value pass-through (Boolean / Integer / Long / Double / BigDecimal / String) return value; } diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReader.java b/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReader.java index 14a30d93ba30..94dd423fb559 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/main/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReader.java +++ b/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>() { + _iterator = JsonUtils.READER_WITH_BIG_DECIMAL.forType(new TypeReference>() { }).readValues(_inputStream); } catch (Exception e) { if (_iterator != null) { diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java index 5c2bda51b873..055ed55c05fb 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONMessageDecoderTest.java +++ b/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> { + @Override + public void init(Set fields, RecordExtractorConfig config) { + } + + @Override + public GenericRow extract(Map 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 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."); diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractorTest.java b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractorTest.java index a67c42d6bfe1..5b80690bbc43 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordExtractorTest.java +++ b/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"); diff --git a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReaderTest.java b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReaderTest.java index 82f83e177904..b015192da135 100644 --- a/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReaderTest.java +++ b/pinot-plugins/pinot-input-format/pinot-json/src/test/java/org/apache/pinot/plugin/inputformat/json/JSONRecordReaderTest.java @@ -21,8 +21,10 @@ 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; @@ -30,9 +32,12 @@ 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> expectedRecordsMap, List expectedPrimaryKeys) @@ -71,14 +94,13 @@ protected void checkValue(RecordReader recordReader, List> 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> 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()); + } + } } diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java index 63726acef1be..eb2f6e31777a 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/JsonUtils.java @@ -277,16 +277,31 @@ public static Map stringToMap(String jsonString) return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonString); } + public static Map stringToMapWithBigDecimal(String jsonString) + throws JsonProcessingException { + return READER_WITH_BIG_DECIMAL.forType(MAP_TYPE_REFERENCE).readValue(jsonString); + } + public static Map bytesToMap(byte[] jsonBytes) throws IOException { return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes); } + public static Map bytesToMapWithBigDecimal(byte[] jsonBytes) + throws IOException { + return READER_WITH_BIG_DECIMAL.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes); + } + public static Map bytesToMap(byte[] jsonBytes, int offset, int length) throws IOException { return DEFAULT_READER.forType(MAP_TYPE_REFERENCE).readValue(jsonBytes, offset, length); } + public static Map 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); diff --git a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/JsonUtilsTest.java b/pinot-spi/src/test/java/org/apache/pinot/spi/utils/JsonUtilsTest.java index 3a9691c22052..4c856fb96707 100644 --- a/pinot-spi/src/test/java/org/apache/pinot/spi/utils/JsonUtilsTest.java +++ b/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 stringMap = JsonUtils.stringToMapWithBigDecimal(json); + Map bytesMap = JsonUtils.bytesToMapWithBigDecimal(jsonBytes); + Map 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 {