changing the dedup store to become pluggable by raghavgautam · Pull Request #10639 · apache/pinot

raghavgautam · 2023-04-18T22:33:11Z

This patch is the first patch for memory problem discussed in #10571

It makes the LocalKeyValue store to be pluggable. Currently, it is hard-coded to be ConcurrentHashMap. This will make it easy to plugin different implementations.

raghavgautam · 2023-04-18T22:42:13Z

...local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentHashMapKeyValueStore.java

+  public void compact() {
+  }
+
+  private static final class ByteArray {


We need this because java.lang.Array picks gets it's hashCode() & equals() method from java.lang.Object which does not work for HashMap keys.
https://docs.oracle.com/javase/8/docs/api/java/lang/reflect/Array.html

use https://github.com/apache/pinot/blob/master/pinot-spi/src/main/java/org/apache/pinot/spi/utils/ByteArray.java ?

Will make the change.

codecov-commenter · 2023-04-18T23:23:43Z

Codecov Report

❌ Patch coverage is 67.05882% with 28 lines in your changes missing coverage. Please review.
✅ Project coverage is 70.27%. Comparing base (f6c6d14) to head (724aec9).
⚠️ Report is 4035 commits behind head on master.

Files with missing lines	Patch %	Lines
...up/ConcurrentMapPartitionDedupMetadataManager.java	66.00%	17 Missing ⚠️
.../local/dedup/TableDedupMetadataManagerFactory.java	47.05%	7 Missing and 2 partials ⚠️
...ent/local/dedup/BaseTableDedupMetadataManager.java	81.81%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #10639      +/-   ##
============================================
- Coverage     70.35%   70.27%   -0.09%     
+ Complexity     6464     6433      -31     
============================================
  Files          2103     2114      +11     
  Lines        112769   114072    +1303     
  Branches      16981    17227     +246     
============================================
+ Hits          79341    80160     +819     
- Misses        27877    28306     +429     
- Partials       5551     5606      +55

Flag	Coverage Δ
integration1	`24.30% <0.00%> (-0.17%)`	⬇️
integration2	`24.03% <1.17%> (-0.22%)`	⬇️
unittests1	`67.82% <65.88%> (-0.01%)`	⬇️
unittests2	`13.81% <0.00%> (-0.07%)`	⬇️

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

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

yupeng9 · 2023-04-19T03:05:41Z

...local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentHashMapKeyValueStore.java

+
+  private final Map<ByteArray, byte[]> _map;
+
+  public ConcurrentHashMapKeyValueStore(byte[] id) {


we do not use the input?

For concurrent hashmap this is unused, I will remove it.

yupeng9 · 2023-04-19T03:09:02Z

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

-    _hashFunction = hashFunction;
+    _dedupConfig = dedupConfig;
+    try {
+      byte[] id = (tableNameWithType + "@" + partitionId).getBytes();


why do we need table name in the key?

We don't need it for concurrent hashmap. But for persistent stores, this is needed to identify which table-partition combination we are storing/querying for.

yupeng9 · 2023-04-19T03:09:59Z

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

+    if (pk instanceof ByteArray) {
+      return ((ByteArray) pk).getBytes();
+    }
+    throw new RuntimeException("Invalid primary key: " + pk);


the msg can better be "unsupported pk class"

Will make the change.

yupeng9 · 2023-04-19T03:11:21Z

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

  }
+
+  public String getKeyStore() {
+    return _keyStore;


this has a default value?

for @JsonProperty, if a default value has not been unspecified it is taken to be empty string

right, can you add a comment on the default impl of this store?

Will make the change.

yupeng9 · 2023-04-19T03:11:30Z

pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/dedup/LocalKeyValueStore.java

+
+    void putBatch(List<Pair<byte[], byte[]>> keyValues);
+
+    long getKeyCount();


does this need to be long?

At 1K/sec this will get 1B keys in 12 days and int only goes till 2B.

yupeng9 · 2023-04-20T18:43:25Z

LGTM. @Jackie-Jiang can you also take a look?

ankitsultana · 2023-04-21T00:19:45Z

pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/dedup/LocalKeyValueStore.java

+import org.apache.commons.lang3.tuple.Pair;
+
+public interface LocalKeyValueStore {
+    byte[] get(byte[] key);


Can we use Generics here to allow the ConcurrentHashMap kv-store to skip serialization?

When we stick with in-memory hash-map we wouldn't want to add the overhead of serialization during ingestion.

If we try generics, the code will have to know that serialization is not needed so, it will be coupled to the ConcurrentHashMapKeyValueStore implementation. This is undesirable. The serialization that we are using is simple enough to not add significant overhead.

ankitsultana · 2023-04-21T00:20:15Z

...l/src/test/java/org/apache/pinot/segment/local/dedup/ConcurrentHashMapKeyValueStoreTest.java

+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+public class ConcurrentHashMapKeyValueStoreTest {


Code format seems incorrect. (we use 2 spaces instead of 4 for indent)

Will fix. Unsure why checkstyle + CI didn't catch this.

ankitsultana · 2023-04-21T00:26:03Z

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

-        _primaryKeyToSegmentMap.putIfAbsent(HashUtils.hashPrimaryKey(pk, _hashFunction), indexSegment) != null;
-    if (!present) {
+    byte[] keyBytes = serializePrimaryKey(HashUtils.hashPrimaryKey(pk, _dedupConfig.getHashFunction()));
+    if (Objects.isNull(_keyValueStore.get(keyBytes))) {


This is a non-atomic operation (contrast with putIfAbsent).

Will fix this by bringing back the putIfAbsent.

ankitsultana · 2023-04-24T17:43:28Z

...local/src/main/java/org/apache/pinot/segment/local/dedup/ConcurrentHashMapKeyValueStore.java

+  public void put(byte[] key, byte[] value) {
+    _map.put(new ByteArray(key), value);
+  }
+  public byte[] putIfAbsent(byte[] key, byte[] value) {


Missing space between methods.

ankitsultana · 2023-04-24T17:43:34Z

pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/dedup/LocalKeyValueStore.java

+  void delete(byte[] key);
+
+  void put(byte[] key, byte[] value);
+  byte[] putIfAbsent(byte[] key, byte[] value);


Missing space between methods.

ankitsultana · 2023-04-24T17:43:58Z

pinot-spi/src/main/java/org/apache/pinot/spi/ingestion/dedup/LocalKeyValueStore.java

+import java.util.List;
+import org.apache.commons.lang3.tuple.Pair;
+
+public interface LocalKeyValueStore {


Let's also add doc comments. Particularly for compact.

Jackie-Jiang

Making it pluggable at KV store level will add extra overhead to the default implementation because it will force us to serialize everything. Instead, we can make it pluggable at metadata manager level so that there is no performance penalty to the default implementation. You may take a look at #9186 of how we made upsert metadata manager pluggable. This PR should be very similar to that

raghavgautam · 2023-04-27T18:02:56Z

Making it pluggable at KV store level will add extra overhead to the default implementation because it will force us to serialize everything. Instead, we can make it pluggable at metadata manager level so that there is no performance penalty to the default implementation. You may take a look at #9186 of how we made upsert metadata manager pluggable. This PR should be very similar to that

@Jackie-Jiang The serialization that we are using is simple enough to not add significant overhead. It will probably cost 100 or so CPU cycles per record. Moreover, this overhead happens only during ingestion time and does not affect query performance. Keeping the LocalKeyValueStore pluggable will make it reusable in other places as well.

Jackie-Jiang

@Jackie-Jiang The serialization that we are using is simple enough to not add significant overhead. It will probably cost 100 or so CPU cycles per record. Moreover, this overhead happens only during ingestion time and does not affect query performance. Keeping the LocalKeyValueStore pluggable will make it reusable in other places as well.

It is not that trivial comparing to other operations when updating a record. Adding a record to KV store could be much cheaper than serializing a primary key, meaning this change could cause several times performance degradation. Also, certain operation requires reference comparison, and should be handled differently if reference comparison is not possible

Jackie-Jiang · 2023-04-27T19:29:11Z

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

-        }
-      });
+      byte[] pkBytes = serializePrimaryKey(pk);
+      if (Objects.deepEquals(_keyValueStore.get(pkBytes), segmentBytes)) {


This can potentially cause race condition when the segment is added again during the segment removal, and that's why we perform the reference check before

This reverts commit 02af6a9.

This reverts commit 6a8c2e3.

This reverts commit 73adcac.

This reverts commit f09b6b1.

This reverts commit 8e60fff.

yupeng9 · 2023-05-01T15:21:26Z

@Jackie-Jiang The serialization that we are using is simple enough to not add significant overhead. It will probably cost 100 or so CPU cycles per record. Moreover, this overhead happens only during ingestion time and does not affect query performance. Keeping the LocalKeyValueStore pluggable will make it reusable in other places as well.

It is not that trivial comparing to other operations when updating a record. Adding a record to KV store could be much cheaper than serializing a primary key, meaning this change could cause several times performance degradation. Also, certain operation requires reference comparison, and should be handled differently if reference comparison is not possible

@raghavgautam maybe we can have some quick benchmark to see how much overhead it adds, and it can be helpful to decide which layer we shall do the abstraction

raghavgautam · 2023-05-02T19:08:16Z

@Jackie-Jiang I have addressed your comment. Can you take another look ?

yupeng9 · 2023-05-09T21:49:03Z

@Jackie-Jiang I made a pass on the most recent change, and it looks good to me. Feel free to review this post-merge when you are back.

changing the dedup store to become pluggable

8e60fff

raghavgautam commented Apr 18, 2023

View reviewed changes

yupeng9 reviewed Apr 19, 2023

View reviewed changes

raghavgautam added 2 commits April 19, 2023 11:52

addressing review comments

f09b6b1

addressing review comments - 2

73adcac

yupeng9 approved these changes Apr 20, 2023

View reviewed changes

ankitsultana reviewed Apr 21, 2023

View reviewed changes

addressing review comments - 3

6a8c2e3

ankitsultana reviewed Apr 24, 2023

View reviewed changes

addressing review comments - 4

02af6a9

ankitsultana approved these changes Apr 24, 2023

View reviewed changes

Jackie-Jiang reviewed Apr 26, 2023

View reviewed changes

Jackie-Jiang reviewed Apr 27, 2023

View reviewed changes

raghavgautam added 5 commits April 28, 2023 14:24

Revert "addressing review comments - 4"

6987cfd

This reverts commit 02af6a9.

Revert "addressing review comments - 3"

1ba56a7

This reverts commit 6a8c2e3.

Revert "addressing review comments - 2"

9758cd7

This reverts commit 73adcac.

Revert "addressing review comments"

db9b01e

This reverts commit f09b6b1.

Revert "changing the dedup store to become pluggable"

ae79080

This reverts commit 8e60fff.

raghavgautam added 2 commits May 2, 2023 10:39

making dedup code pluggable similar to upsert

d56464a

fixing checkstyle issue

724aec9

raghavgautam requested a review from Jackie-Jiang May 3, 2023 22:35

yupeng9 merged commit 53469c0 into apache:master May 9, 2023

raghavgautam deleted the pluggable-dedup branch May 10, 2023 00:03


		private final Map<ByteArray, byte[]> _map;

		public ConcurrentHashMapKeyValueStore(byte[] id) {


		void putBatch(List<Pair<byte[], byte[]>> keyValues);

		long getKeyCount();

Conversation

raghavgautam commented Apr 18, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Apr 18, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yupeng9 commented Apr 20, 2023

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Uh oh!

raghavgautam commented Apr 27, 2023

Uh oh!

Jackie-Jiang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yupeng9 commented May 1, 2023

Uh oh!

raghavgautam commented May 2, 2023

Uh oh!

yupeng9 commented May 9, 2023

Uh oh!

Reviewers

codecov-commenter commented Apr 18, 2023 •

edited

Loading