This is an automated email from the ASF dual-hosted git repository.
raghavyadav01 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 3b760adb71a Optimise the OPEN_STRUCT per-row consumption path, and
follow-up fixes from #18643 review (#19093)
3b760adb71a is described below
commit 3b760adb71a91dfad56de242a5b3803823f3311b
Author: tarun11Mavani <[email protected]>
AuthorDate: Thu Sep 17 01:21:02 2026 +0530
Optimise the OPEN_STRUCT per-row consumption path, and follow-up fixes from
#18643 review (#19093)
* fix(open_struct): address PR #18643 review comments
Review-comment fixes from the storage PR (#18643) that were not included
in the merged version:
- OpenStructColumnSplitter: computeIfAbsent instead of containsKey+put,
Utf8.encodedLength instead of getBytes(UTF_8).length
- MutableOpenStructIndex: volatile copy-on-write Javadoc clarification,
pass PinotDataType directly to tryCoerce instead of converting per call
- MutableKeyColumn: cache PinotDataType destType to avoid repeated
ColumnDataType.fromDataTypeSV().toPinotDataType() conversion
- MutableOpenStructDataSource: snapshot keyColumns map before iterating
- MutableSegmentImpl: remove unused getOpenStructIndex() method
- IndexLoadingConfig: persist OPEN_STRUCT segment metadata so child
configs survive refreshIndexConfigs() dirty rebuilds
* fix(open_struct): add segment-lifecycle hardening for OPEN_STRUCT columns
- ImmutableSegmentImpl: warn log when an OPEN_STRUCT parent's
schema/fieldSpec
is missing or mismatched, so dense/sparse child data silently becomes
unqueryable
- MutableSegmentImpl: checkState guard requiring the open_struct_index to be
enabled when building a DataSource for an OPEN_STRUCT column
- ImmutableOpenStructDataSource: getMapValue()/readValue() to reconstruct
the
full key-value map (dense + sparse) for a doc, used by the seal path and
minion tasks
* fix(open_struct): write full column metadata for sparse child column
writeSparseJsonColumn() hand-built a partial PropertiesConfiguration for
the OPEN_STRUCT sparse ($__sparse__) column, hard-coding a subset of
keys instead of routing through the standard metadata writer. This
branch's ColumnMetadataImpl still tolerates missing bitsPerElement/
lengthOfEachEntry/maxNumberOfMultiValues via UNAVAILABLE defaults, so
it wasn't fatal here, but the metadata was still wrong (e.g. CARDINALITY
was approximated as a non-null doc count rather than the real distinct
value count), and a sibling branch's stricter reader throws
NoSuchElementException on this exact gap.
Route the sparse column through the same AbstractColumnStatisticsCollector
+ BaseSegmentCreator.addColumnMetadataInfo() path that writeDenseKeyColumn()
already uses, instead of a hand-rolled subset of keys that had drifted out
of sync with what the reader expects.
Extend testSparseJsonColumnWritten to round-trip the produced properties
through ColumnMetadataImpl.fromPropertiesConfiguration() as a regression
check.
* perf(open_struct): batch coercion-failure meter to segment close
Keeps a metered-value call, which rebuilds the metric name and hits the
registry, off the per-row consuming path. Matches what
OpenStructColumnSplitter already does at seal time.
* perf(open_struct): batch inference-failure meter to segment close
Completes the batching started for the coercion meter. All three
MutableOpenStructIndex meters now accumulate per row and emit once at
close, consistent with OpenStructColumnSplitter.
* perf(open_struct): gate per-row type inference behind a per-key flag
Only a key with no declared child spec whose stored type fell back to
STRING can produce an inference failure, and that is fixed at allocation.
Every other key now skips both the child-spec lookup and the inference
call on the per-row consuming path. Metering behaviour is unchanged.
* test(open_struct): cover the inferable-value branch of the per-key
inference gate
The existing tests only fed uninferable values through meterIfUninferable,
so dropping its inferDataType null-check would still have passed. Verified
by mutation: removing the guard fails this test.
* fix(open_struct): address code review — drop shared-config metadata
cache, dedup value reads
Revert the _openStructSegmentMetadata cache added to IndexLoadingConfig.
A single IndexLoadingConfig is shared across all segments of a table and
mutated concurrently (BaseTableDataManager#reloadSegments fans out per
segment on _segmentReloadExecutor), so caching one segment's metadata for
replay by refreshIndexConfigs() could apply one segment's OPEN_STRUCT child
index config to another. Reverting also removes the reentrant double pass
through addOpenStructChildConfigs on every segment load. IndexLoadingConfig
is now identical to master; the dirty-rebuild issue the cache targeted is
filed separately, along with the pre-existing shared-config hazard.
Also:
- ImmutableOpenStructDataSource#readValue now delegates to
PinotSegmentColumnReader instead of re-implementing the null-vector check
and the dictionary/raw per-type dispatch. This also fixes the raw-forward-
index-plus-materialized-dictionary case, which the hand-rolled version
would have sent through getDictId on a raw index.
- OpenStructColumnSplitter#writeSparseJsonColumn folds its stats collection
into the existing write loop, removing a third full pass over numDocs.
Tests:
- Dense mocks now report isDictionaryEncoded/isSingleValue, matching how
writeColumnIndexes actually writes dense child columns.
- New DataProvider covers the raw no-dictionary read path across all seven
stored types, which no test previously reached.
- New MutableSegmentImplOpenStructTest covers both sides of the
toDataSource open_struct_index precondition.
* perf(open_struct): cache per-key readers across an OPEN_STRUCT
map-reconstruction scan
getMapValue previously constructed and closed a PinotSegmentColumnReader per
key per doc, forcing a fresh forward-index context (and, for the raw LZ4
sparse column, a fresh decompression buffer) on every call. Add
OpenStructDataSource#openMapValueReader(), a scan-scoped Closeable reader
that
caches one reader per key for the life of a sequential scan; wire it into
PinotSegmentRecordReader and the seal-time SegmentColumnarIndexCreator loop
instead of the stateless per-doc call.
Also fixes a crash surfaced by the new sealed-segment round-trip test:
loading an OPEN_STRUCT segment without an explicit table schema (e.g. via
the
deprecated PinotSegmentRecordReader(File) constructor) makes
SegmentMetadataImpl self-derive a schema from every physical column on disk,
including materialized OPEN_STRUCT children (event$clicks,
event$__sparse__).
PinotSegmentRecordReader#addColumnReader now skips those — they have no
DataSource of their own, only their parent does.
* perf(open_struct): meter coercion/inference failures live instead of
batching to close()
Batching these to close() hid the signal for the whole consuming-segment
lifetime and lost it if the server died first. Cache the PinotMeter returned
by ServerMetrics#addMeteredTableValue and pass it back in on the next
occurrence (the reused-meter overload) so each failure marks live while
still
skipping the per-value metric-name rebuild and registry lookup.
Also fix close()/flushMeters() for the one counter that stays batched
(ignored-key drops): wrap the key-column close loop in a finally, and zero
the counter after emitting so a second close() can't double-count it.
* fix(open_struct): reject non-coercible declared child key types at
validation
MutableKeyColumn now caches ColumnDataType.fromDataTypeSV(storedType)
.toPinotDataType() at allocation, and the fresh-key path in
MutableOpenStructIndex#index computes it as a tryCoerce argument rather than
inside tryCoerce's own try/catch. A declared child type that this conversion
chain can't handle (STRUCT, LIST, nested OPEN_STRUCT, UNKNOWN) now throws
uncaught out of index() on the very first row for that key, instead of being
counted as a coercion failure and dropped.
Reject those types in OpenStructIndexType#validate() so a bad declared type
is caught at config time, not on the consuming path.
* fix(open_struct): fix sparse column null default and drop its stats sort
The sparse column's DimensionFieldSpec left its default null value at
FieldSpec's STRING default ("null") while absent docs actually store "" on
disk, so disk and metadata disagreed. Set the spec's default explicitly to
match.
Also stop routing the sparse column's stats through
StringColumnPreIndexStatsCollector, which sorts every distinct blob
(close to one per doc here) at seal() purely to feed a dictionary this raw,
no-dictionary column never builds. Give NoDictColumnStatisticsCollector a
schema-less (FieldSpec, FieldConfig, PartitionFunction) constructor,
matching
the pattern the other per-type collectors already expose for OPEN_STRUCT's
synthetic columns, and use it here instead: same cardinality/min/max/length
stats, no sort.
* test(open_struct): strengthen sparse column metadata regression assertions
testSparseJsonColumnWritten only asserted dataType/hasDictionary, which the
old hand-rolled metadata already got right, so it passed against pre-PR
metadata too. Assert what's actually new: forward-index encoding is RAW,
cardinality is the real distinct-value count (2, not the old nonNullCount
approximation of 1), and the longest-element length is populated. Also drop
the comment referencing an internal sibling-branch name that means nothing
to readers here.
* fix(open_struct): split and clarify the OPEN_STRUCT parent-skip log
message
"schema is present or fieldSpec is X" read oddly when only one half applies.
Split into the two actual cases (no schema at all vs. a
non-ComplexFieldSpec)
and include the segment name, since this fires per segment load.
* fix(open_struct): guard isMaterializedOpenStructChild against null column
metadata map
Mutable/consuming segments' SegmentMetadataImpl always has a null
column metadata map (set in its 4-arg realtime constructor), so
getColumnMetadataFor()'s unconditional map lookup NPE'd on every
call to PinotSegmentRecordReader.init(MutableSegment, ...) -- i.e.
every getRecord() call on any realtime table, OPEN_STRUCT or not.
Treat a null map as "no materialized children", which mutable
segments never have anyway (only immutable segments materialize
OPEN_STRUCT children on disk via OpenStructColumnSplitter).
* fix(open_struct): address second review round on
ImmutableOpenStructDataSource
Four issues from the latest round of review comments on this PR:
- CachingMapValueReader cached the sparse blob reader under the parent field
name in the same map as per-key readers; a child key literally matching
the parent name shadowed it, silently dropping sparse values.
- getMapValue() and CachingMapValueReader.getMapValue() were duplicated and
had already drifted (the cached path lost the outer path's exception
context). getMapValue() now delegates to openMapValueReader() so there is
one implementation.
- OpenStructIndexType#isCoercible() accepted MAP and OPEN_STRUCT child key
types because ColumnDataType conversion succeeds for both, but
FieldSpec#getDefaultNullValue(DIMENSION, ...) - the call
allocateKeyColumn()
actually makes - has no case for either, throwing uncaught on the
consuming thread. isCoercible() now calls that exact method.
- CachingMapValueReader#close() failed fast on the first reader that threw,
leaking the rest. It now closes every reader, attaching later failures as
suppressed exceptions on the first.
---
.../immutable/ImmutableSegmentImpl.java | 8 +
.../indexsegment/mutable/MutableSegmentImpl.java | 14 +-
.../creator/impl/SegmentColumnarIndexCreator.java | 25 +-
.../impl/openstruct/OpenStructColumnSplitter.java | 57 +--
.../stats/NoDictColumnStatisticsCollector.java | 18 +
.../openstruct/ImmutableOpenStructDataSource.java | 123 +++++++
.../segment/index/openstruct/MutableKeyColumn.java | 23 +-
.../index/openstruct/MutableOpenStructIndex.java | 86 +++--
.../index/openstruct/OpenStructIndexType.java | 33 ++
.../segment/readers/PinotSegmentRecordReader.java | 38 +-
.../mutable/MutableSegmentImplOpenStructTest.java | 106 ++++++
.../openstruct/OpenStructColumnSplitterTest.java | 15 +-
.../ImmutableOpenStructDataSourceTest.java | 389 ++++++++++++++++++++-
.../openstruct/MutableOpenStructIndexTest.java | 155 +++++++-
.../index/openstruct/OpenStructIndexTypeTest.java | 30 ++
.../spi/datasource/OpenStructDataSource.java | 26 ++
16 files changed, 1048 insertions(+), 98 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
index 85d3a0c79a3..54a1895b14b 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java
@@ -143,7 +143,15 @@ public class ImmutableSegmentImpl implements
ImmutableSegment {
Schema schema = segmentMetadata.getSchema();
for (String parent : openStructParents) {
FieldSpec fieldSpec = schema != null ? schema.getFieldSpecFor(parent)
: null;
+ if (schema == null) {
+ LOGGER.warn("Segment '{}': skipping OPEN_STRUCT parent column '{}':
no schema available. "
+ + "Dense/sparse child data on disk will not be queryable.",
segmentMetadata.getName(), parent);
+ continue;
+ }
if (!(fieldSpec instanceof ComplexFieldSpec)) {
+ LOGGER.warn("Segment '{}': skipping OPEN_STRUCT parent column '{}':
fieldSpec is {} "
+ + "(expected ComplexFieldSpec). Dense/sparse child data on
disk will not be queryable.",
+ segmentMetadata.getName(), parent, fieldSpec != null ?
fieldSpec.getClass().getSimpleName() : "null");
continue;
}
ColumnMetadata parentMetadata =
segmentMetadata.getColumnMetadataMap().get(parent);
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 4fec4e25c7d..4225dfea93f 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
@@ -1335,18 +1335,6 @@ public class MutableSegmentImpl implements
MutableSegment {
}
}
- /// Returns the per-column mutable OPEN_STRUCT index, or `null` if the
column is not OPEN_STRUCT
- /// or the index has not been initialized.
- @Nullable
- public MutableOpenStructIndex getOpenStructIndex(String column) {
- IndexContainer container = _indexContainerMap.get(column);
- if (container == null) {
- return null;
- }
- MutableIndex index =
container._mutableIndexes.get(StandardIndexes.openStruct());
- return index instanceof MutableOpenStructIndex ? (MutableOpenStructIndex)
index : null;
- }
-
@Override
public void offload() {
if (_partitionUpsertMetadataManager != null) {
@@ -1734,6 +1722,8 @@ public class MutableSegmentImpl implements MutableSegment
{
DataSource toDataSource() {
if (_fieldSpec.getDataType() == DataType.OPEN_STRUCT) {
MutableIndex idx = _mutableIndexes.get(StandardIndexes.openStruct());
+ Preconditions.checkState(idx instanceof MutableOpenStructIndex,
+ "OPEN_STRUCT column '%s' requires the open_struct_index to be
enabled", _fieldSpec.getName());
return new MutableOpenStructDataSource((ComplexFieldSpec) _fieldSpec,
(MutableOpenStructIndex) idx,
_numDocsIndexed);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
index 69cc5e2e624..0b67f0850cb 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/SegmentColumnarIndexCreator.java
@@ -155,24 +155,27 @@ public class SegmentColumnarIndexCreator extends
BaseSegmentCreator {
@Nullable int[] sortedDocIds, @Nullable RoaringBitmap validDocIds)
throws IOException {
List<IndexCreator> creators =
_colIndexes.get(columnName).getIndexCreators();
- if (sortedDocIds != null) {
- for (int docId : sortedDocIds) {
- if (validDocIds == null || validDocIds.contains(docId)) {
- indexOpenStructDoc(dataSource, docId, creators);
+ try (OpenStructDataSource.MapValueReader mapValueReader =
dataSource.openMapValueReader()) {
+ if (sortedDocIds != null) {
+ for (int docId : sortedDocIds) {
+ if (validDocIds == null || validDocIds.contains(docId)) {
+ indexOpenStructDoc(mapValueReader, docId, creators);
+ }
}
- }
- } else {
- for (int docId = 0; docId < numDocs; docId++) {
- if (validDocIds == null || validDocIds.contains(docId)) {
- indexOpenStructDoc(dataSource, docId, creators);
+ } else {
+ for (int docId = 0; docId < numDocs; docId++) {
+ if (validDocIds == null || validDocIds.contains(docId)) {
+ indexOpenStructDoc(mapValueReader, docId, creators);
+ }
}
}
}
}
- private static void indexOpenStructDoc(OpenStructDataSource dataSource, int
docId, List<IndexCreator> creators)
+ private static void indexOpenStructDoc(OpenStructDataSource.MapValueReader
mapValueReader, int docId,
+ List<IndexCreator> creators)
throws IOException {
- Map<String, Object> value = dataSource.getMapValue(docId);
+ Map<String, Object> value = mapValueReader.getMapValue(docId);
Object toIndex = value != null ? value : Map.of();
for (IndexCreator creator : creators) {
creator.add(toIndex, -1);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
index cfa82d50f33..e50329821fa 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java
@@ -18,9 +18,9 @@
*/
package org.apache.pinot.segment.local.segment.creator.impl.openstruct;
+import com.google.common.base.Utf8;
import java.io.File;
import java.io.IOException;
-import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -41,6 +41,7 @@ import
org.apache.pinot.segment.local.segment.creator.impl.fwd.SingleValueVarByt
import
org.apache.pinot.segment.local.segment.creator.impl.inv.json.OffHeapJsonIndexCreator;
import
org.apache.pinot.segment.local.segment.creator.impl.nullvalue.NullValueVectorCreator;
import
org.apache.pinot.segment.local.segment.creator.impl.stats.AbstractColumnStatisticsCollector;
+import
org.apache.pinot.segment.local.segment.creator.impl.stats.NoDictColumnStatisticsCollector;
import
org.apache.pinot.segment.local.segment.creator.impl.stats.StatsCollectorUtil;
import
org.apache.pinot.segment.local.segment.index.dictionary.DictionaryIndexType;
import
org.apache.pinot.segment.local.segment.index.openstruct.OpenStructSupportedIndexes;
@@ -234,11 +235,9 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
_inferredTypes.putIfAbsent(key, valueType);
}
}
- if (!_presenceBitmaps.containsKey(key)) {
- _presenceBitmaps.put(key, new RoaringBitmap());
- _values.put(key, new ArrayList<>());
- }
- _presenceBitmaps.get(key).add(_numDocs);
+ RoaringBitmap bitmap = _presenceBitmaps.computeIfAbsent(key, k -> new
RoaringBitmap());
+ List<Object> values = _values.computeIfAbsent(key, k -> new
ArrayList<>());
+ bitmap.add(_numDocs);
Object coerced;
try {
PinotDataType sourceType =
PinotDataType.getSingleValueType(rawValue);
@@ -246,10 +245,10 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
coerced = destType.convert(rawValue, sourceType);
} catch (Exception e) {
_coercionFailuresPerKey.merge(key, 1L, Long::sum);
- _presenceBitmaps.get(key).remove(_numDocs);
+ bitmap.remove(_numDocs);
continue;
}
- _values.get(key).add(coerced);
+ values.add(coerced);
}
}
_numDocs++;
@@ -565,7 +564,6 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
String sparseCol = OpenStructNaming.sparseColumnName(_columnName);
int maxLen = 1;
String[] jsonPerDoc = new String[_numDocs];
- int nonNullCount = 0;
for (int docId = 0; docId < _numDocs; docId++) {
Map<String, Object> sparseEntries = new LinkedHashMap<>();
for (String key : sparseKeys) {
@@ -579,14 +577,26 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
try {
String json = JsonUtils.objectToString(sparseEntries);
jsonPerDoc[docId] = json;
- maxLen = Math.max(maxLen,
json.getBytes(StandardCharsets.UTF_8).length);
- nonNullCount++;
+ maxLen = Math.max(maxLen, Utf8.encodedLength(json));
} catch (IOException e) {
throw new RuntimeException("Failed to serialize sparse entries for
docId " + docId, e);
}
}
}
+ // Absent docs store "" in the raw forward index (see loop below) and are
flagged in the null vector, so feed
+ // the same placeholder through the stats collector and record it as the
default null value. Collected inside
+ // the write loop rather than in a pass of its own: it needs the exact
same per-doc branch.
+ DimensionFieldSpec sparseFieldSpec = new DimensionFieldSpec(sparseCol,
DataType.STRING, true);
+ String defaultValue = "";
+ sparseFieldSpec.setDefaultNullValue(defaultValue);
+ // This column is always raw (no dictionary), so it never needs the sorted
unique-values array
+ // StringColumnPreIndexStatsCollector builds for dictionary creation --
the O(n log n) sort at
+ // seal() would be pure overhead here, close to one entry per doc.
NoDictColumnStatisticsCollector
+ // gives the same cardinality/min/max/length stats without it (exact
cardinality up to its
+ // tracking threshold, HyperLogLog beyond that).
+ AbstractColumnStatisticsCollector statsCollector = new
NoDictColumnStatisticsCollector(sparseFieldSpec, null, null);
+
SingleValueVarByteRawIndexCreator fwdCreator = new
SingleValueVarByteRawIndexCreator(
_indexDir, ChunkCompressionType.LZ4, sparseCol, _numDocs,
DataType.STRING, maxLen);
NullValueVectorCreator nullCreator = new NullValueVectorCreator(_indexDir,
sparseCol);
@@ -596,11 +606,13 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
try {
for (int docId = 0; docId < _numDocs; docId++) {
if (jsonPerDoc[docId] != null) {
+ statsCollector.collect(jsonPerDoc[docId]);
fwdCreator.putString(jsonPerDoc[docId]);
if (jsonCreator != null) {
jsonCreator.add(jsonPerDoc[docId]);
}
} else {
+ statsCollector.collect(defaultValue);
fwdCreator.putString("");
nullCreator.setNull(docId);
if (jsonCreator != null) {
@@ -621,22 +633,15 @@ public class OpenStructColumnSplitter implements
ColumnarOpenStructIndexCreator
}
}
+ statsCollector.seal();
+
PropertiesConfiguration props = new PropertiesConfiguration();
- props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.DATA_TYPE),
- DataType.STRING.name());
- props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.COLUMN_TYPE),
- FieldSpec.FieldType.DIMENSION.name());
- props.setProperty(
- V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.IS_SINGLE_VALUED), true);
- props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.TOTAL_DOCS),
- _numDocs);
- props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.CARDINALITY),
- nonNullCount);
- props.setProperty(
- V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.TOTAL_NUMBER_OF_ENTRIES),
- _numDocs);
- props.setProperty(
- V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.HAS_DICTIONARY), false);
+ // Route through the same metadata writer as dense child columns
(BaseSegmentCreator.addColumnMetadataInfo) so
+ // this raw, no-dictionary column carries every property
ColumnMetadataImpl.fromPropertiesConfiguration()
+ // expects, instead of a hand-rolled subset that can silently drift from
what the reader requires.
+ BaseSegmentCreator.addColumnMetadataInfo(props, sparseCol, statsCollector,
_numDocs, sparseFieldSpec,
+ false /* hasDictionary */, 0 /* dictionaryElementSize */,
FieldConfig.EncodingType.RAW,
+ false /* autoGenerated */);
props.setProperty(V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
"hasNullValue"), true);
props.setProperty(
V1Constants.MetadataKeys.Column.getKeyFor(sparseCol,
V1Constants.MetadataKeys.Column.PARENT_COLUMN),
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/NoDictColumnStatisticsCollector.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/NoDictColumnStatisticsCollector.java
index 1f1991584b7..6810d55e010 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/NoDictColumnStatisticsCollector.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/stats/NoDictColumnStatisticsCollector.java
@@ -25,6 +25,9 @@ import java.util.HashSet;
import java.util.Set;
import javax.annotation.Nullable;
import org.apache.pinot.segment.spi.creator.StatsCollectorConfig;
+import org.apache.pinot.segment.spi.partition.PartitionFunction;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.utils.BigDecimalUtils;
import org.apache.pinot.spi.utils.ByteArray;
@@ -73,6 +76,21 @@ public class NoDictColumnStatisticsCollector extends
AbstractColumnStatisticsCol
LOGGER.info("Initialized NoDictColumnStatisticsCollector for column: {}",
column);
}
+ /// Constructs a collector directly from a [FieldSpec], without a
[StatsCollectorConfig]. Lets callers
+ /// that operate outside schema-driven segment generation (e.g. OPEN_STRUCT
materialized child columns,
+ /// whose synthetic columns exist in no schema) reuse this no-dictionary
collector, same as the other
+ /// per-type collectors already do.
+ public NoDictColumnStatisticsCollector(FieldSpec fieldSpec, @Nullable
FieldConfig fieldConfig,
+ @Nullable PartitionFunction partitionFunction) {
+ super(fieldSpec, fieldConfig, partitionFunction);
+ _isFixedWidth = _storedType.isFixedWidth();
+ _isAscii = _storedType == DataType.STRING;
+ _hllPlus = new HyperLogLogPlus(
+ CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_P,
+ CommonConstants.Helix.DEFAULT_HYPERLOGLOG_PLUS_SP);
+ LOGGER.info("Initialized NoDictColumnStatisticsCollector for column: {}",
fieldSpec.getName());
+ }
+
@Override
public void collect(Object entry) {
assert !_sealed;
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java
index 6972604027b..20fe11ec95c 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.segment.local.segment.index.openstruct;
+import java.io.IOException;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -26,6 +28,7 @@ import javax.annotation.Nullable;
import org.apache.pinot.segment.local.segment.index.datasource.BaseDataSource;
import
org.apache.pinot.segment.local.segment.index.datasource.ImmutableDataSource;
import org.apache.pinot.segment.local.segment.index.datasource.NullDataSource;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.spi.Constants;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.datasource.DataSourceMetadata;
@@ -36,6 +39,7 @@ import
org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
import org.apache.pinot.segment.spi.partition.PartitionFunction;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.utils.JsonUtils;
/// Per-key [DataSource] accessor for sealed OPEN_STRUCT segments. Dense keys
get materialized DataSources; sparse keys
@@ -140,6 +144,125 @@ public class ImmutableOpenStructDataSource extends
BaseDataSource implements Ope
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : null;
}
+ @Nullable
+ @Override
+ public Map<String, Object> getMapValue(int docId) {
+ try (MapValueReader reader = openMapValueReader()) {
+ return reader.getMapValue(docId);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close OPEN_STRUCT map value
reader", e);
+ }
+ }
+
+ @Override
+ public MapValueReader openMapValueReader() {
+ return new CachingMapValueReader();
+ }
+
+ /// Caches one [PinotSegmentColumnReader] per key for the life of the
reader, instead of
+ /// constructing one per call the way an unscoped [#getMapValue(int)] would.
For a raw,
+ /// chunk-compressed column (e.g. the sparse blob), a fresh reader per doc
means a fresh
+ /// decompression buffer and, depending on access order, redundant
re-decompression of the same
+ /// chunk; reusing the reader across a sequential scan lets it carry its
decoded-chunk state
+ /// forward. Not thread-safe — for one single-threaded scan only, per
+ /// [OpenStructDataSource#openMapValueReader()].
+ private final class CachingMapValueReader implements MapValueReader {
+ private final Map<String, PinotSegmentColumnReader> _readers = new
HashMap<>();
+ // Kept out of _readers: a child key that happens to match the parent
field name would otherwise share the
+ // same map entry as the sparse blob reader, silently shadowing it.
+ @Nullable
+ private final PinotSegmentColumnReader _sparseReader;
+
+ CachingMapValueReader() {
+ _sparseReader = _sparseDataSource != null ?
createReader(_fieldSpec.getName(), _sparseDataSource) : null;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Nullable
+ @Override
+ public Map<String, Object> getMapValue(int docId) {
+ Map<String, Object> result = null;
+
+ for (Map.Entry<String, DataSource> entry :
_perKeyDataSources.entrySet()) {
+ Object value = readValue(entry.getKey(), entry.getValue(), docId);
+ if (value != null) {
+ if (result == null) {
+ result = new HashMap<>();
+ }
+ result.put(entry.getKey(), value);
+ }
+ }
+
+ if (_sparseReader != null) {
+ Object sparseValue = readValue(_sparseReader, docId);
+ if (sparseValue instanceof String json && !json.isEmpty()) {
+ try {
+ Map<String, Object> sparseMap = JsonUtils.stringToObject(json,
Map.class);
+ if (result == null) {
+ result = new HashMap<>();
+ }
+ result.putAll(sparseMap);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to parse sparse JSON at docId "
+ docId, e);
+ }
+ }
+ }
+
+ return result;
+ }
+
+ @Nullable
+ private Object readValue(String key, DataSource dataSource, int docId) {
+ return readValue(_readers.computeIfAbsent(key, k -> createReader(k,
dataSource)), docId);
+ }
+
+ @Nullable
+ private Object readValue(@Nullable PinotSegmentColumnReader reader, int
docId) {
+ return reader == null ? null : reader.isNull(docId) ? null :
reader.getValue(docId);
+ }
+
+ @Nullable
+ private PinotSegmentColumnReader createReader(String key, DataSource
dataSource) {
+ ForwardIndexReader<?> fwdReader = dataSource.getForwardIndex();
+ if (fwdReader == null) {
+ return null;
+ }
+ return new PinotSegmentColumnReader(key, fwdReader,
dataSource.getDictionary(),
+ dataSource.getNullValueVector(), 0);
+ }
+
+ @Override
+ public void close()
+ throws IOException {
+ IOException firstException = null;
+ for (PinotSegmentColumnReader reader : _readers.values()) {
+ firstException = closeQuietly(reader, firstException);
+ }
+ if (_sparseReader != null) {
+ firstException = closeQuietly(_sparseReader, firstException);
+ }
+ if (firstException != null) {
+ throw firstException;
+ }
+ }
+
+ // Closes every reader even when an earlier one throws, instead of leaking
the rest; extra failures are
+ // attached as suppressed on the first exception, mirroring
try-with-resources semantics.
+ @Nullable
+ private static IOException closeQuietly(PinotSegmentColumnReader reader,
@Nullable IOException firstException) {
+ try {
+ reader.close();
+ return firstException;
+ } catch (IOException e) {
+ if (firstException == null) {
+ return e;
+ }
+ firstException.addSuppressed(e);
+ return firstException;
+ }
+ }
+ }
+
private static class ImmutableOpenStructDataSourceMetadata implements
DataSourceMetadata {
private final FieldSpec _fieldSpec;
private final int _numDocs;
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableKeyColumn.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableKeyColumn.java
index a6b042cf747..1f5d5357981 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableKeyColumn.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableKeyColumn.java
@@ -21,6 +21,7 @@ package
org.apache.pinot.segment.local.segment.index.openstruct;
import java.io.Closeable;
import java.io.IOException;
import java.util.Set;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import
org.apache.pinot.segment.local.realtime.impl.dictionary.MutableDictionaryFactory;
import
org.apache.pinot.segment.local.realtime.impl.forward.FixedByteSVMutableForwardIndex;
import
org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeInvertedIndex;
@@ -31,6 +32,7 @@ import
org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext;
import org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.PinotDataType;
/// A single key's mutable column for an OPEN_STRUCT column: forward index
(dictionary-encoded)
@@ -45,6 +47,8 @@ public class MutableKeyColumn implements Closeable {
private final String _key;
private final DataType _storedType;
+ private final PinotDataType _destType;
+ private final boolean _needsInferenceCheck;
private final MutableForwardIndex _forwardIndex;
private final ThreadSafeMutableRoaringBitmap _presenceBitmap;
private final MutableDictionary _dictionary;
@@ -62,13 +66,16 @@ public class MutableKeyColumn implements Closeable {
public MutableKeyColumn(String key, DataType storedType, Object
defaultNullValue,
PinotDataBufferMemoryManager memoryManager, int capacity) {
- this(key, storedType, defaultNullValue, memoryManager, capacity, key);
+ this(key, storedType, defaultNullValue, memoryManager, capacity, key,
false);
}
public MutableKeyColumn(String key, DataType storedType, Object
defaultNullValue,
- PinotDataBufferMemoryManager memoryManager, int capacity, String
allocationContext) {
+ PinotDataBufferMemoryManager memoryManager, int capacity, String
allocationContext,
+ boolean needsInferenceCheck) {
_key = key;
_storedType = storedType;
+ _needsInferenceCheck = needsInferenceCheck;
+ _destType = ColumnDataType.fromDataTypeSV(storedType).toPinotDataType();
_presenceBitmap = new ThreadSafeMutableRoaringBitmap();
_invertedIndex = new RealtimeInvertedIndex();
@@ -99,6 +106,18 @@ public class MutableKeyColumn implements Closeable {
return _storedType;
}
+ public PinotDataType getDestType() {
+ return _destType;
+ }
+
+ /// Whether a value on this key can ever produce a type-inference failure.
True only for a key
+ /// with no declared child spec whose stored type fell back to STRING; fixed
at allocation, since
+ /// neither the child spec nor the stored type changes afterwards. Lets the
per-row path skip the
+ /// inference call entirely for every other key.
+ public boolean needsInferenceCheck() {
+ return _needsInferenceCheck;
+ }
+
public MutableForwardIndex getForwardIndex() {
return _forwardIndex;
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
index b1c81dac7b2..119c7717a32 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java
@@ -41,6 +41,7 @@ import org.apache.pinot.spi.data.DimensionFieldSpec;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.OpenStructTypeInference;
+import org.apache.pinot.spi.metrics.PinotMeter;
import org.apache.pinot.spi.utils.PinotDataType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,12 +64,23 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
private final PinotDataBufferMemoryManager _memoryManager;
private final int _capacity;
- // Volatile for lock-free reader access; writer always holds the
consuming-thread lock.
+ // Volatile copy-on-write: the writer (consuming thread) creates a fresh
HashMap copy and publishes
+ // atomically via volatile write (see allocateKeyColumn). Readers see a
consistent snapshot of the
+ // entire map. ConcurrentHashMap is NOT appropriate here — it would allow
readers to observe
+ // partially-updated state during a put. Single-writer is guaranteed by the
Pinot consuming thread
+ // model (one thread per partition).
private volatile Map<String, MutableKeyColumn> _keyColumns = new HashMap<>();
// Single-writer (see #index), but close() may run on a different thread, so
volatile for
// visibility; flushed to ServerMetrics on close() to avoid a metered-value
call on every
// ignored key of every consumed row.
private volatile long _ignoredKeyDropCount;
+ // Single-writer (see #index): the consuming thread caches and reuses the
PinotMeter, which skips
+ // the per-value metric-name rebuild and registry lookup
addMeteredTableValue would otherwise do,
+ // while still marking the meter live instead of batching the count to
close().
+ @Nullable
+ private PinotMeter _typeCoercionFailureMeter;
+ @Nullable
+ private PinotMeter _typeInferenceFailureMeter;
public MutableOpenStructIndex(String openStructColumn, String
tableNameWithType, ComplexFieldSpec fieldSpec,
OpenStructIndexConfig config, PinotDataBufferMemoryManager
memoryManager, int capacity) {
@@ -119,7 +131,8 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
// Resolve stored type and coerce BEFORE allocating a column so a
first-row coercion failure
// does not allocate a column that was never usable.
DataType resolvedType = resolveStoredType(key, rawValue, null);
- Object coerced = tryCoerce(key, rawValue, resolvedType);
+ Object coerced = tryCoerce(key, rawValue,
+ ColumnDataType.fromDataTypeSV(resolvedType).toPinotDataType());
if (coerced == null) {
continue;
}
@@ -128,10 +141,10 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
continue;
}
- // Re-resolve against the established type so a later unmappable value
on a STRING key is
- // metered too; for any other established type this is a no-op returning
that type.
- DataType storedType = resolveStoredType(key, rawValue,
keyCol.getStoredType());
- Object coerced = tryCoerce(key, rawValue, storedType);
+ if (keyCol.needsInferenceCheck()) {
+ meterIfUninferable(rawValue);
+ }
+ Object coerced = tryCoerce(key, rawValue, keyCol.getDestType());
if (coerced == null) {
continue;
}
@@ -160,38 +173,51 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
+ " Falling back to STRING.",
_openStructColumn, key, rawValue.getClass().getName());
}
- ServerMetrics serverMetrics = ServerMetrics.get();
- if (serverMetrics != null) {
- serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn,
- ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1);
- }
+ _typeInferenceFailureMeter =
meterFailure(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES,
+ _typeInferenceFailureMeter);
return DataType.STRING;
}
return establishedType != null ? establishedType : inferred;
}
+ /// Counts an inference failure for a value on a STRING-fallback key. This
is the metering-only
+ /// half of [#resolveStoredType]: on the established path that method always
returns the key's
+ /// own stored type, so the return value is unused and only the side effect
matters.
+ private void meterIfUninferable(Object rawValue) {
+ if (OpenStructTypeInference.inferDataType(rawValue) == null) {
+ _typeInferenceFailureMeter =
meterFailure(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES,
+ _typeInferenceFailureMeter);
+ }
+ }
+
/// Coerces rawValue to storedType. Returns null on failure; the caller
drops the entry. Failures
/// are reported through [ServerMeter#OPEN_STRUCT_TYPE_COERCION_FAILURES]
rather than a log line,
/// because this runs per value on the consuming path. Note: a successful
coerce of a
/// "null"-shaped raw value would also return null — but callers gate on
rawValue != null before
/// reaching here.
@Nullable
- private Object tryCoerce(String key, Object rawValue, DataType storedType) {
+ private Object tryCoerce(String key, Object rawValue, PinotDataType
destType) {
try {
PinotDataType sourceType = PinotDataType.getSingleValueType(rawValue);
- PinotDataType destType =
ColumnDataType.fromDataTypeSV(storedType).toPinotDataType();
return destType.convert(rawValue, sourceType);
} catch (Exception e) {
- ServerMetrics serverMetrics = ServerMetrics.get();
- if (serverMetrics != null) {
- // Column-granular for the same reason as the inference meter above.
- serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn,
- ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1);
- }
+ _typeCoercionFailureMeter =
meterFailure(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES,
+ _typeCoercionFailureMeter);
return null;
}
}
+ /// Marks one occurrence on `meter`, reusing `reusedMeter` when present to
skip the metric-name
+ /// rebuild and registry lookup a fresh [ServerMetrics#addMeteredTableValue]
call would do. Returns
+ /// the meter to reuse on the next call (unchanged when no [ServerMetrics]
is registered).
+ private PinotMeter meterFailure(ServerMeter meter, @Nullable PinotMeter
reusedMeter) {
+ ServerMetrics serverMetrics = ServerMetrics.get();
+ if (serverMetrics == null) {
+ return reusedMeter;
+ }
+ return serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn, meter, 1, reusedMeter);
+ }
+
/// Allocates a new MutableKeyColumn for `key` with the resolved
`storedType` and
/// publishes it via volatile copy-on-write.
private MutableKeyColumn allocateKeyColumn(String key, DataType storedType) {
@@ -202,8 +228,9 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
// way, and a doc's resolved value must stay identical before and after
seal.
// See https://github.com/apache/pinot/issues/19466
Object defaultNullValue =
FieldSpec.getDefaultNullValue(FieldSpec.FieldType.DIMENSION, storedType, null);
- MutableKeyColumn newCol =
- new MutableKeyColumn(key, storedType, defaultNullValue,
_memoryManager, _capacity, allocationContext);
+ boolean needsInferenceCheck = !_childFieldSpecs.containsKey(key) &&
storedType == DataType.STRING;
+ MutableKeyColumn newCol = new MutableKeyColumn(key, storedType,
defaultNullValue, _memoryManager, _capacity,
+ allocationContext, needsInferenceCheck);
Map<String, MutableKeyColumn> updated = new HashMap<>(_keyColumns);
updated.put(key, newCol);
_keyColumns = updated;
@@ -290,15 +317,26 @@ public class MutableOpenStructIndex implements
OpenStructIndexReader<ForwardInde
@Override
public void close()
throws IOException {
+ try {
+ flushMeters();
+ } finally {
+ for (MutableKeyColumn keyCol : _keyColumns.values()) {
+ keyCol.close();
+ }
+ }
+ }
+
+ /// Emits the batched ignored-key-drop counter. It accumulates per row on
the consuming path and
+ /// is flushed once here, mirroring what [OpenStructColumnSplitter] does at
seal time. Zeroed after
+ /// emitting so a second close() (e.g. destroy() after commit()) does not
double-count.
+ private void flushMeters() {
if (_ignoredKeyDropCount > 0) {
ServerMetrics serverMetrics = ServerMetrics.get();
if (serverMetrics != null) {
serverMetrics.addMeteredTableValue(_tableNameWithType,
_openStructColumn,
ServerMeter.OPEN_STRUCT_IGNORED_KEY_DROPS, _ignoredKeyDropCount);
}
- }
- for (MutableKeyColumn keyCol : _keyColumns.values()) {
- keyCol.close();
+ _ignoredKeyDropCount = 0;
}
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
index 1b0e5c296aa..ebdad5e4afe 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java
@@ -88,6 +88,39 @@ public class OpenStructIndexType
fieldSpec.getName());
validatePerKeyIndexes(config);
validateIgnoredKeys(config, fieldSpec);
+ if (fieldSpec instanceof ComplexFieldSpec) {
+ validateChildFieldSpecTypes((ComplexFieldSpec) fieldSpec);
+ }
+ }
+ }
+
+ /// Rejects a declared child key type that a key column cannot actually
store (e.g. STRUCT, LIST, MAP, nested
+ /// OPEN_STRUCT, UNKNOWN —
[FieldSpec#getDefaultNullValue(FieldSpec.FieldType,FieldSpec.DataType,String)]
has no
+ /// DIMENSION case for these). Catching this at config validation, rather
than surfacing it as an uncaught
+ /// exception on the first row ingested for such a key, keeps a bad declared
type from taking down the whole
+ /// consuming thread.
+ private void validateChildFieldSpecTypes(ComplexFieldSpec fieldSpec) {
+ Map<String, FieldSpec> childFieldSpecs = fieldSpec.getChildFieldSpecs();
+ if (childFieldSpecs == null) {
+ return;
+ }
+ for (Map.Entry<String, FieldSpec> entry : childFieldSpecs.entrySet()) {
+ FieldSpec.DataType storedType =
entry.getValue().getDataType().getStoredType();
+ Preconditions.checkState(isCoercible(storedType),
+ "OPEN_STRUCT column '%s': child key '%s' declares type '%s', which
cannot be coerced for indexing",
+ fieldSpec.getName(), entry.getKey(), storedType);
+ }
+ }
+
+ private static boolean isCoercible(FieldSpec.DataType storedType) {
+ try {
+ // The exact call allocateKeyColumn() makes to compute a key column's
default null value; a declared type
+ // that fails it here (e.g. MAP, OPEN_STRUCT, which ColumnDataType
conversion alone accepts) would otherwise
+ // throw uncaught on the consuming thread instead of being rejected at
config validation time.
+ FieldSpec.getDefaultNullValue(FieldSpec.FieldType.DIMENSION, storedType,
null);
+ return true;
+ } catch (Exception e) {
+ return false;
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentRecordReader.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentRecordReader.java
index d7be52ed7b2..bb7ab182618 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentRecordReader.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/readers/PinotSegmentRecordReader.java
@@ -30,10 +30,12 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.pinot.common.utils.FileUtils;
import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
import org.apache.pinot.segment.local.segment.readers.sort.PinotSegmentSorter;
+import org.apache.pinot.segment.spi.ColumnMetadata;
import org.apache.pinot.segment.spi.IndexSegment;
import org.apache.pinot.segment.spi.MutableSegment;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.datasource.OpenStructDataSource;
+import org.apache.pinot.segment.spi.index.metadata.ColumnMetadataImpl;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.data.readers.RecordReader;
@@ -59,6 +61,9 @@ public class PinotSegmentRecordReader implements RecordReader
{
// OPEN_STRUCT parent columns have no forward index of their own (they
expose per-key sub-columns),
// so they cannot be read via PinotSegmentColumnReader. Their per-doc value
is reconstructed as a map.
private Map<String, OpenStructDataSource> _openStructDataSources;
+ // Scan-scoped readers opened once per column (see
OpenStructDataSource#openMapValueReader), reused
+ // across every getRecord() call instead of reconstructing per-key
forward-index readers per doc.
+ private Map<String, OpenStructDataSource.MapValueReader>
_openStructMapValueReaders;
private int[] _sortedDocIds;
private boolean _skipDefaultNullValues;
@@ -172,6 +177,7 @@ public class PinotSegmentRecordReader implements
RecordReader {
_columnReaders = new ArrayList<>();
_columnNames = new ArrayList<>();
_openStructDataSources = new HashMap<>();
+ _openStructMapValueReaders = new HashMap<>();
Set<String> columnsInSegment = _indexSegment.getPhysicalColumnNames();
if (CollectionUtils.isEmpty(fieldsToRead)) {
for (String column : columnsInSegment) {
@@ -207,7 +213,19 @@ public class PinotSegmentRecordReader implements
RecordReader {
private void addColumnReader(String column) {
DataSource dataSource = _indexSegment.getDataSourceNullable(column);
if (dataSource instanceof OpenStructDataSource) {
- _openStructDataSources.put(column, (OpenStructDataSource) dataSource);
+ OpenStructDataSource openStructDataSource = (OpenStructDataSource)
dataSource;
+ _openStructDataSources.put(column, openStructDataSource);
+ _openStructMapValueReaders.put(column,
openStructDataSource.openMapValueReader());
+ return;
+ }
+ if (isMaterializedOpenStructChild(column)) {
+ // A materialized OPEN_STRUCT/MAP child (e.g. "event$clicks") is a
physical column on disk but
+ // not independently queryable -- it has no DataSource of its own (see
+ // ImmutableSegmentImpl#_dataSources), only its parent does, and the
parent's OpenStructDataSource
+ // registered above already covers its value. This is reachable whenever
the segment is loaded
+ // without an explicit table schema (e.g. the File-based constructors of
this class), since
+ // SegmentMetadataImpl then self-derives a schema from every physical
column on disk, children
+ // included.
return;
}
PinotSegmentColumnReader reader = new
PinotSegmentColumnReader(_indexSegment, column);
@@ -216,6 +234,17 @@ public class PinotSegmentRecordReader implements
RecordReader {
_columnReaders.add(reader);
}
+ private boolean isMaterializedOpenStructChild(String column) {
+ // Mutable/consuming segments have no column metadata map (never
materialize OPEN_STRUCT children
+ // on disk), so getColumnMetadataFor() would NPE on its unconditional map
lookup.
+ Map<String, ColumnMetadata> columnMetadataMap =
_indexSegment.getSegmentMetadata().getColumnMetadataMap();
+ if (columnMetadataMap == null) {
+ return false;
+ }
+ ColumnMetadata columnMetadata = columnMetadataMap.get(column);
+ return columnMetadata instanceof ColumnMetadataImpl &&
((ColumnMetadataImpl) columnMetadata).isMaterializedChild();
+ }
+
/// Returns the sorted document ids.
@Nullable
public int[] getSortedDocIds() {
@@ -252,7 +281,7 @@ public class PinotSegmentRecordReader implements
RecordReader {
buffer.putDefaultNullValue(column, columnReader.getValue(docId));
}
}
- for (Map.Entry<String, OpenStructDataSource> entry :
_openStructDataSources.entrySet()) {
+ for (Map.Entry<String, OpenStructDataSource.MapValueReader> entry :
_openStructMapValueReaders.entrySet()) {
Map<String, Object> value = entry.getValue().getMapValue(docId);
// A null map means no key is present at this doc; leave the column
unset so the OPEN_STRUCT
// build treats it as an absent/empty struct.
@@ -316,6 +345,11 @@ public class PinotSegmentRecordReader implements
RecordReader {
closeException = e;
}
}
+ if (_openStructMapValueReaders != null) {
+ for (OpenStructDataSource.MapValueReader reader :
_openStructMapValueReaders.values()) {
+ reader.close();
+ }
+ }
if (_destroySegmentOnClose && _indexSegment != null) {
_indexSegment.destroy();
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplOpenStructTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplOpenStructTest.java
new file mode 100644
index 00000000000..fb7870ce90e
--- /dev/null
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplOpenStructTest.java
@@ -0,0 +1,106 @@
+/**
+ * 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.segment.local.indexsegment.mutable;
+
+import java.io.File;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.pinot.common.metadata.segment.SegmentZKMetadata;
+import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager;
+import org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentConfig;
+import
org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory;
+import
org.apache.pinot.segment.local.segment.index.openstruct.MutableOpenStructDataSource;
+import org.apache.pinot.segment.spi.datasource.DataSource;
+import org.apache.pinot.segment.spi.index.StandardIndexes;
+import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
+import org.apache.pinot.spi.data.ComplexFieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.builder.TableNameBuilder;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
+
+
+/// Covers the [MutableSegmentImpl] precondition that an OPEN_STRUCT column
cannot be turned into a
+/// DataSource unless `open_struct_index` is enabled for it. The index is
enabled by default
+/// ([OpenStructIndexConfig#DEFAULT]), so the guard is only reachable when a
table config explicitly
+/// disables it — without the guard that misconfiguration would surface as a
ClassCastException on a
+/// null/foreign index far from its cause.
+public class MutableSegmentImplOpenStructTest {
+ private static final String TABLE_NAME_WITH_TYPE =
+ TableNameBuilder.REALTIME.tableNameWithType("openStructGuardTest");
+ private static final String SEGMENT_NAME = "openStructGuardTest__0__0__" +
UUID.randomUUID();
+ private static final File TEMP_DIR =
+ new File(System.getProperty("java.io.tmpdir"),
MutableSegmentImplOpenStructTest.class.getSimpleName());
+
+ private static MutableSegmentImpl createSegment(boolean
openStructIndexEnabled) {
+ Schema schema = new Schema();
+ schema.addField(new ComplexFieldSpec("event", DataType.OPEN_STRUCT, true,
Map.of()));
+
+ RealtimeSegmentStatsHistory statsHistory =
mock(RealtimeSegmentStatsHistory.class);
+ when(statsHistory.getEstimatedCardinality(anyString())).thenReturn(200);
+ when(statsHistory.getEstimatedAvgColSize(anyString())).thenReturn(32);
+
+ RealtimeSegmentConfig config = new RealtimeSegmentConfig.Builder()
+ .setTableNameWithType(TABLE_NAME_WITH_TYPE)
+ .setSegmentName(SEGMENT_NAME)
+ .setStreamName("openStructGuardTest")
+ .setSchema(schema)
+ .setCapacity(1000)
+ .setAvgNumMultiValues(2)
+ .setIndex("event", StandardIndexes.openStruct(),
+ openStructIndexEnabled ? OpenStructIndexConfig.DEFAULT :
OpenStructIndexConfig.DISABLED)
+ .setSegmentZKMetadata(new SegmentZKMetadata(SEGMENT_NAME))
+ .setMemoryManager(new DirectMemoryManager(SEGMENT_NAME))
+ .setStatsHistory(statsHistory)
+ .setConsumerDir(new File(TEMP_DIR,
UUID.randomUUID().toString()).getAbsolutePath())
+ .build();
+ return new MutableSegmentImpl(config, null);
+ }
+
+ @Test
+ public void testToDataSourceThrowsWhenOpenStructIndexDisabled() {
+ MutableSegmentImpl segment = createSegment(false);
+ try {
+ IllegalStateException e =
+ expectThrows(IllegalStateException.class, () ->
segment.getDataSource("event"));
+ assertTrue(e.getMessage().contains("open_struct_index"), "unexpected
message: " + e.getMessage());
+ } finally {
+ segment.destroy();
+ }
+ }
+
+ @Test
+ public void testToDataSourceSucceedsWhenOpenStructIndexEnabled() {
+ MutableSegmentImpl segment = createSegment(true);
+ try {
+ DataSource dataSource = segment.getDataSource("event");
+ assertNotNull(dataSource);
+ assertTrue(dataSource instanceof MutableOpenStructDataSource);
+ } finally {
+ segment.destroy();
+ }
+ }
+}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
index b8795f02c53..03f97a38213 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java
@@ -38,6 +38,7 @@ import org.apache.pinot.common.request.context.FilterContext;
import org.apache.pinot.common.request.context.predicate.EqPredicate;
import
org.apache.pinot.segment.local.segment.index.readers.json.ImmutableJsonIndexReader;
import org.apache.pinot.segment.spi.V1Constants;
+import org.apache.pinot.segment.spi.index.metadata.ColumnMetadataImpl;
import org.apache.pinot.segment.spi.memory.PinotDataBuffer;
import org.apache.pinot.spi.config.table.FieldConfig;
import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
@@ -219,7 +220,19 @@ public class OpenStructColumnSplitterTest {
}
s.seal();
String sparseCol = OpenStructNaming.sparseColumnName("metrics");
- assertTrue(s.getMaterializedColumnMetadata().containsKey(sparseCol));
+ PropertiesConfiguration props =
s.getMaterializedColumnMetadata().get(sparseCol);
+ assertNotNull(props);
+
+ // Regression: the hand-rolled metadata this replaced never wrote
FieldConfig.EncodingType or
+ // LENGTH_OF_LONGEST_ELEMENT, and approximated CARDINALITY as the non-null
doc count (1 here) rather than the
+ // real distinct-value count (2: the one non-empty json blob, plus the ""
default shared by the 9 absent docs).
+ // Assert the values that only the real
addColumnMetadataInfo()/statsCollector path can produce.
+ ColumnMetadataImpl metadata =
ColumnMetadataImpl.fromPropertiesConfiguration(props, 10, sparseCol);
+ assertEquals(metadata.getFieldSpec().getDataType(), DataType.STRING);
+ assertFalse(metadata.hasDictionary());
+ assertEquals(metadata.getForwardIndexEncoding(),
FieldConfig.EncodingType.RAW);
+ assertEquals(metadata.getCardinality(), 2);
+ assertTrue(metadata.getLengthOfLongestElement() > 0);
}
@Test
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSourceTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSourceTest.java
index 4572ad11207..a82ec18efe1 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSourceTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSourceTest.java
@@ -18,21 +18,60 @@
*/
package org.apache.pinot.segment.local.segment.index.openstruct;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.commons.io.FileUtils;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
import org.apache.pinot.segment.local.segment.index.datasource.NullDataSource;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.local.segment.readers.PinotSegmentRecordReader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.datasource.DataSourceMetadata;
+import
org.apache.pinot.segment.spi.datasource.OpenStructDataSource.MapValueReader;
import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.segment.spi.index.reader.ForwardIndexReader;
+import org.apache.pinot.segment.spi.index.reader.ForwardIndexReaderContext;
import org.apache.pinot.segment.spi.index.reader.JsonIndexReader;
+import org.apache.pinot.segment.spi.index.reader.NullValueVectorReader;
+import org.apache.pinot.spi.config.table.FieldConfig;
+import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.DimensionFieldSpec;
+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.utils.JsonUtils;
+import org.apache.pinot.spi.utils.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
-import static org.testng.Assert.*;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
public class ImmutableOpenStructDataSourceTest {
@@ -232,6 +271,57 @@ public class ImmutableOpenStructDataSourceTest {
return ds;
}
+ /// Dense child columns are written dictionary-encoded
(`OpenStructColumnSplitter#writeColumnIndexes` builds the
+ /// dictionary and the forward index from the same `useDictionary` flag), so
the mock must report
+ /// `isDictionaryEncoded()` — not merely expose a dictionary — to match a
real segment.
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static DataSource mockDenseDataSource(DataType storedType, Object
valueAtDoc0, boolean nullAtDoc1) {
+ DataSource ds = mock(DataSource.class);
+ ForwardIndexReader fwdReader = mock(ForwardIndexReader.class);
+ ForwardIndexReaderContext ctx = mock(ForwardIndexReaderContext.class);
+ when(fwdReader.createContext()).thenReturn(ctx);
+ when(fwdReader.getStoredType()).thenReturn(storedType);
+ when(fwdReader.isSingleValue()).thenReturn(true);
+ when(fwdReader.isDictionaryEncoded()).thenReturn(true);
+
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.getValueType()).thenReturn(storedType);
+ when(fwdReader.getDictId(eq(0), eq(ctx))).thenReturn(42);
+ when(dictionary.get(42)).thenReturn(valueAtDoc0);
+ when(ds.getDictionary()).thenReturn(dictionary);
+
+ when(ds.getForwardIndex()).thenReturn(fwdReader);
+
+ NullValueVectorReader nullReader = mock(NullValueVectorReader.class);
+ when(nullReader.isNull(0)).thenReturn(false);
+ when(nullReader.isNull(1)).thenReturn(nullAtDoc1);
+ when(ds.getNullValueVector()).thenReturn(nullReader);
+
+ return ds;
+ }
+
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static DataSource mockSparseDataSource(String jsonAtDoc0, boolean
nullAtDoc1) {
+ DataSource ds = mock(DataSource.class);
+ ForwardIndexReader fwdReader = mock(ForwardIndexReader.class);
+ ForwardIndexReaderContext ctx = mock(ForwardIndexReaderContext.class);
+ when(fwdReader.createContext()).thenReturn(ctx);
+ when(fwdReader.getStoredType()).thenReturn(DataType.STRING);
+ when(fwdReader.isSingleValue()).thenReturn(true);
+ when(fwdReader.isDictionaryEncoded()).thenReturn(false);
+ when(fwdReader.getString(eq(0), eq(ctx))).thenReturn(jsonAtDoc0);
+ when(fwdReader.getString(eq(1), eq(ctx))).thenReturn("");
+ when(ds.getForwardIndex()).thenReturn(fwdReader);
+ when(ds.getDictionary()).thenReturn(null);
+
+ NullValueVectorReader nullReader = mock(NullValueVectorReader.class);
+ when(nullReader.isNull(0)).thenReturn(false);
+ when(nullReader.isNull(1)).thenReturn(nullAtDoc1);
+ when(ds.getNullValueVector()).thenReturn(nullReader);
+
+ return ds;
+ }
+
@Test
public void testManifestKeyGetsVirtualSparseDataSource() {
String[] blobs = {"{\"region\":\"us\"}", null};
@@ -310,4 +400,301 @@ public class ImmutableOpenStructDataSourceTest {
openStructSpec("event"), Map.of(), null, 5, null);
assertNull(ds.getSparseJsonIndex());
}
+
+ @Test
+ public void testGetMapValueDenseOnly() {
+ DataSource clicksDs = mockDenseDataSource(DataType.INT, 10, true);
+ DataSource nameDs = mockDenseDataSource(DataType.STRING, "hello", false);
+
+ Map<String, DataSource> perKey = new HashMap<>();
+ perKey.put("clicks", clicksDs);
+ perKey.put("name", nameDs);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), perKey, null, 2, null);
+
+ Map<String, Object> doc0 = ds.getMapValue(0);
+ assertNotNull(doc0);
+ assertEquals(doc0.get("clicks"), 10);
+ assertEquals(doc0.get("name"), "hello");
+ }
+
+ /// Raw (no-dictionary) dense child columns are a real, config-selectable
encoding
+ /// (`OpenStructColumnSplitter#writeColumnIndexes` with `useDictionary ==
false`), and they take a different
+ /// read path than the dictionary-encoded case: the per-type dispatch inside
the forward-index read rather
+ /// than a dictId lookup. Cover every stored type OPEN_STRUCT can
materialize.
+ @DataProvider(name = "rawStoredTypes")
+ public static Object[][] rawStoredTypes() {
+ return new Object[][]{
+ {DataType.INT, 7},
+ {DataType.LONG, 7L},
+ {DataType.FLOAT, 1.5f},
+ {DataType.DOUBLE, 2.5d},
+ {DataType.BIG_DECIMAL, new BigDecimal("3.25")},
+ {DataType.STRING, "raw"},
+ {DataType.BYTES, new byte[]{1, 2, 3}}
+ };
+ }
+
+ @Test(dataProvider = "rawStoredTypes")
+ public void testGetMapValueRawDenseColumnPerStoredType(DataType storedType,
Object expected) {
+ DataSource rawDs = mockRawDenseDataSource(storedType, expected);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of("k", rawDs), null, 2, null);
+
+ Map<String, Object> doc0 = ds.getMapValue(0);
+ assertNotNull(doc0);
+ if (storedType == DataType.BYTES) {
+ assertEquals((byte[]) doc0.get("k"), (byte[]) expected);
+ } else {
+ assertEquals(doc0.get("k"), expected);
+ }
+ }
+
+ /// No dictionary and a raw forward index — the combination
`mockDenseDataSource` never produces.
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static DataSource mockRawDenseDataSource(DataType storedType, Object
valueAtDoc0) {
+ DataSource ds = mock(DataSource.class);
+ ForwardIndexReader fwdReader = mock(ForwardIndexReader.class);
+ ForwardIndexReaderContext ctx = mock(ForwardIndexReaderContext.class);
+ when(fwdReader.createContext()).thenReturn(ctx);
+ when(fwdReader.getStoredType()).thenReturn(storedType);
+ when(fwdReader.isSingleValue()).thenReturn(true);
+ when(fwdReader.isDictionaryEncoded()).thenReturn(false);
+ when(ds.getDictionary()).thenReturn(null);
+
+ switch (storedType) {
+ case INT:
+ when(fwdReader.getInt(eq(0), eq(ctx))).thenReturn((Integer)
valueAtDoc0);
+ break;
+ case LONG:
+ when(fwdReader.getLong(eq(0), eq(ctx))).thenReturn((Long) valueAtDoc0);
+ break;
+ case FLOAT:
+ when(fwdReader.getFloat(eq(0), eq(ctx))).thenReturn((Float)
valueAtDoc0);
+ break;
+ case DOUBLE:
+ when(fwdReader.getDouble(eq(0), eq(ctx))).thenReturn((Double)
valueAtDoc0);
+ break;
+ case BIG_DECIMAL:
+ when(fwdReader.getBigDecimal(eq(0), eq(ctx))).thenReturn((BigDecimal)
valueAtDoc0);
+ break;
+ case STRING:
+ when(fwdReader.getString(eq(0), eq(ctx))).thenReturn((String)
valueAtDoc0);
+ break;
+ case BYTES:
+ when(fwdReader.getBytes(eq(0), eq(ctx))).thenReturn((byte[])
valueAtDoc0);
+ break;
+ default:
+ throw new IllegalArgumentException("Unhandled stored type in test
fixture: " + storedType);
+ }
+
+ when(ds.getForwardIndex()).thenReturn(fwdReader);
+
+ NullValueVectorReader nullReader = mock(NullValueVectorReader.class);
+ when(nullReader.isNull(0)).thenReturn(false);
+ when(nullReader.isNull(1)).thenReturn(true);
+ when(ds.getNullValueVector()).thenReturn(nullReader);
+
+ return ds;
+ }
+
+ @Test
+ public void testGetMapValueNullDoc() {
+ DataSource clicksDs = mockDenseDataSource(DataType.INT, 10, true);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of("clicks", clicksDs), null, 2, null);
+
+ // doc 1 has null for clicks → no keys → null map
+ Map<String, Object> doc1 = ds.getMapValue(1);
+ assertNull(doc1);
+ }
+
+ @Test
+ public void testGetMapValueWithSparse() {
+ DataSource clicksDs = mockDenseDataSource(DataType.INT, 10, true);
+ DataSource sparseDs = mockSparseDataSource("{\"rare_key\":\"val\"}", true);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of("clicks", clicksDs), sparseDs, 2,
null);
+
+ Map<String, Object> doc0 = ds.getMapValue(0);
+ assertNotNull(doc0);
+ assertEquals(doc0.get("clicks"), 10);
+ assertEquals(doc0.get("rare_key"), "val");
+ }
+
+ /// A dense child key literally named the same as the parent OPEN_STRUCT
column (e.g. `{"event": {"event": 1,
+ /// "region": "us"}}`) must not shadow the sparse blob reader in the per-key
reader cache: the sparse reader is
+ /// keyed by parent field name, same as this dense key. Exercised through
[ImmutableOpenStructDataSource
+ /// #openMapValueReader()] directly — the bug only lives in the cached
reader a real scan uses, not in the
+ /// unscoped [ImmutableOpenStructDataSource#getMapValue(int)] this test's
data flowed through before it started
+ /// delegating to the same cached reader.
+ @Test
+ public void
testOpenMapValueReaderChildKeyMatchingParentNameDoesNotShadowSparseReader()
throws Exception {
+ DataSource eventKeyDs = mockDenseDataSource(DataType.INT, 5, false);
+ DataSource sparseDs = mockSparseDataSource("{\"region\":\"us\"}", false);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of("event", eventKeyDs), sparseDs, 2,
null);
+
+ try (MapValueReader reader = ds.openMapValueReader()) {
+ Map<String, Object> doc0 = reader.getMapValue(0);
+ assertNotNull(doc0);
+ assertEquals(doc0.get("event"), 5);
+ assertEquals(doc0.get("region"), "us");
+ }
+ }
+
+ @Test
+ public void testGetMapValueMalformedSparseJsonThrowsWithDocContext() {
+ DataSource sparseDs = mockSparseDataSource("not-json", false);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of(), sparseDs, 2, null);
+
+ RuntimeException e = expectThrows(RuntimeException.class, () ->
ds.getMapValue(0));
+ assertTrue(e.getMessage().contains("docId 0"));
+ }
+
+ @Test
+ public void
testOpenMapValueReaderCloseCollectsAndSuppressesReaderCloseFailures() throws
Exception {
+ ForwardIndexReaderContext ctx1 = mock(ForwardIndexReaderContext.class);
+ doThrow(new IOException("reader1 failed")).when(ctx1).close();
+ DataSource ds1 = mockDenseDataSourceWithFailingClose(1, ctx1);
+
+ ForwardIndexReaderContext ctx2 = mock(ForwardIndexReaderContext.class);
+ doThrow(new IOException("reader2 failed")).when(ctx2).close();
+ DataSource ds2 = mockDenseDataSourceWithFailingClose(2, ctx2);
+
+ Map<String, DataSource> perKey = new HashMap<>();
+ perKey.put("a", ds1);
+ perKey.put("b", ds2);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), perKey, null, 1, null);
+
+ MapValueReader reader = ds.openMapValueReader();
+ reader.getMapValue(0);
+
+ IOException e = expectThrows(IOException.class, reader::close);
+ assertEquals(e.getSuppressed().length, 1);
+ }
+
+ /// A dense INT column whose forward index reader context throws on close,
so [PinotSegmentColumnReader#close]
+ /// propagates a failure to exercise the caching reader's
close()-collects-failures behavior.
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private static DataSource mockDenseDataSourceWithFailingClose(int
valueAtDoc0, ForwardIndexReaderContext ctx) {
+ DataSource ds = mock(DataSource.class);
+ ForwardIndexReader fwdReader = mock(ForwardIndexReader.class);
+ when(fwdReader.createContext()).thenReturn(ctx);
+ when(fwdReader.getStoredType()).thenReturn(DataType.INT);
+ when(fwdReader.isSingleValue()).thenReturn(true);
+ when(fwdReader.isDictionaryEncoded()).thenReturn(false);
+ when(fwdReader.getInt(eq(0), eq(ctx))).thenReturn(valueAtDoc0);
+ when(ds.getDictionary()).thenReturn(null);
+ when(ds.getForwardIndex()).thenReturn(fwdReader);
+
+ NullValueVectorReader nullReader = mock(NullValueVectorReader.class);
+ when(nullReader.isNull(0)).thenReturn(false);
+ when(ds.getNullValueVector()).thenReturn(nullReader);
+
+ return ds;
+ }
+
+ @Test
+ public void testGetMapValueSparseOnlyNullDoc() {
+ DataSource sparseDs = mockSparseDataSource("{\"rare_key\":\"val\"}", true);
+
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of(), sparseDs, 2, null);
+
+ // doc 1: sparse is null
+ Map<String, Object> doc1 = ds.getMapValue(1);
+ assertNull(doc1);
+ }
+
+ @Test
+ public void testGetMapValueEmptySegment() {
+ ImmutableOpenStructDataSource ds = new ImmutableOpenStructDataSource(
+ openStructSpec("event"), Map.of(), null, 0, null);
+
+ assertNull(ds.getMapValue(0));
+ }
+
+ /// Builds and seals a real OPEN_STRUCT segment (one dense key, one sparse
key) and reads it back
+ /// with [PinotSegmentRecordReader], which drives
[ImmutableOpenStructDataSource#getMapValue]
+ /// through
[org.apache.pinot.segment.spi.IndexSegment#getPhysicalColumnNames], the
schema-derived
+ /// column set every real caller uses — never the materialized child columns
+ /// (`event$clicks`/`event$__sparse__`) by name. Every other test in this
class drives
+ /// `getMapValue`/`openMapValueReader` against mocked [DataSource]s; this is
the one real
+ /// sealed-segment round trip, so a regression in how `ImmutableSegmentImpl`
wires the parent
+ /// [DataSource] or in the
[ImmutableOpenStructDataSource.CachingMapValueReader] reader cache can't
+ /// hide behind a mock that always answers the way the test expects (e.g. in
the per-key reader
+ /// cache `openMapValueReader()` builds for a sequential scan).
+ @Test
+ public void testSealedSegmentRoundTripThroughRecordReader()
+ throws Exception {
+ File tempDir =
Files.createTempDirectory("ImmutableOpenStructDataSourceTest").toFile();
+ try {
+ Map<String, FieldSpec> children = new HashMap<>();
+ children.put("clicks", new DimensionFieldSpec("clicks", DataType.LONG,
true));
+ ComplexFieldSpec fieldSpec = new ComplexFieldSpec("event",
DataType.OPEN_STRUCT, true, children);
+ Schema schema = new
Schema.SchemaBuilder().setSchemaName("testRoundTrip").addField(fieldSpec).build();
+
+ // "clicks" present on 9/10 docs -> dense; "region" present on 1/10 ->
sparse.
+ OpenStructIndexConfig osConfig = new OpenStructIndexConfig(false, null,
-1, null, 0.5, null, null);
+ ObjectNode indexes = JsonUtils.newObjectNode();
+ indexes.set("open_struct", JsonUtils.objectToJsonNode(osConfig));
+ FieldConfig eventFieldConfig = new
FieldConfig.Builder("event").withIndexes(indexes).build();
+ TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
+ .setTableName("testRoundTrip")
+ .setFieldConfigList(List.of(eventFieldConfig))
+ .setNullHandlingEnabled(true)
+ .build();
+
+ int numDocs = 10;
+ List<GenericRow> rows = new ArrayList<>(numDocs);
+ List<Map<String, Object>> expected = new ArrayList<>(numDocs);
+ for (int docId = 0; docId < numDocs; docId++) {
+ Map<String, Object> value = new HashMap<>();
+ if (docId != 9) {
+ value.put("clicks", (long) docId);
+ }
+ if (docId == 3) {
+ value.put("region", "us");
+ }
+ GenericRow row = new GenericRow();
+ row.putValue("event", value);
+ rows.add(row);
+ expected.add(value.isEmpty() ? null : value);
+ }
+
+ SegmentGeneratorConfig segmentConfig = new
SegmentGeneratorConfig(tableConfig, schema);
+ segmentConfig.setOutDir(tempDir.getAbsolutePath());
+ segmentConfig.setSegmentName("testRoundTripSegment");
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentConfig, new GenericRowRecordReader(rows));
+ driver.build();
+
+ ImmutableSegment segment =
ImmutableSegmentLoader.load(driver.getOutputDirectory(), ReadMode.mmap);
+ try {
+ assertTrue(segment.getDataSource("event") instanceof
ImmutableOpenStructDataSource);
+ try (PinotSegmentRecordReader recordReader = new
PinotSegmentRecordReader()) {
+ recordReader.init(segment);
+ for (int docId = 0; docId < numDocs; docId++) {
+ GenericRow row = new GenericRow();
+ recordReader.next(row);
+ assertEquals(row.getValue("event"), expected.get(docId), "docId="
+ docId);
+ }
+ }
+ } finally {
+ segment.destroy();
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
index b08eb2ae1d0..d9281c8dd58 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndexTest.java
@@ -29,6 +29,7 @@ import
org.apache.pinot.segment.spi.memory.PinotDataBufferMemoryManager;
import org.apache.pinot.spi.config.table.OpenStructIndexConfig;
import org.apache.pinot.spi.data.ComplexFieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.metrics.PinotMeter;
import org.apache.pinot.spi.utils.JsonUtils;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
@@ -42,6 +43,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNull;
@@ -178,39 +180,45 @@ public class MutableOpenStructIndexTest {
throws IOException {
ServerMetrics metrics = mock(ServerMetrics.class);
assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
- OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
- // Unmappable value on a fresh key: falls back to STRING and meters an
inference failure.
- idx.index(0, Map.of("req-42", Map.of("a", 1)));
- // Unmappable value on a key already typed LONG: dropped by coercion,
metered there only.
- idx.index(1, Map.of("clicks", 5L));
- idx.index(2, Map.of("clicks", Map.of("a", 1)));
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ // Unmappable value on a fresh key: falls back to STRING and meters an
inference failure.
+ idx.index(0, Map.of("req-42", Map.of("a", 1)));
+ // Unmappable value on a key already typed LONG: dropped by coercion,
metered there only.
+ idx.index(1, Map.of("clicks", 5L));
+ idx.index(2, Map.of("clicks", Map.of("a", 1)));
+ }
verify(metrics).addMeteredTableValue("testTable_REALTIME", "metrics",
- ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1L);
+ ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1L, null);
verify(metrics).addMeteredTableValue("testTable_REALTIME", "metrics",
- ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1L);
- verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$req-42"), any(), anyLong());
- verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$clicks"), any(), anyLong());
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1L, null);
+ verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$req-42"), any(), anyLong(), any());
+ verify(metrics, never()).addMeteredTableValue(anyString(),
eq("metrics$clicks"), any(), anyLong(), any());
} finally {
ServerMetrics.deregister();
}
}
/// A later unmappable value on a key whose type fell back to STRING is
stored as its serialized
- /// form, so it is metered every time — not just on the first sighting that
established the type.
+ /// form, so it is metered every time — live, per value — not just on the
first sighting that
+ /// established the type.
@Test
- public void testInferenceFailuresMeteredPerValueOnStringFallbackKey()
+ public void testInferenceFailuresMeterLivePerValue()
throws IOException {
ServerMetrics metrics = mock(ServerMetrics.class);
assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
- try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME", openStructSpec(),
- OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
- for (int docId = 0; docId < 3; docId++) {
- idx.index(docId, Map.of("payload", Map.of("a", docId)));
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ for (int docId = 0; docId < 3; docId++) {
+ idx.index(docId, Map.of("payload", Map.of("a", docId)));
+ }
+
+ verify(metrics,
times(3)).addMeteredTableValue(eq("testTable_REALTIME"), eq("metrics"),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES), eq(1L),
any());
}
- verify(metrics, times(3)).addMeteredTableValue("testTable_REALTIME",
"metrics",
- ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES, 1L);
} finally {
ServerMetrics.deregister();
}
@@ -277,4 +285,113 @@ public class MutableOpenStructIndexTest {
ServerMetrics.deregister();
}
}
+
+ /// Coercion failures are metered live, per value, and the returned
[PinotMeter] is cached and
+ /// reused on later occurrences instead of re-resolving the metric
name/registry each time.
+ @Test
+ public void testCoercionFailuresMeterLiveWithReusedMeter()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ PinotMeter reusedMeter = mock(PinotMeter.class);
+ when(metrics.addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES), anyLong(),
any())).thenReturn(reusedMeter);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ // Establish "clicks" as LONG, then feed three unmappable values that
fail coercion.
+ idx.index(0, Map.of("clicks", 5L));
+ idx.index(1, Map.of("clicks", Map.of("a", 1)));
+
+ // First occurrence marks live, with no meter to reuse yet.
+ verify(metrics, times(1)).addMeteredTableValue("testTable_REALTIME",
"metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1L, null);
+
+ idx.index(2, Map.of("clicks", Map.of("a", 2)));
+ idx.index(3, Map.of("clicks", Map.of("a", 3)));
+ }
+
+ verify(metrics, times(3)).addMeteredTableValue(eq("testTable_REALTIME"),
eq("metrics"),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES), eq(1L), any());
+ // The 2nd and 3rd occurrences reuse the meter the 1st returned.
+ verify(metrics, times(2)).addMeteredTableValue("testTable_REALTIME",
"metrics",
+ ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES, 1L, reusedMeter);
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ @Test
+ public void testNoFailureMetersEmittedWhenNoFailures()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ idx.index(0, Map.of("clicks", 5L, "country", "US"));
+ idx.index(1, Map.of("clicks", 7L));
+ }
+
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES), anyLong(),
any());
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES), anyLong(),
any());
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_IGNORED_KEY_DROPS), anyLong());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ @Test
+ public void testInferenceCheckSkippedForNonStringKeys()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ // "clicks" establishes as LONG. Unmappable values on it are a
coercion concern only —
+ // inference must never run for a key whose established type is not
STRING.
+ idx.index(0, Map.of("clicks", 5L));
+ idx.index(1, Map.of("clicks", Map.of("a", 1)));
+ idx.index(2, Map.of("clicks", Map.of("a", 2)));
+ }
+
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES), anyLong(),
any());
+ verify(metrics, times(2)).addMeteredTableValue(eq("testTable_REALTIME"),
eq("metrics"),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES), eq(1L), any());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
+
+ /// A STRING-established key does run the per-row inference check, so this
pins the other side of
+ /// that branch: a later *inferable* value must not be counted. Without the
`inferDataType == null`
+ /// guard inside meterIfUninferable, every row on such a key would meter a
failure.
+ @Test
+ public void testInferenceCheckOnStringKeyDoesNotMeterInferableValues()
+ throws IOException {
+ ServerMetrics metrics = mock(ServerMetrics.class);
+ assertTrue(ServerMetrics.register(metrics), "another ServerMetrics is
already registered");
+ try {
+ try (MutableOpenStructIndex idx = new MutableOpenStructIndex("metrics",
"testTable_REALTIME",
+ openStructSpec(), OpenStructIndexConfig.DEFAULT, _memMgr, 100)) {
+ // "country" has no child spec and infers as STRING, so
needsInferenceCheck() is true and
+ // meterIfUninferable runs on every later row — but both values infer
cleanly.
+ idx.index(0, Map.of("country", "US"));
+ idx.index(1, Map.of("country", "CA"));
+ idx.index(2, Map.of("country", "MX"));
+ }
+
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_INFERENCE_FAILURES), anyLong());
+ verify(metrics, never()).addMeteredTableValue(anyString(), anyString(),
+ eq(ServerMeter.OPEN_STRUCT_TYPE_COERCION_FAILURES), anyLong());
+ } finally {
+ ServerMetrics.deregister();
+ }
+ }
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
index 2e300de3f65..1c91e348e9d 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexTypeTest.java
@@ -141,6 +141,36 @@ public class OpenStructIndexTypeTest {
StandardIndexes.openStruct().validate(fieldIndexConfigs, openStructSpec,
null);
}
+ /// A declared MAP child key passes ColumnDataType conversion (used to be
the only check) but has no DIMENSION
+ /// case in FieldSpec#getDefaultNullValue, the call allocateKeyColumn()
actually makes — must be rejected here
+ /// instead of throwing uncaught on the first row ingested for the key.
+ @Test
+ public void testValidateRejectsMapChildKeyType()
+ throws Exception {
+ OpenStructIndexConfig config = new OpenStructIndexConfig(false, null, -1,
null, 0.5, null, null);
+ FieldIndexConfigs fieldIndexConfigs =
+ new FieldIndexConfigs.Builder().add(StandardIndexes.openStruct(),
config).build();
+ FieldSpec openStructSpec = new ComplexFieldSpec("payload",
FieldSpec.DataType.OPEN_STRUCT, true,
+ Map.of("nested", new ComplexFieldSpec("nested",
FieldSpec.DataType.MAP, true, Map.of())));
+
+ assertThrows(IllegalStateException.class,
+ () -> StandardIndexes.openStruct().validate(fieldIndexConfigs,
openStructSpec, null));
+ }
+
+ /// Same gap as MAP, for a nested OPEN_STRUCT child key.
+ @Test
+ public void testValidateRejectsOpenStructChildKeyType()
+ throws Exception {
+ OpenStructIndexConfig config = new OpenStructIndexConfig(false, null, -1,
null, 0.5, null, null);
+ FieldIndexConfigs fieldIndexConfigs =
+ new FieldIndexConfigs.Builder().add(StandardIndexes.openStruct(),
config).build();
+ FieldSpec openStructSpec = new ComplexFieldSpec("payload",
FieldSpec.DataType.OPEN_STRUCT, true,
+ Map.of("nested", new ComplexFieldSpec("nested",
FieldSpec.DataType.OPEN_STRUCT, true, Map.of())));
+
+ assertThrows(IllegalStateException.class,
+ () -> StandardIndexes.openStruct().validate(fieldIndexConfigs,
openStructSpec, null));
+ }
+
@Test
public void testValidateSkipsIgnoredKeyChecksWhenIndexDisabled()
throws Exception {
diff --git
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/OpenStructDataSource.java
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/OpenStructDataSource.java
index 3be2f7f35aa..9dfdbbe6853 100644
---
a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/OpenStructDataSource.java
+++
b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/datasource/OpenStructDataSource.java
@@ -18,6 +18,8 @@
*/
package org.apache.pinot.segment.spi.datasource;
+import java.io.Closeable;
+import java.io.IOException;
import java.util.Map;
import javax.annotation.Nullable;
import org.apache.pinot.segment.spi.index.column.ColumnIndexContainer;
@@ -95,6 +97,30 @@ public interface OpenStructDataSource extends DataSource {
"Per-doc OPEN_STRUCT map reconstruction is not supported by this
implementation");
}
+ /// Opens a [MapValueReader] scoped to one sequential scan of this data
source, e.g. a full
+ /// segment read by the seal path or a minion task's record reader.
Implementations backed by
+ /// per-key forward-index readers (immutable segments) can cache those
readers for the life of
+ /// the returned instance instead of reconstructing one per key per doc; the
default just
+ /// delegates each call to [#getMapValue(int)], which is already O(1) for
in-memory mutable
+ /// segments.
+ ///
+ /// Not thread-safe: the returned reader is for one single-threaded
sequential scan, not for
+ /// concurrent callers sharing this data source.
+ default MapValueReader openMapValueReader() {
+ return this::getMapValue;
+ }
+
+ /// A [#getMapValue(int)] reader scoped to one scan; see
[#openMapValueReader()].
+ interface MapValueReader extends Closeable {
+ @Nullable
+ Map<String, Object> getMapValue(int docId);
+
+ @Override
+ default void close()
+ throws IOException {
+ }
+ }
+
/// Whether the per-key dictionary's contents correspond exactly to the
values readable from
/// the key column (absent docs folded as the default included) — i.e.
dictionary-based
/// MIN/MAX/DISTINCTCOUNT over it matches a full scan. Sealed segments build
dictionaries
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
