Vamsi-klu commented on code in PR #19088:
URL: https://github.com/apache/pinot/pull/19088#discussion_r3725701821
##########
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:
Good catch, this is a real gap on the aggregation path. index() calls
updateDictionary() and then getOrCreateDocId(), and the key is built straight
from indexContainer._dictId for every dimension and time column, so the
Integer.MIN_VALUE sentinel does flow into the rollup key. Two failed rows with
different real values would collapse into one key, and the default recovery in
addPhysicalColumn happens only after keying. I will move the fallback into the
updateDictionary() catch block: index getDefaultNullValueForIndexing(fieldSpec)
into the dictionary and set _dictId/_dictIds from that, keeping
Integer.MIN_VALUE only if indexing the default also throws. The MIN_VALUE
recovery in addPhysicalColumn stays as a last resort, and the row is still
metered incomplete via dictHadError.
##########
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:
I checked this and the matcher does match the emitted key. Mockito's
ArgumentMatchers.matches() delegates to org.mockito.internal.matchers.Matches,
which calls pattern.matcher(actual).find(), a substring search rather than a
whole-string match (verified against mockito-core 5.23.0, the version pinned in
the root pom). So matches("DICTIONARY-indexingError$") finds the anchored
suffix inside "testTable_REALTIME-DICTIONARY-indexingError". This assertion is
also unchanged from master (it shows up as context in the diff, and the
pre-existing checks at lines 90 and 104 use the same style), and
IndexingFailureTest passes in CI on this PR. I will keep it as is.
##########
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:
Agreed. The null-value fallback at the top of addPhysicalColumn()
substitutes the field default and meters NULL_VALUE, but the hadError flags are
declared after that block, so the method can still return true and addNewRow()
never meters the row as incomplete. The OPEN_STRUCT branch's early return true
has the same gap. I will track the fallback in a flag initialized before the
null check and fold it into all three return paths (single-value, multi-value,
and OPEN_STRUCT) so any default substitution propagates to rowHadError and
triggers recordIncompleteRow() once for the row. This only changes behavior for
raw/no-dictionary columns, since updateDictionary() already flags the null case
for dictionary columns.
--
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]
