perf: Online + Avro path optimizations (#655)

## Summary
guided by flamegraph from one of our customers
- pre-compute window.millis
- eliminate branching in AvroConversions.toChrononRow - by pre-building
a row generator func `Row.fromCached`
- optimize `expandWindowedTileIr` by pre-computing the expander indexes.
- bring-in fast-serde from linkedin which does generic data decoding, by
pre-generating decoder classes.
- optimize FetcherCache.getBatchBytes logic
- early exit when externalParts are null
- bigtablekvstore impl optimizations on how we construct the keys

## Checklist
- [ ] Added Unit Tests
- [x] Covered by existing CI
- [ ] Integration tested
- [ ] Documentation update



<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
  - Added a new method for converting cached data to Chronon row format.
- Introduced a public field to expose the smallest tail hop duration in
milliseconds for group-by serving information.

- **Bug Fixes**
- Improved error handling and null safety in batch response processing
and fetcher logic.
- Enhanced logging with better exception handling and sampling in online
fetcher responses.
- Added exception logging around synchronous KVStore response retrieval.

- **Refactor**
- Optimized windowing and mapping logic for performance and clarity in
several aggregation and expansion methods.
- Modularized derivation and logging logic in the fetcher for better
maintainability.
- Updated method and field names for clarity regarding window resolution
and tail hop durations.
- Refactored Avro serialization/deserialization to use LinkedIn Fast
Avro SerDe with cached conversion functions.
- Improved conversion utilities for Avro to Chronon row representations.
- Centralized Scala-to-Java byte array conversions in tiling utilities.

- **Chores**
- Added LinkedIn Fast Avro SerDe as a dependency and updated related
build and dependency files.
  - Updated utility methods for more consistent data type conversions.

- **Style**
  - Improved code comments and documentation for clarity.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: nikhil-zlai

aggregator/src/main/scala/ai/chronon/aggregator/windowing/Resolution.scala

@@ -67,12 +67,20 @@ object ResolutionUtils {
   /** Find the smallest tail window resolution in a GroupBy. Returns 1D if the GroupBy does not define any windows (all-time aggregates).
     * The window resolutions are: 5 min for a GroupBy a window < 12 hrs, 1 hr for < 12 days, 1 day for > 12 days.
     */
-  def getSmallestWindowResolutionInMillis(groupBy: GroupBy): Long =
-    Option(
-      groupBy.aggregations.toScala.toArray
-        .flatMap(aggregation =>
-          if (aggregation.windows != null) aggregation.windows.toScala
-          else None)
-        .map(FiveMinuteResolution.calculateTailHop)
-    ).filter(_.nonEmpty).map(_.min).getOrElse(WindowUtils.Day.millis)
+  def getSmallestTailHopMillis(groupBy: GroupBy): Long = {
+
+    val tailHops =
+      for (
+        aggs <- Option(groupBy.aggregations).toSeq;
+        agg <- aggs.iterator().toScala;
+        windows <- Option(agg.windows).toSeq;
+        window <- windows.iterator().toScala
+      ) yield {
+        FiveMinuteResolution.calculateTailHop(window)
+      }
+
+    if (tailHops.isEmpty) WindowUtils.Day.millis
+    else tailHops.min
+
+  }
 }

aggregator/src/main/scala/ai/chronon/aggregator/windowing/SawtoothMutationAggregator.scala

@@ -31,7 +31,7 @@ case class FinalBatchIr(collapsed: Array[Any], tailHops: HopsAggregator.OutputAr
   *
   * update/merge/finalize are related to snapshot data. As such they follow the snapshot Schema
   * and aggregators.
-  * However mutations come into play later in the group by and a finalized version of the snapshot
+  * However, mutations come into play later in the group by and a finalized version of the snapshot
   * data is created to be processed with the mutations rows.
   * Since the dataframe inputs are aligned between mutations and snapshot (input) no additional schema is needed.
   */
@@ -109,9 +109,10 @@ class SawtoothMutationAggregator(aggregations: Seq[Aggregation],
   def updateIr(ir: Array[Any], row: Row, queryTs: Long, hasReversal: Boolean = false): Unit = {
     var i: Int = 0
     while (i < windowedAggregator.length) {
+      val windowMillis = windowMappings(i).millis
       val window = windowMappings(i).aggregationPart.window
       val hopIndex = tailHopIndices(i)
-      val rowInWindow = (row.ts >= TsUtils.round(queryTs - window.millis, hopSizes(hopIndex)) && row.ts < queryTs)
+      val rowInWindow = (row.ts >= TsUtils.round(queryTs - windowMillis, hopSizes(hopIndex)) && row.ts < queryTs)
       if (window == null || rowInWindow) {
         if (hasReversal && row.isBefore) {
           windowedAggregator(i).delete(ir, row)
@@ -127,10 +128,11 @@ class SawtoothMutationAggregator(aggregations: Seq[Aggregation],
     val otherIrTs = otherIr.ts
     var i: Int = 0
     while (i < windowedAggregator.length) {
+      val windowMillis = windowMappings(i).millis
       val window = windowMappings(i).aggregationPart.window
       val hopIndex = tailHopIndices(i)
-      val irInWindow =
-        (otherIrTs >= TsUtils.round(queryTs - window.millis, hopSizes(hopIndex)) && otherIrTs < queryTs)
+      lazy val irInWindow =
+        (otherIrTs >= TsUtils.round(queryTs - windowMillis, hopSizes(hopIndex)) && otherIrTs < queryTs)
       if (window == null || irInWindow) {
         ir(i) = windowedAggregator(i).merge(ir(i), otherIr.ir(i))
       }
@@ -143,17 +145,18 @@ class SawtoothMutationAggregator(aggregations: Seq[Aggregation],
   def mergeTailHops(ir: Array[Any], queryTs: Long, batchEndTs: Long, batchIr: FinalBatchIr): Array[Any] = {
     var i: Int = 0
     while (i < windowedAggregator.length) {
+      val windowMillis = windowMappings(i).millis
       val window = windowMappings(i).aggregationPart.window
       if (window != null) { // no hops for unwindowed
         val hopIndex = tailHopIndices(i)
-        val queryTail = TsUtils.round(queryTs - window.millis, hopSizes(hopIndex))
+        val queryTail = TsUtils.round(queryTs - windowMillis, hopSizes(hopIndex))
         val hopIrs = batchIr.tailHops(hopIndex)
         val relevantHops = mutable.ArrayBuffer[Any](ir(i))
         var idx: Int = 0
         while (idx < hopIrs.length) {
           val hopIr = hopIrs(idx)
           val hopStart = hopIr.last.asInstanceOf[Long]
-          if ((batchEndTs - window.millis) + tailBufferMillis > hopStart && hopStart >= queryTail) {
+          if ((batchEndTs - windowMillis) + tailBufferMillis > hopStart && hopStart >= queryTail) {
             relevantHops += hopIr(baseIrIndices(i))
           }
           idx += 1

api/src/main/scala/ai/chronon/api/Extensions.scala

@@ -256,7 +256,7 @@ object Extensions {
     }
   }
 
-  case class WindowMapping(aggregationPart: AggregationPart, baseIrIndex: Int)
+  case class WindowMapping(aggregationPart: AggregationPart, baseIrIndex: Int, millis: Long)
 
   case class UnpackedAggregations(perBucket: Array[AggregationPart], perWindow: Array[WindowMapping])
 
@@ -295,7 +295,8 @@ object Extensions {
                                          )
                                          .orNull,
                                        bucket),
-              counter
+              counter,
+              if (window != null) window.millis else -1
             )
           }
           counter += 1

api/src/main/scala/ai/chronon/api/Row.scala

@@ -123,6 +123,85 @@ object Row {
     }
   }
 
+  private val passThroughFunc: Any => Any = { value: Any => value }
+
+  // recursively traverse a logical struct, and convert it chronon's row type
+  def fromCached[CompositeType, BinaryType, ArrayType, StringType](dataType: DataType,
+                                                                   decomposer: (CompositeType, Int) => Iterator[Any],
+                                                                   debinarizer: BinaryType => Array[Byte],
+                                                                   delister: ArrayType => util.ArrayList[Any],
+                                                                   deStringer: StringType => String): Any => Any = {
+
+    def edit(dataType: DataType): Any => Any =
+      fromCached(dataType, decomposer, debinarizer, delister, deStringer)
+
+    def guard(func: Any => Any): Any => Any = { value =>
+      if (value == null) value else func(value)
+    }
+
+    val baseFunc: Any => Any = dataType match {
+      case StructType(_, fields) =>
+        val length = fields.length
+        val funcs: Array[Any => Any] = fields.map(_.fieldType).map(edit)
+
+        guard { value: Any =>
+          val iter = decomposer(value.asInstanceOf[CompositeType], length)
+
+          val newArr = new Array[Any](length)
+          var idx = 0
+          while (iter.hasNext) {
+            val value = iter.next()
+            newArr.update(idx, funcs(idx)(value))
+            idx += 1
+          }
+          newArr
+        }
+
+      case ListType(elemType) =>
+        val func = edit(elemType)
+
+        guard { value: Any =>
+          val arr = delister(value.asInstanceOf[ArrayType])
+
+          if (func != passThroughFunc) {
+            var idx = 0
+            while (idx < arr.size) {
+              arr.set(idx, func(arr.get(idx)))
+              idx += 1
+            }
+          }
+
+          arr
+
+        }
+
+      case MapType(keyType, valueType) =>
+        val keyFunc = edit(keyType)
+        val valueFunc = edit(valueType)
+
+        guard { value: Any =>
+          val newMap = new util.HashMap[Any, Any]()
+          val map = value.asInstanceOf[util.Map[Any, Any]]
+          val iter = map.entrySet().iterator()
+          while (iter.hasNext) {
+            val entry = iter.next()
+            newMap.put(keyFunc(entry.getKey), valueFunc(entry.getValue))
+          }
+          newMap
+        }
+
+      case BinaryType =>
+        guard { value: Any =>
+          debinarizer(value.asInstanceOf[BinaryType])
+        }
+
+      case StringType => guard { value: Any => deStringer(value.asInstanceOf[StringType]) }
+      case _          => passThroughFunc
+    }
+
+    baseFunc
+  }
+
   // recursively traverse a chronon dataType value, and convert it to an external type
   def to[StructType, BinaryType, ListType, MapType, OutputSchema](
       value: Any,

api/src/main/scala/ai/chronon/api/TilingUtils.scala

@@ -2,6 +2,7 @@ package ai.chronon.api
 
 import ai.chronon.fetcher.TileKey
 
+import java.util
 import scala.jdk.CollectionConverters._
 
 // Convenience functions for working with tiling
@@ -17,13 +18,24 @@ object TilingUtils {
     key
   }
 
+  private def toList(arr: Array[Byte]): java.util.ArrayList[java.lang.Byte] = {
+    if (arr == null) return null
+    val result = new util.ArrayList[java.lang.Byte](arr.length)
+    var idx = 0
+    while (idx < arr.length) {
+      result.add(arr(idx))
+      idx += 1
+    }
+    result
+  }
+
   def buildTileKey(dataset: String,
                    keyBytes: Array[Byte],
                    tileSizeMs: Option[Long],
                    tileStartTs: Option[Long]): TileKey = {
     val tileKey = new TileKey()
     tileKey.setDataset(dataset)
-    tileKey.setKeyBytes(keyBytes.toList.asJava.asInstanceOf[java.util.List[java.lang.Byte]])
+    tileKey.setKeyBytes(toList(keyBytes))
     tileSizeMs.foreach(tileKey.setTileSizeMillis)
     tileStartTs.foreach(tileKey.setTileStartTimestampMillis)
     tileKey

flink/src/main/scala/ai/chronon/flink/FlinkJob.scala

@@ -96,7 +96,7 @@ class FlinkJob(eventSrc: FlinkSource[Map[String, Any]],
         "Tiling is enabled.")
 
     val tilingWindowSizeInMillis: Long =
-      ResolutionUtils.getSmallestWindowResolutionInMillis(groupByServingInfoParsed.groupBy)
+      ResolutionUtils.getSmallestTailHopMillis(groupByServingInfoParsed.groupBy)
 
     // we expect parallelism on the source stream to be set by the source provider
     val sourceSparkProjectedStream: DataStream[Map[String, Any]] =