You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dheerajturaga (via GitHub)" <gi...@apache.org> on 2025/06/14 03:13:55 UTC
[PR] Extend command column in the edge_job table to accomodate more chars [airflow]
dheerajturaga opened a new pull request, #51716:
URL: https://github.com/apache/airflow/pull/51716
The `command` column in the `edge_job` table is hitting the character limit of `1000` during retires of decent sized BashOperator commands. This is causing the EdgeExecutor to crash and exit leaving the scheduler in a bad state. This PR bumps up the char limit to `2048`
Here's the snippet of the error from the scheduler logs
```
[SQL: INSERT INTO edge_job (dag_id, task_id, run_id, map_index, try_number, state, queue, concurrency_slots, command, queued_dttm, edge_worker, last_update) VALUES (%(dag_id)s, %(task_id)s, %(run_id)s, %(map_index)s, %(try_number)s, %(state)s, %(queue)s, %(concurrency_slots)s, %(command)s, %(queued_dttm)s, %(edge_worker)s, %(last_update)s)]
(Background on this error at: https://sqlalche.me/e/14/9h9h)
[2025-06-14T02:26:26.528+0000] {edge_executor.py:343} INFO - Shutting down EdgeExecutor
[2025-06-14T02:26:26.528+0000] {scheduler_job_runner.py:1031} INFO - Exited execute loop
Traceback (most recent call last):
File "/home/airflow/.local/lib/python3.12/site-packages/sqlalchemy/engine/base.py", line 1890, in _execute_context
self.dialect.do_executemany(
File "/home/airflow/.local/lib/python3.12/site-packages/sqlalchemy/dialects/postgresql/psycopg2.py", line 982, in do_executemany
context._psycopg2_fetched_rows = xtras.execute_values(
^^^^^^^^^^^^^^^^^^^^^
File "/home/airflow/.local/lib/python3.12/site-packages/psycopg2/extras.py", line 1299, in execute_values
cur.execute(b''.join(parts))
psycopg2.errors.StringDataRightTruncation: value too long for type character varying(1000)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #51716:
URL: https://github.com/apache/airflow/pull/51716#issuecomment-2972818828
> Looks good! Just will make a short test and CI needs to be green, then LGTM!
Great!, I ran the following in `airflow db shell` when launching breeze to verify `breeze start-airflow --db-reset -e --executor EdgeExecutor`
```
SELECT column_name, data_type, character_maximum_length FROM information_schema.columns WHERE table_name = 'edge_job' AND column_name = 'command' AND table_schema = 'public';
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #51716:
URL: https://github.com/apache/airflow/pull/51716#issuecomment-2972817929
Thanks @amoghrajesh for the quick sanity check!
Thanks @jscheffl for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #51716:
URL: https://github.com/apache/airflow/pull/51716#issuecomment-2972829347
Manual tests with Airflow 3 and Airflow 2.11 went fine!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #51716:
URL: https://github.com/apache/airflow/pull/51716#discussion_r2146875112
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -79,6 +79,13 @@ def _check_db_schema(self, engine: Engine) -> None:
if edge_job_columns and "concurrency_slots" not in edge_job_columns:
EdgeJobModel.metadata.drop_all(engine, tables=[EdgeJobModel.__table__])
+ # Increase command column size in edge_job table to support larger commands
+ if edge_job_columns and "command" in edge_job_columns:
Review Comment:
If think the criteria is not correct. The "command" column is always existing. Also we should ALTER the table only if needed. How about this? So meaning you check the length and only alter if the length is different?
```
with contextlib.suppress(NoSuchTableError):
edge_job_schema = inspector.get_columns("edge_job")
edge_job_columns = [column["name"] for column in edge_job_schema]
for column in edge_job_schema:
if column["name"] == "command":
edge_job_command_len = column["type"].length
# version 0.6.0rc1 added new column concurrency_slots
if edge_job_columns and "concurrency_slots" not in edge_job_columns:
EdgeJobModel.metadata.drop_all(engine, tables=[EdgeJobModel.__table__])
# after version 1.1.0 the command column was changed to VARCHAR(2048)
elif edge_job_command_len and edge_job_command_len != 2048:
with Session(engine) as session:
query = "ALTER TABLE edge_job ALTER COLUMN command TYPE VARCHAR(2048);"
session.execute(text(query))
session.commit()
edge_worker_columns = None
with contextlib.suppress(NoSuchTableError):
edge_worker_columns = [column["name"] for column in inspector.get_columns("edge_worker")]
```
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -79,6 +79,13 @@ def _check_db_schema(self, engine: Engine) -> None:
if edge_job_columns and "concurrency_slots" not in edge_job_columns:
EdgeJobModel.metadata.drop_all(engine, tables=[EdgeJobModel.__table__])
+ # Increase command column size in edge_job table to support larger commands
+ if edge_job_columns and "command" in edge_job_columns:
+ with Session(engine) as session:
+ query = "ALTER TABLE edge_job ALTER COLUMN command TYPE VARCHAR(2048);"
Review Comment:
providers/edge3/src/airflow/providers/edge3/models/edge_job.py:52 must be adjusted accordingly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #51716:
URL: https://github.com/apache/airflow/pull/51716#discussion_r2146980317
##########
providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py:
##########
@@ -73,12 +73,23 @@ def _check_db_schema(self, engine: Engine) -> None:
inspector = inspect(engine)
edge_job_columns = None
with contextlib.suppress(NoSuchTableError):
- edge_job_columns = [column["name"] for column in inspector.get_columns("edge_job")]
+ edge_job_schema = inspector.get_columns("edge_job")
+ edge_job_columns = [column["name"] for column in edge_job_schema]
+ for column in edge_job_schema:
+ if column["name"] == "command":
+ edge_job_command_len = column["type"].length
# version 0.6.0rc1 added new column concurrency_slots
if edge_job_columns and "concurrency_slots" not in edge_job_columns:
EdgeJobModel.metadata.drop_all(engine, tables=[EdgeJobModel.__table__])
+ # version 1.1.0 the command column was changed to VARCHAR(2048)
+ if edge_job_command_len and edge_job_command_len != 2048:
Review Comment:
Just a small nit now... I don't know if this will happen in real life but if the table is dropped inthe lines above the migration will fail here. So to be safe we need `elif`:
```suggestion
elif edge_job_command_len and edge_job_command_len != 2048:
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on PR #51716:
URL: https://github.com/apache/airflow/pull/51716#issuecomment-2972751958
@jscheffl @dheerajturaga from what i see in the edge executor interface, the edge_command is nothing but a model dump of `workload`: https://github.com/apache/airflow/blob/main/providers/edge3/src/airflow/providers/edge3/executors/edge_executor.py#L169
I tried a small experiment
Created a simple dag and ran it couple of times with retries defined.
```
from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.python import PythonOperator
def flaky_task(**context):
y = 1//0
with DAG(
dag_id="retry_example_dag",
start_date=datetime(2025, 1, 1),
schedule=None,
catchup=False,
) as dag:
retrying_task = PythonOperator(
task_id="flaky_task",
python_callable=flaky_task,
retries=5,
retry_delay=timedelta(seconds=5),
)
```
(Made some tweaks so it prints the size of the `workload.model_dump_json()` which is `command` here)
```
2025-06-14 12:50:54.751778 [info ] Task execute_workload[a7517e3a-e357-4bf4-8519-841aae0fa0aa] received [celery.worker.strategy]
2025-06-14 12:50:54.897616 [info ] [a7517e3a-e357-4bf4-8519-841aae0fa0aa] Executing workload in Celery: token='eyJ***' ti=TaskInstance(id=UUID('01976e7e-34bf-70b1-84cc-284ae6aff602'), task_id='flaky_task', dag_id='retry_example_dag', run_id='manual__2025-06-14T12:50:54.260989+00:00', try_number=1, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config=None, parent_context_carrier={}, context_carrier={}, queued_dttm=None) dag_rel_path=PurePosixPath('retry-with-delay.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=retry_example_dag/run_id=manual__2025-06-14T12:50:54.260989+00:00/task_id=flaky_task/attempt=1.log' type='ExecuteTask', len of workload is: 903 [airflow.providers.celery.executors.celery_executor_utils]
2025-06-14 12:50:54.915904 [info ] Secrets backends loaded for worker [supervisor] backend_classes=['LocalFilesystemBackend', 'EnvironmentVariablesBackend'] count=2
2025-06-14 12:50:55.545694 [info ] Task finished [supervisor] duration=0.6320609190006508 exit_code=0 final_state=up_for_retry
2025-06-14 12:50:55.550448 [info ] Task execute_workload[a7517e3a-e357-4bf4-8519-841aae0fa0aa] succeeded in 0.7965272139990702s: None [celery.app.trace]
^L[2025-06-14 12:50:57 +0000] [665] [INFO] Handling signal: winch
2025-06-14 12:51:01.275889 [info ] Task execute_workload[241e8614-d7dd-4de9-b901-32195c238f65] received [celery.worker.strategy]
2025-06-14 12:51:01.280838 [info ] [241e8614-d7dd-4de9-b901-32195c238f65] Executing workload in Celery: token='eyJ***' ti=TaskInstance(id=UUID('01976e7e-39b3-754c-ae51-ce764bb98068'), task_id='flaky_task', dag_id='retry_example_dag', run_id='manual__2025-06-14T12:50:54.260989+00:00', try_number=2, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config=None, parent_context_carrier={}, context_carrier={}, queued_dttm=datetime.datetime(2025, 6, 14, 12, 50, 54, 624148, tzinfo=TzInfo(UTC))) dag_rel_path=PurePosixPath('retry-with-delay.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=retry_example_dag/run_id=manual__2025-06-14T12:50:54.260989+00:00/task_id=flaky_task/attempt=2.log' type='ExecuteTask', len of workload is: 928 [airflow.providers.celery.executors.celery_executor_utils]
2025-06-14 12:51:01.295521 [info ] Secrets backends loaded for worker [supervisor] backend_classes=['LocalFilesystemBackend', 'EnvironmentVariablesBackend'] count=2
2025-06-14 12:51:01.453121 [info ] Task finished [supervisor] duration=0.15831252200587187 exit_code=0 final_state=up_for_retry
2025-06-14 12:51:01.458458 [info ] Task execute_workload[241e8614-d7dd-4de9-b901-32195c238f65] succeeded in 0.18147331300133374s: None [celery.app.trace]
2025-06-14 12:51:06.603739 [info ] Task execute_workload[7b287860-ad2c-4fc9-9a03-90f429aa3666] received [celery.worker.strategy]
2025-06-14 12:51:06.609338 [info ] [7b287860-ad2c-4fc9-9a03-90f429aa3666] Executing workload in Celery: token='eyJ***' ti=TaskInstance(id=UUID('01976e7e-50c7-7f57-a3da-d326eefbde78'), task_id='flaky_task', dag_id='retry_example_dag', run_id='manual__2025-06-14T12:50:54.260989+00:00', try_number=3, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config=None, parent_context_carrier={}, context_carrier={}, queued_dttm=datetime.datetime(2025, 6, 14, 12, 51, 1, 270429, tzinfo=TzInfo(UTC))) dag_rel_path=PurePosixPath('retry-with-delay.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=retry_example_dag/run_id=manual__2025-06-14T12:50:54.260989+00:00/task_id=flaky_task/attempt=3.log' type='ExecuteTask', len of workload is: 928 [airflow.providers.celery.executors.celery_executor_utils]
2025-06-14 12:51:06.624493 [info ] Secrets backends loaded for worker [supervisor] backend_classes=['LocalFilesystemBackend', 'EnvironmentVariablesBackend'] count=2
2025-06-14 12:51:06.758592 [info ] Task finished [supervisor] duration=0.13502606399561046 exit_code=0 final_state=up_for_retry
2025-06-14 12:51:06.762165 [info ] Task execute_workload[7b287860-ad2c-4fc9-9a03-90f429aa3666] succeeded in 0.15771018699888373s: None [celery.app.trace]
2025-06-14 12:51:12.682274 [info ] Task execute_workload[c2480fb0-ba96-4bbd-b5eb-0e74cd2641cc] received [celery.worker.strategy]
2025-06-14 12:51:12.686425 [info ] [c2480fb0-ba96-4bbd-b5eb-0e74cd2641cc] Executing workload in Celery: token='eyJ***' ti=TaskInstance(id=UUID('01976e7e-6580-7a9c-b89b-7b016c42fca4'), task_id='flaky_task', dag_id='retry_example_dag', run_id='manual__2025-06-14T12:50:54.260989+00:00', try_number=4, map_index=-1, pool_slots=1, queue='default', priority_weight=1, executor_config=None, parent_context_carrier={}, context_carrier={}, queued_dttm=datetime.datetime(2025, 6, 14, 12, 51, 6, 599630, tzinfo=TzInfo(UTC))) dag_rel_path=PurePosixPath('retry-with-delay.py') bundle_info=BundleInfo(name='dags-folder', version=None) log_path='dag_id=retry_example_dag/run_id=manual__2025-06-14T12:50:54.260989+00:00/task_id=flaky_task/attempt=4.log' type='ExecuteTask', len of workload is: 928 [airflow.providers.celery.executors.celery_executor_utils
```
So i can see in normal cases, its 903 and then 928.
Considering that we can max out some fields in length like:
• task_id can be up to 36 characters
• dag_id can be up to 250 characters
• Other fields (e.g., run_id, queue, log_path, executor_config, etc.) add more
I think 2048 would be a good length specifier here. I do not see a situation where it could exceed beyond that and anything shorter could be hit real soon and become limiting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Extend command column in the edge_job table to accomodate more chars [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl merged PR #51716:
URL: https://github.com/apache/airflow/pull/51716
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org