This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new d9ca6241562 Match file datasets in exclude_datasets without a doubled
slash (#75921)
d9ca6241562 is described below
commit d9ca6241562a151730406b4c6015087e1f3a24ed
Author: Ramit Kataria <[email protected]>
AuthorDate: Thu Oct 29 01:59:48 2026 -0700
Match file datasets in exclude_datasets without a doubled slash (#75921)
File and other path-based OpenLineage datasets have names that start with
"/",
so joining namespace and name with another "/" made file://host1/x match
only
as file://host1//x, and the pattern a user would naturally write never
matched.
The feature is unreleased, so the format can still change without breaking
existing policies.
The docs also lacked the way to report a single declared asset instead of
every object a hook collects, which was the workaround asked for in #63371,
and did not say that the extras cap limits SQL hook lineage the same way.
---
providers/openlineage/docs/emission_policy.rst | 13 ++++++++++---
.../airflow/providers/openlineage/api/emission_policy.py | 3 ++-
.../airflow/providers/openlineage/utils/emission_policy.py | 8 ++++++--
.../tests/unit/openlineage/utils/test_emission_policy.py | 7 +++++++
4 files changed, 25 insertions(+), 6 deletions(-)
diff --git a/providers/openlineage/docs/emission_policy.rst
b/providers/openlineage/docs/emission_policy.rst
index 06488dfcafd..d8c331e3493 100644
--- a/providers/openlineage/docs/emission_policy.rst
+++ b/providers/openlineage/docs/emission_policy.rst
@@ -282,7 +282,10 @@ example the per-chunk objects of a multipart upload that
hook lineage reports on
- Patterns are always matched with ``re.fullmatch``, independent of
``match_mode``, against
``<namespace>/<name>`` of the OpenLineage dataset:
``s3://my-bucket/staging/part_1`` for a
- dataset with namespace ``s3://my-bucket`` and name ``staging/part_1``.
Airflow assets are
+ dataset with namespace ``s3://my-bucket`` and name ``staging/part_1``. Names
that already start
+ with ``/``, such as file paths, get no extra ``/``:
``file://host1/staging/part_1`` for
+ namespace ``file://host1`` and name ``/staging/part_1``, and
``file/tmp/part_1`` for a local
+ file with namespace ``file``. Airflow assets are
matched after translation to OpenLineage, so assets that cannot be
translated are never
reported, whatever the patterns.
- They apply to every source of task lineage: extractor results,
hook-collected assets,
@@ -294,7 +297,10 @@ example the per-chunk objects of a multipart upload that
hook lineage reports on
that case, since the event exists only to carry the datasets.
- Excluding datasets never changes which source is used. If an extractor
returns datasets that
are all excluded, hook lineage is not consulted as a fallback; if hook
lineage collected assets
- that are all excluded, manually declared ``inlets`` / ``outlets`` are not
used either.
+ that are all excluded, manually declared ``inlets`` / ``outlets`` are not
used either. To
+ report your own assets instead of the ones a hook collects, for example one
asset for an upload
+ prefix instead of one per uploaded part, set ``hook_lineage: false`` for the
task and declare the
+ asset in ``outlets``.
- A rule whose ``scope`` has a ``hook`` key only applies to datasets reported
by that hook, and
only to hook lineage and SQL hook lineage (the other sources have no
reporting hook). Such a
rule may only carry ``exclude_datasets`` or ``hook_lineage``; with a
``hook`` scope,
@@ -343,7 +349,8 @@ In the example above:
OpenLineage reads the hook lineage collector, after
:ref:`[lineage]
max_assets_per_collector<config:lineage__max_assets_per_collector>` has
limited what the collector kept. Excluding assets reduces what gets
reported, but it does not
- make room for assets the cap already discarded.
+ make room for assets the cap already discarded. The same applies to SQL
hook lineage queries
+ and :ref:`[lineage]
max_extras_per_collector<config:lineage__max_extras_per_collector>`.
.. note::
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/api/emission_policy.py
b/providers/openlineage/src/airflow/providers/openlineage/api/emission_policy.py
index 652b89d4e3f..50d3d9489e6 100644
---
a/providers/openlineage/src/airflow/providers/openlineage/api/emission_policy.py
+++
b/providers/openlineage/src/airflow/providers/openlineage/api/emission_policy.py
@@ -189,7 +189,8 @@ def extend_global_openlineage_emission_policy(
:param hook_lineage: Whether to use ``HookLineageCollector`` as a fallback.
:param include_full_task_info: Whether to include the full serialized
operator state.
:param exclude_datasets: Regex patterns matched with ``re.fullmatch``
against
- ``"<namespace>/<name>"`` of each OpenLineage dataset of the task;
matching datasets are
+ ``"<namespace>/<name>"`` of each OpenLineage dataset of the task
(without a doubled ``/``
+ when the name starts with one); matching datasets are
dropped. Replaces, instead of extending, any list from the
``emission_policy`` config,
including lists set for specific hooks.
:raises ValueError: If ``exclude_datasets`` is not a list of valid regex
patterns.
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/utils/emission_policy.py
b/providers/openlineage/src/airflow/providers/openlineage/utils/emission_policy.py
index 89a93b94035..39f007cac58 100644
---
a/providers/openlineage/src/airflow/providers/openlineage/utils/emission_policy.py
+++
b/providers/openlineage/src/airflow/providers/openlineage/utils/emission_policy.py
@@ -80,7 +80,8 @@ Control flag keys (all optional inside ``controls``; the dict
must be non-empty)
in the ``AirflowRunFacet``. When ``false`` (default), only a curated subset
of
task attributes is sent. Only meaningful for task events. Default:
``false``.
- ``exclude_datasets``: list of ``re.fullmatch`` patterns (independent of
``match_mode``)
- matched against ``"<namespace>/<name>"`` of each OpenLineage dataset;
matching datasets
+ matched against ``"<namespace>/<name>"`` of each OpenLineage dataset
(without a doubled ``/``
+ when the name starts with one); matching datasets
are dropped from task events and from the SQL hook lineage events of the
task. Applies
to extractor, hook-collected and manually declared lineage. Default: ``[]``.
@@ -326,7 +327,10 @@ def get_hook_class_name(hook: object | None) -> str | None:
def _get_dataset_identity(dataset: Dataset) -> str:
"""Return the ``"<namespace>/<name>"`` string that ``exclude_datasets``
patterns match."""
- return f"{dataset.namespace}/{dataset.name}"
+ # File and other path-based datasets have names starting with "/"; joining
them with another
+ # "/" would make ``file://host/path`` match as ``file://host//path``.
+ separator = "" if dataset.name.startswith("/") else "/"
+ return f"{dataset.namespace}{separator}{dataset.name}"
def _get_rule_exclude_patterns(rule: Rule) -> tuple[str, ...] | None:
diff --git
a/providers/openlineage/tests/unit/openlineage/utils/test_emission_policy.py
b/providers/openlineage/tests/unit/openlineage/utils/test_emission_policy.py
index 4784bcede1d..d64ceb3a9fc 100644
--- a/providers/openlineage/tests/unit/openlineage/utils/test_emission_policy.py
+++ b/providers/openlineage/tests/unit/openlineage/utils/test_emission_policy.py
@@ -2565,6 +2565,13 @@ class TestExcludeDatasetsResolution:
pytest.param(
["gs://bucket/staging/.*"], Dataset("gs://bucket",
"final/out"), False, id="no-match"
),
+ pytest.param(
+ ["file://host1/staging/.*"],
+ Dataset("file://host1", "/staging/part_1"),
+ True,
+ id="absolute-name",
+ ),
+ pytest.param(["file/tmp/.*"], Dataset("file", "/tmp/part_1"),
True, id="absolute-name-no-host"),
],
)
def test_is_excluded_matches_namespace_and_name(self, patterns, dataset,
excluded):
