From 444c9bd0be733555d58608c5c5c361219504b547 Mon Sep 17 00:00:00 2001 From: Kartik Khare Date: Tue, 8 Sep 2026 10:58:38 +0530 Subject: [PATCH 1/3] Count primary keys not replaced at removal, not from a stale bitmap clone The "Found N primary keys not replaced" warning and the upsertInconsistentRows/partialUpsertKeysNotReplaced metric are both computed from a validDocIds bitmap that was cloned before segment replacement started. Records for those keys can arrive while the replacement runs. Ingestion correctly moves them onto the newer consuming segment, but the replacement is still reading the pre-replacement clone, so keys that were replaced perfectly well are reported as not replaced. The metric is a false positive and the alert built on it is muted in production as a result. Count at the action boundary instead. removeSegment already decides, per key, whether the key still belongs to the segment being removed, inside computeIfPresent under a recordLocation.getSegment() == segment check. That is the authoritative answer. Return how many keys passed that check and report only those, so a key ingestion already moved is never counted. BasePartitionUpsertMetadataManager gains removeSegmentAndGetNumKeysRemoved(IndexSegment, MutableRoaringBitmap), which by default delegates to removeSegment and returns the bitmap cardinality, keeping existing metadata-manager implementations behaving exactly as before. Both ConcurrentMap managers override it and count real removals. The warning and the metric now only fire when the count is above zero. For the consistent-deletes manager the reporting also moves after doRemoveSegment, because that path walks every doc in the segment rather than the valid ones, so the removal itself is where the count becomes knowable. Postmortem: ZD#7481, RCA-299. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HyuTYJvV4ARAXR9wSXdjBi --- .../BasePartitionUpsertMetadataManager.java | 18 ++++- ...rentMapPartitionUpsertMetadataManager.java | 19 +++++ ...rtMetadataManagerForConsistentDeletes.java | 33 +++++--- ...asePartitionUpsertMetadataManagerTest.java | 80 +++++++++++++++++++ ...tadataManagerForConsistentDeletesTest.java | 30 +++++++ ...MapPartitionUpsertMetadataManagerTest.java | 26 ++++++ 6 files changed, 192 insertions(+), 14 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 166ff0825006..93e6cfdc6c65 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 @@ -685,10 +685,12 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable revertSegmentUpsertMetadata(oldSegment, segmentName, validDocIdsForOldSegment); return; } - _logger.warn("Found {} primary keys not replaced for segment: {}", - validDocIdsForOldSegment.getCardinality(), segmentName); - updateInconsistentRowsMetric(segmentName, validDocIdsForOldSegment.getCardinality()); - removeSegment(oldSegment, validDocIdsForOldSegment); + int numKeysStillNotReplaced = + removeSegmentAndGetNumKeysRemoved(oldSegment, validDocIdsForOldSegment); + if (numKeysStillNotReplaced > 0) { + _logger.warn("Found {} primary keys not replaced for segment: {}", numKeysStillNotReplaced, segmentName); + updateInconsistentRowsMetric(segmentName, numKeysStillNotReplaced); + } } } @@ -740,6 +742,14 @@ private MutableRoaringBitmap getValidDocIdsForOldSegment(IndexSegment oldSegment return oldSegment.getValidDocIds() != null ? oldSegment.getValidDocIds().getMutableRoaringBitmap() : null; } + /// Removes candidate keys and returns how many were still owned by the segment at removal time. Implementations + /// backed by concurrent metadata should override this method and count only removals that pass their authoritative + /// ownership check. The default preserves compatibility with existing metadata-manager implementations. + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, MutableRoaringBitmap validDocIds) { + removeSegment(segment, validDocIds); + return validDocIds.getCardinality(); + } + protected abstract void removeSegment(IndexSegment segment, MutableRoaringBitmap validDocIds); protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java index 57559b020694..0049beb3c878 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java @@ -209,11 +209,17 @@ protected void addSegmentWithoutUpsert(ImmutableSegmentImpl segment, ThreadSafeM @Override protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { + removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator); + } + + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { + AtomicInteger numKeysRemoved = new AtomicInteger(); while (primaryKeyIterator.hasNext()) { PrimaryKey primaryKey = primaryKeyIterator.next(); _primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction), (pk, recordLocation) -> { if (recordLocation.getSegment() == segment) { + numKeysRemoved.getAndIncrement(); if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { _previousKeyToRecordLocationMap.remove(pk); } @@ -222,6 +228,7 @@ protected void removeSegment(IndexSegment segment, Iterator primaryK return recordLocation; }); } + return numKeysRemoved.get(); } @Override @@ -291,6 +298,18 @@ protected void removeSegment(IndexSegment segment, MutableRoaringBitmap validDoc } } + @Override + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, MutableRoaringBitmap validDocIds) { + try (PrimaryKeyReader primaryKeyReader = new PrimaryKeyReader(segment, _primaryKeyColumns)) { + return removeSegmentAndGetNumKeysRemoved(segment, + UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, validDocIds)); + } catch (Exception e) { + throw new RuntimeException( + String.format("Caught exception while removing segment: %s, table: %s, message: %s", segment.getSegmentName(), + _tableNameWithType, e.getMessage()), e); + } + } + @Override public void doRemoveExpiredPrimaryKeys() { AtomicInteger numMetadataTTLKeysRemoved = new AtomicInteger(); 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 32352f7a3b0f..ec6cbbe86295 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 @@ -238,18 +238,23 @@ protected void addSegmentWithoutUpsert(ImmutableSegmentImpl segment, ThreadSafeM @Override protected void doRemoveSegment(IndexSegment segment) { + doRemoveSegmentAndGetNumKeysRemoved(segment); + } + + protected int doRemoveSegmentAndGetNumKeysRemoved(IndexSegment segment) { String segmentName = segment.getSegmentName(); _logger.info("Removing {} segment: {}, current primary key count: {}", segment instanceof ImmutableSegment ? "immutable" : "mutable", segmentName, getNumPrimaryKeys()); long startTimeMs = System.currentTimeMillis(); // For ConsistentDeletes, we need to iterate over ALL docs in the segment (not just valid ones) // to properly decrement distinctSegmentCount for every key that was ever in the segment + int numKeysRemoved = 0; try (PrimaryKeyReader primaryKeyReader = new PrimaryKeyReader(segment, _primaryKeyColumns)) { if (shouldRevertMetadataOnInconsistency(segment)) { revertAndRemoveSegment(segment, UpsertUtils.getRecordIterator(primaryKeyReader, segment.getSegmentMetadata().getTotalDocs())); } else { - removeSegment(segment, + numKeysRemoved = removeSegmentAndGetNumKeysRemoved(segment, UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, segment.getSegmentMetadata().getTotalDocs())); } } catch (Exception e) { @@ -262,6 +267,7 @@ protected void doRemoveSegment(IndexSegment segment) { updatePrimaryKeyGauge(numPrimaryKeys); _logger.info("Finished removing segment: {} in {}ms, current primary key count: {}", segmentName, System.currentTimeMillis() - startTimeMs, numPrimaryKeys); + return numKeysRemoved; } protected void removeSegment(IndexSegment segment, MutableRoaringBitmap validDocIds) { @@ -301,19 +307,19 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable doAddOrReplaceSegment((ImmutableSegmentImpl) segment, validDocIds, queryableDocIds, recordInfoIterator, oldSegment, validDocIdsForOldSegment); } - if (validDocIdsForOldSegment != null && !validDocIdsForOldSegment.isEmpty()) { - if (shouldRevertMetadataOnInconsistency(oldSegment)) { - revertSegmentUpsertMetadata(oldSegment, segmentName, validDocIdsForOldSegment); - return; - } - _logger.warn("Found {} primary keys not replaced for segment: {}", - validDocIdsForOldSegment.getCardinality(), segmentName); - updateInconsistentRowsMetric(segmentName, validDocIdsForOldSegment.getCardinality()); + if (validDocIdsForOldSegment != null && !validDocIdsForOldSegment.isEmpty() + && shouldRevertMetadataOnInconsistency(oldSegment)) { + revertSegmentUpsertMetadata(oldSegment, segmentName, validDocIdsForOldSegment); + return; } // 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 // distinctSegmentCount by 1 - doRemoveSegment(oldSegment); + int numKeysStillNotReplaced = doRemoveSegmentAndGetNumKeysRemoved(oldSegment); + if (numKeysStillNotReplaced > 0) { + _logger.warn("Found {} primary keys not replaced for segment: {}", numKeysStillNotReplaced, segmentName); + updateInconsistentRowsMetric(segmentName, numKeysStillNotReplaced); + } } finally { segmentLock.unlock(); } @@ -321,6 +327,11 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable @Override protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { + removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator); + } + + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { + AtomicInteger numKeysRemoved = new AtomicInteger(); // We need to decrease the distinctSegmentCount for each unique primary key in this deleting segment by 1 // as the occurrence of the key in this segment is being removed. We are taking a set of unique primary keys // to avoid double counting the same key in the same segment. @@ -330,6 +341,7 @@ protected void removeSegment(IndexSegment segment, Iterator primaryK _primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction), (pk, recordLocation) -> { if (recordLocation.getSegment() == segment) { + numKeysRemoved.getAndIncrement(); if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { _previousKeyToRecordLocationMap.remove(pk); } @@ -343,6 +355,7 @@ protected void removeSegment(IndexSegment segment, Iterator primaryK RecordLocation.decrementSegmentCount(recordLocation.getDistinctSegmentCount())); }); } + return numKeysRemoved.get(); } @Override diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java index ca3e2da05a55..18815fb9e6f8 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManagerTest.java @@ -998,6 +998,54 @@ public void testTrackNewlyAddedSegments() { "Failed to remove stale segments"); } + @Test + public void testReplaceSegmentDoesNotReportKeyMovedBeforeRemoval() + throws IOException { + UpsertContext upsertContext = mock(UpsertContext.class); + when(upsertContext.getConsistencyMode()).thenReturn(UpsertConfig.ConsistencyMode.NONE); + + ThreadSafeMutableRoaringBitmap liveValidDocIds = createDocIds(0); + IndexSegment oldSegment = mock(IndexSegment.class); + when(oldSegment.getSegmentName()).thenReturn("segment"); + when(oldSegment.getValidDocIds()).thenReturn(liveValidDocIds); + ImmutableSegmentImpl newSegment = mock(ImmutableSegmentImpl.class); + when(newSegment.getSegmentName()).thenReturn("segment"); + + try (ActionBoundaryMetadataManager upsertMetadataManager = + new ActionBoundaryMetadataManager("myTable", 0, upsertContext, () -> liveValidDocIds.remove(0), 0)) { + upsertMetadataManager.replaceSegment(newSegment, new ThreadSafeMutableRoaringBitmap(), null, + List.of().iterator(), oldSegment); + + assertEquals(upsertMetadataManager._numInconsistentRows, 0); + assertEquals(upsertMetadataManager._candidateValidDocIds.toArray(), new int[]{0}); + upsertMetadataManager.stop(); + } + } + + @Test + public void testReplaceSegmentReportsOnlyKeysStillOwnedAtRemoval() + throws IOException { + UpsertContext upsertContext = mock(UpsertContext.class); + when(upsertContext.getConsistencyMode()).thenReturn(UpsertConfig.ConsistencyMode.NONE); + + ThreadSafeMutableRoaringBitmap liveValidDocIds = createDocIds(0, 1); + IndexSegment oldSegment = mock(IndexSegment.class); + when(oldSegment.getSegmentName()).thenReturn("segment"); + when(oldSegment.getValidDocIds()).thenReturn(liveValidDocIds); + ImmutableSegmentImpl newSegment = mock(ImmutableSegmentImpl.class); + when(newSegment.getSegmentName()).thenReturn("segment"); + + try (ActionBoundaryMetadataManager upsertMetadataManager = + new ActionBoundaryMetadataManager("myTable", 0, upsertContext, () -> liveValidDocIds.remove(0), 1)) { + upsertMetadataManager.replaceSegment(newSegment, new ThreadSafeMutableRoaringBitmap(), null, + List.of().iterator(), oldSegment); + + assertEquals(upsertMetadataManager._numInconsistentRows, 1); + assertEquals(upsertMetadataManager._candidateValidDocIds.toArray(), new int[]{0, 1}); + upsertMetadataManager.stop(); + } + } + @Test public void testResolveComparisonTies() { // Build a record info list for testing @@ -1140,4 +1188,36 @@ protected int getPrevKeyToRecordLocationSize() { protected void clearPrevKeyToRecordLocation() { } } + + private static class ActionBoundaryMetadataManager extends DummyPartitionUpsertMetadataManager { + private final Runnable _duringSegmentReplacement; + private final int _numKeysRemovedAtActionBoundary; + private int _numInconsistentRows; + private MutableRoaringBitmap _candidateValidDocIds; + + private ActionBoundaryMetadataManager(String tableNameWithType, int partitionId, UpsertContext context, + Runnable duringSegmentReplacement, int numKeysRemovedAtActionBoundary) { + super(tableNameWithType, partitionId, context); + _duringSegmentReplacement = duringSegmentReplacement; + _numKeysRemovedAtActionBoundary = numKeysRemovedAtActionBoundary; + } + + @Override + protected void doAddOrReplaceSegment(ImmutableSegmentImpl segment, ThreadSafeMutableRoaringBitmap validDocIds, + @Nullable ThreadSafeMutableRoaringBitmap queryableDocIds, Iterator recordInfoIterator, + @Nullable IndexSegment oldSegment, @Nullable MutableRoaringBitmap validDocIdsForOldSegment) { + _duringSegmentReplacement.run(); + } + + @Override + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, MutableRoaringBitmap validDocIds) { + _candidateValidDocIds = validDocIds.clone(); + return _numKeysRemovedAtActionBoundary; + } + + @Override + protected void updateInconsistentRowsMetric(String segmentName, int numKeysStillNotReplaced) { + _numInconsistentRows += numKeysStillNotReplaced; + } + } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java index 52a688907f89..28d5a7e2ab74 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java @@ -1350,4 +1350,34 @@ public void testNoRevertForImmutableSegmentReplacement() upsertMetadataManager.stop(); upsertMetadataManager.close(); } + + @Test + public void testRemoveSegmentCountsOnlyKeysStillOwnedBySegment() + throws IOException { + ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes upsertMetadataManager = + new ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes(REALTIME_TABLE_NAME, 0, + _contextBuilder.build()); + IndexSegment oldSegment = mock(IndexSegment.class); + IndexSegment newerSegment = mock(IndexSegment.class); + PrimaryKey oldSegmentKey = makePrimaryKey(10); + PrimaryKey newerSegmentKey = makePrimaryKey(20); + Object oldSegmentMapKey = HashUtils.hashPrimaryKey(oldSegmentKey, HashFunction.NONE); + Object newerSegmentMapKey = HashUtils.hashPrimaryKey(newerSegmentKey, HashFunction.NONE); + upsertMetadataManager._primaryKeyToRecordLocationMap.put(oldSegmentMapKey, + new ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.RecordLocation(oldSegment, 0, 100, 1)); + upsertMetadataManager._primaryKeyToRecordLocationMap.put(newerSegmentMapKey, + new ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.RecordLocation(newerSegment, 1, 200, 2)); + + int numKeysRemoved = upsertMetadataManager.removeSegmentAndGetNumKeysRemoved(oldSegment, + List.of(oldSegmentKey, newerSegmentKey).iterator()); + + assertEquals(numKeysRemoved, 1); + assertFalse(upsertMetadataManager._primaryKeyToRecordLocationMap.containsKey(oldSegmentMapKey)); + ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.RecordLocation newerLocation = + upsertMetadataManager._primaryKeyToRecordLocationMap.get(newerSegmentMapKey); + assertSame(newerLocation.getSegment(), newerSegment); + assertEquals(newerLocation.getDistinctSegmentCount(), 1); + upsertMetadataManager.stop(); + upsertMetadataManager.close(); + } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java index a3dcfb12821c..f71be5b3499a 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java @@ -206,6 +206,32 @@ public void testAddReplaceRemoveSegment() verifyAddReplaceRemoveSegment(HashFunction.MURMUR3, true); } + @Test + public void testRemoveSegmentCountsOnlyKeysStillOwnedBySegment() + throws IOException { + ConcurrentMapPartitionUpsertMetadataManager upsertMetadataManager = + new ConcurrentMapPartitionUpsertMetadataManager(REALTIME_TABLE_NAME, 0, _contextBuilder.build()); + IndexSegment oldSegment = mock(IndexSegment.class); + IndexSegment newerSegment = mock(IndexSegment.class); + PrimaryKey oldSegmentKey = makePrimaryKey(10); + PrimaryKey newerSegmentKey = makePrimaryKey(20); + Object oldSegmentMapKey = HashUtils.hashPrimaryKey(oldSegmentKey, HashFunction.NONE); + Object newerSegmentMapKey = HashUtils.hashPrimaryKey(newerSegmentKey, HashFunction.NONE); + upsertMetadataManager._primaryKeyToRecordLocationMap.put(oldSegmentMapKey, + new RecordLocation(oldSegment, 0, 100)); + upsertMetadataManager._primaryKeyToRecordLocationMap.put(newerSegmentMapKey, + new RecordLocation(newerSegment, 1, 200)); + + int numKeysRemoved = upsertMetadataManager.removeSegmentAndGetNumKeysRemoved(oldSegment, + List.of(oldSegmentKey, newerSegmentKey).iterator()); + + assertEquals(numKeysRemoved, 1); + assertFalse(upsertMetadataManager._primaryKeyToRecordLocationMap.containsKey(oldSegmentMapKey)); + assertSame(upsertMetadataManager._primaryKeyToRecordLocationMap.get(newerSegmentMapKey).getSegment(), newerSegment); + upsertMetadataManager.stop(); + upsertMetadataManager.close(); + } + @Test public void testRemoveExpiredPrimaryKeys() throws IOException { From 71347758cdcf0dd94fb5a69cd723367389781375 Mon Sep 17 00:00:00 2001 From: Kartik Khare Date: Tue, 8 Sep 2026 11:36:25 +0530 Subject: [PATCH 2/3] Name a bounded sample of the removed primary keys at DEBUG The count alone tells you a table is inconsistent but not which rows, so the next step is always a hunt. Keep the first few primary keys that fail the ownership check and log them. Free when nobody is looking. The list is not allocated unless the manager's logger has DEBUG enabled, so the steady state is one null check per removed key. The keys themselves cost nothing to obtain: they are already in hand at the increment, and UpsertUtils.getPrimaryKeyIterator hands out a fresh PrimaryKey per doc (PrimaryKeyReader.getPrimaryKey(int) allocates), so holding a reference is safe and needs no copy. Bounded at 8 keys per removal. DEBUG rather than WARN on purpose. A primary key is customer data, and every other log line in this package reports key counts rather than key values. The count stays at WARN where it was. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HyuTYJvV4ARAXR9wSXdjBi --- .../upsert/BasePartitionUpsertMetadataManager.java | 4 ++++ ...ConcurrentMapPartitionUpsertMetadataManager.java | 13 +++++++++++++ ...onUpsertMetadataManagerForConsistentDeletes.java | 13 +++++++++++++ 3 files changed, 30 insertions(+) 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 93e6cfdc6c65..72ccd78e33fc 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 @@ -742,6 +742,10 @@ private MutableRoaringBitmap getValidDocIdsForOldSegment(IndexSegment oldSegment return oldSegment.getValidDocIds() != null ? oldSegment.getValidDocIds().getMutableRoaringBitmap() : null; } + /// How many primary keys to name in the DEBUG log when a segment's keys are removed. Bounded because the count + /// alone can be large, and because primary keys are customer data. + protected static final int NUM_SAMPLED_REMOVED_KEYS = 8; + /// Removes candidate keys and returns how many were still owned by the segment at removal time. Implementations /// backed by concurrent metadata should override this method and count only removals that pass their authoritative /// ownership check. The default preserves compatibility with existing metadata-manager implementations. diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java index 0049beb3c878..3cad8478d902 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java @@ -19,8 +19,10 @@ package org.apache.pinot.segment.local.upsert; import com.google.common.annotations.VisibleForTesting; +import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.ConcurrentHashMap; @@ -214,12 +216,19 @@ protected void removeSegment(IndexSegment segment, Iterator primaryK protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { AtomicInteger numKeysRemoved = new AtomicInteger(); + // Naming the keys is only worth its memory when someone is reading DEBUG, so the list is not allocated + // otherwise. Everything else in this package logs key counts rather than key values, because a primary key is + // customer data. + List sampledKeysRemoved = _logger.isDebugEnabled() ? new ArrayList<>(NUM_SAMPLED_REMOVED_KEYS) : null; while (primaryKeyIterator.hasNext()) { PrimaryKey primaryKey = primaryKeyIterator.next(); _primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction), (pk, recordLocation) -> { if (recordLocation.getSegment() == segment) { numKeysRemoved.getAndIncrement(); + if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_REMOVED_KEYS) { + sampledKeysRemoved.add(primaryKey); + } if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { _previousKeyToRecordLocationMap.remove(pk); } @@ -228,6 +237,10 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

primaryK protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { AtomicInteger numKeysRemoved = new AtomicInteger(); + // Naming the keys is only worth its memory when someone is reading DEBUG, so the list is not allocated + // otherwise. Everything else in this package logs key counts rather than key values, because a primary key is + // customer data. + List sampledKeysRemoved = _logger.isDebugEnabled() ? new ArrayList<>(NUM_SAMPLED_REMOVED_KEYS) : null; // We need to decrease the distinctSegmentCount for each unique primary key in this deleting segment by 1 // as the occurrence of the key in this segment is being removed. We are taking a set of unique primary keys // to avoid double counting the same key in the same segment. @@ -342,6 +348,9 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

{ if (recordLocation.getSegment() == segment) { numKeysRemoved.getAndIncrement(); + if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_REMOVED_KEYS) { + sampledKeysRemoved.add(primaryKey); + } if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { _previousKeyToRecordLocationMap.remove(pk); } @@ -355,6 +364,10 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

Date: Tue, 8 Sep 2026 13:16:02 +0530 Subject: [PATCH 3/3] Name the unreplaced keys in the existing warning instead of a separate DEBUG line Follow-up on the DEBUG sample: put the keys in the warning that already reports the count, so one line carries both. Found 2 primary keys not replaced for segment: X, first 2: [[200], [300]] Flipping the DEBUG line to WARN in place would have been wrong. removeSegmentAndGetNumKeysRemoved is also on the plain segment-removal path, where removing every key the segment owns is normal, so a warning there would fire on every retention deletion and name customer keys for a healthy operation. Instead the sample is collected into a caller-supplied list. The replacement path passes one and reports it; the plain removal path passes null and collects nothing, so it neither samples nor logs. The base default cannot tell which candidates were actually removed, so it names no keys and logKeysNotReplaced falls back to the count-only message. The isDebugEnabled gate is gone with the DEBUG line. The cost is now one null check per removed key on the plain path, and one bounded ArrayList per replacement. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01HyuTYJvV4ARAXR9wSXdjBi --- .../BasePartitionUpsertMetadataManager.java | 33 ++++++++++++++----- ...rentMapPartitionUpsertMetadataManager.java | 21 ++++-------- ...rtMetadataManagerForConsistentDeletes.java | 28 +++++++--------- ...asePartitionUpsertMetadataManagerTest.java | 3 +- ...tadataManagerForConsistentDeletesTest.java | 5 ++- ...MapPartitionUpsertMetadataManagerTest.java | 5 ++- 6 files changed, 53 insertions(+), 42 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 72ccd78e33fc..9c641403ad67 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 @@ -685,10 +685,11 @@ public void replaceSegment(ImmutableSegment segment, @Nullable ThreadSafeMutable revertSegmentUpsertMetadata(oldSegment, segmentName, validDocIdsForOldSegment); return; } + List sampledKeysNotReplaced = new ArrayList<>(NUM_SAMPLED_KEYS_NOT_REPLACED); int numKeysStillNotReplaced = - removeSegmentAndGetNumKeysRemoved(oldSegment, validDocIdsForOldSegment); + removeSegmentAndGetNumKeysRemoved(oldSegment, validDocIdsForOldSegment, sampledKeysNotReplaced); if (numKeysStillNotReplaced > 0) { - _logger.warn("Found {} primary keys not replaced for segment: {}", numKeysStillNotReplaced, segmentName); + logKeysNotReplaced(segmentName, numKeysStillNotReplaced, sampledKeysNotReplaced); updateInconsistentRowsMetric(segmentName, numKeysStillNotReplaced); } } @@ -742,18 +743,32 @@ private MutableRoaringBitmap getValidDocIdsForOldSegment(IndexSegment oldSegment return oldSegment.getValidDocIds() != null ? oldSegment.getValidDocIds().getMutableRoaringBitmap() : null; } - /// How many primary keys to name in the DEBUG log when a segment's keys are removed. Bounded because the count - /// alone can be large, and because primary keys are customer data. - protected static final int NUM_SAMPLED_REMOVED_KEYS = 8; + /// How many of the unreplaced primary keys to name in the log. Bounded because the count itself can be large, + /// and because a primary key is customer data. + protected static final int NUM_SAMPLED_KEYS_NOT_REPLACED = 8; - /// Removes candidate keys and returns how many were still owned by the segment at removal time. Implementations - /// backed by concurrent metadata should override this method and count only removals that pass their authoritative - /// ownership check. The default preserves compatibility with existing metadata-manager implementations. - protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, MutableRoaringBitmap validDocIds) { + /// Removes candidate keys and returns how many were still owned by the segment at removal time, appending up to + /// [#NUM_SAMPLED_KEYS_NOT_REPLACED] of them to `sampledKeysRemoved` when it is given. Implementations backed by + /// concurrent metadata should override this and count only removals that pass their authoritative ownership check. + /// The default preserves compatibility with existing metadata-manager implementations and names no keys, since it + /// cannot tell which of the candidates were actually removed. + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, MutableRoaringBitmap validDocIds, + @Nullable List sampledKeysRemoved) { removeSegment(segment, validDocIds); return validDocIds.getCardinality(); } + /// Logs the unreplaced-key count, naming a few of the keys when the implementation could identify them. + protected void logKeysNotReplaced(String segmentName, int numKeysStillNotReplaced, + List sampledKeysNotReplaced) { + if (sampledKeysNotReplaced.isEmpty()) { + _logger.warn("Found {} primary keys not replaced for segment: {}", numKeysStillNotReplaced, segmentName); + } else { + _logger.warn("Found {} primary keys not replaced for segment: {}, first {}: {}", numKeysStillNotReplaced, + segmentName, sampledKeysNotReplaced.size(), sampledKeysNotReplaced); + } + } + protected abstract void removeSegment(IndexSegment segment, MutableRoaringBitmap validDocIds); protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java index 3cad8478d902..3b992bc6db57 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java @@ -19,7 +19,6 @@ package org.apache.pinot.segment.local.upsert; import com.google.common.annotations.VisibleForTesting; -import java.util.ArrayList; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -211,22 +210,19 @@ protected void addSegmentWithoutUpsert(ImmutableSegmentImpl segment, ThreadSafeM @Override protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { - removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator); + removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator, null); } - protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator, + @Nullable List sampledKeysRemoved) { AtomicInteger numKeysRemoved = new AtomicInteger(); - // Naming the keys is only worth its memory when someone is reading DEBUG, so the list is not allocated - // otherwise. Everything else in this package logs key counts rather than key values, because a primary key is - // customer data. - List sampledKeysRemoved = _logger.isDebugEnabled() ? new ArrayList<>(NUM_SAMPLED_REMOVED_KEYS) : null; while (primaryKeyIterator.hasNext()) { PrimaryKey primaryKey = primaryKeyIterator.next(); _primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction), (pk, recordLocation) -> { if (recordLocation.getSegment() == segment) { numKeysRemoved.getAndIncrement(); - if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_REMOVED_KEYS) { + if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_KEYS_NOT_REPLACED) { sampledKeysRemoved.add(primaryKey); } if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { @@ -237,10 +233,6 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

sampledKeysRemoved) { try (PrimaryKeyReader primaryKeyReader = new PrimaryKeyReader(segment, _primaryKeyColumns)) { return removeSegmentAndGetNumKeysRemoved(segment, - UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, validDocIds)); + UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, validDocIds), sampledKeysRemoved); } catch (Exception e) { throw new RuntimeException( String.format("Caught exception while removing segment: %s, table: %s, message: %s", segment.getSegmentName(), 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 3e6084b09d40..55c91271c9e3 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 @@ -240,10 +240,11 @@ protected void addSegmentWithoutUpsert(ImmutableSegmentImpl segment, ThreadSafeM @Override protected void doRemoveSegment(IndexSegment segment) { - doRemoveSegmentAndGetNumKeysRemoved(segment); + doRemoveSegmentAndGetNumKeysRemoved(segment, null); } - protected int doRemoveSegmentAndGetNumKeysRemoved(IndexSegment segment) { + protected int doRemoveSegmentAndGetNumKeysRemoved(IndexSegment segment, + @Nullable List sampledKeysRemoved) { String segmentName = segment.getSegmentName(); _logger.info("Removing {} segment: {}, current primary key count: {}", segment instanceof ImmutableSegment ? "immutable" : "mutable", segmentName, getNumPrimaryKeys()); @@ -257,7 +258,8 @@ protected int doRemoveSegmentAndGetNumKeysRemoved(IndexSegment segment) { UpsertUtils.getRecordIterator(primaryKeyReader, segment.getSegmentMetadata().getTotalDocs())); } else { numKeysRemoved = removeSegmentAndGetNumKeysRemoved(segment, - UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, segment.getSegmentMetadata().getTotalDocs())); + UpsertUtils.getPrimaryKeyIterator(primaryKeyReader, segment.getSegmentMetadata().getTotalDocs()), + sampledKeysRemoved); } } catch (Exception e) { throw new RuntimeException( @@ -317,9 +319,10 @@ && shouldRevertMetadataOnInconsistency(oldSegment)) { // 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 // distinctSegmentCount by 1 - int numKeysStillNotReplaced = doRemoveSegmentAndGetNumKeysRemoved(oldSegment); + List sampledKeysNotReplaced = new ArrayList<>(NUM_SAMPLED_KEYS_NOT_REPLACED); + int numKeysStillNotReplaced = doRemoveSegmentAndGetNumKeysRemoved(oldSegment, sampledKeysNotReplaced); if (numKeysStillNotReplaced > 0) { - _logger.warn("Found {} primary keys not replaced for segment: {}", numKeysStillNotReplaced, segmentName); + logKeysNotReplaced(segmentName, numKeysStillNotReplaced, sampledKeysNotReplaced); updateInconsistentRowsMetric(segmentName, numKeysStillNotReplaced); } } finally { @@ -329,15 +332,12 @@ && shouldRevertMetadataOnInconsistency(oldSegment)) { @Override protected void removeSegment(IndexSegment segment, Iterator primaryKeyIterator) { - removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator); + removeSegmentAndGetNumKeysRemoved(segment, primaryKeyIterator, null); } - protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator) { + protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator primaryKeyIterator, + @Nullable List sampledKeysRemoved) { AtomicInteger numKeysRemoved = new AtomicInteger(); - // Naming the keys is only worth its memory when someone is reading DEBUG, so the list is not allocated - // otherwise. Everything else in this package logs key counts rather than key values, because a primary key is - // customer data. - List sampledKeysRemoved = _logger.isDebugEnabled() ? new ArrayList<>(NUM_SAMPLED_REMOVED_KEYS) : null; // We need to decrease the distinctSegmentCount for each unique primary key in this deleting segment by 1 // as the occurrence of the key in this segment is being removed. We are taking a set of unique primary keys // to avoid double counting the same key in the same segment. @@ -348,7 +348,7 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

{ if (recordLocation.getSegment() == segment) { numKeysRemoved.getAndIncrement(); - if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_REMOVED_KEYS) { + if (sampledKeysRemoved != null && sampledKeysRemoved.size() < NUM_SAMPLED_KEYS_NOT_REPLACED) { sampledKeysRemoved.add(primaryKey); } if (_context.isTableTypeInconsistentDuringConsumption() && segment instanceof MutableSegment) { @@ -364,10 +364,6 @@ protected int removeSegmentAndGetNumKeysRemoved(IndexSegment segment, Iterator

sampledKeysRemoved) { _candidateValidDocIds = validDocIds.clone(); return _numKeysRemovedAtActionBoundary; } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java index 28d5a7e2ab74..292f40a70a22 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletesTest.java @@ -1368,10 +1368,13 @@ public void testRemoveSegmentCountsOnlyKeysStillOwnedBySegment() upsertMetadataManager._primaryKeyToRecordLocationMap.put(newerSegmentMapKey, new ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.RecordLocation(newerSegment, 1, 200, 2)); + List sampledKeysRemoved = new ArrayList<>(); int numKeysRemoved = upsertMetadataManager.removeSegmentAndGetNumKeysRemoved(oldSegment, - List.of(oldSegmentKey, newerSegmentKey).iterator()); + List.of(oldSegmentKey, newerSegmentKey).iterator(), sampledKeysRemoved); assertEquals(numKeysRemoved, 1); + // Only the key still owned by the old segment is named, not the one ingestion already moved. + assertEquals(sampledKeysRemoved, List.of(oldSegmentKey)); assertFalse(upsertMetadataManager._primaryKeyToRecordLocationMap.containsKey(oldSegmentMapKey)); ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.RecordLocation newerLocation = upsertMetadataManager._primaryKeyToRecordLocationMap.get(newerSegmentMapKey); diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java index f71be5b3499a..3a1125e1020e 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerTest.java @@ -222,10 +222,13 @@ public void testRemoveSegmentCountsOnlyKeysStillOwnedBySegment() upsertMetadataManager._primaryKeyToRecordLocationMap.put(newerSegmentMapKey, new RecordLocation(newerSegment, 1, 200)); + List sampledKeysRemoved = new ArrayList<>(); int numKeysRemoved = upsertMetadataManager.removeSegmentAndGetNumKeysRemoved(oldSegment, - List.of(oldSegmentKey, newerSegmentKey).iterator()); + List.of(oldSegmentKey, newerSegmentKey).iterator(), sampledKeysRemoved); assertEquals(numKeysRemoved, 1); + // Only the key still owned by the old segment is named, not the one ingestion already moved. + assertEquals(sampledKeysRemoved, List.of(oldSegmentKey)); assertFalse(upsertMetadataManager._primaryKeyToRecordLocationMap.containsKey(oldSegmentMapKey)); assertSame(upsertMetadataManager._primaryKeyToRecordLocationMap.get(newerSegmentMapKey).getSegment(), newerSegment); upsertMetadataManager.stop();