refactor: fetcher sub package + kill old stats in fetcher (#423)

Author: nikhil-zlai

online/src/main/java/ai/chronon/online/JavaExternalSourceHandler.java

@@ -16,6 +16,7 @@
 
 package ai.chronon.online;
 
+import ai.chronon.online.fetcher.Fetcher;
 import scala.collection.Seq;
 import scala.compat.java8.FutureConverters;
 import scala.concurrent.Future;

online/src/main/java/ai/chronon/online/JavaFetcher.java

@@ -16,8 +16,8 @@
 
 package ai.chronon.online;
 
-import ai.chronon.online.Fetcher.Request;
-import ai.chronon.online.Fetcher.Response;
+import ai.chronon.online.fetcher.Fetcher;
+import ai.chronon.online.fetcher.FetcherResponseWithTs;
 import scala.collection.Iterator;
 import scala.collection.Seq;
 import scala.Option;
@@ -120,9 +120,9 @@ public JavaFetcher build() {
   }
 
 
-  public static List<JavaResponse> toJavaResponses(Seq<Response> responseSeq) {
+  public static List<JavaResponse> toJavaResponses(Seq<Fetcher.Response> responseSeq) {
     List<JavaResponse> result = new ArrayList<>(responseSeq.size());
-    Iterator<Response> it = responseSeq.iterator();
+    Iterator<Fetcher.Response> it = responseSeq.iterator();
     while (it.hasNext()) {
       result.add(new JavaResponse(it.next()));
     }
@@ -139,44 +139,21 @@ private CompletableFuture<List<JavaResponse>> convertResponsesWithTs(Future<Fetc
     });
   }
 
-  public static List<JavaStatsResponse> toJavaStatsResponses(Seq<Fetcher.StatsResponse> responseSeq) {
-    List<JavaStatsResponse> result = new ArrayList<>(responseSeq.size());
-    Iterator<Fetcher.StatsResponse> it = responseSeq.iterator();
-    while(it.hasNext()) {
-      result.add(toJavaStatsResponse(it.next()));
-    }
-    return result;
-  }
-
-  public static JavaStatsResponse toJavaStatsResponse(Fetcher.StatsResponse response) {
-    return new JavaStatsResponse(response);
-  }
-  public static JavaSeriesStatsResponse toJavaSeriesStatsResponse(Fetcher.SeriesStatsResponse response) {
-    return new JavaSeriesStatsResponse(response);
-  }
-
-  private CompletableFuture<List<JavaStatsResponse>> convertStatsResponses(Future<Seq<Fetcher.StatsResponse>> responses) {
-    return FutureConverters
-            .toJava(responses)
-            .toCompletableFuture()
-            .thenApply(JavaFetcher::toJavaStatsResponses);
-  }
-
-  private Seq<Request> convertJavaRequestList(List<JavaRequest> requests, boolean isGroupBy, long startTs) {
-    ArrayBuffer<Request> scalaRequests = new ArrayBuffer<>();
+  private Seq<Fetcher.Request> convertJavaRequestList(List<JavaRequest> requests, boolean isGroupBy, long startTs) {
+    ArrayBuffer<Fetcher.Request> scalaRequests = new ArrayBuffer<>();
     for (JavaRequest request : requests) {
-      Request convertedRequest = request.toScalaRequest();
+      Fetcher.Request convertedRequest = request.toScalaRequest();
       scalaRequests.$plus$eq(convertedRequest);
     }
-    Seq<Request> scalaRequestsSeq = scalaRequests.toSeq();
+    Seq<Fetcher.Request> scalaRequestsSeq = scalaRequests.toSeq();
     instrument(requests.stream().map(jReq -> jReq.name).collect(Collectors.toList()), isGroupBy, "java.request_conversion.latency.millis", startTs);
     return scalaRequestsSeq;
   }
 
   public CompletableFuture<List<JavaResponse>> fetchGroupBys(List<JavaRequest> requests) {
     long startTs = System.currentTimeMillis();
     // Convert java requests to scala requests
-    Seq<Request> scalaRequests = convertJavaRequestList(requests, true, startTs);
+    Seq<Fetcher.Request> scalaRequests = convertJavaRequestList(requests, true, startTs);
     // Get responses from the fetcher
     Future<FetcherResponseWithTs> scalaResponses = this.fetcher.withTs(this.fetcher.fetchGroupBys(scalaRequests));
     // Convert responses to CompletableFuture
@@ -186,7 +163,7 @@ public CompletableFuture<List<JavaResponse>> fetchGroupBys(List<JavaRequest> req
   public CompletableFuture<List<JavaResponse>> fetchJoin(List<JavaRequest> requests) {
     long startTs = System.currentTimeMillis();
     // Convert java requests to scala requests
-    Seq<Request> scalaRequests = convertJavaRequestList(requests, false, startTs);
+    Seq<Fetcher.Request> scalaRequests = convertJavaRequestList(requests, false, startTs);
     // Get responses from the fetcher
     Future<FetcherResponseWithTs> scalaResponses = this.fetcher.withTs(this.fetcher.fetchJoin(scalaRequests, Option.empty()));
     // Convert responses to CompletableFuture
@@ -213,10 +190,4 @@ private Metrics.Context getJoinContext(String joinName) {
   private Metrics.Context getGroupByContext(String groupByName) {
     return new Metrics.Context("group_by.fetch", null, groupByName, null, false, null, null, null, null);
   }
-
-  public CompletableFuture<JavaSeriesStatsResponse> fetchConsistencyMetricsTimeseries(JavaStatsRequest request) {
-    Future<Fetcher.SeriesStatsResponse> response = this.fetcher.fetchConsistencyMetricsTimeseries(request.toScalaRequest());
-    // Convert responses to CompletableFuture
-    return FutureConverters.toJava(response).toCompletableFuture().thenApply(JavaFetcher::toJavaSeriesStatsResponse);
-  }
 }

online/src/main/java/ai/chronon/online/JavaRequest.java

@@ -16,6 +16,7 @@
 
 package ai.chronon.online;
 
+import ai.chronon.online.fetcher.Fetcher;
 import scala.Option;
 import ai.chronon.api.ScalaJavaConversions;
 

online/src/main/java/ai/chronon/online/JavaResponse.java

@@ -17,6 +17,7 @@
 package ai.chronon.online;
 
 import ai.chronon.api.ScalaJavaConversions;
+import ai.chronon.online.fetcher.Fetcher;
 
 import java.util.Map;
 

online/src/main/java/ai/chronon/online/JavaSeriesStatsResponse.java

@@ -19,6 +19,7 @@ package ai.chronon.online
 import ai.chronon.api.Constants
 import ai.chronon.api.StructType
 import ai.chronon.online.KVStore._
+import ai.chronon.online.fetcher.Fetcher
 import org.apache.spark.sql.SparkSession
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
@@ -259,15 +260,17 @@ abstract class Api(userConf: Map[String, String]) extends Serializable {
   final def buildFetcher(debug: Boolean = false,
                          callerName: String = null,
                          disableErrorThrows: Boolean = false): Fetcher =
-    new Fetcher(genKvStore,
-                Constants.MetadataDataset,
-                logFunc = responseConsumer,
-                debug = debug,
-                externalSourceRegistry = externalRegistry,
-                timeoutMillis = timeoutMillis,
-                callerName = callerName,
-                flagStore = flagStore,
-                disableErrorThrows = disableErrorThrows)
+    new Fetcher(
+      genKvStore,
+      Constants.MetadataDataset,
+      logFunc = responseConsumer,
+      debug = debug,
+      externalSourceRegistry = externalRegistry,
+      timeoutMillis = timeoutMillis,
+      callerName = callerName,
+      flagStore = flagStore,
+      disableErrorThrows = disableErrorThrows
+    )
 
   final def buildJavaFetcher(callerName: String = null, disableErrorThrows: Boolean = false): JavaFetcher = {
     new JavaFetcher(genKvStore,

online/src/main/java/ai/chronon/online/JavaStatsRequest.java

@@ -17,8 +17,8 @@
 package ai.chronon.online
 
 import ai.chronon.api.Constants
-import ai.chronon.online.Fetcher.Request
-import ai.chronon.online.Fetcher.Response
+import ai.chronon.online.fetcher.Fetcher.Request
+import ai.chronon.online.fetcher.Fetcher.Response
 
 import scala.collection.Seq
 import scala.collection.mutable

online/src/main/java/ai/chronon/online/JavaStatsResponse.java

@@ -7,7 +7,7 @@ import ai.chronon.api.LongType
 import ai.chronon.api.StringType
 import ai.chronon.api.StructField
 import ai.chronon.api.StructType
-import ai.chronon.online.Fetcher.Request
+import ai.chronon.online.fetcher.Fetcher
 
 import scala.collection.Seq
 
@@ -64,7 +64,7 @@ object OnlineDerivationUtil {
 
   def applyDeriveFunc(
       deriveFunc: DerivationFunc,
-      request: Request,
+      request: Fetcher.Request,
       baseMap: Map[String, AnyRef]
   ): Map[String, AnyRef] = {
     val requestTs = request.atMillis.getOrElse(System.currentTimeMillis())