Jackie-Jiang commented on code in PR #19088:
URL: https://github.com/apache/pinot/pull/19088#discussion_r4449000588
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/PartitionDedupMetadataManager.java:
##########
@@ -55,6 +55,14 @@ default void replaceSegment(IndexSegment oldSegment,
IndexSegment newSegment) {
/// Remove the expired primary keys from the metadata when TTL is enabled.
void removeExpiredPrimaryKeys();
+ /// Returns true if the primary key is already present and within TTL. Does
not insert the key.
+ ///
+ /// Call this before writing a new consuming row so a failed write cannot
leave a ghost key that drops a later retry.
+ /// Default is absent so existing test doubles keep the previous
always-insert behavior until they override it.
+ default boolean isRecordPresent(DedupRecordInfo dedupRecordInfo) {
+ return false;
Review Comment:
[P1] Preserve deduplication for existing metadata-manager plugins
A custom metadata manager implementing the existing interface inherits false
here. MutableSegmentImpl now publishes the row before calling
checkRecordPresentOrUpdate(), and ignores that method's duplicate result. Thus
a previously working plugin can return true for an existing key and still
retain every duplicate after this change. These are supported production
extensions via DedupConfig.metadataManagerClass, not just test doubles. Please
preserve the old contract for implementations without the new capability, or
provide a compatible migration. Add a regression using a legacy implementation
that only overrides checkRecordPresentOrUpdate(), and ingest the same key twice.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -674,82 +697,105 @@ public boolean index(GenericRow row, @Nullable
StreamMessageMetadata metadata)
boolean canTakeMore;
// NOTE: out-of-order records can not be dropped or marked when
consistent upsert view is enabled.
- // Since Indexing the record and updation of _numDocsIndexed counter
happens before updating the upsert
- // metadata, we wouldn't be able to actually drop or mark those records
as dropped. This order is important for
- // consistent upsert view, otherwise the latest doc can be missed by
query due to 'docId < _numDocs' check
- // in query filter operators. Here the record becomes queryable before
validDocIds bitmaps are updated.
- if (_upsertConsistencyMode != UpsertConfig.ConsistencyMode.NONE) {
- updateDictionary(updatedRow);
- addNewRow(numDocsIndexed, updatedRow);
- numDocsIndexed++;
- canTakeMore = numDocsIndexed < _capacity;
- _numDocsIndexed = numDocsIndexed;
- // Index the record and update _numDocsIndexed counter before updating
the upsert metadata so that the record
- // becomes queryable before validDocIds bitmaps are updated. This
order is important for consistent upsert view,
- // otherwise the latest doc can be missed by query due to 'docId <
_numDocs' check in query filter operators.
- // NOTE: out-of-order records can not be dropped or marked when
consistent upsert view is enabled.
- _partitionUpsertMetadataManager.addRecord(this, recordInfo);
- } else {
- // if record doesn't need to be dropped, then persist in segment and
update metadata hashmap
- // we are doing metadata update first followed by segment data update
here, there can be a scenario where
- // segment indexing or addNewRow call errors out in those scenario,
there can be metadata inconsistency where
- // a key is pointing to some other key's docID
- // TODO fix this metadata mismatch scenario
- boolean isOutOfOrderRecord =
!_partitionUpsertMetadataManager.addRecord(this, recordInfo);
+ // Peek first so drop/mark can skip the write without mutating
validDocIds. Then finish the physical row
+ // before addRecord so metadata never points at a hole (issue #16316).
+ if (!_partitionUpsertMetadataManager.isAcceptingRecords()) {
+ updateIndexedAndIngestionTime(metadata);
+ return canAddMore();
+ }
+ if (_upsertConsistencyMode == UpsertConfig.ConsistencyMode.NONE) {
+ boolean isOutOfOrderRecord =
_partitionUpsertMetadataManager.isOutOfOrderRecord(recordInfo);
if (_upsertOutOfOrderRecordColumn != null) {
updatedRow.putValue(_upsertOutOfOrderRecordColumn,
BooleanUtils.toInt(isOutOfOrderRecord));
}
- if (!isOutOfOrderRecord || !_upsertDropOutOfOrderRecord) {
- updateDictionary(updatedRow);
- addNewRow(numDocsIndexed, updatedRow);
- // Update number of documents indexed before handling the upsert
metadata so that the record becomes queryable
- // once validated
- numDocsIndexed++;
+ if (isOutOfOrderRecord && _upsertDropOutOfOrderRecord) {
+ // Not addRecord(): the check above ran without the key's lock, and
if the key's segment was removed since
+ // then, addRecord() would treat it as new and mark the unwritten
doc `numDocsIndexed` valid.
+ _partitionUpsertMetadataManager.dropOutOfOrderRecord(this,
recordInfo);
+ updateIndexedAndIngestionTime(metadata);
+ return canAddMore();
}
- canTakeMore = numDocsIndexed < _capacity;
- _numDocsIndexed = numDocsIndexed;
}
+ indexPhysicalRow(numDocsIndexed, updatedRow);
+ numDocsIndexed++;
+ canTakeMore = numDocsIndexed < _capacity;
+ _numDocsIndexed = numDocsIndexed;
+ // Index the record and update _numDocsIndexed counter before updating
the upsert metadata so that the record
+ // becomes queryable before validDocIds bitmaps are updated. This order
is important for consistent upsert view,
+ // otherwise the latest doc can be missed by query due to 'docId <
_numDocs' check in query filter operators.
+ _partitionUpsertMetadataManager.addRecord(this, recordInfo);
Review Comment:
[P1] Keep repaired upsert fields consistent with the metadata being claimed
recordInfo was captured before indexPhysicalRow(), but that method can
replace primary-key, comparison, or delete fields with defaults. This call then
claims metadata for the original values while publishing a different physical
row. For example, a recoverable dictionary failure on comparison value 20 can
store its default while metadata still claims version 20, causing a later valid
version 10 to be rejected; rebuilding metadata from the segment yields
different ordering. A repaired primary key similarly disagrees with the map
key, and a repaired delete flag can disagree with the queryable bitmap. Please
handle metadata-defining field failures explicitly: avoid validating a
substituted row under the original metadata, or finalize the physical values
before constructing metadata with a defined recovery policy. Cover PK,
comparison, and delete-field fallback.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -852,192 +898,527 @@ private void validateNumMultiValues(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) {
+ RowWriteState state = _rowWriteState;
+ boolean dictHadError = updateDictionary(row, state);
+ boolean rowHadError = addNewRow(docId, row, state);
+ if (dictHadError || rowHadError) {
+ recordIncompleteRow();
+ }
+ if (state._repairFailed) {
+ markUnrecoverable(state);
+ throwPendingRowIndexingExceptionIfStrict();
+ throw new IllegalStateException(
+ "Mutable segment " + _segmentName + " is terminal after an
unrecoverable indexing failure");
+ }
+ }
+
+ /// @return {@code true} if any column required a default/fallback while
updating dictionaries.
+ /// When `continueOnError` is false, dictionary failures are stashed and
rethrown after the row is published.
+ private boolean updateDictionary(GenericRow row, RowWriteState state) {
+ 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) {
+ recordIndexingError("DICTIONARY");
+ hadError = true;
+ state.markIncomplete();
+ 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) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ hadError = true;
+ state.noteError(e);
+ indexContainer._dictId = Integer.MIN_VALUE;
+ indexContainer._dictIds = null;
+ try {
+ // Index the field default instead of leaving the sentinel: with
metrics aggregation the rollup key is built
+ // from the dict ids (see getOrCreateDocId), and Integer.MIN_VALUE
is not a real dict id.
+ Object defaultValue =
getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ if (indexContainer._fieldSpec.isSingleValueField()) {
+ indexContainer._dictId = dictionary.index(defaultValue);
+ } else {
+ indexContainer._dictIds = dictionary.index((Object[])
defaultValue);
+ }
+ row.putDefaultNullValue(column, defaultValue);
+ indexContainer._minValue = dictionary.getMinVal();
+ indexContainer._maxValue = dictionary.getMaxVal();
+ } catch (Exception fallbackError) {
+ _logger.error("Failed to index default null value into dictionary
for column: {}", column, fallbackError);
+ state.noteRepairFailure(fallbackError);
+ }
}
- updateIndexCapacityThresholdBreached(dictionary, entry.getKey());
+ updateIndexCapacityThresholdBreached(dictionary, column);
}
+ return hadError;
}
- private void addNewRow(int docId, GenericRow row) {
+ /// Indexes a new physical row. Always completes the row so seal/query
lengths stay aligned with [_numDocsIndexed]
+ /// (issue #16316). Forward indexes are written first from a finalized
value; secondary indexes then consume that
+ /// same value. On unrecoverable repair the segment is marked terminal.
+ ///
+ /// @return {@code true} if any column required a default/fallback while
indexing
+ private boolean addNewRow(int docId, GenericRow row, RowWriteState state) {
+ 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, state))
{
+ rowHadError = true;
+ }
+ } else if (!addPhysicalColumn(docId, row, column, indexContainer,
state)) {
+ 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) {
+ recordOrDeferIndexingError("ROW", e);
+ rowHadError = true;
+ state.noteError(e);
+ try {
+ indexDefaultNullColumn(docId, column, indexContainer, state);
+ } catch (Exception fallbackError) {
+ _logger.error("Failed to index default null for column: {} at docId:
{}", column, docId, fallbackError);
+ state.noteRepairFailure(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 {
+ // Multi-column text is a row-level secondary index. It is written
only after every column has a finalized
+ // value in `_multiColumnValues`. There is no per-doc rollback; a
failure leaves this doc absent from the
+ // text index while forward indexes remain complete. Counted once as a
secondary-index error.
+ _multiColumnTextIndex.add(_multiColumnValues);
+ } catch (Exception e) {
+ recordOrDeferIndexingError("MULTI_COLUMN_TEXT", e);
+ rowHadError = true;
+ state.noteError(e);
+ } finally {
+ Collections.fill(_multiColumnValues, null);
}
+ }
+ if (state._repairFailed) {
+ markUnrecoverable(state);
+ }
+ 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,
+ RowWriteState state) {
+ ValueAggregator valueAggregator = indexContainer._valueAggregator;
+ String sourceColumn = indexContainer._sourceColumn;
+ Object value = row.getValue(sourceColumn);
+ if (value == null &&
sourceColumn.equals(AggregationFunctionColumnPair.STAR)) {
+ assert valueAggregator.getAggregationType() ==
AggregationFunctionType.COUNT;
+ value = 1;
+ }
+
+ MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ try {
+ value = valueAggregator.getInitialAggregatedValue(value);
+ addAggregatedForwardValue(forwardIndex, valueAggregator, dataType,
value, docId, column);
+ indexContainer._valuesInfo.updateSVNumValues();
+ state._columnsWithCanonicalForward.add(column);
+ return true;
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.forward(), e);
+ state.noteError(e);
+ try {
+ indexDefaultAggregatedValue(docId, column, indexContainer, state);
+ } catch (Exception fallbackError) {
+ state.noteRepairFailure(fallbackError);
}
+ return false;
+ }
+ }
- FieldSpec fieldSpec = indexContainer._fieldSpec;
- DataType dataType = fieldSpec.getDataType();
+ private void addAggregatedForwardValue(MutableIndex forwardIndex,
ValueAggregator valueAggregator, DataType dataType,
+ Object value, int docId, String column) {
+ maybeIntercept(column, StandardIndexes.forward(), value, docId);
+ 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);
+ }
+ }
- if (fieldSpec.isSingleValueField()) {
- // Update numValues info
- indexContainer._valuesInfo.updateSVNumValues();
-
- // Route OPEN_STRUCT values to the dedicated mutable index.
OPEN_STRUCT has no forward
- // index / dictionary / min-max, so the standard per-IndexType loop
and the comparable
- // tracking below would be no-ops at best and crash at worst (Map is
not Comparable).
- if (dataType == DataType.OPEN_STRUCT) {
- MutableIndex openStructIndex =
indexContainer._mutableIndexes.get(StandardIndexes.openStruct());
- if (openStructIndex != null) {
- openStructIndex.add(value, -1, docId);
- }
- continue;
+ /// Returns {@code true} when the physical column was written from the row
value without error and without falling
+ /// back to the field default.
+ private boolean addPhysicalColumn(int docId, GenericRow row, String column,
IndexContainer indexContainer,
+ RowWriteState state) {
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ boolean isNull = row.isNullValue(column);
+ Object value = row.getValue(column);
+ boolean defaultSubstituted = false;
+ if (value == null) {
+ recordIndexingError("NULL_VALUE");
+ value = getDefaultNullValueForIndexing(fieldSpec);
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ }
+
+ if (fieldSpec.isSingleValueField()) {
+ if (dataType == DataType.OPEN_STRUCT) {
+ return addOpenStructColumn(docId, column, indexContainer, value,
defaultSubstituted, state);
+ }
+ int dictId = indexContainer._dictId;
+ if (indexContainer._dictionary != null && dictId == Integer.MIN_VALUE) {
+ try {
+ Object defaultValue = getDefaultNullValueForIndexing(fieldSpec);
+ dictId = indexContainer._dictionary.index(defaultValue);
+ indexContainer._dictId = dictId;
+ value = defaultValue;
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ } catch (Exception e) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ state.noteError(e);
+ state.noteRepairFailure(e);
+ return false;
}
+ }
+ Object parsedJsonValue = defaultSubstituted ? null :
row.getParsedJsonValue(column);
+ boolean hadError = writeSingleValueIndexes(docId, column,
indexContainer, fieldSpec, dataType, value, dictId,
+ isNull, parsedJsonValue, state);
+ return !hadError && !defaultSubstituted;
+ }
- // Update indexes
- int dictId = indexContainer._dictId;
- for (Map.Entry<IndexType, MutableIndex> indexEntry :
indexContainer._mutableIndexes.entrySet()) {
- try {
- MutableIndex mutableIndex = indexEntry.getValue();
- if (mutableIndex instanceof MutableJsonIndexImpl) {
- // Feed the JSON index the parsed value cached by the
DataTypeTransformer (if any), so it flattens the
- // document directly instead of re-parsing the string this
column was serialized into for the forward
- // index; the index falls back to that string for documents it
cannot flatten directly. This is
- // intentionally scoped to Pinot's built-in JSON index
implementation.
- Object parsedValue = row.getParsedJsonValue(column);
- if (parsedValue != null && value instanceof String) {
- ((MutableJsonIndexImpl) mutableIndex).addParsed(parsedValue,
(String) value);
- updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
- continue;
- }
- }
- mutableIndex.add(value, dictId, docId);
- updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
- } catch (Exception e) {
- recordIndexingError(indexEntry.getKey(), e);
- }
+ Object[] values = value instanceof Object[] ? (Object[]) value : new
Object[]{value};
+ int[] dictIds = indexContainer._dictIds;
+ if (indexContainer._dictionary != null && dictIds == null) {
+ try {
+ Object[] defaultValues = (Object[])
getDefaultNullValueForIndexing(fieldSpec);
+ dictIds = indexContainer._dictionary.index(defaultValues);
+ indexContainer._dictIds = dictIds;
+ values = defaultValues;
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ } catch (Exception e) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ state.noteError(e);
+ state.noteRepairFailure(e);
+ return false;
+ }
+ }
+ boolean hadError = writeMultiValueIndexes(docId, column, indexContainer,
dataType, values, dictIds, isNull, state);
+ return !hadError && !defaultSubstituted;
+ }
+
+ private boolean addOpenStructColumn(int docId, String column, IndexContainer
indexContainer, Object value,
+ boolean defaultSubstituted, RowWriteState state) {
+ MutableIndex openStructIndex =
indexContainer._mutableIndexes.get(StandardIndexes.openStruct());
+ if (openStructIndex != null) {
+ try {
+ maybeIntercept(column, StandardIndexes.openStruct(), value, docId);
+ openStructIndex.add(value, -1, docId);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.openStruct(), e);
+ state.noteError(e);
+ try {
+ Object defaultValue =
getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ maybeIntercept(column, StandardIndexes.openStruct(), defaultValue,
docId);
+ openStructIndex.add(defaultValue, -1, docId);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception fallbackError) {
+ state.noteRepairFailure(fallbackError);
+ return false;
}
+ defaultSubstituted = true;
+ }
+ } else {
+ state._columnsWithCanonicalForward.add(column);
+ }
+ indexContainer._valuesInfo.updateSVNumValues();
+ return !defaultSubstituted;
+ }
- if (dictId < 0) {
- // Update min/max value from raw value
- // NOTE: Skip updating min/max value for aggregated metrics because
the value will change over time.
- if (!isAggregateMetricsEnabled() || fieldSpec.getFieldType() !=
FieldSpec.FieldType.METRIC) {
- Comparable comparable = toComparableValue(value, dataType, column);
- if (indexContainer._minValue == null) {
- indexContainer._minValue = comparable;
- indexContainer._maxValue = comparable;
- } else {
- if (comparable.compareTo(indexContainer._minValue) < 0) {
- indexContainer._minValue = comparable;
- }
- if (comparable.compareTo(indexContainer._maxValue) > 0) {
- indexContainer._maxValue = comparable;
- }
- }
+ /// Writes the canonical forward index first, then every secondary index
from that same finalized value.
+ private boolean writeSingleValueIndexes(int docId, String column,
IndexContainer indexContainer, FieldSpec fieldSpec,
+ DataType dataType, Object value, int dictId, boolean isNull, @Nullable
Object parsedJsonValue,
+ RowWriteState state) {
+ boolean hadError = false;
+ MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
+ if (forwardIndex != null &&
!state._columnsWithCanonicalForward.contains(column)) {
+ try {
+ maybeIntercept(column, StandardIndexes.forward(), value, docId);
+ forwardIndex.add(value, dictId, docId);
+ updateIndexCapacityThresholdBreached(forwardIndex,
StandardIndexes.forward(), column);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.forward(), e);
+ state.noteError(e);
+ hadError = true;
+ try {
+ Object defaultValue = getDefaultNullValueForIndexing(fieldSpec);
+ int defaultDictId = dictId;
+ if (indexContainer._dictionary != null) {
+ defaultDictId = indexContainer._dictionary.index(defaultValue);
+ indexContainer._dictId = defaultDictId;
}
+ value = defaultValue;
+ dictId = defaultDictId;
Review Comment:
[P1] Keep the registered aggregation key consistent with repaired dimensions
getOrCreateDocId() has already registered the original dimension dictionary
IDs before this forward-index fallback changes the dimension and _dictId to the
default. With aggregation enabled, a forward failure for dim=42, metric=1 can
publish dim=DEFAULT while _recordIdMap still maps the key for 42 to that
document. A later successful dim=42, metric=2 then aggregates into the
default-valued row, permanently assigning valid input to the wrong group.
Please ensure a repaired dimension cannot disagree with the registered key, or
treat key-column repair as unrecoverable. The existing dictionary-failure
aggregation test fails before key creation and does not cover this
forward-failure case.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -1129,77 +1582,159 @@ private void aggregateMetrics(GenericRow row, int
docId) {
}
MutableForwardIndex forwardIndex =
(MutableForwardIndex)
indexContainer._mutableIndexes.get(StandardIndexes.forward());
- DataType dataType = metricFieldSpec.getDataType();
-
- switch (valueAggregator.getAggregatedValueType()) {
- case DOUBLE:
- double oldDoubleValue;
- double newDoubleValue;
- switch (dataType) {
- case INT:
- oldDoubleValue = forwardIndex.getInt(docId);
- newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, value);
- forwardIndex.setInt(docId, (int) newDoubleValue);
- break;
- case LONG:
- oldDoubleValue = forwardIndex.getLong(docId);
- newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, value);
- forwardIndex.setLong(docId, (long) newDoubleValue);
- break;
- case FLOAT:
- oldDoubleValue = forwardIndex.getFloat(docId);
- newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, value);
- forwardIndex.setFloat(docId, (float) newDoubleValue);
- break;
- case DOUBLE:
- oldDoubleValue = forwardIndex.getDouble(docId);
- newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, value);
- forwardIndex.setDouble(docId, newDoubleValue);
- break;
- default:
- throw new
UnsupportedOperationException(String.format("Aggregation type %s of %s not
supported for %s",
- valueAggregator.getAggregatedValueType(),
valueAggregator.getAggregationType(), dataType));
- }
- break;
- case LONG:
- long oldLongValue;
- long newLongValue;
- switch (dataType) {
- case INT:
- oldLongValue = forwardIndex.getInt(docId);
- newLongValue = (long)
valueAggregator.applyRawValue(oldLongValue, value);
- forwardIndex.setInt(docId, (int) newLongValue);
- break;
- case LONG:
- oldLongValue = forwardIndex.getLong(docId);
- newLongValue = (long)
valueAggregator.applyRawValue(oldLongValue, value);
- forwardIndex.setLong(docId, newLongValue);
- break;
- case FLOAT:
- oldLongValue = (long) forwardIndex.getFloat(docId);
- newLongValue = (long)
valueAggregator.applyRawValue(oldLongValue, value);
- forwardIndex.setFloat(docId, (float) newLongValue);
- break;
- case DOUBLE:
- oldLongValue = (long) forwardIndex.getDouble(docId);
- newLongValue = (long)
valueAggregator.applyRawValue(oldLongValue, value);
- forwardIndex.setDouble(docId, (double) newLongValue);
- break;
- default:
- throw new
UnsupportedOperationException(String.format("Aggregation type %s of %s not
supported for %s",
- valueAggregator.getAggregatedValueType(),
valueAggregator.getAggregationType(), dataType));
+ prepared.add(
+ prepareMetricWrite(column, valueAggregator, forwardIndex,
metricFieldSpec.getDataType(), value, docId));
+ }
+
+ List<PreparedMetricWrite> committed = new ArrayList<>();
+ try {
+ for (PreparedMetricWrite write : prepared) {
+ maybeIntercept(write._column, StandardIndexes.forward(),
write.interceptValue(), docId);
+ commitMetricWrite(write, docId);
+ committed.add(write);
+ }
+ } catch (Exception e) {
+ Exception restoreError = null;
+ for (int i = committed.size() - 1; i >= 0; i--) {
+ try {
+ maybeIntercept(committed.get(i)._column, StandardIndexes.forward(),
committed.get(i).restoreInterceptValue(),
+ docId);
+ restoreMetricWrite(committed.get(i), docId);
+ } catch (Exception restoreException) {
+ if (restoreError == null) {
+ restoreError = restoreException;
+ } else {
+ restoreError.addSuppressed(restoreException);
}
- break;
- case BYTES:
- Object oldValue =
valueAggregator.deserializeAggregatedValue(forwardIndex.getBytes(docId));
- Object newValue = valueAggregator.applyRawValue(oldValue, value);
- forwardIndex.setBytes(docId,
valueAggregator.serializeAggregatedValue(newValue));
- break;
- default:
- throw new UnsupportedOperationException(
- String.format("Aggregation type %s of %s not supported for %s",
valueAggregator.getAggregatedValueType(),
- valueAggregator.getAggregationType(), dataType));
+ }
+ }
+ if (restoreError != null) {
+ e.addSuppressed(restoreError);
+ _unrecoverableIndexingFailure = true;
+ }
+ throw wrapIndexingException("AGGREGATE_METRICS", e);
+ }
+ }
+
+ private PreparedMetricWrite prepareMetricWrite(String column,
ValueAggregator valueAggregator,
+ MutableForwardIndex forwardIndex, DataType dataType, Object rawValue,
int docId) {
+ switch (valueAggregator.getAggregatedValueType()) {
+ case DOUBLE: {
+ double oldDoubleValue;
+ double newDoubleValue;
+ MetricWriteKind kind;
+ switch (dataType) {
+ case INT:
+ oldDoubleValue = forwardIndex.getInt(docId);
+ newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, rawValue);
+ kind = MetricWriteKind.INT;
+ break;
+ case LONG:
+ oldDoubleValue = forwardIndex.getLong(docId);
+ newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, rawValue);
+ kind = MetricWriteKind.LONG;
+ break;
+ case FLOAT:
+ oldDoubleValue = forwardIndex.getFloat(docId);
+ newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, rawValue);
+ kind = MetricWriteKind.FLOAT;
+ break;
+ case DOUBLE:
+ oldDoubleValue = forwardIndex.getDouble(docId);
+ newDoubleValue = (double)
valueAggregator.applyRawValue(oldDoubleValue, rawValue);
+ kind = MetricWriteKind.DOUBLE;
+ break;
+ default:
+ throw new UnsupportedOperationException(String.format("Aggregation
type %s of %s not supported for %s",
+ valueAggregator.getAggregatedValueType(),
valueAggregator.getAggregationType(), dataType));
+ }
+ return PreparedMetricWrite.numeric(column, forwardIndex, kind,
oldDoubleValue, newDoubleValue);
}
+ case LONG: {
+ long oldLongValue;
+ long newLongValue;
+ MetricWriteKind kind;
+ switch (dataType) {
+ case INT:
+ oldLongValue = forwardIndex.getInt(docId);
+ newLongValue = (long) valueAggregator.applyRawValue(oldLongValue,
rawValue);
+ kind = MetricWriteKind.INT;
+ break;
+ case LONG:
+ oldLongValue = forwardIndex.getLong(docId);
+ newLongValue = (long) valueAggregator.applyRawValue(oldLongValue,
rawValue);
+ kind = MetricWriteKind.LONG;
+ break;
+ case FLOAT:
+ oldLongValue = (long) forwardIndex.getFloat(docId);
+ newLongValue = (long) valueAggregator.applyRawValue(oldLongValue,
rawValue);
+ kind = MetricWriteKind.FLOAT;
+ break;
+ case DOUBLE:
+ oldLongValue = (long) forwardIndex.getDouble(docId);
+ newLongValue = (long) valueAggregator.applyRawValue(oldLongValue,
rawValue);
+ kind = MetricWriteKind.DOUBLE;
+ break;
+ default:
+ throw new UnsupportedOperationException(String.format("Aggregation
type %s of %s not supported for %s",
+ valueAggregator.getAggregatedValueType(),
valueAggregator.getAggregationType(), dataType));
+ }
+ return PreparedMetricWrite.numeric(column, forwardIndex, kind,
oldLongValue, newLongValue);
+ }
+ case BYTES: {
+ Object oldValue =
valueAggregator.deserializeAggregatedValue(forwardIndex.getBytes(docId));
+ Object newValue = valueAggregator.applyRawValue(oldValue, rawValue);
+ return PreparedMetricWrite.bytes(column, forwardIndex,
valueAggregator.serializeAggregatedValue(oldValue),
+ valueAggregator.serializeAggregatedValue(newValue));
Review Comment:
[P1] Capture the original aggregate bytes before applying the update
applyRawValue() can mutate oldValue and return that same object:
AvgValueAggregator calls value.apply(), and HLL aggregators mutate their
sketch. Serializing oldValue afterward therefore saves the updated aggregate as
the rollback value. If this metric commits and a later metric write fails,
restoreMetricWrite() restores the new bytes, silently retaining only part of
the failed row. Please capture the original bytes before applyRawValue(), or
clone the old aggregate first. Extend rollback coverage with a mutable BYTES
aggregate such as AVG followed by a failing metric; the current numeric LONG
tests cannot expose this aliasing.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:
##########
@@ -852,192 +898,527 @@ private void validateNumMultiValues(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) {
+ RowWriteState state = _rowWriteState;
+ boolean dictHadError = updateDictionary(row, state);
+ boolean rowHadError = addNewRow(docId, row, state);
+ if (dictHadError || rowHadError) {
+ recordIncompleteRow();
+ }
+ if (state._repairFailed) {
+ markUnrecoverable(state);
+ throwPendingRowIndexingExceptionIfStrict();
+ throw new IllegalStateException(
+ "Mutable segment " + _segmentName + " is terminal after an
unrecoverable indexing failure");
+ }
+ }
+
+ /// @return {@code true} if any column required a default/fallback while
updating dictionaries.
+ /// When `continueOnError` is false, dictionary failures are stashed and
rethrown after the row is published.
+ private boolean updateDictionary(GenericRow row, RowWriteState state) {
+ 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) {
+ recordIndexingError("DICTIONARY");
+ hadError = true;
+ state.markIncomplete();
+ 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) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ hadError = true;
+ state.noteError(e);
+ indexContainer._dictId = Integer.MIN_VALUE;
+ indexContainer._dictIds = null;
+ try {
+ // Index the field default instead of leaving the sentinel: with
metrics aggregation the rollup key is built
+ // from the dict ids (see getOrCreateDocId), and Integer.MIN_VALUE
is not a real dict id.
+ Object defaultValue =
getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ if (indexContainer._fieldSpec.isSingleValueField()) {
+ indexContainer._dictId = dictionary.index(defaultValue);
+ } else {
+ indexContainer._dictIds = dictionary.index((Object[])
defaultValue);
+ }
+ row.putDefaultNullValue(column, defaultValue);
+ indexContainer._minValue = dictionary.getMinVal();
+ indexContainer._maxValue = dictionary.getMaxVal();
+ } catch (Exception fallbackError) {
+ _logger.error("Failed to index default null value into dictionary
for column: {}", column, fallbackError);
+ state.noteRepairFailure(fallbackError);
+ }
}
- updateIndexCapacityThresholdBreached(dictionary, entry.getKey());
+ updateIndexCapacityThresholdBreached(dictionary, column);
}
+ return hadError;
}
- private void addNewRow(int docId, GenericRow row) {
+ /// Indexes a new physical row. Always completes the row so seal/query
lengths stay aligned with [_numDocsIndexed]
+ /// (issue #16316). Forward indexes are written first from a finalized
value; secondary indexes then consume that
+ /// same value. On unrecoverable repair the segment is marked terminal.
+ ///
+ /// @return {@code true} if any column required a default/fallback while
indexing
+ private boolean addNewRow(int docId, GenericRow row, RowWriteState state) {
+ 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, state))
{
+ rowHadError = true;
+ }
+ } else if (!addPhysicalColumn(docId, row, column, indexContainer,
state)) {
+ 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) {
+ recordOrDeferIndexingError("ROW", e);
+ rowHadError = true;
+ state.noteError(e);
+ try {
+ indexDefaultNullColumn(docId, column, indexContainer, state);
+ } catch (Exception fallbackError) {
+ _logger.error("Failed to index default null for column: {} at docId:
{}", column, docId, fallbackError);
+ state.noteRepairFailure(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 {
+ // Multi-column text is a row-level secondary index. It is written
only after every column has a finalized
+ // value in `_multiColumnValues`. There is no per-doc rollback; a
failure leaves this doc absent from the
+ // text index while forward indexes remain complete. Counted once as a
secondary-index error.
+ _multiColumnTextIndex.add(_multiColumnValues);
+ } catch (Exception e) {
+ recordOrDeferIndexingError("MULTI_COLUMN_TEXT", e);
+ rowHadError = true;
+ state.noteError(e);
+ } finally {
+ Collections.fill(_multiColumnValues, null);
}
+ }
+ if (state._repairFailed) {
+ markUnrecoverable(state);
+ }
+ 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,
+ RowWriteState state) {
+ ValueAggregator valueAggregator = indexContainer._valueAggregator;
+ String sourceColumn = indexContainer._sourceColumn;
+ Object value = row.getValue(sourceColumn);
+ if (value == null &&
sourceColumn.equals(AggregationFunctionColumnPair.STAR)) {
+ assert valueAggregator.getAggregationType() ==
AggregationFunctionType.COUNT;
+ value = 1;
+ }
+
+ MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ try {
+ value = valueAggregator.getInitialAggregatedValue(value);
+ addAggregatedForwardValue(forwardIndex, valueAggregator, dataType,
value, docId, column);
+ indexContainer._valuesInfo.updateSVNumValues();
+ state._columnsWithCanonicalForward.add(column);
+ return true;
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.forward(), e);
+ state.noteError(e);
+ try {
+ indexDefaultAggregatedValue(docId, column, indexContainer, state);
+ } catch (Exception fallbackError) {
+ state.noteRepairFailure(fallbackError);
}
+ return false;
+ }
+ }
- FieldSpec fieldSpec = indexContainer._fieldSpec;
- DataType dataType = fieldSpec.getDataType();
+ private void addAggregatedForwardValue(MutableIndex forwardIndex,
ValueAggregator valueAggregator, DataType dataType,
+ Object value, int docId, String column) {
+ maybeIntercept(column, StandardIndexes.forward(), value, docId);
+ 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);
+ }
+ }
- if (fieldSpec.isSingleValueField()) {
- // Update numValues info
- indexContainer._valuesInfo.updateSVNumValues();
-
- // Route OPEN_STRUCT values to the dedicated mutable index.
OPEN_STRUCT has no forward
- // index / dictionary / min-max, so the standard per-IndexType loop
and the comparable
- // tracking below would be no-ops at best and crash at worst (Map is
not Comparable).
- if (dataType == DataType.OPEN_STRUCT) {
- MutableIndex openStructIndex =
indexContainer._mutableIndexes.get(StandardIndexes.openStruct());
- if (openStructIndex != null) {
- openStructIndex.add(value, -1, docId);
- }
- continue;
+ /// Returns {@code true} when the physical column was written from the row
value without error and without falling
+ /// back to the field default.
+ private boolean addPhysicalColumn(int docId, GenericRow row, String column,
IndexContainer indexContainer,
+ RowWriteState state) {
+ FieldSpec fieldSpec = indexContainer._fieldSpec;
+ DataType dataType = fieldSpec.getDataType();
+ boolean isNull = row.isNullValue(column);
+ Object value = row.getValue(column);
+ boolean defaultSubstituted = false;
+ if (value == null) {
+ recordIndexingError("NULL_VALUE");
+ value = getDefaultNullValueForIndexing(fieldSpec);
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ }
+
+ if (fieldSpec.isSingleValueField()) {
+ if (dataType == DataType.OPEN_STRUCT) {
+ return addOpenStructColumn(docId, column, indexContainer, value,
defaultSubstituted, state);
+ }
+ int dictId = indexContainer._dictId;
+ if (indexContainer._dictionary != null && dictId == Integer.MIN_VALUE) {
+ try {
+ Object defaultValue = getDefaultNullValueForIndexing(fieldSpec);
+ dictId = indexContainer._dictionary.index(defaultValue);
+ indexContainer._dictId = dictId;
+ value = defaultValue;
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ } catch (Exception e) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ state.noteError(e);
+ state.noteRepairFailure(e);
+ return false;
}
+ }
+ Object parsedJsonValue = defaultSubstituted ? null :
row.getParsedJsonValue(column);
+ boolean hadError = writeSingleValueIndexes(docId, column,
indexContainer, fieldSpec, dataType, value, dictId,
+ isNull, parsedJsonValue, state);
+ return !hadError && !defaultSubstituted;
+ }
- // Update indexes
- int dictId = indexContainer._dictId;
- for (Map.Entry<IndexType, MutableIndex> indexEntry :
indexContainer._mutableIndexes.entrySet()) {
- try {
- MutableIndex mutableIndex = indexEntry.getValue();
- if (mutableIndex instanceof MutableJsonIndexImpl) {
- // Feed the JSON index the parsed value cached by the
DataTypeTransformer (if any), so it flattens the
- // document directly instead of re-parsing the string this
column was serialized into for the forward
- // index; the index falls back to that string for documents it
cannot flatten directly. This is
- // intentionally scoped to Pinot's built-in JSON index
implementation.
- Object parsedValue = row.getParsedJsonValue(column);
- if (parsedValue != null && value instanceof String) {
- ((MutableJsonIndexImpl) mutableIndex).addParsed(parsedValue,
(String) value);
- updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
- continue;
- }
- }
- mutableIndex.add(value, dictId, docId);
- updateIndexCapacityThresholdBreached(mutableIndex,
indexEntry.getKey(), column);
- } catch (Exception e) {
- recordIndexingError(indexEntry.getKey(), e);
- }
+ Object[] values = value instanceof Object[] ? (Object[]) value : new
Object[]{value};
+ int[] dictIds = indexContainer._dictIds;
+ if (indexContainer._dictionary != null && dictIds == null) {
+ try {
+ Object[] defaultValues = (Object[])
getDefaultNullValueForIndexing(fieldSpec);
+ dictIds = indexContainer._dictionary.index(defaultValues);
+ indexContainer._dictIds = dictIds;
+ values = defaultValues;
+ isNull = true;
+ defaultSubstituted = true;
+ state.markIncomplete();
+ } catch (Exception e) {
+ recordOrDeferIndexingError("DICTIONARY", e);
+ state.noteError(e);
+ state.noteRepairFailure(e);
+ return false;
+ }
+ }
+ boolean hadError = writeMultiValueIndexes(docId, column, indexContainer,
dataType, values, dictIds, isNull, state);
+ return !hadError && !defaultSubstituted;
+ }
+
+ private boolean addOpenStructColumn(int docId, String column, IndexContainer
indexContainer, Object value,
+ boolean defaultSubstituted, RowWriteState state) {
+ MutableIndex openStructIndex =
indexContainer._mutableIndexes.get(StandardIndexes.openStruct());
+ if (openStructIndex != null) {
+ try {
+ maybeIntercept(column, StandardIndexes.openStruct(), value, docId);
+ openStructIndex.add(value, -1, docId);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.openStruct(), e);
+ state.noteError(e);
+ try {
+ Object defaultValue =
getDefaultNullValueForIndexing(indexContainer._fieldSpec);
+ maybeIntercept(column, StandardIndexes.openStruct(), defaultValue,
docId);
+ openStructIndex.add(defaultValue, -1, docId);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception fallbackError) {
+ state.noteRepairFailure(fallbackError);
+ return false;
}
+ defaultSubstituted = true;
+ }
+ } else {
+ state._columnsWithCanonicalForward.add(column);
+ }
+ indexContainer._valuesInfo.updateSVNumValues();
+ return !defaultSubstituted;
+ }
- if (dictId < 0) {
- // Update min/max value from raw value
- // NOTE: Skip updating min/max value for aggregated metrics because
the value will change over time.
- if (!isAggregateMetricsEnabled() || fieldSpec.getFieldType() !=
FieldSpec.FieldType.METRIC) {
- Comparable comparable = toComparableValue(value, dataType, column);
- if (indexContainer._minValue == null) {
- indexContainer._minValue = comparable;
- indexContainer._maxValue = comparable;
- } else {
- if (comparable.compareTo(indexContainer._minValue) < 0) {
- indexContainer._minValue = comparable;
- }
- if (comparable.compareTo(indexContainer._maxValue) > 0) {
- indexContainer._maxValue = comparable;
- }
- }
+ /// Writes the canonical forward index first, then every secondary index
from that same finalized value.
+ private boolean writeSingleValueIndexes(int docId, String column,
IndexContainer indexContainer, FieldSpec fieldSpec,
+ DataType dataType, Object value, int dictId, boolean isNull, @Nullable
Object parsedJsonValue,
+ RowWriteState state) {
+ boolean hadError = false;
+ MutableIndex forwardIndex =
indexContainer._mutableIndexes.get(StandardIndexes.forward());
+ if (forwardIndex != null &&
!state._columnsWithCanonicalForward.contains(column)) {
+ try {
+ maybeIntercept(column, StandardIndexes.forward(), value, docId);
+ forwardIndex.add(value, dictId, docId);
+ updateIndexCapacityThresholdBreached(forwardIndex,
StandardIndexes.forward(), column);
+ state._columnsWithCanonicalForward.add(column);
+ } catch (Exception e) {
+ recordOrDeferIndexingError(StandardIndexes.forward(), e);
+ state.noteError(e);
+ hadError = true;
+ try {
+ Object defaultValue = getDefaultNullValueForIndexing(fieldSpec);
+ int defaultDictId = dictId;
+ if (indexContainer._dictionary != null) {
+ defaultDictId = indexContainer._dictionary.index(defaultValue);
+ indexContainer._dictId = defaultDictId;
Review Comment:
[P1] Refresh cached dictionary bounds after inserting the fallback value
_minValue and _maxValue were copied from the dictionary in
updateDictionary(), before this insertion. If the default is a new extreme,
those cached bounds remain stale. For example, with null handling disabled,
repairing the only dictionary INT row from 42 to Integer.MIN_VALUE leaves
datasource bounds [42,42]. ColumnValueSegmentPruner then incorrectly prunes the
consuming segment for WHERE col < 0 even though its repaired row matches.
Please refresh the cached bounds in fallback dictionary-insertion paths, and
cover the datasource bounds/range-pruning behavior with null handling disabled.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -735,12 +736,28 @@ private boolean processStreamEvents(MessageBatch
messageBatch, long idlePipeSlee
_serverMetrics.addMeteredGlobalValue(ServerMeter.REALTIME_BYTES_CONSUMED,
recordSerializedValueLength);
}
} catch (Exception e) {
- _numRowsErrored++;
- _numBytesDropped += rowSizeInBytes;
String errorMessage =
"Caught exception while indexing the record at offset: " +
offset + " , row: " + transformedRow;
_segmentLogger.error(errorMessage, e);
_realtimeTableDataManager.addSegmentError(_segmentNameStr, new
SegmentErrorInfo(now(), errorMessage, e));
+ // continueOnError=false finishes a started row, then rethrows.
A published repair is consumed;
+ // only an unpublished failure is a drop. An unrecoverable
repair is terminal.
+ if (_realtimeSegment.getNumDocsIndexed() > docsBeforeIndex) {
+ indexedMessageCount++;
+ _lastRowMetadata = metadata;
+ _lastConsumedTimestampMs = System.currentTimeMillis();
+ realtimeRowsConsumedMeter =
+ _serverMetrics.addMeteredTableValue(_clientId,
ServerMeter.REALTIME_ROWS_CONSUMED, 1,
+ realtimeRowsConsumedMeter);
+
_serverMetrics.addMeteredGlobalValue(ServerMeter.REALTIME_ROWS_CONSUMED, 1L);
+ } else {
+ _numRowsErrored++;
+ _numBytesDropped += rowSizeInBytes;
+ }
+ canTakeMore = _realtimeSegment.canTakeMoreRows();
+ if (!canTakeMore) {
+ throw new RuntimeException(errorMessage, e);
Review Comment:
[P1] Distinguish normal capacity completion from an unrecoverable failure
With the default continueOnError=false, a recoverable indexing error on the
row that fills the segment publishes the repaired row and then rethrows.
canTakeMoreRows() is false simply because _numDocsIndexed == _capacity, so this
branch throws before _currentOffset and _numRowsIndexed advance.
PartitionConsumer then transitions to State.ERROR instead of completing the
segment normally. A malformed JSON secondary-index value on the
capacity-filling row is one reachable example. Please check unrecoverable state
separately and let successfully published rows advance their offsets and
complete normally. StatelessRealtimeSegmentWriter.indexTransformedRow has the
same conflation. A real-segment regression with capacity one and a recoverable
JSON error would cover the distinction missing from the mocked tests.
--
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]
