apache-airflow-providers-snowflake: 5.3.1

[mtsadler-branch]

Apache Airflow Provider(s)

snowflake

Versions of Apache Airflow Providers

Unmentioned version lock: snowflake-connector-python<3.5.0

Upgrading to latest snowflake-connector-python creates issue with Timestamps rendering as unparsable epochs during pandas_sql.read_query()

Apache Airflow version

2.8.4

Operating System

N/A

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

    result = self.get_hook().get_pandas_df(query)
  File "/home/airflow/.local/lib/python3.9/site-packages/airflow/providers/common/sql/hooks/sql.py", line 227, in get_pandas_df
    return psql.read_sql(sql, con=conn, params=parameters, **kwargs)
  File "/home/airflow/.local/lib/python3.9/site-packages/pandas/io/sql.py", line 602, in read_sql
    return pandas_sql.read_query(
  File "/home/airflow/.local/lib/python3.9/site-packages/pandas/io/sql.py", line 2116, in read_query
    cursor = self.execute(*args)
  File "/home/airflow/.local/lib/python3.9/site-packages/pandas/io/sql.py", line 2068, in execute
    raise ex from exc
pandas.io.sql.DatabaseError: Execution failed on sql '
...
Timestamp '(seconds_since_epoch=1712074200000000000)' is not recognized; 73)

What you think should happen instead

No response

How to reproduce

install:

snowflake-connector-python==3.5.0
apache-airflow-providers-snowflake==5.3.1

run:

from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
query = '...'
SnowflakeHook.get_pandas_df(query)

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Would love someone takes a look and fixes those incompatibilities.

[mtsadler-branch]

Sorry, I mis-spoke when opening the issue.

It looks like apache-airflow-providers-snowflake==5.3.1 requires pyarrow==10.0.1

However, snowflake-connector-python==3.5.0 removes the pyarrow dependency.

When installing these 3 packages, pandas.Timestamp() are still parsed correctly when using SnowflakeHook:

snowflake-connector-python==3.5.0
pyarrow==10.0.1
apache-airflow-providers-snowflake==5.3.1

To rephrase, I believe apache-airflow-providers-snowflake==5.3.1 should have a dependency on pyarrow==10.0.1

[Taragolis]

What happen if you use latest version of package: https://pypi.org/project/snowflake-connector-python/#history ?

[potiuk]

Yes. I think I also would like to the bottom of this. I believe (after a quick look) snowlfake connector dropped pyarrow because it's well, optional, and it's very likely that what you are doing and the query you are running is triggering this error becasue of some implicit conversion of timestamp. I saw Snowflake suggesting FROM_TIMESTAMP and other ways of getting timestamp that does not require pyarrow to do some kind of conversion.

So I think 1st you should take a look at the queries you use and see if by modifying your queries you could get rid of the pyarrow requirement - following what snowflake did. I'd wait for your report on attempting to do so if latest version does not work for you already.

And even if there are cases where pyarrow might be useful, then at MOST this would call for an optional feature of the provider, not requirement. For example we could have apache-airflow-providers-snowflake[pyarrow] to install additionally pyarrow. This would be easy to do - we have mechanisms for that - and then you will have to install it this way in the future if you need pyarrow. But I think the questions above should be answered first.

[potiuk]

Maarked it as pending-response @mtsadler-branch

[mtsadler-branch]

To reproduce the error, I have the following PyPi packages installed:

apache-airflow==2.8.4
apache-airflow-providers-snowflake==5.3.1
pandas==1.3.5
pyarrow==10.0.1
snowflake-connector-python==3.5.0

And ran the following code:

from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
import pandas as pd
from snowflake.connector.pandas_tools import pd_writer


hook = SnowflakeHook(
    # Add SF connection details here...
)

# Create table
temp_sf = {
    "database": "test",
    "schema": "scratch",
    "table": "temp_table",
}
table_name = f"{temp_sf['database']}.{temp_sf['schema']}.{temp_sf['table']}"
create_table_query = f"""
    CREATE OR REPLACE TABLE {table_name}
    (COL1 INT, COL2 TIMESTAMP_NTZ(9)) AS
    SELECT 1, '2021-01-01T01:00:00.000000000'::timestamp_ntz
"""
results = hook.get_pandas_df(create_table_query)


# Append new data
new_data = pd.DataFrame({
    "COL1": [4, 5, 6],
    "COL2": [
        '2021-01-04T04:00:00.000000000',
        '2021-01-05T05:00:00.000000000',
        '2021-01-06T06:00:00.000000000',
    ],
})
engine = hook.get_sqlalchemy_engine()
with engine.connect() as conn:
    # Regardless of whether pyarrow is installed, this will append data to Snowflake table
    # However, if pyarrow isn't installed, then COL2 will have invalid timestamps
    new_data.to_sql(
        name=table_name,
        con=conn,
        if_exists="append",
        index=False,
        method=pd_writer,
    )
# If pyarrow is installed, this will return the correct data
# If pyarrow isn't installed, this will error: Timestamp '(seconds_since_epoch=1712074200000000000)' is not recognized
results = hook.get_pandas_df(f"SELECT * FROM {table_name}")
print(results)

Results:
You'll see DataFrame.to_sql(con=SnowflakeHook.get_sqlalchemy_engine().connect()) works as expected with pyarrow==10.0.1.

But when pyarrow isn't installed, the data written to COL2 isn't the correct format for Timestamp(), which causes an error when trying to query COL2.

Open Questions:
Is there a different pattern for inserting/appending data to an existing table, which doesn't require pyarrow?

[potiuk]

Open Questions:
Is there a different pattern for inserting/appending data to an existing table, which doesn't require pyarrow?

That's likely what you should check. following Snowflake documentation and experimenting with snowflake connector. I suggest to open isssue in the connector repository - there is likely a reason they dropped pyarrow, probably there is a way to do the same without Pyarrow.

[potiuk]

30 seconds googling from my side:

This is what is called under-the-hood:psql.read_sql(sql, con=conn, params=parameters, **kwargs) -> what I see is that Pandas string gets converted to Timestamp (and I guess pandas uses pyarrow for that). But pandas has native Timstamp object pd.Timestamp('20180310') so I guess you should utilize that.

You seem to be using an implicit conversion of Pandas that is happening in this case.

But those are all guesses - I have no experience with either of the technologies, just googled it. But for me it looks like it's not snowflake provider that depends on pyarrow but your queries implicitly do, so nothing to do in provider really - if you want to write queries and use implicit dependency on Pyarrow, then you need to declare it as dependency yourself.

[potiuk]

Converting to a discussion - that does not seem like provider's dependency missing, simply user implicitly depending on Pandas' optional feature of using Pyarrow for data conversion.

[mtsadler-branch]

For anyone stumbling across this thread, if you cast the datetime64[ns] to formatted strings before .to_sql(...) it seems to resolve the issue.

date_cols = df.select_dtypes(include=['datetime64[ns]']).columns
    for col in date_cols:
        df[col] = df[col].dt.strftime('%Y-%m-%dT%H:%M:%SZ')
engine = hook.get_sqlalchemy_engine()
with engine.connect() as conn:
    df.to_sql(
        name=table_name,
        con=conn,
        if_exists="append",
        index=False,
        method=pd_writer,
    )