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

Code before:
        return self.get_hook(conn_id=self.source_conn_id, hook_params=self.source_hook_params)

    @cached_property
    def destination_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.destination_conn_id, hook_params=self.destination_hook_params)

    def get_paginated_sql(self, offset: int) -> str:
        """Format the paginated SQL statement using the current format."""
        return self.paginated_sql_statement_clause.format(self.sql, self.page_size, offset)

    def render_template_fields(
        self,
        context: Context,
        jinja_env: jinja2.Environment | None = None,
    ) -> None:
        super().render_template_fields(context=context, jinja_env=jinja_env)

        # Make sure string are converted to integers
        if isinstance(self.page_size, str):
            self.page_size = int(self.page_size)
        commit_every = self.insert_args.get("commit_every")
        if isinstance(commit_every, str):
            self.insert_args["commit_every"] = int(commit_every)

    def _process_rows(self, rows: list[Any], context: Context):
        return self._rows_processor(rows, **context)  # type: ignore

    def _insert_rows(self, rows: list[Any], context: Context):
        rows = self._process_rows(rows=rows, context=context)

        self.log.info("Inserting %d rows into %s", len(rows), self.destination_conn_id)

        self.destination_hook.insert_rows(table=self.destination_table, rows=rows, **self.insert_args)

    def execute(self, context: Context):
        if self.preoperator:
            self.log.info("Running preoperator")
            self.log.info(self.preoperator)
            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__,
            )
        else:
            if isinstance(self.sql, str):

Review comment (suggestion):
An optimization could be to check if it is null, and if so, just skip processing 

Code after:
        return hook

    @cached_property
    def source_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.source_conn_id, hook_params=self.source_hook_params)

    @cached_property
    def destination_hook(self) -> DbApiHook:
        return self.get_hook(conn_id=self.destination_conn_id, hook_params=self.destination_hook_params)

    def get_paginated_sql(self, offset: int) -> str:
        """Format the paginated SQL statement using the current format."""
        return self.paginated_sql_statement_clause.format(self.sql, self.page_size, offset)

    def render_template_fields(
        self,
        context: Context,
        jinja_env: jinja2.Environment | None = None,
    ) -> None:
        super().render_template_fields(context=context, jinja_env=jinja_env)

        # Make sure string are converted to integers
        if isinstance(self.page_size, str):
            self.page_size = int(self.page_size)
        commit_every = self.insert_args.get("commit_every")
        if isinstance(commit_every, str):
            self.insert_args["commit_every"] = int(commit_every)

    def _insert_rows(self, rows: list[Any], context: Context):
        if self._rows_processor:
            rows = self._rows_processor(rows, **context)

        self.log.info("Inserting %d rows into %s", len(rows), self.destination_conn_id)

        self.destination_hook.insert_rows(table=self.destination_table, rows=rows, **self.insert_args)

    def execute(self, context: Context):
        if self.preoperator:
            self.log.info("Running preoperator")
            self.log.info(self.preoperator)
            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__,
            )