raghavyadav01 commented on code in PR #19093:
URL: https://github.com/apache/pinot/pull/19093#discussion_r4018897684
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java:
##########
@@ -140,6 +144,141 @@ public JsonIndexReader getSparseJsonIndex() {
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : 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 (_sparseDataSource != null) {
+ Object sparseValue = readValue(_fieldSpec.getName(), _sparseDataSource,
docId);
+ if (sparseValue instanceof String) {
+ String json = (String) sparseValue;
+ if (!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;
+ }
+
+ /// Reads the value of `key` at `docId`, or `null` when the doc is null or
the column has no
+ /// forward index. Delegates the null-vector check and the dictionary/raw
per-type read dispatch to
+ /// [PinotSegmentColumnReader] rather than re-deriving them here, so this
path cannot drift from the
+ /// reader every other column read in the engine already goes through.
OPEN_STRUCT child columns are
+ /// always single-valued, hence the 0 maxNumValuesPerMVEntry.
+ @Nullable
+ private static Object readValue(String key, DataSource dataSource, int
docId) {
+ ForwardIndexReader<?> fwdReader = dataSource.getForwardIndex();
+ if (fwdReader == null) {
+ return null;
+ }
+ try (PinotSegmentColumnReader reader = new PinotSegmentColumnReader(key,
fwdReader, dataSource.getDictionary(),
+ dataSource.getNullValueVector(), 0)) {
+ return reader.isNull(docId) ? null : reader.getValue(docId);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to read value from OPEN_STRUCT key
forward index", e);
+ }
+ }
+
+ @Override
+ public MapValueReader openMapValueReader() {
+ return new CachingMapValueReader();
+ }
+
+ /// Caches one [PinotSegmentColumnReader] per key for the life of the
reader, instead of the
+ /// per-call construct-and-close [#readValue]/[#getMapValue] does. 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<>();
+
+ @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 (_sparseDataSource != null) {
+ Object sparseValue = readValue(_fieldSpec.getName(),
_sparseDataSource, docId);
Review Comment:
The reader cache is keyed by child key name, but the sparse blob is cached
under the parent column name. A child key that matches the parent (`{"event":
{"event": 1, "region": "us"}}`) hits the dense reader here, so `instanceof
String` fails and every sparse key is silently dropped. Could the sparse reader
get its own field instead of sharing the map?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java:
##########
@@ -140,6 +144,141 @@ public JsonIndexReader getSparseJsonIndex() {
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : 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 (_sparseDataSource != null) {
+ Object sparseValue = readValue(_fieldSpec.getName(), _sparseDataSource,
docId);
+ if (sparseValue instanceof String) {
+ String json = (String) sparseValue;
+ if (!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;
+ }
+
+ /// Reads the value of `key` at `docId`, or `null` when the doc is null or
the column has no
+ /// forward index. Delegates the null-vector check and the dictionary/raw
per-type read dispatch to
+ /// [PinotSegmentColumnReader] rather than re-deriving them here, so this
path cannot drift from the
+ /// reader every other column read in the engine already goes through.
OPEN_STRUCT child columns are
+ /// always single-valued, hence the 0 maxNumValuesPerMVEntry.
+ @Nullable
+ private static Object readValue(String key, DataSource dataSource, int
docId) {
+ ForwardIndexReader<?> fwdReader = dataSource.getForwardIndex();
+ if (fwdReader == null) {
+ return null;
+ }
+ try (PinotSegmentColumnReader reader = new PinotSegmentColumnReader(key,
fwdReader, dataSource.getDictionary(),
+ dataSource.getNullValueVector(), 0)) {
+ return reader.isNull(docId) ? null : reader.getValue(docId);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to read value from OPEN_STRUCT key
forward index", e);
+ }
+ }
+
+ @Override
+ public MapValueReader openMapValueReader() {
+ return new CachingMapValueReader();
+ }
+
+ /// Caches one [PinotSegmentColumnReader] per key for the life of the
reader, instead of the
+ /// per-call construct-and-close [#readValue]/[#getMapValue] does. 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<>();
+
+ @SuppressWarnings("unchecked")
+ @Nullable
+ @Override
+ public Map<String, Object> getMapValue(int docId) {
Review Comment:
This is the same body as `getMapValue` above, and the two copies have
already drifted — the cached `readValue` dropped the exception wrapper that
adds read context. Could the outer one just delegate: `try (MapValueReader r =
openMapValueReader()) { return r.getMapValue(docId); }`?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/OpenStructIndexType.java:
##########
@@ -88,6 +89,36 @@ public void validate(FieldIndexConfigs indexConfigs,
FieldSpec fieldSpec, TableC
fieldSpec.getName());
validatePerKeyIndexes(config);
validateIgnoredKeys(config, fieldSpec);
+ if (fieldSpec instanceof ComplexFieldSpec) {
+ validateChildFieldSpecTypes((ComplexFieldSpec) fieldSpec);
+ }
+ }
+ }
+
+ /// Rejects a declared child key type that the consuming and sealed-build
paths cannot coerce a
+ /// value to (e.g. STRUCT, LIST, nested OPEN_STRUCT, UNKNOWN —
[ColumnDataType#fromDataTypeSV] or
+ /// the resulting [ColumnDataType#toPinotDataType] throws 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 {
Review Comment:
`fromDataTypeSV` maps MAP to MAP and OPEN_STRUCT to OBJECT, and
`toPinotDataType()` accepts both, so a child declared as either passes this
check. It then reaches `allocateKeyColumn`, where
`FieldSpec.getDefaultNullValue(DIMENSION, MAP, null)` throws uncaught on the
consuming thread — the same failure this was meant to close. Worth restricting
to the types a key column can actually store?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java:
##########
@@ -140,6 +144,141 @@ public JsonIndexReader getSparseJsonIndex() {
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : 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 (_sparseDataSource != null) {
+ Object sparseValue = readValue(_fieldSpec.getName(), _sparseDataSource,
docId);
+ if (sparseValue instanceof String) {
+ String json = (String) sparseValue;
+ if (!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;
+ }
+
+ /// Reads the value of `key` at `docId`, or `null` when the doc is null or
the column has no
+ /// forward index. Delegates the null-vector check and the dictionary/raw
per-type read dispatch to
+ /// [PinotSegmentColumnReader] rather than re-deriving them here, so this
path cannot drift from the
+ /// reader every other column read in the engine already goes through.
OPEN_STRUCT child columns are
+ /// always single-valued, hence the 0 maxNumValuesPerMVEntry.
+ @Nullable
+ private static Object readValue(String key, DataSource dataSource, int
docId) {
+ ForwardIndexReader<?> fwdReader = dataSource.getForwardIndex();
+ if (fwdReader == null) {
+ return null;
+ }
+ try (PinotSegmentColumnReader reader = new PinotSegmentColumnReader(key,
fwdReader, dataSource.getDictionary(),
+ dataSource.getNullValueVector(), 0)) {
+ return reader.isNull(docId) ? null : reader.getValue(docId);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to read value from OPEN_STRUCT key
forward index", e);
+ }
+ }
+
+ @Override
+ public MapValueReader openMapValueReader() {
+ return new CachingMapValueReader();
+ }
+
+ /// Caches one [PinotSegmentColumnReader] per key for the life of the
reader, instead of the
+ /// per-call construct-and-close [#readValue]/[#getMapValue] does. 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<>();
+
+ @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 (_sparseDataSource != null) {
+ Object sparseValue = readValue(_fieldSpec.getName(),
_sparseDataSource, docId);
+ if (sparseValue instanceof String) {
+ String json = (String) sparseValue;
+ if (!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) {
+ PinotSegmentColumnReader reader = _readers.computeIfAbsent(key, k ->
createReader(k, dataSource));
+ if (reader == null) {
+ return null;
+ }
+ return 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 {
+ for (PinotSegmentColumnReader reader : _readers.values()) {
Review Comment:
If one reader's `close()` throws, the rest leak their forward-index
contexts. Same shape in `PinotSegmentRecordReader#close`, where a column-reader
failure skips the map-value readers entirely. Worth collecting and suppressing
so the loop always finishes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
