Vamsi-klu commented on code in PR #19088:
URL: https://github.com/apache/pinot/pull/19088#discussion_r3755547427
##########
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:
Fixed in `4a0dfe3`. The `updateDictionary` catch now indexes the column's
default null value into the dictionary before falling back to the sentinel, and
it also puts the default back on the row. Without that second half the forward
index would hold the default while the secondary indexes still held the raw
value, and `setNull` would never fire. The sentinel is kept only if indexing
the default throws as well, and the row is metered incomplete either way. Min
and max are refreshed from the dictionary in the same block, otherwise seal can
still throw the class of error this PR exists to prevent.
One correction on the framing: two failed rows still share the default-value
key after this, by design. The real defect was that `Integer.MIN_VALUE` is not
a real dict id, so a failed row was keyed on something that could never match
the default the row actually stored, and a later row legitimately carrying that
default then landed on a second docId with identical dimension content. The new
test pins exactly that: two failed rows roll into docId 0, then a row carrying
the default joins the same docId. Before the fix it produces two docIds.
##########
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 do not think this one holds. Mockito's `Matches` delegates to
`Matcher.find()`, which is substring semantics rather than full-match, so the
anchored suffix does match inside
`testTable_REALTIME-DICTIONARY-indexingError`. The assertion is also unchanged
from master and green in CI, so I have left 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:
Fixed in `4a0dfe3`. `addPhysicalColumn` tracks a `defaultSubstituted` flag
now, folded into the SV, MV and OPEN_STRUCT return paths, so `addNewRow` meters
the row incomplete exactly once per row no matter how many columns fell back.
There is a test with one row that only substitutes and one that substitutes and
also fails its JSON index, asserting exactly two meters for two rows.
--
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]
