Copilot commented on code in PR #19088:
URL: https://github.com/apache/pinot/pull/19088#discussion_r3669414589
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -820,182 +831,419 @@ private void validateLengthOfMVColumns(GenericRow row)
}
}
- private void updateDictionary(GenericRow row) {
+ /// Runs dictionary + forward/secondary indexing for a new docId and meters
an incomplete row when either step had
+ /// to fall back to defaults (issue #16316).
+ private void indexPhysicalRow(int docId, GenericRow row) {
+ boolean dictHadError = updateDictionary(row);
+ boolean rowHadError = addNewRow(docId, row);
+ if (dictHadError || rowHadError) {
+ recordIncompleteRow();
+ }
+ }
+
+ /// @return {@code true} if any column required a default/fallback while
updating dictionaries
+ private boolean updateDictionary(GenericRow row) {
+ boolean hadError = false;
for (Map.Entry<String, IndexContainer> entry :
_indexContainerMap.entrySet()) {
IndexContainer indexContainer = entry.getValue();
MutableDictionary dictionary = indexContainer._dictionary;
if (dictionary == null) {
continue;
}
-
- Object value = row.getValue(entry.getKey());
- if (value == null) {
- recordIndexingError("DICTIONARY");
- } else {
+ String column = entry.getKey();
+ Object value = row.getValue(column);
+ try {
+ if (value == null) {
+ // Prefer default-null dict entry so addNewRow can still complete
the forward index for this docId
+ // (fail-soft; issue #16316). Meter and fall back to field-spec
default.
+ recordIndexingError("DICTIONARY");
+ hadError = true;
+ value = getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ row.putDefaultNullValue(column, value);
+ }
if (indexContainer._fieldSpec.isSingleValueField()) {
indexContainer._dictId = dictionary.index(value);
} else {
indexContainer._dictIds = dictionary.index((Object[]) value);
}
-
// Update min/max value from dictionary
indexContainer._minValue = dictionary.getMinVal();
indexContainer._maxValue = dictionary.getMaxVal();
+ } catch (Exception e) {
+ // Do not abort the row mid-dictionary: remaining columns still get a
chance, and addNewRow will fill
+ // defaults for this column if dict ids are unset (Integer.MIN_VALUE /
null).
+ hadError = true;
+ recordIndexingError("DICTIONARY", e);
+ indexContainer._dictId = Integer.MIN_VALUE;
+ indexContainer._dictIds = null;
}
Review Comment:
In updateDictionary(), the exception path sets _dictId=Integer.MIN_VALUE.
When metrics aggregation is enabled, getOrCreateDocId() uses these dictIds as
the rollup key, so a dict failure can silently roll up unrelated rows together
(or otherwise corrupt aggregation keying).
Instead of leaving a sentinel, try to fall back to the field default-null
value and index that into the dictionary so key columns remain valid; only fall
back to Integer.MIN_VALUE if even the default cannot be indexed.
This issue also appears in the following locations of the same file:
- line 1028
- line 1119
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -820,182 +831,419 @@ private void validateLengthOfMVColumns(GenericRow row)
}
}
- private void updateDictionary(GenericRow row) {
+ /// Runs dictionary + forward/secondary indexing for a new docId and meters
an incomplete row when either step had
+ /// to fall back to defaults (issue #16316).
+ private void indexPhysicalRow(int docId, GenericRow row) {
+ boolean dictHadError = updateDictionary(row);
+ boolean rowHadError = addNewRow(docId, row);
+ if (dictHadError || rowHadError) {
+ recordIncompleteRow();
+ }
+ }
+
+ /// @return {@code true} if any column required a default/fallback while
updating dictionaries
+ private boolean updateDictionary(GenericRow row) {
+ boolean hadError = false;
for (Map.Entry<String, IndexContainer> entry :
_indexContainerMap.entrySet()) {
IndexContainer indexContainer = entry.getValue();
MutableDictionary dictionary = indexContainer._dictionary;
if (dictionary == null) {
continue;
}
-
- Object value = row.getValue(entry.getKey());
- if (value == null) {
- recordIndexingError("DICTIONARY");
- } else {
+ String column = entry.getKey();
+ Object value = row.getValue(column);
+ try {
+ if (value == null) {
+ // Prefer default-null dict entry so addNewRow can still complete
the forward index for this docId
+ // (fail-soft; issue #16316). Meter and fall back to field-spec
default.
+ recordIndexingError("DICTIONARY");
+ hadError = true;
+ value = getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ row.putDefaultNullValue(column, value);
+ }
if (indexContainer._fieldSpec.isSingleValueField()) {
indexContainer._dictId = dictionary.index(value);
} else {
indexContainer._dictIds = dictionary.index((Object[]) value);
}
-
// Update min/max value from dictionary
indexContainer._minValue = dictionary.getMinVal();
indexContainer._maxValue = dictionary.getMaxVal();
+ } catch (Exception e) {
+ // Do not abort the row mid-dictionary: remaining columns still get a
chance, and addNewRow will fill
+ // defaults for this column if dict ids are unset (Integer.MIN_VALUE /
null).
+ hadError = true;
+ recordIndexingError("DICTIONARY", e);
+ indexContainer._dictId = Integer.MIN_VALUE;
+ indexContainer._dictIds = null;
}
- updateIndexCapacityThresholdBreached(dictionary, entry.getKey());
+ updateIndexCapacityThresholdBreached(dictionary, column);
}
+ return hadError;
}
- private void addNewRow(int docId, GenericRow row) {
+ /// Indexes a new physical row. Fail-soft (issue #16316): every physical
column must end up with a forward-index
+ /// (or OPEN_STRUCT) entry for [docId] so seal/query lengths stay aligned
with [_numDocsIndexed]. On forward-index
+ /// failure the column is completed with the field default/null instead of
being left blank. Secondary index
+ /// failures are still metered and swallowed. Aggregation path failures also
fall back to a default initial value.
+ ///
+ /// @return {@code true} if any column required a default/fallback while
indexing
+ private boolean addNewRow(int docId, GenericRow row) {
+ boolean rowHadError = false;
for (Map.Entry<String, IndexContainer> entry :
_indexContainerMap.entrySet()) {
String column = entry.getKey();
IndexContainer indexContainer = entry.getValue();
-
- // Handle ingestion aggregation
- ValueAggregator valueAggregator = indexContainer._valueAggregator;
- if (valueAggregator != null) {
- String sourceColumn = indexContainer._sourceColumn;
- // NOTE: value can be null if the column is not specified in the
schema.
- Object value = row.getValue(sourceColumn);
- // Handle COUNT(*)
- if (value == null &&
sourceColumn.equals(AggregationFunctionColumnPair.STAR)) {
- assert valueAggregator.getAggregationType() ==
AggregationFunctionType.COUNT;
- value = 1;
+ try {
+ if (indexContainer._valueAggregator != null) {
+ if (!addAggregatedColumn(docId, row, column, indexContainer)) {
+ rowHadError = true;
+ }
+ } else if (!addPhysicalColumn(docId, row, column, indexContainer)) {
+ rowHadError = true;
}
-
- // Update numValues info
- indexContainer._valuesInfo.updateSVNumValues();
-
- MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
- FieldSpec fieldSpec = indexContainer._fieldSpec;
-
- DataType dataType = fieldSpec.getDataType();
- value = valueAggregator.getInitialAggregatedValue(value);
- // BIG_DECIMAL is actually stored as byte[] and hence can be supported
here.
- switch (dataType.getStoredType()) {
- case INT:
- forwardIndex.add(((Number) value).intValue(), -1, docId);
- break;
- case LONG:
- forwardIndex.add(((Number) value).longValue(), -1, docId);
- break;
- case FLOAT:
- forwardIndex.add(((Number) value).floatValue(), -1, docId);
- break;
- case DOUBLE:
- forwardIndex.add(((Number) value).doubleValue(), -1, docId);
- break;
- case BIG_DECIMAL:
- case BYTES:
- forwardIndex.add(valueAggregator.serializeAggregatedValue(value),
-1, docId);
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported data type: " + dataType + " for aggregation: " +
column);
+ } catch (Exception e) {
+ // Last-resort complete-the-row so a single bad column cannot leave a
half-written docId.
+ rowHadError = true;
+ recordIndexingError("ROW", e);
+ try {
+ indexDefaultNullColumn(docId, indexContainer);
+ } catch (Exception fallbackError) {
+ _logger.error("Failed to index default null for column: {} at docId:
{}", column, docId, fallbackError);
}
- continue;
}
+ }
- // Update the null value vector even if a null value is somehow produced
- if (indexContainer._nullValueVector != null && row.isNullValue(column)) {
- indexContainer._nullValueVector.setNull(docId);
+ if (_multiColumnValues != null) {
+ try {
+ _multiColumnTextIndex.add(_multiColumnValues);
+ } catch (Exception e) {
+ rowHadError = true;
+ recordIndexingError("MULTI_COLUMN_TEXT", e);
+ } finally {
+ Collections.fill(_multiColumnValues, null);
}
+ }
+ return rowHadError;
+ }
- Object value = row.getValue(column);
- if (value == null) {
- // the value should not be null unless something is broken upstream
but this will lead to inappropriate reuse
- // of the dictionary id if this somehow happens. An NPE here can
corrupt indexes leading to incorrect query
- // results, hence the extra care. A metric will already have been
emitted when trying to update the dictionary.
- continue;
- }
+ /// Returns {@code true} when the aggregated column was written without
error.
+ private boolean addAggregatedColumn(int docId, GenericRow row, String
column, IndexContainer indexContainer) {
+ ValueAggregator valueAggregator = indexContainer._valueAggregator;
+ String sourceColumn = indexContainer._sourceColumn;
+ // NOTE: value can be null if the column is not specified in the schema.
+ Object value = row.getValue(sourceColumn);
+ // Handle COUNT(*)
+ if (value == null &&
sourceColumn.equals(AggregationFunctionColumnPair.STAR)) {
+ assert valueAggregator.getAggregationType() ==
AggregationFunctionType.COUNT;
+ value = 1;
+ }
- FieldSpec fieldSpec = indexContainer._fieldSpec;
- DataType dataType = fieldSpec.getDataType();
+ MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ try {
+ value = valueAggregator.getInitialAggregatedValue(value);
+ // BIG_DECIMAL is actually stored as byte[] and hence can be supported
here.
+ switch (dataType.getStoredType()) {
+ case INT:
+ forwardIndex.add(((Number) value).intValue(), -1, docId);
+ break;
+ case LONG:
+ forwardIndex.add(((Number) value).longValue(), -1, docId);
+ break;
+ case FLOAT:
+ forwardIndex.add(((Number) value).floatValue(), -1, docId);
+ break;
+ case DOUBLE:
+ forwardIndex.add(((Number) value).doubleValue(), -1, docId);
+ break;
+ case BIG_DECIMAL:
+ case BYTES:
+ forwardIndex.add(valueAggregator.serializeAggregatedValue(value),
-1, docId);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unsupported data type: " + dataType + " for aggregation: " +
column);
+ }
+ indexContainer._valuesInfo.updateSVNumValues();
+ return true;
+ } catch (Exception e) {
+ recordIndexingError(StandardIndexes.forward(), e);
+ indexDefaultAggregatedValue(docId, indexContainer);
+ return false;
+ }
+ }
- if (fieldSpec.isSingleValueField()) {
- // Update numValues info
+ /// Returns {@code true} when the physical column was written without error.
+ private boolean addPhysicalColumn(int docId, GenericRow row, String column,
IndexContainer indexContainer) {
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ boolean isNull = row.isNullValue(column);
+ Object value = row.getValue(column);
+ if (value == null) {
+ // Should not happen after NullValueTransformer, but complete the row
with defaults rather than leaving a hole.
+ recordIndexingError("NULL_VALUE");
+ value = getDefaultNullValueForIndexing(fieldSpec);
+ isNull = true;
+ }
Review Comment:
When row.getValue(column) is null, addPhysicalColumn() falls back to the
field default and logs an indexing error, but it can still return true (because
this fallback isn't included in the hadError/return value). That means callers
won't meter the row as incomplete even though a default was substituted.
If the intent is that any default substitution counts as an incomplete row
(per PR description and method contract), the method should propagate this
fallback as a failure signal (or otherwise ensure recordIncompleteRow() is
triggered once per row).
##########
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java:
##########
@@ -133,8 +133,46 @@ public void testIndexingFailures()
assertEquals(_mutableSegment.getDataSource(JSON_COL).getJsonIndex().getMatchingDocIds("valid
= 'json'"),
ImmutableRoaringBitmap.bitmapOf(0, 2, 3));
assertTrue(_mutableSegment.getDataSource(STRING_COL).getNullValueVector().isNull(3));
- // null string value skipped
+ // Fail-soft (#16316): null string is completed with the field default so
forward lengths stay aligned.
+ GenericRow nullResult = _mutableSegment.getRecord(3, new GenericRow());
+ assertEquals(nullResult.getValue(STRING_COL),
FieldSpec.DEFAULT_DIMENSION_NULL_VALUE_OF_STRING);
verify(_serverMetrics,
times(1)).addMeteredTableValue(matches("DICTIONARY-indexingError$"),
eq(ServerMeter.INDEXING_FAILURES), eq(1L));
Review Comment:
This matcher won't match the actual metric key being emitted.
MutableSegmentImpl.recordIndexingError("DICTIONARY") prefixes the key with the
realtime table name (e.g. "testTable_REALTIME-DICTIONARY-indexingError"), so
matching only "DICTIONARY-indexingError$" will fail.
Update the expectation to include the table prefix (or use a regex with
".*-" prefix) so the test asserts the same key that production code emits.
--
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]
