klsince commented on code in PR #13636:
URL: https://github.com/apache/pinot/pull/13636#discussion_r1716301593
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BaseTableDedupMetadataManager.java:
##########
@@ -19,39 +19,60 @@
package org.apache.pinot.segment.local.dedup;
import com.google.common.base.Preconditions;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.pinot.common.metrics.ServerMetrics;
import org.apache.pinot.segment.local.data.manager.TableDataManager;
import org.apache.pinot.spi.config.table.DedupConfig;
-import org.apache.pinot.spi.config.table.HashFunction;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.spi.data.Schema;
-abstract class BaseTableDedupMetadataManager implements
TableDedupMetadataManager {
+public abstract class BaseTableDedupMetadataManager implements
TableDedupMetadataManager {
protected final Map<Integer, PartitionDedupMetadataManager>
_partitionMetadataManagerMap = new ConcurrentHashMap<>();
protected String _tableNameWithType;
- protected List<String> _primaryKeyColumns;
- protected ServerMetrics _serverMetrics;
- protected HashFunction _hashFunction;
+ protected DedupContext _dedupContext;
@Override
public void init(TableConfig tableConfig, Schema schema, TableDataManager
tableDataManager,
ServerMetrics serverMetrics) {
_tableNameWithType = tableConfig.getTableName();
- _primaryKeyColumns = schema.getPrimaryKeyColumns();
- Preconditions.checkArgument(!CollectionUtils.isEmpty(_primaryKeyColumns),
+ List<String> primaryKeyColumns = schema.getPrimaryKeyColumns();
+ Preconditions.checkArgument(!CollectionUtils.isEmpty(primaryKeyColumns),
"Primary key columns must be configured for dedup enabled table: %s",
_tableNameWithType);
- _serverMetrics = serverMetrics;
-
DedupConfig dedupConfig = tableConfig.getDedupConfig();
Preconditions.checkArgument(dedupConfig != null, "Dedup must be enabled
for table: %s", _tableNameWithType);
- _hashFunction = dedupConfig.getHashFunction();
+ double metadataTTL = dedupConfig.getMetadataTTL();
+ String dedupTimeColumn = dedupConfig.getDedupTimeColumn();
+ if (metadataTTL > 0) {
+ dedupTimeColumn = dedupConfig.getDedupTimeColumn();
Review Comment:
can remove this stmt, as you already did it in L51
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentMapPartitionDedupMetadataManager.java:
##########
@@ -19,105 +19,80 @@
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.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 =
Review Comment:
I assume we can convert user's times to timestamp as in millis?
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/dedup/BasePartitionDedupMetadataManager.java:
##########
@@ -0,0 +1,212 @@
+/**
+ * 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 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) {
+ if (!startOperation()) {
+ _logger.info("Skip adding segment: {} because metadata manager is
already stopped", segment.getSegmentName());
+ return;
+ }
+ try (DedupUtils.DedupRecordInfoReader dedupRecordInfoReader = new
DedupUtils.DedupRecordInfoReader(segment,
+ _primaryKeyColumns, _dedupTimeColumn)) {
+ Iterator<DedupRecordInfo> dedupRecordInfoIterator =
+ DedupUtils.getDedupRecordInfoIterator(dedupRecordInfoReader,
segment.getSegmentMetadata().getTotalDocs());
+ doAddSegment(segment, dedupRecordInfoIterator);
+ removeExpiredPrimaryKeys();
Review Comment:
in upsert manager, removeExpiredPrimaryKeys() is only called when starting a
new consuming segment, so do we need to call this method when add/remove
segment? that might slow down adding/removing segment, while on helix threads.
--
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]
