zozo123 opened a new pull request, #76156:
URL: https://github.com/apache/airflow/pull/76156
related: #74195
This PR covers only the **external producer with checkpointed delivery**
part of #74195. It complements #74550, which adds the version sensor, and does
not touch the sensor.
## What it adds
`DatabricksDeltaTableVersionProducerOperator` runs in a recurring producer
Dag. On each run it reads the version of a Unity Catalog Delta table. If the
table has a new data version committed outside Airflow, the operator publishes
one asset event for that version. If nothing changed, the task is skipped.
```python
orders = UnityTableIdentity(host="adb-123.azuredatabricks.net",
catalog="main", schema="sales", table="orders")
with DAG("orders_delta_producer", schedule="*/10 * * * *",
max_active_runs=1, catchup=False):
DatabricksDeltaTableVersionProducerOperator(
task_id="publish_orders_versions",
unity_table=orders,
outlets=[orders.to_asset()],
sql_warehouse_name="analytics",
)
```
Consumers schedule on the asset as usual and can pin their reads with
`VERSION AS OF {{ ... .extra['version'] }}`.
## Design
- **The checkpoint is an asset event.** The checkpoint is the newest event
this producer published on the asset, found by
`extra["databricks_delta_checkpoint_key"]`. The key defaults to
`"<dag_id>.<task_id>"`. The operator reads it through `inlet_events`, and the
asset is added to inlets automatically. There is no Variable, XCom or external
state store.
- On Airflow 3 the lookup is filtered on the server
(`.extra(...).ascending(False).limit(1)`) when the accessor supports it.
- On older versions it falls back to iterating the accessor.
- **Atomic publish and checkpoint.** The outlet event is recorded together
with the task's success state, so publishing an event and advancing the
checkpoint are one metadata-database write. A run that fails or crashes before
success publishes nothing and leaves the checkpoint where it was. The next run
publishes the same version again.
- **At-least-once delivery with deduplication.** Each event carries
`dedup_id = "<asset uri>@<delta table id>:<version>"`. Retries and redeliveries
reuse the same `dedup_id`. The default `max_active_tis_per_dag=1` avoids
concurrent publishing.
- **Coalescing.** Several commits between two runs produce one event for the
latest version. The event also records `previous_version`.
- **Maintenance commits.** `OPTIMIZE`, `VACUUM START` and `VACUUM END` are
ignored by default (`ignored_operations` matches case-insensitively). This uses
the same parameter name and default as #74550.
- If the newest commit is a maintenance commit, the operator scans the
history back to the checkpoint for a data commit.
- If the history has expired, or the gap is larger than `max_history_rows`
(default 1000), it publishes, to be safe.
- **Recreated tables.** If the Delta table id changes or the version goes
backwards, the operator either publishes a `change="recreated"` event (the
default) or fails (`on_recreation="fail"`).
- **First run.** With no checkpoint, the operator publishes an `enrollment`
event for the current version. With `baseline_version=N`, it publishes only
versions newer than N.
- **Safety checks.**
- The connection host must match `unity_table.host`, case-insensitively.
This is checked before any query runs.
- Non-Delta tables are rejected.
- Identifiers are quoted with backticks.
- Errors raise a new `DatabricksDeltaTableError`.
- **Cost.** Each run issues two metadata queries: `DESCRIBE DETAIL` and
`DESCRIBE HISTORY ... LIMIT n`.
- **Event extras.** The extra reuses the keys from #74550 (`version`,
`observed_version`, `table_identity`, `operation`, `timestamp`). It also adds
`change`, `previous_version`, `delta_table_id`, `dedup_id` and the checkpoint
key. It reuses `UnityTableIdentity` and `create_asset` from
`assets/databricks.py`, so there is no new URI scheme.
## Tests
`tests/unit/databricks/operators/test_databricks_delta_table.py` has 43
tests, all passing locally. They use a fake Delta table that serves `DESCRIBE
DETAIL` and `DESCRIBE HISTORY`, including expired history. They cover:
- argument validation
- enrollment and baseline
- coalescing and maintenance-only commits
- data commits hidden behind maintenance commits
- expired history and the scan window
- recreation in both modes
- non-Delta and missing tables
- host mismatch, with no queries issued
- checkpoint scoping, the server-side filter path and corrupt checkpoints
- delivery after a crash (same `dedup_id`) and concurrent runs
One `db_test`, for Airflow 3.2+, runs the producer Dag 5 times with
`dag.test()` against the real metadata DB. Commits arrive in the order `-, -,
WRITE v3, OPTIMIZE v4, -`. The test checks that the states are `success,
skipped, success, skipped, skipped` and that exactly two events are stored
(`enrollment@2` and `new_version@3`). This exercises the real `inlet_events`
checkpoint path.
The PR also adds a how-to page (`docs/operators/delta_table_producer.rst`)
and a system example, and registers both in `provider.yaml`.
## Known limitations and follow-ups
- The history-reading logic overlaps with #74550. It could be shared after
that PR lands.
- Other tasks that publish events for the same asset (for example `COPY
INTO` with outlets) also trigger consumers, and their events may not have a
`version`. The docs explain how consumers can tell them apart.
- An `AssetWatcher`/trigger-based producer is not included and would be a
possible follow-up.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: AI coding assistant (Grok Bot) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
