support flink_value_extractor_factory jni #220

java/rocksjni/flink_compactionfilterjni.cc

@@ -183,6 +183,15 @@ void Java_org_rocksdb_FlinkCompactionFilter_disposeFlinkCompactionFilterConfigHo
   delete config_holder;
 }
 
+jlong Java_org_rocksdb_FlinkCompactionFilter_createFlinkValueExtractorFactory(
+    JNIEnv* env, jclass /* jcls */, jlong config_holder_handle) {
+  using namespace TERARKDB_NAMESPACE::flink;
+  auto config_holder =
+      *(reinterpret_cast<std::shared_ptr<FlinkCompactionFilter::ConfigHolder>*>(
+          config_holder_handle));
+  return reinterpret_cast<jlong>(new FlinkValueExtractorFactory(config_holder));
+}
+
 /*
  * Class:     org_rocksdb_FlinkCompactionFilter
  * Method:    createNewFlinkCompactionFilter0

java/rocksjni/options.cc

@@ -3305,6 +3305,20 @@ void Java_org_rocksdb_ColumnFamilyOptions_setMergeOperator(
           mergeOperatorHandle));
 }
 
+/*
+ * Class:     org_rocksdb_ColumnFamilyOptions
+ * Method:    setCompactionFilterHandle
+ * Signature: (JJ)V
+ */
+void Java_org_rocksdb_ColumnFamilyOptions_setValueExtractorFactoryHandle(
+    JNIEnv* /*env*/, jobject /*jobj*/, jlong jopt_handle,
+    jlong jvaluemeta_extractor_factory_handle) {
+  reinterpret_cast<TERARKDB_NAMESPACE::ColumnFamilyOptions*>(jopt_handle)
+      ->value_meta_extractor_factory =
+      reinterpret_cast<TERARKDB_NAMESPACE::ValueExtractorFactory*>(
+          jvaluemeta_extractor_factory_handle);
+}
+
 /*
  * Class:     org_rocksdb_ColumnFamilyOptions
  * Method:    setCompactionFilterHandle

java/rocksjni/value_extractor.cc

@@ -0,0 +1,29 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+//
+// This file implements the "bridge" between Java and C++ for
+// TERARKDB_NAMESPACE::CompactionFilter.
+
+#include "rocksdb/value_extractor.h"
+
+#include <jni.h>
+
+#include "include/org_rocksdb_AbstractValueExtractorFactory.h"
+
+// <editor-fold desc="org.rocksdb.AbstractCompactionFilter">
+
+/*
+ * Class:     org_rocksdb_AbstractCompactionFilter
+ * Method:    disposeInternal
+ * Signature: (J)V
+ */
+void Java_org_rocksdb_AbstractValueExtractorFactory_disposeInternal(
+    JNIEnv* /*env*/, jobject /*jobj*/, jlong handle) {
+  auto* vef =
+      reinterpret_cast<TERARKDB_NAMESPACE::ValueExtractorFactory*>(handle);
+  assert(vef != nullptr);
+  delete vef;
+}
+// </editor-fold>

java/src/main/java/org/rocksdb/AbstractValueExtractoFactory.java

@@ -0,0 +1,40 @@
+// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
+//  This source code is licensed under both the GPLv2 (found in the
+//  COPYING file in the root directory) and Apache 2.0 License
+//  (found in the LICENSE.Apache file in the root directory).
+
+package org.rocksdb;
+
+/**
+ * Each compaction will create a new {@link AbstractCompactionFilter}
+ * allowing the application to know about different compactions
+ *
+ * @param <T> The concrete type of the compaction filter
+ */
+public abstract class AbstractValueExtractorFactory<T extends AbstractSlice<?>>
+    extends RocksObject {
+
+  public AbstractValueExtractorFactory(final long nativeHandle) {
+    super(nativeHandle);
+  }
+
+
+  /**
+   * A name which identifies this compaction filter
+   *
+   * The name will be printed to the LOG file on start up for diagnosis
+   */
+  public abstract String name();
+
+  /**
+   * We override {@link RocksCallbackObject#disposeInternal()}
+   * as disposing of a TERARKDB_NAMESPACE::AbstractCompactionFilterFactory requires
+   * a slightly different approach as it is a std::shared_ptr
+   */
+  @Override
+  protected void disposeInternal() {
+    disposeInternal(nativeHandle_);
+  }
+
+  private native void disposeInternal(final long handle);
+}

java/src/main/java/org/rocksdb/FlinkCompactionFilter.java

@@ -26,7 +26,7 @@ public FlinkCompactionFilter(ConfigHolder configHolder, TimeProvider timeProvide
    public FlinkCompactionFilter(ConfigHolder configHolder, TimeProvider timeProvider, Logger logger) {
     super(createNewFlinkCompactionFilter0(configHolder.nativeHandle_, timeProvider, logger == null ? 0 : logger.nativeHandle_));
   }
-
+   private native static long createFlinkValueExtractorFactory(long configHolderHandle);
    private native static long createNewFlinkCompactionFilter0(long configHolderHandle, TimeProvider timeProvider, long loggerHandle);
   private native static long createNewFlinkCompactionFilterConfigHolder();
   private native static void disposeFlinkCompactionFilterConfigHolder(long configHolderHandle);
@@ -118,6 +118,18 @@ public interface TimeProvider {
     long currentTimestamp();
   }
 
+  public static class FlinkValueExtractorFactory
+    extends AbstractValueExtractorFactory<Slice> {
+
+      public FlinkValueExtractor(ConfigHolder configHolder) {
+          super(createFlinkValueExtractorFactory(configHolder));
+      }
+      @Override
+      public String name() {
+        return "FlinkValueExtractorFactory";
+      }
+  }
+
    public static class FlinkCompactionFilterFactory extends AbstractCompactionFilterFactory<FlinkCompactionFilter> {
     private final ConfigHolder configHolder;
     private final TimeProvider timeProvider;

utilities/flink/flink_compaction_filter.h

@@ -33,7 +33,7 @@ static const std::size_t JAVA_MAX_SIZE = static_cast<std::size_t>(0x7fffffff);
  * Note: this compaction filter is a special implementation, designed for usage
  * only in Apache Flink project.
  */
-class FlinkCompactionFilter : public CompactionFilter, ValueExtractor {
+class FlinkCompactionFilter : public CompactionFilter, public ValueExtractor {
  public:
   enum StateType {
     // WARNING!!! Do not change the order of enum entries as it is important for
@@ -144,7 +144,7 @@ class FlinkCompactionFilter : public CompactionFilter, ValueExtractor {
                     std::string* skip_until) const override;
 
   Status Extract(const Slice& key, const Slice& value,
-                         std::string* output) const override;
+                 std::string* output) const override;
 
   bool IgnoreSnapshots() const override { return true; }
 
@@ -196,5 +196,22 @@ static const FlinkCompactionFilter::Config DISABLED_CONFIG =
                                   std::numeric_limits<int64_t>::max(),
                                   std::numeric_limits<int64_t>::max(), nullptr};
 
+class FlinkValueExtractorFactory : public ValueExtractorFactory {
+ public:
+  const char* Name() const override {
+    return "flink.ValueTimeStampExtractorFactory";
+  }
+  explicit FlinkValueExtractorFactory(
+      std::shared_ptr<FlinkCompactionFilter::ConfigHolder> config_holder)
+      : config_holder_(std::move(config_holder)) {}
+  std::unique_ptr<ValueExtractor> CreateValueExtractor(
+      const Context& context) const {
+    return std::unique_ptr<ValueExtractor>(
+        new FlinkCompactionFilter(config_holder_, nullptr, nullptr));
+  };
+ private:
+  std::shared_ptr<FlinkCompactionFilter::ConfigHolder> config_holder_;
+};
+
 }  // namespace flink
 }  // namespace TERARKDB_NAMESPACE