This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 18ae067f98b Parse-once for JSON-index ingestion: cache the parsed Map
and flatten it directly (#18756)
18ae067f98b is described below
commit 18ae067f98b8f6b198389313dde58e9c27a9a5eb
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Oct 12 22:01:51 2026 -0700
Parse-once for JSON-index ingestion: cache the parsed Map and flatten it
directly (#18756)
* Parse-once for JSON-index ingestion: cache the parsed JSON and flatten it
directly
A JSON-typed column with a JSON index is parsed twice per row: the value
(JSON
text in the common case) is parsed and re-serialized to a canonical string
for
the forward index, and the JSON index then re-parses that string. For large
documents (e.g. log messages with stack traces) the re-tokenization
dominates.
DataTypeTransformer now parses the JSON once: it caches the parsed value on
the
GenericRow and reuses its canonical string for the forward index, so the
JSON
index flattens the cached value via the new
MutableJsonIndex.addParsed(Object) /
JsonUtils.flattenParsed(Object) instead of re-parsing. This covers JSON
columns
fed as text (parse the string once, cache the JsonNode), as Maps/Lists, or
as
JsonNodes.
Behavior-preserving (the flattened records are byte-for-byte identical to
the
string path): flattenParsed re-parses each DecimalNode leaf the way the
string
path does (so an integral decimal like 2.0 -> "2", scientific notation, and
values past 2^53 stay identical) and falls back to serialize+reparse only
for
Float / byte[] leaves from a non-JSON RecordReader. Gated by
supportsParsedValue()
so an index that does not optimize the parsed path keeps getting the
serialized
string.
The cache self-invalidates: GenericRow.putValue/putValues/removeValue/
putDefaultNullValue drop the parsed entry, so a transformer that rewrites
the
value after DataTypeTransformer (e.g. SanitizationTransformer trimming an
over-length JSON string) cannot leave a stale parsed node for the index.
SPI surface (all additive, source/binary compatible):
MutableJsonIndex.addParsed
+ supportsParsedValue default methods, GenericRow parsed-value cache
accessors,
JsonUtils.flattenParsed overload.
~3-7x JSON-index ingestion throughput on multi-KB documents
(BenchmarkJsonFlatten);
~no change for tiny JSON.
* Keep cached JSON indexing consistent with forward values
Preserve decimal tokens and sorted Map ordering, and reuse the forward
string for Float/binary inputs. Reproduce with Map values containing
BigDecimal("2.0"), dotted keys, or Avro-style leaves.
* Respect JSON parser limits and classify cached JSON values once
A cached Map/List/JsonNode was flattened directly even when the string path
cannot parse its forward value (e.g. a name longer than Jackson's
50,000-char
limit or an integer longer than 1,000 digits). With skipInvalidJson, the
consuming index then matched the document while the immutable index rebuilt
from the forward value skipped it.
The JSON index now classifies each cached value once, checking nesting
depth, name length, string length, and integer length against the
constraints of the reader the string path uses. MutableSegmentImpl passes
the forward value with the cached value, so a value that cannot be flattened
directly (Float/binary leaves, limit violations) indexes that string without
being serialized again, and the transformer no longer runs a preflight walk.
Nested decimal tokens are re-parsed inside a container, as in the forward
value, and a failure while converting the value defers to the string path,
which covers decimal tokens its reader accepts at input buffer boundaries.
SanitizationTransformer no longer writes back unchanged values, which
dropped the cached value of every JSON column with a max-length strategy.
* Avoid deep recursion in JSON parser-limit tests
Flattening a 1000-level document recursively needs about 950 KB of stack
with the JaCoCo agent, so testFlattenParsedRespectsParserConstraints
overflowed the 1 MB test thread stack in CI. Documents at the depth limit
are now only classified, and documents one level past it are compared
against a hand-built forward value instead of being serialized.
---
.../apache/pinot/perf/BenchmarkJsonFlatten.java | 112 ++++++++
.../indexsegment/mutable/MutableSegmentImpl.java | 13 +
.../realtime/impl/json/MutableJsonIndexImpl.java | 20 ++
.../recordtransformer/DataTypeTransformer.java | 89 ++++++-
.../recordtransformer/SanitizationTransformer.java | 8 +-
.../indexsegment/mutable/IndexingFailureTest.java | 38 +++
.../recordtransformer/DataTypeTransformerTest.java | 132 +++++++++
.../segment/local/segment/index/JsonIndexTest.java | 159 +++++++++++
.../apache/pinot/spi/data/readers/GenericRow.java | 38 +++
.../java/org/apache/pinot/spi/utils/JsonUtils.java | 266 +++++++++++++++++++
.../pinot/spi/data/readers/GenericRowTest.java | 36 +++
.../org/apache/pinot/spi/utils/JsonUtilsTest.java | 294 +++++++++++++++++++++
12 files changed, 1197 insertions(+), 8 deletions(-)
diff --git
a/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonFlatten.java
b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonFlatten.java
new file mode 100644
index 00000000000..6452abbadd6
--- /dev/null
+++ b/pinot-perf/src/main/java/org/apache/pinot/perf/BenchmarkJsonFlatten.java
@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.perf;
+
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.pinot.spi.config.table.JsonIndexConfig;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.apache.pinot.spi.utils.MapUtils;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+
+/// Compares the cost of building the JSON index input for a realtime row from
a re-serialized JSON **string**
+/// (the current path: `DataTypeTransformer` serializes the Map, then the JSON
index re-parses it) versus from the
+/// already-parsed **Map** (the parse-once cache: the index flattens the Map
directly via `valueToTree`).
+///
+/// - `flattenFromString` - current per-row index cost (`stringToJsonNode` +
flatten).
+/// - `flattenFromMap` - classification plus `valueToTree` + flatten when
supported.
+/// - `serializeMap` - the Map to String serialization that stays (the forward
index stores it).
+/// Float/binary leaves decline the optimization after classification and
flatten the serialized forward value.
+@State(Scope.Benchmark)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.MILLISECONDS)
+@Fork(1)
+@Warmup(iterations = 5, time = 1)
+@Measurement(iterations = 8, time = 1)
+public class BenchmarkJsonFlatten {
+
+ // A representative structured log line (nested "log" object, ids, an array).
+ private static final String MESSAGE_JSON =
+ "{\"x_gr_trace_id\":\"a1b2c3d4-e5f6-7890-abcd-ef1234567890\","
+ + "\"user_id\":\"user-99174432\","
+ +
"\"path\":\"/retail_purchase_order/modules/serialized_inventory.py\","
+ +
"\"status\":\"500\",\"event\":\"request.error\",\"level\":\"ERROR\","
+ +
"\"stream\":\"stderr\",\"time\":\"2026-06-12T10:31:13.123456789Z\","
+ + "\"msg\":\"File
\\\"/retail_purchase_order/modules/serialized_inventory.py\\\", line 222, in "
+ + "create_integration_partner_entries\","
+ + "\"log\":{\"msg\":\"create_integration_partner_entries
failed\",\"message\":\"partner missing\","
+ +
"\"level\":\"error\",\"logger\":\"retail.purchase_order.serialized_inventory\",\"line\":222},"
+ +
"\"http\":{\"method\":\"POST\",\"status_code\":500,\"duration_ms\":1843,\"host\":\"api-internal\"},"
+ +
"\"tags\":[\"retail\",\"purchase_order\",\"serialized_inventory\",\"prod\"]}";
+
+ private Map<String, Object> _messageMap;
+ private String _forwardValue;
+ private JsonIndexConfig _config;
+
+ @Param({"JSON", "FLOAT_BINARY"})
+ public String _inputType;
+
+ @Setup
+ public void setup()
+ throws Exception {
+ _messageMap = JsonUtils.stringToObject(MESSAGE_JSON, Map.class);
+ if ("FLOAT_BINARY".equals(_inputType)) {
+ _messageMap.put("ratio", 0.1f);
+ _messageMap.put("payload", new byte[]{1, 2, 3});
+ }
+ _forwardValue = MapUtils.toString(_messageMap);
+ _config = new JsonIndexConfig();
+ _config.setMaxLevels(2);
+ }
+
+ @Benchmark
+ public Object flattenFromString()
+ throws Exception {
+ return JsonUtils.flatten(_forwardValue, _config);
+ }
+
+ @Benchmark
+ public Object flattenFromMap()
+ throws Exception {
+ return JsonUtils.flattenParsed(_messageMap, _forwardValue, _config);
+ }
+
+ @Benchmark
+ public Object serializeMap() {
+ return MapUtils.toString(_messageMap);
+ }
+
+ public static void main(String[] args)
+ throws Exception {
+ new Runner(new
OptionsBuilder().include(BenchmarkJsonFlatten.class.getSimpleName()).build()).run();
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
index aa699f38570..c74a24ca27e 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java
@@ -61,6 +61,7 @@ import
org.apache.pinot.segment.local.realtime.impl.dictionary.BaseOffHeapMutabl
import
org.apache.pinot.segment.local.realtime.impl.dictionary.SameValueMutableDictionary;
import
org.apache.pinot.segment.local.realtime.impl.forward.SameValueMutableForwardIndex;
import
org.apache.pinot.segment.local.realtime.impl.invertedindex.MultiColumnRealtimeLuceneTextIndex;
+import org.apache.pinot.segment.local.realtime.impl.json.MutableJsonIndexImpl;
import
org.apache.pinot.segment.local.realtime.impl.nullvalue.MutableNullValueVector;
import
org.apache.pinot.segment.local.segment.index.datasource.MutableDataSource;
import
org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
@@ -963,6 +964,18 @@ public class MutableSegmentImpl implements MutableSegment {
for (Map.Entry<IndexType, MutableIndex> indexEntry :
indexContainer._mutableIndexes.entrySet()) {
try {
MutableIndex mutableIndex = indexEntry.getValue();
+ if (mutableIndex instanceof MutableJsonIndexImpl) {
+ // Feed the JSON index the parsed value cached by the
DataTypeTransformer (if any), so it flattens the
+ // document directly instead of re-parsing the string this
column was serialized into for the forward
+ // index; the index falls back to that string for documents it
cannot flatten directly. This is
+ // intentionally scoped to Pinot's built-in JSON index
implementation.
+ Object parsedValue = row.getParsedJsonValue(column);
+ if (parsedValue != null && value instanceof String) {
+ ((MutableJsonIndexImpl) mutableIndex).addParsed(parsedValue,
(String) value);
+ updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
+ continue;
+ }
+ }
mutableIndex.add(value, dictId, docId);
updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
} catch (Exception e) {
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/json/MutableJsonIndexImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/json/MutableJsonIndexImpl.java
index 8010ab5d20d..f9fde3c6ea3 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/json/MutableJsonIndexImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/realtime/impl/json/MutableJsonIndexImpl.java
@@ -115,6 +115,26 @@ public class MutableJsonIndexImpl implements
MutableJsonIndex {
}
}
+ /// Indexes an already-parsed JSON document (e.g. a Map cached on the
GenericRow), flattening it directly via
+ /// {@link JsonUtils#flattenParsed(Object, String, JsonIndexConfig)} instead
of re-parsing `jsonString`, the string
+ /// the document was serialized into for the forward index. A document that
cannot be flattened directly is indexed
+ /// from `jsonString`, as {@link #add(String)} does.
+ public void addParsed(Object parsedValue, String jsonString)
+ throws IOException {
+ try {
+ List<Map<String, String>> flattenedRecords =
+ JsonUtils.flattenParsed(parsedValue, jsonString, _jsonIndexConfig);
+ _writeLock.lock();
+ try {
+ addFlattenedRecords(flattenedRecords);
+ } finally {
+ _writeLock.unlock();
+ }
+ } finally {
+ _nextDocId++;
+ }
+ }
+
/// Adds the flattened records for the next document.
private void addFlattenedRecords(List<Map<String, String>> records) {
int numRecords = records.size();
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformer.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformer.java
index c7c70f61ff6..24551220e9e 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformer.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformer.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.segment.local.recordtransformer;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -27,12 +29,16 @@ import java.util.UUID;
import javax.annotation.Nullable;
import org.apache.pinot.common.utils.ThrottledLogger;
import org.apache.pinot.segment.local.utils.DataTypeTransformerUtils;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigs;
+import org.apache.pinot.segment.spi.index.FieldIndexConfigsUtil;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.ingestion.IngestionConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.recordtransformer.RecordTransformer;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.PinotDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -51,6 +57,11 @@ public class DataTypeTransformer implements
RecordTransformer {
private static final Logger LOGGER =
LoggerFactory.getLogger(DataTypeTransformer.class);
private final Map<String, PinotDataType> _dataTypes;
+ /// JSON-typed columns that have a JSON index. For these the parsed value is
cached on the row (before it is
+ /// serialized to a string for the forward index) so the JSON index can
flatten it directly instead of re-parsing.
+ /// Only the consuming segment's JSON index reads the cache; caching costs a
reference per row elsewhere. Empty for
+ /// the source-field usage, which runs before indexing.
+ private final Set<String> _jsonCacheColumns;
private final boolean _continueOnError;
private final ThrottledLogger _throttledLogger;
// UUID primary key columns for upsert/dedup tables; non-null and non-empty
when applicable.
@@ -64,18 +75,20 @@ public class DataTypeTransformer implements
RecordTransformer {
/// Creates a [DataTypeTransformer] that converts the (non-virtual) schema
columns to the data types defined in the
/// [Schema].
public DataTypeTransformer(TableConfig tableConfig, Schema schema) {
- this(tableConfig, extractSchemaDataTypes(schema), schema);
+ this(tableConfig, extractSchemaDataTypes(schema), schema,
computeJsonCacheColumns(tableConfig, schema));
}
/// Creates a [DataTypeTransformer] that converts the given columns to the
provided [PinotDataType]s. This is useful
/// for fixing the data types of source fields before other transformers
(such as [ExpressionTransformer]) consume
/// them.
public DataTypeTransformer(TableConfig tableConfig, Map<String,
PinotDataType> dataTypes) {
- this(tableConfig, dataTypes, null);
+ this(tableConfig, dataTypes, null, Set.of());
}
- private DataTypeTransformer(TableConfig tableConfig, Map<String,
PinotDataType> dataTypes, @Nullable Schema schema) {
+ private DataTypeTransformer(TableConfig tableConfig, Map<String,
PinotDataType> dataTypes,
+ @Nullable Schema schema, Set<String> jsonCacheColumns) {
_dataTypes = dataTypes;
+ _jsonCacheColumns = jsonCacheColumns;
IngestionConfig ingestionConfig = tableConfig.getIngestionConfig();
_continueOnError = ingestionConfig != null &&
ingestionConfig.isContinueOnError();
_throttledLogger = new ThrottledLogger(LOGGER, ingestionConfig);
@@ -116,6 +129,29 @@ public class DataTypeTransformer implements
RecordTransformer {
return _dataTypes.isEmpty();
}
+ /// Returns the JSON-typed columns that have an (enabled) JSON index, i.e.
the columns worth caching the parsed value
+ /// for. If the index configs cannot be resolved, returns an empty set and
the optimization is disabled (the JSON
+ /// index simply re-parses the serialized string, as before).
+ private static Set<String> computeJsonCacheColumns(TableConfig tableConfig,
Schema schema) {
+ Set<String> columns = new HashSet<>();
+ try {
+ Map<String, FieldIndexConfigs> indexConfigsByColumn =
+ FieldIndexConfigsUtil.createIndexConfigsByColName(tableConfig,
schema);
+ for (FieldSpec fieldSpec : schema.getAllFieldSpecs()) {
+ if (!fieldSpec.isVirtualColumn() && fieldSpec.getDataType() ==
FieldSpec.DataType.JSON) {
+ FieldIndexConfigs configs =
indexConfigsByColumn.get(fieldSpec.getName());
+ if (configs != null &&
configs.getConfig(StandardIndexes.json()).isEnabled()) {
+ columns.add(fieldSpec.getName());
+ }
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Failed to compute JSON parse-once cache columns; disabling
the optimization", e);
+ return Set.of();
+ }
+ return columns;
+ }
+
@Override
public Set<String> getInputColumns() {
return _dataTypes.keySet();
@@ -131,8 +167,40 @@ public class DataTypeTransformer implements
RecordTransformer {
&& value instanceof CharSequence) {
validateCanonicalUuidPrimaryKey(column, value.toString());
}
- value = DataTypeTransformerUtils.transformValue(column, value,
entry.getValue());
+ // For a JSON column with a JSON index, cache the parsed value so the
index flattens it directly instead of
+ // re-parsing the string the value is serialized into for the forward
index. The parsed value is cached AFTER
+ // the forward value is written (below), because putValue invalidates
the parsed-JSON cache: a later transformer
+ // that rewrites the value (e.g. sanitization trimming an over-length
JSON string) must not leave a stale node.
+ Object parsedToCache = null;
+ if (value != null && !_jsonCacheColumns.isEmpty() &&
_jsonCacheColumns.contains(column)) {
+ if (value instanceof Map || value instanceof List || value
instanceof JsonNode) {
+ // Already-parsed JSON: serialize it for the forward index (below)
and cache the parsed value for the index,
+ // which flattens the forward string instead for a value it cannot
flatten directly (e.g. Float leaves).
+ parsedToCache = value;
+ value = DataTypeTransformerUtils.transformValue(column, value,
entry.getValue());
+ } else if (value instanceof String) {
+ // The column value is JSON text. The JSON conversion parses it
(preserving BigDecimal) and re-serializes it
+ // to the canonical string for the forward index; the JSON index
then re-parses that string. Parse it once
+ // here instead: reuse the node's canonical string for the forward
index and cache the node for the index.
+ JsonNode parsed = tryParseJson((String) value);
+ if (parsed != null) {
+ parsedToCache = parsed;
+ // parsed.toString() equals PinotDataType.JSON.convert(value)
for well-formed JSON (byte-identical).
+ value = parsed.toString();
+ } else {
+ // Not well-formed JSON: the normal conversion wraps it as a
JSON string.
+ value = DataTypeTransformerUtils.transformValue(column, value,
entry.getValue());
+ }
+ } else {
+ value = DataTypeTransformerUtils.transformValue(column, value,
entry.getValue());
+ }
+ } else {
+ value = DataTypeTransformerUtils.transformValue(column, value,
entry.getValue());
+ }
record.putValue(column, value);
+ if (parsedToCache != null) {
+ record.putParsedJsonValue(column, parsedToCache);
+ }
} catch (Exception e) {
if (!_continueOnError) {
throw new RuntimeException("Caught exception while transforming data
type for column: " + column, e);
@@ -178,4 +246,17 @@ public class DataTypeTransformer implements
RecordTransformer {
+ "routing. Non-canonical values cause silent dedup failures in
multi-partition realtime tables.");
}
}
+
+ /// Parses JSON text the way [PinotDataType]'s JSON conversion does
(preserving `BigDecimal` floats so the
+ /// canonical string is byte-identical), or returns `null` if the text is
not well-formed JSON so the caller
+ /// falls back to the normal conversion (which wraps a non-JSON string as a
JSON string).
+ @Nullable
+ private static JsonNode tryParseJson(String jsonText) {
+ try {
+ return JsonUtils.stringToJsonNodeWithBigDecimal(jsonText);
+ } catch (IOException e) {
+ // Not well-formed JSON; let the caller fall back to the normal
conversion (which wraps it as a JSON string).
+ return null;
+ }
+ }
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/SanitizationTransformer.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/SanitizationTransformer.java
index a71f4aab5a8..a3feefb019f 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/SanitizationTransformer.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/recordtransformer/SanitizationTransformer.java
@@ -72,11 +72,11 @@ public class SanitizationTransformer implements
RecordTransformer {
SanitizedColumnInfo info = entry.getValue();
Object value = record.getValue(columnName);
SanitizationResult result =
SanitizationTransformerUtils.sanitizeValue(info, value);
- if (result != null) {
+ // An unsanitized value is unchanged (multi-value arrays are updated in
place), so it is not written back, which
+ // would also drop a parsed JSON value cached for the column.
+ if (result != null && result.isSanitized()) {
record.putValue(columnName, result.getValue());
- if (result.isSanitized()) {
- record.markSanitized();
- }
+ record.markSanitized();
}
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java
index 044728c3ad8..bb65e6bedf7 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java
@@ -26,11 +26,13 @@ import java.util.Set;
import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
+import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
import org.apache.pinot.spi.config.table.JsonIndexConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.stream.StreamMessageMetadata;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -137,4 +139,40 @@ public class IndexingFailureTest implements
PinotBuffersAfterMethodCheckRule {
verify(_serverMetrics,
times(1)).addMeteredTableValue(matches("DICTIONARY-indexingError$"),
eq(ServerMeter.INDEXING_FAILURES), eq(1L));
}
+
+ @Test
+ public void testJsonIndexUsesParsedCache()
+ throws IOException {
+ GenericRow row = new GenericRow();
+ row.putValue(INT_COL, 0);
+ row.putValue(STRING_COL, "a");
+ row.putValue(JSON_COL, "not valid json");
+ row.putParsedJsonValue(JSON_COL,
JsonUtils.stringToJsonNode("{\"cached\":\"value\"}"));
+
+ _mutableSegment.index(row, METADATA);
+
+
assertEquals(_mutableSegment.getDataSource(JSON_COL).getJsonIndex().getMatchingDocIds("cached
= 'value'"),
+ ImmutableRoaringBitmap.bitmapOf(0));
+ verify(_serverMetrics,
never()).addMeteredTableValue(matches("indexingError$"),
eq(ServerMeter.INDEXING_FAILURES),
+ anyLong());
+ }
+
+ @Test
+ public void testJsonIndexUsesForwardValueForUnsupportedParsedCache()
+ throws IOException {
+ GenericRow row = new GenericRow();
+ row.putValue(INT_COL, 0);
+ row.putValue(STRING_COL, "a");
+ row.putValue(JSON_COL, "{\"forward\":\"value\"}");
+ // A Float leaf cannot be flattened directly, so the index flattens the
forward value instead.
+ row.putParsedJsonValue(JSON_COL, Map.of("cached", 0.5f));
+
+ _mutableSegment.index(row, METADATA);
+
+ JsonIndexReader jsonIndex =
_mutableSegment.getDataSource(JSON_COL).getJsonIndex();
+ assertEquals(jsonIndex.getMatchingDocIds("forward = 'value'"),
ImmutableRoaringBitmap.bitmapOf(0));
+ assertEquals(jsonIndex.getMatchingDocIds("cached = '0.5'"),
ImmutableRoaringBitmap.bitmapOf());
+ verify(_serverMetrics,
never()).addMeteredTableValue(matches("indexingError$"),
eq(ServerMeter.INDEXING_FAILURES),
+ anyLong());
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformerTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformerTest.java
index 48dc788ce6e..63ed4889e3c 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformerTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/recordtransformer/DataTypeTransformerTest.java
@@ -18,30 +18,162 @@
*/
package org.apache.pinot.segment.local.recordtransformer;
+import com.fasterxml.jackson.core.StreamReadConstraints;
+import com.fasterxml.jackson.databind.JsonNode;
+import java.math.BigInteger;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.pinot.segment.local.utils.DataTypeTransformerUtils;
import org.apache.pinot.spi.config.table.DedupConfig;
+import org.apache.pinot.spi.config.table.JsonIndexConfig;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.config.table.UpsertConfig;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertEqualsNoOrder;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
public class DataTypeTransformerTest {
private static final String COLUMN = "testColumn";
+ /// The JSON parse-once cache: for a JSON column with a JSON index, the
DataTypeTransformer caches the parsed Map on
+ /// the row (so the index can flatten it directly) and still serializes the
column value to a string. For a JSON
+ /// column without a JSON index, nothing is cached.
+ @Test
+ public void testJsonParseOnceCache()
+ throws Exception {
+ Schema schema = new
Schema.SchemaBuilder().addSingleValueDimension("jsonCol",
FieldSpec.DataType.JSON).build();
+ Map<String, Object> doc = new HashMap<>();
+ doc.put("a", 1);
+ doc.put("b", "x");
+
+ TableConfig withIndex = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t")
+ .setJsonIndexColumns(List.of("jsonCol")).build();
+ GenericRow indexed = new GenericRow();
+ indexed.putValue("jsonCol", doc);
+ new DataTypeTransformer(withIndex, schema).transform(indexed);
+ // The column value is serialized for the forward index...
+ assertTrue(indexed.getValue("jsonCol") instanceof String);
+ // ...and the parsed Map is cached for the JSON index to flatten directly.
+ assertEquals(indexed.getParsedJsonValue("jsonCol"), doc);
+ // JSON decoders store arrays nested in maps as Object[]. They remain
eligible for the parsed path.
+ GenericRow arrays = new GenericRow();
+ arrays.putValue("jsonCol", Map.of("array", new Object[]{Map.of("n", 1),
2}));
+ new DataTypeTransformer(withIndex, schema).transform(arrays);
+ String arraysForwardValue = (String) arrays.getValue("jsonCol");
+ JsonIndexConfig config = new JsonIndexConfig();
+ assertEquals(JsonUtils.flattenParsed(arrays.getParsedJsonValue("jsonCol"),
arraysForwardValue, config),
+ JsonUtils.flatten(arraysForwardValue, config));
+
+ // Avro-style leaves, and values exceeding the string path's parser
limits, are cached unchanged: the JSON index
+ // classifies the value and flattens the forward string instead, without
another tree conversion or serialization.
+ BigInteger longInteger = new
BigInteger("9".repeat(StreamReadConstraints.defaults().getMaxNumberLength() +
1));
+ for (Object unsafe : new Object[]{0.1f, new byte[]{1, 2, 3}, longInteger})
{
+ GenericRow fallback = new GenericRow();
+ fallback.putValue("jsonCol", Map.of("value", unsafe));
+ new DataTypeTransformer(withIndex, schema).transform(fallback);
+ assertEquals(fallback.getParsedJsonValue("jsonCol"), Map.of("value",
unsafe));
+ GenericRow reference = new GenericRow();
+ reference.putValue("jsonCol", Map.of("value", unsafe));
+ new DataTypeTransformer(new
TableConfigBuilder(TableType.OFFLINE).setTableName("t").build(), schema)
+ .transform(reference);
+ assertEquals(fallback.getValue("jsonCol"),
reference.getValue("jsonCol"));
+ }
+
+ // No JSON index on the column -> no caching (the index, if any, would
re-parse as before).
+ TableConfig noIndex = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t").build();
+ GenericRow unindexed = new GenericRow();
+ unindexed.putValue("jsonCol", doc);
+ new DataTypeTransformer(noIndex, schema).transform(unindexed);
+ assertTrue(unindexed.getValue("jsonCol") instanceof String);
+ assertNull(unindexed.getParsedJsonValue("jsonCol"));
+ }
+
+ /// The common case: a JSON column is fed as JSON text (e.g. a stringified
log payload). With a JSON index, the
+ /// transformer parses it once, caches the JsonNode for the index, and the
forward-index string is byte-identical to
+ /// the normal conversion. A non-JSON string falls through to the normal
conversion with nothing cached.
+ @Test
+ public void testJsonParseOnceCacheFromString() {
+ Schema schema = new
Schema.SchemaBuilder().addSingleValueDimension("jsonCol",
FieldSpec.DataType.JSON).build();
+ TableConfig withIndex = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t")
+ .setJsonIndexColumns(List.of("jsonCol")).build();
+ TableConfig noIndex = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t").build();
+
+ // Includes a fractional number (BigDecimal canonicalization) and an
integer to cover both flatten paths.
+ for (String json : new String[]{"{\"b\":\"x\",\"a\":1}",
"{\"bd\":1234567890.5,\"n\":42}"}) {
+ // Reference: the same column without a JSON index serializes via the
normal PinotDataType conversion.
+ GenericRow reference = new GenericRow();
+ reference.putValue("jsonCol", json);
+ new DataTypeTransformer(noIndex, schema).transform(reference);
+ assertNull(reference.getParsedJsonValue("jsonCol"));
+
+ GenericRow row = new GenericRow();
+ row.putValue("jsonCol", json);
+ new DataTypeTransformer(withIndex, schema).transform(row);
+ // Forward-index value is byte-identical to the normal conversion...
+ assertEquals(row.getValue("jsonCol"), reference.getValue("jsonCol"));
+ // ...and the parsed node is cached for the JSON index to flatten
directly.
+ assertTrue(row.getParsedJsonValue("jsonCol") instanceof JsonNode);
+ }
+
+ // A non-JSON string is not cached and is serialized exactly as the normal
conversion would.
+ GenericRow nonJsonReference = new GenericRow();
+ nonJsonReference.putValue("jsonCol", "not json");
+ new DataTypeTransformer(noIndex, schema).transform(nonJsonReference);
+ GenericRow nonJson = new GenericRow();
+ nonJson.putValue("jsonCol", "not json");
+ new DataTypeTransformer(withIndex, schema).transform(nonJson);
+ assertEquals(nonJson.getValue("jsonCol"),
nonJsonReference.getValue("jsonCol"));
+ assertNull(nonJson.getParsedJsonValue("jsonCol"));
+ }
+
+ /// A transformer that runs after DataTypeTransformer and rewrites a JSON
column's value must not leave a stale parsed
+ /// node cached: SanitizationTransformer trims an over-length JSON string,
which has to invalidate the cache so the
+ /// JSON index re-parses the trimmed forward value instead of flattening the
original (full) document.
+ @Test
+ public void testParsedJsonCacheInvalidatedBySanitization() {
+ Schema schema = new
Schema.SchemaBuilder().addSingleValueDimension("jsonCol",
FieldSpec.DataType.JSON).build();
+ FieldSpec jsonSpec = schema.getFieldSpecFor("jsonCol");
+ jsonSpec.setMaxLength(8);
+
jsonSpec.setMaxLengthExceedStrategy(FieldSpec.MaxLengthExceedStrategy.TRIM_LENGTH);
+ TableConfig withIndex = new
TableConfigBuilder(TableType.OFFLINE).setTableName("t")
+ .setJsonIndexColumns(List.of("jsonCol")).build();
+
+ GenericRow row = new GenericRow();
+ row.putValue("jsonCol", "{\"a\":\"averylongvalue\",\"b\":2}");
+ new DataTypeTransformer(withIndex, schema).transform(row);
+ // After type transformation the full parsed node is cached for the JSON
index.
+ assertTrue(row.getParsedJsonValue("jsonCol") instanceof JsonNode);
+
+ // Sanitization trims the forward-index string; the stale node must be
dropped so the index re-parses the trimmed
+ // value (which the forward index now holds) rather than flattening the
full document.
+ new SanitizationTransformer(schema).transform(row);
+ assertTrue(((String) row.getValue("jsonCol")).length() <= 8);
+ assertNull(row.getParsedJsonValue("jsonCol"));
+
+ // A value within the max length is left unchanged, so its cached node
survives sanitization.
+ GenericRow shortRow = new GenericRow();
+ shortRow.putValue("jsonCol", "{\"a\":1}");
+ new DataTypeTransformer(withIndex, schema).transform(shortRow);
+ new SanitizationTransformer(schema).transform(shortRow);
+ assertEquals(shortRow.getValue("jsonCol"), "{\"a\":1}");
+ assertTrue(shortRow.getParsedJsonValue("jsonCol") instanceof JsonNode);
+ assertFalse(shortRow.isSanitized());
+ }
+
@Test
public void testStandardize() {
/// Tests for Map
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
index 2a4b9e88188..477e0d993a2 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/JsonIndexTest.java
@@ -19,11 +19,16 @@
package org.apache.pinot.segment.local.segment.index;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.StreamReadConstraints;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -32,6 +37,7 @@ import org.apache.pinot.common.metrics.ServerMeter;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.segment.local.PinotBuffersAfterMethodCheckRule;
import org.apache.pinot.segment.local.realtime.impl.json.MutableJsonIndexImpl;
+import org.apache.pinot.segment.local.recordtransformer.DataTypeTransformer;
import
org.apache.pinot.segment.local.segment.creator.impl.inv.json.OffHeapJsonIndexCreator;
import
org.apache.pinot.segment.local.segment.creator.impl.inv.json.OnHeapJsonIndexCreator;
import org.apache.pinot.segment.local.segment.index.json.JsonIndexType;
@@ -42,7 +48,13 @@ import
org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.JsonIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
+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.utils.JsonUtils;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.roaringbitmap.RoaringBitmap;
import org.roaringbitmap.buffer.ImmutableRoaringBitmap;
import org.testng.annotations.AfterMethod;
@@ -96,6 +108,153 @@ public class JsonIndexTest implements
PinotBuffersAfterMethodCheckRule {
FileUtils.deleteDirectory(INDEX_DIR);
}
+ /// The realtime parse-once path must build an identical index: indexing
each document via `addParsed(Map, String)`
+ /// (the cache path) must yield the same matching docIds as `add(String)`
(the re-parse path) for every filter.
+ @Test
+ public void testMutableJsonIndexParsedMatchesString()
+ throws Exception {
+ String[] docs = {
+
"{\"name\":\"adam\",\"age\":20,\"addresses\":[{\"country\":\"us\"},{\"country\":\"ca\"}]}",
+ "{\"name\":\"bob\",\"age\":30,\"tags\":[\"x\",\"y\"],\"active\":true}",
+ "{\"name\":\"carol\",\"nested\":{\"a\":1,\"b\":\"z\"}}"
+ };
+ JsonIndexConfig config = new JsonIndexConfig();
+ try (MutableJsonIndexImpl fromString = new MutableJsonIndexImpl(config,
"table__0__0", "col");
+ MutableJsonIndexImpl fromParsed = new MutableJsonIndexImpl(config,
"table__0__0", "col")) {
+ for (String doc : docs) {
+ fromString.add(doc);
+ fromParsed.addParsed(JsonUtils.stringToObject(doc, Map.class), doc);
+ }
+ for (String filter : new String[]{
+ "name='adam'", "name='bob'", "name='carol'", "age='20'", "age='30'",
"active='true'", "name='nobody'"}) {
+ assertEquals(fromParsed.getMatchingDocIds(filter),
fromString.getMatchingDocIds(filter), "filter: " + filter);
+ }
+ }
+ }
+
+ @Test
+ public void testMutableJsonIndexCachedMapMatchesForwardValue()
+ throws Exception {
+ Map<String, Object> collision = new LinkedHashMap<>();
+ collision.put("a.b", 2);
+ collision.put("a", Map.of("b", 1));
+ List<Map<String, Object>> docs = List.of(Map.of("a", new
BigDecimal("2.0")),
+ Map.of("a", new BigDecimal("123456789012345678900")), collision,
Map.of("nested", collision));
+ Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("col",
DataType.JSON).build();
+ TableConfig withIndex = new
TableConfigBuilder(TableType.REALTIME).setTableName("table")
+ .setJsonIndexColumns(List.of("col")).build();
+ TableConfig noIndex = new
TableConfigBuilder(TableType.REALTIME).setTableName("table").build();
+ DataTypeTransformer cachedTransformer = new DataTypeTransformer(withIndex,
schema);
+ DataTypeTransformer referenceTransformer = new
DataTypeTransformer(noIndex, schema);
+ JsonIndexConfig config = new JsonIndexConfig();
+ String[] forwardValues = new String[docs.size()];
+ try (MutableJsonIndexImpl fromParsed = new MutableJsonIndexImpl(config,
"table__0__0", "col")) {
+ for (int i = 0; i < docs.size(); i++) {
+ GenericRow reference = new GenericRow();
+ reference.putValue("col", docs.get(i));
+ referenceTransformer.transform(reference);
+ forwardValues[i] = (String) reference.getValue("col");
+ GenericRow cached = new GenericRow();
+ cached.putValue("col", docs.get(i));
+ cachedTransformer.transform(cached);
+ assertEquals(cached.getValue("col"), forwardValues[i]);
+ assertNotNull(cached.getParsedJsonValue("col"));
+ fromParsed.addParsed(cached.getParsedJsonValue("col"),
forwardValues[i]);
+ }
+ createIndex(true, config, forwardValues);
+ File indexFile = new File(INDEX_DIR, ON_HEAP_COLUMN_NAME +
V1Constants.Indexes.JSON_INDEX_FILE_EXTENSION);
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(indexFile);
+ JsonIndexReader fromString = new ImmutableJsonIndexReader(buffer,
docs.size())) {
+ Map<String, int[]> expected = Map.of("a='2.0'", ids(0), "a='2'",
empty(),
+ "a='123456789012345678900'", ids(1), "\"a.b\"='2'", ids(2),
"\"a.b\"='1'", empty(),
+ "\"nested.a.b\"='2'", ids(3), "\"nested.a.b\"='1'", empty());
+ for (Map.Entry<String, int[]> entry : expected.entrySet()) {
+ assertDocIds(fromString, entry.getKey(), entry.getValue());
+ assertEquals(fromParsed.getMatchingDocIds(entry.getKey()),
fromString.getMatchingDocIds(entry.getKey()),
+ "filter: " + entry.getKey());
+ }
+ }
+ }
+ }
+
+ /// A document whose forward value the string path cannot parse (here a
name, an integer, or a decimal longer than
+ /// Jackson's limits) must not be indexed from its cached parsed value: with
`skipInvalidJson`, the immutable index
+ /// built from the forward value after sealing skips the whole document, so
the consuming index must skip it as well.
+ @Test
+ public void testMutableJsonIndexCachedValueRespectsParserConstraints()
+ throws Exception {
+ StreamReadConstraints constraints = StreamReadConstraints.defaults();
+ String longName = "k".repeat(constraints.getMaxNameLength() + 1);
+ String longInteger = "9".repeat(constraints.getMaxNumberLength() + 1);
+ List<Object> docs = List.of(Map.of("ok", "v", longName, "v"), Map.of("ok",
"v", "n", new BigInteger(longInteger)),
+ JsonNodeFactory.instance.objectNode().put("ok", "v").put(longName,
"v"),
+ Map.of("ok", "v", "d", new BigDecimal("1." +
longInteger.substring(1))), Map.of("ok", "v"));
+ Schema schema = new Schema.SchemaBuilder().addSingleValueDimension("col",
DataType.JSON).build();
+ TableConfig withIndex = new
TableConfigBuilder(TableType.REALTIME).setTableName("table")
+ .setJsonIndexColumns(List.of("col")).build();
+ TableConfig noIndex = new
TableConfigBuilder(TableType.REALTIME).setTableName("table").build();
+ DataTypeTransformer cachedTransformer = new DataTypeTransformer(withIndex,
schema);
+ DataTypeTransformer referenceTransformer = new
DataTypeTransformer(noIndex, schema);
+ JsonIndexConfig config = new JsonIndexConfig();
+ config.setSkipInvalidJson(true);
+ String[] forwardValues = new String[docs.size()];
+ try (MutableJsonIndexImpl consuming = new MutableJsonIndexImpl(config,
"table__0__0", "col")) {
+ for (int i = 0; i < docs.size(); i++) {
+ GenericRow reference = new GenericRow();
+ reference.putValue("col", docs.get(i));
+ referenceTransformer.transform(reference);
+ forwardValues[i] = (String) reference.getValue("col");
+ GenericRow cached = new GenericRow();
+ cached.putValue("col", docs.get(i));
+ cachedTransformer.transform(cached);
+ assertEquals(cached.getValue("col"), forwardValues[i]);
+ // Like MutableSegmentImpl, index the cached value along with the
forward value.
+ Object parsedValue = cached.getParsedJsonValue("col");
+ assertNotNull(parsedValue, "doc: " + i);
+ consuming.addParsed(parsedValue, forwardValues[i]);
+ }
+ createIndex(true, config, forwardValues);
+ File indexFile = new File(INDEX_DIR, ON_HEAP_COLUMN_NAME +
V1Constants.Indexes.JSON_INDEX_FILE_EXTENSION);
+ try (PinotDataBuffer buffer =
PinotDataBuffer.mapReadOnlyBigEndianFile(indexFile);
+ JsonIndexReader sealed = new ImmutableJsonIndexReader(buffer,
docs.size())) {
+ assertDocIds(sealed, "ok='v'", ids(4));
+ for (String filter : new String[]{"ok='v'", "n='" + longInteger +
"'"}) {
+ assertEquals(consuming.getMatchingDocIds(filter),
sealed.getMatchingDocIds(filter), "filter: " + filter);
+ }
+ Map<String, RoaringBitmap> skipped = getMatchingDocsMap(sealed, "$");
+ assertEquals(skipped, Map.of(JsonUtils.SKIPPED_VALUE_REPLACEMENT,
RoaringBitmap.bitmapOf(0, 1, 2, 3)));
+ assertEquals(consuming.getMatchingFlattenedDocsMap("$", null),
skipped);
+ }
+ }
+ }
+
+ /// The JSON-text fast path: a column fed as JSON text is parsed once into a
JsonNode (with BigDecimal, as
+ /// PinotDataType does) and that node is fed to the index via `addParsed`.
It must yield the same matching docIds
+ /// as re-parsing the serialized string. The docs include a fractional
number to exercise the DecimalNode fallback.
+ @Test
+ public void testMutableJsonIndexParsedNodeMatchesString()
+ throws Exception {
+ String[] docs = {
+
"{\"name\":\"adam\",\"age\":20,\"score\":98.6,\"addresses\":[{\"country\":\"us\"},{\"country\":\"ca\"}]}",
+ "{\"name\":\"bob\",\"age\":30,\"tags\":[\"x\",\"y\"],\"active\":true}",
+ "{\"name\":\"carol\",\"nested\":{\"a\":1,\"b\":\"z\"}}"
+ };
+ JsonIndexConfig config = new JsonIndexConfig();
+ try (MutableJsonIndexImpl fromString = new MutableJsonIndexImpl(config,
"table__0__0", "col");
+ MutableJsonIndexImpl fromParsed = new MutableJsonIndexImpl(config,
"table__0__0", "col")) {
+ for (String doc : docs) {
+ // The forward index stores the canonical string (node.toString());
the index re-parses it on the string path.
+ String canonical =
JsonUtils.stringToJsonNodeWithBigDecimal(doc).toString();
+ fromString.add(canonical);
+ fromParsed.addParsed(JsonUtils.stringToJsonNodeWithBigDecimal(doc),
canonical);
+ }
+ for (String filter : new String[]{
+ "name='adam'", "name='bob'", "name='carol'", "age='20'",
"score='98.6'", "active='true'", "name='nobody'"}) {
+ assertEquals(fromParsed.getMatchingDocIds(filter),
fromString.getMatchingDocIds(filter), "filter: " + filter);
+ }
+ }
+ }
+
@Test
public void testAddRecordWithUnpairedSurrogateDoesNotShiftDocIds()
throws IOException {
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/GenericRow.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/GenericRow.java
index 64221285379..228aee6643f 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/GenericRow.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/data/readers/GenericRow.java
@@ -66,6 +66,11 @@ public class GenericRow implements Serializable {
private final Set<String> _nullValueFields = new HashSet<>();
private boolean _incomplete;
private boolean _sanitized;
+ /// Transient, per-row cache of already-parsed JSON values (column -> {@code
Map}/{@code List}/{@code JsonNode}).
+ /// Populated by the DataTypeTransformer for columns that feed a JSON index,
so the index can flatten the parsed
+ /// value directly instead of re-parsing the string the document was just
serialized into. Not part of the row's
+ /// value, equality, copy, or serialized state - purely an ingestion-time
optimization that is cleared per row.
+ private transient Map<String, Object> _parsedJsonCache;
/// Initializes the generic row from the given generic row (shallow copy).
The row should be new created or cleared
/// before calling this method.
@@ -193,22 +198,52 @@ public class GenericRow implements Serializable {
/// Sets the value for the given field.
public void putValue(String fieldName, @Nullable Object value) {
_fieldToValueMap.put(fieldName, value);
+ invalidateParsedJsonValue(fieldName);
}
/// Sets the values per the given map from fields to values.
public void putValues(Map<String, Object> fieldToValueMap) {
_fieldToValueMap.putAll(fieldToValueMap);
+ if (_parsedJsonCache != null) {
+ _parsedJsonCache.keySet().removeAll(fieldToValueMap.keySet());
+ }
}
/// Removes the value for the given field.
public Object removeValue(String fieldName) {
+ invalidateParsedJsonValue(fieldName);
return _fieldToValueMap.remove(fieldName);
}
+ /// Caches an already-parsed JSON value for the given field so a downstream
JSON index can flatten it directly
+ /// instead of re-parsing the serialized string. See {@link
#_parsedJsonCache}.
+ public void putParsedJsonValue(String fieldName, Object parsedValue) {
+ if (_parsedJsonCache == null) {
+ _parsedJsonCache = new HashMap<>();
+ }
+ _parsedJsonCache.put(fieldName, parsedValue);
+ }
+
+ /// Returns the cached already-parsed JSON value for the given field, or
{@code null} if none was cached.
+ @Nullable
+ public Object getParsedJsonValue(String fieldName) {
+ return _parsedJsonCache != null ? _parsedJsonCache.get(fieldName) : null;
+ }
+
+ /// Drops any cached parsed JSON value for the given field. Invoked whenever
the field's value is overwritten or
+ /// removed so a downstream JSON index never flattens a parsed node that no
longer matches the serialized value (for
+ /// example after a sanitization transformer trims an over-length JSON
string). See {@link #_parsedJsonCache}.
+ private void invalidateParsedJsonValue(String fieldName) {
+ if (_parsedJsonCache != null) {
+ _parsedJsonCache.remove(fieldName);
+ }
+ }
+
/// Sets the `defaultNullValue` for the given `nullField`.
public void putDefaultNullValue(String fieldName, Object defaultNullValue) {
_fieldToValueMap.put(fieldName, defaultNullValue);
_nullValueFields.add(fieldName);
+ invalidateParsedJsonValue(fieldName);
}
/// Marks a field as `null`.
@@ -235,6 +270,9 @@ public class GenericRow implements Serializable {
public void clear() {
_fieldToValueMap.clear();
_nullValueFields.clear();
+ if (_parsedJsonCache != null) {
+ _parsedJsonCache.clear();
+ }
_incomplete = false;
_sanitized = false;
}
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 c6dc75b0298..1fae429f6a2 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
@@ -21,18 +21,22 @@ package org.apache.pinot.spi.utils;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.StreamReadConstraints;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.databind.cfg.JsonNodeFeature;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateSerializer;
import com.fasterxml.jackson.datatype.jsr310.ser.LocalTimeSerializer;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.MapDifference;
import com.google.common.collect.Maps;
@@ -42,6 +46,8 @@ import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.math.BigDecimal;
+import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.time.LocalTime;
@@ -104,6 +110,11 @@ public class JsonUtils {
// time, and the serializers themselves are stateless.
private static final JavaTimeModule JAVA_TIME_MODULE = buildJavaTimeModule();
private static final ObjectMapper DEFAULT_MAPPER =
newObjectMapperWithJavaTime();
+ // Keep decimal tokens intact until normalization, and match the forward
value's MapUtils key ordering.
+ private static final ObjectMapper PARSED_JSON_MAPPER =
+
newObjectMapperWithJavaTime().configure(JsonNodeFeature.STRIP_TRAILING_BIGDECIMAL_ZEROES,
false);
+ private static final ObjectMapper PARSED_JSON_MAP_MAPPER =
+
PARSED_JSON_MAPPER.copy().enable(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS);
public static final ObjectReader DEFAULT_READER = DEFAULT_MAPPER.reader();
public static final ObjectWriter DEFAULT_WRITER = DEFAULT_MAPPER.writer();
public static final ObjectWriter DEFAULT_PRETTY_WRITER =
DEFAULT_MAPPER.writerWithDefaultPrettyPrinter();
@@ -783,6 +794,261 @@ public class JsonUtils {
}
}
+ /// Flattens an already-parsed JSON value ({@link Map} / {@link List} /
{@link JsonNode}) for the JSON index, avoiding
+ /// the string tokenization that {@link #flatten(String, JsonIndexConfig)}
performs on `jsonString`, the value's
+ /// serialized form. Used by the realtime JSON index for the value cached on
the `GenericRow` before it is serialized
+ /// into the forward-index value, so the document is parsed once at
ingestion instead of being serialized and
+ /// re-parsed here. The result matches flattening `jsonString` (Maps
serialize via {@link MapUtils#toString(Map)},
+ /// Lists and JsonNodes via {@link #objectToString(Object)}): a value that
cannot be flattened identically, or whose
+ /// direct conversion fails, flattens `jsonString` instead. Jackson's
hash-flooding defense, which rejects a document
+ /// with many field names colliding in its name table, is not replicated:
such a document is indexed here although
+ /// the string path rejects it (or skips it with `skipInvalidJson`).
+ public static List<Map<String, String>> flattenParsed(Object jsonValue,
String jsonString,
+ JsonIndexConfig jsonIndexConfig)
+ throws IOException {
+ // A single traversal checks the value against the string path's parser
constraints and finds BigDecimal leaves.
+ int classification = classifyParsedValue(jsonValue);
+ if (classification == FLATTEN_UNSAFE) {
+ return flatten(jsonString, jsonIndexConfig);
+ }
+ JsonNode toFlatten;
+ try {
+ JsonNode jsonNode = jsonValue instanceof JsonNode
+ ? (JsonNode) jsonValue
+ : (jsonValue instanceof Map ? PARSED_JSON_MAP_MAPPER :
PARSED_JSON_MAPPER).valueToTree(jsonValue);
+ // A DecimalNode (a BigDecimal float, e.g. from
stringToJsonNodeWithBigDecimal) renders its plain value, but the
+ // string path re-parses its serialized token as an integer or double
(possibly in scientific notation).
+ // Re-parse just those leaves the same way so flatten matches, without
re-tokenizing the
+ // whole document as the serialize+reparse fallback would.
+ toFlatten = classification == FLATTEN_SAFE_WITH_BIG_DECIMAL ?
normalizeBigDecimalNodes(jsonNode) : jsonNode;
+ } catch (Exception e) {
+ // Let the string path decide how to index the document. For example,
its reader accepts a decimal token one
+ // digit longer than the number-length limit at an input buffer
boundary, which the normalization rejects.
+ return flatten(jsonString, jsonIndexConfig);
+ }
+ try {
+ return flatten(toFlatten, jsonIndexConfig);
+ } catch (Exception e) {
+ // The string path flattens the same tree, so it fails the same way.
+ if (jsonIndexConfig.getSkipInvalidJson()) {
+ return SKIPPED_FLATTENED_RECORD;
+ } else {
+ throw e;
+ }
+ }
+ }
+
+ /// Returns whether {@link #flattenParsed} flattens the value directly
rather than its serialized form.
+ @VisibleForTesting
+ static boolean canFlattenParsedValue(@Nullable Object value) {
+ return classifyParsedValue(value) != FLATTEN_UNSAFE;
+ }
+
+ private static final int FLATTEN_UNSAFE = 0;
+ private static final int FLATTEN_SAFE = 1;
+ private static final int FLATTEN_SAFE_WITH_BIG_DECIMAL = 2;
+
+ /// Constraints of the reader {@link #flatten(String, JsonIndexConfig)}
parses with. A parsed value must not bypass
+ /// them, or the consuming index would contain a document that the string
path (e.g. the immutable index built from
+ /// the forward value after the segment is sealed) rejects or skips.
Classification checks nesting depth, name length,
+ /// string length, and integer length; decimal tokens are checked when
{@link #normalizeBigDecimalNodes(JsonNode)}
+ /// re-parses them with the same reader.
+ private static final StreamReadConstraints FLATTEN_READ_CONSTRAINTS =
+ DEFAULT_READER.getFactory().streamReadConstraints();
+ /// The longest number token that an int, long, or double leaf can serialize
to is shorter than this.
+ private static final int MAX_FIXED_SIZE_NUMBER_LENGTH = 32;
+ /// Whether any parsed value can be flattened directly. Classification does
not track document length (which requires
+ /// serializing) or token count, and a number limit shorter than a
primitive's token would require counting every
+ /// number, so such constraints send every value through the string path.
+ private static final boolean PARSED_FLATTEN_SUPPORTED =
+ !FLATTEN_READ_CONSTRAINTS.hasMaxDocumentLength() &&
!FLATTEN_READ_CONSTRAINTS.hasMaxTokenCount()
+ && FLATTEN_READ_CONSTRAINTS.getMaxNumberLength() >=
MAX_FIXED_SIZE_NUMBER_LENGTH;
+ /// Bounds of log10(2) for the digit count bounds in {@link
#isWithinIntegerLength(BigInteger)}.
+ private static final double LOG10_2_LOWER_BOUND = 0.30102;
+ private static final double LOG10_2_UPPER_BOUND = 0.30103;
+
+ /// Classifies whether a parsed value can be flattened directly so the
records are byte-for-byte identical to
+ /// flattening its serialized string (both paths render each leaf via {@link
JsonNode#asText()}). {@code Integer},
+ /// {@code Long}, {@code Double} and {@code BigInteger} leaves render
identically to the string path
+ /// ({@link #FLATTEN_SAFE}); a {@code BigDecimal} leaf renders its plain
value while the string path re-parses it (an
+ /// integer token as an int/long, a floating-point token as a double), so it
is re-parsed the same way first
+ /// ({@link #FLATTEN_SAFE_WITH_BIG_DECIMAL}), preserving scale-zero integers
and tokens like 2.0; e.g.
+ /// {@code "1234567890.5"} -> {@code "1.2345678905E9"}. A {@code Float},
binary, POJO, or other unsupported leaf does
+ /// not round-trip, and the string path would reject a value exceeding the
limits checked against
+ /// {@link #FLATTEN_READ_CONSTRAINTS}; both are flattened from the
serialized form instead ({@link #FLATTEN_UNSAFE}).
+ private static int classifyParsedValue(@Nullable Object value) {
+ return PARSED_FLATTEN_SUPPORTED ? classifyForFlatten(value, 0) :
FLATTEN_UNSAFE;
+ }
+
+ /// Classifies a {@link Map} / {@link List} / {@code Object[]} / {@link
JsonNode} / leaf value nested inside `depth`
+ /// containers. See {@link #classifyParsedValue(Object)}.
+ private static int classifyForFlatten(@Nullable Object value, int depth) {
+ if (value instanceof JsonNode) {
+ return classifyForFlatten((JsonNode) value, depth);
+ }
+ if (value instanceof Map) {
+ return classifyEntriesForFlatten(((Map<?, ?>) value).entrySet(), depth);
+ }
+ if (value instanceof List) {
+ return classifyElementsForFlatten((List<?>) value, depth);
+ }
+ if (value instanceof Object[]) {
+ return classifyElementsForFlatten((Object[]) value, depth);
+ }
+ if (value instanceof String) {
+ return isWithinStringLength((String) value) ? FLATTEN_SAFE :
FLATTEN_UNSAFE;
+ }
+ if (value instanceof BigInteger) {
+ return isWithinIntegerLength((BigInteger) value) ? FLATTEN_SAFE :
FLATTEN_UNSAFE;
+ }
+ if (value instanceof BigDecimal) {
+ return FLATTEN_SAFE_WITH_BIG_DECIMAL;
+ }
+ boolean supportedLeaf = value == null || value instanceof Boolean || value
instanceof Character
+ || value instanceof Byte || value instanceof Short || value instanceof
Integer || value instanceof Long
+ || value instanceof Double;
+ return supportedLeaf ? FLATTEN_SAFE : FLATTEN_UNSAFE;
+ }
+
+ /// Classifies a {@link JsonNode} nested inside `depth` containers. See
{@link #classifyParsedValue(Object)}. A
+ /// {@code DecimalNode} (a {@code BigDecimal} float) needs normalization,
and a {@code FloatNode} or a binary / POJO
+ /// node does not round-trip. A {@link JsonNode} built by a non-JSON
RecordReader may exceed the reader constraints.
+ private static int classifyForFlatten(JsonNode node, int depth) {
+ switch (node.getNodeType()) {
+ case NUMBER:
+ if (node.isFloat()) {
+ return FLATTEN_UNSAFE;
+ }
+ if (node.isBigDecimal()) {
+ return FLATTEN_SAFE_WITH_BIG_DECIMAL;
+ }
+ return !node.isBigInteger() ||
isWithinIntegerLength(node.bigIntegerValue()) ? FLATTEN_SAFE : FLATTEN_UNSAFE;
+ case OBJECT:
+ return classifyEntriesForFlatten(node.properties(), depth);
+ case ARRAY:
+ return classifyElementsForFlatten(node, depth);
+ case STRING:
+ return isWithinStringLength(node.textValue()) ? FLATTEN_SAFE :
FLATTEN_UNSAFE;
+ case BOOLEAN:
+ case NULL:
+ return FLATTEN_SAFE;
+ default:
+ // BINARY, POJO, MISSING
+ return FLATTEN_UNSAFE;
+ }
+ }
+
+ /// Classifies the entries of an object nested inside `depth` containers.
+ private static int classifyEntriesForFlatten(Iterable<? extends Map.Entry<?,
?>> entries, int depth) {
+ if (depth >= FLATTEN_READ_CONSTRAINTS.getMaxNestingDepth()) {
+ return FLATTEN_UNSAFE;
+ }
+ int result = FLATTEN_SAFE;
+ for (Map.Entry<?, ?> entry : entries) {
+ Object key = entry.getKey();
+ if (!(key instanceof String) || ((String) key).length() >
FLATTEN_READ_CONSTRAINTS.getMaxNameLength()) {
+ return FLATTEN_UNSAFE;
+ }
+ int classification = classifyForFlatten(entry.getValue(), depth + 1);
+ if (classification == FLATTEN_UNSAFE) {
+ return FLATTEN_UNSAFE;
+ }
+ result = Math.max(result, classification);
+ }
+ return result;
+ }
+
+ /// Classifies the elements of an array nested inside `depth` containers.
+ private static int classifyElementsForFlatten(Iterable<?> elements, int
depth) {
+ if (depth >= FLATTEN_READ_CONSTRAINTS.getMaxNestingDepth()) {
+ return FLATTEN_UNSAFE;
+ }
+ int result = FLATTEN_SAFE;
+ for (Object element : elements) {
+ int classification = classifyForFlatten(element, depth + 1);
+ if (classification == FLATTEN_UNSAFE) {
+ return FLATTEN_UNSAFE;
+ }
+ result = Math.max(result, classification);
+ }
+ return result;
+ }
+
+ /// Same as {@link #classifyElementsForFlatten(Iterable, int)} for an {@code
Object[]}, without wrapping it in a List.
+ private static int classifyElementsForFlatten(Object[] elements, int depth) {
+ if (depth >= FLATTEN_READ_CONSTRAINTS.getMaxNestingDepth()) {
+ return FLATTEN_UNSAFE;
+ }
+ int result = FLATTEN_SAFE;
+ for (Object element : elements) {
+ int classification = classifyForFlatten(element, depth + 1);
+ if (classification == FLATTEN_UNSAFE) {
+ return FLATTEN_UNSAFE;
+ }
+ result = Math.max(result, classification);
+ }
+ return result;
+ }
+
+ private static boolean isWithinStringLength(String value) {
+ return value.length() <= FLATTEN_READ_CONSTRAINTS.getMaxStringLength();
+ }
+
+ /// Returns whether the string path can parse the integer token. Like the
parser, this counts digits without the sign.
+ private static boolean isWithinIntegerLength(BigInteger value) {
+ int maxLength = FLATTEN_READ_CONSTRAINTS.getMaxNumberLength();
+ // The magnitude is in [2^(bitLength - 1), 2^(bitLength + 1)), so only
values near the limit need their digits
+ // counted.
+ int bitLength = value.bitLength();
+ if ((bitLength + 1) * LOG10_2_UPPER_BOUND + 1 <= maxLength) {
+ return true;
+ }
+ if ((bitLength - 1) * LOG10_2_LOWER_BOUND >= maxLength) {
+ return false;
+ }
+ return value.toString().length() - (value.signum() < 0 ? 1 : 0) <=
maxLength;
+ }
+
+ /// Returns a copy of the tree with every {@code DecimalNode} re-parsed the
way the string path does (serialize via
+ /// {@code toString}, parse with the default reader), so {@link #flatten}'s
{@code asText()} is identical -- whether
+ /// the value renders as an integer, a double, or in scientific notation --
while sharing all other leaf nodes. Only
+ /// called when the tree actually contains a {@code BigDecimal} node. Using
{@code doubleValue()} instead would be
+ /// wrong: the serialized token determines whether the result is an integer
or double, and integer tokens past
+ /// 2^53 must retain their precision. The re-parse also applies the reader's
number-length limit, so a token the
+ /// string path rejects fails here as well.
+ private static JsonNode normalizeBigDecimalNodes(JsonNode node)
+ throws IOException {
+ // A decimal document is a bare token in the forward value, which the
string path parses at the end of the input.
+ return node.isBigDecimal() ? stringToJsonNode(node.toString()) :
normalizeNestedBigDecimalNodes(node);
+ }
+
+ private static JsonNode normalizeNestedBigDecimalNodes(JsonNode node)
+ throws IOException {
+ if (node.isBigDecimal()) {
+ // Parse a token near the number-length limit inside a container, as in
the forward value: at the end of the
+ // input, Jackson accepts a plain decimal token one digit longer than
the limit.
+ String token = node.toString();
+ return token.length() <= FLATTEN_READ_CONSTRAINTS.getMaxNumberLength()
+ ? stringToJsonNode(token)
+ : stringToJsonNode("[" + token + "]").get(0);
+ }
+ if (node.isObject()) {
+ ObjectNode result = JsonNodeFactory.instance.objectNode();
+ for (Map.Entry<String, JsonNode> field : node.properties()) {
+ result.set(field.getKey(),
normalizeNestedBigDecimalNodes(field.getValue()));
+ }
+ return result;
+ }
+ if (node.isArray()) {
+ ArrayNode result = JsonNodeFactory.instance.arrayNode(node.size());
+ for (JsonNode child : node) {
+ result.add(normalizeNestedBigDecimalNodes(child));
+ }
+ return result;
+ }
+ return node;
+ }
+
/// Generates the JsonSchemaTreeNode tree from the given json index config
indexPaths to represent which path we
/// should flatten/index in the json record.
/// @param jsonIndexConfig
diff --git
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/GenericRowTest.java
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/GenericRowTest.java
index 5f57798e1c9..da9cb52aa70 100644
---
a/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/GenericRowTest.java
+++
b/pinot-spi/src/test/java/org/apache/pinot/spi/data/readers/GenericRowTest.java
@@ -168,4 +168,40 @@ public class GenericRowTest {
second.putDefaultNullValue("one", null);
Assert.assertEquals(first, second);
}
+
+ @Test
+ public void testParsedJsonValueInvalidatedOnValueChange() {
+ // The parse-once cache must not outlive the value it mirrors: overwriting
or removing a field drops its cached
+ // parsed JSON node, so a downstream JSON index never flattens content the
serialized value no longer holds.
+ GenericRow row = new GenericRow();
+ row.putValue("j", "{\"a\":1}");
+ row.putParsedJsonValue("j", "parsed-node");
+ Assert.assertNotNull(row.getParsedJsonValue("j"));
+
+ // putValue (e.g. a sanitization transformer trimming the JSON string)
invalidates the now-stale node.
+ row.putValue("j", "{\"a\":");
+ Assert.assertNull(row.getParsedJsonValue("j"));
+
+ // putValues invalidates the overwritten keys.
+ row.putParsedJsonValue("j", "parsed-node");
+ HashMap<String, Object> values = new HashMap<>();
+ values.put("j", "x");
+ row.putValues(values);
+ Assert.assertNull(row.getParsedJsonValue("j"));
+
+ // putDefaultNullValue invalidates.
+ row.putParsedJsonValue("j", "parsed-node");
+ row.putDefaultNullValue("j", "default");
+ Assert.assertNull(row.getParsedJsonValue("j"));
+
+ // removeValue invalidates.
+ row.putParsedJsonValue("k", "parsed-node");
+ row.removeValue("k");
+ Assert.assertNull(row.getParsedJsonValue("k"));
+
+ // clear() drops everything for row reuse.
+ row.putParsedJsonValue("m", "parsed-node");
+ row.clear();
+ Assert.assertNull(row.getParsedJsonValue("m"));
+ }
}
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 bae79fa8c8a..3ed1eab1c12 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
@@ -18,16 +18,22 @@
*/
package org.apache.pinot.spi.utils;
+import com.fasterxml.jackson.core.StreamReadConstraints;
import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
import com.google.common.collect.ImmutableSet;
import java.io.File;
import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.BigInteger;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.LocalDate;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -43,13 +49,301 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertEqualsNoOrder;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
public class JsonUtilsTest {
private static final String JSON_FILE = "json_util_test.json";
+ /// The parse-once index path: `flattenParsed(parsedValue)` must produce the
same flattened records as the old
+ /// serialize-then-reparse path (the records are a set, so order is
ignored). Covers diverse leaf types - including
+ /// non-Jackson types (Float / BigInteger / BigDecimal / large long) that a
non-JSON RecordReader could place on a
+ /// JSON column - plus nested objects/arrays, a top-level list, null, and
empty containers, across maxLevels.
+ @Test
+ public void testFlattenParsedValueMatchesString()
+ throws Exception {
+ Map<String, Object> nested = new HashMap<>();
+ nested.put("a", "x");
+ nested.put("arr", Arrays.asList(1, 2, 3));
+ Map<String, Object> doc = new HashMap<>();
+ doc.put("s", "line 1\nline 2 \"quoted\" é");
+ doc.put("i", 7);
+ doc.put("lng", 9007199254740993L);
+ doc.put("f", 0.1f);
+ doc.put("d", 3.140625);
+ doc.put("bd", new BigDecimal("1234567890.5"));
+ doc.put("bi", new BigInteger("123456789012345678901234567890"));
+ doc.put("bool", true);
+ doc.put("nul", null);
+ doc.put("obj", nested);
+ doc.put("arrObj", List.of(Map.of("k", "v1"), Map.of("k", "v2")));
+ doc.put("empty", new HashMap<>());
+ // JSON-column Map inputs use MapUtils' canonical key ordering for the
forward value.
+ String oldPathInput = MapUtils.toString(doc);
+
+ for (int maxLevels : new int[]{-1, 1, 2, 3}) {
+ JsonIndexConfig config = new JsonIndexConfig();
+ config.setMaxLevels(maxLevels);
+ List<Map<String, String>> fromParsed = JsonUtils.flattenParsed(doc,
oldPathInput, config);
+ List<Map<String, String>> fromString = JsonUtils.flatten(oldPathInput,
config);
+ assertEqualsNoOrder(fromParsed.toArray(), fromString.toArray(),
+ "flattenParsed(Map) must match serialize+reparse at maxLevels=" +
maxLevels);
+ }
+
+ // Top-level list input (the cache stores List values too).
+ List<Object> topList = List.of(Map.of("k", "a"), Map.of("k", "b"));
+ JsonIndexConfig config = new JsonIndexConfig();
+ assertTrue(JsonUtils.canFlattenParsedValue(topList));
+ assertEqualsNoOrder(flattenParsed(topList, config).toArray(),
+ JsonUtils.flatten(JsonUtils.objectToString(topList),
config).toArray());
+ }
+
+ @Test
+ public void testFlattenParsedPreservesDecimalRepresentationAndMapOrder()
+ throws Exception {
+ Map<String, Object> collision = new LinkedHashMap<>();
+ collision.put("a.b", 2);
+ collision.put("a", Map.of("b", 1));
+ Map<String, Object> decimals = Map.of("scaled", new BigDecimal("2.0"),
+ "large", new BigDecimal("123456789012345678900"),
+ "array", new Object[]{new BigDecimal("0.0100"), Map.of("n", new
BigDecimal("9007199254740993.0"))});
+ JsonIndexConfig config = new JsonIndexConfig();
+ for (Map<String, Object> doc : List.<Map<String, Object>>of(decimals,
collision, Map.of("nested", collision))) {
+ assertTrue(JsonUtils.canFlattenParsedValue(doc));
+ assertEqualsNoOrder(flattenParsed(doc, config).toArray(),
+ JsonUtils.flatten(MapUtils.toString(doc), config).toArray());
+ }
+ // A List forward value preserves map insertion order instead of using
MapUtils' sorted writer.
+ List<Object> list = List.of(collision, decimals);
+ assertTrue(JsonUtils.canFlattenParsedValue(list));
+ assertEqualsNoOrder(flattenParsed(list, config).toArray(),
+ JsonUtils.flatten(JsonUtils.objectToString(list), config).toArray());
+ }
+
+ @Test
+ public void testFlattenParsedJsonNodeMatchesString()
+ throws Exception {
+ // A JSON column fed as text (the common case) is parsed once and the
JsonNode is cached for the JSON index. The
+ // node is parsed with BigDecimal (matching PinotDataType's JSON
canonicalization), so flattenParsed(node) must
+ // match flatten(node.toString()) -- the string the forward index stores
and the index used to re-parse.
+ String noDecimal = "{\"s\":\"a\\nb
\\\"q\\\"\",\"i\":7,\"lng\":9007199254740993,"
+ + "\"bi\":123456789012345678901234567890,\"bool\":true,\"nul\":null,"
+ +
"\"obj\":{\"a\":\"x\",\"arr\":[1,2,3]},\"arr\":[{\"k\":\"v1\"},{\"k\":\"v2\"}]}";
+ // Numbers parse to DecimalNode under BigDecimal, which can render
differently than the re-parsed string path: a
+ // fractional value as a double (e.g. "1234567890.5" -> "1.2345678905E9")
and an INTEGRAL-valued decimal as an
+ // integer ("2.0" -> "2", not "2.0"; past 2^53 the integer must be
preserved exactly). flattenParsed re-parses each
+ // DecimalNode the same way, so it must still match -- across integral,
large-magnitude, tiny, high-precision,
+ // exponent, negative, >2^53, nested and array values.
+ String withDecimal =
"{\"f\":0.1,\"d\":3.140625,\"bd\":1234567890.5,\"tiny\":0.0000001,"
+ +
"\"prec\":3.141592653589793238462643383279,\"neg\":-9876543210.125,\"exp\":1.5e10,"
+ +
"\"whole\":2.0,\"negWhole\":-5.0,\"zero\":0.0,\"expWhole\":1.5e1,\"bigWhole\":9007199254740993.0,"
+ +
"\"arr\":[0.5,123456.789,2,1234567890.5,7.0,-100.00],\"nested\":{\"x\":98765.4321,\"w\":42.0},"
+ + "\"i\":42,\"s\":\"hi\"}";
+ for (String json : new String[]{noDecimal, withDecimal}) {
+ JsonNode node = JsonUtils.stringToJsonNodeWithBigDecimal(json);
+ assertTrue(JsonUtils.canFlattenParsedValue(node));
+ for (int maxLevels : new int[]{-1, 1, 2, 3}) {
+ JsonIndexConfig config = new JsonIndexConfig();
+ config.setMaxLevels(maxLevels);
+ assertEqualsNoOrder(JsonUtils.flattenParsed(node, node.toString(),
config).toArray(),
+ JsonUtils.flatten(node.toString(), config).toArray(),
+ "flattenParsed(JsonNode) must match flatten(node.toString()) for "
+ json + " at maxLevels=" + maxLevels);
+ }
+ }
+ }
+
+ /// A parsed value must not bypass the parser constraints of the string
path; otherwise the consuming index would
+ /// differ from the immutable index built from the forward value (e.g. a
matching document vs. a skipped one with
+ /// `skipInvalidJson`). Values at the parser limits stay on the parsed path.
+ @Test
+ public void testFlattenParsedRespectsParserConstraints()
+ throws Exception {
+ StreamReadConstraints constraints = StreamReadConstraints.defaults();
+ String maxName = "k".repeat(constraints.getMaxNameLength());
+ String maxString = "s".repeat(constraints.getMaxStringLength());
+ String maxDigits = "9".repeat(constraints.getMaxNumberLength());
+ int maxDepth = constraints.getMaxNestingDepth();
+ JsonIndexConfig config = new JsonIndexConfig();
+
+ List<Object> withinLimits = List.of(Map.of(maxName, "v"), Map.of("s",
maxString),
+ Map.of("n", new BigInteger(maxDigits), "neg", new BigInteger("-" +
maxDigits)),
+ JsonNodeFactory.instance.objectNode().put(maxName, new BigInteger("-"
+ maxDigits)),
+ Map.of("d", new BigDecimal("1." + maxDigits.substring(1))));
+ for (int i = 0; i < withinLimits.size(); i++) {
+ Object value = withinLimits.get(i);
+ assertTrue(JsonUtils.canFlattenParsedValue(value), "withinLimits[" + i +
"]");
+ assertEqualsNoOrder(flattenParsed(value, config).toArray(),
+ JsonUtils.flatten(forwardValue(value), config).toArray(),
"withinLimits[" + i + "]");
+ }
+ // Documents nested this deep are only classified: serializing and
flattening them recursively (on either path) can
+ // need more stack than a test thread has.
+ List<Object> deepWithinLimits = List.of(nestedMap(maxDepth),
nestedList(maxDepth), nestedArray(maxDepth),
+ nestedJsonNode(maxDepth), nestedObjectNode(maxDepth), Map.of("node",
nestedJsonNode(maxDepth - 1)));
+ for (int i = 0; i < deepWithinLimits.size(); i++) {
+ assertTrue(JsonUtils.canFlattenParsedValue(deepWithinLimits.get(i)),
"deepWithinLimits[" + i + "]");
+ }
+
+ JsonIndexConfig skipInvalidJson = new JsonIndexConfig();
+ skipInvalidJson.setSkipInvalidJson(true);
+ // A decimal document is a bare token in its forward value, which Jackson
currently accepts with one digit over the
+ // number-length limit at the end of the input. Both paths must agree
either way.
+ JsonNode decimalDocument = JsonNodeFactory.instance.numberNode(new
BigDecimal("1." + maxDigits));
+ assertEquals(flattenParsed(decimalDocument, skipInvalidJson),
+ JsonUtils.flatten(forwardValue(decimalDocument), skipInvalidJson));
+
+ String longName = maxName + "k";
+ String longDigits = maxDigits + "9";
+ List<Object> exceedingLimits = List.of(Map.of("ok", "v", longName, "v"),
Map.of("ok", "v", "s", maxString + "s"),
+ Map.of("ok", "v", "n", new BigInteger(longDigits)), List.of("v", new
BigInteger("-" + longDigits)),
+ new Object[]{Map.of(longName, 1)},
JsonNodeFactory.instance.objectNode().put("ok", "v").put(longName, "v"),
+ JsonNodeFactory.instance.arrayNode().add(maxString + "s"),
+ JsonNodeFactory.instance.objectNode().put("ok", "v").put("n", new
BigInteger(longDigits)),
+ Map.of("node", JsonNodeFactory.instance.objectNode().put(longName,
"v")));
+ for (int i = 0; i < exceedingLimits.size(); i++) {
+ assertFalse(JsonUtils.canFlattenParsedValue(exceedingLimits.get(i)),
"exceedingLimits[" + i + "]");
+ assertSkippedOnBothPaths(exceedingLimits.get(i), config,
skipInvalidJson);
+ }
+
+ // Decimal tokens are checked when the parsed path re-parses them,
including a token that becomes too long when its
+ // BigDecimal is rendered for the forward value.
+ BigDecimal longDecimal = new BigDecimal("1." + maxDigits);
+ List<Object> decimalsExceedingLimits = List.of(Map.of("ok", "v", "d",
longDecimal),
+ JsonNodeFactory.instance.objectNode().put("ok", "v").put("d",
longDecimal),
+ JsonUtils.stringToJsonNodeWithBigDecimal("{\"ok\":\"v\",\"d\":" +
maxDigits.substring(3) + "e5}"));
+ for (int i = 0; i < decimalsExceedingLimits.size(); i++) {
+
assertTrue(JsonUtils.canFlattenParsedValue(decimalsExceedingLimits.get(i)),
"decimalsExceedingLimits[" + i + "]");
+ assertSkippedOnBothPaths(decimalsExceedingLimits.get(i), config,
skipInvalidJson);
+ }
+
+ // Objects nested one level deeper than the limit can still be serialized,
so they are cached, but the string path
+ // rejects their forward value. Build that forward value directly instead
of serializing such a deep document.
+ String deepForwardValue = "{\"a\":".repeat(maxDepth + 1) + "\"v\"" +
"}".repeat(maxDepth + 1);
+ assertEquals(JsonUtils.flatten(deepForwardValue, skipInvalidJson),
JsonUtils.SKIPPED_FLATTENED_RECORD);
+ for (Object value : List.of(nestedMap(maxDepth + 1),
nestedObjectNode(maxDepth + 1))) {
+ assertFalse(JsonUtils.canFlattenParsedValue(value));
+ assertEquals(JsonUtils.flattenParsed(value, deepForwardValue,
skipInvalidJson),
+ JsonUtils.SKIPPED_FLATTENED_RECORD);
+ assertThrows(IOException.class, () -> JsonUtils.flattenParsed(value,
deepForwardValue, config));
+ }
+
+ // An array nested deeper than the limit cannot be serialized, so the
value has no forward value and is not cached.
+ List<Object> tooDeep = List.of(nestedList(maxDepth + 1),
nestedArray(maxDepth + 1), nestedJsonNode(maxDepth + 1),
+ Map.of("node", nestedJsonNode(maxDepth)));
+ for (int i = 0; i < tooDeep.size(); i++) {
+ assertFalse(JsonUtils.canFlattenParsedValue(tooDeep.get(i)), "tooDeep["
+ i + "]");
+ }
+ }
+
+ /// Over 32,768 chars, the string path reads the forward value in chunks,
and a plain decimal token one digit over the
+ /// number-length limit is accepted when it ends a chunk. The parsed path
then defers to the string path instead of
+ /// rejecting the token in its normalization.
+ @Test
+ public void testFlattenParsedDefersToStringPathAtReaderBufferBoundaries()
+ throws Exception {
+ BigDecimal longDecimal = new BigDecimal("1." +
"9".repeat(StreamReadConstraints.defaults().getMaxNumberLength()));
+ JsonIndexConfig skipInvalidJson = new JsonIndexConfig();
+ skipInvalidJson.setSkipInvalidJson(true);
+ int numAccepted = 0;
+ for (int padding = 33_000; padding < 37_000; padding += 7) {
+ Map<String, Object> value = Map.of("a", "x".repeat(padding), "d",
longDecimal, "z", 1);
+ String forwardValue = forwardValue(value);
+ List<Map<String, String>> expected = JsonUtils.flatten(forwardValue,
skipInvalidJson);
+ if (!expected.equals(JsonUtils.SKIPPED_FLATTENED_RECORD)) {
+ numAccepted++;
+ }
+ assertEquals(JsonUtils.flattenParsed(value, forwardValue,
skipInvalidJson), expected, "padding: " + padding);
+ }
+ assertTrue(numAccepted > 0,
+ "The string path no longer accepts the decimal token at reader buffer
boundaries; revisit this fallback");
+ }
+
+ /// The JSON index passes the forward value along with the cached value, so
a value that cannot be flattened directly
+ /// is flattened from that string instead of being serialized again, while a
supported value ignores it.
+ @Test
+ public void testFlattenParsedFallsBackToSerializedForm()
+ throws Exception {
+ JsonIndexConfig config = new JsonIndexConfig();
+ String serializedForm = "{\"serialized\":\"form\"}";
+ List<Object> unsupported = List.of(Map.of("f", 0.1f), Map.of("b", new
byte[]{1, 2, 3}),
+ JsonNodeFactory.instance.objectNode().put("f", 0.1f),
+ Map.of("k".repeat(StreamReadConstraints.defaults().getMaxNameLength()
+ 1), "v"));
+ for (int i = 0; i < unsupported.size(); i++) {
+ assertEquals(JsonUtils.flattenParsed(unsupported.get(i), serializedForm,
config),
+ JsonUtils.flatten(serializedForm, config), "unsupported[" + i + "]");
+ }
+ Map<String, Object> supported = Map.of("a", 1, "d", new BigDecimal("2.0"));
+ assertEquals(JsonUtils.flattenParsed(supported, serializedForm, config),
+ JsonUtils.flatten(MapUtils.toString(supported), config));
+ }
+
+ private static void assertSkippedOnBothPaths(Object value, JsonIndexConfig
config, JsonIndexConfig skipInvalidJson)
+ throws IOException {
+ String forwardValue = forwardValue(value);
+ assertEquals(JsonUtils.flatten(forwardValue, skipInvalidJson),
JsonUtils.SKIPPED_FLATTENED_RECORD);
+ assertEquals(JsonUtils.flattenParsed(value, forwardValue,
skipInvalidJson), JsonUtils.SKIPPED_FLATTENED_RECORD);
+ assertThrows(IOException.class, () -> JsonUtils.flatten(forwardValue,
config));
+ assertThrows(IOException.class, () -> JsonUtils.flattenParsed(value,
forwardValue, config));
+ }
+
+ /// Flattens a parsed value as the realtime JSON index does, along with the
forward value it was serialized into.
+ private static List<Map<String, String>> flattenParsed(Object value,
JsonIndexConfig config)
+ throws IOException {
+ return JsonUtils.flattenParsed(value, forwardValue(value), config);
+ }
+
+ /// Serializes a parsed value the way the forward index does for JSON
columns.
+ @SuppressWarnings("unchecked")
+ private static String forwardValue(Object value)
+ throws IOException {
+ return value instanceof Map ? MapUtils.toString((Map<String, Object>)
value) : JsonUtils.objectToString(value);
+ }
+
+ private static Object nestedMap(int depth) {
+ Object value = "v";
+ for (int i = 0; i < depth; i++) {
+ value = Map.of("a", value);
+ }
+ return value;
+ }
+
+ private static Object nestedList(int depth) {
+ Object value = "v";
+ for (int i = 0; i < depth; i++) {
+ value = List.of(value);
+ }
+ return value;
+ }
+
+ private static Object nestedArray(int depth) {
+ Object value = "v";
+ for (int i = 0; i < depth; i++) {
+ value = new Object[]{value};
+ }
+ return value;
+ }
+
+ private static JsonNode nestedJsonNode(int depth) {
+ JsonNode value = JsonNodeFactory.instance.textNode("v");
+ for (int i = 0; i < depth; i++) {
+ value = i % 2 == 0
+ ? JsonNodeFactory.instance.arrayNode().add(value)
+ : JsonNodeFactory.instance.objectNode().set("a", value);
+ }
+ return value;
+ }
+
+ private static JsonNode nestedObjectNode(int depth) {
+ JsonNode value = JsonNodeFactory.instance.textNode("v");
+ for (int i = 0; i < depth; i++) {
+ value = JsonNodeFactory.instance.objectNode().set("a", value);
+ }
+ return value;
+ }
+
@Test
public void testFlatten()
throws IOException {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
