Apache Airflow Documents LLM Approval And Cuts Metadata Memory

Apache Airflow is the Python DAG orchestrator that still sits under a large share of batch ETL. In the last seven days, main received 139 commits and 1279 files changed. The lockfile and CI noise is real, but operators should look at how LLMFileAnalysisOperator is meant to pause for a person, how metadata deletes and terminal Dag runs stop paying per row, and a first IBM Db2 hook that is still marked not-ready.

The common AI provider already shipped LLMFileAnalysisOperator for sending a stored file to a model. This week the project documents the LLM and human in the loop (HITL) parameters the operator inherits from LLMOperator: require_approval, approval_timeout, and allow_modifications. Set require_approval=True and the task pauses after analysis until a reviewer accepts the output. allow_modifications=True lets that reviewer edit the text before it is returned.

The common AI docs now include an example operators can copy:

LLMFileAnalysisOperator(
    task_id="analyze_contract_with_approval",
    prompt="Summarize the key obligations and flag any unusual termination clauses.",
    llm_conn_id="pydanticai_default",
    file_path="s3://legal/contracts/vendor-agreement.pdf",
    file_conn_id="aws_default",
    require_approval=True,
    approval_timeout=timedelta(hours=1),
    allow_modifications=True,
)

A related commit adds .md to the supported formats, including gzipped markdown. Markdown files in object storage no longer need a rename step before analysis.

Vendor hosted agents got a shared tool contract rather than another operator facade. BaseManagedAgentToolset covers the case where Bedrock AgentCore, Azure AI Foundry, or Vertex AI Agent Engine runs the loop on the vendor side. Airflow can only consult that loop. Errors split into three buckets: the model can rephrase, the call is terminal, or the task should retry. Replay is opt in per implementation, because a cached answer can skip a side effect the orchestrator cannot see.

Clicking Test on a LangChain or LlamaIndex connection used to report that the hook had no test_connection method. Those hooks now implement it, matching PydanticAIHook and MCPHook.

TaskInstance.set_state is wrapped with @provide_session. When the terminal state path skipped pending tasks without passing a session, every mapped child opened its own connection: a refresh SELECT, a merge SELECT, an UPDATE, and a COMMIT. On an idle PostgreSQL 17 host with a 2.5 ms RTT, marking one Dag run failed across 3917 mapped task instances took 280 seconds, about 19500 statements and 3900 connections. That is the documented sql_alchemy_pool_enabled=False plus pgbouncer layout, not a misconfigured pool.

Passing the ambient session drops the per task instance cost from 79 ms to 7.5 ms on the same run:

ti.set_state(TaskInstanceState.SKIPPED, session=session)

The Dag run state update and the running task path already used that session, so commit boundaries do not change. Large mapped arrays win. Nothing in the commit claims a behavior change for teardown handling.

Deletes had the same shape of waste. delete_dag forced SQLAlchemy synchronize_session="fetch" on every bulk delete. Fetch reads every deleted primary key back to update an identity map that only held the DagModel row. The extra heap was about 211 bytes per row on PostgreSQL, or about 1 GiB in the API server for a Dag with five million task instances. The delete path now uses the default strategy. The same fetch flag was dropped from queued asset event deletes in the public assets route.

A new provider package, apache-airflow-providers-ibm-db2, adds Db2Hook and a Db2 dialect. Connection type is ibmdb2. Default port is 50000. The hook speaks ibm_db_dbi, tests with SELECT 1 FROM SYSIBM.SYSDUMMY1, and maps extra JSON keys onto the Db2 connection string in uppercase (SECURITY=SSL, SSLServerCertificate, CurrentSchema).

Db2 has no REPLACE INTO. The dialect emits MERGE INTO ... USING (VALUES (...)) and refuses to build the statement without a primary key or an explicit replace_index. Identity columns marked autoincrement=True are stripped from insert column lists so GENERATED ALWAYS AS IDENTITY does not break bulk loads.

provider.yaml still says state: not-ready and lifecycle: incubation at 0.1.0. Treat this as a preview hook, not something to point SQLExecuteQueryOperator at in production this week.

SageMaker operators inherit region_name, verify, and botocore_config from AwsBaseOperator. They did not pass those fields to the trigger after deferral. The second half of the task then hit AWS with defaults. An explicit region, a custom botocore config, or an explicit SSL verify setting stopped applying once the triggerer took over.

A few other provider fixes in the same window are the kind that only show up after a Dag has been running:

  • MSGraphAsyncOperator dropped path_parameters on later pages, so a URL template such as users/{user_id}/mailFolders/{mailFolder_id}/messages lost its IDs after the first page.
  • InfluxDBHook methods crashed with AttributeError if a Dag called them before get_conn(). The default port was also Neo4j Bolt (7687). HTTP now defaults to 8086 and HTTPS to 443.
  • Connection Test in the UI hid request errors behind a red icon instead of the shared error toaster.

The version inflation checker stopped treating every nested with statement as the end of the Dag. Tasks after a nested with open(...) inside a Dag block were invisible to the runtime varying value scan, so datetime.now() and random() in those tasks were never flagged. If that checker is part of Dag review, rerun it against files that open handles during parse.

Variable.set no longer rewrites team_name on update. Under core.multi_team, a later set that passed a team name used to move the variable. Existing team assignments now stick; passing team_name on update is a no op.

Dag cache metrics used to emit under api_server.dag_bag.* even when the scheduler owned the cache. Operators comparing hit rate across components were reading mixed series. New code requires an explicit namespace at construction.

Budget triggerer capacity for HITL waits, not just worker slots. The example timeout is one hour. The IBM Db2 provider still needs a ready state and a real release before it belongs on a production SQL path.