[Feature] Deduplication #8708

saurabhd336 · 2022-05-16T11:26:24Z

This PR adds support for enabling deduplication for realtime table, via a top level table config. At a high level, primaryKey (as defined in the table schema) hashes are stored into in-memory data structures, and each incoming row is validated against it. Duplicate rows are dropped. The expectation while using this feature, is for the stream to be partitioned by the primary key, strictReplicaGroup routing to be enabled and the configured stream consumer type to be lowLevel. These requirements are therefore mandated via tableConfig API's input validations.

Design doc: https://docs.google.com/document/d/17sOSRQ1slff30z7jDc0ec5qKwv0xSfPkDjpMOY07POQ/edit?usp=sharing

How to use

https://docs.pinot.apache.org/basics/data-import/dedup

codecov-commenter · 2022-05-16T17:44:40Z

Codecov Report

Merging #8708 (38f4b55) into master (9abf15f) will decrease coverage by 55.68%.
The diff coverage is 0.00%.

@@              Coverage Diff              @@
##             master    #8708       +/-   ##
=============================================
- Coverage     69.81%   14.13%   -55.69%     
+ Complexity     4622      168     -4454     
=============================================
  Files          1735     1695       -40     
  Lines         91320    89448     -1872     
  Branches      13644    13440      -204     
=============================================
- Hits          63759    12642    -51117     
- Misses        23144    75868    +52724     
+ Partials       4417      938     -3479

Flag	Coverage Δ
integration1	`?`
integration2	`?`
unittests1	`?`
unittests2	`14.13% <0.00%> (-0.02%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
...a/org/apache/pinot/common/metrics/ServerGauge.java	`0.00% <0.00%> (-95.66%)`	⬇️
...a/org/apache/pinot/common/metrics/ServerMeter.java	`0.00% <0.00%> (-100.00%)`	⬇️
...he/pinot/common/utils/config/TableConfigUtils.java	`0.00% <0.00%> (-85.93%)`	⬇️
...manager/realtime/LLRealtimeSegmentDataManager.java	`0.00% <0.00%> (-71.12%)`	⬇️
...ata/manager/realtime/RealtimeTableDataManager.java	`0.00% <0.00%> (-67.76%)`	⬇️
...ent/local/dedup/PartitionDedupMetadataManager.java	`0.00% <0.00%> (ø)`
...segment/local/dedup/TableDedupMetadataManager.java	`0.00% <0.00%> (ø)`
...l/indexsegment/immutable/ImmutableSegmentImpl.java	`0.00% <0.00%> (-68.12%)`	⬇️
...local/indexsegment/mutable/MutableSegmentImpl.java	`0.00% <0.00%> (-59.29%)`	⬇️
...ent/local/realtime/impl/RealtimeSegmentConfig.java	`0.00% <0.00%> (-92.37%)`	⬇️
... and 1380 more

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 9abf15f...38f4b55. Read the comment docs.

Jackie-Jiang

Good job extracting several common properties from upsert and dedup

Jackie-Jiang · 2022-05-25T20:59:13Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableState.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TableState {


Suggest modeling it as a util class (TableStateUtils) and have one static method public static boolean isAllSegmentsLoaded(HelixManager helixManager, String tableNameWithType). The _allSegmentsLoaded can still be tracked within the metadata manager. We don't want this util class to track the loaded flag, instead it should always re-calculate the state.

_allSegmentsLoaded will need to present in both upsert and dedupe metadata classes separately. Here its with just one instance of this class. Is that okay?

The reason why I suggest modeling this class as a util and not tracking _allSegmentsLoaded within this class is because we may reuse this util method for other features, and we don't want to couple this "check once then always true" semantic into this util method/class

(minor) Let's rename it to TableStateUtils

Jackie-Jiang · 2022-05-25T21:37:48Z

.../src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java

@@ -538,6 +539,7 @@ private void processStreamEvents(MessageBatch messagesAndOffsets, long idlePipeS
        }
        for (GenericRow transformedRow : reusedResult.getTransformedRows()) {
          try {
+            // TODO(saurabh): we may have dropped the record due to dedup. Should we increment indexedMessageCount?


IMO it is okay to increase the value since we are just tracking the row count fed into the index(). We should use another metrics to track the rows ignored because of the dedup

(minor) Since we already track the dropped records, we can remove this TODO and consider changing it to a comment

Ack. I think with the metric added, this is no longer needed

Jackie-Jiang · 2022-05-25T21:38:29Z

.../src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java

@@ -1307,6 +1310,7 @@ public LLRealtimeSegmentDataManager(SegmentZKMetadata segmentZKMetadata, TableCo
    Set<String> fstIndexColumns = indexLoadingConfig.getFSTIndexColumns();
    _fstIndexColumns = new ArrayList<>(fstIndexColumns);

+    boolean dedupEnabled = (tableConfig.getDedupConfig() != null && tableConfig.getDedupConfig().isDedupEnabled());


This flag is redundant. It is implicit on the presence of partitionDedupMetadataManager

Jackie-Jiang · 2022-05-25T21:39:24Z

...core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java

@@ -111,7 +116,9 @@ public class RealtimeTableDataManager extends BaseTableDataManager {
  private static final int MIN_INTERVAL_BETWEEN_STATS_UPDATES_MINUTES = 30;

  private UpsertConfig.Mode _upsertMode;
+  private boolean _isDedupEnabled;


This flag is redundant, and is implicit on the presence of _tableDedupMetadataManager

Jackie-Jiang · 2022-05-25T21:42:34Z

...core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java

+    Integer partitionGroupId = SegmentUtils
+        .getRealtimeSegmentPartitionId(segmentName, _tableNameWithType, _helixManager, _primaryKeyColumns.get(0));
+    Preconditions.checkNotNull(partitionGroupId, String
+        .format("PartitionGroupId is not available for segment: '%s' (upsert-enabled table: %s)", segmentName,


Suggested change

.format("PartitionGroupId is not available for segment: '%s' (upsert-enabled table: %s)", segmentName,

.format("PartitionGroupId is not available for segment: '%s' (dedup-enabled table: %s)", segmentName,

Jackie-Jiang · 2022-05-25T22:23:01Z

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/TableConfig.java

@@ -288,8 +303,8 @@ public void setTierConfigsList(List<TierConfig> tierConfigsList) {
  }

  @JsonIgnore
-  public UpsertConfig.HashFunction getHashFunction() {
-    return _upsertConfig == null ? UpsertConfig.HashFunction.NONE : _upsertConfig.getHashFunction();
+  public HashFunction getHashFunction() {


We should remove this method. The hash function can come from both upsert config and dedup config

Jackie-Jiang · 2022-05-25T22:23:48Z

pinot-spi/src/main/java/org/apache/pinot/spi/config/table/DedupConfig.java

+  private final HashFunction _hashFunction;
+
+  @JsonCreator
+  public DedupConfig(@JsonProperty(value = "dedupEnabled", required = true) final boolean dedupEnabled,


(minor) We don't usually put final for local variables or parameters

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java

Jackie-Jiang · 2022-05-25T22:27:02Z

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java

+        "Upsert/Dedup table must use strict replica-group (i.e. strictReplicaGroup) based routing");
+
+    // specifically for upsert
+    if (tableConfig.getUpsertConfig() != null) {


Non-null config doesn't mean it is enabled

Suggested change

if (tableConfig.getUpsertConfig() != null) {

if (tableConfig.getUpsertMode() != UpsertConfig.Mode.NONE) {

Jackie-Jiang · 2022-05-25T22:28:46Z

pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java

-  static void validateUpsertConfig(TableConfig tableConfig, Schema schema) {
-    if (tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE) {
+  static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema) {
+    if (tableConfig.getUpsertMode() == UpsertConfig.Mode.NONE && tableConfig.getDedupConfig() == null) {


Non-null dedup config doesn't mean it is enabled. We either remove the dedupEnabled field and treat non-null dedup as dedup-enabled, or check the flag.

Idea here is, if the config json doesn't have dedupeConfig field, no need to run the validaiton

Understood. We should also skip the validation when DedupConfig is available, but dedup is not enabled

Jackie-Jiang

LGTM with some non-blocking comments. Good job!

Jackie-Jiang · 2022-06-01T23:44:57Z

.../src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java

@@ -538,6 +539,7 @@ private void processStreamEvents(MessageBatch messagesAndOffsets, long idlePipeS
        }
        for (GenericRow transformedRow : reusedResult.getTransformedRows()) {
          try {
+            // TODO(saurabh): we may have dropped the record due to dedup. Should we increment indexedMessageCount?


(minor) Since we already track the dropped records, we can remove this TODO and consider changing it to a comment

Jackie-Jiang · 2022-06-01T23:46:03Z

...-local/src/main/java/org/apache/pinot/segment/local/dedup/PartitionDedupMetadataManager.java

+  private final HashFunction _hashFunction;
+  private boolean _allSegmentsLoaded;
+
+  // TODO(saurabh) : We can replace this with a ocncurrent Set


(minor) Remove this TODO

Jackie-Jiang · 2022-06-01T23:46:44Z

...-local/src/main/java/org/apache/pinot/segment/local/dedup/PartitionDedupMetadataManager.java

+
+  // TODO(saurabh) : We can replace this with a ocncurrent Set
+  @VisibleForTesting
+  final ConcurrentHashMap<Object, IndexSegment> _primaryKeySet = new ConcurrentHashMap<>();


Suggested change

final ConcurrentHashMap<Object, IndexSegment> _primaryKeySet = new ConcurrentHashMap<>();

final ConcurrentHashMap<Object, IndexSegment> _primaryKeyToSegmentMap = new ConcurrentHashMap<>();

Jackie-Jiang · 2022-06-01T23:58:58Z

...-local/src/main/java/org/apache/pinot/segment/local/dedup/PartitionDedupMetadataManager.java

+  }
+
+  public boolean checkRecordPresentOrUpdate(RecordInfo recordInfo, IndexSegment indexSegment) {
+    if (!_allSegmentsLoaded) {


Let's move the if check into the waitTillAllSegmentsLoaded() for thread safety. It is single threaded now, but in case that changes

Could you help me understand the thread safety concerns with this? I don't any, single threaded or multi threaded.

Infact, moving this if check inside waitTillAllSegmentsLoaded() would lead to unnecessary serialization even when all segments have already been loaded. Even in single threaded env, that's a heavy lock acquisition cost, when _allSegmentsLoaded is already true.

To the point where, I think we should reduce the critical section here https://github.com/apache/pinot/blob/master/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/upsert/PartialUpsertHandler.java#L74, once _allSegmentsLoaded has been set to true, no need to enter a syncronized block.

Do let me know your thoughts

I misused the word thread-safety. I was suggesting adding an extra if check in the synchronized block to avoid potential unnecessary checks when multiple threads invoke waitTillAllSegmentsLoaded().
Good point on reducing the critical section in PartialUpsertHandler. We should first check the flag, then enter the critical section

Jackie-Jiang · 2022-06-02T00:00:49Z

...al/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java

+    }
+
+    if (isDedupEnabled() && _partitionDedupMetadataManager.checkRecordPresentOrUpdate(recordInfo, this)) {
+      _logger.info("Dropped row {} since its primary key already exists", row);


Don't log anything here, it can flood the log

Jackie-Jiang · 2022-06-02T00:02:35Z

...al/src/main/java/org/apache/pinot/segment/local/indexsegment/mutable/MutableSegmentImpl.java

+      if (_serverMetrics != null) {
+        _serverMetrics.addMeteredTableValue(_realtimeTableName, ServerMeter.REALTIME_DEDUP_DROPPED, 1);
+      }
+      return numDocsIndexed < _capacity;


(minor)

Suggested change

return numDocsIndexed < _capacity;

return true;

Jackie-Jiang · 2022-06-02T00:05:53Z

...-segment-local/src/main/java/org/apache/pinot/segment/local/utils/tablestate/TableState.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TableState {


(minor) Let's rename it to TableStateUtils

Jackie-Jiang · 2022-06-02T00:06:36Z

pinot-segment-local/src/test/resources/data/test_dedup_data.json

+    "description" : "second",
+    "secondsSinceEpoch": 1567205392
+  }
+]


(nit) empty line

Jackie-Jiang · 2022-06-02T00:06:40Z

pinot-segment-local/src/test/resources/data/test_dedup_schema.json

+    }
+  },
+  "primaryKeyColumns": ["event_id"]
+}


(nit) empty line

Jackie-Jiang · 2022-06-02T00:08:58Z

...-local/src/main/java/org/apache/pinot/segment/local/dedup/PartitionDedupMetadataManager.java

+  }
+
+  @VisibleForTesting
+  public static Iterator<RecordInfo> getRecordInfoIterator(IndexSegment segment, List<String> primaryKeyColumns) {


Suggest returning an iterator of PrimaryKey. For dedup, we don't need the docId and comparisonValue information from the RecordInfo. Similar for the checkRecordPresentOrUpdate() which can just take the PrimaryKey object. This is not a blocker, so maybe put a TODO and address it later

Ack. Didn't see any big impact of changing the method signature to accept PK, hence made that change too.

KKcorps · 2022-06-03T14:59:17Z

@saurabhd336 Please add documentation for this in Import Data section. You can create a new page titled Stream Ingestion with Deduplication

saurabhd336 marked this pull request as draft May 16, 2022 11:26

saurabhd336 force-pushed the dedupConfig branch 2 times, most recently from f684376 to 3d42ad5 Compare May 23, 2022 07:13

saurabhd336 marked this pull request as ready for review May 23, 2022 08:43

saurabhd336 force-pushed the dedupConfig branch from 31d784e to 1934df5 Compare May 24, 2022 09:35

saurabhd336 changed the title ~~(WIP) Dedup config~~ Dedup config May 24, 2022

saurabhd336 force-pushed the dedupConfig branch from 1934df5 to 590a954 Compare May 25, 2022 05:45

KKcorps requested a review from Jackie-Jiang May 25, 2022 06:04

Jackie-Jiang reviewed May 25, 2022

View reviewed changes

saurabhd336 force-pushed the dedupConfig branch 7 times, most recently from d41aa14 to f594261 Compare June 1, 2022 09:01

Saurabh Dubey added 13 commits June 1, 2022 15:54

Add dedup config to table config

9ea1575

Add dedup config to builders and indexers

e5fb51d

Fix tests

a690403

API changes

9e4fd37

Support segment being loaded

87f9c60

Add segment

7253329

Segment add / remove support

d71039e

Add isAllSegmentsLoaded check before ingestion

7983b2a

Set primaryKeyColumns correctly

5192d6f

Handle reload, keep segment reference as state

7fb1bfc

Review comments

cdc7ea5

Review comments

829eee5

Review comments

b665624

Saurabh Dubey added 2 commits June 1, 2022 15:54

Add metrics

f4c0b7f

Fix tests

a22ef7a

saurabhd336 force-pushed the dedupConfig branch from 08da312 to e0cf1be Compare June 1, 2022 10:24

Add tests

6c35b1b

saurabhd336 force-pushed the dedupConfig branch from e0cf1be to 6c35b1b Compare June 1, 2022 11:19

Saurabh Dubey added 2 commits June 1, 2022 17:03

Metrics

f151f28

Remove hashFunction

9f360b4

saurabhd336 force-pushed the dedupConfig branch from f50819e to 9f360b4 Compare June 1, 2022 11:59

Jackie-Jiang approved these changes Jun 2, 2022

View reviewed changes

saurabhd336 force-pushed the dedupConfig branch 2 times, most recently from 693659f to b56bfe2 Compare June 2, 2022 05:57

Review comments

38f4b55

saurabhd336 force-pushed the dedupConfig branch from b56bfe2 to 38f4b55 Compare June 2, 2022 08:09

Jackie-Jiang added feature release-notes Referenced by PRs that need attention when compiling the next release notes labels Jun 2, 2022

Jackie-Jiang changed the title ~~Dedup config~~ [Feature] Deduplication Jun 2, 2022

Jackie-Jiang merged commit 2b82366 into apache:master Jun 2, 2022

Jackie-Jiang mentioned this pull request Jul 5, 2023

Deduping mechanism in pinot #5552

Closed

	.format("PartitionGroupId is not available for segment: '%s' (upsert-enabled table: %s)", segmentName,
	.format("PartitionGroupId is not available for segment: '%s' (dedup-enabled table: %s)", segmentName,

	if (tableConfig.getUpsertConfig() != null) {
	if (tableConfig.getUpsertMode() != UpsertConfig.Mode.NONE) {

	final ConcurrentHashMap<Object, IndexSegment> _primaryKeySet = new ConcurrentHashMap<>();
	final ConcurrentHashMap<Object, IndexSegment> _primaryKeyToSegmentMap = new ConcurrentHashMap<>();

[Feature] Deduplication #8708

[Feature] Deduplication #8708

Conversation

saurabhd336 commented May 16, 2022 • edited Loading

How to use

codecov-commenter commented May 16, 2022 • edited Loading

Codecov Report

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

saurabhd336 Jun 2, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KKcorps commented Jun 3, 2022 • edited Loading

saurabhd336 commented May 16, 2022 •

edited

Loading

codecov-commenter commented May 16, 2022 •

edited

Loading

saurabhd336 Jun 2, 2022 •

edited

Loading

KKcorps commented Jun 3, 2022 •

edited

Loading