You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "safaehar (via GitHub)" <gi...@apache.org> on 2026/10/02 17:21:50 UTC
[PR] Make db clean --skip-archive delete rows directly [airflow]
safaehar opened a new pull request, #74113:
URL: https://github.com/apache/airflow/pull/74113
`airflow db clean --skip-archive` currently copies every selected batch into an archive table, deletes the source rows, and then drops the archive table. This still incurs the archive writes and requires schema `CREATE`/`DROP` privileges even though the archive is not preserved.
This change gives `--skip-archive` a direct-delete path:
- PostgreSQL and SQLite delete primary keys selected by the existing limited cleanup query.
- MySQL uses a derived primary-key selection with `EXISTS`, avoiding MySQL's unsupported `LIMIT` in an `IN` subquery.
- Each batch remains a separate transaction and retains the current selection, keep-last, composite-primary-key, rollback, and progress-output behavior.
- Cleanup without `--skip-archive` continues to create and retain archive tables.
The CLI help and cleanup documentation now state that `--skip-archive` deletes directly without creating archive tables.
## PostgreSQL WAL comparison
On a disposable PostgreSQL 13 database, deleting 1,000,000 rows from the same two-column table and measuring with `pg_wal_lsn_diff(pg_current_wal_lsn(), :start_lsn)` produced:
- archive path (`CREATE TABLE AS`, joined `DELETE`, `DROP`): 172,750,560 bytes
- direct-delete path: 100,445,792 bytes
The direct path generated approximately 42% less WAL in this comparison.
## Tests
- `uv run --project airflow-core pytest airflow-core/tests/unit/utils/test_db_cleanup.py -xvs` (SQLite: 99 passed, 3 backend-specific skipped)
- Focused PostgreSQL run covering skip-archive batching, keep-last, composite XCom keys, archive regression, and failure handling (11 passed)
- `prek run --stage pre-commit --files airflow-core/src/airflow/utils/db_cleanup.py airflow-core/tests/unit/utils/test_db_cleanup.py airflow-core/src/airflow/cli/cli_config.py airflow-core/docs/howto/usage-cli.rst`
closes: #42003
related: #73417
related: #67187
related: #28051
related: #66177
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (Codex)
Generated-by: Codex following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
* When adding dependency, check compliance with the ASF 3rd Party License Policy.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "safaehar (via GitHub)" <gi...@apache.org>.
safaehar commented on code in PR #74113:
URL: https://github.com/apache/airflow/pull/74113#discussion_r4401045322
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -605,44 +615,51 @@ def _do_delete(
)
except BaseException:
- error_raised = True
- # Roll back the failed transaction so its locks are released before
- # the archive table is dropped in the ``finally`` block below.
- # ``rollback()`` itself can raise (e.g. the connection died); suppress
- # it so it does not shadow the original error being re-raised.
with suppress(Exception):
session.rollback()
raise
- finally:
- if target_table is not None and skip_archive:
- # Drop the archive table on the session's own connection. Binding
- # the drop to ``session.get_bind()`` (the Engine) would check out a
- # *second* pooled connection, and on MySQL its ``DROP TABLE`` blocks
- # indefinitely on the metadata lock still held by this session's
- # open transaction when the DELETE above failed -- the ``db clean``
- # hang reported in #66177.
- try:
- target_table.drop(bind=session.connection())
- session.commit()
- except Exception:
- # If we are already unwinding from a delete failure, a cleanup
- # error here must not replace the original exception (Python
- # makes a ``finally``-raised error the top-level one). Log and
- # let the original delete error keep propagating. On the success
- # path (no delete error), a drop/commit failure is a real
- # problem, so re-raise it.
- if not error_raised:
- raise
- logger.warning(
- "Failed to drop archive table %s while cleaning up after a "
- "delete failure; propagating the original delete error instead.",
- target_table_name,
- exc_info=True,
- )
print("Finished Performing Delete")
+def _delete_directly(
Review Comment:
Agreed. the source table does not change between batches, so reflecting it each time is unnecessary. I’ll cache the reflected table in _do_delete() and pass it to _delete_directly(), while keeping the initialization lazy so an empty cleanup does not perform reflection
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "safaehar (via GitHub)" <gi...@apache.org>.
safaehar commented on code in PR #74113:
URL: https://github.com/apache/airflow/pull/74113#discussion_r4401033909
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -507,6 +507,21 @@ def _do_delete(
else:
print("Performing Delete...")
+ if skip_archive:
+ try:
+ _delete_directly(
+ limited_query=limited_query,
+ source_table_name=source_table_name,
+ session=session,
+ dialect_name=dialect_name,
+ )
+ session.commit()
+ except BaseException:
+ with suppress(Exception):
+ session.rollback()
+ raise
+ continue
Review Comment:
the continue is intentional here. The direct DELETE embeds limited_query, including its skip_if_referenced guard, so the condition is reevaluated within the DELETE itself. Unlike the archive path, there is no separately copied row that can remain both archived and live. Adding a count would introduce another query per batch and a new count/delete race. The regression test covers both whole and partial direct skips. I’ll add a comment to make this distinction clearer.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "safaehar (via GitHub)" <gi...@apache.org>.
safaehar commented on PR #74113:
URL: https://github.com/apache/airflow/pull/74113#issuecomment-5991931491
> Has the new direct-delete path been tested against MySQL, both with and without batching? We should verify that the derived-table query works in both cases, including composite primary keys.
>
> Also, I believe the CI failure should be fixed by PR #74122
@SameerMesiah97
Thanks, this is now covered. I parameterized the composite-primary-key XCom regression test to run both without batching (batch_size=None) and with batching (batch_size=2). Both cases pass against MySQL 8.0 using Airflow’s published CI image, exercising the derived-table EXISTS delete in both modes.
I also rebased onto the latest main, which includes #74122. The fresh CI run is in progress
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "safaehar (via GitHub)" <gi...@apache.org>.
safaehar commented on code in PR #74113:
URL: https://github.com/apache/airflow/pull/74113#discussion_r4270939112
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -494,44 +503,52 @@ def _do_delete(
continue
except BaseException:
- error_raised = True
- # Roll back the failed transaction so its locks are released before
- # the archive table is dropped in the ``finally`` block below.
- # ``rollback()`` itself can raise (e.g. the connection died); suppress
- # it so it does not shadow the original error being re-raised.
with suppress(Exception):
session.rollback()
raise
- finally:
- if target_table is not None and skip_archive:
- # Drop the archive table on the session's own connection. Binding
- # the drop to ``session.get_bind()`` (the Engine) would check out a
- # *second* pooled connection, and on MySQL its ``DROP TABLE`` blocks
- # indefinitely on the metadata lock still held by this session's
- # open transaction when the DELETE above failed -- the ``db clean``
- # hang reported in #66177.
- try:
- target_table.drop(bind=session.connection())
- session.commit()
- except Exception:
- # If we are already unwinding from a delete failure, a cleanup
- # error here must not replace the original exception (Python
- # makes a ``finally``-raised error the top-level one). Log and
- # let the original delete error keep propagating. On the success
- # path (no delete error), a drop/commit failure is a real
- # problem, so re-raise it.
- if not error_raised:
- raise
- logger.warning(
- "Failed to drop archive table %s while cleaning up after a "
- "delete failure; propagating the original delete error instead.",
- target_table_name,
- exc_info=True,
- )
print("Finished Performing Delete")
+def _delete_directly(
+ *,
+ limited_query: Select,
+ source_table_name: str,
+ session: Session,
+ dialect_name: str,
+) -> None:
+ metadata = reflect_tables([source_table_name], session)
+ source_table = metadata.tables[source_table_name]
+ pk_cols = list(source_table.primary_key.columns)
+ pk_query = limited_query.with_only_columns(
+ *[literal_column(f"{_BASE_TABLE_ALIAS}.{col.name}").label(col.name) for col in pk_cols]
+ )
+ rows_to_delete = pk_query.subquery("rows_to_delete")
+
+ if dialect_name == "mysql":
+ delete = source_table.delete().where(
+ select(literal(1))
+ .select_from(rows_to_delete)
+ .where(and_(*[source_table.c[col.name] == rows_to_delete.c[col.name] for col in pk_cols]))
+ .exists()
+ )
+ elif len(pk_cols) == 1:
+ pk_col = pk_cols[0]
+ delete = source_table.delete().where(
+ source_table.c[pk_col.name].in_(select(rows_to_delete.c[pk_col.name]))
+ )
+ else:
+ delete = source_table.delete().where(
+ tuple_(*[source_table.c[col.name] for col in pk_cols]).in_(
+ select(*[rows_to_delete.c[col.name] for col in pk_cols])
+ )
+ )
+
+ logger.debug("direct delete statement:\n%s", delete.compile())
+ session.execute(delete)
+ session.commit()
Review Comment:
adressed !
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on code in PR #74113:
URL: https://github.com/apache/airflow/pull/74113#discussion_r4270252543
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -494,44 +503,52 @@ def _do_delete(
continue
except BaseException:
- error_raised = True
- # Roll back the failed transaction so its locks are released before
- # the archive table is dropped in the ``finally`` block below.
- # ``rollback()`` itself can raise (e.g. the connection died); suppress
- # it so it does not shadow the original error being re-raised.
with suppress(Exception):
session.rollback()
raise
- finally:
- if target_table is not None and skip_archive:
- # Drop the archive table on the session's own connection. Binding
- # the drop to ``session.get_bind()`` (the Engine) would check out a
- # *second* pooled connection, and on MySQL its ``DROP TABLE`` blocks
- # indefinitely on the metadata lock still held by this session's
- # open transaction when the DELETE above failed -- the ``db clean``
- # hang reported in #66177.
- try:
- target_table.drop(bind=session.connection())
- session.commit()
- except Exception:
- # If we are already unwinding from a delete failure, a cleanup
- # error here must not replace the original exception (Python
- # makes a ``finally``-raised error the top-level one). Log and
- # let the original delete error keep propagating. On the success
- # path (no delete error), a drop/commit failure is a real
- # problem, so re-raise it.
- if not error_raised:
- raise
- logger.warning(
- "Failed to drop archive table %s while cleaning up after a "
- "delete failure; propagating the original delete error instead.",
- target_table_name,
- exc_info=True,
- )
print("Finished Performing Delete")
+def _delete_directly(
+ *,
+ limited_query: Select,
+ source_table_name: str,
+ session: Session,
+ dialect_name: str,
+) -> None:
+ metadata = reflect_tables([source_table_name], session)
+ source_table = metadata.tables[source_table_name]
+ pk_cols = list(source_table.primary_key.columns)
+ pk_query = limited_query.with_only_columns(
+ *[literal_column(f"{_BASE_TABLE_ALIAS}.{col.name}").label(col.name) for col in pk_cols]
+ )
+ rows_to_delete = pk_query.subquery("rows_to_delete")
+
+ if dialect_name == "mysql":
+ delete = source_table.delete().where(
+ select(literal(1))
+ .select_from(rows_to_delete)
+ .where(and_(*[source_table.c[col.name] == rows_to_delete.c[col.name] for col in pk_cols]))
+ .exists()
+ )
+ elif len(pk_cols) == 1:
+ pk_col = pk_cols[0]
+ delete = source_table.delete().where(
+ source_table.c[pk_col.name].in_(select(rows_to_delete.c[pk_col.name]))
+ )
+ else:
+ delete = source_table.delete().where(
+ tuple_(*[source_table.c[col.name] for col in pk_cols]).in_(
+ select(*[rows_to_delete.c[col.name] for col in pk_cols])
+ )
+ )
+
+ logger.debug("direct delete statement:\n%s", delete.compile())
+ session.execute(delete)
+ session.commit()
Review Comment:
Could we keep the transaction boundary in the existing batch orchestration rather than committing inside `_delete_directly()`?
The existing cleanup loop already commits batches, so this is not a new behavioral regression, but the new helper could remain responsible only for executing the DELETE while its caller preserves the commit-per-batch behavior.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #74113:
URL: https://github.com/apache/airflow/pull/74113#issuecomment-5963346549
Nice approach. Skipping the archive entirely instead of create-then-drop also removes the need for that `finally` and the workaround from #66177.
There's going to be a conflict with #73568 which adds some warnings int he same block you are editing, would you consider putting this in draft until that one merges and them rebasing on top of it? #73417 is also in the same block so there's a lot of churn in there right now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #74113:
URL: https://github.com/apache/airflow/pull/74113#discussion_r4342621250
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -605,44 +615,51 @@ def _do_delete(
)
except BaseException:
- error_raised = True
- # Roll back the failed transaction so its locks are released before
- # the archive table is dropped in the ``finally`` block below.
- # ``rollback()`` itself can raise (e.g. the connection died); suppress
- # it so it does not shadow the original error being re-raised.
with suppress(Exception):
session.rollback()
raise
- finally:
- if target_table is not None and skip_archive:
- # Drop the archive table on the session's own connection. Binding
- # the drop to ``session.get_bind()`` (the Engine) would check out a
- # *second* pooled connection, and on MySQL its ``DROP TABLE`` blocks
- # indefinitely on the metadata lock still held by this session's
- # open transaction when the DELETE above failed -- the ``db clean``
- # hang reported in #66177.
- try:
- target_table.drop(bind=session.connection())
- session.commit()
- except Exception:
- # If we are already unwinding from a delete failure, a cleanup
- # error here must not replace the original exception (Python
- # makes a ``finally``-raised error the top-level one). Log and
- # let the original delete error keep propagating. On the success
- # path (no delete error), a drop/commit failure is a real
- # problem, so re-raise it.
- if not error_raised:
- raise
- logger.warning(
- "Failed to drop archive table %s while cleaning up after a "
- "delete failure; propagating the original delete error instead.",
- target_table_name,
- exc_info=True,
- )
print("Finished Performing Delete")
+def _delete_directly(
Review Comment:
It would likely be more efficient to build the table once and pass it into this method, no? Maybe I'm missing a reason to redo it each pass like this?
##########
airflow-core/src/airflow/utils/db_cleanup.py:
##########
@@ -507,6 +507,21 @@ def _do_delete(
else:
print("Performing Delete...")
+ if skip_archive:
+ try:
+ _delete_directly(
+ limited_query=limited_query,
+ source_table_name=source_table_name,
+ session=session,
+ dialect_name=dialect_name,
+ )
+ session.commit()
+ except BaseException:
+ with suppress(Exception):
+ session.rollback()
+ raise
+ continue
Review Comment:
As written, this continue is right for the fix but will cause another because it makes the whole skip_if_referenced block around L613 unreachable.
One fix would be to have `_delete_directly` return the number of rows deleted and compare that against a count of the selection, warning when it deleted fewer than it found. `deleted == 0` won't do, since L597-600 rejects it for staying silent on a partial skip. You can throw the warning at L604 into a helper that both paths can call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on PR #74113:
URL: https://github.com/apache/airflow/pull/74113#issuecomment-6220563266
https://github.com/apache/airflow/pull/73568 merged. Ping me when you get the CI passing on this one and I'll give it a proper review. Thanks for your patience!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Make db clean --skip-archive delete rows directly [airflow]
Posted by "safaehar (via GitHub)" <gi...@apache.org>.
safaehar commented on PR #74113:
URL: https://github.com/apache/airflow/pull/74113#issuecomment-6251679321
> #73568 merged. Ping me when you get the CI passing on this one and I'll give it a proper review. Thanks for your patience!
@ferruzzi CI passed, thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org