fix: remove references to custom json, float essential apis to top

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

@@ -264,14 +264,14 @@ object Builders {
         online: Boolean = false,
         production: Boolean = false,
         customJson: String = null,
-        dependencies: Seq[String] = null,
         namespace: String = null,
         team: String = null,
         samplePercent: Double = 100,
         consistencySamplePercent: Double = 5,
         tableProperties: Map[String, String] = Map.empty,
         historicalBackfill: Boolean = true,
-        driftSpec: DriftSpec = null
+        driftSpec: DriftSpec = null,
+        additionalOutputPartitionColumns: Seq[String] = Seq.empty
     ): MetaData = {
       val result = new MetaData()
       result.setName(name)
@@ -298,6 +298,11 @@ object Builders {
         result.setTableProperties(tableProperties.toJava)
       if (driftSpec != null)
         result.setDriftSpec(driftSpec)
+
+      if (additionalOutputPartitionColumns.nonEmpty) {
+        result.setAdditionalOutputPartitionColumns(additionalOutputPartitionColumns.toJava)
+      }
+
       result
     }
   }

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

@@ -154,20 +154,6 @@ object Extensions {
     @deprecated("Use `name` instead.")
     def nameToFilePath: String = metaData.name.replaceFirst("\\.", "/")
 
-    // helper function to extract values from customJson
-    def customJsonLookUp(key: String): Any = {
-      if (metaData.customJson == null) return null
-      val mapper = new ObjectMapper()
-      val typeRef = new TypeReference[java.util.HashMap[String, Object]]() {}
-      val jMap: java.util.Map[String, Object] = mapper.readValue(metaData.customJson, typeRef)
-      jMap.toScala.get(key).orNull
-    }
-
-    def owningTeam: String = {
-      val teamOverride = Try(customJsonLookUp(Constants.TeamOverride).asInstanceOf[String]).toOption
-      teamOverride.getOrElse(metaData.team)
-    }
-
     // if drift spec is set but tile size is not set, default to 30 minutes
     def driftTileSize: Option[Window] = {
       Option(metaData.getDriftSpec) match {
@@ -460,13 +446,6 @@ object Extensions {
       }
     }
 
-    // Check if tiling is enabled for a given GroupBy. Defaults to false if the 'enable_tiling' flag isn't set.
-    def isTilingEnabled: Boolean =
-      groupBy.getMetaData.customJsonLookUp("enable_tiling") match {
-        case s: Boolean => s
-        case _          => false
-      }
-
     def semanticHash: String = {
       val newGroupBy = groupBy.deepCopy()
       newGroupBy.unsetMetaData()

api/src/test/scala/ai/chronon/api/test/ExtensionsTest.scala

@@ -16,17 +16,11 @@
 
 package ai.chronon.api.test
 
-import ai.chronon.api.Accuracy
-import ai.chronon.api.Builders
-import ai.chronon.api.Constants
 import ai.chronon.api.Extensions._
-import ai.chronon.api.GroupBy
 import ai.chronon.api.ScalaJavaConversions._
-import org.junit.Assert.assertEquals
-import org.junit.Assert.assertFalse
-import org.junit.Assert.assertTrue
-import org.mockito.Mockito.spy
-import org.mockito.Mockito.when
+import ai.chronon.api.{Accuracy, Builders, Constants, GroupBy}
+import org.junit.Assert.{assertEquals, assertTrue}
+import org.mockito.Mockito.{spy, when}
 import org.scalatest.flatspec.AnyFlatSpec
 
 import java.util.Arrays
@@ -41,24 +35,6 @@ class ExtensionsTest extends AnyFlatSpec {
     )
   }
 
-  it should "owning team" in {
-    val metadata =
-      Builders.MetaData(
-        customJson = "{\"check_consistency\": true, \"lag\": 0, \"team_override\": \"ml_infra\"}",
-        team = "chronon"
-      )
-
-    assertEquals(
-      "ml_infra",
-      metadata.owningTeam
-    )
-
-    assertEquals(
-      "chronon",
-      metadata.team
-    )
-  }
-
   it should "row identifier" in {
     val labelPart = Builders.LabelPart();
     val res = labelPart.rowIdentifier(Arrays.asList("yoyo", "yujia"), "ds")
@@ -140,19 +116,4 @@ class ExtensionsTest extends AnyFlatSpec {
     assertTrue(keys.contains(Constants.TimeColumn))
     assertEquals(4, keys.size)
   }
-
-  it should "is tiling enabled" in {
-    def buildGroupByWithCustomJson(customJson: String = null): GroupBy =
-      Builders.GroupBy(
-        metaData = Builders.MetaData(name = "featureGroupName", customJson = customJson)
-      )
-
-    // customJson not set defaults to false
-    assertFalse(buildGroupByWithCustomJson().isTilingEnabled)
-    assertFalse(buildGroupByWithCustomJson("{}").isTilingEnabled)
-
-    assertTrue(buildGroupByWithCustomJson("{\"enable_tiling\": true}").isTilingEnabled)
-    assertFalse(buildGroupByWithCustomJson("{\"enable_tiling\": false}").isTilingEnabled)
-    assertFalse(buildGroupByWithCustomJson("{\"enable_tiling\": \"string instead of bool\"}").isTilingEnabled)
-  }
 }

api/thrift/api.thrift

@@ -249,13 +249,19 @@ struct MetaData {
 
     4: optional string outputNamespace
 
-    5: optional map<string, string> tableProperties
+    /**
+    * By default we will just partition the output by the date column - set via "spark.chronon.partition.column"
+    * With this we will partition the output with the specified additional columns
+    **/
+    5: optional list<string> additionalOutputPartitionColumns
+
+    6: optional map<string, string> tableProperties
 
     // tag_key -> tag_value - tags allow for repository wide querying, deprecations etc
     // this is object level tag - applies to all columns produced by the object - GroupBy, Join, Model etc
-    6: optional map<string, string> tags
+    20: optional map<string, string> tags
     // column -> tag_key -> tag_value
-    7: optional map<string, map<string, string>> columnTags
+    21: optional map<string, map<string, string>> columnTags
 
     // marking this as true means that the conf can be served online
     // once marked online, a conf cannot be changed - compiling the conf won't be allowed
@@ -286,8 +292,6 @@ struct MetaData {
     204: optional common.ExecutionInfo executionInfo
 }
 
-
-
 // Equivalent to a FeatureSet in chronon terms
 struct GroupBy {
     1: optional MetaData metaData

flink/src/main/scala/ai/chronon/flink/window/FlinkRowAggregators.scala

@@ -5,7 +5,7 @@ import ai.chronon.api.Constants
 import ai.chronon.api.DataType
 import ai.chronon.api.GroupBy
 import ai.chronon.api.Row
-import ai.chronon.api.ScalaJavaConversions.ListOps
+import ai.chronon.api.ScalaJavaConversions.{IteratorOps, ListOps}
 import ai.chronon.flink.types.TimestampedIR
 import ai.chronon.flink.types.TimestampedTile
 import ai.chronon.online.TileCodec
@@ -42,6 +42,23 @@ class FlinkRowAggregationFunction(
   private val valueColumns: Array[String] = inputSchema.map(_._1).toArray // column order matters
   private val timeColumnAlias: String = Constants.TimeColumn
 
+  private val isMutation: Boolean = {
+    groupBy.getSources
+      .iterator()
+      .toScala
+      .exists(source => source.isSetEntities && source.getEntities.isSetMutationTopic)
+  }
+
+  private val reversalIndex = {
+    val result = inputSchema.indexWhere(_._1 == Constants.ReversalColumn)
+
+    if (isMutation)
+      require(result >= 0,
+              s"Please specify source.query.reversal_column for CDC sources, only found, ${inputSchema.map(_._1)}")
+
+    result
+  }
+
   /*
    * Initialize the transient rowAggregator.
    * Running this method is an idempotent operation:
@@ -60,6 +77,7 @@ class FlinkRowAggregationFunction(
       element: Map[String, Any],
       accumulatorIr: TimestampedIR
   ): TimestampedIR = {
+
     // Most times, the time column is a Long, but it could be a Double.
     val tsMills = Try(element(timeColumnAlias).asInstanceOf[Long])
       .getOrElse(element(timeColumnAlias).asInstanceOf[Double].toLong)
@@ -79,7 +97,14 @@ class FlinkRowAggregationFunction(
     )
 
     val partialAggregates = Try {
-      rowAggregator.update(accumulatorIr.ir, row)
+      val isDelete = isMutation && row.getAs[Boolean](reversalIndex)
+
+      if (isDelete) {
+        rowAggregator.delete(accumulatorIr.ir, row)
+      } else {
+        rowAggregator.update(accumulatorIr.ir, row)
+      }
+
     }
 
     partialAggregates match {

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

@@ -89,7 +89,6 @@ class GroupByServingInfoParsed(val groupByServingInfo: GroupByServingInfo, parti
   // Start tiling specific variables
 
   lazy val tiledCodec: TileCodec = new TileCodec(groupBy, valueChrononSchema.fields.map(sf => (sf.name, sf.fieldType)))
-  lazy val isTilingEnabled: Boolean = groupByOps.isTilingEnabled
 
   // End tiling specific variables
 

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

@@ -25,22 +25,12 @@ object MetadataEndPoint {
   val ConfByKeyEndPointName = "CHRONON_METADATA"
   val NameByTeamEndPointName = "CHRONON_ENTITY_BY_TEAM"
 
-  private def getTeamFromMetadata(metaData: MetaData): String = {
-    val team = metaData.team
-    if (metaData.customJson != null && metaData.customJson.nonEmpty) {
-      implicit val formats = DefaultFormats
-      val customJson = parse(metaData.customJson)
-      val teamFromJson: String = (customJson \ "team_override").extractOpt[String].getOrElse("")
-      if (teamFromJson.nonEmpty) teamFromJson else team
-    } else team
-  }
-
   private def parseTeam[Conf <: TBase[_, _]: Manifest: ClassTag](conf: Conf): String = {
     conf match {
-      case join: Join                 => "joins/" + getTeamFromMetadata(join.metaData)
-      case groupBy: GroupBy           => "group_bys/" + getTeamFromMetadata(groupBy.metaData)
-      case stagingQuery: StagingQuery => "staging_queries/" + getTeamFromMetadata(stagingQuery.metaData)
-      case model: Model               => "models/" + getTeamFromMetadata(model.metaData)
+      case join: Join                 => "joins/" + join.metaData.team
+      case groupBy: GroupBy           => "group_bys/" + groupBy.metaData.team
+      case stagingQuery: StagingQuery => "staging_queries/" + stagingQuery.metaData.team
+      case model: Model               => "models/" + model.metaData.team
       case _ =>
         logger.error(s"Failed to parse team from $conf")
         throw new Exception(s"Failed to parse team from $conf")

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

@@ -98,7 +98,7 @@ object Metrics {
         environment = environment,
         join = join.metaData.cleanName,
         production = join.metaData.isProduction,
-        team = join.metaData.owningTeam
+        team = join.metaData.team
       )
     }
 
@@ -108,7 +108,7 @@ object Metrics {
         groupBy = groupBy.metaData.cleanName,
         production = groupBy.metaData.isProduction,
         accuracy = groupBy.inferredAccuracy,
-        team = groupBy.metaData.owningTeam,
+        team = groupBy.metaData.team,
         join = groupBy.sources.toScala
           .find(_.isSetJoinSource)
           .map(_.getJoinSource.join.metaData.cleanName)
@@ -127,7 +127,7 @@ object Metrics {
         environment = environment,
         groupBy = stagingQuery.metaData.cleanName,
         production = stagingQuery.metaData.isProduction,
-        team = stagingQuery.metaData.owningTeam
+        team = stagingQuery.metaData.team
       )
     }
 

online/src/main/scala/ai/chronon/online/fetcher/FetchContext.scala

@@ -13,18 +13,6 @@ case class FetchContext(kvStore: KVStore,
                         disableErrorThrows: Boolean = false,
                         executionContextOverride: ExecutionContext = null) {
 
-  def isTilingEnabled: Boolean = {
-    Option(flagStore)
-      .map(_.isSet(FlagStoreConstants.TILING_ENABLED, Map.empty[String, String].toJava))
-      .exists(_.asInstanceOf[Boolean])
-  }
-
-  def shouldStreamingDecodeThrow(groupByName: String): Boolean = {
-    Option(flagStore)
-      .exists(
-        _.isSet("disable_streaming_decoding_error_throws", Map("group_by_streaming_dataset" -> groupByName).toJava))
-  }
-
   def getOrCreateExecutionContext: ExecutionContext = {
     Option(executionContextOverride).getOrElse(FlexibleExecutionContext.buildExecutionContext)
   }

online/src/main/scala/ai/chronon/online/fetcher/GroupByFetcher.scala

@@ -86,7 +86,7 @@ class GroupByFetcher(fetchContext: FetchContext, metadataStore: MetadataStore)
         case Accuracy.TEMPORAL =>
           // Build a tile key for the streaming request
           // When we build support for layering, we can expand this out into a utility that builds n tile keys for n layers
-          val keyBytes = if (fetchContext.isTilingEnabled) {
+          val keyBytes = {
 
             val tileKey = TilingUtils.buildTileKey(
               groupByServingInfo.groupByOps.streamingDataset,
@@ -96,8 +96,6 @@ class GroupByFetcher(fetchContext: FetchContext, metadataStore: MetadataStore)
             )
 
             TilingUtils.serializeTileKey(tileKey)
-          } else {
-            streamingKeyBytes
           }
 
           Some(