Persistent Cache Implementation using RocksDB #198
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
| /** | ||
| * Copyright DataStax, Inc 2021. | ||
| * | ||
| * Licensed 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 com.datastax.oss.cdc.cache; | ||
|
|
||
| import com.github.benmanes.caffeine.cache.stats.CacheStats; | ||
|
|
||
| import java.util.List; | ||
|
|
||
| public interface MutationCache<K> { | ||
|
There was a problem hiding this comment. Interface level description on what this cache will be used for. (de-duplication) |
||
| /** | ||
| * Add a mutation MD5 digest to the cache for the given mutation key. | ||
| * @param mutationKey the key for the mutation, typically a partition key or a unique identifier | ||
| * @param md5Digest the MD5 digest of the mutation to be added | ||
| * @return a list of MD5 digests for the given mutation key, which may include the newly added digest | ||
| */ | ||
| List<String> addMutationMd5(K mutationKey, String md5Digest); | ||
|
|
||
| /** | ||
| * Retrieve the list of MD5 digests for the given mutation key. | ||
| * @param mutationKey the key for the mutation | ||
| * @return a list of MD5 digests associated with the mutation key, or an empty list if none exist | ||
| */ | ||
| List<String> getMutationCRCs(K mutationKey); | ||
|
|
||
| /** | ||
| * Check if a mutation with the given key and MD5 digest has already been processed. | ||
| * @param mutationKey the key for the mutation | ||
| * @param md5Digest the MD5 digest of the mutation | ||
| * @return true if the mutation has been processed, false otherwise | ||
| */ | ||
| boolean isMutationProcessed(K mutationKey, String md5Digest); | ||
|
|
||
| /** | ||
| * Gives the current statistics of the cache, such as hit rate, miss rate, and size. | ||
| * <a href="https://github.com/ben-manes/caffeine/wiki/Statistics">Caffeine Statistics wiki</a> | ||
| */ | ||
| CacheStats stats(); | ||
|
|
||
| /** | ||
| * Gives the estimated size of the cache. | ||
| */ | ||
| long estimatedSize(); | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,142 @@ | ||
| /** | ||
| * Copyright DataStax, Inc 2021. | ||
| * | ||
| * Licensed 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 com.datastax.oss.cdc.cache; | ||
|
|
||
| import com.github.benmanes.caffeine.cache.Cache; | ||
| import com.github.benmanes.caffeine.cache.Caffeine; | ||
| import com.github.benmanes.caffeine.cache.RemovalCause; | ||
| import com.github.benmanes.caffeine.cache.stats.CacheStats; | ||
| import org.rocksdb.Options; | ||
| import org.rocksdb.RocksDBException; | ||
| import org.rocksdb.TtlDB; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.function.Function; | ||
|
|
||
| public class PersistentCache<K> implements MutationCache<K> { | ||
| /** | ||
| * The mutation cache | ||
| */ | ||
| private final Cache<K, List<String>> mutationCache; | ||
|
|
||
| private final TtlDB rocksDB; | ||
| /** | ||
| * Max number of cached digests per cached entry. | ||
| */ | ||
| long maxDigests; | ||
|
|
||
| private final Function<K, byte[]> keySerializer; | ||
|
|
||
| static { | ||
| TtlDB.loadLibrary(); | ||
| } | ||
|
|
||
| public PersistentCache(long maxDigests, long maxCapacity, Duration expireAfter, String dbPath, Function<K, byte[]> keySerializer) throws RocksDBException { | ||
| this.maxDigests = maxDigests; | ||
| this.keySerializer = keySerializer; | ||
|
|
||
| Options options = new Options().setCreateIfMissing(true); | ||
| this.rocksDB = TtlDB.open(options, dbPath, (int) expireAfter.getSeconds(), false); | ||
|
|
||
| this.mutationCache = Caffeine.newBuilder() | ||
| .expireAfterWrite(expireAfter.getSeconds(), TimeUnit.SECONDS) | ||
| .maximumSize(maxCapacity) | ||
| .recordStats() | ||
| .removalListener((K key, List<String> value, RemovalCause cause) -> { | ||
| try { | ||
| // If the removal cause is not SIZE, we delete the key from RocksDB | ||
| // This is to avoid deleting the key when it is removed due to the size limit | ||
| if (cause != RemovalCause.SIZE ) { | ||
| rocksDB.delete(this.keySerializer.apply(key)); | ||
| } | ||
| } catch (RocksDBException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| }) | ||
| .build(); | ||
| } | ||
|
|
||
|
|
||
| private List<String> valueDeserializer(byte[] data){ | ||
| return data == null || data.length == 0 ? null : List.of(new String(data).split(",")); | ||
| } | ||
|
|
||
| private byte[] valueSerializer(List<String> data){ | ||
| return data.stream() | ||
| .reduce((s1, s2) -> s1 + "," + s2) | ||
| .orElse("") | ||
| .getBytes(); | ||
| } | ||
|
|
||
| public List<String> getMutationCRCs(K mutationKey) { | ||
| return mutationCache.get(mutationKey, k -> { | ||
| try { | ||
| return valueDeserializer(rocksDB.get(this.keySerializer.apply(k))); | ||
| } catch (RocksDBException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public void putMutationCRCs(K key, List<String> value) { | ||
| mutationCache.asMap().compute(key, (k, v) -> { | ||
| try { | ||
| rocksDB.put(keySerializer.apply(k), valueSerializer(value)); | ||
| } catch (RocksDBException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| return value; | ||
| }); | ||
| } | ||
|
|
||
| public List<String> addMutationMd5(K mutationKey, String md5Digest) { | ||
| List<String> crcs = getMutationCRCs(mutationKey); | ||
| if(crcs == null) { | ||
| crcs = new ArrayList<>(1); | ||
| crcs.add(md5Digest); | ||
| } else { | ||
| if (!crcs.contains(md5Digest)) { | ||
| if (crcs.size() >= maxDigests) { | ||
| // remove the oldest digest | ||
| crcs.remove(0); | ||
| } | ||
| crcs.add(md5Digest); | ||
| } | ||
| } | ||
| putMutationCRCs(mutationKey, crcs); | ||
| return crcs; | ||
| } | ||
|
|
||
| public boolean isMutationProcessed(K mutationKey, String md5Digest) { | ||
| List<String> digests = getMutationCRCs(mutationKey); | ||
| return digests != null && digests.contains(md5Digest); | ||
| } | ||
|
|
||
| public CacheStats stats() { | ||
| return mutationCache.stats(); | ||
| } | ||
|
|
||
| public long estimatedSize() { | ||
| return mutationCache.estimatedSize(); | ||
| } | ||
|
|
||
| public void close() { | ||
| rocksDB.close(); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
360 minutes is too high. Why don't we make it 180.
cc: @srinath-ctds