Jackie-Jiang commented on a change in pull request #6899:
URL: https://github.com/apache/incubator-pinot/pull/6899#discussion_r651107004
##########
File path:
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartitionUpsertMetadataManager.java
##########
@@ -60,128 +64,175 @@
private final String _tableNameWithType;
private final int _partitionId;
private final ServerMetrics _serverMetrics;
+ private final PartialUpsertHandler _partialUpsertHandler;
- public PartitionUpsertMetadataManager(String tableNameWithType, int
partitionId, ServerMetrics serverMetrics) {
+ // TODO(upset): consider an off-heap KV store to persist this index to
improve the recovery speed.
+ @VisibleForTesting
+ final ConcurrentHashMap<PrimaryKey, RecordLocation>
_primaryKeyToRecordLocationMap = new ConcurrentHashMap<>();
+
+ // Reused for reading previous record during partial upsert
+ private final GenericRow _reuse = new GenericRow();
+ // Stores the result of updateRecord()
+ private GenericRow _result;
+
+ public PartitionUpsertMetadataManager(String tableNameWithType, int
partitionId, ServerMetrics serverMetrics,
+ @Nullable PartialUpsertHandler partialUpsertHandler) {
_tableNameWithType = tableNameWithType;
_partitionId = partitionId;
_serverMetrics = serverMetrics;
+ _partialUpsertHandler = partialUpsertHandler;
}
- public ConcurrentHashMap<PrimaryKey, RecordLocation>
getPrimaryKeyToRecordLocationMap() {
- return _primaryKeyToRecordLocationMap;
+ /**
+ * Initializes the upsert metadata for the given immutable segment.
+ */
+ public void addSegment(ImmutableSegmentImpl immutableSegment,
Iterator<RecordInfo> recordInfoIterator) {
+ addSegment(immutableSegment, recordInfoIterator, new
ThreadSafeMutableRoaringBitmap());
}
- // TODO(upset): consider an off-heap KV store to persist this index to
improve the recovery speed.
@VisibleForTesting
- final ConcurrentHashMap<PrimaryKey, RecordLocation>
_primaryKeyToRecordLocationMap = new ConcurrentHashMap<>();
-
- /**
- * Initializes the upsert metadata for the given immutable segment, returns
the valid doc ids for the segment.
- */
- public ThreadSafeMutableRoaringBitmap addSegment(String segmentName,
Iterator<RecordInfo> recordInfoIterator) {
+ void addSegment(ImmutableSegmentImpl immutableSegment, Iterator<RecordInfo>
recordInfoIterator,
+ ThreadSafeMutableRoaringBitmap validDocIds) {
+ String segmentName = immutableSegment.getSegmentName();
LOGGER.info("Adding upsert metadata for segment: {}", segmentName);
+ immutableSegment.enableUpsert(this, validDocIds);
- ThreadSafeMutableRoaringBitmap validDocIds = new
ThreadSafeMutableRoaringBitmap();
while (recordInfoIterator.hasNext()) {
RecordInfo recordInfo = recordInfoIterator.next();
_primaryKeyToRecordLocationMap.compute(recordInfo._primaryKey,
(primaryKey, currentRecordLocation) -> {
if (currentRecordLocation != null) {
// Existing primary key
- if (segmentName.equals(currentRecordLocation.getSegmentName())) {
- // The current record location has the same segment name
-
- // Update the record location when the new timestamp is greater
than or equal to the current timestamp.
- // There are 2 scenarios:
- // 1. The current record location is pointing to the same
segment (the segment being added). In this case,
- // we want to update the record location when there is a tie
to keep the newer record. Note that the
- // record info iterator will return records with incremental
doc ids.
- // 2. The current record location is pointing to the old segment
being replaced. This could happen when
- // committing a consuming segment, or reloading a completed
segment. In this case, we want to update
- // the record location when there is a tie because the record
locations should point to the new added
- // segment instead of the old segment being replaced. Also,
do not update the valid doc ids for the old
- // segment because it has not been replaced yet.
+ // The current record is in the same segment
+ // Update the record location when there is a tie to keep the newer
record. Note that the record info iterator
+ // will return records with incremental doc ids.
+ IndexSegment currentSegment = currentRecordLocation.getSegment();
+ if (immutableSegment == currentSegment) {
if (recordInfo._timestamp >= currentRecordLocation.getTimestamp())
{
- // Only update the valid doc ids for the new segment
- if (validDocIds == currentRecordLocation.getValidDocIds()) {
- validDocIds.remove(currentRecordLocation.getDocId());
- }
+ validDocIds.remove(currentRecordLocation.getDocId());
validDocIds.add(recordInfo._docId);
- return new RecordLocation(segmentName, recordInfo._docId,
recordInfo._timestamp, validDocIds);
+ return new RecordLocation(immutableSegment, recordInfo._docId,
recordInfo._timestamp);
} else {
return currentRecordLocation;
}
- } else {
- // The current record location is pointing to a different segment
-
- // Update the record location when getting a newer timestamp, or
the timestamp is the same as the current
- // timestamp, but the segment has a larger sequence number (the
segment is newer than the current segment).
- if (recordInfo._timestamp > currentRecordLocation.getTimestamp()
|| (
- recordInfo._timestamp == currentRecordLocation.getTimestamp()
- &&
LLCSegmentName.isLowLevelConsumerSegmentName(segmentName)
- &&
LLCSegmentName.isLowLevelConsumerSegmentName(currentRecordLocation.getSegmentName())
- && LLCSegmentName.getSequenceNumber(segmentName) >
LLCSegmentName
-
.getSequenceNumber(currentRecordLocation.getSegmentName()))) {
-
currentRecordLocation.getValidDocIds().remove(currentRecordLocation.getDocId());
+ }
+
+ // The current record is in an old segment being replaced
+ // This could happen when committing a consuming segment, or
reloading a completed segment. In this case, we
+ // want to update the record location when there is a tie because
the record locations should point to the new
+ // added segment instead of the old segment being replaced. Also, do
not update the valid doc ids for the old
+ // segment because it has not been replaced yet.
+ String currentSegmentName = currentSegment.getSegmentName();
+ if (segmentName.equals(currentSegmentName)) {
+ if (recordInfo._timestamp >= currentRecordLocation.getTimestamp())
{
validDocIds.add(recordInfo._docId);
- return new RecordLocation(segmentName, recordInfo._docId,
recordInfo._timestamp, validDocIds);
+ return new RecordLocation(immutableSegment, recordInfo._docId,
recordInfo._timestamp);
} else {
return currentRecordLocation;
}
}
+
+ // The current record is in a different segment
+ // Update the record location when getting a newer timestamp, or the
timestamp is the same as the current
+ // timestamp, but the segment has a larger sequence number (the
segment is newer than the current segment).
+ if (recordInfo._timestamp > currentRecordLocation.getTimestamp() || (
+ recordInfo._timestamp == currentRecordLocation.getTimestamp() &&
LLCSegmentName
+ .isLowLevelConsumerSegmentName(segmentName) && LLCSegmentName
+ .isLowLevelConsumerSegmentName(currentSegmentName)
+ && LLCSegmentName.getSequenceNumber(segmentName) >
LLCSegmentName
+ .getSequenceNumber(currentSegmentName))) {
+ assert currentSegment.getValidDocIds() != null;
+
currentSegment.getValidDocIds().remove(currentRecordLocation.getDocId());
+ validDocIds.add(recordInfo._docId);
+ return new RecordLocation(immutableSegment, recordInfo._docId,
recordInfo._timestamp);
+ } else {
+ return currentRecordLocation;
+ }
} else {
// New primary key
validDocIds.add(recordInfo._docId);
- return new RecordLocation(segmentName, recordInfo._docId,
recordInfo._timestamp, validDocIds);
+ return new RecordLocation(immutableSegment, recordInfo._docId,
recordInfo._timestamp);
}
});
}
// Update metrics
_serverMetrics.setValueOfPartitionGauge(_tableNameWithType, _partitionId,
ServerGauge.UPSERT_PRIMARY_KEYS_COUNT,
_primaryKeyToRecordLocationMap.size());
- return validDocIds;
}
/**
- * Updates the upsert metadata for a new consumed record in the given
consuming segment.
+ * Updates the upsert metadata for a new consumed record in the given
consuming segment. Returns the merged record if
+ * partial-upsert is enabled.
*/
- public void updateRecord(String segmentName, RecordInfo recordInfo,
ThreadSafeMutableRoaringBitmap validDocIds) {
+ public GenericRow updateRecord(IndexSegment segment, RecordInfo recordInfo,
GenericRow record) {
+ // For partial-upsert, need to ensure all previous records are loaded
before inserting new records.
+ if (_partialUpsertHandler != null) {
+ while (!_partialUpsertHandler.isAllSegmentsLoaded()) {
+ LOGGER
+ .info("Sleeping 1 second waiting for all segments loaded for
partial-upsert table: {}", _tableNameWithType);
+ try {
+ //noinspection BusyWait
+ Thread.sleep(1000L);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ _result = record;
_primaryKeyToRecordLocationMap.compute(recordInfo._primaryKey,
(primaryKey, currentRecordLocation) -> {
if (currentRecordLocation != null) {
// Existing primary key
// Update the record location when the new timestamp is greater than
or equal to the current timestamp. Update
// the record location when there is a tie to keep the newer record.
if (recordInfo._timestamp >= currentRecordLocation.getTimestamp()) {
-
currentRecordLocation.getValidDocIds().remove(currentRecordLocation.getDocId());
- validDocIds.add(recordInfo._docId);
- return new RecordLocation(segmentName, recordInfo._docId,
recordInfo._timestamp, validDocIds);
+ IndexSegment currentSegment = currentRecordLocation.getSegment();
+ if (_partialUpsertHandler != null) {
+ // Partial upsert
+ GenericRow previousRecord =
currentSegment.getRecord(currentRecordLocation.getDocId(), _reuse);
Review comment:
The entry will be removed before destroying the segment
(`removeSegment()` will be called)
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
