[FSTORE-1795] Update api to use data source and deprecate direct storage connector usage

docs/templates/api/data_source_api.md

@@ -0,0 +1,13 @@
+# Data Source
+
+## Retrieval
+
+{{ds_get}}
+
+## Properties
+
+{{data_source_properties}}
+
+## Methods
+
+{{data_source_methods}}

java/beam/src/main/java/com/logicalclocks/hsfs/beam/StreamFeatureGroup.java

@@ -68,7 +68,7 @@ public StreamFeatureGroup(FeatureStore featureStore, @NonNull String name, Integ
     this.onlineTopicName = onlineTopicName;
     this.eventTime = eventTime;
     this.onlineConfig = onlineConfig;
-    this.storageConnector = storageConnector;
+    this.dataSource.setStorageConnector(storageConnector);
     this.dataSource.setPath(path);
   }
 

java/flink/src/main/java/com/logicalclocks/hsfs/flink/StreamFeatureGroup.java

@@ -75,7 +75,7 @@ public StreamFeatureGroup(FeatureStore featureStore, @NonNull String name, Integ
     this.notificationTopicName = notificationTopicName;
     this.eventTime = eventTime;
     this.onlineConfig = onlineConfig;
-    this.storageConnector = storageConnector;
+    this.dataSource.setStorageConnector(storageConnector);
     this.dataSource.setPath(path);
   }
 

java/hsfs/src/main/java/com/logicalclocks/hsfs/DataSource.java

@@ -53,4 +53,8 @@ public class DataSource extends RestDto<DataSource> {
   @Setter
   private String path = "";
 
+  @Getter
+  @Setter
+  private StorageConnector storageConnector = null;
+
 }

java/hsfs/src/main/java/com/logicalclocks/hsfs/FeatureGroupBase.java

@@ -135,10 +135,6 @@ public abstract class FeatureGroupBase<T> {
   @Setter
   protected OnlineConfig onlineConfig;
 
-  @Getter
-  @Setter
-  protected StorageConnector storageConnector;
-
   @Getter
   @Setter
   protected DataSource dataSource;

java/hsfs/src/main/java/com/logicalclocks/hsfs/FeatureStoreBase.java

@@ -205,6 +205,34 @@ public StorageConnector getStorageConnector(String name) throws FeatureStoreExce
     return storageConnectorApi.getByName(this, name, StorageConnector.class);
   }
 
+  /**
+   * Get a previously created data source from the feature store.
+   *
+   * <p>data sources encapsulate all information needed for the execution engine to read and write to a specific
+   * storage.
+   *
+   * <p>If you want to connect to the online feature store, see the getOnlineDataSource` method to get the
+   * JDBC connector for the Online Feature Store.
+   *
+   * <pre>
+   * {@code
+   *        // get feature store handle
+   *        FeatureStore fs = HopsworksConnection.builder().build().getFeatureStore();
+   *        DataSource ds = fs.getDataSource("ds_name");
+   * }
+   * </pre>
+   *
+   * @param name Name of the data source to retrieve.
+   * @return DataSource Data source object.
+   * @throws FeatureStoreException If unable to retrieve DataSource from the feature store.
+   * @throws IOException Generic IO exception.
+   */
+  public DataSource getDataSource(String name) throws FeatureStoreException, IOException {
+    DataSource dataSource = new DataSource();
+    dataSource.setStorageConnector(getStorageConnector(name));
+    return dataSource;
+  }
+
   /**
    * Get a previously created HopsFs compliant storage connector from the feature store.
    *

java/hsfs/src/main/java/com/logicalclocks/hsfs/StreamFeatureGroup.java

@@ -60,8 +60,8 @@ public StreamFeatureGroup(FeatureStoreBase featureStore, @NonNull String name, I
     this.onlineTopicName = onlineTopicName;
     this.eventTime = eventTime;
     this.timeTravelFormat = timeTravelFormat;
-    this.storageConnector = storageConnector;
     this.onlineConfig = onlineConfig;
+    this.dataSource.setStorageConnector(storageConnector);
     this.dataSource.setPath(path);
   }
 

java/hsfs/src/main/java/com/logicalclocks/hsfs/TrainingDatasetBase.java

@@ -115,7 +115,7 @@ public class TrainingDatasetBase {
 
   @Getter
   @Setter
-  protected StorageConnector storageConnector;
+  protected DataSource dataSource;
 
   @Getter
   @Setter
@@ -138,14 +138,19 @@ public TrainingDatasetBase(Integer version, String description, DataFormat dataF
                          TrainingDatasetType trainingDatasetType, Float validationSize, Float testSize,
                          String trainStart, String trainEnd, String validationStart,
                          String validationEnd, String testStart, String testEnd, Integer timeSplitSize,
-                         FilterLogic extraFilterLogic, Filter extraFilter)
+                         FilterLogic extraFilterLogic, Filter extraFilter, DataSource dataSource)
       throws FeatureStoreException, ParseException {
     this.version = version;
     this.description = description;
     this.dataFormat = dataFormat != null ? dataFormat : DataFormat.PARQUET;
     this.coalesce = coalesce != null ? coalesce : false;
-    this.location = location;
-    this.storageConnector = storageConnector;
+    if (dataSource == null) {
+      this.dataSource = new DataSource();
+      this.dataSource.setStorageConnector(storageConnector);
+      this.dataSource.setPath(location);
+    } else {
+      this.dataSource = dataSource;
+    }
     this.trainSplit = trainSplit;
     this.splits = splits == null ? Lists.newArrayList() : splits;
     this.seed = seed;
@@ -155,7 +160,7 @@ public TrainingDatasetBase(Integer version, String description, DataFormat dataF
     this.eventStartTime = eventStartTime != null ? FeatureGroupUtils.getDateFromDateString(eventStartTime) : null;
     this.eventEndTime = eventEndTime != null ? FeatureGroupUtils.getDateFromDateString(eventEndTime) : null;
     this.trainingDatasetType = trainingDatasetType != null ? trainingDatasetType :
-        getTrainingDatasetType(storageConnector);
+        getTrainingDatasetType(dataSource);
     setValTestSplit(validationSize, testSize);
     setTimeSeriesSplits(timeSplitSize, trainStart, trainEnd, validationStart, validationEnd, testStart, testEnd);
     if (extraFilter != null) {
@@ -225,10 +230,10 @@ public void setLabel(List<String> label) {
     this.label = label.stream().map(String::toLowerCase).collect(Collectors.toList());
   }
 
-  public TrainingDatasetType getTrainingDatasetType(StorageConnector storageConnector) {
-    if (storageConnector == null) {
+  public TrainingDatasetType getTrainingDatasetType(DataSource dataSource) {
+    if (dataSource == null || dataSource.getStorageConnector() == null) {
       return TrainingDatasetType.HOPSFS_TRAINING_DATASET;
-    } else if (storageConnector.getStorageConnectorType() == StorageConnectorType.HOPSFS) {
+    } else if (dataSource.getStorageConnector().getStorageConnectorType() == StorageConnectorType.HOPSFS) {
       return TrainingDatasetType.HOPSFS_TRAINING_DATASET;
     } else {
       return TrainingDatasetType.EXTERNAL_TRAINING_DATASET;

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

@@ -88,7 +88,6 @@ public ExternalFeatureGroup(FeatureStore featureStore, @NonNull String name, Int
     this.description = description;
     this.primaryKeys = primaryKeys != null
         ? primaryKeys.stream().map(String::toLowerCase).collect(Collectors.toList()) : null;
-    this.storageConnector = storageConnector;
     this.features = features;
     this.statisticsConfig = statisticsConfig != null ? statisticsConfig : new StatisticsConfig();
     this.eventTime = eventTime;
@@ -97,6 +96,7 @@ public ExternalFeatureGroup(FeatureStore featureStore, @NonNull String name, Int
     this.topicName = topicName;
     this.notificationTopicName = notificationTopicName;
     this.onlineConfig = onlineConfig;
+    this.dataSource.setStorageConnector(storageConnector);
     this.dataSource.setPath(path);
     this.dataSource.setQuery(query);
   }

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

@@ -88,7 +88,7 @@ public FeatureGroup(FeatureStore featureStore, @NonNull String name, Integer ver
     this.notificationTopicName = notificationTopicName;
     this.eventTime = eventTime;
     this.onlineConfig = onlineConfig;
-    this.storageConnector = storageConnector;
+    this.dataSource.setStorageConnector(storageConnector);
     this.dataSource.setPath(path);
   }