[Do not merge] Generic Serializer for Incremental Aggregation

aggregator/src/main/scala/ai/chronon/aggregator/row/RowAggregator.scala

@@ -70,6 +70,11 @@ class RowAggregator(val inputSchema: Seq[(String, DataType)], val aggregationPar
     .toArray
     .zip(columnAggregators.map(_.irType))
 
+  val incrementalOutputSchema = aggregationParts
+    .map(_.incrementalOutputColumnName)
+    .toArray
+    .zip(columnAggregators.map(_.irType))
+
   val outputSchema: Array[(String, DataType)] = aggregationParts
     .map(_.outputColumnName)
     .toArray

aggregator/src/test/scala/ai/chronon/aggregator/test/SawtoothOnlineAggregatorTest.scala

@@ -143,7 +143,7 @@ class SawtoothOnlineAggregatorTest extends TestCase {
         operation = Operation.HISTOGRAM,
         inputColumn = "action",
         windows = Seq(
-          new Window(3, TimeUnit.DAYS),
+          new Window(3, TimeUnit.DAYS)
         )
       )
     )
@@ -162,15 +162,15 @@ class SawtoothOnlineAggregatorTest extends TestCase {
 
     val finalBatchIr = FinalBatchIr(
       Array[Any](
-        null,                       // collapsed (T-1 -> T)
+        null // collapsed (T-1 -> T)
       ),
       Array(
-        Array.empty,                // 1‑day hops (not used)
-        Array(                      // 1-hour hops
-          hop(1, 1746745200000L),   // 2025-05-08 23:00:00 UTC
-          hop(1, 1746766800000L),   // 2025-05-09 05:00:00 UTC
+        Array.empty, // 1‑day hops (not used)
+        Array( // 1-hour hops
+              hop(1, 1746745200000L), // 2025-05-08 23:00:00 UTC
+              hop(1, 1746766800000L) // 2025-05-09 05:00:00 UTC
         ),
-        Array.empty                  // 5‑minute hops (not used)
+        Array.empty // 5‑minute hops (not used)
       )
     )
     val queryTs = batchEndTs + 100

api/py/ai/chronon/group_by.py

@@ -362,7 +362,7 @@ def GroupBy(
     tags: Optional[Dict[str, str]] = None,
     derivations: Optional[List[ttypes.Derivation]] = None,
     deprecation_date: Optional[str] = None,
-    description: Optional[str] = None,
+    is_incremental: Optional[bool] = False,
     **kwargs,
 ) -> ttypes.GroupBy:
     """
@@ -570,6 +570,7 @@ def _normalize_source(source):
         backfillStartDate=backfill_start_date,
         accuracy=accuracy,
         derivations=derivations,
+        isIncremental=is_incremental,
     )
     validate_group_by(group_by)
     return group_by

api/py/test/sample/scripts/spark_submit.sh

@@ -28,13 +28,14 @@
 
 set -euxo pipefail
 CHRONON_WORKING_DIR=${CHRONON_TMPDIR:-/tmp}/${USER}
+echo $CHRONON_WORKING_DIR
 mkdir -p ${CHRONON_WORKING_DIR}
 export TEST_NAME="${APP_NAME}_${USER}_test"
 unset PYSPARK_DRIVER_PYTHON
 unset PYSPARK_PYTHON
 unset SPARK_HOME
 unset SPARK_CONF_DIR
-export LOG4J_FILE="${CHRONON_WORKING_DIR}/log4j_file"
+export LOG4J_FILE="${CHRONON_WORKING_DIR}/log4j.properties"
 cat > ${LOG4J_FILE} << EOF
 log4j.rootLogger=INFO, stdout
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
@@ -47,6 +48,9 @@ EOF
 $SPARK_SUBMIT_PATH \
 --driver-java-options " -Dlog4j.configuration=file:${LOG4J_FILE}" \
 --conf "spark.executor.extraJavaOptions= -XX:ParallelGCThreads=4 -XX:+UseParallelGC -XX:+UseCompressedOops" \
+--conf "spark.driver.extraJavaOptions=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005 -Dlog4j.configuration=file:${LOG4J_FILE}" \
+--conf "spark.sql.warehouse.dir=/home/chaitu/projects/chronon/spark-warehouse" \
+--conf "javax.jdo.option.ConnectionURL=jdbc:derby:;databaseName=/home/chaitu/projects/chronon/hive-metastore/metastore_db;create=true" \
 --conf spark.sql.shuffle.partitions=${PARALLELISM:-4000} \
 --conf spark.dynamicAllocation.maxExecutors=${MAX_EXECUTORS:-1000} \
 --conf spark.default.parallelism=${PARALLELISM:-4000} \
@@ -77,3 +81,6 @@ tee ${CHRONON_WORKING_DIR}/${APP_NAME}_spark.log
 
 
 
+
+#--conf "spark.driver.extraJavaOptions=-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=5005 -Dlog4j.rootLogger=INFO,console" \
+

api/py/test/sample/teams.json

@@ -5,7 +5,7 @@
         },
         "common_env": {
             "VERSION": "latest",
-            "SPARK_SUBMIT_PATH": "[TODO]/path/to/spark-submit",
+            "SPARK_SUBMIT_PATH": "spark-submit",
             "JOB_MODE": "local[*]",
             "HADOOP_DIR": "[STREAMING-TODO]/path/to/folder/containing",
             "CHRONON_ONLINE_CLASS": "[ONLINE-TODO]your.online.class",

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

@@ -98,7 +98,7 @@ object Extensions {
     def cleanName: String = metaData.name.sanitize
 
     def outputTable = s"${metaData.outputNamespace}.${metaData.cleanName}"
-
+    def incrementalOutputTable = s"${metaData.outputNamespace}.${metaData.cleanName}_inc"
     def preModelTransformsTable = s"${metaData.outputNamespace}.${metaData.cleanName}_pre_mt"
     def outputLabelTable = s"${metaData.outputNamespace}.${metaData.cleanName}_labels"
     def outputFinalView = s"${metaData.outputNamespace}.${metaData.cleanName}_labeled"
@@ -179,6 +179,10 @@ object Extensions {
 
     def outputColumnName =
       s"${aggregationPart.inputColumn}_$opSuffix${aggregationPart.window.suffix}${bucketSuffix}"
+
+    def incrementalOutputColumnName =
+      s"${aggregationPart.inputColumn}_$opSuffix${bucketSuffix}"
+
   }
 
   implicit class AggregationOps(aggregation: Aggregation) {

api/thrift/api.thrift

@@ -287,6 +287,7 @@ struct GroupBy {
     6: optional string backfillStartDate
     // Optional derivation list
     7: optional list<Derivation> derivations
+    8: optional bool isIncremental
 }
 
 struct JoinPart {

online/src/main/scala/ai/chronon/online/SparkConversions.scala

@@ -163,4 +163,78 @@ object SparkConversions {
       extraneousRecord
     )
   }
+
+  /**
+    * Converts a single Spark column value to Chronon normalized IR format.
+    *
+    * This is the inverse of toSparkRow() - used when reading pre-computed IR values
+    * from Spark DataFrames. Each IR column in the DataFrame is converted based on its
+    * Chronon IR type.
+    *
+    * Examples:
+    * - Count IR: Long → Long (pass-through, primitives stay primitives)
+    * - Sum IR: Double → Double (pass-through)
+    * - Average IR: Spark Row(sum, count) → Array[Any](sum, count)
+    * - UniqueCount IR: Spark Array[T] → java.util.ArrayList[T]
+    * - Histogram IR: Spark Map[K,V] → java.util.HashMap[K,V]
+    * - ApproxPercentile IR: Array[Byte] → Array[Byte] (pass-through for binary)
+    *
+    * @param sparkValue The value from a Spark DataFrame column
+    * @param irType The Chronon IR type for this column (from RowAggregator.incrementalOutputSchema)
+    * @return Normalized IR value ready for denormalize()
+    */
+  def fromSparkValue(sparkValue: Any, irType: api.DataType): Any = {
+    if (sparkValue == null) return null
+
+    (sparkValue, irType) match {
+      // Primitives - pass through (Count, Sum, Min, Max, Binary sketches)
+      case (v,
+            api.IntType | api.LongType | api.ShortType | api.ByteType | api.FloatType | api.DoubleType |
+            api.StringType | api.BooleanType | api.BinaryType) =>
+        v
+
+      // Spark Row → Array[Any] (Average, Variance, Skew, Kurtosis, FirstK/LastK)
+      case (row: Row, api.StructType(_, fields)) =>
+        val arr = new Array[Any](fields.length)
+        fields.zipWithIndex.foreach {
+          case (field, idx) =>
+            arr(idx) = fromSparkValue(row.get(idx), field.fieldType)
+        }
+        arr
+
+      // Spark mutable.WrappedArray → util.ArrayList (UniqueCount, TopK, BottomK)
+      case (arr: mutable.WrappedArray[_], api.ListType(elementType)) =>
+        val result = new util.ArrayList[Any](arr.length)
+        arr.foreach { elem =>
+          result.add(fromSparkValue(elem, elementType))
+        }
+        result
+
+      // Spark native Array → util.ArrayList (alternative array representation)
+      case (arr: Array[_], api.ListType(elementType)) =>
+        val result = new util.ArrayList[Any](arr.length)
+        arr.foreach { elem =>
+          result.add(fromSparkValue(elem, elementType))
+        }
+        result
+
+      // Spark scala.collection.Map → util.HashMap (Histogram)
+      case (map: scala.collection.Map[_, _], api.MapType(keyType, valueType)) =>
+        val result = new util.HashMap[Any, Any]()
+        map.foreach {
+          case (k, v) =>
+            result.put(
+              fromSparkValue(k, keyType),
+              fromSparkValue(v, valueType)
+            )
+        }
+        result
+
+      case (value, tpe) =>
+        throw new IllegalArgumentException(
+          s"Cannot convert Spark value $value (${value.getClass.getSimpleName}) " +
+            s"to Chronon IR type $tpe"
+        )
+    }
+  }
 }

spark/src/main/scala/ai/chronon/spark/DataRange.scala

@@ -53,6 +53,11 @@ case class PartitionRange(start: String, end: String)(implicit tableUtils: Table
     }
   }
 
+  def daysBetween: Int = {
+    if (start == null || end == null) 0
+    else Stream.iterate(start)(tableUtils.partitionSpec.after).takeWhile(_ <= end).size
+  }
+
   def isSingleDay: Boolean = {
     start == end
   }

spark/src/main/scala/ai/chronon/spark/Driver.scala

@@ -466,7 +466,8 @@ object Driver {
         tableUtils,
         args.stepDays.toOption,
         args.startPartitionOverride.toOption,
-        !args.runFirstHole()
+        !args.runFirstHole(),
+        args.groupByConf.isIncremental
       )
 
       if (args.shouldExport()) {