Jackie-Jiang commented on code in PR #17503:
URL: https://github.com/apache/pinot/pull/17503#discussion_r2696417137
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java:
##########
@@ -120,6 +118,10 @@ protected void doAddOrReplaceSegment(ImmutableSegmentImpl
segment, ThreadSafeMut
if (validDocIdsForOldSegment != null) {
validDocIdsForOldSegment.remove(currentDocId);
}
+ if
(!_previousKeyToRecordLocationMap.containsKey(primaryKey)
Review Comment:
When we replace a consuming segment, we should remove the key from the map
right?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java:
##########
@@ -746,12 +747,11 @@ void
revertSegmentUpsertMetadataWithRetry(ImmutableSegment segment, ThreadSafeMu
}
}
- protected abstract void removeNewlyAddedKeys(IndexSegment oldSegment);
-
protected abstract void eraseKeyToPreviousLocationMap();
Review Comment:
I don't think we need this method. The map should be cleared during the
process of `revertCurrentSegmentUpsertMetadata()`. Explicitly clear it can lead
to bugs where certain keys are not handled
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java:
##########
@@ -270,20 +274,6 @@ public void replaceSegment(ImmutableSegment segment,
@Nullable ThreadSafeMutable
}
}
- @Override
- protected void removeNewlyAddedKeys(IndexSegment oldSegment) {
- int removedKeys = 0;
- for (Map.Entry<Object, RecordLocation> entry : _newlyAddedKeys.entrySet())
{
- if (entry.getValue().getSegment() == oldSegment) {
- _primaryKeyToRecordLocationMap.remove(entry.getKey());
- removeDocId(oldSegment, entry.getValue().getDocId());
- removedKeys++;
- }
- }
- _logger.info("Removed newly added {} keys for the segment: {} out of :
{}", removedKeys,
- oldSegment.getSegmentName(), _previousKeyToRecordLocationMap.size());
- }
-
@Override
protected void removeSegment(IndexSegment segment, Iterator<PrimaryKey>
primaryKeyIterator) {
Review Comment:
When removing a consuming segment (e.g. a consuming segment is reset), we
should also revert the keys
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManager.java:
##########
@@ -94,10 +95,7 @@ protected void doAddOrReplaceSegment(ImmutableSegmentImpl
segment, ThreadSafeMut
if (comparisonResult >= 0) {
replaceDocId(segment, validDocIds, queryableDocIds,
currentDocId, newDocId, recordInfo);
RecordLocation newRecordLocation = new
RecordLocation(segment, newDocId, newComparisonValue);
- // Track the record location of the newly added keys
- if (_newlyAddedKeys.containsKey(primaryKey)) {
- _newlyAddedKeys.put(primaryKey, newRecordLocation);
- }
+ _previousKeyToRecordLocationMap.put(primaryKey,
currentRecordLocation);
Review Comment:
I don't follow this. Why do we want to put this record?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java:
##########
@@ -193,7 +197,7 @@ currentSegmentName, getAuthoritativeCreationTime(segment),
// New primary key
addDocId(segment, validDocIds, queryableDocIds, newDocId,
recordInfo);
RecordLocation newRecordLocation = new RecordLocation(segment,
newDocId, newComparisonValue, 1);
- _newlyAddedKeys.put(primaryKey, newRecordLocation);
+ _previousKeyToRecordLocationMap.put(primaryKey,
newRecordLocation);
Review Comment:
Why do we add keys here?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/BasePartitionUpsertMetadataManager.java:
##########
@@ -746,12 +747,11 @@ void
revertSegmentUpsertMetadataWithRetry(ImmutableSegment segment, ThreadSafeMu
}
}
- protected abstract void removeNewlyAddedKeys(IndexSegment oldSegment);
-
protected abstract void eraseKeyToPreviousLocationMap();
protected abstract void revertCurrentSegmentUpsertMetadata(IndexSegment
oldSegment,
- ThreadSafeMutableRoaringBitmap validDocIds,
ThreadSafeMutableRoaringBitmap queryableDocIds);
+ ThreadSafeMutableRoaringBitmap validDocIds,
ThreadSafeMutableRoaringBitmap queryableDocIds,
+ ImmutableSegment segment, MutableRoaringBitmap validDocIdsForOldSegment);
Review Comment:
Please add javadoc about the purpose of these 2 methods
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/ConcurrentMapPartitionUpsertMetadataManagerForConsistentDeletes.java:
##########
@@ -183,8 +179,16 @@ protected void doAddOrReplaceSegment(ImmutableSegmentImpl
segment, ThreadSafeMut
currentSegmentName, getAuthoritativeCreationTime(segment),
getAuthoritativeCreationTime(currentSegment)))) {
replaceDocId(segment, validDocIds, queryableDocIds,
currentSegment, currentDocId, newDocId, recordInfo);
- return new RecordLocation(segment, newDocId,
newComparisonValue,
+ RecordLocation newRecordLocation = new RecordLocation(segment,
newDocId, newComparisonValue,
RecordLocation.incrementSegmentCount(currentDistinctSegmentCount));
+ // Track previous location for revert purposes:
+ // - If current key is in consuming segment: Don't track - key
is moving to immutable segment,
+ // consuming segment no longer owns it, so no revert needed
for this key
+ // - If current key is in immutable segment: Track current
location to enable revert if needed
+ if (!(currentSegment instanceof MutableSegment)) {
Review Comment:
When current segment is mutable, we should remove the key
--
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]
