caoterry commented on PR #73983: URL: https://github.com/apache/airflow/pull/73983#issuecomment-6588637149
@ashb I rebased onto main and re-measured, since the code behind the numbers above has moved. The workflows need approving again after the force-push. - The migration is now `0154` (3.5.0, revises `16d7301f124b`). The indexes and the downgrade are unchanged. Round-tripped again on SQLite (the migration round-trip check), Postgres 16 and MySQL 8.4. - The per-key lookup is now [`AssetManager._get_latest_pending_apdr`](https://github.com/apache/airflow/blob/88e5c356eaa13047ab04dd04a73c394ddc9aca00/airflow-core/src/airflow/assets/manager.py#L768-L787). It also filters `created_dag_run_id IS NULL` (#71074) and only runs when `batch_asset_events` is on (#68517). The [scheduler's pending scan](https://github.com/apache/airflow/blob/88e5c356eaa13047ab04dd04a73c394ddc9aca00/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L2493-L2510) is unchanged. Same setup as before: Postgres 16, 100,000 `asset_partition_dag_run` rows for one Dag, 99,500 fired, and the newest 500 pending with `pending_partition_key` set as batching would set it. Main at `16d7301f124b` vs this PR, `EXPLAIN (ANALYZE, BUFFERS)`, warm cache: | Query | Before | After | |---|---|---| | Lookup, key with a pending row | Seq Scan + Sort, 1,037 buffers, 9.17 ms | Index Scan Backward on `idx_apdr_target_dag_id_partition_key_id`, 4 buffers, 0.035 ms | | Lookup, key already fired (no pending row) | Seq Scan + Sort, 1,037 buffers, 9.82 ms | Same index, 5 buffers, 0.056 ms | | Scheduler pending scan, 500 rows (`FOR UPDATE SKIP LOCKED`) | Seq Scan, 1,037 buffers for the scan, 9.43 ms | Index Scan on `idx_apdr_created_dag_run_id_created_at_id`, 12 buffers for the scan, 0.95 ms | | Deleting 100 `dag_run` rows, `apdr_created_dag_run_id_fkey` cascade trigger | 1,386 ms for 100 calls | 1.19 ms for 100 calls | These timings come from a laptop, so compare within a row rather than against my earlier comment. The plan shapes match it. <details><summary>Plans: before (main, 16d7301f124b)</summary> ``` -- lookup, key with a pending row Limit (cost=2537.01..2537.02 rows=1 width=94) (actual time=9.149..9.150 rows=1 loops=1) Buffers: shared hit=1037 -> Sort (cost=2537.01..2537.02 rows=1 width=94) (actual time=9.147..9.148 rows=1 loops=1) Sort Key: id DESC Sort Method: quicksort Memory: 25kB Buffers: shared hit=1037 -> Seq Scan on asset_partition_dag_run (cost=0.00..2537.00 rows=1 width=94) (actual time=8.939..9.113 rows=1 loops=1) Filter: ((created_dag_run_id IS NULL) AND ((partition_key)::text = 'acct_99800'::text) AND ((target_dag_id)::text = 'consumer'::text)) Rows Removed by Filter: 99999 Buffers: shared hit=1037 Execution Time: 9.169 ms -- lookup, key already fired Limit (cost=2537.01..2537.02 rows=1 width=94) (actual time=9.798..9.799 rows=0 loops=1) Buffers: shared hit=1037 -> Sort (cost=2537.01..2537.02 rows=1 width=94) (actual time=9.797..9.797 rows=0 loops=1) Sort Key: id DESC Sort Method: quicksort Memory: 25kB Buffers: shared hit=1037 -> Seq Scan on asset_partition_dag_run (cost=0.00..2537.00 rows=1 width=94) (actual time=9.791..9.791 rows=0 loops=1) Filter: ((created_dag_run_id IS NULL) AND ((partition_key)::text = 'acct_50000'::text) AND ((target_dag_id)::text = 'consumer'::text)) Rows Removed by Filter: 100000 Buffers: shared hit=1037 Execution Time: 9.816 ms -- pending scan (scheduler loop) Limit (cost=2066.67..2072.92 rows=500 width=106) (actual time=8.914..9.359 rows=500 loops=1) Buffers: shared hit=1538 -> LockRows (cost=2066.67..2072.92 rows=500 width=106) (actual time=8.913..9.313 rows=500 loops=1) Buffers: shared hit=1538 -> Sort (cost=2066.67..2067.92 rows=500 width=106) (actual time=8.834..8.863 rows=500 loops=1) Sort Key: asset_partition_dag_run.created_at, asset_partition_dag_run.id Sort Method: quicksort Memory: 75kB Buffers: shared hit=1038 -> Nested Loop (cost=0.00..2044.26 rows=500 width=106) (actual time=8.518..8.702 rows=500 loops=1) Join Filter: ((asset_partition_dag_run.target_dag_id)::text = (dag.dag_id)::text) Buffers: shared hit=1038 -> Seq Scan on dag (cost=0.00..1.01 rows=1 width=15) (actual time=0.011..0.012 rows=1 loops=1) Filter: ((is_paused IS FALSE) AND (is_draining IS FALSE) AND (is_stale IS FALSE)) Buffers: shared hit=1 -> Seq Scan on asset_partition_dag_run (cost=0.00..2037.00 rows=500 width=100) (actual time=8.502..8.603 rows=500 loops=1) Filter: (created_dag_run_id IS NULL) Rows Removed by Filter: 99500 Buffers: shared hit=1037 Execution Time: 9.431 ms -- dag_run delete, 100 runs Trigger for constraint apdr_created_dag_run_id_fkey: time=1386.306 calls=100 Execution Time: 1421.502 ms ``` </details> <details><summary>Plans: after (this PR, f954ddd21484)</summary> ``` -- lookup, key with a pending row Limit (cost=0.42..8.44 rows=1 width=94) (actual time=0.019..0.019 rows=1 loops=1) Buffers: shared hit=4 -> Index Scan Backward using idx_apdr_target_dag_id_partition_key_id on asset_partition_dag_run (cost=0.42..8.44 rows=1 width=94) (actual time=0.018..0.018 rows=1 loops=1) Index Cond: (((target_dag_id)::text = 'consumer'::text) AND ((partition_key)::text = 'acct_99800'::text)) Filter: (created_dag_run_id IS NULL) Buffers: shared hit=4 Execution Time: 0.035 ms -- lookup, key already fired Limit (cost=0.42..8.44 rows=1 width=94) (actual time=0.033..0.033 rows=0 loops=1) Buffers: shared hit=5 -> Index Scan Backward using idx_apdr_target_dag_id_partition_key_id on asset_partition_dag_run (cost=0.42..8.44 rows=1 width=94) (actual time=0.032..0.032 rows=0 loops=1) Index Cond: (((target_dag_id)::text = 'consumer'::text) AND ((partition_key)::text = 'acct_50000'::text)) Filter: (created_dag_run_id IS NULL) Rows Removed by Filter: 1 Buffers: shared hit=5 Execution Time: 0.056 ms -- pending scan (scheduler loop) Limit (cost=798.62..804.87 rows=500 width=106) (actual time=0.396..0.876 rows=500 loops=1) Buffers: shared hit=964 -> LockRows (cost=798.62..805.25 rows=530 width=106) (actual time=0.395..0.832 rows=500 loops=1) Buffers: shared hit=964 -> Sort (cost=798.62..799.95 rows=530 width=106) (actual time=0.381..0.410 rows=500 loops=1) Sort Key: asset_partition_dag_run.created_at, asset_partition_dag_run.id Sort Method: quicksort Memory: 75kB Buffers: shared hit=13 -> Nested Loop (cost=0.42..774.64 rows=530 width=106) (actual time=0.025..0.257 rows=500 loops=1) Join Filter: ((asset_partition_dag_run.target_dag_id)::text = (dag.dag_id)::text) Buffers: shared hit=13 -> Seq Scan on dag (cost=0.00..1.01 rows=1 width=15) (actual time=0.009..0.010 rows=1 loops=1) Filter: ((is_paused IS FALSE) AND (is_draining IS FALSE) AND (is_stale IS FALSE)) Buffers: shared hit=1 -> Index Scan using idx_apdr_created_dag_run_id_created_at_id on asset_partition_dag_run (cost=0.42..767.00 rows=530 width=100) (actual time=0.013..0.159 rows=500 loops=1) Index Cond: (created_dag_run_id IS NULL) Buffers: shared hit=12 Execution Time: 0.947 ms -- dag_run delete, 100 runs Trigger for constraint apdr_created_dag_run_id_fkey: time=1.185 calls=100 Execution Time: 5.999 ms ``` </details> -- 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]
