def METHOD_NAME(name: str, sql_only: bool = False) -> t.Type[BaseOperator]:
    name = name.lower()
    try:
        if name == "spark":
            from sqlmesh.schedulers.airflow.operators.spark_submit import (
                SQLMeshSparkSubmitOperator,
            )
            return SQLMeshSparkSubmitOperator
        if name in ("databricks", "databricks-submit", "databricks-sql"):
            if name == "databricks-submit" or (name == "databricks" and not sql_only):
                from sqlmesh.schedulers.airflow.operators.databricks import (
                    SQLMeshDatabricksSubmitOperator,
                )
                return SQLMeshDatabricksSubmitOperator
            if name == "databricks-sql" or (name == "databricks" and sql_only):
                from sqlmesh.schedulers.airflow.operators.databricks import (
                    SQLMeshDatabricksSQLOperator,
                )
                return SQLMeshDatabricksSQLOperator
        if name == "snowflake":
            from sqlmesh.schedulers.airflow.operators.snowflake import (
                SQLMeshSnowflakeOperator,
            )
            return SQLMeshSnowflakeOperator
        if name == "bigquery":
            from sqlmesh.schedulers.airflow.operators.bigquery import (
                SQLMeshBigQueryOperator,
            )
            return SQLMeshBigQueryOperator
        if name == "redshift":
            from sqlmesh.schedulers.airflow.operators.redshift import (
                SQLMeshRedshiftOperator,
            )
            return SQLMeshRedshiftOperator
        if name in ("postgres", "postgresql"):
            from sqlmesh.schedulers.airflow.operators.postgres import (
                SQLMeshPostgresOperator,
            )
            return SQLMeshPostgresOperator
    except ImportError:
        raise SQLMeshError(f"Failed to automatically discover an operator for '{name}'.'")
    raise ValueError(f"Unsupported engine name '{name}'.")