Repository: apache/airflow
File: providers/common/sql/src/airflow/providers/common/sql/operators/generic_transfer.py (Python)

Code before:
        "source_conn_id",
        "destination_conn_id",
        "sql",
        "destination_table",
        "preoperator",
        "insert_args",
        "page_size",
        "paginated_sql_statement_clause",
    )
    template_ext: Sequence[str] = (
        ".sql",
        ".hql",
    )
    template_fields_renderers = {"preoperator": "sql"}
    ui_color = "#b0f07c"

    def __init__(
        self,
        *,
        sql: str | list[str],
        destination_table: str,
        source_conn_id: str,
        source_hook_params: dict | None = None,
        destination_conn_id: str,
        destination_hook_params: dict | None = None,
        rows_processor: Callable[[Any, Context], Any] = lambda rows, **context: rows,
        preoperator: str | list[str] | None = None,
        insert_args: dict | None = None,
        page_size: int | None = None,
        paginated_sql_statement_clause: str | None = None,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.sql = sql
        self.destination_table = destination_table
        self.source_conn_id = source_conn_id
        self.source_hook_params = source_hook_params
        self.destination_conn_id = destination_conn_id
        self.destination_hook_params = destination_hook_params
        self._rows_processor = rows_processor
        self.preoperator = preoperator
        self.insert_args = insert_args or {}
        self.page_size = page_size
        self.paginated_sql_statement_clause = paginated_sql_statement_clause or "{} LIMIT {} OFFSET {}"

    @classmethod
    def get_hook(cls, conn_id: str, hook_params: dict | None = None) -> DbApiHook:
        """
        Return DbApiHook for this connection id.

        :param conn_id: connection id

Review comment (performance):
Maybe it is a better idea to make it nullable? I can see it becoming a performance overhead for a large number of rows

Code after:

    template_fields: Sequence[str] = (
        "source_conn_id",
        "destination_conn_id",
        "sql",
        "destination_table",
        "preoperator",
        "insert_args",
        "page_size",
        "paginated_sql_statement_clause",
    )
    template_ext: Sequence[str] = (
        ".sql",
        ".hql",
    )
    template_fields_renderers = {"preoperator": "sql"}
    ui_color = "#b0f07c"

    def __init__(
        self,
        *,
        sql: str | list[str],
        destination_table: str,
        source_conn_id: str,
        source_hook_params: dict | None = None,
        destination_conn_id: str,
        destination_hook_params: dict | None = None,
        rows_processor: Callable[..., list[Any]] | None = None,
        # rows_processor is called as rows_processor(rows, **context);
        # context keys vary, so Callable[..., list[Any]] is intentional.
        preoperator: str | list[str] | None = None,
        insert_args: dict | None = None,
        page_size: int | None = None,
        paginated_sql_statement_clause: str | None = None,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.sql = sql
        self.destination_table = destination_table
        self.source_conn_id = source_conn_id
        self.source_hook_params = source_hook_params
        self.destination_conn_id = destination_conn_id
        self.destination_hook_params = destination_hook_params
        self._rows_processor = rows_processor
        self.preoperator = preoperator
        self.insert_args = insert_args or {}
        self.page_size = page_size
        self.paginated_sql_statement_clause = paginated_sql_statement_clause or "{} LIMIT {} OFFSET {}"

    @classmethod
    def get_hook(cls, conn_id: str, hook_params: dict | None = None) -> DbApiHook: