Fail-soft MutableSegmentImpl.index to avoid partial-row corruption - #19088
Fail-soft MutableSegmentImpl.index to avoid partial-row corruption#19088Vamsi-klu wants to merge 3 commits into
Conversation
Exceptions mid-index left some columns updated and others not, causing seal failures and stuck consuming segments. Finish the row with default/null fills, meter incomplete rows, and clear stale aggregateMetrics multi-value FIXME.
Oversized BigDecimal values no longer abort mid-row; the row completes with the aggregator default so the mutable segment remains consistent and sealable.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19088 +/- ##
============================================
- Coverage 65.47% 65.43% -0.04%
Complexity 1421 1421
============================================
Files 3426 3426
Lines 217315 217498 +183
Branches 34509 34543 +34
============================================
+ Hits 142283 142325 +42
- Misses 63513 63638 +125
- Partials 11519 11535 +16
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
Ready for review — all required CI checks are green on this PR. Issue: #16316 Could the following folks take a look when convenient? Formal GitHub "Request review" is unavailable from a fork contributor account on Thank you! |
Jackie-Jiang
added
enhancement
There was a problem hiding this comment.
Pull request overview
This PR hardens realtime mutable segment ingestion by making MutableSegmentImpl.index() “fail-soft” at the row level: when mid-row indexing/transform/dictionary/secondary-index errors occur, the segment still completes the physical row (using default/null fallbacks) so column lengths remain aligned and the segment stays sealable, while metering incomplete rows for observability.
Changes:
- Add fail-soft indexing flow (
indexPhysicalRow, updatedupdateDictionary/addNewRow) to complete rows with defaults on errors and meter incomplete rows. - Treat forward-index failures as row-integrity issues (attempt fallback write instead of silently leaving holes) and add row-level incomplete-row metering.
- Extend tests to validate fail-soft behavior for ingestion aggregations and for secondary-index (JSON) failure scenarios.
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java |
Implements fail-soft row completion, error metering, and forward/secondary index fallback behavior during mutable segment ingestion. |
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/IndexingFailureTest.java |
Updates/extends unit tests to assert row completion behavior and incomplete-row metering when indexing fails mid-row. |
pinot-segment-local/src/test/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImplIngestionAggregationTest.java |
Adjusts aggregation ingestion test to validate fail-soft behavior (row completes with aggregator default instead of throwing). |
Comments suppressed due to low confidence (2)
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:1032
- addPhysicalColumn() relies on iterating indexContainer._mutableIndexes (a HashMap via MutableIndexes), so index write order is nondeterministic. If a non-forward index happens to be written before the forward index, and the forward index then fails and is replaced with a default value, the earlier index can still contain the original value for this docId, leaving the column internally inconsistent (filters/secondary indexes disagree with forward value).
To preserve the PR's per-row consistency goal, the forward index should be written first (including any default fallback), and only then should other indexes be updated using the finalized value/dictId (or the column should be treated as fully defaulted).
boolean forwardWritten = false;
boolean hadError = false;
for (Map.Entry<IndexType, MutableIndex> indexEntry : indexContainer._mutableIndexes.entrySet()) {
IndexType indexType = indexEntry.getKey();
try {
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java:1124
- Same issue as the single-value path: the multi-value branch iterates indexContainer._mutableIndexes (HashMap) in nondeterministic order. If a secondary index is written before the MV forward index and the forward write then fails and falls back to defaultValues, the column can end up with secondary-index entries for the original values but a forward entry for the default values.
Consider explicitly writing the MV forward index first (including fallback), then writing all other indexes using the final values/dictIds so the column remains self-consistent for that docId.
boolean forwardWritten = false;
boolean hadError = false;
indexContainer._valuesInfo.updateVarByteMVMaxRowLengthInBytes(values, dataType.getStoredType());
for (Map.Entry<IndexType, MutableIndex> indexEntry : indexContainer._mutableIndexes.entrySet()) {
IndexType indexType = indexEntry.getKey();
try {
| } 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; | ||
| } |
| verify(_serverMetrics, times(1)).addMeteredTableValue(matches("DICTIONARY-indexingError$"), | ||
| eq(ServerMeter.INDEXING_FAILURES), eq(1L)); |
| 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; | ||
| } |
Why
MutableSegmentImpl.index()is not transactional. On exception mid-row (dictionary/forward/agg), some columns were updated and others were not.RealtimeSegmentDataManagercatches the error and continues consuming on the same mutable segment, so corruption becomes durable until seal fails (IndexOutOfBoundsException) or queries lie.Production effect: a single bad row can poison a consuming partition (seal failure, stuck consumption, data loss risk).
Impact
aggregateMetricsFIXME (aggregation is already disabled/rejected for MV dims).How
index()/updateDictionary/addNewRow: on failure, fill remaining columns with field default/null values for that docId and advance doc count consistently.Test plan
IndexingFailureTestfor mid-row failure → segment remains consistent / sealable../mvnw -pl pinot-segment-local -am -Dtest=IndexingFailureTest -Dsurefire.failIfNoSpecifiedTests=false testRelated
fixes: #16316
related: #16317
Reviewers
Suggested: Jackie-Jiang (issue guidance on fail-soft defaults)
Was generative AI tooling used to co-author this PR?
Generated-by: Grok Build (xAI)