当我的sql抛出空值时，如何在SqlSensor(Airflow)中获得成功？

from airflow import DAG
from airflow.operators.sensors import SqlSensor
from airflow.hooks.mysql_hook import MySqlHook

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2022, 1, 1),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'sql_sensor_example',
    default_args=default_args,
    schedule_interval='@daily',
)

sql_sensor_task = SqlSensor(
    task_id='sql_sensor_task',
    conn_id='mysql_conn',
    sql='SELECT COUNT(*) FROM my_table',
    timeout=60,
    mode='poke',
    dag=dag,
)

def handle_empty_result(context):
    result = context['task_instance'].xcom_pull(task_ids='sql_sensor_task')
    if result is None:
        raise ValueError('SQL query returned empty result')

sql_sensor_task.set_upstream(handle_empty_result)