Allow filtering hook-collected assets in OpenLineage emission policy

providers/openlineage/docs/emission_policy.rst

@@ -144,13 +144,39 @@ pattern.
        inputs/outputs. Task events only.
        **No effect when ``extract_operator_metadata`` is ``false``** — the entire extraction
        pipeline (including hook lineage) is skipped.
+   * - ``exclude_hook_lineage_assets``
+     - ``[]``
+     - List of regex patterns; hook-collected assets whose URI matches any of them are dropped.
+       Task events only. **No effect when ``hook_lineage`` is ``false``** — hook lineage is not
+       collected at all.
+   * - ``exclude_hook_lineage_hooks``
+     - ``[]``
+     - List of regex patterns; hook-collected assets reported by a hook whose **fully-qualified**
+       class name (``module.ClassName``, as with the ``operator`` scope key) matches any of them are
+       dropped. Task events only. **No effect when ``hook_lineage`` is ``false``**.
    * - ``include_full_task_info``
      - ``false``
      - Whether to include the full serialized operator state in ``AirflowRunFacet``. When
        ``false``, only a curated subset of task attributes is sent. When ``true``, all
        serializable task parameters are included, which may significantly increase event size.
        Task events only.
 
+Both exclude lists are always matched with ``re.fullmatch``, independent of the rule's
+``match_mode`` (which governs ``scope`` only). Unlike the boolean controls they are **replaced**
+rather than merged across tiers: the most specific matching rule that sets a list wins outright,
+so a task-scoped rule can narrow — or clear, with ``[]`` — a broader global exclusion. Both
+filters apply to hook-collected assets only; SQL-based hook lineage is unaffected.
+
+.. note::
+
+    **Exclusion happens after the collector cap.** These filters run when OpenLineage reads the
+    hook lineage collector, which is after
+    :ref:`[lineage] max_assets_per_collector<config:lineage__max_assets_per_collector>` has
+    already limited what the collector retained. Excluding assets reduces what gets reported, but
+    it does not create room for assets the cap already discarded — a task that writes more objects
+    than the cap allows never collected the later ones in the first place. Reducing the number of
+    assets a hook registers requires changes in the hooks themselves.
+
 ``locked`` (top-level, default ``false``) is an admin-only floor lock: when ``true``, the
 control fields carried by this rule's ``controls`` dict cannot be overridden by per-Dag /
 per-task authoring flags (see :ref:`emission_policy_authoring:openlineage`). The rule still
@@ -193,6 +219,7 @@ suppresses DAG-level events for X while leaving task events enabled.
       {"scope": {"dag_id": "expensive_dag"}, "controls": {"extract_operator_metadata": false}},
       {"scope": {"dag_id": "expensive_dag", "task_id": "send_report"}, "controls": {"extract_operator_metadata": true}},
       {"scope": {"dag_id": "my_dag", "task_id": "sensitive_task"}, "controls": {"hook_lineage": false}},
+      {"scope": {"dag_id": "chunked_dag", "task_id": "upload"}, "controls": {"exclude_hook_lineage_assets": ["s3://my-bucket/staging/part_.*"]}},
       {"scope": {"dag_id": "reporting_dag"}, "controls": {"emit_dag_events": false}},
       {"scope": {"dag_id": "full_control_dag"}, "controls": {"emit": false}},
       {"scope": {"dag_id": "full_control_dag", "task_id": "critical_task"}, "controls": {"emit": true}},
@@ -212,6 +239,8 @@ In the example above:
   with a more-specific task-scoped rule and extracts normally.
 - ``sensitive_task`` in ``my_dag`` runs its extractor normally but skips the
   ``HookLineageCollector`` fallback.
+- ``upload`` in ``chunked_dag`` keeps hook lineage but drops the staging part objects it writes,
+  so only the assets that are not staging parts are reported.
 - DAG-level events for ``reporting_dag`` are suppressed; task events are still emitted.
 - Both task events and DAG-level events for ``full_control_dag`` are suppressed — except
   ``critical_task``, which re-enables task event emission via a task-scoped rule.
@@ -253,6 +282,11 @@ wins **unless** the matching conf rule is marked with ``"locked": true``.
     # Suppress DAG-run events but keep task events
     extend_global_openlineage_emission_policy(dag, emit_dag_events=False)
 
+    # Keep hook lineage, but drop the per-chunk objects a loop uploads
+    extend_global_openlineage_emission_policy(
+        extract, exclude_hook_lineage_assets=["s3://my-bucket/staging/part_.*"]
+    )
+
 Key semantics:
 
 - **Resolution priority**: authoring flags > unlocked conf rules > built-in defaults.

providers/openlineage/src/airflow/providers/openlineage/api/emission_policy.py

@@ -91,12 +91,16 @@
     EMIT,
     EMIT_DAG_EVENTS,
     EMIT_TASK_EVENTS,
+    EXCLUDE_HOOK_LINEAGE_ASSETS,
+    EXCLUDE_HOOK_LINEAGE_HOOKS,
     EXTRACT_OPERATOR_METADATA,
     HOOK_LINEAGE,
     INCLUDE_FULL_TASK_INFO,
     INCLUDE_SOURCE_CODE,
     OL_EMISSION_POLICY_PARAM,
+    ControlValue,
     _merge_param,
+    find_invalid_pattern,
 )
 
 if TYPE_CHECKING:
@@ -120,6 +124,8 @@ def extend_global_openlineage_emission_policy(
     extract_operator_metadata: bool | None = None,
     include_source_code: bool | None = None,
     hook_lineage: bool | None = None,
+    exclude_hook_lineage_assets: list[str] | None = None,
+    exclude_hook_lineage_hooks: list[str] | None = None,
     include_full_task_info: bool | None = None,
 ) -> T:
     """
@@ -132,8 +138,9 @@ def extend_global_openlineage_emission_policy(
     When called on a **DAG**, flags are applied as follows:
 
     - Task-relevant flags (``emit``, ``emit_task_events``, ``extract_operator_metadata``,
-      ``include_source_code``, ``hook_lineage``, ``include_full_task_info``) are
-      **propagated to all tasks** in the Dag at the time of the call.
+      ``include_source_code``, ``hook_lineage``, ``exclude_hook_lineage_assets``,
+      ``exclude_hook_lineage_hooks``, ``include_full_task_info``) are **propagated to all
+      tasks** in the Dag at the time of the call.
     - DAG-run-level flags (``emit``, ``emit_dag_events``) are stored on the Dag itself.
     - ``emit_dag_events`` is meaningless on a task and logs a warning if provided.
 
@@ -179,7 +186,13 @@ def extend_global_openlineage_emission_policy(
     :param extract_operator_metadata: Whether to run operator-specific extractor-based metadata collection.
     :param include_source_code: Whether to include operator source code in Python/Bash operator events.
     :param hook_lineage: Whether to use ``HookLineageCollector`` as a fallback.
+    :param exclude_hook_lineage_assets: Regex patterns; hook-collected assets whose URI matches
+        any of them are dropped.  Has no effect when ``hook_lineage`` is disabled.
+    :param exclude_hook_lineage_hooks: Regex patterns; hook-collected assets reported by a hook
+        whose fully-qualified class name (``module.ClassName``) matches any of them are dropped.
+        Has no effect when ``hook_lineage`` is disabled.
     :param include_full_task_info: Whether to include the full serialized operator state.
+    :raises ValueError: If either exclusion argument is not a list of valid regex patterns.
     :return: The same *obj* — allows use as a decorator or in chained calls.
     """
     if isinstance(obj, XComArg):
@@ -191,6 +204,8 @@ def extend_global_openlineage_emission_policy(
             extract_operator_metadata=extract_operator_metadata,
             include_source_code=include_source_code,
             hook_lineage=hook_lineage,
+            exclude_hook_lineage_assets=exclude_hook_lineage_assets,
+            exclude_hook_lineage_hooks=exclude_hook_lineage_hooks,
             include_full_task_info=include_full_task_info,
         )
         return obj
@@ -206,7 +221,20 @@ def extend_global_openlineage_emission_policy(
             "configuration with a global rule ('scope': {}) instead."
         )
 
-    provided: dict[str, bool] = {
+    # Validate eagerly: a bad pattern stored here would otherwise surface much later as an
+    # re.error inside hook lineage extraction, which the extractor manager swallows — the Dag
+    # would silently lose all lineage instead of reporting the typo.
+    for field, patterns in (
+        (EXCLUDE_HOOK_LINEAGE_ASSETS, exclude_hook_lineage_assets),
+        (EXCLUDE_HOOK_LINEAGE_HOOKS, exclude_hook_lineage_hooks),
+    ):
+        if patterns is None:
+            continue
+        invalid = find_invalid_pattern(field, patterns)
+        if invalid:
+            raise ValueError(f"extend_global_openlineage_emission_policy(): {invalid}")
+
+    provided: dict[str, ControlValue] = {
         k: v
         for k, v in {
             EMIT: emit,
@@ -215,6 +243,8 @@ def extend_global_openlineage_emission_policy(
             EXTRACT_OPERATOR_METADATA: extract_operator_metadata,
             INCLUDE_SOURCE_CODE: include_source_code,
             HOOK_LINEAGE: hook_lineage,
+            EXCLUDE_HOOK_LINEAGE_ASSETS: exclude_hook_lineage_assets,
+            EXCLUDE_HOOK_LINEAGE_HOOKS: exclude_hook_lineage_hooks,
             INCLUDE_FULL_TASK_INFO: include_full_task_info,
         }.items()
         if v is not None

providers/openlineage/src/airflow/providers/openlineage/extractors/manager.py

@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import re
 from collections.abc import Iterator
 from typing import TYPE_CHECKING
 
@@ -137,7 +138,7 @@ def extract_metadata(
                 # If no inputs and outputs are present - check Hook Lineage if enabled
                 if (not task_metadata.inputs) and (not task_metadata.outputs):
                     if controls.hook_lineage:
-                        hook_lineage = self.get_hook_lineage(task_instance, task_instance_state)
+                        hook_lineage = self.get_hook_lineage(task_instance, task_instance_state, controls)
                         if hook_lineage is not None:
                             task_metadata = task_metadata.merge(hook_lineage)
                         else:  # Last resort - check manual annotations
@@ -166,7 +167,7 @@ def extract_metadata(
                 # internally. An uncaught exception here would propagate up to the listener's
                 # @print_warning decorator, silently suppressing the task-level event.
                 try:
-                    hook_lineage = self.get_hook_lineage(task_instance, task_instance_state)
+                    hook_lineage = self.get_hook_lineage(task_instance, task_instance_state, controls)
                 except Exception as e:
                     self.log.warning(
                         "Failed to extract OpenLineage hook lineage %s: %s. Task event will be emitted without lineage.",
@@ -243,10 +244,39 @@ def extract_inlets_and_outlets(
                 task_metadata.outputs.append(ol)
                 seen.add((ol.namespace, ol.name))
 
+    def _is_excluded_asset(
+        self,
+        asset_info,
+        exclude_assets: tuple[str, ...],
+        exclude_hooks: tuple[str, ...],
+    ) -> bool:
+        """
+        Return ``True`` if *asset_info* matches any asset-URI or hook-class exclusion pattern.
+
+        Hooks are matched on their fully-qualified class name, mirroring how the ``operator``
+        scope key identifies operators, so two identically named classes from different modules
+        stay distinguishable.
+        """
+        if exclude_assets:
+            uri = getattr(asset_info.asset, "uri", None)
+            if uri and any(re.fullmatch(p, uri) for p in exclude_assets):
+                self.log.debug("Excluding hook-collected asset %r by asset pattern.", uri)
+                return True
+
+        if exclude_hooks:
+            hook_type = type(asset_info.context)
+            hook_name = f"{hook_type.__module__}.{hook_type.__name__}"
+            if any(re.fullmatch(p, hook_name) for p in exclude_hooks):
+                self.log.debug("Excluding hook-collected asset reported by hook %r.", hook_name)
+                return True
+
+        return False
+
     def get_hook_lineage(
         self,
         task_instance=None,
         task_instance_state: TaskInstanceState | None = None,
+        controls: EmissionPolicy | None = None,
     ) -> OperatorLineage | None:
         """
         Extract lineage from the Hook Lineage Collector.
@@ -259,8 +289,13 @@ def get_hook_lineage(
           When ``task_instance`` is provided, each extra is parsed and separate per-query
           OpenLineage events are emitted.
 
+        Assets matching the ``exclude_hook_lineage_assets`` / ``exclude_hook_lineage_hooks``
+        patterns of *controls* are dropped. SQL-based lineage is not filtered.
+
         Returns ``None`` when nothing was collected.
         """
+        if controls is None:
+            controls = EmissionPolicy.defaults()
         try:
             from airflow.providers.common.compat.lineage.hook import get_hook_lineage_collector
             from airflow.providers.common.sql.hooks.lineage import SqlJobHookLineageExtra
@@ -276,16 +311,22 @@ def get_hook_lineage(
         self.log.debug("OpenLineage will extract lineage from Hook Lineage Collector.")
         collected = collector.collected_assets
 
-        # Asset-based inputs/outputs - keep only assets that can be translated to OL datasets
+        exclude_assets = controls.exclude_hook_lineage_assets
+        exclude_hooks = controls.exclude_hook_lineage_hooks
+
+        # Asset-based inputs/outputs - keep only assets that are not excluded by policy and
+        # that can be translated to OL datasets
         inputs = [
             asset
             for asset_info in collected.inputs
-            if (asset := translate_airflow_asset(asset_info.asset, asset_info.context)) is not None
+            if not self._is_excluded_asset(asset_info, exclude_assets, exclude_hooks)
+            and (asset := translate_airflow_asset(asset_info.asset, asset_info.context)) is not None
         ]
         outputs = [
             asset
             for asset_info in collected.outputs
-            if (asset := translate_airflow_asset(asset_info.asset, asset_info.context)) is not None
+            if not self._is_excluded_asset(asset_info, exclude_assets, exclude_hooks)
+            and (asset := translate_airflow_asset(asset_info.asset, asset_info.context)) is not None
         ]
 
         # SQL-based lineage - keep only SQL extra with query_text or job_id.