klsince commented on code in PR #13636:
URL: https://github.com/apache/pinot/pull/13636#discussion_r1719008024
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -554,7 +564,12 @@ private void buildDedupMeta(ImmutableSegmentImpl
immutableSegment) {
PartitionDedupMetadataManager partitionDedupMetadataManager =
_tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId);
immutableSegment.enableDedup(partitionDedupMetadataManager);
- partitionDedupMetadataManager.addSegment(immutableSegment);
+ SegmentDataManager oldSegmentManager =
_segmentDataManagerMap.get(segmentName);
Review Comment:
looks like you fixed `// TODO: Change dedup handling to handle segment
replacement` commented above L539
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java:
##########
@@ -19,106 +19,124 @@
package org.apache.pinot.segment.local.dedup;
import com.google.common.annotations.VisibleForTesting;
-import java.util.HashMap;
+import com.google.common.util.concurrent.AtomicDouble;
+import java.io.IOException;
import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.metrics.ServerGauge;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.HashUtils;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.spi.config.table.HashFunction;
-import org.apache.pinot.spi.data.readers.PrimaryKey;
-import org.apache.pinot.spi.utils.ByteArray;
-class ConcurrentMapPartitionDedupMetadataManager implements
PartitionDedupMetadataManager {
- private final String _tableNameWithType;
- private final List<String> _primaryKeyColumns;
- private final int _partitionId;
- private final ServerMetrics _serverMetrics;
- private final HashFunction _hashFunction;
+class ConcurrentMapPartitionDedupMetadataManager extends
BasePartitionDedupMetadataManager {
@VisibleForTesting
- final ConcurrentHashMap<Object, IndexSegment> _primaryKeyToSegmentMap = new
ConcurrentHashMap<>();
+ final AtomicDouble _largestSeenTime = new AtomicDouble(0);
+ @VisibleForTesting
+ final ConcurrentHashMap<Object, Pair<IndexSegment, Double>>
_primaryKeyToSegmentAndTimeMap =
+ new ConcurrentHashMap<>();
- public ConcurrentMapPartitionDedupMetadataManager(String tableNameWithType,
List<String> primaryKeyColumns,
- int partitionId, ServerMetrics serverMetrics, HashFunction hashFunction)
{
- _tableNameWithType = tableNameWithType;
- _primaryKeyColumns = primaryKeyColumns;
- _partitionId = partitionId;
- _serverMetrics = serverMetrics;
- _hashFunction = hashFunction;
+ protected ConcurrentMapPartitionDedupMetadataManager(String
tableNameWithType, int partitionId,
+ DedupContext dedupContext) {
+ super(tableNameWithType, partitionId, dedupContext);
}
- public void addSegment(IndexSegment segment) {
- // Add all PKs to _primaryKeyToSegmentMap
- Iterator<PrimaryKey> primaryKeyIterator = getPrimaryKeyIterator(segment);
- while (primaryKeyIterator.hasNext()) {
- PrimaryKey pk = primaryKeyIterator.next();
- _primaryKeyToSegmentMap.put(HashUtils.hashPrimaryKey(pk, _hashFunction),
segment);
+ @Override
+ protected void doAddOrReplaceSegment(IndexSegment oldSegment, IndexSegment
newSegment,
+ Iterator<DedupRecordInfo> dedupRecordInfoIteratorOfNewSegment) {
+ String segmentName = newSegment.getSegmentName();
+ while (dedupRecordInfoIteratorOfNewSegment.hasNext()) {
+ DedupRecordInfo dedupRecordInfo =
dedupRecordInfoIteratorOfNewSegment.next();
+ double dedupTime = dedupRecordInfo.getDedupTime();
+ _largestSeenTime.getAndUpdate(time -> Math.max(time, dedupTime));
+
_primaryKeyToSegmentAndTimeMap.compute(HashUtils.hashPrimaryKey(dedupRecordInfo.getPrimaryKey(),
_hashFunction),
+ (primaryKey, segmentAndTime) -> {
+ if (segmentAndTime == null) {
+ return Pair.of(newSegment, dedupTime);
+ } else {
+ // when oldSegment is null, it means we are adding a new segment
+ // when oldSegment is not null, it means we are replacing an
existing segment
+ if (oldSegment == null) {
+ _logger.warn("When adding a new segment: dedup record in
segment: {} with primary key: {} and dedup "
Review Comment:
nit: "... new segment: record ..." (removing 'dedup')
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java:
##########
@@ -19,106 +19,124 @@
package org.apache.pinot.segment.local.dedup;
import com.google.common.annotations.VisibleForTesting;
-import java.util.HashMap;
+import com.google.common.util.concurrent.AtomicDouble;
+import java.io.IOException;
import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.metrics.ServerGauge;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.HashUtils;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.spi.config.table.HashFunction;
-import org.apache.pinot.spi.data.readers.PrimaryKey;
-import org.apache.pinot.spi.utils.ByteArray;
-class ConcurrentMapPartitionDedupMetadataManager implements
PartitionDedupMetadataManager {
- private final String _tableNameWithType;
- private final List<String> _primaryKeyColumns;
- private final int _partitionId;
- private final ServerMetrics _serverMetrics;
- private final HashFunction _hashFunction;
+class ConcurrentMapPartitionDedupMetadataManager extends
BasePartitionDedupMetadataManager {
@VisibleForTesting
- final ConcurrentHashMap<Object, IndexSegment> _primaryKeyToSegmentMap = new
ConcurrentHashMap<>();
+ final AtomicDouble _largestSeenTime = new AtomicDouble(0);
+ @VisibleForTesting
+ final ConcurrentHashMap<Object, Pair<IndexSegment, Double>>
_primaryKeyToSegmentAndTimeMap =
+ new ConcurrentHashMap<>();
- public ConcurrentMapPartitionDedupMetadataManager(String tableNameWithType,
List<String> primaryKeyColumns,
- int partitionId, ServerMetrics serverMetrics, HashFunction hashFunction)
{
- _tableNameWithType = tableNameWithType;
- _primaryKeyColumns = primaryKeyColumns;
- _partitionId = partitionId;
- _serverMetrics = serverMetrics;
- _hashFunction = hashFunction;
+ protected ConcurrentMapPartitionDedupMetadataManager(String
tableNameWithType, int partitionId,
+ DedupContext dedupContext) {
+ super(tableNameWithType, partitionId, dedupContext);
}
- public void addSegment(IndexSegment segment) {
- // Add all PKs to _primaryKeyToSegmentMap
- Iterator<PrimaryKey> primaryKeyIterator = getPrimaryKeyIterator(segment);
- while (primaryKeyIterator.hasNext()) {
- PrimaryKey pk = primaryKeyIterator.next();
- _primaryKeyToSegmentMap.put(HashUtils.hashPrimaryKey(pk, _hashFunction),
segment);
+ @Override
+ protected void doAddOrReplaceSegment(IndexSegment oldSegment, IndexSegment
newSegment,
+ Iterator<DedupRecordInfo> dedupRecordInfoIteratorOfNewSegment) {
+ String segmentName = newSegment.getSegmentName();
+ while (dedupRecordInfoIteratorOfNewSegment.hasNext()) {
+ DedupRecordInfo dedupRecordInfo =
dedupRecordInfoIteratorOfNewSegment.next();
+ double dedupTime = dedupRecordInfo.getDedupTime();
+ _largestSeenTime.getAndUpdate(time -> Math.max(time, dedupTime));
+
_primaryKeyToSegmentAndTimeMap.compute(HashUtils.hashPrimaryKey(dedupRecordInfo.getPrimaryKey(),
_hashFunction),
+ (primaryKey, segmentAndTime) -> {
+ if (segmentAndTime == null) {
+ return Pair.of(newSegment, dedupTime);
+ } else {
Review Comment:
no need for `else`, as you have 'return' in the if-branch.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -554,7 +564,12 @@ private void buildDedupMeta(ImmutableSegmentImpl
immutableSegment) {
PartitionDedupMetadataManager partitionDedupMetadataManager =
_tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId);
immutableSegment.enableDedup(partitionDedupMetadataManager);
- partitionDedupMetadataManager.addSegment(immutableSegment);
+ SegmentDataManager oldSegmentManager =
_segmentDataManagerMap.get(segmentName);
Review Comment:
nit: perhaps rename buildDedupMeta to handleDedup() for a bit naming
consistency with handleUpsert().
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java:
##########
@@ -0,0 +1,241 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.dedup;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.metrics.ServerGauge;
+import org.apache.pinot.common.metrics.ServerMetrics;
+import org.apache.pinot.common.metrics.ServerTimer;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.spi.config.table.HashFunction;
+import org.apache.pinot.spi.data.readers.PrimaryKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public abstract class BasePartitionDedupMetadataManager implements
PartitionDedupMetadataManager {
+ protected final String _tableNameWithType;
+ protected final List<String> _primaryKeyColumns;
+ protected final int _partitionId;
+ protected final ServerMetrics _serverMetrics;
+ protected final HashFunction _hashFunction;
+ protected final double _metadataTTL;
+ protected final String _dedupTimeColumn;
+ protected final Logger _logger;
+
+ // The following variables are always accessed within synchronized block
+ private boolean _stopped;
+ // Initialize with 1 pending operation to indicate the metadata manager can
take more operations
+ private int _numPendingOperations = 1;
+ private boolean _closed;
+
+ protected BasePartitionDedupMetadataManager(String tableNameWithType, int
partitionId, DedupContext dedupContext) {
+ _tableNameWithType = tableNameWithType;
+ _partitionId = partitionId;
+ _primaryKeyColumns = dedupContext.getPrimaryKeyColumns();
+ _hashFunction = dedupContext.getHashFunction();
+ _serverMetrics = dedupContext.getServerMetrics();
+ _metadataTTL = dedupContext.getMetadataTTL() >= 0 ?
dedupContext.getMetadataTTL() : 0;
+ _dedupTimeColumn = dedupContext.getDedupTimeColumn();
+ if (_metadataTTL > 0) {
+ Preconditions.checkArgument(_dedupTimeColumn != null,
+ "When metadataTTL is configured, metadata time column must be
configured for dedup enabled table: %s",
+ tableNameWithType);
+ }
+ _logger = LoggerFactory.getLogger(tableNameWithType + "-" + partitionId +
"-" + getClass().getSimpleName());
+ }
+
+ @Override
+ public boolean checkRecordPresentOrUpdate(PrimaryKey pk, IndexSegment
indexSegment) {
+ throw new UnsupportedOperationException(
+ "checkRecordPresentOrUpdate(PrimaryKey pk, IndexSegment indexSegment)
is " + "deprecated!");
+ }
+
+ @Override
+ public void addSegment(IndexSegment segment) {
+ addOrReplaceSegment(null, segment);
+ }
+
+ @Override
+ public void replaceSegment(IndexSegment oldSegment, IndexSegment newSegment)
{
+ addOrReplaceSegment(oldSegment, newSegment);
+ }
+
+ private void addOrReplaceSegment(@Nullable IndexSegment oldSegment,
IndexSegment newSegment) {
+ if (!startOperation()) {
Review Comment:
nit: it may be a bit more readable to move the if-check and the
try-catch-finally code into addSegment/replaceSegment so that we don't need to
check oldSegment==null, like the add/replace methods in upsert manager.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -554,7 +564,12 @@ private void buildDedupMeta(ImmutableSegmentImpl
immutableSegment) {
PartitionDedupMetadataManager partitionDedupMetadataManager =
_tableDedupMetadataManager.getOrCreatePartitionManager(partitionGroupId);
immutableSegment.enableDedup(partitionDedupMetadataManager);
- partitionDedupMetadataManager.addSegment(immutableSegment);
+ SegmentDataManager oldSegmentManager =
_segmentDataManagerMap.get(segmentName);
+ if (oldSegmentManager != null) {
+
partitionDedupMetadataManager.replaceSegment(oldSegmentManager.getSegment(),
immutableSegment);
+ } else {
+ partitionDedupMetadataManager.addSegment(immutableSegment);
Review Comment:
not sure if you have logs inside, we can log msg here as in handleUpsert()
when adding or replacing segments.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java:
##########
@@ -19,106 +19,124 @@
package org.apache.pinot.segment.local.dedup;
import com.google.common.annotations.VisibleForTesting;
-import java.util.HashMap;
+import com.google.common.util.concurrent.AtomicDouble;
+import java.io.IOException;
import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.common.metrics.ServerGauge;
-import org.apache.pinot.common.metrics.ServerMetrics;
-import org.apache.pinot.segment.local.segment.readers.PinotSegmentColumnReader;
import org.apache.pinot.segment.local.utils.HashUtils;
import org.apache.pinot.segment.spi.IndexSegment;
-import org.apache.pinot.spi.config.table.HashFunction;
-import org.apache.pinot.spi.data.readers.PrimaryKey;
-import org.apache.pinot.spi.utils.ByteArray;
-class ConcurrentMapPartitionDedupMetadataManager implements
PartitionDedupMetadataManager {
- private final String _tableNameWithType;
- private final List<String> _primaryKeyColumns;
- private final int _partitionId;
- private final ServerMetrics _serverMetrics;
- private final HashFunction _hashFunction;
+class ConcurrentMapPartitionDedupMetadataManager extends
BasePartitionDedupMetadataManager {
@VisibleForTesting
- final ConcurrentHashMap<Object, IndexSegment> _primaryKeyToSegmentMap = new
ConcurrentHashMap<>();
+ final AtomicDouble _largestSeenTime = new AtomicDouble(0);
+ @VisibleForTesting
+ final ConcurrentHashMap<Object, Pair<IndexSegment, Double>>
_primaryKeyToSegmentAndTimeMap =
+ new ConcurrentHashMap<>();
- public ConcurrentMapPartitionDedupMetadataManager(String tableNameWithType,
List<String> primaryKeyColumns,
- int partitionId, ServerMetrics serverMetrics, HashFunction hashFunction)
{
- _tableNameWithType = tableNameWithType;
- _primaryKeyColumns = primaryKeyColumns;
- _partitionId = partitionId;
- _serverMetrics = serverMetrics;
- _hashFunction = hashFunction;
+ protected ConcurrentMapPartitionDedupMetadataManager(String
tableNameWithType, int partitionId,
+ DedupContext dedupContext) {
+ super(tableNameWithType, partitionId, dedupContext);
}
- public void addSegment(IndexSegment segment) {
- // Add all PKs to _primaryKeyToSegmentMap
- Iterator<PrimaryKey> primaryKeyIterator = getPrimaryKeyIterator(segment);
- while (primaryKeyIterator.hasNext()) {
- PrimaryKey pk = primaryKeyIterator.next();
- _primaryKeyToSegmentMap.put(HashUtils.hashPrimaryKey(pk, _hashFunction),
segment);
+ @Override
+ protected void doAddOrReplaceSegment(IndexSegment oldSegment, IndexSegment
newSegment,
+ Iterator<DedupRecordInfo> dedupRecordInfoIteratorOfNewSegment) {
+ String segmentName = newSegment.getSegmentName();
+ while (dedupRecordInfoIteratorOfNewSegment.hasNext()) {
+ DedupRecordInfo dedupRecordInfo =
dedupRecordInfoIteratorOfNewSegment.next();
+ double dedupTime = dedupRecordInfo.getDedupTime();
+ _largestSeenTime.getAndUpdate(time -> Math.max(time, dedupTime));
+
_primaryKeyToSegmentAndTimeMap.compute(HashUtils.hashPrimaryKey(dedupRecordInfo.getPrimaryKey(),
_hashFunction),
+ (primaryKey, segmentAndTime) -> {
+ if (segmentAndTime == null) {
+ return Pair.of(newSegment, dedupTime);
+ } else {
+ // when oldSegment is null, it means we are adding a new segment
+ // when oldSegment is not null, it means we are replacing an
existing segment
+ if (oldSegment == null) {
+ _logger.warn("When adding a new segment: dedup record in
segment: {} with primary key: {} and dedup "
+ + "time: {} already exists in segment: {} with dedup
time: {}", segmentName,
+ dedupRecordInfo.getPrimaryKey(), dedupTime,
segmentAndTime.getLeft().getSegmentName(),
+ segmentAndTime.getRight());
+ } else {
+ if (segmentAndTime.getLeft() != oldSegment) {
+ _logger.warn("When replacing a segment: dedup record in
segment: {} with primary key: {} and dedup "
Review Comment:
would this happen?
as the get-check-update logic is done within ConcurrentMap.compute() and
it's atomic, no?
--
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]
