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

Code before:
        **kwargs,
    ):
        super().__init__(
            conn_id=conn_id,
            database=database,
            hook_params=hook_params,
            **kwargs,
        )
        self.table_name = table_name
        self.schema = schema
        self._columns: list | None = list(columns) if columns else None
        self.ignored_columns = set(ignored_columns or {})
        self.rows = rows or []
        self._rows_processor = rows_processor
        self.preoperator = preoperator
        self.postoperator = postoperator
        self.insert_args = insert_args or {}
        self.do_xcom_push = False

    @property
    def table_name_with_schema(self) -> str:
        if self.schema is not None:
            return f"{self.schema}.{self.table_name}"
        return self.table_name

    @cached_property
    def columns(self):
        if self._columns is None:
            self._columns = self.get_db_hook().dialect.get_column_names(self.table_name_with_schema)
        return self._columns

    @property
    def column_names(self) -> list[str]:
        if self.ignored_columns:
            return [column for column in self.columns if column not in self.ignored_columns]
        return self.columns

    def _insert_rows(self, context: Context):
        rows = self.rows.resolve(context=context) if isinstance(self.rows, XComArg) else self.rows

        if self._rows_processor:
            rows = self._rows_processor(rows, **context)

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

        self.get_db_hook().insert_rows(
            table=self.table_name_with_schema,
            rows=rows,
            target_fields=self.column_names,
            **self.insert_args,
        )

Review comment (suggestion):
Before removing this, XComArgs were resolved before `execute()`. Now `self.rows` could be an `XComArg` when `if not self.rows` runs, and `XComArg` is always truthy, so the skip-on-empty check won’t fire (e.g., and XCom resolving to an empty list).

Code after:
        **kwargs,
    ):
        super().__init__(
            conn_id=conn_id,
            database=database,
            hook_params=hook_params,
            **kwargs,
        )
        self.table_name = table_name
        self.schema = schema
        self._columns: list | None = list(columns) if columns else None
        self.ignored_columns = set(ignored_columns or {})
        self.rows = rows or []
        self._rows_processor = rows_processor
        self.preoperator = preoperator
        self.postoperator = postoperator
        self.insert_args = insert_args or {}
        self.do_xcom_push = False

    @property
    def table_name_with_schema(self) -> str:
        if self.schema is not None:
            return f"{self.schema}.{self.table_name}"
        return self.table_name

    @cached_property
    def columns(self):
        if self._columns is None:
            self._columns = self.get_db_hook().dialect.get_column_names(self.table_name_with_schema)
        return self._columns

    @property
    def column_names(self) -> list[str]:
        if self.ignored_columns:
            return [column for column in self.columns if column not in self.ignored_columns]
        return self.columns

    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.conn_id)

        self.get_db_hook().insert_rows(
            table=self.table_name_with_schema,
            rows=rows,
            target_fields=self.column_names,
            **self.insert_args,
        )

    def execute(self, context: Context) -> Any: