Why does _generate_insert_sql use %s as a placeholder instead of ?, or how to use GenericTransfer?

[jplauri]

I'm trying to use GenericTransfer to insert rows from a table to another using MS SQL.

Suppose there are two toy tables:

CREATE TABLE SOURCE_TEST (
    id INT IDENTITY PRIMARY KEY,
    payload INTEGER NOT NULL
);

CREATE TABLE DESTINATION_TEST (
    id INT IDENTITY PRIMARY KEY,
    payload INTEGER NOT NULL
);

INSERT INTO SOURCE_TEST(payload) VALUES (1);
INSERT INTO SOURCE_TEST(payload) VALUES (2);
INSERT INTO SOURCE_TEST(payload) VALUES (3);

Specifically, our goal is to copy all the rows from SOURCE_TEST to DESTINATION_TEST. Let's write a simple DAG for this:

from os import environ
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.generic_transfer import GenericTransfer

default_args = {
    'owner': 'airflow',
    'email': ['x@y.com'],
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=1),
    # 'queue': 'bash_queue',
    # 'pool': 'backfill',
    # 'priority_weight': 10,
    # 'end_date': datetime(2016, 1, 1),
    # 'wait_for_downstream': False,
    # 'dag': dag,
    # 'sla': timedelta(minutes=5),
    'execution_timeout': timedelta(minutes=30)
}


with DAG(
    dag_id='test_generic_transfer',
    default_args=default_args,
    description='Try out GenericTransfer',
    schedule_interval=None,
    start_date=datetime(2021, 1, 1),
    catchup=False,
    template_searchpath=environ["AIRFLOW_HOME"] + "/dags/sandbox/sql/", # Where the .sql file should be found
    tags=['sandbox']
) as dag:
    transfer = GenericTransfer(
        task_id="transfer",
        sql='read_from_source.sql', # Simple SELECT *, see below
        source_conn_id="my_db",
        destination_conn_id="my_db",
        destination_table="DESTINATION_TEST",
        preoperator="TRUNCATE TABLE DESTINATION_TEST",
        insert_args={
            "target_fields": ["payload"]
        }
    )

Here, we've also written read_from_source.sql which has the contents SELECT payload FROM SOURCE_TEST. Also, suppose my_db is a predefined ODBC connection for which "test connection" works. Now, when executing the DAG, the task fails with:
pyodbc.ProgrammingError: ('The SQL contains 0 parameter markers, but 1 parameters were supplied', 'HY000')
Let's look into what's roughly going on: the last line of execute calls insert_rows of the DbApiHook base, which in turn calls the static helper _generate_insert_sql. This method seems to use "%s" as the placeholder value, but I'm not sure why. Should the placeholder value not be ? instead?

I can't imagine the placeholder value simply being a bug, so what am I doing wrong i.e., how to use GenericTransfer?

[potiuk]

I think MsSQLHook lacks the proper implementaiton of the _generate_insert_sql . Every database has it's own form of prepared statements - Postgres/Mysql use '%s' (and this is the default implementation) , mssql uses ? - and that's why _generate_insert_sql needs to be provided by Hooks that do not support '%s' syntax. I think the best way woudl be to add a PR fixing it.

[jplauri]

Unless I'm missing something, it does indeed seem that simple. One thing to note is that MS SQL also doesn't have "REPLACE INTO" (in case replace=True). However, even though I'm no SQL expert either, I've successfully used this pattern for UPSERT before: https://sqlperformance.com/2020/09/locking/upsert-anti-pattern

But handling the UPSERT nicely seems like the most difficult thing to do here. One lazy(?) alternative is to raise a NotImplementedError in case the replace flag is set.

[jplauri]

Sorry for the spam, but I see that the PostgresHook is a little specialized in this as well. That is, it uses replace_index which is exactly something we should have for the MsSqlHook too if wanting to replace. I think I can draw inspiration from this hook.

[potiuk]

Yep.

[arunaj]

Hi @jplauri, did you end up implementing your own _generate_insert_sql for MsSQLHook? So far I have only found this related issue and this implementation that uses sqlalchemy with INSERT instead of MERGE. I wanted to make sure it doesn't already exist somewhere before attempting my own, since I am new to Airflow/Hooks.

[jplauri]

Yes, I did. See here: #19596

There's maybe a better solution available now (see one of the latest comments). This functionality would still be useful, I think.