Add offlineGroupBy option for external source

[Hui-Cheng-AirBnb]

Summary

Add and validate offline groupby option for external source. This is the first part of the task to support backfill for external sources. This PR does:

1. Add offline GroupBy to ExternalSource
2. Validate the GroupBy schema is compatible with the External source.
3. Add offline GroupBy as part of the join parts for offline backfill

Why / Goal

We want to introduce a first-level abstraction for External source to support offline backfill. The idea is that an ExternalSource now can take in an optional GroupBy definition. During offline backfill, Chronon will dispatch to the GroupBy for backfill.

Test Plan

• Added Unit Tests
• Covered by existing CI
• Integration tested

========= Test configs =======
Test configs are at https://git.musta.ch/airbnb/ml_models/pull/27812

========= Export schema with External offline groupby =========

python \
    run_on_bigair.py run \
    production/joins/zipline_test/test_online_join_external.v2 \
    --envs "VERSION:hui-cheng-ext-gb-1-0.0.110-SNAPSHOT" \
    -- \
    --ds=2025-10-12 \
    --mode=analyze \
    --skip-timestamp-check \
    --skip-table-permission-check \
    --export-schema

• BigAir UI link: https://mli.airbnb.tools/bigair/workflow-run/4DPr4cRADSwqzr6dW3cXD8
• airflow_url: https://airflow-neon.a.musta.ch/tree?dag_id=chronon_runpy_wf.hui_cheng.4DPr4cRADSwqzr6dW3cXD8

========= Backfill join with External offline groupby ===========

python \
    run_on_bigair.py run \
    production/joins/zipline_test/test_online_join_external.v2 \
    --envs "VERSION:hui-cheng-ext-gb-1-0.0.110-SNAPSHOT" \
    -- \
    --ds=2025-10-12 \
    --mode=backfill

• BigAir UI link: https://mli.airbnb.tools/bigair/workflow-run/30pM325mrOoLRgCNEU9Xva
• airflow_url: https://airflow-neon.a.musta.ch/tree?dag_id=chronon_runpy_wf.hui_cheng.30pM325mrOoLRgCNEU9Xva

https://superset.a.musta.ch/sqllab/p/Oa8yamlbN8K/

======== Backfill join WITHOUT External offline groupby ========

python \
    run_on_bigair.py run \
    production/joins/zipline_test/test_online_join_small.v2 \
    --envs "VERSION:hui-cheng-ext-gb-1-0.0.110-SNAPSHOT" \
    -- \
    --ds=2025-10-12 \
    --mode=backfill

• BigAir UI link: https://mli.airbnb.tools/bigair/workflow-run/5OHqpaOMUsyqMZzpJDSSRV
• airflow_url: https://airflow-neon.a.musta.ch/tree?dag_id=chronon_runpy_wf.hui_cheng.5OHqpaOMUsyqMZzpJDSSRV

https://superset.a.musta.ch/sqllab/p/gKx6rAQ2kxz/

Compare with backfilled table with official Chronon build
https://superset.a.musta.ch/sqllab/p/4RJ9e9zBwJA/

========= Export schema WITHOUT External offline groupby =====

python \
    run_on_bigair.py run \
    production/joins/zipline_test/test_online_join_small.v2 \
    --envs "VERSION:hui-cheng-ext-gb-1-0.0.110-SNAPSHOT" \
    -- \
    --ds=2025-10-12 \
    --mode=analyze \
    --skip-timestamp-check \
    --skip-table-permission-check \
    --export-schema

• BigAir UI link: https://mli.airbnb.tools/bigair/workflow-run/38Q3SKmezh9Xt4dgZPJMju
• airflow_url: https://airflow-neon.a.musta.ch/tree?dag_id=chronon_runpy_wf.hui_cheng.38Q3SKmezh9Xt4dgZPJMju

Check schema matches with the schema exported from official package
https://superset.a.musta.ch/sqllab/p/6B3NRO5A13R/

Checklist

• Documentation update

Reviewers

[hzding621]

We may also need to update:

• https://github.com/airbnb/chronon/blob/main/api/py/ai/chronon/repo/join_backfill.py#L78
• https://github.com/airbnb/chronon/blob/main/spark/src/main/scala/ai/chronon/spark/Join.scala#L218

These logics are used in join backfill DAGs where each join_parts are run in parallel.

Today, production run uses: https://github.com/airbnb/chronon/blob/main/spark/src/main/scala/ai/chronon/spark/JoinBase.scala#L477 which has been handled in this PR. For dev run, we use join_backfill.py to orchestrate a fine-grained DAG, where each task runs, computeLeft, computeJoinPartOpt, and computeFinalJoin in different spark jobs. We have unit test for this scenario in JoinFlowTest.scala

[Hui-Cheng-AirBnb]

We may also need to update:

• https://github.com/airbnb/chronon/blob/main/api/py/ai/chronon/repo/join_backfill.py#L78
• https://github.com/airbnb/chronon/blob/main/spark/src/main/scala/ai/chronon/spark/Join.scala#L218

These logics are used in join backfill DAGs where each join_parts are run in parallel.

Today, production run uses: https://github.com/airbnb/chronon/blob/main/spark/src/main/scala/ai/chronon/spark/JoinBase.scala#L477 which has been handled in this PR. For dev run, we use join_backfill.py to orchestrate a fine-grained DAG, where each task runs, computeLeft, computeJoinPartOpt, and computeFinalJoin in different spark jobs. We have unit test for this scenario in JoinFlowTest.scala

Thanks @hzding621 , I've made a change and added unit tests. Let me know if you think we should do a test dev run (if yes, please share instructions of dev run)