feast-dev · ntkathole · Aug 5, 2026 · Aug 9, 2026 · Aug 10, 2026
@@ -25,6 +25,11 @@
 _ol_processor: Optional[Any] = None
 
 
+def get_ol_processor() -> Optional[Any]:
+    """Return the global OpenLineage processor, if initialized."""
+    return _ol_processor
+
+
 def register_all_routes(app: FastAPI, grpc_handler, server=None, store=None):
     app.include_router(get_entity_router(grpc_handler))
     app.include_router(get_data_source_router(grpc_handler))
@@ -133,19 +138,86 @@ def _register_openlineage_consumer(app: FastAPI, feast_store):
 
         # Wire the local processor into Feast's own OL emitter so Feast events
         # are also stored in the consumer DB automatically.
+        # The emitter is lazy-initialized, so it may be None at startup.
+        # Two-pronged approach:
+        #   1. If already initialized, wire now.
+        #   2. Store processor globally so _init_openlineage_emitter() can
+        #      pick it up when the emitter is lazily created later.
         try:
-            if feast_store and hasattr(feast_store, "_openlineage_emitter"):
-                emitter = feast_store._openlineage_emitter
-                if emitter and hasattr(emitter, "_client") and emitter._client:
-                    emitter._client.set_local_processor(processor)
-                    logger.info("Feast OL emitter wired to local consumer processor")
+            emitter = getattr(feast_store, "_openlineage_emitter", None)
+            if emitter and hasattr(emitter, "_client") and emitter._client:
+                emitter._client.set_local_processor(processor)
+                logger.info(
+                    "Feast OL emitter wired to local consumer processor (eager)"
+                )
         except Exception as wire_err:
             logger.debug(f"Could not wire emitter to local processor: {wire_err}")
 
+        def _build_allowed_namespaces_fn(fs):
+            """Build a callback that derives OL namespace access from Feast RBAC.
+
+            Uses ``permitted_resources`` with ``DESCRIBE`` on all known projects.
+            The projects the current user may describe become the allowed OL
+            namespaces (mapped through ``resolve_namespace``).
+
+            External producer namespaces (e.g. ``spark://ml-team``,
+            ``airflow://prod-cluster``) are included when
+            ``consumer.namespace_mapping`` maps them to a Feast project the
+            user is allowed to DESCRIBE.  This is the only purpose of
+            ``namespace_mapping`` — it is a **read-side RBAC bridge**, not a
+            routing or rewrite mechanism.  Ingest stores events as-is;
+            producers own their namespace.
+            """
+
+            def _get_allowed():
+                try:
+                    from feast.openlineage.identity import resolve_namespace
+                    from feast.permissions.action import AuthzedAction
+                    from feast.permissions.security_manager import (
+                        get_security_manager,
+                        permitted_resources,
+                    )
+
+                    sm = get_security_manager()
+                    if sm is None:
+                        return None
+
+                    all_projects = fs.registry.list_projects(allow_cache=True)
+                    if not all_projects:
+                        return None
+
+                    allowed_projects = permitted_resources(
+                        all_projects, AuthzedAction.DESCRIBE
+                    )
+
+                    ol_ns_config = getattr(ol_config, "namespace", "feast")
+                    allowed_project_names = {p.name for p in allowed_projects}
+
+                    namespaces = set()
+                    for p in allowed_projects:
+                        namespaces.add(resolve_namespace(ol_ns_config, p.name))
+
+                    # Include external namespaces whose namespace_mapping
+                    # target is a project the user can DESCRIBE.
+                    consumer_cfg = getattr(ol_config, "consumer", None)
+                    ns_map = getattr(consumer_cfg, "namespace_mapping", None) or {}
+                    for ext_ns, mapped_project in ns_map.items():
+                        if mapped_project in allowed_project_names:
+                            namespaces.add(ext_ns)
+
+                    return list(namespaces) if namespaces else None
+                except Exception:
+                    return None
+
+            return _get_allowed
+
+        get_allowed_namespaces = _build_allowed_namespaces_fn(feast_store)
+
         consumer_router = get_consumer_router(
             config=ol_config,
             store=ol_store,
             processor=processor,
+            get_allowed_namespaces=get_allowed_namespaces,
         )
 
         app.include_router(consumer_router)

@@ -86,6 +86,7 @@ def get_object_relationships_path(
             "featureView",
             "featureService",
             "feature",
+            "savedDataset",
         ]
         if object_type not in valid_types:
             raise ValueError(
@@ -175,6 +176,7 @@ def get_complete_registry_data(
                 "featureServices": project_resources.get("featureServices", []),
                 "features": project_resources.get("features", []),
                 "labels": project_resources.get("labels", []),
+                "savedDatasets": project_resources.get("savedDatasets", []),
             },
             "relationships": lineage_response.get("relationships", []),
             "indirectRelationships": lineage_response.get("indirectRelationships", []),
@@ -186,6 +188,7 @@ def get_complete_registry_data(
                 "featureServices": pagination.get("featureServices", {}),
                 "features": pagination.get("features", {}),
                 "labels": pagination.get("labels", {}),
+                "savedDatasets": pagination.get("savedDatasets", {}),
                 "relationships": lineage_response.get("relationshipsPagination", {}),
                 "indirectRelationships": lineage_response.get(
                     "indirectRelationshipsPagination", {}
@@ -274,6 +277,8 @@ def get_complete_registry_data_all(
                 feat["project"] = project_name
             for lbl in project_resources.get("labels", []):
                 lbl["project"] = project_name
+            for sd in project_resources.get("savedDatasets", []):
+                sd["project"] = project_name
             all_data.append(
                 {
                     "project": project_name,
@@ -285,6 +290,7 @@ def get_complete_registry_data_all(
                         "featureServices": project_resources.get("featureServices", []),
                         "features": project_resources.get("features", []),
                         "labels": project_resources.get("labels", []),
+                        "savedDatasets": project_resources.get("savedDatasets", []),
                     },
                     "relationships": lineage_response.get("relationships", []),
                     "indirectRelationships": lineage_response.get(

@@ -376,6 +376,7 @@ def _init_openlineage_emitter(self) -> Optional[Any]:
                 ol_config = self.config.openlineage.to_openlineage_config()
                 emitter = FeastOpenLineageEmitter(ol_config)
                 if emitter.is_enabled:
+                    self._wire_local_processor(emitter)
                     return emitter
         except ImportError:
             # OpenLineage not installed, silently skip
@@ -384,6 +385,21 @@ def _init_openlineage_emitter(self) -> Optional[Any]:
             warnings.warn(f"Failed to initialize OpenLineage emitter: {e}")
         return None
 
+    def _wire_local_processor(self, emitter: Any) -> None:
+        """Wire the local OL consumer processor into the emitter so
+        Feast-produced events are also stored in the consumer DB."""
+        try:
+            from feast.api.registry.rest import get_ol_processor
+
+            processor = get_ol_processor()
+            if processor and hasattr(emitter, "_client") and emitter._client:
+                emitter._client.set_local_processor(processor)
+                _logger.info(
+                    "Feast OL emitter wired to local consumer processor (lazy)"
+                )
+        except Exception as e:
+            _logger.debug(f"Could not wire emitter to local processor: {e}")
+
     def __repr__(self) -> str:
         # Show lazy loading status without triggering initialization
         registry_status = "not loaded" if self._registry is None else "loaded"
@@ -532,6 +548,7 @@ def _submit_and_process_materialization_jobs(
         regular_fvs: list,
         previous_states: dict,
         date_range: "_MaterializationDateRange",
+        openlineage_run_id: Optional[str] = None,
     ) -> None:
         """
         Submit all tasks to the engine in one call and process the results.
@@ -544,8 +561,22 @@ def _submit_and_process_materialization_jobs(
         )
 
         batch_start = time.monotonic()
+        materialize_kwargs: Dict[str, Any] = {}
+        if openlineage_run_id and self.openlineage_emitter is not None:
+            from feast.openlineage.identity import (
+                LineageParentContext,
+                materialize_job_name,
+            )
+
+            materialize_kwargs["lineage_parent"] = LineageParentContext(
+                job_namespace=self.openlineage_emitter.namespace_for(self.project),
+                job_name=materialize_job_name(self.project),
+                run_id=openlineage_run_id,
+            )
         try:
-            jobs = provider.batch_engine.materialize(self.registry, tasks)
+            jobs = provider.batch_engine.materialize(
+                self.registry, tasks, **materialize_kwargs
+            )
         except Exception:
             self._rollback_fv_states(regular_fvs, previous_states)
             raise
@@ -602,6 +633,7 @@ def _materialize_fvs_batch(
         end_date: datetime,
         tqdm_builder,
         disable_event_timestamp: bool = False,
+        openlineage_run_id: Optional[str] = None,
     ) -> None:
         """Batch path: collect all FVs, submit to engine in one call.
 
@@ -638,6 +670,7 @@ def _materialize_fvs_batch(
                 regular_fvs,
                 previous_states,
                 date_range,
+                openlineage_run_id=openlineage_run_id,
             )
 
     @property
@@ -1874,9 +1907,18 @@ def _mlflow_log_apply(
             _logger.debug("MLflow apply logging failed: %s", e)
 
     def _emit_openlineage_apply(self, objects: List[Any]):
-        """Emit OpenLineage events for applied objects."""
+        """Emit OpenLineage events for applied objects.
+
+        Skips when using a remote registry — the RegistryServer already
+        emits OL events in its Apply* handlers, so emitting here would
+        double-count every object.
+        """
         if self.openlineage_emitter is None:
             return
+        from feast.infra.registry.remote import RemoteRegistry
+
+        if isinstance(self._registry, RemoteRegistry):
+            return
         try:
             self.openlineage_emitter.emit_apply(objects, self.project)
         except Exception as e:
@@ -1915,21 +1957,9 @@ def teardown(self):
     def _teardown_openlineage(self):
         """Clean up OpenLineage data for this project's namespace during teardown."""
         try:
-            if (
-                hasattr(self.config, "openlineage")
-                and self.config.openlineage is not None
-                and self.config.openlineage.enabled
-            ):
-                ol_config = self.config.openlineage.to_openlineage_config()
-                consumer_cfg = getattr(ol_config, "consumer", None)
-                if consumer_cfg and getattr(consumer_cfg, "enabled", False):
-                    conn_str = getattr(consumer_cfg, "connection_string", None)
-                    if conn_str:
-                        from feast.openlineage.store import OpenLineageStore
-
-                        ol_store = OpenLineageStore(connection_string=conn_str)
-                        namespace = f"{self.project}/{self.project}"
-                        ol_store.purge_namespace(namespace)
+            emitter = self.openlineage_emitter
+            if emitter is not None:
+                emitter.teardown_project(self.project)
         except Exception as e:
             warnings.warn(f"Failed to clean up OpenLineage data during teardown: {e}")
 
@@ -2565,6 +2595,7 @@ def tqdm_builder(length):
                     regular_fvs_with_dates,
                     end_date_tz,
                     tqdm_builder,
+                    openlineage_run_id=ol_run_id,
                 )
             else:
                 for feature_view, start_date in regular_fvs_with_dates:
@@ -2757,6 +2788,7 @@ def tqdm_builder(length):
                     end_date,
                     tqdm_builder,
                     disable_event_timestamp=disable_event_timestamp,
+                    openlineage_run_id=ol_run_id,
                 )
             else:
                 for feature_view, fv_start in regular_fvs_with_dates:
@@ -2882,7 +2914,11 @@ def _emit_openlineage_materialize_start(
             return None
         try:
             run_id, success = self.openlineage_emitter.emit_materialize_start(
-                feature_views, start_date, end_date, self.project
+                feature_views,
+                start_date,
+                end_date,
+                self.project,
+                online_store=getattr(self.config, "online_store", None),
             )
             # Return run_id only if START was successfully emitted
             # This prevents orphaned COMPLETE/FAIL events
@@ -2901,7 +2937,10 @@ def _emit_openlineage_materialize_complete(
             return
         try:
             self.openlineage_emitter.emit_materialize_complete(
-                run_id, feature_views, self.project
+                run_id,
+                feature_views,
+                self.project,
+                online_store=getattr(self.config, "online_store", None),
             )
         except Exception as e:
             warnings.warn(f"Failed to emit OpenLineage materialize complete event: {e}")

@@ -1,5 +1,5 @@
 from abc import ABC, abstractmethod
-from typing import List, Sequence, Union
+from typing import TYPE_CHECKING, List, Optional, Sequence, Union
 
 import pyarrow as pa
 
@@ -19,6 +19,9 @@
 from feast.on_demand_feature_view import OnDemandFeatureView
 from feast.stream_feature_view import StreamFeatureView
 
+if TYPE_CHECKING:
+    from feast.openlineage.identity import LineageParentContext
+
 
 class ComputeEngine(ABC):
     """
@@ -104,10 +107,24 @@ def materialize(
         self,
         registry: BaseRegistry,
         tasks: Union[MaterializationTask, List[MaterializationTask]],
+        *,
+        lineage_parent: Optional["LineageParentContext"] = None,
         **kwargs,
     ) -> List[MaterializationJob]:
+        """Materialize features for the given tasks.
+
+        Args:
+            registry: Feature registry
+            tasks: One or more materialization tasks
+            lineage_parent: Optional OpenLineage parent run. Engines that emit
+                their own lineage (e.g. SparkApplication) should attach this as
+                a parentRun; others ignore it.
+            **kwargs: Engine-specific options
+        """
         if isinstance(tasks, MaterializationTask):
             tasks = [tasks]
+        if lineage_parent is not None:
+            kwargs["lineage_parent"] = lineage_parent
         return [self._materialize_one(registry, task, **kwargs) for task in tasks]
 
     def _materialize_one(