Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@
"""

from alembic import op
from sqlalchemy import Column, ForeignKeyConstraint, Integer
from sqlalchemy import Column, ForeignKeyConstraint, Integer, text

from airflow.models.base import StringID
from airflow.utils.sqlalchemy import ExtendedJSON
Expand All @@ -47,12 +47,12 @@ def upgrade():
with op.batch_alter_table("task_instance") as batch_op:
# I think we always use this name for TaskInstance after 7b2661a43ba3?
batch_op.drop_constraint("task_instance_pkey", type_="primary")
batch_op.add_column(Column("map_index", Integer, nullable=False, default=-1))
batch_op.add_column(Column("map_index", Integer, nullable=False, server_default=text("-1")))
batch_op.create_primary_key("task_instance_pkey", ["dag_id", "task_id", "run_id", "map_index"])

# Re-create task_reschedule's constraints.
with op.batch_alter_table("task_reschedule") as batch_op:
batch_op.add_column(Column("map_index", Integer, nullable=False, default=-1))
batch_op.add_column(Column("map_index", Integer, nullable=False, server_default=text("-1")))
batch_op.create_foreign_key(
"task_reschedule_ti_fkey",
"task_instance",
Expand Down
3 changes: 2 additions & 1 deletion airflow/models/taskinstance.py
Original file line number Diff line number Diff line change
Expand Up @@ -60,6 +60,7 @@
func,
inspect,
or_,
text,
tuple_,
)
from sqlalchemy.ext.associationproxy import association_proxy
Expand Down Expand Up @@ -343,7 +344,7 @@ class TaskInstance(Base, LoggingMixin):
task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True, nullable=False)
dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True, nullable=False)
run_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True, nullable=False)
map_index = Column(Integer, primary_key=True, nullable=False, default=-1)
map_index = Column(Integer, primary_key=True, nullable=False, server_default=text("-1"))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This one can possibly have default and server_default.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I checked some sources and realised default isn’t actually much useful in ORM, only if we use the SQL expression builder. And with server_default set, default only really matters for UPDATE, which actually seems a bit wrong to me, so I’ll omit default until they are proven useful.


start_date = Column(UtcDateTime)
end_date = Column(UtcDateTime)
Expand Down
4 changes: 2 additions & 2 deletions airflow/models/taskreschedule.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import datetime
from typing import TYPE_CHECKING

from sqlalchemy import Column, ForeignKeyConstraint, Index, Integer, String, asc, desc
from sqlalchemy import Column, ForeignKeyConstraint, Index, Integer, String, asc, desc, text
from sqlalchemy.ext.associationproxy import association_proxy
from sqlalchemy.orm import relationship

Expand All @@ -41,7 +41,7 @@ class TaskReschedule(Base):
task_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
dag_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
run_id = Column(String(ID_LEN, **COLLATION_ARGS), nullable=False)
map_index = Column(Integer, nullable=False, default=-1)
map_index = Column(Integer, nullable=False, server_default=text("-1"))
try_number = Column(Integer, nullable=False)
start_date = Column(UtcDateTime, nullable=False)
end_date = Column(UtcDateTime, nullable=False)
Expand Down