<p dir="auto"><strong>Apache Airflow version</strong>: 1.10.15, but happens in all of them</p>
<p dir="auto"><strong>Environment</strong>: GCP Composer -&gt; Kubernetes + Celery</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>: Google</li>
</ul>
<p dir="auto"><strong>What happened</strong>:<br>
Errors appear while trying to use SqlSensor against a BigQuery View.<br>
For context I'll explain a little bit: BigQuery seemingly executes only SQL, but in reality there are two versions of the language. For backwards compatibility the so called <code class="notranslate">legacySQL</code> is set by default.</p>
<p dir="auto">I'm hot-fixing it like so:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="class BigQuerySqlSensor(SqlSensor):
    &quot;&quot;&quot;
    Overwrites hook config when using SqlSensor with a BigQuery
    connection
    &quot;&quot;&quot;
    def _get_hook(self):
        hook = super()._get_hook()
        hook.use_legacy_sql = False
        hook.location = 'europe-west1'
        return hook"><pre class="notranslate"><span class="pl-k">class</span> <span class="pl-v">BigQuerySqlSensor</span>(<span class="pl-v">SqlSensor</span>):
    <span class="pl-s">"""</span>
<span class="pl-s">    Overwrites hook config when using SqlSensor with a BigQuery</span>
<span class="pl-s">    connection</span>
<span class="pl-s">    """</span>
    <span class="pl-k">def</span> <span class="pl-en">_get_hook</span>(<span class="pl-s1">self</span>):
        <span class="pl-s1">hook</span> <span class="pl-c1">=</span> <span class="pl-en">super</span>().<span class="pl-en">_get_hook</span>()
        <span class="pl-s1">hook</span>.<span class="pl-s1">use_legacy_sql</span> <span class="pl-c1">=</span> <span class="pl-c1">False</span>
        <span class="pl-s1">hook</span>.<span class="pl-s1">location</span> <span class="pl-c1">=</span> <span class="pl-s">'europe-west1'</span>
        <span class="pl-k">return</span> <span class="pl-s1">hook</span></pre></div>
<p dir="auto"><strong>How to reproduce it</strong>:<br>
The issue I'm experiencing is trying to run a simple query agains a view</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="check_resource_has_data = SqlSensor(
    task_id=&quot;check_resource_has_data&quot;,
    conn_id='google_cloud_default',
    sql=f'SELECT COUNT(*) &gt; 0 FROM `{dataset_id}.{resource_id}`',
)"><pre class="notranslate"><span class="pl-s1">check_resource_has_data</span> <span class="pl-c1">=</span> <span class="pl-v">SqlSensor</span>(
    <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"check_resource_has_data"</span>,
    <span class="pl-s1">conn_id</span><span class="pl-c1">=</span><span class="pl-s">'google_cloud_default'</span>,
    <span class="pl-s1">sql</span><span class="pl-c1">=</span><span class="pl-s">f'SELECT COUNT(*) &gt; 0 FROM `<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">dataset_id</span><span class="pl-kos">}</span></span>.<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">resource_id</span><span class="pl-kos">}</span></span>`'</span>,
)</pre></div>
<p dir="auto"><em>First Error</em>: As I am using a standard language but the hook automatically sets this to legacy BigQuery will try to run this as Legacy, fail to parse and return an erro</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2021-07-28 16:03:05,499] {taskinstance.py:1152} ERROR - BigQuery job failed. Final error was: {'reason': 'invalid', 'location': '`my-project.test.cashflow_abacus_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}. The job was: {'kind': 'bigquery#job', 'etag': 'random_etag==', 'id': 'my-project:US.job_some_randome_id_123412', 'selfLink': 'https://bigquery.googleapis.com/bigquery/v2/projects/my-project/jobs/job_some_randome_id_123412?location=US', 'user_email': 'composer-data-orchestration@my-project.iam.gserviceaccount.com', 'configuration': {'query': {'query': '
                    SELECT COUNT(*) &gt; 0
                    FROM `my-project.test.my_view`
                ', 'priority': 'INTERACTIVE', 'useLegacySql': True}, 
                'jobType': 'QUERY'}, 'jobReference': {'projectId': 
                'my-project', 'jobId': 'job_some_randome_id_123412', 'location': 'SO'}, 'statistics': {'creationTime': '1627488185084', 'startTime': '1627488185109', 'endTime': '1627488185109'}, 'status': {'errorResult': {'reason': 'invalid', 'location': '`my-project.test.my_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}, 'errors': [{'reason': 'invalid', 'location': '`my-project.test.my_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}], 'state': 'DONE'}}"><pre class="notranslate"><code class="notranslate">[2021-07-28 16:03:05,499] {taskinstance.py:1152} ERROR - BigQuery job failed. Final error was: {'reason': 'invalid', 'location': '`my-project.test.cashflow_abacus_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}. The job was: {'kind': 'bigquery#job', 'etag': 'random_etag==', 'id': 'my-project:US.job_some_randome_id_123412', 'selfLink': 'https://bigquery.googleapis.com/bigquery/v2/projects/my-project/jobs/job_some_randome_id_123412?location=US', 'user_email': 'composer-data-orchestration@my-project.iam.gserviceaccount.com', 'configuration': {'query': {'query': '
                    SELECT COUNT(*) &gt; 0
                    FROM `my-project.test.my_view`
                ', 'priority': 'INTERACTIVE', 'useLegacySql': True}, 
                'jobType': 'QUERY'}, 'jobReference': {'projectId': 
                'my-project', 'jobId': 'job_some_randome_id_123412', 'location': 'SO'}, 'statistics': {'creationTime': '1627488185084', 'startTime': '1627488185109', 'endTime': '1627488185109'}, 'status': {'errorResult': {'reason': 'invalid', 'location': '`my-project.test.my_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}, 'errors': [{'reason': 'invalid', 'location': '`my-project.test.my_view`', 'message': 'Invalid table name: `my-project.test.my_view`
[Try using standard SQL (https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql)].'}], 'state': 'DONE'}}
</code></pre></div>
<p dir="auto"><em>Second Error</em>: In order to switch the "dialect" you have several options. One of them, the closest to the code is prepending the query with a comment line like so <code class="notranslate"># standardSQL</code> (<a href="https://cloud.google.com/bigquery/docs/reference/standard-sql/enabling-standard-sql#sql-prefix" rel="nofollow">docs</a>). But</p>
<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">There's no way to pass information to the hock that the SqlSensor is using underneath. It would be interesting to add a <code class="notranslate">hook_params</code> to the SqlSensor to be able to config the underlaying hooks:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="# airflow/sensors/sql.py
class SqlSensor(BaseSensorOperator):
    def __init__(
        self, *, conn_id, sql, hook_config: Dict, parameters=None, success=None, failure=None, fail_on_empty=False, **kwargs
    ):
        self.conn_id = conn_id
        # init all the params...
        self.hook_config = hook_config or {}
        super().__init__(**kwargs)

    def _get_hook(self):
        conn = BaseHook.get_connection(self.conn_id)
        # ...
        return conn.get_hook(**self.hook_config)"><pre class="notranslate"><span class="pl-c"># airflow/sensors/sql.py</span>
<span class="pl-k">class</span> <span class="pl-v">SqlSensor</span>(<span class="pl-v">BaseSensorOperator</span>):
    <span class="pl-k">def</span> <span class="pl-en">__init__</span>(
        <span class="pl-s1">self</span>, <span class="pl-c1">*</span>, <span class="pl-s1">conn_id</span>, <span class="pl-s1">sql</span>, <span class="pl-s1">hook_config</span>: <span class="pl-v">Dict</span>, <span class="pl-s1">parameters</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">success</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">failure</span><span class="pl-c1">=</span><span class="pl-c1">None</span>, <span class="pl-s1">fail_on_empty</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>
    ):
        <span class="pl-s1">self</span>.<span class="pl-s1">conn_id</span> <span class="pl-c1">=</span> <span class="pl-s1">conn_id</span>
        <span class="pl-c"># init all the params...</span>
        <span class="pl-s1">self</span>.<span class="pl-s1">hook_config</span> <span class="pl-c1">=</span> <span class="pl-s1">hook_config</span> <span class="pl-c1">or</span> {}
        <span class="pl-en">super</span>().<span class="pl-en">__init__</span>(<span class="pl-c1">**</span><span class="pl-s1">kwargs</span>)

    <span class="pl-k">def</span> <span class="pl-en">_get_hook</span>(<span class="pl-s1">self</span>):
        <span class="pl-s1">conn</span> <span class="pl-c1">=</span> <span class="pl-v">BaseHook</span>.<span class="pl-en">get_connection</span>(<span class="pl-s1">self</span>.<span class="pl-s1">conn_id</span>)
        <span class="pl-c"># ...</span>
        <span class="pl-k">return</span> <span class="pl-s1">conn</span>.<span class="pl-en">get_hook</span>(<span class="pl-c1">**</span><span class="pl-s1">self</span>.<span class="pl-s1">hook_config</span>)</pre></div>
<p dir="auto">And in the connection</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="# airflow/models/connection.py
class Connection(Base, LoggingMixin):
    # ...
    def get_hook(self, **kwargs):
        &quot;&quot;&quot;Return hook based on conn_type.&quot;&quot;&quot;
        # locate hook class ...
        return hook_class(**{conn_id_param: self.conn_id}, **kwargs)"><pre class="notranslate"><span class="pl-c"># airflow/models/connection.py</span>
<span class="pl-k">class</span> <span class="pl-v">Connection</span>(<span class="pl-v">Base</span>, <span class="pl-v">LoggingMixin</span>):
    <span class="pl-c"># ...</span>
    <span class="pl-k">def</span> <span class="pl-en">get_hook</span>(<span class="pl-s1">self</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
        <span class="pl-s">"""Return hook based on conn_type."""</span>
        <span class="pl-c"># locate hook class ...</span>
        <span class="pl-k">return</span> <span class="pl-en">hook_class</span>(<span class="pl-c1">**</span>{<span class="pl-s1">conn_id_param</span>: <span class="pl-s1">self</span>.<span class="pl-s1">conn_id</span>}, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>)</pre></div>
<p dir="auto"><strong>Description</strong></p>
<p dir="auto">A sql sensor which uses Standard SQL due to default one uses legacy sql</p>
<p dir="auto"><strong>Use case / motivation</strong></p>
<p dir="auto">Currently (correct me if I am wrong!), the sql sensor only supports legacy sql. If I want to poke a BQ table, I do not think I can do that using standard sql right now.</p>
<p dir="auto"><strong>Are you willing to submit a PR?</strong></p>
<p dir="auto">If community approves of this idea, sure!</p>