[FSTORE-743][append][java] Add support for insert to external feature groups online (#983)

Author: moritzmeister

java/spark/src/main/java/com/logicalclocks/hsfs/spark/ExternalFeatureGroup.java

@@ -88,7 +88,7 @@ public ExternalFeatureGroup(FeatureStore featureStore, @NonNull String name, Int
                               ExternalDataFormat dataFormat, String path, Map<String, String> options,
                               @NonNull StorageConnector storageConnector, String description,
                               List<String> primaryKeys, List<Feature> features, StatisticsConfig statisticsConfig,
-                              String eventTime) {
+                              String eventTime, boolean onlineEnabled, String onlineTopicName) {
     this();
     this.timeTravelFormat = null;
     this.featureStore = featureStore;
@@ -108,6 +108,8 @@ public ExternalFeatureGroup(FeatureStore featureStore, @NonNull String name, Int
     this.features = features;
     this.statisticsConfig = statisticsConfig != null ? statisticsConfig : new StatisticsConfig();
     this.eventTime = eventTime;
+    this.onlineEnabled = onlineEnabled;
+    this.onlineTopicName = onlineTopicName;
   }
 
   public ExternalFeatureGroup() {
@@ -135,17 +137,17 @@ public Dataset<Row> read() throws FeatureStoreException, IOException {
 
   @Override
   public Dataset<Row> read(boolean online) throws FeatureStoreException, IOException {
-    return null;
+    return selectAll().read(online);
   }
 
   @Override
   public Dataset<Row> read(Map<String, String> readOptions) throws FeatureStoreException, IOException {
-    return null;
+    return selectAll().read(false, readOptions);
   }
 
   @Override
   public Dataset<Row> read(boolean online, Map<String, String> readOptions) throws FeatureStoreException, IOException {
-    return null;
+    return selectAll().read(online, readOptions);
   }
 
   @Override
@@ -176,18 +178,7 @@ public void show(int numRows) throws FeatureStoreException, IOException {
 
   @Override
   public void show(int numRows, boolean online) throws FeatureStoreException, IOException {
-
-  }
-
-  @Override
-  public void insert(Dataset<Row> featureData) throws IOException, FeatureStoreException, ParseException {
-
-  }
-
-  @Override
-  public void insert(Dataset<Row> featureData, Map<String, String> writeOptions)
-      throws FeatureStoreException, IOException, ParseException {
-
+    read(true).show(numRows);
   }
 
   @Override
@@ -238,6 +229,77 @@ public void insert(Dataset<Row> featureData, boolean overwrite, Map<String, Stri
 
   }
 
+  /**
+   * Incrementally insert data to the online storage of an external feature group. The feature group has to be online
+   * enabled to perform this operation.
+   * The `features` dataframe can be a Spark DataFrame or RDD.
+   * If statistics are enabled, statistics are recomputed for the entire feature group.
+   * If the feature group doesn't exist, the insert method will create the necessary metadata the first time it is
+   * invoked and write the specified `features` dataframe as feature group to the online feature store.
+   *
+   * <pre>
+   * {@code
+   *        // get feature store handle
+   *        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
+   *        // get feature group handle
+   *        ExternalFeatureGroup fg = fs.getExternalFeatureGroup("electricity_prices", 1);
+   *        // insert data
+   *        fg.insert(featureData, writeOptions);
+   * }
+   * </pre>
+   *
+   * @param featureData Spark DataFrame, RDD. Features to be saved.
+   * @throws IOException Generic IO exception.
+   * @throws FeatureStoreException If client is not connected to Hopsworks; cannot run read query on storage and/or
+   *                               can't reconcile schema.
+   */
+  @Override
+  public void insert(Dataset<Row> featureData)
+      throws FeatureStoreException, IOException {
+
+    featureGroupEngine.insert(this, featureData, null);
+
+    codeEngine.saveCode(this);
+    computeStatistics();
+  }
+
+  /**
+   * Incrementally insert data to the online storage of an external feature group. The feature group has to be online
+   * enabled to perform this operation.
+   * The `features` dataframe can be a Spark DataFrame or RDD.
+   * If statistics are enabled, statistics are recomputed for the entire feature group.
+   * If the feature group doesn't exist, the insert method will create the necessary metadata the first time it is
+   * invoked and write the specified `features` dataframe as feature group to the online feature store.
+   *
+   * <pre>
+   * {@code
+   *        // get feature store handle
+   *        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
+   *        // get feature group handle
+   *        ExternalFeatureGroup fg = fs.getExternalFeatureGroup("electricity_prices", 1);
+   *        // Define additional write options (for example for Spark)
+   *        Map<String, String> writeOptions = = new HashMap<String, String>();
+   *        // insert data
+   *        fg.insert(featureData, writeOptions);
+   * }
+   * </pre>
+   *
+   * @param featureData Spark DataFrame, RDD. Features to be saved.
+   * @param writeOptions Additional write options as key-value pairs.
+   * @throws IOException Generic IO exception.
+   * @throws FeatureStoreException If client is not connected to Hopsworks; cannot run read query on storage and/or
+   *                               can't reconcile schema.
+   */
+  @Override
+  public void insert(Dataset<Row> featureData, Map<String, String> writeOptions)
+      throws FeatureStoreException, IOException, ParseException {
+
+    featureGroupEngine.insert(this, featureData, writeOptions);
+
+    codeEngine.saveCode(this);
+    computeStatistics();
+  }
+
   @Override
   public void commitDeleteRecord(Dataset<Row> featureData) throws FeatureStoreException, IOException, ParseException {
 

java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/FeatureGroupEngine.java

@@ -157,6 +157,25 @@ public void insert(StreamFeatureGroup streamFeatureGroup, Dataset<Row> featureDa
         SparkEngine.getInstance().getKafkaConfig(streamFeatureGroup, writeOptions));
   }
 
+  public void insert(ExternalFeatureGroup externalFeatureGroup, Dataset<Row> featureData,
+      Map<String, String> writeOptions)
+      throws FeatureStoreException, IOException {
+
+    if (!externalFeatureGroup.getOnlineEnabled()) {
+      throw new FeatureStoreException("Online storage is not enabled for this feature group. External feature groups "
+          + "can only store data in online storage. To create an offline only external feature group, use the `save` "
+          + "method.");
+    }
+
+    if (externalFeatureGroup.getId() == null) {
+      externalFeatureGroup = saveExternalFeatureGroup(externalFeatureGroup);
+    }
+
+    SparkEngine.getInstance().writeOnlineDataframe(externalFeatureGroup, featureData,
+        externalFeatureGroup.getOnlineTopicName(),
+        SparkEngine.getInstance().getKafkaConfig(externalFeatureGroup, writeOptions));
+  }
+
   @Deprecated
   public StreamingQuery insertStream(FeatureGroup featureGroup, Dataset<Row> featureData, String queryName,
                                      String outputMode, boolean awaitTermination, Long timeout,
@@ -461,6 +480,7 @@ public ExternalFeatureGroup saveExternalFeatureGroup(ExternalFeatureGroup extern
 
     ExternalFeatureGroup apiFg = saveExtennalFeatureGroupMetaData(externalFeatureGroup, ExternalFeatureGroup.class);
     externalFeatureGroup.setId(apiFg.getId());
+    externalFeatureGroup.setOnlineTopicName(apiFg.getOnlineTopicName());
 
     return externalFeatureGroup;
   }

java/spark/src/test/java/com/logicalclocks/hsfs/spark/TestExternalFeatureGroup.java

@@ -50,7 +50,7 @@ public void testExternalFeatureGroupPrimaryKey() throws FeatureStoreException, I
 
     ExternalFeatureGroup externalFeatureGroup = new ExternalFeatureGroup(featureStore, "fgName", 1,
         "SELECT *", ExternalDataFormat.HUDI, "path", null, storageConnector, "description",
-        Collections.singletonList("primaryKey"), features, null, "featureA");
+        Collections.singletonList("primaryKey"), features, null, "featureA", false, "topic");
 
     Exception pkException = assertThrows(FeatureStoreException.class, () -> {
       externalFeatureGroupEngine.saveExternalFeatureGroup(externalFeatureGroup);
@@ -76,7 +76,7 @@ public void testFeatureGroupEventTimeFeature() throws FeatureStoreException, IOE
 
     ExternalFeatureGroup externalFeatureGroup = new ExternalFeatureGroup(featureStore, "fgName", 1,
         "SELECT *", ExternalDataFormat.HUDI, "path", null, storageConnector, "description",
-        Collections.singletonList("featureA"), features, null, "eventTime");
+        Collections.singletonList("featureA"), features, null, "eventTime", false, "topic");
 
     Exception pkException = assertThrows(FeatureStoreException.class, () -> {
       externalFeatureGroupEngine.saveExternalFeatureGroup(externalFeatureGroup);