Apparently the Jinja template does not work when using dynamic task mapping with SQLExecuteQueryOperator

[Talvarenga]

Apache Airflow version

2.6.2

What happened

We are trying to use dynamic task mapping with SQLExecuteQueryOperator on Trino. Our use case is to expand the sql parameter to the operator by calling some SQL files.

Without dynamic task mapping it works perfectly, but when used with the dynamic task mapping, it is unable to recognize the Path, and instead tries to execute the path as query.

I believe it has some relation with the template_searchpath parameter.

What you think should happen instead

It should have worked similar with or without dynamic task mapping.

How to reproduce

Deployed the following DAG in Airflow

from airflow.models import DAG
from datetime import datetime, timedelta
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

DEFAULT_ARGS = {
    'start_date': datetime(2023, 7, 16),
}

with DAG (dag_id= 'trino_dinamic_map',
    template_searchpath = '/opt/airflow',
	description = "Esta é um dag para o projeto exemplo",
    schedule = None,
	default_args = DEFAULT_ARGS,
    catchup=False,
) as dag:


    trino_call = SQLExecuteQueryOperator(
        task_id= 'trino_call',
        conn_id='con_id',
        sql = 'queries/insert_delta_dp_raw_table1.sql',
        handler=list
    )

    trino_insert = SQLExecuteQueryOperator.partial(
        task_id="trino_insert_table",
        conn_id='con_id',
        handler=list
    ).expand_kwargs([{'sql': 'queries/insert_delta_dp_raw_table1.sql'}, {'sql': 'queries/insert_delta_dp_raw_table2.sql'}, {'sql': 'queries/insert_delta_dp_raw_table3.sql'}])


trino_call >> trino_insert

In the sql file it can be any query, for the test I used a create table. Queries are located in /opt/airflow/queries

CREATE TABLE database_config.data_base_name.TABLE_NAME (
    "JOB_NAME" VARCHAR(60) NOT NULL,
    "JOB_ID" DECIMAL NOT NULL,
    "JOB_STATUS" VARCHAR(10),
    "JOB_STARTED_DATE" VARCHAR(10),
    "JOB_STARTED_TIME" VARCHAR(10),
    "JOB_FINISHED_DATE" VARCHAR(10),
    "JOB_FINISHED_TIME" VARCHSAR(10)
)

task_1 (without dynamic task mapping) completes successfully, while task_2(with dynamic task mapping) fails.

Looking at the error logs, there was a failure when executing the query, not recognizing the query content but the path.
Here is the traceback:

trino.exceptions.TrinoUserError: TrinoUserError(type=USER_ERROR, name=SYNTAX_ERROR, message="line 1:1: mismatched input 'queries'. Expecting: 'ALTER', 'ANALYZE', 'CALL', 'COMMENT', 'COMMIT', 'CREATE', 'DEALLOCATE', 'DELETE', 'DENY', 'DESC', 'DESCRIBE', 'DROP', 'EXECUTE', 'EXPLAIN', 'GRANT', 'INSERT', 'MERGE', 'PREPARE', 'REFRESH', 'RESET', 'REVOKE', 'ROLLBACK', 'SET', 'SHOW', 'START', 'TRUNCATE', 'UPDATE', 'USE', ", query_id=20230629_114146_04418_qbcnd)

Operating System

Red Hat Enterprise Linux 8.8

Versions of Apache Airflow Providers

No response

Deployment

Docker-Compose

Deployment details

No response

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]

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.

[raphaelauv]

same than #31801 (comment) ?

[Talvarenga]

I think in my case the problem is to identify "template_searchpath = '/opt/airflow'" while using the expand method, while in #31801 the problem is to convert the jinja2 template. I don't know if the cause of the problem is the same.

[hussein-awala]

I will take a look