From cdab840cb7039cb6591d4639b7f17ef8c1365fd2 Mon Sep 17 00:00:00 2001
From: Kacper Muda <mudakacper@gmail.com>
Date: Thu, 6 Nov 2025 12:31:54 +0100
Subject: [PATCH] feat(openlineage): Add parentRunFacet for DAG events

---
 providers/openlineage/docs/guides/user.rst    |  50 ++++
 .../providers/openlineage/plugins/listener.py |  43 +++-
 .../providers/openlineage/plugins/macros.py   |  70 ++++--
 .../openlineage/plugins/openlineage.py        |   2 +
 .../providers/openlineage/utils/utils.py      | 168 ++++++++++++-
 .../example_openlineage_trigger_dag.py        |  13 +-
 .../openlineage_trigger_dag.json              | 160 +++++++++++-
 .../unit/openlineage/plugins/test_macros.py   | 210 +++++++++++++++-
 .../unit/openlineage/utils/test_utils.py      | 233 ++++++++++++++++++
 9 files changed, 897 insertions(+), 52 deletions(-)

diff --git a/providers/openlineage/docs/guides/user.rst b/providers/openlineage/docs/guides/user.rst
index 730d7d0f2ffc6..50961dc08a5de 100644
--- a/providers/openlineage/docs/guides/user.rst
+++ b/providers/openlineage/docs/guides/user.rst
@@ -478,6 +478,56 @@ You can enable this automation by setting ``spark_inject_transport_info`` option
   AIRFLOW__OPENLINEAGE__SPARK_INJECT_TRANSPORT_INFO=true
 
 
+Passing parent information to Airflow DAG
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+To enable full OpenLineage lineage tracking across dependent DAGs, you can pass parent and root job information
+through the DAG's ``dag_run.conf``. When a DAG run configuration includes an ``openlineage`` section with valid metadata,
+this information is automatically parsed and converted into DAG run's ``parentRunFacet``, from which the root information
+is also propagated to all tasks. If no DAG run ``openlineage`` configuration is provided, the DAG run will not contain
+``parentRunFacet`` and root of all tasks will default to Dag run.
+
+The ``openlineage`` dict in conf should contain the following keys:
+
+
+*(all three values must be included to create a parent reference)*
+
+- **parentRunId** — the unique run ID (uuid) of the direct parent job
+- **parentJobName** — the name of the parent job
+- **parentJobNamespace** — the namespace of the parent job
+
+*(all three values must be included to create a root reference, otherwise parent will be used as root)*
+
+- **rootParentRunId** — the run ID (uuid) of the top-level (root) job
+- **rootParentJobName** — the name of the top-level (root) job
+- **rootParentJobNamespace** — the namespace of the top-level (root) job
+
+.. note::
+
+  We highly recommend providing all six OpenLineage identifiers (parent and root) to ensure complete lineage tracking. If the root information is missing, the parent set will be used as the root; if any of the three parent fields are missing, no parent facet will be created. Partial or mixed configurations are not supported - either all three parent or all three root values must be provided together.
+
+
+Example:
+
+.. code-block:: shell
+
+    curl -X POST "http://<AIRFLOW_HOST>/api/v2/dags/my_dag_name/dagRuns" \
+    -H "Content-Type: application/json" \
+    -d '{
+      "logical_date": "2019-08-24T14:15:22Z",
+      "conf": {
+        "openlineage": {
+          "parentRunId": "3bb703d1-09c1-4a42-8da5-35a0b3216072",
+          "parentJobNamespace": "prod_biz",
+          "parentJobName": "get_files",
+          "rootParentRunId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e",
+          "rootParentJobNamespace": "prod_analytics",
+          "rootParentJobName": "generate_report_sales_e2e"
+        }
+      }
+    }'
+
+
 Troubleshooting
 ===============
 
diff --git a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
index bd1931e3ca9f8..d7e87c8696527 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
@@ -41,7 +41,9 @@
     get_airflow_mapped_task_facet,
     get_airflow_run_facet,
     get_dag_documentation,
+    get_dag_parent_run_facet,
     get_job_name,
+    get_root_information_from_dagrun_conf,
     get_task_documentation,
     get_task_parent_run_facet,
     get_user_provided_run_facets,
@@ -224,7 +226,11 @@ def on_running():
                 task=task_metadata,
                 run_facets={
                     **get_user_provided_run_facets(task_instance, TaskInstanceState.RUNNING),
-                    **get_task_parent_run_facet(parent_run_id=parent_run_id, parent_job_name=dag.dag_id),
+                    **get_task_parent_run_facet(
+                        parent_run_id=parent_run_id,
+                        parent_job_name=dag.dag_id,
+                        **get_root_information_from_dagrun_conf(getattr(dagrun, "conf", {})),
+                    ),
                     **get_airflow_mapped_task_facet(task_instance),
                     **get_airflow_run_facet(dagrun, dag, task_instance, task, task_uuid),
                     **debug_facet,
@@ -351,7 +357,11 @@ def on_success():
                 nominal_end_time=data_interval_end,
                 run_facets={
                     **get_user_provided_run_facets(task_instance, TaskInstanceState.SUCCESS),
-                    **get_task_parent_run_facet(parent_run_id=parent_run_id, parent_job_name=dag.dag_id),
+                    **get_task_parent_run_facet(
+                        parent_run_id=parent_run_id,
+                        parent_job_name=dag.dag_id,
+                        **get_root_information_from_dagrun_conf(getattr(dagrun, "conf", {})),
+                    ),
                     **get_airflow_run_facet(dagrun, dag, task_instance, task, task_uuid),
                     **get_airflow_debug_facet(),
                 },
@@ -489,7 +499,11 @@ def on_failure():
                 job_description_type=doc_type,
                 run_facets={
                     **get_user_provided_run_facets(task_instance, TaskInstanceState.FAILED),
-                    **get_task_parent_run_facet(parent_run_id=parent_run_id, parent_job_name=dag.dag_id),
+                    **get_task_parent_run_facet(
+                        parent_run_id=parent_run_id,
+                        parent_job_name=dag.dag_id,
+                        **get_root_information_from_dagrun_conf(getattr(dagrun, "conf", {})),
+                    ),
                     **get_airflow_run_facet(dagrun, dag, task_instance, task, task_uuid),
                     **get_airflow_debug_facet(),
                 },
@@ -540,7 +554,11 @@ def on_state_change():
                 "job_description": None,
                 "job_description_type": None,
                 "run_facets": {
-                    **get_task_parent_run_facet(parent_run_id=parent_run_id, parent_job_name=ti.dag_id),
+                    **get_task_parent_run_facet(
+                        parent_run_id=parent_run_id,
+                        parent_job_name=ti.dag_id,
+                        **get_root_information_from_dagrun_conf(getattr(dagrun, "conf", {})),
+                    ),
                     **get_airflow_debug_facet(),
                 },
             }
@@ -645,8 +663,6 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
             )
             data_interval_end = dag_run.data_interval_end.isoformat() if dag_run.data_interval_end else None
 
-            run_facets = {**get_airflow_dag_run_facet(dag_run)}
-
             date = dag_run.logical_date
             if AIRFLOW_V_3_0_PLUS and date is None:
                 date = dag_run.run_after
@@ -660,7 +676,6 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
                 start_date=dag_run.start_date,
                 nominal_start_time=data_interval_start,
                 nominal_end_time=data_interval_end,
-                run_facets=run_facets,
                 clear_number=dag_run.clear_number,
                 owners=[x.strip() for x in dag_run.dag.owner.split(",")] if dag_run.dag else None,
                 job_description=doc,
@@ -669,6 +684,10 @@ def on_dag_run_running(self, dag_run: DagRun, msg: str) -> None:
                 # AirflowJobFacet should be created outside ProcessPoolExecutor that pickles objects,
                 # as it causes lack of some TaskGroup attributes and crashes event emission.
                 job_facets=get_airflow_job_facet(dag_run=dag_run),
+                run_facets={
+                    **get_airflow_dag_run_facet(dag_run),
+                    **get_dag_parent_run_facet(getattr(dag_run, "conf", {})),
+                },
             )
         except BaseException as e:
             self.log.warning("OpenLineage received exception in method on_dag_run_running", exc_info=e)
@@ -716,7 +735,10 @@ def on_dag_run_success(self, dag_run: DagRun, msg: str) -> None:
                 job_description_type=doc_type,
                 task_ids=task_ids,
                 dag_run_state=dag_run.get_state(),
-                run_facets={**get_airflow_dag_run_facet(dag_run)},
+                run_facets={
+                    **get_airflow_dag_run_facet(dag_run),
+                    **get_dag_parent_run_facet(getattr(dag_run, "conf", {})),
+                },
             )
         except BaseException as e:
             self.log.warning("OpenLineage received exception in method on_dag_run_success", exc_info=e)
@@ -765,7 +787,10 @@ def on_dag_run_failed(self, dag_run: DagRun, msg: str) -> None:
                 dag_run_state=dag_run.get_state(),
                 task_ids=task_ids,
                 msg=msg,
-                run_facets={**get_airflow_dag_run_facet(dag_run)},
+                run_facets={
+                    **get_airflow_dag_run_facet(dag_run),
+                    **get_dag_parent_run_facet(getattr(dag_run, "conf", {})),
+                },
             )
         except BaseException as e:
             self.log.warning("OpenLineage received exception in method on_dag_run_failed", exc_info=e)
diff --git a/providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py
index ac9c75b4a941b..508c89eafda47 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py
@@ -20,7 +20,7 @@
 
 from airflow.providers.openlineage import conf
 from airflow.providers.openlineage.plugins.adapter import OpenLineageAdapter
-from airflow.providers.openlineage.utils.utils import get_job_name
+from airflow.providers.openlineage.utils.utils import get_job_name, get_root_information_from_dagrun_conf
 from airflow.providers.openlineage.version_compat import AIRFLOW_V_3_0_PLUS
 
 if TYPE_CHECKING:
@@ -102,7 +102,7 @@ def lineage_root_parent_id(task_instance: TaskInstance):
     """
     return "/".join(
         (
-            lineage_job_namespace(),
+            lineage_root_job_namespace(task_instance),
             lineage_root_job_name(task_instance),
             lineage_root_run_id(task_instance),
         )
@@ -110,10 +110,16 @@ def lineage_root_parent_id(task_instance: TaskInstance):
 
 
 def lineage_root_job_name(task_instance: TaskInstance):
+    root_parent_job_name = _get_ol_root_id("root_parent_job_name", task_instance)
+    if root_parent_job_name:
+        return root_parent_job_name
     return task_instance.dag_id
 
 
 def lineage_root_run_id(task_instance: TaskInstance):
+    root_parent_run_id = _get_ol_root_id("root_parent_run_id", task_instance)
+    if root_parent_run_id:
+        return root_parent_run_id
     return OpenLineageAdapter.build_dag_run_id(
         dag_id=task_instance.dag_id,
         logical_date=_get_logical_date(task_instance),
@@ -121,32 +127,44 @@ def lineage_root_run_id(task_instance: TaskInstance):
     )
 
 
+def lineage_root_job_namespace(task_instance: TaskInstance):
+    root_parent_job_namespace = _get_ol_root_id("root_parent_job_namespace", task_instance)
+    if root_parent_job_namespace:
+        return root_parent_job_namespace
+    return conf.namespace()
+
+
+def _get_ol_root_id(id_key: str, task_instance: TaskInstance) -> str | None:
+    dr_conf = _get_dag_run_conf(task_instance=task_instance)
+    ol_root_info = get_root_information_from_dagrun_conf(dr_conf=dr_conf)
+    if ol_root_info and ol_root_info.get(id_key):
+        return ol_root_info[id_key]
+    return None
+
+
+def _get_dagrun_from_ti(task_instance: TaskInstance):
+    context = task_instance.get_template_context()
+    if getattr(task_instance, "dag_run", None):
+        return task_instance.dag_run
+    return context["dag_run"]
+
+
+def _get_dag_run_conf(task_instance: TaskInstance) -> dict:
+    dr = _get_dagrun_from_ti(task_instance=task_instance)
+    return dr.conf or {}
+
+
 def _get_dag_run_clear_number(task_instance: TaskInstance):
-    # todo: remove when min airflow version >= 3.0
-    if AIRFLOW_V_3_0_PLUS:
-        context = task_instance.get_template_context()
-        if hasattr(task_instance, "dag_run"):
-            dag_run = task_instance.dag_run
-        else:
-            dag_run = context["dag_run"]
-        return dag_run.clear_number
-    return task_instance.dag_run.clear_number
+    dr = _get_dagrun_from_ti(task_instance=task_instance)
+    return dr.clear_number
 
 
 def _get_logical_date(task_instance):
-    # todo: remove when min airflow version >= 3.0
     if AIRFLOW_V_3_0_PLUS:
-        context = task_instance.get_template_context()
-        if hasattr(task_instance, "dag_run"):
-            dag_run = task_instance.dag_run
-        else:
-            dag_run = context["dag_run"]
-        if hasattr(dag_run, "logical_date") and dag_run.logical_date:
-            date = dag_run.logical_date
-        else:
-            date = dag_run.run_after
-    elif hasattr(task_instance, "logical_date"):
-        date = task_instance.logical_date
-    else:
-        date = task_instance.execution_date
-    return date
+        dr = _get_dagrun_from_ti(task_instance=task_instance)
+        if getattr(dr, "logical_date", None):
+            return dr.logical_date
+        return dr.run_after
+    if getattr(task_instance, "logical_date", None):
+        return task_instance.logical_date
+    return task_instance.execution_date
diff --git a/providers/openlineage/src/airflow/providers/openlineage/plugins/openlineage.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/openlineage.py
index 964cac27254e2..8e422d84ac604 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/plugins/openlineage.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/plugins/openlineage.py
@@ -28,6 +28,7 @@
         lineage_job_namespace,
         lineage_parent_id,
         lineage_root_job_name,
+        lineage_root_job_namespace,
         lineage_root_parent_id,
         lineage_root_run_id,
         lineage_run_id,
@@ -51,6 +52,7 @@ class OpenLineageProviderPlugin(AirflowPlugin):
             lineage_parent_id,
             lineage_root_run_id,
             lineage_root_job_name,
+            lineage_root_job_namespace,
             lineage_root_parent_id,
         ]
         listeners = [get_openlineage_listener()]
diff --git a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
index bd3901e84e76c..ac86ce28453af 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
@@ -147,27 +147,175 @@ def get_job_name(task: TaskInstance | RuntimeTaskInstance) -> str:
     return f"{task.dag_id}.{task.task_id}"
 
 
-def get_task_parent_run_facet(
-    parent_run_id: str, parent_job_name: str, parent_job_namespace: str = conf.namespace()
+def _get_parent_run_facet(
+    parent_run_id: str,
+    parent_job_name: str,
+    parent_job_namespace: str = conf.namespace(),
+    root_parent_run_id: str | None = None,
+    root_parent_job_name: str | None = None,
+    root_parent_job_namespace: str | None = None,
 ) -> dict[str, Any]:
-    """
-    Retrieve the parent run facet for task-level events.
-
-    This facet currently always points to the DAG-level run ID and name,
-    as external events for DAG runs are not yet handled.
-    """
+    """Create the parent run facet from identifiers."""
+    root_parent_run_id = root_parent_run_id or parent_run_id
+    root_parent_job_name = root_parent_job_name or parent_job_name
+    root_parent_job_namespace = root_parent_job_namespace or parent_job_namespace
     return {
         "parent": parent_run.ParentRunFacet(
             run=parent_run.Run(runId=parent_run_id),
             job=parent_run.Job(namespace=parent_job_namespace, name=parent_job_name),
             root=parent_run.Root(
-                run=parent_run.RootRun(runId=parent_run_id),
-                job=parent_run.RootJob(namespace=parent_job_namespace, name=parent_job_name),
+                run=parent_run.RootRun(runId=root_parent_run_id),
+                job=parent_run.RootJob(namespace=root_parent_job_namespace, name=root_parent_job_name),
             ),
         )
     }
 
 
+def get_task_parent_run_facet(
+    parent_run_id: str,
+    parent_job_name: str,
+    parent_job_namespace: str = conf.namespace(),
+    root_parent_run_id: str | None = None,
+    root_parent_job_name: str | None = None,
+    root_parent_job_namespace: str | None = None,
+) -> dict[str, Any]:
+    """Retrieve the parent run facet."""
+    return _get_parent_run_facet(
+        parent_run_id=parent_run_id,
+        parent_job_namespace=parent_job_namespace,
+        parent_job_name=parent_job_name,
+        root_parent_run_id=root_parent_run_id,
+        root_parent_job_namespace=root_parent_job_namespace,
+        root_parent_job_name=root_parent_job_name,
+    )
+
+
+def _get_openlineage_data_from_dagrun_conf(dr_conf: dict | None) -> dict:
+    """Return the 'openlineage' section from a DAG run config if valid, otherwise an empty dict."""
+    if not dr_conf or not isinstance(dr_conf, dict):
+        return {}
+    ol_data = dr_conf.get("openlineage")
+    return ol_data if isinstance(ol_data, dict) else {}
+
+
+def get_root_information_from_dagrun_conf(dr_conf: dict | None) -> dict[str, str]:
+    """Extract root parent run and job information from a DAG run config."""
+    ol_data = _get_openlineage_data_from_dagrun_conf(dr_conf)
+    if not ol_data:
+        log.debug("No 'openlineage' data found in DAG run config.")
+        return {}
+
+    root_run_id = ol_data.get("rootParentRunId", "")
+    root_namespace = ol_data.get("rootParentJobNamespace", "")
+    root_name = ol_data.get("rootParentJobName", "")
+
+    all_root_info = (root_run_id, root_namespace, root_name)
+    if not all(all_root_info):
+        if any(all_root_info):
+            log.warning(
+                "Incomplete root OpenLineage information in DAG run config. "
+                "No root information will be used. Found values: "
+                "rootParentRunId='%s', rootParentJobNamespace='%s', rootParentJobName='%s'.",
+                root_run_id,
+                root_namespace,
+                root_name,
+            )
+        else:
+            log.debug("No 'openlineage' root information found in DAG run config.")
+        return {}
+
+    try:  # Validate that runId is correct UUID
+        parent_run.RootRun(runId=root_run_id)
+    except ValueError:
+        log.warning(
+            "Invalid OpenLineage rootParentRunId '%s' in DAG run config - expected a valid UUID.",
+            root_run_id,
+        )
+        return {}
+
+    log.debug(
+        "Extracted valid root OpenLineage identifiers from DAG run config: "
+        "rootParentRunId='%s', rootParentJobNamespace='%s', rootParentJobName='%s'.",
+        root_run_id,
+        root_namespace,
+        root_name,
+    )
+    return {
+        "root_parent_run_id": root_run_id,
+        "root_parent_job_namespace": root_namespace,
+        "root_parent_job_name": root_name,
+    }
+
+
+def get_dag_parent_run_facet(dr_conf: dict | None) -> dict[str, parent_run.ParentRunFacet]:
+    """Build the OpenLineage parent run facet from a DAG run config."""
+    ol_data = _get_openlineage_data_from_dagrun_conf(dr_conf)
+    if not ol_data:
+        log.debug("No 'openlineage' data found in DAG run config.")
+        return {}
+
+    parent_run_id = ol_data.get("parentRunId", "")
+    parent_job_namespace = ol_data.get("parentJobNamespace", "")
+    parent_job_name = ol_data.get("parentJobName", "")
+
+    all_parent_info = (parent_run_id, parent_job_namespace, parent_job_name)
+    if not all(all_parent_info):
+        if any(all_parent_info):
+            log.warning(
+                "Incomplete parent OpenLineage information in DAG run config. "
+                "ParentRunFacet will NOT be created. Found values: "
+                "parentRunId='%s', parentJobNamespace='%s', parentJobName='%s'.",
+                parent_run_id,
+                parent_job_namespace,
+                parent_job_name,
+            )
+        else:
+            log.debug("No 'openlineage' parent information found in DAG run config.")
+        return {}
+
+    try:  # Validate that runId is correct UUID
+        parent_run.RootRun(runId=parent_run_id)
+    except ValueError:
+        log.warning(
+            "Invalid OpenLineage parentRunId '%s' in DAG run config - expected a valid UUID.",
+            parent_run_id,
+        )
+        return {}
+
+    log.debug(
+        "Extracted valid parent OpenLineage identifiers from DAG run config: "
+        "parentRunId='%s', parentJobNamespace='%s', parentJobName='%s'.",
+        parent_run_id,
+        parent_job_namespace,
+        parent_job_name,
+    )
+
+    root_info = get_root_information_from_dagrun_conf(dr_conf)
+    if root_info and all(root_info.values()):
+        root_parent_run_id = root_info["root_parent_run_id"]
+        root_parent_job_namespace = root_info["root_parent_job_namespace"]
+        root_parent_job_name = root_info["root_parent_job_name"]
+    else:
+        log.debug(
+            "Missing OpenLineage root identifiers in DAG run config, "
+            "parent identifiers will be used as root instead."
+        )
+        root_parent_run_id, root_parent_job_namespace, root_parent_job_name = (
+            parent_run_id,
+            parent_job_namespace,
+            parent_job_name,
+        )
+
+    return _get_parent_run_facet(
+        parent_run_id=parent_run_id,
+        parent_job_namespace=parent_job_namespace,
+        parent_job_name=parent_job_name,
+        root_parent_run_id=root_parent_run_id,
+        root_parent_job_namespace=root_parent_job_namespace,
+        root_parent_job_name=root_parent_job_name,
+    )
+
+
 def _truncate_string_to_byte_size(s: str, max_size: int = _MAX_DOC_BYTES) -> str:
     """
     Truncate a string to a maximum UTF-8 byte size, ensuring valid encoding.
diff --git a/providers/openlineage/tests/system/openlineage/example_openlineage_trigger_dag.py b/providers/openlineage/tests/system/openlineage/example_openlineage_trigger_dag.py
index 513898a51f974..fb524344bab4d 100644
--- a/providers/openlineage/tests/system/openlineage/example_openlineage_trigger_dag.py
+++ b/providers/openlineage/tests/system/openlineage/example_openlineage_trigger_dag.py
@@ -20,6 +20,7 @@
 It checks:
     - task's trigger_dag_id
     - DAGRun START and COMPLETE events, for the triggered DAG
+    - propagation of OL parent and root info from DAGRun conf
 """
 
 from __future__ import annotations
@@ -47,7 +48,17 @@
         trigger_dag_id="openlineage_trigger_dag_child__notrigger",
         trigger_run_id=f"openlineage_trigger_dag_triggering_child_{datetime.now().isoformat()}",
         wait_for_completion=True,
-        conf={"some_config": "value1"},
+        conf={
+            "some_config": "value1",
+            "openlineage": {
+                "parentRunId": "3bb703d1-09c1-4a42-8da5-35a0b3216072",
+                "parentJobNamespace": "prod_biz",
+                "parentJobName": "get_files",
+                "rootParentRunId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e",
+                "rootParentJobNamespace": "prod_analytics",
+                "rootParentJobName": "generate_report_sales_e2e",
+            },
+        },
         poke_interval=10,
     )
 
diff --git a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_trigger_dag.json b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_trigger_dag.json
index 45f742efad6d9..3695408b28fbd 100644
--- a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_trigger_dag.json
+++ b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_trigger_dag.json
@@ -21,7 +21,15 @@
                     },
                     "dagRun": {
                         "conf": {
-                            "some_config": "value1"
+                            "some_config": "value1",
+                            "openlineage": {
+                                "parentRunId": "3bb703d1-09c1-4a42-8da5-35a0b3216072",
+                                "parentJobNamespace": "prod_biz",
+                                "parentJobName": "get_files",
+                                "rootParentRunId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e",
+                                "rootParentJobNamespace": "prod_analytics",
+                                "rootParentJobName": "generate_report_sales_e2e"
+                            }
                         },
                         "dag_id": "openlineage_trigger_dag_child__notrigger",
                         "data_interval_end": "{{ is_datetime(result) }}",
@@ -34,6 +42,26 @@
                     "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
                     "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/[\\d-]+\\/OpenLineage.json\\#\\/\\$defs\\/RunFacet\") }}"
                 },
+                "parent": {
+                  "job": {
+                    "namespace": "prod_biz",
+                    "name": "get_files"
+                  },
+                  "run": {
+                    "runId": "3bb703d1-09c1-4a42-8da5-35a0b3216072"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "generate_report_sales_e2e",
+                      "namespace": "prod_analytics"
+                    },
+                    "run": {
+                      "runId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
+                },
                 "nominalTime": {
                     "nominalEndTime": "{{ is_datetime(result) }}",
                     "nominalStartTime": "{{ is_datetime(result) }}",
@@ -153,7 +181,15 @@
                     },
                     "dagRun": {
                         "conf": {
-                            "some_config": "value1"
+                            "some_config": "value1",
+                            "openlineage": {
+                                "parentRunId": "3bb703d1-09c1-4a42-8da5-35a0b3216072",
+                                "parentJobNamespace": "prod_biz",
+                                "parentJobName": "get_files",
+                                "rootParentRunId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e",
+                                "rootParentJobNamespace": "prod_analytics",
+                                "rootParentJobName": "generate_report_sales_e2e"
+                            }
                         },
                         "dag_id": "openlineage_trigger_dag_child__notrigger",
                         "data_interval_end": "{{ is_datetime(result) }}",
@@ -167,6 +203,26 @@
                     "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
                     "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/[\\d-]+\\/OpenLineage.json\\#\\/\\$defs\\/RunFacet\") }}"
                 },
+                "parent": {
+                  "job": {
+                    "namespace": "prod_biz",
+                    "name": "get_files"
+                  },
+                  "run": {
+                    "runId": "3bb703d1-09c1-4a42-8da5-35a0b3216072"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "generate_report_sales_e2e",
+                      "namespace": "prod_analytics"
+                    },
+                    "run": {
+                      "runId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
+                },
                 "nominalTime": {
                     "nominalEndTime": "{{ is_datetime(result) }}",
                     "nominalStartTime": "{{ is_datetime(result) }}",
@@ -237,6 +293,66 @@
             }
         }
     },
+    {
+        "eventType": "START",
+        "run": {
+            "facets": {
+                "parent": {
+                  "job": {
+                    "namespace": "{{ result is string }}",
+                    "name": "openlineage_trigger_dag_child__notrigger"
+                  },
+                  "run": {
+                    "runId": "{{ is_uuid(result) }}"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "generate_report_sales_e2e",
+                      "namespace": "prod_analytics"
+                    },
+                    "run": {
+                      "runId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
+                }
+            }
+        },
+        "job": {
+            "name": "openlineage_trigger_dag_child__notrigger.do_nothing_task"
+        }
+    },
+    {
+        "eventType": "COMPLETE",
+        "run": {
+            "facets": {
+                "parent": {
+                  "job": {
+                    "namespace": "{{ result is string }}",
+                    "name": "openlineage_trigger_dag_child__notrigger"
+                  },
+                  "run": {
+                    "runId": "{{ is_uuid(result) }}"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "generate_report_sales_e2e",
+                      "namespace": "prod_analytics"
+                    },
+                    "run": {
+                      "runId": "9d3b14f7-de91-40b6-aeef-e887e2c7673e"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
+                }
+            }
+        },
+        "job": {
+            "name": "openlineage_trigger_dag_child__notrigger.do_nothing_task"
+        }
+    },
     {
         "eventType": "START",
         "run": {
@@ -245,6 +361,26 @@
                     "task": {
                         "trigger_dag_id": "openlineage_trigger_dag_child__notrigger"
                     }
+                },
+                "parent": {
+                  "job": {
+                    "namespace": "{{ result is string }}",
+                    "name": "openlineage_trigger_dag"
+                  },
+                  "run": {
+                    "runId": "{{ is_uuid(result) }}"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "openlineage_trigger_dag",
+                      "namespace": "{{ result is string }}"
+                    },
+                    "run": {
+                      "runId": "{{ is_uuid(result) }}"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
                 }
             }
         },
@@ -260,6 +396,26 @@
                     "task": {
                         "trigger_dag_id": "openlineage_trigger_dag_child__notrigger"
                     }
+                },
+                "parent": {
+                  "job": {
+                    "namespace": "{{ result is string }}",
+                    "name": "openlineage_trigger_dag"
+                  },
+                  "run": {
+                    "runId": "{{ is_uuid(result) }}"
+                  },
+                  "root": {
+                    "job": {
+                      "name": "openlineage_trigger_dag",
+                      "namespace": "{{ result is string }}"
+                    },
+                    "run": {
+                      "runId": "{{ is_uuid(result) }}"
+                    }
+                  },
+                  "_producer": "{{ regex_match(result, \"^https:\\/\\/github.com/apache/airflow/tree/providers-openlineage\\/[\\d]+\\.[\\d]+\\.[\\d]+.*$\") }}",
+                  "_schemaURL": "{{ regex_match(result, \"^https:\\/\\/openlineage.io\\/spec\\/facets\\/[\\d-]+\\/ParentRunFacet.json\\#\\/\\$defs\\/ParentRunFacet$\") }}"
                 }
             }
         },
diff --git a/providers/openlineage/tests/unit/openlineage/plugins/test_macros.py b/providers/openlineage/tests/unit/openlineage/plugins/test_macros.py
index 520421be76862..3cdde5d183f75 100644
--- a/providers/openlineage/tests/unit/openlineage/plugins/test_macros.py
+++ b/providers/openlineage/tests/unit/openlineage/plugins/test_macros.py
@@ -21,12 +21,13 @@
 
 import pytest
 
-from airflow import __version__
 from airflow.providers.openlineage.conf import namespace
 from airflow.providers.openlineage.plugins.macros import (
     lineage_job_name,
     lineage_job_namespace,
     lineage_parent_id,
+    lineage_root_job_name,
+    lineage_root_job_namespace,
     lineage_root_run_id,
     lineage_run_id,
 )
@@ -35,10 +36,10 @@
 
 _DAG_NAMESPACE = namespace()
 
-if __version__.startswith("2."):
-    LOGICAL_DATE_KEY = "execution_date"
-else:
+if AIRFLOW_V_3_0_PLUS:
     LOGICAL_DATE_KEY = "logical_date"
+else:
+    LOGICAL_DATE_KEY = "execution_date"
 
 
 def test_lineage_job_namespace():
@@ -130,3 +131,204 @@ def test_lineage_root_run_id_with_runtime_task_instance(create_runtime_ti):
         assert lineage_root_run_id(runtime_ti) is not None
     except AttributeError as e:
         pytest.fail(f"lineage_root_run_id should not throw AttributeError with RuntimeTaskInstance: {e}")
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_run_id_no_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_id="test_dag",
+        run_id="test_run_id",
+        conf=None,
+    )
+
+    result = lineage_root_run_id(runtime_ti)
+    assert result == "01937fbb-4680-70b3-b49b-1de6b041527a"
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_run_id_with_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_id="test_dag",
+        run_id="test_run_id",
+        conf={
+            "openlineage": {
+                "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+                "rootParentJobNamespace": "rootns",
+                "rootParentJobName": "rootjob",
+            }
+        },
+    )
+
+    result = lineage_root_run_id(runtime_ti)
+    assert result == "22222222-2222-2222-2222-222222222222"
+
+
+def test_lineage_root_run_id_without_conf_af2():
+    date = datetime(2020, 1, 1, 1, 1, 1, 0, tzinfo=timezone.utc)
+    conf = {}
+    dag_run = mock.MagicMock(run_id="run_id", conf=conf)
+    dag_run.logical_date = date
+    dag_run.clear_number = 1
+    task_instance = mock.MagicMock(
+        dag_id="dag_id",
+        task_id="task_id",
+        dag_run=dag_run,
+        logical_date=date,
+        try_number=1,
+    )
+
+    call_result1 = lineage_root_run_id(task_instance)
+    call_result2 = lineage_root_run_id(task_instance)
+
+    # random part value does not matter, it just has to be the same for the same TaskInstance
+    assert call_result1 == call_result2
+    assert call_result1 == "016f5e9e-c4c8-7c30-9eda-d9c646d633ea"
+
+
+def test_lineage_root_run_id_with_conf_af2():
+    conf = {
+        "openlineage": {
+            "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+            "rootParentJobNamespace": "rootns",
+            "rootParentJobName": "rootjob",
+        }
+    }
+    task_instance = mock.MagicMock(
+        dag_run=mock.MagicMock(conf=conf),
+    )
+
+    call_result1 = lineage_root_run_id(task_instance)
+    call_result2 = lineage_root_run_id(task_instance)
+
+    assert call_result1 == call_result2
+    assert call_result1 == "22222222-2222-2222-2222-222222222222"
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_job_name_no_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_id="test_dag",
+        run_id="test_run_id",
+        conf=None,
+    )
+
+    result = lineage_root_job_name(runtime_ti)
+    assert result == "test_dag"
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_job_name_with_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_id="test_dag",
+        run_id="test_run_id",
+        conf={
+            "openlineage": {
+                "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+                "rootParentJobNamespace": "rootns",
+                "rootParentJobName": "rootjob",
+            }
+        },
+    )
+
+    result = lineage_root_job_name(runtime_ti)
+    assert result == "rootjob"
+
+
+def test_lineage_root_job_name_without_conf_af2():
+    task_instance = mock.MagicMock(
+        dag_id="dag_id",
+        dag_run=mock.MagicMock(conf={}),
+    )
+
+    result = lineage_root_job_name(task_instance)
+    assert result == "dag_id"
+
+
+def test_lineage_root_job_name_with_conf_af2():
+    conf = {
+        "openlineage": {
+            "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+            "rootParentJobNamespace": "rootns",
+            "rootParentJobName": "rootjob",
+        }
+    }
+    task_instance = mock.MagicMock(dag_run=mock.MagicMock(conf=conf))
+
+    result = lineage_root_job_name(task_instance)
+    assert result == "rootjob"
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_job_namespace_no_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(task=task, dag_id="test_dag", run_id="test_run_id", conf=None)
+
+    result = lineage_root_job_namespace(runtime_ti)
+    assert result == _DAG_NAMESPACE
+
+
+@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test only for Airflow 3.0+")
+def test_lineage_root_job_namespace_with_conf_af3(create_runtime_ti):
+    from airflow.providers.common.compat.sdk import BaseOperator
+
+    task = BaseOperator(task_id="test_task")
+
+    runtime_ti = create_runtime_ti(
+        task=task,
+        dag_id="test_dag",
+        run_id="test_run_id",
+        conf={
+            "openlineage": {
+                "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+                "rootParentJobNamespace": "rootns",
+                "rootParentJobName": "rootjob",
+            }
+        },
+    )
+
+    result = lineage_root_job_namespace(runtime_ti)
+    assert result == "rootns"
+
+
+def test_lineage_root_job_namespace_without_conf_af2():
+    task_instance = mock.MagicMock(dag_run=mock.MagicMock(conf={}))
+
+    result = lineage_root_job_namespace(task_instance)
+    assert result == _DAG_NAMESPACE
+
+
+def test_lineage_root_job_namespace_with_conf_af2():
+    conf = {
+        "openlineage": {
+            "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+            "rootParentJobNamespace": "rootns",
+            "rootParentJobName": "rootjob",
+        }
+    }
+    task_instance = mock.MagicMock(dag_run=mock.MagicMock(conf=conf))
+
+    result = lineage_root_job_namespace(task_instance)
+    assert result == "rootns"
diff --git a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
index 3bbea4aafac4d..198280426a307 100644
--- a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
+++ b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
@@ -24,6 +24,7 @@
 
 import pendulum
 import pytest
+from openlineage.client.facet_v2 import parent_run
 from uuid6 import uuid7
 
 from airflow import DAG
@@ -31,6 +32,7 @@
 from airflow.models.taskinstance import TaskInstance, TaskInstanceState
 from airflow.providers.common.compat.assets import Asset
 from airflow.providers.common.compat.sdk import BaseOperator, TaskGroup, task, timezone
+from airflow.providers.openlineage.conf import namespace
 from airflow.providers.openlineage.plugins.facets import AirflowDagRunFacet, AirflowJobFacet
 from airflow.providers.openlineage.utils.utils import (
     _MAX_DOC_BYTES,
@@ -40,6 +42,7 @@
     TaskInfo,
     TaskInfoComplete,
     TaskInstanceInfo,
+    _get_openlineage_data_from_dagrun_conf,
     _get_task_groups_details,
     _get_tasks_details,
     _truncate_string_to_byte_size,
@@ -47,11 +50,14 @@
     get_airflow_job_facet,
     get_airflow_state_run_facet,
     get_dag_documentation,
+    get_dag_parent_run_facet,
     get_fully_qualified_class_name,
     get_job_name,
     get_operator_class,
     get_operator_provider_version,
+    get_root_information_from_dagrun_conf,
     get_task_documentation,
+    get_task_parent_run_facet,
     get_user_provided_run_facets,
 )
 from airflow.providers.standard.operators.empty import EmptyOperator
@@ -465,6 +471,233 @@ def test_get_operator_class_mapped_operator():
     assert op_class == MockOperator
 
 
+@pytest.mark.parametrize("dr_conf", (None, {}))
+def test_get_openlineage_data_from_dagrun_conf_none_conf(dr_conf):
+    _dr_conf = None if dr_conf is None else {}
+    assert _get_openlineage_data_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == _dr_conf  # Assert conf is not changed
+
+
+def test_get_openlineage_data_from_dagrun_conf_no_openlineage_key():
+    dr_conf = {"something_else": {"a": 1}}
+    assert _get_openlineage_data_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {"something_else": {"a": 1}}  # Assert conf is not changed
+
+
+def test_get_openlineage_data_from_dagrun_conf_invalid_type():
+    dr_conf = {"openlineage": "not_a_dict"}
+    assert _get_openlineage_data_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {"openlineage": "not_a_dict"}  # Assert conf is not changed
+
+
+def test_get_openlineage_data_from_dagrun_conf_valid_dict():
+    dr_conf = {"openlineage": {"key": "value"}}
+    assert _get_openlineage_data_from_dagrun_conf(dr_conf) == {"key": "value"}
+    assert dr_conf == {"openlineage": {"key": "value"}}  # Assert conf is not changed
+
+
+@pytest.mark.parametrize("dr_conf", (None, {}))
+def test_get_root_information_from_dagrun_conf_no_conf(dr_conf):
+    _dr_conf = None if dr_conf is None else {}
+    assert get_root_information_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == _dr_conf  # Assert conf is not changed
+
+
+def test_get_root_information_from_dagrun_conf_no_openlineage():
+    dr_conf = {"something": "else"}
+    assert get_root_information_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {"something": "else"}  # Assert conf is not changed
+
+
+def test_get_root_information_from_dagrun_conf_openlineage_not_dict():
+    dr_conf = {"openlineage": "my_value"}
+    assert get_root_information_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {"openlineage": "my_value"}  # Assert conf is not changed
+
+
+def test_get_root_information_from_dagrun_conf_missing_keys():
+    dr_conf = {"openlineage": {"rootParentRunId": "id_only"}}
+    assert get_root_information_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {"openlineage": {"rootParentRunId": "id_only"}}  # Assert conf is not changed
+
+
+def test_get_root_information_from_dagrun_conf_invalid_run_id():
+    dr_conf = {
+        "openlineage": {
+            "rootParentRunId": "not_uuid",
+            "rootParentJobNamespace": "ns",
+            "rootParentJobName": "jobX",
+        }
+    }
+    assert get_root_information_from_dagrun_conf(dr_conf) == {}
+    assert dr_conf == {  # Assert conf is not changed
+        "openlineage": {
+            "rootParentRunId": "not_uuid",
+            "rootParentJobNamespace": "ns",
+            "rootParentJobName": "jobX",
+        }
+    }
+
+
+def test_get_root_information_from_dagrun_conf_valid_data():
+    dr_conf = {
+        "openlineage": {
+            "rootParentRunId": "11111111-1111-1111-1111-111111111111",
+            "rootParentJobNamespace": "ns",
+            "rootParentJobName": "jobX",
+        }
+    }
+    expected = {
+        "root_parent_run_id": "11111111-1111-1111-1111-111111111111",
+        "root_parent_job_namespace": "ns",
+        "root_parent_job_name": "jobX",
+    }
+    assert get_root_information_from_dagrun_conf(dr_conf) == expected
+    assert dr_conf == {  # Assert conf is not changed
+        "openlineage": {
+            "rootParentRunId": "11111111-1111-1111-1111-111111111111",
+            "rootParentJobNamespace": "ns",
+            "rootParentJobName": "jobX",
+        }
+    }
+
+
+@pytest.mark.parametrize("dr_conf", (None, {}))
+def test_get_dag_parent_run_facet_no_conf(dr_conf):
+    _dr_conf = None if dr_conf is None else {}
+    assert get_dag_parent_run_facet(dr_conf) == {}
+    assert dr_conf == _dr_conf  # Assert conf is not changed
+
+
+def test_get_dag_parent_run_facet_missing_keys():
+    dr_conf = {"openlineage": {"parentRunId": "11111111-1111-1111-1111-111111111111"}}
+    assert get_dag_parent_run_facet(dr_conf) == {}
+    # Assert conf is not changed
+    assert dr_conf == {"openlineage": {"parentRunId": "11111111-1111-1111-1111-111111111111"}}
+
+
+def test_get_dag_parent_run_facet_valid_no_root():
+    dr_conf = {
+        "openlineage": {
+            "parentRunId": "11111111-1111-1111-1111-111111111111",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+        }
+    }
+
+    result = get_dag_parent_run_facet(dr_conf)
+    parent_facet = result.get("parent")
+
+    assert isinstance(parent_facet, parent_run.ParentRunFacet)
+    assert parent_facet.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.job.namespace == "ns"
+    assert parent_facet.job.name == "jobA"
+    assert parent_facet.root is not None  # parent is used as root, since root is missing
+    assert parent_facet.root.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.root.job.namespace == "ns"
+    assert parent_facet.root.job.name == "jobA"
+
+    assert dr_conf == {  # Assert conf is not changed
+        "openlineage": {
+            "parentRunId": "11111111-1111-1111-1111-111111111111",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+        }
+    }
+
+
+def test_get_dag_parent_run_facet_invalid_uuid():
+    dr_conf = {
+        "openlineage": {
+            "parentRunId": "not_uuid",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+        }
+    }
+
+    result = get_dag_parent_run_facet(dr_conf)
+    assert result == {}
+    assert dr_conf == {  # Assert conf is not changed
+        "openlineage": {
+            "parentRunId": "not_uuid",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+        }
+    }
+
+
+def test_get_dag_parent_run_facet_valid_with_root():
+    dr_conf = {
+        "openlineage": {
+            "parentRunId": "11111111-1111-1111-1111-111111111111",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+            "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+            "rootParentJobNamespace": "rootns",
+            "rootParentJobName": "rootjob",
+        }
+    }
+
+    result = get_dag_parent_run_facet(dr_conf)
+    parent_facet = result.get("parent")
+
+    assert isinstance(parent_facet, parent_run.ParentRunFacet)
+    assert parent_facet.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.job.namespace == "ns"
+    assert parent_facet.job.name == "jobA"
+    assert parent_facet.root is not None
+    assert parent_facet.root.run.runId == "22222222-2222-2222-2222-222222222222"
+    assert parent_facet.root.job.namespace == "rootns"
+    assert parent_facet.root.job.name == "rootjob"
+
+    assert dr_conf == {  # Assert conf is not changed
+        "openlineage": {
+            "parentRunId": "11111111-1111-1111-1111-111111111111",
+            "parentJobNamespace": "ns",
+            "parentJobName": "jobA",
+            "rootParentRunId": "22222222-2222-2222-2222-222222222222",
+            "rootParentJobNamespace": "rootns",
+            "rootParentJobName": "rootjob",
+        }
+    }
+
+
+def test_get_task_parent_run_facet_defaults():
+    result = get_task_parent_run_facet(
+        parent_run_id="11111111-1111-1111-1111-111111111111",
+        parent_job_name="jobA",
+    )
+    parent_facet = result.get("parent")
+
+    assert isinstance(parent_facet, parent_run.ParentRunFacet)
+    assert parent_facet.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.job.namespace == namespace()
+    assert parent_facet.job.name == "jobA"
+    assert parent_facet.root.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.root.job.namespace == namespace()
+    assert parent_facet.root.job.name == "jobA"
+
+
+def test_get_task_parent_run_facet_custom_root_values():
+    result = get_task_parent_run_facet(
+        parent_run_id="11111111-1111-1111-1111-111111111111",
+        parent_job_name="jobA",
+        parent_job_namespace="ns",
+        root_parent_run_id="22222222-2222-2222-2222-222222222222",
+        root_parent_job_name="rjob",
+        root_parent_job_namespace="rns",
+    )
+
+    parent_facet = result.get("parent")
+    assert isinstance(parent_facet, parent_run.ParentRunFacet)
+    assert parent_facet.run.runId == "11111111-1111-1111-1111-111111111111"
+    assert parent_facet.job.namespace == "ns"
+    assert parent_facet.job.name == "jobA"
+    assert parent_facet.root.run.runId == "22222222-2222-2222-2222-222222222222"
+    assert parent_facet.root.job.namespace == "rns"
+    assert parent_facet.root.job.name == "rjob"
+
+
 def test_get_tasks_details():
     class TestMappedOperator(BaseOperator):
         def __init__(self, value, **kwargs):
