aoelvp94 commented on code in PR #76363:
URL: https://github.com/apache/airflow/pull/76363#discussion_r4454297640
##########
providers/apache/arrow/src/airflow/providers/apache/arrow/hooks/adbc.py:
##########
@@ -276,27 +274,30 @@ def insert_rows(
replace=False,
*,
executemany=False,
- fast_executemany=False,
+ fast_executemany=True,
Review Comment:
Good catch. I went with the fallback so the default stays safe:
`fast_executemany` is now a hint, like on PostgresHook. If the first
`adbc_ingest` raises `NotSupportedError`, that batch and the rest are bound
with `executemany` instead, with one info log. A `NotSupportedError` on a later
batch, or any other ingest error, is still raised, so real failures aren't
hidden. Tests cover all three cases, and `adbc.rst` Usage now says ingest is
the default and `fast_executemany=False` binds.
https://github.com/apache/airflow/commit/4f3420ec4bac0467cb14a9f9dc28e43b88ed8a94
##########
providers/apache/arrow/src/airflow/providers/apache/arrow/hooks/adbc.py:
##########
@@ -238,34 +238,32 @@ def _to_record_batch(cls, rows, schema: Schema) ->
RecordBatch:
schema=schema,
)
- @classmethod
- def _execute_executemany(cls, cursor, statement: str, record_batch:
RecordBatch) -> None:
- """
- Execute a statement using cursor.executemany with a RecordBatch.
-
- ADBC's executemany() accepts Arrow data directly and routes it through
- bind_stream internally, making this the Arrow-native fast path.
- """
- cursor.executemany(statement, record_batch)
-
def _resolve_execute_batch(
- self, cursor, executemany: bool, fast_executemany: bool
- ) -> Callable[[Cursor, str, RecordBatch], None]:
+ self,
+ cursor: Cursor,
+ *,
+ table_name: str,
+ schema_name: str | None,
+ sql: str,
+ replace: bool,
+ fast_executemany: bool,
+ ) -> Callable[[RecordBatch], object]:
"""
- Return the batch-execute callable for the given cursor.
-
- ADBC's executemany() natively accepts a RecordBatch, so it is always
- the fast path. fast_executemany is passed through to drivers that
- expose that knob (silently ignored otherwise).
+ Return the callable that loads one RecordBatch through ``cursor``.
+
+ With ``fast_executemany`` the batch goes through the driver's bulk
ingest
+ (:meth:`adbc_driver_manager.dbapi.Cursor.adbc_ingest`, ``COPY`` on
PostgreSQL).
+ Otherwise it is bound to the generated ``INSERT`` statement with
+ ``cursor.executemany``, which most drivers execute once per row.
+ ``replace=True`` always binds, because bulk ingest cannot carry the
+ dialect's conflict clause.
"""
- if fast_executemany:
- with contextlib.suppress(AttributeError):
- cursor.fast_executemany = True
- self.log.info(
- "Fast_executemany is enabled for conn_id '%s'!",
- self.get_conn_id(),
- )
- return self._execute_executemany
+ if fast_executemany and replace:
+ self.log.info("Bulk ingest cannot replace rows; binding them to
the statement instead")
Review Comment:
Agreed, moved it to `log.debug`, since the caller never asked for ingest.
Added a test that checks it no longer logs at info level.
https://github.com/apache/airflow/commit/4f3420ec4bac0467cb14a9f9dc28e43b88ed8a94
--
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]
