tarun11Mavani commented on code in PR #19093:
URL: https://github.com/apache/pinot/pull/19093#discussion_r4023964659
##########
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:
isCoercible now calls FieldSpec.getDefaultNullValue(DIMENSION, storedType,
null) directly — the exact call allocateKeyColumn makes — instead of the looser
ColumnDataType conversion.
##########
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:
done.
--
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]
