Both Snowflake and Postgresql Don't Commit Database Changes.

[prime-minister-of-fun]

I'm trying to get a feel for using Airflow. Running version 2.7.1 in standalone, sqlite backend, and nothing fancy in the config file. Running on a laptop, with postgresql running on the same laptop. I checked the auth credentials. They work.

I have a very simple task, open a file with an UPDATE query in it, run the query. Task returns success, but, the column/row isn't updated. Same behavior for snowflake. I can even append a COMMIT as a second query and still the UPDATE shows as successful, yet the row is not changed. Is there something in setting up the connection (pg_for_me) I am missing?

Here's what's in the flat file:

UPDATE public.public_statistics set date_exported = CURRENT_TIMESTAMP where county_fips in (31109);

Here's the code for the Postgresql job.

import os
from airflow.models import Variable
from airflow.models.xcom import XCom
from airflow import AirflowException
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator

def update_pg_row():

 os.chdir('/home/mpapet/airflow/dags/datasets/pickup')
 process_file_list = list()

 all_files_list = os.listdir("./")
 #KISS any file here has not been imported to snowflake

 for i in all_files_list:
  if i.startswith('update_row_'):
    process_file_list.append(i)


 if len(process_file_list) == 0:
   print("no fips update files to process")
   return()

 print(process_file_list)
 print("length of file_list {}".format(len(process_file_list)))
 for f in process_file_list:
  file_stats = os.stat(f)
  print("processing {}".format(f))
  if file_stats.st_size > 10:
   fh = open(f, "r+")
   qry = fh.read()
   print("!!!!!!!!!!!! DEBUG !!!!!!!!!!!!!!!")
   print(qry)
   try:
    conn_op = SQLExecuteQueryOperator(
     conn_id='pg_for_me',
     task_id='update-pg-row-as-synced',
     autocommit=True,
     sql=f
     )
   except Exception as e:
    print('cannot get queryoperator')
    raise AirflowException(e)
  else:
   #should never be here.
   print("{} is too small??  {}".format(f,file_stats.st_size))

Here's the logging for the task:
*** Found local files: *** * /home/mpapet/airflow/logs/dag_id=row-by-row/run_id=manual__2023-09-19T06:05:22.830374+00:00/task_id=update-exported-row/attempt=1.log [2023-09-19, 06:05:31 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: row-by-row.update-exported-row manual__2023-09-19T06:05:22.830374+00:00 [queued]> [2023-09-19, 06:05:31 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: row-by-row.update-exported-row manual__2023-09-19T06:05:22.830374+00:00 [queued]> [2023-09-19, 06:05:31 UTC] {taskinstance.py:1359} INFO - Starting attempt 1 of 1 [2023-09-19, 06:05:31 UTC] {taskinstance.py:1380} INFO - Executing <Task(PythonOperator): update-exported-row> on 2023-09-19 06:05:22.830374+00:00 [2023-09-19, 06:05:31 UTC] {standard_task_runner.py:57} INFO - Started process 10407 to run task [2023-09-19, 06:05:31 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'row-by-row', 'update-exported-row', 'manual__2023-09-19T06:05:22.830374+00:00', '--job-id', '145', '--raw', '--subdir', 'DAGS_FOLDER/row-by-row0.py', '--cfg-path', '/tmp/tmpwujr2rdi'] [2023-09-19, 06:05:31 UTC] {standard_task_runner.py:85} INFO - Job 145: Subtask update-exported-row [2023-09-19, 06:05:31 UTC] {task_command.py:415} INFO - Running <TaskInstance: row-by-row.update-exported-row manual__2023-09-19T06:05:22.830374+00:00 [running]> on host monkey.papet.home [2023-09-19, 06:05:31 UTC] {taskinstance.py:1660} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='row-by-row' AIRFLOW_CTX_TASK_ID='update-exported-row' AIRFLOW_CTX_EXECUTION_DATE='2023-09-19T06:05:22.830374+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='manual__2023-09-19T06:05:22.830374+00:00' [2023-09-19, 06:05:31 UTC] {logging_mixin.py:151} INFO - ['update_row_2023-09-18-23:05:29.sql'] [2023-09-19, 06:05:31 UTC] {logging_mixin.py:151} INFO - length of file_list 1 [2023-09-19, 06:05:31 UTC] {logging_mixin.py:151} INFO - processing update_row_2023-09-18-23:05:29.sql [2023-09-19, 06:05:31 UTC] {logging_mixin.py:151} INFO - !!!!!!!!!!!! DEBUG !!!!!!!!!!!!!!! [2023-09-19, 06:05:31 UTC] {logging_mixin.py:151} INFO - UPDATE public.public_statistics set date_exported = CURRENT_TIMESTAMP where county_fips in (31109); COMMIT; [2023-09-19, 06:05:31 UTC] {python.py:194} INFO - Done. Returned value was: None [2023-09-19, 06:05:31 UTC] {taskinstance.py:1398} INFO - Marking task as SUCCESS. dag_id=row-by-row, task_id=update-exported-row, execution_date=20230919T060522, start_date=20230919T060531, end_date=20230919T060531 [2023-09-19, 06:05:31 UTC] {local_task_job_runner.py:228} INFO - Task exited with return code 0 [2023-09-19, 06:05:31 UTC] {taskinstance.py:2776} INFO - 1 downstream tasks scheduled from follow-on schedule check

[Taragolis]

Executing <Task(PythonOperator)

Just wondering are you trying to run Operator inside of PythonOperator/task flow ?

[Taragolis]

Yes. I take it that's wrong? That's an honest question.

When you create task (initialise Operator Object) then nothing actually happen except creation Python object.
But when you create task outside of DAG context then Airflow Worker (or Scheduler in case of LocalExecutor) nothing know about it because it not parsed during DAG Processing.

Actual task execution happen when worker run execute method of operator. I've predict next question "What happen if manually run execute method?" and the answer: It is not recommend, and it is not how Airflow actually work, because this methods only intend to use by Airflow Worker in additional quite a few preparation steps happen before worker execute this method and without this steps manually call execute method might have unpredictable behaviour: sometimes it seems to work, sometimes it failed immediately.

I don't see how I can accomplish something as simple as an ETL from one database to another without LOTS of tasks that need to pass input/output to the next task.

• XCom
• TaskFlow a bit simplify to pass input from one task to another. In addition Jarek @potiuk create good article on Medium: Generic Airflow Transfers made easy
• Jinja Templating a bit simplify dynamic staff, for example it could use for render SQL scripts, see also Templates reference
• If you you need to Low-level iteration, instead of Operators inside of operators, you could use Hooks and their methods, most of the Operators internally uses Hooks, so if you need operate over PostgreSQL you could use PostgresHook and their methods, for Snowflake you could use SnowflakeHook, and etc.

[prime-minister-of-fun]

Thank you so much for taking the time to write all that out. It will get me back on track.

[prime-minister-of-fun]

Hi, this still isn't working. I put everything for the PostgreSQL task inside the task, switched to the TaskFlow format.
The result is success, with the row in postgreSQL not updated.
I don't mean to be difficult, but, it's not obvious how this is wrong. I'll try the hook method and see if that changes anything.

[prime-minister-of-fun]

PostgresHook actually performs the update.

SQLExecuteQueryOperator does not notify you in any way that an update fails. Despite the existence of an autocommit option, it doesn't commit and returns success. I am very much out of my comfort zone to submit a bug report on this because I know so little. So, here it stays.

from  airflow.providers.postgres.hooks.postgres import PostgresHook
###LOTS MORE CODE###
try:
 conn_op = PostgresHook(
 postgres_conn_id='pg_for_me',
 schema='bls',
 autocommit=True
 )
except Exception as e:
 print('cannot get postgresHook')
 raise AirflowException(e)

try:
 conn = conn_op.get_conn()
 cursor = conn.cursor()
except Exception as e:
 print("Could not get cursor")
 raise AirflowException(e)```

[Taragolis]

If you want to open an issue it would be nice to have example how to reproduce this behaviour with SQLExecuteQueryOperator. Some simple DAG which could run everyone with minimal changes in it.