[DOP-34706] Include connections via symlink in hierarchy (#430)

* [DOP-34706] Include connections via symlink in hierarchy

* [DOP-34706] Moving fixtures and tests into separate files

* [DOP-34706] Separate join in two unions

* [DOP-34706] Separate IO queries

* [DOP-34706] move select to variable

* [DOP-34706] remove prefix from query

* [DOP-34706] remove prefix from query

* [DOP-34706] remove CTE for core part of the query

Author: TiGrib

data_rentgen/db/repositories/job_dependency.py

@@ -5,7 +5,6 @@
 
 from sqlalchemy import (
     ARRAY,
-    CTE,
     CompoundSelect,
     DateTime,
     Integer,
@@ -21,6 +20,7 @@
     tuple_,
 )
 
+from data_rentgen.db.models.dataset_symlink import DatasetSymlink
 from data_rentgen.db.models.input import Input
 from data_rentgen.db.models.job_dependency import JobDependency
 from data_rentgen.db.models.output import Output
@@ -104,13 +104,14 @@ async def get_dependencies(
         infer_from_lineage: bool = False,
     ) -> list[JobDependency]:
         core_query = self._get_core_hierarchy_query(include_indirect=infer_from_lineage)
+        core_subquery = core_query.subquery()
 
-        query: Select | CompoundSelect
+        query: Select
         match direction:
             case "UPSTREAM":
-                query = select(core_query).where(core_query.c.to_job_id == any_(bindparam("job_ids")))
+                query = select(core_subquery).where(core_subquery.c.to_job_id == any_(bindparam("job_ids")))
             case "DOWNSTREAM":
-                query = select(core_query).where(core_query.c.from_job_id == any_(bindparam("job_ids")))
+                query = select(core_subquery).where(core_subquery.c.from_job_id == any_(bindparam("job_ids")))
 
         result = await self._session.execute(
             query,
@@ -125,37 +126,58 @@ def _get_core_hierarchy_query(
         self,
         *,
         include_indirect: bool = False,
-    ) -> CTE:
+    ) -> Select | CompoundSelect:
         query: Select | CompoundSelect
         query = select(
             JobDependency.from_job_id,
             JobDependency.to_job_id,
             JobDependency.type,
         )
         if include_indirect:
-            query = query.union(
-                select(
-                    Output.job_id.label("from_job_id"),
-                    Input.job_id.label("to_job_id"),
-                    literal("INFERRED_FROM_LINEAGE").label("type"),
-                )
-                .distinct()
+            # Where clause and columns are common part for all unions
+            where_clauses = [
+                Input.created_at >= bindparam("since"),
+                Output.created_at >= bindparam("since"),
+                Output.created_at >= Input.created_at,
+                Output.job_id != Input.job_id,
+                or_(
+                    bindparam("until", type_=DateTime(timezone=True)).is_(None),
+                    and_(
+                        Input.created_at <= bindparam("until"),
+                        Output.created_at <= bindparam("until"),
+                    ),
+                ),
+            ]
+            inferred_columns = select(
+                Output.job_id.label("from_job_id"),
+                Input.job_id.label("to_job_id"),
+                literal("INFERRED_FROM_LINEAGE").label("type"),
+            ).distinct()
+
+            # IO connections via same dataset
+            direct_connection = inferred_columns.join(
+                Input,
+                Output.dataset_id == Input.dataset_id,
+            ).where(*where_clauses)
+            # IO connections Output.d_id == Symlink.to_d_id Symlink.from_d_id == Input.d_id
+            via_symlinks_from_output = (
+                inferred_columns.join(DatasetSymlink, Output.dataset_id == DatasetSymlink.to_dataset_id)
                 .join(
                     Input,
-                    Output.dataset_id == Input.dataset_id,
+                    DatasetSymlink.from_dataset_id == Input.dataset_id,
                 )
-                .where(
-                    Input.created_at >= bindparam("since"),
-                    Output.created_at >= bindparam("since"),
-                    Output.created_at >= Input.created_at,
-                    Output.job_id != Input.job_id,
-                    or_(
-                        bindparam("until", type_=DateTime(timezone=True)).is_(None),
-                        and_(
-                            Input.created_at <= bindparam("until"),
-                            Output.created_at <= bindparam("until"),
-                        ),
-                    ),
+                .where(*where_clauses)
+            )
+            # IO connections Input.d_id == Symlink.to_d_id Symlink.from_d_id == Output.d_id
+            via_symlinks_from_input = (
+                inferred_columns.join(DatasetSymlink, Input.dataset_id == DatasetSymlink.to_dataset_id)
+                .join(
+                    Output,
+                    DatasetSymlink.from_dataset_id == Output.dataset_id,
                 )
+                .where(*where_clauses)
             )
-        return query.cte("jobs_hierarchy_core_query").prefix_with("NOT MATERIALIZED", dialect="postgresql")
+
+            query = query.union(direct_connection, via_symlinks_from_input, via_symlinks_from_output)
+
+        return query

data_rentgen/server/settings/cors.py

@@ -50,10 +50,10 @@ class CORSSettings(BaseModel):
         default=False,
         description="If ``True``, cookies should be supported for cross-origin request",
     )
-    allow_methods: list[str] = Field(default=["GET"], description="HTTP Methods allowed for CORS")
+    allow_methods: list[str] = Field(default=["GET", "POST"], description="HTTP Methods allowed for CORS")
     # https://github.com/snok/asgi-correlation-id#cors
     allow_headers: list[str] = Field(
-        default=["X-Request-ID", "X-Request-With"],
+        default=["X-Request-ID", "X-Request-With", "Access-Control-Allow-Origin"],
         description="HTTP headers allowed for CORS",
     )
     expose_headers: list[str] = Field(default=["X-Request-ID"], description="HTTP headers exposed from backend")

tests/conftest.py

@@ -29,6 +29,7 @@
     "tests.test_server.fixtures.keycloak",
     "tests.test_server.fixtures.factories.address",
     "tests.test_server.fixtures.factories.dataset",
+    "tests.test_server.fixtures.factories.job_dependencies",
     "tests.test_server.fixtures.factories.job_type",
     "tests.test_server.fixtures.factories.job",
     "tests.test_server.fixtures.factories.lineage",