Add Apache Arrow provider

[dabla]

Apache Arrow provider (apache-airflow-providers-apache-arrow)

Adds a new Airflow provider backed by Apache Arrow and the
Arrow Database Connectivity (ADBC) standard.

What's included

AdbcHook

A general-purpose DbApiHook subclass that connects to any ADBC-compatible database via the
adbc-driver-manager. It exposes the full Airflow SQL hook surface (get_records, run,
insert_rows, …) while internally operating on Arrow RecordBatch objects for zero-copy,
columnar data transfer.

Key capabilities:

• Arrow-native inserts — insert_rows converts rows to Arrow RecordBatch objects and
  uses the cursor's native bind API when available, falling back to executemany otherwise.
• Chunked bulk loading — rows are committed in configurable chunks (commit_every) to
  bound transaction size and lock duration.
• Dynamic driver discovery — the driver shared library is located automatically from the
  installed wheel; the entrypoint can also be overridden via connection extras.
• Dialect-aware — SQL placeholder style is normalised per-dialect before execution.
• Connection extras — driver, entrypoint, db_kwargs, conn_kwargs, and dialect
  are all configurable from the Airflow connection UI without code changes.

Supported databases (via optional extras)

Extra	Driver package
sqlite	adbc-driver-sqlite
postgresql	adbc-driver-postgresql
snowflake	adbc-driver-snowflake
bigquery	adbc-driver-bigquery
flightsql	adbc-driver-flightsql

Installation

# Core provider only
pip install apache-airflow-providers-apache-arrow

# With a specific driver
pip install "apache-airflow-providers-apache-arrow[postgresql]"

# With all bundled drivers
pip install "apache-airflow-providers-apache-arrow[all]"

Requirements

• apache-airflow >= 2.11.0
• apache-airflow-providers-common-sql >= 1.28.2
• adbc-driver-manager >= 1.7.0
• pyarrow >= 16.1.0 (Python < 3.13) / >= 18.0.0 (Python ≥ 3.13)
• more-itertools >= 9.0.0

Link to the discussion thread on the devlist: https://lists.apache.org/thread/v1nxnr5gocvzs12pknjxkw64mrtpg2o0

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[potiuk]

Just for anyone looking here - this is a draft for discussion between me, @dabla and @zeroshade - we will still need to start a DISCUSSION thread for the new provider - and we think Arrow and ADBC is a good addition. But we have to first discuss the approach :)

[vikramkoka]

As @potiuk mentioned, I believe this needs a devlist conversation first.
Looking forward to it

[potiuk]

As @potiuk mentioned, I believe this needs a devlist conversation first. Looking forward to it

Yep. This one is mostly to gather learnigs, get feedback from @zeroshade and see how we can turn it into a "convincing" devlist proposal - by showing some use cases and small POC of implementation and what it allows :).

We'll experiment a bit with it and gather our thoughts and see what can come out of it.

[dabla]

Just for anyone looking here - this is a draft for discussion between me, @dabla and @zeroshade - we will still need to start a DISCUSSION thread for the new provider - and we think Arrow and ADBC is a good addition. But we have to first discuss the approach :)

A discussion was already started on the devlist in beginning of march this year.

[kaxil]

The filtered schema keeps the table's column order, but _generate_insert_sql below emits the column list in target_fields order, and _to_record_batch indexes rows positionally against this schema. When the two orders differ, values land in the wrong columns.

With target_fields=["last", "first"] against a table declared (first, last), the filtered schema comes back as [first, last], so the SQL says INSERT INTO t (last, first) while the RecordBatch is {first: 'Smith', last: 'Alice'}. Same-typed columns silently swap; differently-typed ones blow up with ArrowInvalid: Could not convert 'Alice' with type str: tried to convert to int64.

Ordering the filtered schema by target_fields instead of by the table schema fixes both cases:

fields = {field.name: field for field in table_schema}
table_schema = schema([fields[name] for name in target_fields])

[kaxil]

The docstring says commit_every=0 inserts all rows in one transaction, but more_itertools.chunked(rows, 0) yields no chunks at all, so nothing is inserted and the method logs "Loaded a total of 0 rows" without raising. DbApiHook.insert_rows reads 0 as "don't commit mid-way", so this also diverges from the base class.

chunked(rows, commit_every or None) restores the documented behaviour, since chunked(rows, None) yields a single chunk containing everything.

[kaxil]

adbc_driver_manager.dbapi.Cursor has no bind method, neither on 1.11.0 nor on the >=1.7.0 floor declared in pyproject.toml (its public surface is adbc_ingest, adbc_prepare, execute, executemany, ...). So hasattr(cursor, "bind") is always False against a real driver, _resolve_execute_batch always returns _execute_executemany, and both _execute_native_bind and the "Native Arrow bind supported!" log are unreachable in production.

test_insert_rows_native_bind only passes because setup_method assigns self.cur.bind = mock.MagicMock() onto a plain MagicMock, so the test greenlights a path that cannot execute. Since the Arrow-native transfer is the headline of the provider, worth pinning down what the intended fast path is: cursor.adbc_ingest, or executemany with the RecordBatch (which does work, I checked it against the SQLite driver)?

[kaxil]

Routing through to_pydict() collapses duplicate column names, so get_records silently drops columns whenever a query selects the same name twice. SELECT a.id, b.id FROM a JOIN b ... produces an Arrow table with two id columns, but to_pydict() keeps only the last, and the zip then yields 1-tuples where every other DbApiHook returns 2-tuples.

Iterating the columns positionally avoids it and behaves the same for the normal and empty-result cases:

table = cursor.fetch_arrow_table()
return list(zip(*(column.to_pylist() for column in table.columns)))

[kaxil]

autocommit is pinned to False here and there's no set_autocommit override, so with supports_autocommit = True the base _create_autocommit_connection ends up running conn.autocommit = autocommit against an ADBC dbapi.Connection. That only sets an unused Python attribute: after conn.autocommit = True the connection is still in manual-commit mode, and conn.commit() keeps succeeding instead of erroring. insert_rows(..., autocommit=True) and run(..., autocommit=True) therefore do nothing at the driver level.

ADBC's actual lever is the autocommit= kwarg on connect() or adbc_connection.set_options(**{"adbc.connection.autocommit": "true"}). Worth noting that wiring it up properly also needs the conn.commit() inside the insert_rows chunk loop made conditional: once autocommit is really on, SQLite raises ProgrammingError: INVALID_STATE: No active transaction, cannot commit on that call.

[kaxil]

These four keys aren't ADBC option names. Against the real SQLite driver each one is rejected: NotSupportedError: NOT_IMPLEMENTED: [SQLite] Unknown connection option autocommit='true', and the same for read_only, current_catalog and current_db_schema. The dotted form is the canonical spelling, and of the four only adbc.connection.autocommit is actually accepted.

The db_kwargs table above has the mirror-image problem: adbc.connection.autocommit is listed there, but it's a connection option and belongs under conn_kwargs. timeout (the key test_get_conn_forwards_db_kwargs asserts) is rejected as an unknown database option too. The tests can't catch any of this because they patch connect, so the passthrough assertions pass while the documented values would fail against a driver. Distinct from the earlier db_kwargs threads, which were about whether to forward options at all rather than which names are valid.