Add accept_none to SQLTableCheckOperator (#63210)

* issue-30082: Adding accept_none to SQLTableCheckOperator

* Update providers/common/sql/src/airflow/providers/common/sql/operators/sql.py

Co-authored-by: Elad Kalif <45845474+eladkal@users.noreply.github.com>

* fix static checks

---------

Co-authored-by: Elad Kalif <45845474+eladkal@users.noreply.github.com>

Author: jroachgolf84

providers/common/sql/src/airflow/providers/common/sql/operators/sql.py

@@ -714,6 +714,7 @@ class SQLTableCheckOperator(BaseSQLOperator):
 
     :param conn_id: the connection ID used to connect to the database
     :param database: name of database which overwrite the defined one in connection
+    :param accept_none: If True, an empty table (row count=0) will not trigger a failure.
 
     .. seealso::
         For more information on how to use this operator, take a look at the guide:
@@ -738,20 +739,32 @@ def __init__(
         partition_clause: str | None = None,
         conn_id: str | None = None,
         database: str | None = None,
+        accept_none: bool = False,
         **kwargs,
     ):
         super().__init__(conn_id=conn_id, database=database, **kwargs)
 
         self.table = table
         self.checks = checks
         self.partition_clause = _initialize_partition_clause(partition_clause)
+        self.accept_none = accept_none
         self.sql = f"SELECT check_name, check_result FROM ({self._generate_sql_query()}) AS check_table"
 
     def execute(self, context: Context):
         hook = self.get_db_hook()
         records = hook.get_records(self.sql)
 
         if not records:
+            # accept_none prevents an error from being thrown if there are no records in the table
+            if self.accept_none:
+                self.log.warning(
+                    "No records found for table: %s, but accept_none=True, "
+                    "therefore, no tests are being marked as failed.",
+                    self.table,
+                )
+                return
+
+            # Otherwise, we'll raise an exception
             self._raise_exception(f"The following query returned zero rows: {self.sql}")
 
         self.log.info("Record:\n%s", records)

providers/common/sql/tests/unit/common/sql/operators/test_sql.py

@@ -605,6 +605,21 @@ def test_sql_check_partition_clause_templating(self, conn_id):
         finally:
             hook.run(["DROP TABLE employees"])
 
+    def test_sql_check_accept_none_true(self, monkeypatch):
+        """Validate that empty table does not fail when accept_none=True."""
+        records = []
+        operator = self._construct_operator(monkeypatch, self.checks, records)
+        operator.accept_none = True
+        operator.execute(context=MagicMock())
+
+    def test_sql_check_accept_none_false(self, monkeypatch):
+        """Validate that empty table throws an exception when accept_none=False."""
+        records = []
+        operator = self._construct_operator(monkeypatch, self.checks, records)
+        operator.accept_none = False  # This is default, technically
+        with pytest.raises(AirflowException):
+            operator.execute(context=MagicMock())
+
     def test_pass_all_checks_check(self, monkeypatch):
         records = [("row_count_check", 1), ("column_sum_check", "y")]
         operator = self._construct_operator(monkeypatch, self.checks, records)