You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "raghavyadav01 (via GitHub)" <gi...@apache.org> on 2026/09/08 17:46:16 UTC
Re: [PR] Optimise the OPEN_STRUCT per-row consumption path, and follow-up fixes from #18643 review [pinot]
raghavyadav01 commented on code in PR #19093:
URL: https://github.com/apache/pinot/pull/19093#discussion_r3960628759
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java:
##########
@@ -152,6 +156,62 @@ public JsonIndexReader getSparseJsonIndex() {
return _sparseDataSource != null ? _sparseDataSource.getJsonIndex() : null;
}
+ @SuppressWarnings("unchecked")
+ @Nullable
+ @Override
+ public Map<String, Object> getMapValue(int docId) {
Review Comment:
Can a minion task actually reach this? `PinotSegmentRecordReader` iterates `getPhysicalColumnNames()`, which includes the `col$key` / `col$__sparse__` children, and the immutable segment has no DataSource for those, so the reader throws on the first child before it gets to the parent. Worth skipping `isMaterializedChild()` columns in `addColumnReader` and adding a sealed-segment round-trip test, otherwise this path is only covered by mocks.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java:
##########
@@ -63,12 +63,20 @@ 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;
+ // Batched for the same reason as _ignoredKeyDropCount: keep a metered-value call, which rebuilds
+ // the metric name and hits the registry, off the per-row consuming path. Flushed in close().
+ private volatile long _typeCoercionFailureCount;
Review Comment:
Doesn't batching these to `close()` hide the signal for the whole consuming-segment lifetime, and lose it if the server dies first? Could we cache the `PinotMeter` and use the `addMeteredTableValue(..., reusedMeter)` overload instead? That avoids the name rebuild but keeps the metric live.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/ImmutableOpenStructDataSource.java:
##########
@@ -152,6 +156,62 @@ 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(),
Review Comment:
This builds a new `PinotSegmentColumnReader` (and so a new `ForwardIndexReaderContext`) per key per doc. For the raw LZ4 sparse column that's a direct buffer alloc and a full chunk re-decompression on every doc, and both callers are whole-segment loops. Can we hold one reader per key for the scan instead? Same concern I raised on #18643, deferred then because the immutable side had no `getMapValue`.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java:
##########
@@ -119,7 +127,8 @@ public void index(int docId, @Nullable Object value) {
// 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());
Review Comment:
Now that `fromDataTypeSV(...)` sits outside `tryCoerce`'s try/catch, a child spec declared as LIST/STRUCT/UNKNOWN (child types aren't validated anywhere) throws out of `index()` and errors the whole row, where before it was a counted coercion failure. The splitter still has the same expression inside its try. Keep it inside for the fresh-key path, or reject those child types in `OpenStructIndexType.validate`?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -575,14 +572,20 @@ private void writeSparseJsonColumn(List<String> sparseKeys)
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 = "";
Review Comment:
The comment says "" is recorded as the default null value, but `sparseFieldSpec` still carries the STRING default `"null"`, and that's what ends up in `DEFAULT_NULL_VALUE`. Dense keys keep disk and metadata aligned here. Either `setDefaultNullValue("")` on the spec or reword the comment?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitter.java:
##########
@@ -575,14 +572,20 @@ private void writeSparseJsonColumn(List<String> sparseKeys)
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 = "";
+ AbstractColumnStatisticsCollector statsCollector = StatsCollectorUtil.createStatsCollector(sparseFieldSpec, null);
Review Comment:
`StringColumnPreIndexStatsCollector` keeps a set entry for every distinct blob (near one per doc here) and sorts them all at seal, which is O(n log n) string compares on the commit path for cardinality/min/max a raw column never reads. Since the commit path is what the stacked PR is trimming, would a minimal `ColumnStatistics` built from the loop you already have (lengths, doc counts) fed into the same `addColumnMetadataInfo` call work?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentImpl.java:
##########
@@ -144,6 +144,10 @@ public ImmutableSegmentImpl(
for (String parent : openStructParents) {
FieldSpec fieldSpec = schema != null ? schema.getFieldSpecFor(parent) : null;
if (!(fieldSpec instanceof ComplexFieldSpec)) {
+ LOGGER.warn("Skipping OPEN_STRUCT parent column '{}': schema is {} or fieldSpec is {} "
Review Comment:
Nit: worth adding the segment name here since this fires per segment load, and splitting the two cases. "schema is present or fieldSpec is DimensionFieldSpec" reads oddly when only one applies.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructIndex.java:
##########
@@ -290,15 +299,30 @@ public DataType getStoredType() {
@Override
public void close()
throws IOException {
- if (_ignoredKeyDropCount > 0) {
- ServerMetrics serverMetrics = ServerMetrics.get();
- if (serverMetrics != null) {
- serverMetrics.addMeteredTableValue(_tableNameWithType, _openStructColumn,
- ServerMeter.OPEN_STRUCT_IGNORED_KEY_DROPS, _ignoredKeyDropCount);
- }
- }
+ flushMeters();
for (MutableKeyColumn keyCol : _keyColumns.values()) {
keyCol.close();
}
}
+
+ /// Emits the batched ingestion counters. Counters accumulate per row on the consuming path and
+ /// are flushed once here, mirroring what [OpenStructColumnSplitter] does at seal time.
+ private void flushMeters() {
Review Comment:
Two small things: the counters aren't zeroed after emitting, so a second `close()` double-counts, and `flushMeters()` runs before the key-column close loop with no `finally`. Could we zero after emit and flush in a `finally`?
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/creator/impl/openstruct/OpenStructColumnSplitterTest.java:
##########
@@ -219,7 +220,17 @@ public void testSparseJsonColumnWritten()
}
s.seal();
String sparseCol = OpenStructNaming.sparseColumnName("metrics");
- assertTrue(s.getMaterializedColumnMetadata().containsKey(sparseCol));
+ PropertiesConfiguration props = s.getMaterializedColumnMetadata().get(sparseCol);
+ assertNotNull(props);
+
+ // Regression: the sparse column's metadata used to omit several properties ColumnMetadataImpl.
+ // fromPropertiesConfiguration() reads via config.getInt() with no default (e.g. CARDINALITY). This branch's
+ // reader still tolerates a missing bitsPerElement/lengthOfEachEntry/maxNumberOfMultiValues via UNAVAILABLE
+ // defaults, but the sibling early-release branch's stricter reader throws NoSuchElementException on the same
Review Comment:
"The sibling early-release branch" won't mean anything to readers here. Also `CARDINALITY` is the only no-default read and the old block already wrote it, so this test passes on the pre-PR metadata. Could we assert what's actually new instead, e.g. encoding RAW, cardinality 2 (old code wrote 1), longest element length?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/openstruct/MutableOpenStructDataSource.java:
##########
@@ -116,8 +116,9 @@ public boolean isFullyMaterialized() {
@Override
public Map<String, DataSource> getDataSources() {
+ Map<String, MutableKeyColumn> snapshot = _index.getKeyColumns();
Review Comment:
Rebase note: this snapshot doesn't change behaviour (`getDataSource(key)` re-reads the live map, and the published map is never mutated), and it conflicts with #19439 which simplified this loop to `result.put(key, getDataSource(key))`. Probably simplest to take master's version.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org