Change lazy loads from joined to selectin (#8659)

src/prefect/server/database/orm_models.py

@@ -151,7 +151,7 @@ def result_artifact_id(cls):
     def _result_artifact(cls):
         return sa.orm.relationship(
             "Artifact",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.result_artifact_id],
             primaryjoin="Artifact.id==%s.result_artifact_id" % cls.__name__,
         )
@@ -236,7 +236,7 @@ def result_artifact_id(cls):
     def _result_artifact(cls):
         return sa.orm.relationship(
             "Artifact",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.result_artifact_id],
             primaryjoin="Artifact.id==%s.result_artifact_id" % cls.__name__,
         )
@@ -524,7 +524,7 @@ def work_queue_id(cls):
     def _state(cls):
         return sa.orm.relationship(
             "FlowRunState",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.state_id],
             primaryjoin="FlowRunState.id==%s.state_id" % cls.__name__,
         )
@@ -582,7 +582,7 @@ def parent_task_run(cls):
     def work_queue(cls):
         return sa.orm.relationship(
             "WorkQueue",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.work_queue_id],
         )
 
@@ -700,7 +700,7 @@ def state_id(cls):
     def _state(cls):
         return sa.orm.relationship(
             "TaskRunState",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.state_id],
             primaryjoin="TaskRunState.id==%s.state_id" % cls.__name__,
         )
@@ -868,7 +868,7 @@ def flow(cls):
     @declared_attr
     def work_queue(cls):
         return sa.orm.relationship(
-            "WorkQueue", lazy="joined", foreign_keys=[cls.work_queue_id]
+            "WorkQueue", lazy="selectin", foreign_keys=[cls.work_queue_id]
         )
 
     @declared_attr
@@ -959,7 +959,7 @@ def block_type_id(cls):
 
     @declared_attr
     def block_type(cls):
-        return sa.orm.relationship("BlockType", lazy="joined")
+        return sa.orm.relationship("BlockType", lazy="selectin")
 
     @declared_attr
     def __table_args__(cls):
@@ -1011,7 +1011,7 @@ def block_type_id(cls):
 
     @declared_attr
     def block_type(cls):
-        return sa.orm.relationship("BlockType", lazy="joined")
+        return sa.orm.relationship("BlockType", lazy="selectin")
 
     @declared_attr
     def block_schema_id(cls):
@@ -1023,7 +1023,7 @@ def block_schema_id(cls):
 
     @declared_attr
     def block_schema(cls):
-        return sa.orm.relationship("BlockSchema", lazy="joined")
+        return sa.orm.relationship("BlockSchema", lazy="selectin")
 
     @declared_attr
     def __table_args__(cls):
@@ -1142,7 +1142,7 @@ def work_pool_id(cls):
     def work_pool(cls):
         return sa.orm.relationship(
             "WorkPool",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.work_pool_id],
         )
 
@@ -1240,7 +1240,7 @@ def block_document_id(cls):
     def block_document(cls):
         return sa.orm.relationship(
             "BlockDocument",
-            lazy="joined",
+            lazy="selectin",
             foreign_keys=[cls.block_document_id],
         )
 

src/prefect/server/models/flow_runs.py

@@ -13,7 +13,7 @@
 import sqlalchemy as sa
 from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.orm import joinedload, load_only
+from sqlalchemy.orm import load_only, selectinload
 
 import prefect.server.models as models
 import prefect.server.schemas as schemas
@@ -93,7 +93,7 @@ async def create_flow_run(
             .limit(1)
             .execution_options(populate_existing=True)
             .options(
-                joinedload(db.FlowRun.work_queue).joinedload(db.WorkQueue.work_pool)
+                selectinload(db.FlowRun.work_queue).selectinload(db.WorkQueue.work_pool)
             )
         )
         result = await session.execute(query)
@@ -158,7 +158,9 @@ async def read_flow_run(
     result = await session.execute(
         sa.select(db.FlowRun)
         .where(db.FlowRun.id == flow_run_id)
-        .options(joinedload(db.FlowRun.work_queue).joinedload(db.WorkQueue.work_pool))
+        .options(
+            selectinload(db.FlowRun.work_queue).selectinload(db.WorkQueue.work_pool)
+        )
     )
     return result.scalar()
 
@@ -266,7 +268,9 @@ async def read_flow_runs(
     query = (
         select(db.FlowRun)
         .order_by(sort.as_sql_sort(db))
-        .options(joinedload(db.FlowRun.work_queue).joinedload(db.WorkQueue.work_pool))
+        .options(
+            selectinload(db.FlowRun.work_queue).selectinload(db.WorkQueue.work_pool)
+        )
     )
 
     if columns: