Add rows processor to GenericTransfer by dabla · Pull Request #61143 · apache/airflow

dabla · 2026-01-27T18:47:58Z

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

This PR extends the GenericTransfer operator with a rows_processor parameter, similar to SQLInsertRowsOperator.
The rows_processor allows users to post-process rows returned by the source hook before they are inserted into the destination hook.

This enables lightweight transformations (filtering, normalization, enrichment, etc.) without introducing an intermediate task, while keeping database-specific type handling encapsulated in the hook itself.

def rows_processor(rows, **context):
    # Example: normalize string values before insertion
    return [
        tuple(value.strip() if isinstance(value, str) else value for value in row)
        for row in rows
    ]

GenericTransfer(
    task_id="transfer_data",
    source_conn_id="source_db",
    destination_conn_id="target_db",
    sql="SELECT id, name FROM source_table",
    destination_table="target_table",
    rows_processor=rows_processor,
)

In addition, this PR makes SQLExecuteQueryTrigger non-blocking when running on Airflow 3.2 or higher.

Previously, the trigger had to remain blocking due to a bug in the CommsDecoder that could raise a “Response read out of order” error when handling asynchronous trigger events. With this issue fixed in Airflow 3.2, the trigger can now safely run in a non-blocking mode, improving triggerer scalability and resource usage.

Read the 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.
For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

…wsOperator

Nataneljpwd

Looks great, I have left a few comments and questions, mainly nitpicks

…onstructor of GenericTransfer

…onstructor of SQLInsertRowsOperator

Nataneljpwd

Looks great!
Only 1 question regarding the python interface file

dabla · 2026-02-01T10:34:11Z

Looks great! Only 1 question regarding the python interface file

Thx for to review @Nataneljpwd, really appreciated, well done! I think I've fixed all issues, awaiting CI/CD now.

Nataneljpwd

Looks great, 1 minor nitpick for type hinting, other than that, looks great

… removed override of render_template_fields

dabla · 2026-02-03T12:37:50Z

@Nataneljpwd thx for the review and helping fixing the mypy issues!

Nataneljpwd

Everything looks amazing, nice work!

…s/sql.py Co-authored-by: Shahar Epstein <60007259+shahar1@users.noreply.github.com>

* refactor: Fixed docstring for rows_processor parameter in SQLInsertRowsOperator * refactor: Added rows_processor parameter in GenericTransfer * refactor: Make SQLExecuteQueryTrigger non-blocking from Airflow 3.2+ * refactor: Make inserting of rows in sync and deferred mode more DRY * refactor: Updated GenericTransfer type * refactor: Ignore mypy for _rows_processor * refactor: Removed _process_rows and made rows_processor optional in constructor of GenericTransfer * refactor: Removed _process_rows and made rows_processor optional in constructor of SQLInsertRowsOperator * refactor: Made _get_records in trigger protected * refactor: Updated typing of rows_processor in GenericTransfer interface * refactor: Fixed typing of rows parameter in _insert_rows method * refactor: Refactored unit test for SQLInsertRowsOperator * refactor: Fixted typing of _insert_rows method * refactor: Updated docstring for rows_processor parameter * refactor: Changed typing of context parameter from Context to Any in rows_processor parameter * refactor: Changed typing of rows_processor * refactor: Refactored _insert_rows method in SQLInsertRowsOperator and removed override of render_template_fields * refactor: Removed unused import of jinja2 * docs: Added comment to explain why rows_processor is typed like this * Update providers/common/sql/src/airflow/providers/common/sql/operators/sql.py Co-authored-by: Shahar Epstein <60007259+shahar1@users.noreply.github.com> * refactor: Removed get_records * refactor: Evaluate XComArgs first before evaluating rows * refactor: Fixed check on rows in SQLInsertRowsOperator --------- Co-authored-by: David Blain <david.blain@b-holding.be> Co-authored-by: Shahar Epstein <60007259+shahar1@users.noreply.github.com>

dabla added 3 commits January 27, 2026 18:39

refactor: Fixed docstring for rows_processor parameter in SQLInsertRo…

dc7b9f6

…wsOperator

refactor: Added rows_processor parameter in GenericTransfer

d37a16c

refactor: Make SQLExecuteQueryTrigger non-blocking from Airflow 3.2+

18e0560

boring-cyborg Bot added area:providers provider:common-sql labels Jan 27, 2026

dabla added 6 commits January 27, 2026 19:48

Merge branch 'main' into feature/add-rows-processor-generic-transfer

2fdba24

Merge branch 'main' into feature/add-rows-processor-generic-transfer

2a47f9a

refactor: Make inserting of rows in sync and deferred mode more DRY

8edc348

refactor: Updated GenericTransfer type

5e4c530

refactor: Ignore mypy for _rows_processor

18d9c6d

Merge branch 'main' into feature/add-rows-processor-generic-transfer

bca6b89

dabla requested a review from ephraimbuddy January 28, 2026 19:17

dabla added 2 commits January 28, 2026 20:17

Merge branch 'main' into feature/add-rows-processor-generic-transfer

9541a74

Merge branch 'main' into feature/add-rows-processor-generic-transfer

bb87f54

Nataneljpwd reviewed Jan 30, 2026

View reviewed changes

dabla added 4 commits January 31, 2026 12:31

refactor: Removed _process_rows and made rows_processor optional in c…

7c3a2e6

…onstructor of GenericTransfer

refactor: Removed _process_rows and made rows_processor optional in c…

1f31246

…onstructor of SQLInsertRowsOperator

refactor: Made _get_records in trigger protected

d5c9017

Merge branch 'main' into feature/add-rows-processor-generic-transfer

a670485

Nataneljpwd approved these changes Jan 31, 2026

View reviewed changes

Comment thread providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.pyi Outdated

dabla added 2 commits February 1, 2026 11:31

refactor: Updated typing of rows_processor in GenericTransfer interface

4930c9a

Merge branch 'main' into feature/add-rows-processor-generic-transfer

64eb707

dabla added 3 commits February 1, 2026 11:34

Merge branch 'main' into feature/add-rows-processor-generic-transfer

b5fbe45

refactor: Fixed typing of rows parameter in _insert_rows method

2c2164a

refactor: Refactored unit test for SQLInsertRowsOperator

68df300

dabla requested a review from Nataneljpwd February 1, 2026 16:33

Nataneljpwd reviewed Feb 1, 2026

View reviewed changes

Comment thread providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py Outdated

dabla added 2 commits February 1, 2026 20:50

refactor: Fixted typing of _insert_rows method

b973c03

refactor: Updated docstring for rows_processor parameter

aa8aafd

dabla added 3 commits February 2, 2026 15:02

refactor: Refactored _insert_rows method in SQLInsertRowsOperator and…

23869fe

… removed override of render_template_fields

refactor: Removed unused import of jinja2

ec1ff73

Merge branch 'main' into feature/add-rows-processor-generic-transfer

e6f52a7

dabla requested a review from Nataneljpwd February 2, 2026 18:21

Nataneljpwd reviewed Feb 2, 2026

View reviewed changes

Comment thread providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py

dabla added 4 commits February 2, 2026 20:58

Merge branch 'main' into feature/add-rows-processor-generic-transfer

a3eeb84

Merge branch 'main' into feature/add-rows-processor-generic-transfer

1828cc9

docs: Added comment to explain why rows_processor is typed like this

1627210

Merge branch 'main' into feature/add-rows-processor-generic-transfer

80fe566

Nataneljpwd approved these changes Feb 4, 2026

View reviewed changes

shahar1 reviewed Feb 6, 2026

View reviewed changes

dabla and others added 5 commits February 6, 2026 14:18

Update providers/common/sql/src/airflow/providers/common/sql/operator…

55e91da

…s/sql.py Co-authored-by: Shahar Epstein <60007259+shahar1@users.noreply.github.com>

refactor: Removed get_records

6980f9f

refactor: Evaluate XComArgs first before evaluating rows

2477b5b

Merge branch 'main' into feature/add-rows-processor-generic-transfer

51be7a5

refactor: Fixed check on rows in SQLInsertRowsOperator

c1e2b33

dabla requested a review from shahar1 February 6, 2026 17:39

potiuk approved these changes Feb 10, 2026

View reviewed changes

potiuk merged commit d70216c into apache:main Feb 10, 2026
90 checks passed

shahar1 mentioned this pull request Feb 11, 2026

Status of testing Providers that were prepared on February 10, 2026 #61766

Closed

81 tasks

Conversation

dabla commented Jan 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Was generative AI tooling used to co-author this PR?

Uh oh!

Nataneljpwd left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Nataneljpwd left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

dabla commented Feb 1, 2026

Uh oh!

Nataneljpwd left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

dabla commented Feb 3, 2026

Uh oh!

Nataneljpwd left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

dabla commented Jan 27, 2026 •

edited

Loading