mobuchowski commented on code in PR #75303:
URL: https://github.com/apache/airflow/pull/75303#discussion_r4350417328
##########
providers/standard/src/airflow/providers/standard/utils/openlineage.py:
##########
@@ -80,9 +80,19 @@ def _get_openlineage_parent_info(ti: TaskInstance |
RuntimeTI) -> dict[str, str]
"rootParentJobNamespace": lineage_root_job_namespace(ti),
}
+ try:
+ from airflow.providers.openlineage.plugins.macros import
lineage_job_type, lineage_root_job_type
+ except ImportError: # Job type macros are missing in older versions of
the OpenLineage provider
+ return parent_info
+
+ parent_info["parentJobType"] = lineage_job_type()
+ if (root_job_type := lineage_root_job_type(ti)) is not None:
+ parent_info["rootParentJobType"] = root_job_type
+ return parent_info
+
def _inject_openlineage_parent_info_to_dagrun_conf(
- dr_conf: dict | None, ol_parent_info: dict[str, str]
+ dr_conf: dict | None, ol_parent_info: dict[str, Any]
Review Comment:
`_inject_openlineage_parent_info_to_dagrun_conf` decides whether to inject
by checking the user's existing `openlineage` conf for any key in
`forbidden_keys`. If it finds none, it merges with `{**ol_parent_info,
**current_ol_dr_conf}`, so the user's keys win key by key and the injected
values fill in the rest.
The new `parentJobType` / `rootParentJobType` keys aren't in
`forbidden_keys`. Also, the existing entries `rootJobName` / `rootJobNamespace`
are misspelled: the real keys are `rootParentJobName` /
`rootParentJobNamespace`. That typo predates this PR, but it now affects job
types too. Two examples:
- A user triggers with `conf={"openlineage": {"rootParentJobName": "x",
"rootParentJobNamespace": "y"}}`. No forbidden key matches, so injection goes
ahead. The child Dag gets the user's root name and namespace together with the
injected `rootParentRunId` and, new in this PR, the injected
`rootParentJobType`. That's the TriggerDagRunOperator Dag's type attached to
job `y/x`.
- A user sets only `rootParentJobType` (or `parentJobType`). Their type is
attached to the injected run ids of a different job.
Both are uncommon inputs. Still, the description says "a type is never
paired with a different job", and this path gets around it. Suggested fix: add
both job type keys to `forbidden_keys`, and fix the two root key names while
you're here:
```python
forbidden_keys = (
"parentRunId",
"parentJobName",
"parentJobNamespace",
"parentJobType",
"rootParentRunId",
"rootParentJobName",
"rootParentJobNamespace",
"rootParentJobType",
)
```
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -55,6 +55,20 @@ def lineage_job_name(task_instance: TaskInstance):
return get_job_name(task_instance)
+def lineage_job_type() -> dict[str, str]:
Review Comment:
The docs list these next to the other `lineage_*` macros and say they "can
be forwarded to a child job together with the parent information". Every
example in that section uses the macros inside a string template such as `"{{
macros.OpenLineageProviderPlugin.lineage_job_namespace() }}"`. The other macros
return strings, so that works for them.
These two return a Python dict, or `None`. Unless
`render_template_as_native_obj=True` is set, Airflow renders templates to text,
so you get the Python repr instead:
- `"{{ macros.OpenLineageProviderPlugin.lineage_job_type() }}"` renders to
`"{'processingType': 'BATCH', 'integration': 'AIRFLOW', 'jobType': 'TASK'}"`.
That has single quotes, so it isn't valid JSON for a Spark conf or an HTTP
payload.
- When the root type is unknown, `lineage_root_job_type(task_instance)`
renders to the literal string `"None"`.
- Someone who passes these by hand in a `TriggerDagRunOperator` conf gets a
string where a dict is expected. `_get_job_type_from_ol_data` then logs
"Invalid OpenLineage parentJobType" and drops it.
The automatic injection path calls the macros directly from Python, so it
isn't affected. Only the template use the docs suggest is. Could you either
note in `macros.rst` that these return dicts and need native rendering (or
should be called from Python), or add string forms, for example a
`|tojson`-friendly example or `lineage_job_type_json()`?
##########
providers/standard/tests/unit/standard/utils/test_openlineage.py:
##########
@@ -137,18 +156,39 @@ def _mock_version(package):
mock_root_run_id.assert_called_once_with(ti)
mock_root_job_name.assert_called_once_with(ti)
mock_root_job_namespace.assert_called_once_with(ti)
+ mock_job_type.assert_called_once_with()
+ mock_root_job_type.assert_called_once_with(ti)
- # Verify result structure
- assert isinstance(result, dict)
assert result == expected_values
- assert set(result.keys()) == {
- "parentRunId",
- "parentJobName",
- "parentJobNamespace",
- "rootParentRunId",
- "rootParentJobName",
- "rootParentJobNamespace",
- }
+
+
+def test_get_openlineage_parent_info_without_job_type_macros(monkeypatch):
Review Comment:
Nit: the repo's testing guidelines ask for `spec`/`autospec` on mocks and
prefer `@mock.patch` decorators over `with patch(...)`. The six
`patch(f"{OL_MACROS_PATH}.lineage_*", return_value=...)` calls here use
neither. The updated test above already added `autospec=True` for the new
patches, so matching that here would be consistent.
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -430,6 +458,7 @@ def get_task_parent_run_facet(
root_parent_run_id=root_parent_run_id,
root_parent_job_namespace=root_parent_job_namespace,
root_parent_job_name=root_parent_job_name,
+ root_parent_job_type=root_parent_job_type,
Review Comment:
Nothing is broken today. For task events, `get_task_parent_run_facet` never
sets `parent_job_type`. The parent of a task is always its Dag, which we know
is `BATCH/AIRFLOW/DAG`. When the Dag run has no OpenLineage conf, the root (the
Dag itself) gets no type either.
Meanwhile `lineage_root_job_type(ti)` returns `{BATCH, AIRFLOW, DAG}` for
that same root, and TriggerDagRunOperator injects it. So in a child Dag's
events the root job has `jobType: DAG`, but the same root job has no `jobType`
facet in its own Dag's task events. Is leaving that out deliberate for this PR,
or worth passing the Dag type as `parent_job_type` here?
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -441,7 +470,30 @@ def _get_openlineage_data_from_dagrun_conf(dr_conf: dict |
None) -> dict:
return ol_data if isinstance(ol_data, dict) else {}
-def get_root_information_from_dagrun_conf(dr_conf: dict | None) -> dict[str,
str]:
+def _get_job_type_from_ol_data(ol_data: dict, key: str) -> dict[str, str] |
None:
+ """Return a copy of the job type facet fields stored under ``key`` if they
are valid, otherwise None."""
+ job_type = ol_data.get(key)
+ if job_type is None:
+ return None
+ expected_fields = {"processingType", "integration", "jobType"}
+ # The spec allows partial job type, but only complete information is
supported for now.
+ if (
+ not isinstance(job_type, dict)
+ or job_type.keys() != expected_fields
Review Comment:
This check drops the whole job type if the dict has any key beyond
`processingType` / `integration` / `jobType`, and the `unknown-field` test case
shows that's on purpose. Nothing in this PR sends extra keys, but a producer
outside Airflow might forward a serialized `JobTypeJobFacet`, which includes
`_producer` / `_schemaURL`. That type would then be thrown away with a warning,
even though the three fields we need are all there. Would it be better to
require the three fields and ignore the rest?
```python
if not isinstance(job_type, dict) or not all(
isinstance(job_type.get(f), str) and job_type[f] for f in expected_fields
):
...
return {f: job_type[f] for f in expected_fields}
```
If strict matching is deliberate (for example, to reject anything that isn't
the macro's own output), feel free to ignore this.
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/macros.py:
##########
@@ -154,6 +189,10 @@ def _get_ol_root_id(id_key: str, task_instance:
TaskInstance) -> str | None:
return None
+def _build_airflow_job_type(job_type: Literal["DAG", "TASK"]) -> dict[str,
str]:
Review Comment:
Optional: `_build_airflow_job_type` repeats the `BATCH` / `AIRFLOW` values
that `OpenLineageAdapter` uses when it builds the real `jobType` facet
(`adapter.py:651`). `_JOB_TYPE_DAG` / `_JOB_TYPE_TASK` already exist in
`adapter.py:75`. This module already imports from `adapter`, so pulling in the
constants keeps the macros in sync with the emitted events:
```python
from airflow.providers.openlineage.plugins.adapter import _JOB_TYPE_DAG,
_JOB_TYPE_TASK, OpenLineageAdapter
...
return _build_airflow_job_type(_JOB_TYPE_TASK)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
