@Override
  public void putCacheValue(
      Cache<ByteArrayWrapper, Set<ByteArrayWrapper>> cache,
      ByteArrayWrapper keyWrap,
      ByteArrayWrapper valueWrap) {
    if (initialized()) {
      byte[] joinKeyBytes = keyWrap.bytes;
      synchronized (rocksDBLock) {
        byte[] uniqueKeysDeserialized = rocksDB.get(columnFamilyHandle, joinKeyBytes);
        if (uniqueKeysDeserialized != null) {
          Set<ByteArrayWrapper> set = ByteArraySetSerializer.deserialize(uniqueKeysDeserialized);
          if (!set.contains(valueWrap)) {
            set.add(valueWrap);
            uniqueKeysDeserialized = ByteArraySetSerializer.serialize(set);
            rocksDB.put(columnFamilyHandle, joinKeyBytes, uniqueKeysDeserialized);
          }
        } else {
          Set<ByteArrayWrapper> set = new HashSet<>();
          set.add(valueWrap);
          uniqueKeysDeserialized = ByteArraySetSerializer.serialize(set);
          rocksDB.put(columnFamilyHandle, joinKeyBytes, uniqueKeysDeserialized);
        }
      }
      return;
    }
    tmpInitializationMap.compute(
        keyWrap,
        (keyWrapper, oldSet) -> {
          if (oldSet == null) {
            oldSet = Sets.newHashSet();
          }
          oldSet.add(valueWrap);
          return oldSet;
        });
  }
Retrieve the elements of the key.

<p>Fetch the Collection from guava cache, if not present, fetch the result from RocksDB. if
present, just return the result.

@return not null, but may be empty.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/lookup/RocksDBSetSpilledState.java