Unable to execute a file, by placing the absolute path of it, ending with .sql extension to the AWS SqlToS3Operator

[Altonymous]

Apache Airflow version

2.8.1

If "Other Airflow 2 version" selected, which one?

No response

What happened?

Unable to pass in an absolute path to a sql file/template and have it load appropriately in the amazon sql to s3 transfer class:
https://github.com/apache/airflow/blob/2.8.1/airflow/providers/amazon/aws/transfers/sql_to_s3.py#L64

Traceback (most recent call last):
  File "/opt/venv/lib/python3.11/site-packages/pandas/io/sql.py", line 2675, in execute
    cur.execute(sql, *args)
  File "/opt/venv/lib/python3.11/site-packages/MySQLdb/cursors.py", line 179, in execute
    res = self._query(mogrified_query)
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/MySQLdb/cursors.py", line 330, in _query
    db.query(q)
  File "/opt/venv/lib/python3.11/site-packages/MySQLdb/connections.py", line 261, in query
    _mysql.connection.query(self, query)
MySQLdb.ProgrammingError: (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'extract/last_event.sql' at line 1")
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
  File "/opt/venv/lib/python3.11/site-packages/airflow/models/taskinstance.py", line 433, in _execute_task
    result = execute_callable(context=context, **execute_callable_kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/airflow/decorators/base.py", line 241, in execute
    return_value = super().execute(context)
                   ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/airflow/operators/python.py", line 199, in execute
    return_value = self.execute_callable()
                   ^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/airflow/operators/python.py", line 216, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/dags/aggregates/users_last_event.py", line 48, in extract
    ).execute(context=None)
      ^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/airflow/providers/amazon/aws/transfers/sql_to_s3.py", line 178, in execute
    data_df = sql_hook.get_pandas_df(sql=self.query, parameters=self.parameters)
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/airflow/providers/common/sql/hooks/sql.py", line 221, in get_pandas_df
    return psql.read_sql(sql, con=conn, params=parameters, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/pandas/io/sql.py", line 706, in read_sql
    return pandas_sql.read_query(
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/pandas/io/sql.py", line 2739, in read_query
    cursor = self.execute(sql, params)
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/venv/lib/python3.11/site-packages/pandas/io/sql.py", line 2687, in execute
    raise ex from exc
pandas.errors.DatabaseError: Execution failed on sql 'extract/last_event.sql': (1064, "You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'extract/last_event.sql' at line 1")

What you think should happen instead?

I would expect it to use the sql/template in the file.

How to reproduce

import pendulum

from airflow.decorators import dag, task
from airflow.providers.amazon.aws.transfers.sql_to_s3 import SqlToS3Operator

dag(
    schedule_interval=None,
    start_date=pendulum.datetime(2024, 1, 1, tz="UTC"),
    catchup=False,
    tags=["prototype"],
    template_searchpath=["/app/dags/templates/users"],
)
def users_last_event_taskflow():
    @task()
    def extract(s3_bucket, s3_key, sql_conn_id, aws_conn_id):
        SqlToS3Operator(
            query="/app/path/to/sql/template.sql",
            s3_bucket=s3_bucket,
            s3_key=s3_key,
            sql_conn_id=sql_conn_id,
            aws_conn_id=aws_conn_id,
            task_id="sql_to_s3_task",
            replace=True,
        ).execute(context=None)

    extract("s3_bucket", "s3_key", "mysql_default", "aws_default")

users_last_event_taskflow_etl = users_last_event_taskflow()

Operating System

debian

Versions of Apache Airflow Providers

apache-airflow==2.8.1
apache-airflow-providers-amazon==8.17.0
apache-airflow-providers-mysql==5.1.0

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

Every time.

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

[potiuk]

You should put just filename not "folder/filename" in your sql. the search path is the folder where the file is looked for (but it does not search subfolcers).

[Altonymous]

@potiuk,

I also missed a provider in my report. Didn't realize my copy/pasta mistake, until now. Not sure if it's relevant.

apache-airflow-providers-common-sql[pandas]==1.10.1

Tried using just the file name and updating the template_path to be the exact directory the file lives in with the same result...

template_searchpath=["/app/dags/templates/users/"]

query="last_event.sql",

last_event.sql contents:
select 1=1 as last_event;

"You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'last_event.sql' at line 1")

[potiuk]

Then no idea, It should work, maybe someone will be able to investigate it.

[potiuk]

Ah no . Of course your DAG is completely wrong. Why are you instantiating Operator inside a task-flow decorated task? It makes no sense whatsoever. You either use the operator (classic) or decaorate your task. with @task decorator. Putting operator INISDE @task decorated method makes completely no sense (and that's the reason why it does not work because the operator is not processed by execution engine to process templates if you just instalatiate it and use it.

[potiuk]

You can see how you can combine task-flow tasks and classic operator tasks in single DAG here https://airflow.apache.org/docs/apache-airflow/stable/tutorial/taskflow.html#adding-dependencies-between-decorated-and-traditional-tasks - but certainly you do not do it by instantiating classic operator in task-flow decorated task