You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by xi...@apache.org on 2026/03/16 00:48:00 UTC
(pinot) branch master updated: Added more tests for #17503 (#17882)
This is an automated email from the ASF dual-hosted git repository.
xiangfu 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 5f4ad42594f Added more tests for #17503 (#17882)
5f4ad42594f is described below
commit 5f4ad42594f31038306f63fc79fc0b594cfea1c2
Author: Chaitanya Deepthi <45...@users.noreply.github.com>
AuthorDate: Sun Mar 15 17:47:54 2026 -0700
Added more tests for #17503 (#17882)
* Remove the key from metadata map only if the key is newly added and is not present in any immutable segment location
* Test the _previousKeyToRecordLocationMap after segment commit and remove
* Fix the checkstyle issues
* Change the comment
---
...oncurrentMapPartitionUpsertMetadataManager.java | 15 ++--
...nUpsertMetadataManagerForConsistentDeletes.java | 15 ++--
...rrentMapPartitionUpsertMetadataManagerTest.java | 96 ++++++++++++++++++++++
3 files changed, 112 insertions(+), 14 deletions(-)
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 a6c0721e5fe..be9c0b17307 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
@@ -234,11 +234,11 @@ public class ConcurrentMapPartitionUpsertMetadataManager extends BasePartitionUp
_primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction),
(pk, recordLocation) -> {
RecordLocation prevLocation = _previousKeyToRecordLocationMap.get(pk);
- if (prevLocation == null) {
- _previousKeyToRecordLocationMap.remove(pk);
- return null;
- }
if (recordLocation.getSegment() == segment) {
+ if (prevLocation == null) {
+ _previousKeyToRecordLocationMap.remove(pk);
+ return null;
+ }
// Revert to previous segment location
IndexSegment prevSegment = prevLocation.getSegment();
ThreadSafeMutableRoaringBitmap prevValidDocIds = prevSegment.getValidDocIds();
@@ -265,10 +265,11 @@ public class ConcurrentMapPartitionUpsertMetadataManager extends BasePartitionUp
_previousKeyToRecordLocationMap.remove(pk);
} else {
_logger.warn(
- "Consuming segment: {} has already ingested the primary key for docId {} from segment: {}, suggesting"
+ "Consuming segment: {} has added the primary key for docId: {} from the segment: {}, suggesting"
+ " that consumption is occurring concurrently with segment replacement, which is undesirable "
- + "for consistency.", recordLocation.getSegment().getSegmentName(), primaryKeyEntry.getKey(),
- segment.getSegmentName());
+ + "for consistency between replicas for the table: {}.",
+ recordLocation.getSegment().getSegmentName(), primaryKeyEntry.getKey(), segment.getSegmentName(),
+ _tableNameWithType);
}
return recordLocation;
});
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 7535da9d3e7..d21120a0450 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
@@ -361,11 +361,11 @@ public class ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes
_primaryKeyToRecordLocationMap.computeIfPresent(HashUtils.hashPrimaryKey(primaryKey, _hashFunction),
(pk, recordLocation) -> {
RecordLocation prevLocation = _previousKeyToRecordLocationMap.get(pk);
- if (prevLocation == null) {
- _previousKeyToRecordLocationMap.remove(pk);
- return null;
- }
if (recordLocation.getSegment() == segment) {
+ if (prevLocation == null) {
+ _previousKeyToRecordLocationMap.remove(pk);
+ return null;
+ }
// Revert to previous segment location
IndexSegment prevSegment = prevLocation.getSegment();
ThreadSafeMutableRoaringBitmap prevValidDocIds = prevSegment.getValidDocIds();
@@ -398,10 +398,11 @@ public class ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes
_previousKeyToRecordLocationMap.remove(pk);
} else {
_logger.warn(
- "Consuming segment {} has already ingested the primary key for docId {} from segment {}, suggesting"
+ "Consuming segment: {} has added the primary key for docId: {} from the segment: {}, suggesting"
+ " that consumption is occurring concurrently with segment replacement, which is undesirable "
- + "for consistency.", recordLocation.getSegment().getSegmentName(), primaryKeyEntry.getKey(),
- segment.getSegmentName());
+ + "for consistency between replicas for the table: {}.",
+ recordLocation.getSegment().getSegmentName(), primaryKeyEntry.getKey(), segment.getSegmentName(),
+ _tableNameWithType);
}
if (!uniquePrimaryKeys.add(pk)) {
return recordLocation;
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 681f7d53bc3..e9cd61dace2 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
@@ -82,6 +82,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.when;
@@ -2361,6 +2362,101 @@ public class ConcurrentMapPartitionUpsertMetadataManagerTest {
assertEquals(recordLocationMap.size(), 1);
checkRecordLocation(recordLocationMap, 10, segment2, 0, 1500, HashFunction.NONE);
+ // Prev key map must be cleared after segment replacement irrespective of mode
+ assertTrue(upsertMetadataManager._previousKeyToRecordLocationMap.isEmpty());
+
+ upsertMetadataManager.stop();
+ upsertMetadataManager.close();
+ }
+
+ /**
+ * Verifies that _previousKeyToRecordLocationMap is cleared after segment commit (replace) and after segment remove,
+ * when not in PROTECTED mode. With PROTECTED mode the same replace assertion is covered by
+ * testProtectedModeRevertsMetadataForConsumingSegmentSeal.
+ */
+ @Test
+ public void testPrevKeyToRecordLocationMapClearedAfterSegmentCommitInNonProtectedMode()
+ throws IOException {
+ UpsertContext upsertContext =
+ _contextBuilder.setConsistencyMode(UpsertConfig.ConsistencyMode.NONE).setDropOutOfOrderRecord(true).build();
+
+ ConcurrentMapPartitionUpsertMetadataManager upsertMetadataManager =
+ new ConcurrentMapPartitionUpsertMetadataManager(REALTIME_TABLE_NAME, 0, upsertContext);
+ Set<IndexSegment> trackedSegments = upsertMetadataManager._trackedSegments;
+
+ // Add immutable segment, then consuming segment with records (populates _previousKeyToRecordLocationMap)
+ int numRecords = 3;
+ int[] primaryKeys = new int[]{10, 20, 30};
+ int[] timestamps = new int[]{100, 200, 300};
+ ThreadSafeMutableRoaringBitmap validDocIds1 = new ThreadSafeMutableRoaringBitmap();
+ List<PrimaryKey> primaryKeysList1 = getPrimaryKeyList(numRecords, primaryKeys);
+ ImmutableSegmentImpl segment1 = mockImmutableSegmentWithTimestamps(1, validDocIds1, null, primaryKeysList1,
+ timestamps);
+ upsertMetadataManager.addSegment(segment1, validDocIds1, null,
+ getRecordInfoListWithIntegerComparison(numRecords, primaryKeys, timestamps, null).iterator());
+ trackedSegments.add(segment1);
+
+ int[] consumingPrimaryKeys = new int[]{10, 20, 30};
+ int[] consumingTimestamps = new int[]{150, 250, 350};
+ ThreadSafeMutableRoaringBitmap validDocIds2 = new ThreadSafeMutableRoaringBitmap();
+ MutableSegment mutableSegment = mockMutableSegmentWithDataSource(2, validDocIds2, null, consumingPrimaryKeys);
+ upsertMetadataManager.addRecord(mutableSegment, new RecordInfo(makePrimaryKey(10), 0, 150, false));
+ upsertMetadataManager.addRecord(mutableSegment, new RecordInfo(makePrimaryKey(20), 1, 250, false));
+ upsertMetadataManager.addRecord(mutableSegment, new RecordInfo(makePrimaryKey(30), 2, 350, false));
+ trackedSegments.add(mutableSegment);
+
+ assertFalse(upsertMetadataManager._previousKeyToRecordLocationMap.isEmpty(),
+ "Prev key map should be populated after consuming segment adds records over immutable");
+
+ // Commit: replace consuming with immutable (no PROTECTED mode)
+ int[] newTimestamps = new int[]{150, 250, 350};
+ ThreadSafeMutableRoaringBitmap newValidDocIds = new ThreadSafeMutableRoaringBitmap();
+ newValidDocIds.add(0);
+ newValidDocIds.add(1);
+ newValidDocIds.add(2);
+ List<PrimaryKey> newPrimaryKeyList = getPrimaryKeyList(3, consumingPrimaryKeys);
+ ImmutableSegmentImpl newSegment =
+ mockImmutableSegmentWithTimestamps(2, newValidDocIds, null, newPrimaryKeyList, newTimestamps);
+ doReturn(mutableSegment.getSegmentName()).when(newSegment).getSegmentName();
+ upsertMetadataManager.replaceSegment(newSegment, newValidDocIds, null,
+ getRecordInfoListWithIntegerComparison(3, consumingPrimaryKeys, newTimestamps, null).iterator(),
+ mutableSegment);
+ trackedSegments.remove(mutableSegment);
+ trackedSegments.add(newSegment);
+
+ assertTrue(upsertMetadataManager._previousKeyToRecordLocationMap.isEmpty(),
+ "Prev key map must be cleared after segment commit irrespective of mode");
+
+ // Also verify prev key map is cleared after removeSegment: set up again and remove consuming segment
+ int[] primaryKeysB = new int[]{40, 50, 60};
+ int[] timestampsB = new int[]{400, 500, 600};
+ ThreadSafeMutableRoaringBitmap validDocIds1B = new ThreadSafeMutableRoaringBitmap();
+ List<PrimaryKey> primaryKeysList1B = getPrimaryKeyList(3, primaryKeysB);
+ ImmutableSegmentImpl segment1B = mockImmutableSegmentWithTimestamps(3, validDocIds1B, null, primaryKeysList1B,
+ timestampsB);
+ upsertMetadataManager.addSegment(segment1B, validDocIds1B, null,
+ getRecordInfoListWithIntegerComparison(3, primaryKeysB, timestampsB, null).iterator());
+ trackedSegments.add(segment1B);
+
+ int[] consumingPrimaryKeysB = new int[]{40, 50, 60};
+ ThreadSafeMutableRoaringBitmap validDocIds2B = new ThreadSafeMutableRoaringBitmap();
+ validDocIds2B.add(0);
+ validDocIds2B.add(1);
+ validDocIds2B.add(2);
+ MutableSegment mutableSegmentB = mockMutableSegmentWithDataSource(4, validDocIds2B, null, consumingPrimaryKeysB);
+ upsertMetadataManager.addRecord(mutableSegmentB, new RecordInfo(makePrimaryKey(40), 0, 450, false));
+ upsertMetadataManager.addRecord(mutableSegmentB, new RecordInfo(makePrimaryKey(50), 1, 550, false));
+ upsertMetadataManager.addRecord(mutableSegmentB, new RecordInfo(makePrimaryKey(60), 2, 650, false));
+ trackedSegments.add(mutableSegmentB);
+
+ assertFalse(upsertMetadataManager._previousKeyToRecordLocationMap.isEmpty(),
+ "Prev key map should be populated after second consuming segment adds records over immutable");
+
+ upsertMetadataManager.removeSegment(mutableSegmentB);
+ trackedSegments.remove(mutableSegmentB);
+
+ assertTrue(upsertMetadataManager._previousKeyToRecordLocationMap.isEmpty(),
+ "Prev key map must be cleared after segment remove irrespective of mode");
upsertMetadataManager.stop();
upsertMetadataManager.close();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org