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: [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]
