potiuk commented on code in PR #64321:
URL: https://github.com/apache/airflow/pull/64321#discussion_r3173443438
##########
providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py:
##########
@@ -55,6 +55,7 @@ class GenericTransfer(BaseOperator):
:param insert_args: extra params for `insert_rows` method.
:param page_size: number of records to be read in paginated mode
(optional).
:param paginated_sql_statement_clause: SQL statement clause to be used for
pagination (optional).
+ :param deferrable: Run operator in the deferrable mode
Review Comment:
nit: this only takes effect in paginated mode (`page_size` set + `sql` is a
`str`); the non-paginated branch ignores it. Worth one extra clause so users
don't expect it to control non-paginated execution too — e.g. `Run the operator
in deferrable mode (only effective in paginated mode, i.e. when page_size is
set and sql is a string).`
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py:
##########
@@ -159,14 +162,29 @@ def execute(self, context: Context):
self.destination_hook.run(self.preoperator)
if self.page_size and isinstance(self.sql, str):
- self.defer(
- trigger=SQLExecuteQueryTrigger(
- conn_id=self.source_conn_id,
- hook_params=self.source_hook_params,
- sql=self.get_paginated_sql(0),
- ),
- method_name=self.execute_complete.__name__,
- )
+ if self.deferrable:
+ self.defer(
+ trigger=SQLExecuteQueryTrigger(
+ conn_id=self.source_conn_id,
+ hook_params=self.source_hook_params,
+ sql=self.get_paginated_sql(0),
+ ),
+ method_name=self.execute_complete.__name__,
+ )
+ else:
+ offset = 0
+ while True:
+ paginated_sql = self.get_paginated_sql(offset)
+ self.log.info("Executing: \n %s", paginated_sql)
+ if rows := self.source_hook.get_records(paginated_sql):
+ self._insert_rows(rows=rows, context=context)
+ offset += self.page_size
Review Comment:
nit: the deferred branch's `execute_complete` logs `"Offset increased to
%d"` after each page; the new sync loop has no equivalent. A
`self.log.info("Offset increased to %d", offset)` here keeps the two paths'
task logs comparable for users debugging long transfers.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
##########
providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py:
##########
@@ -159,14 +162,29 @@ def execute(self, context: Context):
self.destination_hook.run(self.preoperator)
if self.page_size and isinstance(self.sql, str):
- self.defer(
- trigger=SQLExecuteQueryTrigger(
- conn_id=self.source_conn_id,
- hook_params=self.source_hook_params,
- sql=self.get_paginated_sql(0),
- ),
- method_name=self.execute_complete.__name__,
- )
+ if self.deferrable:
+ self.defer(
+ trigger=SQLExecuteQueryTrigger(
+ conn_id=self.source_conn_id,
+ hook_params=self.source_hook_params,
+ sql=self.get_paginated_sql(0),
+ ),
+ method_name=self.execute_complete.__name__,
+ )
+ else:
+ offset = 0
+ while True:
+ paginated_sql = self.get_paginated_sql(offset)
+ self.log.info("Executing: \n %s", paginated_sql)
+ if rows := self.source_hook.get_records(paginated_sql):
+ self._insert_rows(rows=rows, context=context)
Review Comment:
nit (optimisation): when the total row count is an exact multiple of
`page_size`, the loop always issues one trailing empty query before
terminating. Cheap to short-circuit:
```python
if rows := self.source_hook.get_records(paginated_sql):
self._insert_rows(rows=rows, context=context)
if len(rows) < self.page_size:
break
offset += self.page_size
else:
...
```
Fine to leave for a follow-up if you'd rather keep the diff minimal.
---
Drafted-by: Claude Opus 4.7; reviewed by @potiuk before posting
--
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]
