kaxil commented on PR #75404:
URL: https://github.com/apache/airflow/pull/75404#issuecomment-6345232080
I ran the old and new queries side by side on SQLite, Postgres 14 and MySQL
8.0 to see whether the rewrite changes which assets get orphaned. On my data it
didn't.
I copied `asset`, `asset_alias`, `asset_alias_asset` and the three reference
tables from the models at 42b1201 (same columns, PKs, NOT NULLs and indexes; I
left out FKs so seeding didn't need Dag rows). There were 12,000 assets. 20 of
them had N rows in each of the four reference tables. For the rest, each table
got 1 to 3 rows with p=0.15, so about half were unreferenced and many were
referenced from one table only. Both queries were built from SQLAlchemy as
written (old from main, new from this branch), and I compared the full CTE rows
with each other and with a ground truth computed in Python.
With N=15 (about 50k joined rows per hot asset in the old query), single
runs:
| Backend | Old orphan / active | New orphan / active | Old time (orphan /
active) | New time (orphan / active) |
|---|---|---|---|---|
| SQLite | 6226 / 5774 | 6226 / 5774 | 0.84s / 0.09s | 0.020s / 0.018s |
| Postgres 14 | 6226 / 5774 | 6226 / 5774 | 0.73s / 0.15s | 0.063s / 0.025s |
| MySQL 8.0 | 6226 / 5774 | 6226 / 5774 | 1.06s / 0.25s | 0.048s / 0.064s |
The orphan and active ID sets were the same for old, new and the Python
ground truth on all three backends. With N=30 on Postgres the old query took
2.66s / 1.91s and the new one 0.055s / 0.026s.
From the schema side: every column the old query counts (`dag_id` on the
three reference tables, `alias_id` on `asset_alias_asset`) is a NOT NULL PK
column, so after the outer joins it is non-null exactly when a matching row
exists. The join multiplies the counts, but a count of zero still means no
references.
<details>
<summary>Generated SQL (Postgres; MySQL and SQLite compile to the same
shape)</summary>
Old orphan CTE (main):
```sql
WITH anon_1 AS
(SELECT asset.id AS id, asset.name AS name, asset.uri AS uri, asset."group"
AS "group", asset.extra AS extra, asset.created_at AS created_at,
asset.updated_at AS updated_at
FROM asset LEFT OUTER JOIN dag_schedule_asset_reference ON asset.id =
dag_schedule_asset_reference.asset_id LEFT OUTER JOIN
task_outlet_asset_reference ON asset.id = task_outlet_asset_reference.asset_id
LEFT OUTER JOIN task_inlet_asset_reference ON asset.id =
task_inlet_asset_reference.asset_id LEFT OUTER JOIN asset_alias_asset ON
asset_alias_asset.asset_id = asset.id GROUP BY asset.id
HAVING count(dag_schedule_asset_reference.dag_id) +
count(task_outlet_asset_reference.dag_id) +
count(task_inlet_asset_reference.dag_id) + count(asset_alias_asset.alias_id) =
%(param_1)s)
SELECT anon_1.id
FROM anon_1
```
New orphan CTE (this PR):
```sql
WITH anon_1 AS
(SELECT asset.id AS id, asset.name AS name, asset.uri AS uri, asset."group"
AS "group", asset.extra AS extra, asset.created_at AS created_at,
asset.updated_at AS updated_at
FROM asset
WHERE NOT ((EXISTS (SELECT *
FROM dag_schedule_asset_reference
WHERE dag_schedule_asset_reference.asset_id = asset.id)) OR (EXISTS (SELECT *
FROM task_outlet_asset_reference
WHERE task_outlet_asset_reference.asset_id = asset.id)) OR (EXISTS (SELECT *
FROM task_inlet_asset_reference
WHERE task_inlet_asset_reference.asset_id = asset.id)) OR (EXISTS (SELECT *
FROM asset_alias_asset
WHERE asset_alias_asset.asset_id = asset.id))))
SELECT anon_1.id
FROM anon_1
```
The active CTEs are the same with `!=` in the `HAVING` (old) and without the
outer `NOT` (new).
</details>
<details>
<summary>Postgres 14 plans (N=15, <code>EXPLAIN (ANALYZE, COSTS OFF, TIMING
OFF)</code>, hash-condition, bucket/memory and planning-time lines
trimmed)</summary>
New:
```
Seq Scan on asset (actual rows=6226 loops=1)
Filter: ((NOT (hashed SubPlan 8)) AND (NOT (hashed SubPlan 2)) AND (NOT
(hashed SubPlan 4)) AND (NOT (hashed SubPlan 6)))
Rows Removed by Filter: 5774
SubPlan 8
-> Seq Scan on asset_alias_asset (actual rows=3996 loops=1)
SubPlan 2
-> Seq Scan on dag_schedule_asset_reference (actual rows=3966 loops=1)
SubPlan 4
-> Seq Scan on task_outlet_asset_reference (actual rows=3920 loops=1)
SubPlan 6
-> Seq Scan on task_inlet_asset_reference (actual rows=3886 loops=1)
Execution Time: 25.381 ms
```
Old:
```
HashAggregate (actual rows=6226 loops=1)
Group Key: asset.id
Filter: ((((count(dag_schedule_asset_reference.dag_id) +
count(task_outlet_asset_reference.dag_id)) +
count(task_inlet_asset_reference.dag_id)) + count(asset_alias_asset.alias_id))
= 0)
Rows Removed by Filter: 5774
-> Hash Right Join (actual rows=1033613 loops=1)
-> Seq Scan on task_outlet_asset_reference (actual rows=3920
loops=1)
-> Hash (actual rows=85844 loops=1)
-> Hash Right Join (actual rows=85844 loops=1)
-> Seq Scan on task_inlet_asset_reference (actual
rows=3886 loops=1)
-> Hash (actual rows=20448 loops=1)
-> Hash Right Join (actual rows=20448 loops=1)
-> Seq Scan on
dag_schedule_asset_reference (actual rows=3966 loops=1)
-> Hash (actual rows=14139 loops=1)
-> Hash Right Join (actual
rows=14139 loops=1)
-> Seq Scan on
asset_alias_asset (actual rows=3996 loops=1)
-> Hash (actual rows=12000
loops=1)
-> Seq Scan on asset
(actual rows=12000 loops=1)
Execution Time: 96.882 ms
```
</details>
In the Postgres run the old plan built 1,033,613 joined rows for 12,000
assets before the `HashAggregate`.
A small note on the description: in this plan Postgres rewrote `NOT (... OR
...)` as four `AND NOT` filters and ran each `EXISTS` as a hashed subplan over
a seq scan, so the `asset_id` indexes weren't used. That may change with larger
tables or a smaller `work_mem`; I only looked at this one size.
On tests: all 16 `-k orphan` tests in `test_scheduler_job.py` pass on this
branch with SQLite. I then deleted each `EXISTS` line in turn and reran `-k
asset_orphaning`:
| `EXISTS` removed | Failing tests |
|---|---|
| `DagScheduleAssetReference` | `test_asset_orphaning` |
| `TaskOutletAssetReference` | `test_asset_orphaning`,
`test_asset_orphaning_asset_referenced_from_every_table` |
| `TaskInletAssetReference` | none (4 passed) |
| `alias_association_table` |
`test_asset_orphaning_keeps_alias_materialized_asset_active` |
That's the gap from my inline comment on the new test. Adding an inlet-only
asset there would cover it.
--
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]
