This is an automated email from the ASF dual-hosted git repository.
KKcorps pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 1aa0e4d06b9 Surface FULL-upsert inconsistencies during segment
replacement (#19110)
1aa0e4d06b9 is described below
commit 1aa0e4d06b9028b198aa85c6cdace274dce6a821
Author: Chaitanya Deepthi <[email protected]>
AuthorDate: Wed Jul 29 03:41:01 2026 -0700
Surface FULL-upsert inconsistencies during segment replacement (#19110)
Two minimal edits:
1. Flatten the outer isTableTypeInconsistentDuringConsumption gate in
BasePartitionUpsertMetadataManager.replaceSegment and the
ConsistentDeletes
override. shouldRevertMetadataOnInconsistency already implies this
predicate,
so the outer gate is redundant. Removing it also makes the non-revert
branch
run logInconsistentResults for FULL upsert tables, not just for
dropOutOfOrderRecord / outOfOrderRecordColumn / partial upsert tables.
2. Add an else branch to logInconsistentResults that emits
REALTIME_UPSERT_INCONSISTENT_ROWS for FULL upsert tables (with or without
deleteRecordColumn). Previously nothing was emitted for that case,
leaving
cross-replica drift on plain FULL upsert tables unalertable.
Behavior on the revert path, the partial-upsert path, and the
dropOutOfOrderRecord path is unchanged.
---
.../upsert/BasePartitionUpsertMetadataManager.java | 35 ++++++++++------------
...nUpsertMetadataManagerForConsistentDeletes.java | 13 ++++----
2 files changed, 22 insertions(+), 26 deletions(-)
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
index 837ce9c6431..7894275fa48 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java
@@ -693,16 +693,15 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
validDocIdsForOldSegment = getValidDocIdsForOldSegment(oldSegment);
}
if (validDocIdsForOldSegment != null &&
!validDocIdsForOldSegment.isEmpty()) {
- if (_context.isTableTypeInconsistentDuringConsumption()) {
- if (shouldRevertMetadataOnInconsistency(oldSegment)) {
- // If there are still valid docs in the old segment, validate and
revert the metadata of the
- // consuming segment in place
- revertSegmentUpsertMetadata(oldSegment, segmentName,
validDocIdsForOldSegment);
- return;
- } else {
- logInconsistentResults(segmentName,
validDocIdsForOldSegment.getCardinality());
- }
+ if (shouldRevertMetadataOnInconsistency(oldSegment)) {
+ // If there are still valid docs in the old segment, validate and
revert the metadata of the
+ // consuming segment in place
+ revertSegmentUpsertMetadata(oldSegment, segmentName,
validDocIdsForOldSegment);
+ return;
}
+ _logger.warn("Found {} primary keys not replaced for segment: {}",
+ validDocIdsForOldSegment.getCardinality(), segmentName);
+ updateInconsistentRowsMetric(segmentName,
validDocIdsForOldSegment.getCardinality());
removeSegment(oldSegment, validDocIdsForOldSegment);
}
}
@@ -737,23 +736,21 @@ public abstract class BasePartitionUpsertMetadataManager
implements PartitionUps
}
int numKeysStillNotReplaced = getPrevKeyToRecordLocationSize();
if (numKeysStillNotReplaced > 0) {
- logInconsistentResults(segmentName, numKeysStillNotReplaced);
+ _logger.warn("Found {} primary keys not replaced for segment: {} after
revert attempt",
+ numKeysStillNotReplaced, segmentName);
+ updateInconsistentRowsMetric(segmentName, numKeysStillNotReplaced);
// Clear the map when inconsistencies still exist for the consuming
segments
clearPrevKeyToRecordLocation();
}
}
- protected void logInconsistentResults(String segmentName, int
numKeysStillNotReplaced) {
- _logger.error("Found {} primary keys not replaced for segment: {}. "
- + "Proceeding with current state which may cause inconsistency. To
correct this behaviour from now, set "
- + "cluster config:
`pinot.server.consuming.segment.consistency.mode` to `PROTECTED`",
- numKeysStillNotReplaced, segmentName);
- if (_context.isDropOutOfOrderRecord() ||
_context.getOutOfOrderRecordColumn() != null) {
- _serverMetrics.addMeteredTableValue(_tableNameWithType,
ServerMeter.REALTIME_UPSERT_INCONSISTENT_ROWS,
- numKeysStillNotReplaced);
- } else if (_partialUpsertHandler != null) {
+ protected void updateInconsistentRowsMetric(String segmentName, int
numKeysStillNotReplaced) {
+ if (_partialUpsertHandler != null) {
_serverMetrics.addMeteredTableValue(_tableNameWithType,
ServerMeter.PARTIAL_UPSERT_KEYS_NOT_REPLACED,
numKeysStillNotReplaced);
+ } else {
+ _serverMetrics.addMeteredTableValue(_tableNameWithType,
ServerMeter.REALTIME_UPSERT_INCONSISTENT_ROWS,
+ numKeysStillNotReplaced);
}
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
index 929df2ac3ac..43961880f17 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java
@@ -304,14 +304,13 @@ public class
ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes
oldSegment, validDocIdsForOldSegment);
}
if (validDocIdsForOldSegment != null &&
!validDocIdsForOldSegment.isEmpty()) {
- if (_context.isTableTypeInconsistentDuringConsumption()) {
- if (shouldRevertMetadataOnInconsistency(oldSegment)) {
- revertSegmentUpsertMetadata(oldSegment, segmentName,
validDocIdsForOldSegment);
- return;
- } else {
- logInconsistentResults(segmentName,
validDocIdsForOldSegment.getCardinality());
- }
+ if (shouldRevertMetadataOnInconsistency(oldSegment)) {
+ revertSegmentUpsertMetadata(oldSegment, segmentName,
validDocIdsForOldSegment);
+ return;
}
+ _logger.warn("Found {} primary keys not replaced for segment: {}",
+ validDocIdsForOldSegment.getCardinality(), segmentName);
+ updateInconsistentRowsMetric(segmentName,
validDocIdsForOldSegment.getCardinality());
}
// we want to always remove a segment in case of
enableDeletedKeysCompactionConsistency = true
// this is to account for the removal of primary-key in the
to-be-removed segment and reduce
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
