Write daily IRs for incremental aggregation by kambstreat · Pull Request #998 · airbnb/chronon

kambstreat · 2025-06-03T18:42:31Z

Summary

This PR address the following CHIP : #984

Why / Goal

The goal of the PR is to reduce the time taken to compute aggregations. Added daily level partial aggregates for batch feature computation, which avoids reading past day's event level data.

With this change, we see, 4x improvement for computing features. (except first time run )

Test Plan

Added Unit Tests
Covered by existing CI
Integration tested

Checklist

Documentation update

Reviewers

pengyu-hou · 2025-06-10T18:19:59Z

      s"${aggregationPart.inputColumn}_$opSuffix${aggregationPart.window.suffix}${bucketSuffix}"
+
+    def incOutputColumnName =
+      s"${aggregationPart.inputColumn}_$opSuffix${bucketSuffix}"


should we still keep the aggregationPart.window.suffix? Otherwise, how do we reconstruct the final output column?

@pengyu-hou not sure I get it. I can not use the window.suffix right as the intermediate incremental is daily aggregation.

pengyu-hou · 2025-06-10T18:26:14Z

    .toArray
    .zip(columnAggregators.map(_.irType))

+  val incSchema = aggregationParts


I am thinking that we should use a full incremental in the code and we can keep inc as the suffix for the table so the table names are not getting too long. What do you think?

Suggested change

val incSchema = aggregationParts

val incrementalSchema = aggregationParts

yes. good to use full name.

…hronon into kchakka/incremental

Co-authored-by: Pengyu Hou <3771747+pengyu-hou@users.noreply.github.com> Signed-off-by: chaitanya <1847554+kambstreat@users.noreply.github.com>

…hronon into kchakka/incremental

…ntal

…hronon into kchakka/incremental

….scala GroupBy.scala — bug fixes - Fix aggregationParts returning empty when no holes exist: Previously derived from the hole-filling loop, so when the incremental table was already up-to-date, convertIncrementalDfToHops received an empty list and produced wrong hops. Now derived directly from groupByConf unconditionally. - Fix maxWindow.get NPE: Replaced with getOrElse(throw ...) that gives a clear error message when incremental mode is used on a GroupBy with no windowed aggregations. - Remove unused imports: Operation and com.google.common.collect.Table. - Fix typo: "aggregatiosn" → "aggregations" in scaladoc. GroupByIncrementalTest.scala — test improvements - Enable UNIQUE_COUNT and BOUNDED_UNIQUE_COUNT: Uncommented both operations in testIncrementalBasicAggregations. Comparison converts array IRs to their sizes (since element ordering/MD5 hashing differs between Chronon and SQL) and validates against count(distinct ...) from SQL. - Rewrite testIncrementalStatisticalAggregations: Replaced schema-only checks with a full incremental-vs-non-incremental snapshotEvents comparison for SKEW, KURTOSIS, APPROX_PERCENTILE, APPROX_UNIQUE_COUNT, and APPROX_HISTOGRAM_K. - Remove redundant column-existence assertions: Dropped assertTrue("IR must contain X") checks that were superseded by the value comparison. - Remove unused BinaryType import.

Introduces an `is_incremental` mode for SNAPSHOT event GroupBy backfills. Per-day partial aggregates (IRs) are cached in a `<table>_daily_inc` table so the expensive raw-event hop aggregation is computed once per day and reused across backfill steps, rather than re-scanning the full window on every step. Squashed from PR #998 (kambstreat/chronon kchakka/incremental) onto main, to serve as the base for continued ownership and follow-up improvements. Co-authored-by: chaitu <kambstreat@users.noreply.github.com> Co-authored-by: Pengyu Hou <3771747+pengyu-hou@users.noreply.github.com>

@PARAM

* Add incremental SNAPSHOT GroupBy aggregation Introduces an `is_incremental` mode for SNAPSHOT event GroupBy backfills. Per-day partial aggregates (IRs) are cached in a `<table>_daily_inc` table so the expensive raw-event hop aggregation is computed once per day and reused across backfill steps, rather than re-scanning the full window on every step. Squashed from PR #998 (kambstreat/chronon kchakka/incremental) onto main, to serve as the base for continued ownership and follow-up improvements. Co-authored-by: chaitu <kambstreat@users.noreply.github.com> Co-authored-by: Pengyu Hou <3771747+pengyu-hou@users.noreply.github.com> * Make incremental IR build chunked/restartable and skew-resilient Two scalability fixes to the incremental SNAPSHOT GroupBy path, surfaced by benchmarking a ~500M-rows/day, 730-day-window source that OOM'd the cold build: 1. Chunk the daily-IR hole filling by stepDays. computeIncrementalDf now splits each unfilled hole into PartitionRange.steps(stepDays) sub-ranges, each committed separately. This makes the large cold / full-recompute build restartable (resumes via unfilledRanges) and bounds per-write shuffle/memory instead of materializing the whole window in one insertInto. stepDays is threaded from computeBackfill through fromIncrementalDf; default None keeps the original single-write behavior. 2. Replace groupByKey with map-side-combining aggregateByKey in convertIncrementalDfToHops, so a hot key's daily hops are merged incrementally rather than shuffled to one task and held as a single Iterable. Output type and ordering are unchanged. Adds testIncrementalChunkedBuild which runs computeBackfill in incremental mode with stepDays and asserts the output equals a non-incremental backfill. Existing correctness tests (which compare incremental vs non-incremental across all IR types) remain green. * Reduce per-row overhead in incremental hops conversion Two behavior-preserving micro-optimizations in convertIncrementalDfToHops, which runs once per cached IR row (~hundreds of millions on large windows): - Hoist convertSparkToJava from a per-row inner `def` to an object-level method so it is no longer re-allocated for every row. - Resolve the partition column index and the per-aggregation IR column indices once on the driver, instead of doing a name->index fieldIndex lookup per column per row. Indices (and partitionSpec) are captured in the executor closure rather than the DataFrame. No change to computed values; the existing incremental-vs-normal correctness tests across all IR types remain green. * Fix incremental IR write clobbering neighbor partitions computeIncrementalDf ran hopsAggregate over a window-widened source scan (getIntersectedRange pushes the read back by the query window), so it emitted daily IR hops for days outside the requested range. With dynamic partition overwrite, save() then wrote ALL of those partitions - so filling one day's hole overwrote neighboring partitions with window-truncated, incomplete data, gutting them to a few rows. This caused systematic, one-directional event loss in the final GroupBy output (observed at scale: a different day silently reduced to ~hundreds of rows per run, ~16M undercounted keys downstream). It was invisible to existing tests because their narrow ranges recomputed the clobbered neighbors correctly. Fix: clamp the IR write to exactly the requested range before save(), so only the target partition(s) are ever written. The widened input scan is still needed for correct per-day aggregation; only the output is clamped. Adds testIncrementalWriteIsClampedToRange: a single-day computeIncrementalDf over a 30-day input must write exactly one partition (writes 26 without the fix). * Split incremental build from read (materialize vs fromIncrementalDf) Separates the two halves of incremental mode so the _daily_inc table can be built once and consumed by multiple readers (backfill, upload, later join): - Rename the object-level computeIncrementalDf -> materializeIncrementalDf: the idempotent "producer" that fills _daily_inc holes for [range.start - maxWindow, range.end]. No-op when already complete. - fromIncrementalDf becomes a mode-agnostic read adapter (_daily_inc -> GroupBy with overridden hopsAggregate) and gains buildIfMissing (default true): * true = ensure-then-read (ad hoc / standalone; current behavior unchanged) * false = read-only (production, behind a dedicated producer node that is the sole writer; consumers run as pure readers) Pure refactor: computeBackfill and all existing callers use the default buildIfMissing=true, so behavior is unchanged. The instance computeIncrementalDf (per-hole worker) is untouched. All GroupByIncrementalTest cases pass. This is the seam for follow-up PRs: snapshot-events incremental upload and join right-parts consume fromIncrementalDf; production wires materializeIncrementalDf as a producer node the consumers depend on. * Keep computeIncrementalDf name; trim redundant comments - Revert the object-level method name back to computeIncrementalDf (from materializeIncrementalDf); it coexists with the instance-level overload as before. Docstring disambiguates the two. - Remove comments that restate the code (per-branch line/step narration, type-conversion case labels, @PARAM boilerplate) while keeping the "why" comments: the write clamp rationale, aggregateByKey skew resilience, the finalize=false note, and the closure/serialization notes. No behavior change; all GroupByIncrementalTest cases pass. * scalafmt * restrict to snapshot + event * Address review: tag incremental table, fix accuracy guard, require daily resolution Responds to PR review feedback: - Tag the _daily_inc table as Chronon-generated with a dedicated table type. Adds Constants.TableType.GroupByIncremental and tags the write in computeIncrementalDf (was previously untagged, unlike the backfill output). - Fix the Python is_incremental accuracy guard: accuracy is optional and inferred as SNAPSHOT when unset (absent a streaming source). The guard now accepts the inferred-SNAPSHOT case instead of rejecting accuracy=None. The Scala backstop already uses inferredAccuracy. - Require DailyResolution in the instance computeIncrementalDf, since the incremental IRs are daily; guards against a non-daily resolution silently producing wrong IRs. Also de-flakes testIncrementalFirstLast: it perturbed ts with rand() to make timestamps unique, but rand() can collide, leaving FIRST/LAST tie-breaks ambiguous. Replaced with a deterministic per-day row-number offset so no two events in a day share a ts. Verified stable across repeated runs. * Dedup shared daily IR columns in incremental mode Aggregations sharing the same (operation, input_column, bucket) but differing only by window collapse to the same daily IR column name, since the incremental column name drops the window suffix (e.g. SUM(price, 5d) and SUM(price, 10d) both -> price_sum). This is valid in normal mode but previously produced duplicate columns in _daily_inc. The collapsed IR values are identical, so persist one column per distinct name (uniqueIncrementalIndices) on both the schema and the written values. The read path resolves each part by column name, so all collapsed windows map back to the single shared column and the windowed output is reconstructed correctly. Adds testIncrementalDedupsSharedColumns: verifies a single price_sum column is written and the final windowed output (price_sum_5d / price_sum_10d) matches the non-incremental backfill. Addresses PR review comment on Extensions.scala (duplicate columns across windows). * Guard resolution in overridden hopsAggregate Follow-up to the DailyResolution check: the incremental GroupBy's overridden hopsAggregate ignores its resolution argument and always returns the cached daily hops. Add require(resolution == DailyResolution) inside the override so a non-daily consumer (e.g. a TEMPORAL upload calling hopsAggregate with FiveMinuteResolution) fails loudly instead of silently receiving daily hops as if finer-grained. * Trim verbose comments, remove spark.stop() from tests, scalafmt - Condense the wordier incremental-mode comments/scaladocs in GroupBy.scala to one or two lines while keeping the rationale. - Remove spark.stop() from GroupByIncrementalTest cases: it shut down the shared singleton SparkSession mid-suite, causing cross-test flakiness (an unrelated test could fail in a full-suite run). Verified the suite is stable across repeated full runs. - scalafmt. --------- Co-authored-by: chaitu <kambstreat@users.noreply.github.com>

kambstreat added 3 commits May 30, 2025 13:11

code to write daily irs

eaacfa6

store incremental agg and compute final IRs

40b6cb2

Store hops to inc tables

a014b6e

pengyu-hou reviewed Jun 10, 2025

View reviewed changes

kambstreat added 9 commits June 13, 2025 18:10

add code changes to generate final output from IR for AVG

32d559e

change function structure and variable names

37293df

remove unused functions

6263706

change function defs

cb4325b

make changes

796ef96

change function order

f218b23

add new field is_incremental to python api

b1d4ee9

get argument for isIncremental in scala spark backend

2ab7659

add unit test for incremental groupby

238c781

pengyu-hou reviewed Jun 20, 2025

View reviewed changes

Comment thread spark/src/test/scala/ai/chronon/spark/test/GroupByTest.scala

pengyu-hou reviewed Jul 17, 2025

View reviewed changes

Comment thread api/src/main/scala/ai/chronon/api/Extensions.scala Outdated

kambstreat added 14 commits July 17, 2025 21:00

reuse table ccreation

8edfd27

Update GroupByTest

e903683

Add GroupByTest for events

0bdc4fc

changes for incrementalg

7987931

resolve merge conflicts

2b26d45

add last hole logic for incrementnal bacckfill

7b62a43

fix syntax

aeeb5ec

fix bug : backfill only for missing holes

9180d23

fix none error for inc Table

ee81672

add incremental table queryable range

29a3f28

add logging for tableUtils

aa16010

add log

ff41cc9

fill incremental holes

aa25f9f

modify incremental aggregation parts

3efe8cd

kambstreat and others added 26 commits February 6, 2026 22:28

resolve merge conflicts

bb9b7d2

scala formatted

2f78a05

fix for scala 2.13

b296d70

fix for scala 2.13

3202e7d

change unit test

fa2fcba

add toSeq for scala 2.13 compatibility

b0e90bf

fix last/first tests

7138a66

add new test file for incremental aggregations

325302c

fix failed ci

40aa755

remove obsolete comment

b847a40

Merge branch 'kchakka/incremental' of https://github.com/kambstreat/c…

7692f24

…hronon into kchakka/incremental

revert to master files tor spark_submit.sh and teams.json

b385cda

remove log statements during debug

464395b

Add daily_inc suffix to incremental table

1002744

Co-authored-by: Pengyu Hou <3771747+pengyu-hou@users.noreply.github.com> Signed-off-by: chaitanya <1847554+kambstreat@users.noreply.github.com>

fix bug in flatten function

cd141ab

Merge branch 'kchakka/incremental' of https://github.com/kambstreat/c…

9a25d10

…hronon into kchakka/incremental

revert changes in GroupByTest

a9f3c78

Merge branch 'main' of github.com:airbnb/chronon into kchakka/increme…

0a8499d

…ntal

remove comment

91d2768

Merge branch 'main' into kchakka/incremental

5649e0c

Merge branch 'kchakka/incremental' of https://github.com/kambstreat/c…

c68ae66

…hronon into kchakka/incremental

Merge branch 'main' into kchakka/incremental

eea3bbf

Merge branch 'kchakka/incremental' of https://github.com/kambstreat/c…

127410c

…hronon into kchakka/incremental

Merge branch 'main' into kchakka/incremental

84d3ece

Merge branch 'main' into kchakka/incremental

1ad74fe

Navigation Menu

Uh oh!

Write daily IRs for incremental aggregation - #998

Write daily IRs for incremental aggregation#998
kambstreat wants to merge 67 commits into
airbnb:mainfrom
kambstreat:kchakka/incremental

kambstreat commented Jun 3, 2025 •

edited

Loading

Uh oh!

pengyu-hou Jun 10, 2025

Uh oh!

kambstreat Jun 11, 2025

Uh oh!

pengyu-hou Jun 10, 2025

Uh oh!

kambstreat Jun 11, 2025

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

	val incSchema = aggregationParts
	val incrementalSchema = aggregationParts

Uh oh!

Conversation

kambstreat commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Why / Goal

Test Plan

Checklist

Reviewers

Uh oh!

pengyu-hou Jun 10, 2025

Choose a reason for hiding this comment

Uh oh!

kambstreat Jun 11, 2025

Choose a reason for hiding this comment

Uh oh!

pengyu-hou Jun 10, 2025

Choose a reason for hiding this comment

Uh oh!

kambstreat Jun 11, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

kambstreat commented Jun 3, 2025 •

edited

Loading