Issue 1:
Apache Airflow Provider(s)

common-sql

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==6.0.0
apache-airflow-providers-apache-hive==4.0.1
apache-airflow-providers-apache-livy==3.1.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.4.0
apache-airflow-providers-common-sql==1.3.0
apache-airflow-providers-databricks==3.3.0
apache-airflow-providers-dbt-cloud==2.2.0
apache-airflow-providers-elasticsearch==4.2.1
apache-airflow-providers-ftp==3.1.0
apache-airflow-providers-google==8.4.0
apache-airflow-providers-http==4.0.0
apache-airflow-providers-imap==3.0.0
apache-airflow-providers-microsoft-azure==4.3.0
apache-airflow-providers-postgres==5.2.2
apache-airflow-providers-redis==3.0.0
apache-airflow-providers-sftp==4.1.0
apache-airflow-providers-snowflake==3.3.0
apache-airflow-providers-sqlite==3.2.1
apache-airflow-providers-ssh==3.2.0

Apache Airflow version

2.4.3

Operating System

Debian Bullseye

Deployment

Astronomer

Deployment details

No response

What happened

With the latest version of common-sql provider, the get_records from hook is now a ordinary dictionary, causing this KeyError with SqlSensor:

[2022-11-29, 00:39:18 UTC] {taskinstance.py:1851} ERROR - Task failed with exception
Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/airflow/sensors/base.py", line 189, in execute
poke_return = self.poke(context)
File "/usr/local/lib/python3.9/site-packages/airflow/providers/common/sql/sensors/sql.py", line 98, in poke
first_cell = records[0][0]
KeyError: 0

I have only tested with Snowflake, I haven't tested it with other databases. Reverting back to 1.2.0 solves the issue.

What you think should happen instead

It should return an iterable list as usual with the query.

How to reproduce

from datetime import datetime

from airflow import DAG
from airflow.providers.common.sql.sensors.sql import SqlSensor

with DAG(
dag_id="sql_provider_snowflake_test",
schedule=None,
start_date=datetime(2022, 1, 1),
catchup=False,
):
t1 = SqlSensor(
task_id="snowflake_test",
conn_id="snowflake",
sql="select 0",
fail_on_empty=False,
poke_interval=20,
mode="poke",
timeout=60 * 5,
)

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow Provider(s)

common-sql

Versions of Apache Airflow Providers

apache-airflow-providers-google==8.2.0

apache-airflow-providers-http==4.0.0

apache-airflow-providers-salesforce==5.0.0

apache-airflow-providers-slack==5.1.0

apache-airflow-providers-snowflake==3.2.0

Issue:

apache-airflow-providers-common-sql==1.3.0

Apache Airflow version

2.4.3

Operating System

Debian GNU/Linux 11 (bullseye)

Deployment

Astronomer

Deployment details

No response

What happened

Problem occurred when upgrading from common-sql=1.2.0 to common-sql=1.3.0

Getting a KEY_ERROR when running a unique_check and null_check on a column.

1.3.0 log:

1.2.0 log:

What you think should happen instead

Potential causes:

seems to be indexing based on the test query column COL_NAME instead of the table column STRIPE_ID

the record from the test changed types went from a tuple to a list of dictionaries.

no tolerance is specified for these tests, so .get('tolerance') looks like it will cause an error without a default specified like .get('tolerance', None)

Expected behavior:

these tests continue to pass with the upgrade

tolerance is not a required key.

How to reproduce

from datetime import datetime
from airflow import DAG

from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.providers.common.sql.operators.sql import SQLColumnCheckOperator

my_conn_id = "snowflake_default"

default_args={"conn_id": my_conn_id}

with DAG(
dag_id="airflow_providers_example",
schedule=None,
start_date=datetime(2022, 11, 27),
default_args=default_args,
) as dag:

create_table = SnowflakeOperator(
task_id="create_table",
sql=""" CREATE OR REPLACE TABLE testing AS (
SELECT
1 AS row_num,
'not null' AS field

UNION ALL

SELECT
2 AS row_num,
'test' AS field

UNION ALL

SELECT
3 AS row_num,
'test 2' AS field
)""",
)

column_checks = SQLColumnCheckOperator(
task_id="column_checks",
table="testing",
column_mapping={
"field": {"unique_check": {"equal_to": 0}, "null_check": {"equal_to": 0}}
},
)

create_table >> column_checks

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct