Add ix_jobs_pipeline_fetch_q index

Author: r4victor

src/dstack/_internal/server/migrations/versions/2026/03_10_1130_8b6d5d8c1b9a_add_ix_jobs_pipeline_fetch_q_index.py

@@ -0,0 +1,51 @@
+"""Add ix_jobs_pipeline_fetch_q index
+
+Revision ID: 8b6d5d8c1b9a
+Revises: 6026b29d78c7
+Create Date: 2026-03-10 11:30:00.000000+00:00
+
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "8b6d5d8c1b9a"
+down_revision = "6026b29d78c7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.get_context().autocommit_block():
+        op.drop_index(
+            "ix_jobs_pipeline_fetch_q",
+            table_name="jobs",
+            if_exists=True,
+            postgresql_concurrently=True,
+        )
+        op.create_index(
+            "ix_jobs_pipeline_fetch_q",
+            "jobs",
+            [sa.literal_column("last_processed_at ASC")],
+            unique=False,
+            sqlite_where=sa.text("(status NOT IN ('TERMINATED', 'ABORTED', 'FAILED', 'DONE'))"),
+            postgresql_where=sa.text(
+                "(status NOT IN ('TERMINATED', 'ABORTED', 'FAILED', 'DONE'))"
+            ),
+            postgresql_concurrently=True,
+        )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.get_context().autocommit_block():
+        op.drop_index(
+            "ix_jobs_pipeline_fetch_q",
+            table_name="jobs",
+            if_exists=True,
+            postgresql_concurrently=True,
+        )
+    # ### end Alembic commands ###

src/dstack/_internal/server/models.py

@@ -515,6 +515,15 @@ class JobModel(PipelineModelMixin, BaseModel):
     should be processed only one-by-one.
     """
 
+    __table_args__ = (
+        Index(
+            "ix_jobs_pipeline_fetch_q",
+            last_processed_at.asc(),
+            postgresql_where=status.not_in(JobStatus.finished_statuses()),
+            sqlite_where=status.not_in(JobStatus.finished_statuses()),
+        ),
+    )
+
 
 class GatewayModel(PipelineModelMixin, BaseModel):
     __tablename__ = "gateways"