# Copyright 2026 The Feast Authors

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# https://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

"""

Feast OpenLineage Emitter.

This module provides high-level functions for emitting OpenLineage events

from Feast operations like materialization, feature retrieval, and registry changes.

"""

import logging

import uuid

from datetime import datetime

from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union

if TYPE_CHECKING:

from feast import FeatureService, FeatureView

from feast.infra.registry.base_registry import BaseRegistry

from feast.labeling.label_view import LabelView

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.stream_feature_view import StreamFeatureView

from feast.openlineage.client import FeastOpenLineageClient

from feast.openlineage.config import OpenLineageConfig

try:

from openlineage.client.event_v2 import (

InputDataset,

OutputDataset,

RunState,

)

OPENLINEAGE_AVAILABLE = True

except ImportError:

OPENLINEAGE_AVAILABLE = False

logger = logging.getLogger(__name__)

class FeastOpenLineageEmitter:

"""

High-level emitter for Feast OpenLineage events.

This class provides methods for emitting lineage events for various

Feast operations including:

- Registry apply (feature view, feature service definitions)

- Materialization (batch and incremental)

- Feature retrieval (online and historical)

- Data source relationships

Example:

from feast import FeatureStore

from feast.openlineage import FeastOpenLineageEmitter, OpenLineageConfig

config = OpenLineageConfig(transport_type="http", transport_url="http://localhost:5000")

emitter = FeastOpenLineageEmitter(config)

fs = FeatureStore(repo_path="feature_repo")

# Emit lineage for registry

emitter.emit_registry_lineage(fs.registry, fs.project)

"""

def __init__(

self,

config: Optional[OpenLineageConfig] = None,

client: Optional[FeastOpenLineageClient] = None,

):

"""

Initialize the Feast OpenLineage Emitter.

Args:

config: OpenLineage configuration

client: Optional pre-configured FeastOpenLineageClient

"""

self._config = config or OpenLineageConfig.from_env()

self._client = client or FeastOpenLineageClient(self._config)

@property

def is_enabled(self) -> bool:

"""Check if the emitter is enabled."""

return self._client.is_enabled

@property

def namespace(self) -> str:

"""Get the default namespace."""

return self._config.namespace

def namespace_for(self, project: str) -> str:

"""Public OpenLineage namespace for a Feast project."""

from feast.openlineage.identity import resolve_namespace

return resolve_namespace(self._config.namespace, project)

def _get_namespace(self, project: str) -> str:

"""Backward-compatible alias for :meth:`namespace_for`."""

return self.namespace_for(project)

@staticmethod

def _match_storage_to_datasources(

saved_dataset: Any,

registered_data_sources: Optional[List[Any]],

) -> List[str]:

"""Match a SavedDataset's storage location against registered DataSources.

Compares the physical identifiers (URI, table, path) from the

SavedDataset's storage with those from each registered DataSource.

Returns:

List of matched DataSource names.

"""

if not registered_data_sources or not hasattr(saved_dataset, "storage"):

return []

try:

storage_proto = saved_dataset.storage.to_proto()

except Exception:

return []

from feast.lineage.registry_lineage import (

_extract_datasource_identifiers,

_extract_storage_identifiers,

)

storage_ids = _extract_storage_identifiers(storage_proto)

if not storage_ids:

return []

matched: List[str] = []

seen: set = set()

for ds in registered_data_sources:

if not hasattr(ds, "name") or not ds.name or ds.name in seen:

continue

try:

ds_ids = _extract_datasource_identifiers(ds.to_proto())

except Exception:

continue

if storage_ids & ds_ids:

seen.add(ds.name)

matched.append(ds.name)

return matched

def _job_kind_facets(self, kind: str, project: str) -> Dict[str, Any]:

from feast.openlineage.facets import FeastJobKindFacet

from feast.openlineage.identity import FeastJobKind

kind_value = kind.value if isinstance(kind, FeastJobKind) else str(kind)

return {

"feast_jobKind": FeastJobKindFacet(

kind=kind_value,

feast_project=project,

)

}

def teardown_project(self, project: str) -> None:

"""Purge consumer-store lineage for this project's namespace.

No-op when the OpenLineage consumer is disabled or has no connection.

"""

consumer_cfg = getattr(self._config, "consumer", None)

if not consumer_cfg or not getattr(consumer_cfg, "enabled", False):

return

conn_str = getattr(consumer_cfg, "connection_string", None)

if not conn_str:

return

from feast.openlineage.store import OpenLineageStore

OpenLineageStore(connection_string=conn_str).purge_namespace(

self.namespace_for(project)

)

def emit_registry_lineage(

self,

registry: "BaseRegistry",

project: str,

allow_cache: bool = True,

) -> List[bool]:

"""

Emit lineage events for all objects in a Feast registry.

This method emits JobEvents for feature views and feature services,

and DatasetEvents for data sources and entities.

Args:

registry: Feast registry

project: Project name

allow_cache: Whether to use cached registry data

Returns:

List of success/failure indicators for each event

"""

if not self.is_enabled:

return []

from feast.feature_view import FeatureView

from feast.labeling.label_view import LabelView

from feast.on_demand_feature_view import OnDemandFeatureView

from feast.stream_feature_view import StreamFeatureView

results = []

# Get all feature views at once (includes FeatureView, StreamFeatureView, OnDemandFeatureView, LabelView)

all_feature_views: list = []

try:

all_feature_views = registry.list_all_feature_views(

project=project, allow_cache=allow_cache

)

except Exception as e:

logger.error(f"Error listing all feature views: {e}")

# Emit lineage events for each feature view type

for fv in all_feature_views:

try:

if isinstance(fv, OnDemandFeatureView):

result = self.emit_on_demand_feature_view_lineage(fv, project)

elif isinstance(fv, StreamFeatureView):

result = self.emit_stream_feature_view_lineage(fv, project)

elif isinstance(fv, LabelView):

result = self.emit_label_view_lineage(fv, project)

elif isinstance(fv, FeatureView):

result = self.emit_feature_view_lineage(fv, project)

else:

continue

results.append(result)

except Exception as e:

logger.error(f"Error emitting lineage for feature view {fv.name}: {e}")

# Emit events for feature services

try:

feature_services = registry.list_feature_services(

project=project, allow_cache=allow_cache

)

for fs in feature_services:

result = self.emit_feature_service_lineage(

fs, all_feature_views, project

)

results.append(result)

except Exception as e:

logger.error(f"Error emitting feature service lineage: {e}")

# Emit events for saved datasets

try:

saved_datasets = registry.list_saved_datasets(

project=project, allow_cache=allow_cache

)

# Fetch registered data sources for storage-based matching

registered_data_sources = []

try:

registered_data_sources = registry.list_data_sources(

project=project, allow_cache=allow_cache

)

except Exception:

pass

for sd in saved_datasets:

result = self.emit_saved_dataset_lineage(

sd, project, registered_data_sources=registered_data_sources

)

results.append(result)

except Exception as e:

logger.error(f"Error emitting saved dataset lineage: {e}")

logger.info(

f"Emitted {sum(results)}/{len(results)} lineage events for registry"

)

return results

def emit_feature_view_lineage(

self,

feature_view: "FeatureView",

project: str,

) -> bool:

"""

Emit lineage for a feature view definition.

Args:

feature_view: The feature view

project: Project name

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

from feast.openlineage.mappers import feature_view_to_job

try:

namespace = self._get_namespace(project)

job, inputs, outputs = feature_view_to_job(

feature_view,

namespace=namespace,

)

# Emit a RunEvent with COMPLETE state to create lineage connection

result = self._client.emit_run_event(

job_name=job.name,

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job.facets,

namespace=namespace,

)

return result

except Exception as e:

logger.error(

f"Error emitting feature view lineage for {feature_view.name}: {e}"

)

return False

def emit_label_view_lineage(

self,

label_view: "LabelView",

project: str,

) -> bool:

"""

Emit lineage for a label view definition.

Args:

label_view: The label view

project: Project name

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

from feast.openlineage.mappers import feature_view_to_job

try:

namespace = self._get_namespace(project)

job, inputs, outputs = feature_view_to_job(

label_view,

namespace=namespace,

job_type="LABEL_VIEW",

)

result = self._client.emit_run_event(

job_name=job.name,

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job.facets,

namespace=namespace,

)

return result

except Exception as e:

logger.error(

f"Error emitting label view lineage for {label_view.name}: {e}"

)

return False

def emit_stream_feature_view_lineage(

self,

stream_feature_view: "StreamFeatureView",

project: str,

) -> bool:

"""

Emit lineage for a stream feature view definition.

Args:

stream_feature_view: The stream feature view

project: Project name

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

from feast.openlineage.mappers import feature_view_to_job

try:

namespace = self._get_namespace(project)

# StreamFeatureView inherits from FeatureView

job, inputs, outputs = feature_view_to_job(

stream_feature_view,

namespace=namespace,

)

# Emit a RunEvent with COMPLETE state to create lineage connection

return self._client.emit_run_event(

job_name=f"stream_{job.name}",

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job.facets,

namespace=namespace,

)

except Exception as e:

logger.error(

f"Error emitting stream feature view lineage for {stream_feature_view.name}: {e}"

)

return False

def emit_on_demand_feature_view_lineage(

self,

odfv: "OnDemandFeatureView",

project: str,

) -> bool:

"""

Emit lineage for an on-demand feature view definition.

Args:

odfv: The on-demand feature view

project: Project name

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

from feast.openlineage.facets import FeastFeatureViewFacet

from feast.openlineage.mappers import feast_field_to_schema_field

try:

from openlineage.client.facet_v2 import schema_dataset

namespace = self._get_namespace(project)

# Build inputs from sources

inputs = []

for source_name, fv_proj in odfv.source_feature_view_projections.items():

inputs.append(

InputDataset(

namespace=namespace,

name=fv_proj.name,

)

)

for source_name, req_source in odfv.source_request_sources.items():

req_name = getattr(req_source, "name", source_name)

inputs.append(

InputDataset(

namespace=namespace,

name=req_name,

)

)

# Build output with feast_featureView facet on the dataset

output_facets: Dict[str, Any] = {

"feast_featureView": FeastFeatureViewFacet(

name=odfv.name,

ttl_seconds=0,

entities=[],

features=[f.name for f in odfv.features] if odfv.features else [],

online_enabled=True,

offline_enabled=True,

mode="ON_DEMAND",

description=odfv.description if odfv.description else "",

owner=odfv.owner if hasattr(odfv, "owner") and odfv.owner else "",

tags=odfv.tags if odfv.tags else {},

),

}

if odfv.features:

output_facets["schema"] = schema_dataset.SchemaDatasetFacet(

fields=[feast_field_to_schema_field(f) for f in odfv.features]

)

outputs = [

OutputDataset(

namespace=namespace,

name=odfv.name,

facets=output_facets, # type: ignore[arg-type]

)

]

from feast.openlineage.facets import FeastProjectFacet

from feast.openlineage.identity import FeastJobKind

job_facets = {

"feast_project": FeastProjectFacet(project_name=project),

**self._job_kind_facets(FeastJobKind.DEFINITION, project),

}

# Emit a RunEvent with COMPLETE state to create lineage connection

return self._client.emit_run_event(

job_name=f"on_demand_feature_view_{odfv.name}",

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job_facets,

namespace=namespace,

)

except Exception as e:

logger.error(

f"Error emitting on-demand feature view lineage for {odfv.name}: {e}"

)

return False

def emit_feature_service_lineage(

self,

feature_service: "FeatureService",

feature_views: List[

Union["FeatureView", "OnDemandFeatureView", "StreamFeatureView"]

],

project: str,

) -> bool:

"""

Emit lineage for a feature service definition.

Args:

feature_service: The feature service

feature_views: List of all available feature views (FeatureView,

OnDemandFeatureView, StreamFeatureView)

project: Project name

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

from feast.openlineage.mappers import feature_service_to_job

try:

# Find the feature views referenced by this service

namespace = self._get_namespace(project)

fv_names = {proj.name for proj in feature_service.feature_view_projections}

referenced_fvs = [fv for fv in feature_views if fv.name in fv_names]

job, inputs, outputs = feature_service_to_job(

feature_service,

referenced_fvs,

namespace=namespace,

)

# Emit a RunEvent with COMPLETE state to create lineage connection

return self._client.emit_run_event(

job_name=job.name,

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job.facets,

namespace=namespace,

)

except Exception as e:

logger.error(

f"Error emitting feature service lineage for {feature_service.name}: {e}"

)

return False

def emit_saved_dataset_lineage(

self,

saved_dataset: Any,

project: str,

registered_data_sources: Optional[List[Any]] = None,

) -> bool:

"""

Emit lineage for a saved dataset definition.

Emits two events:

1. A **RunEvent** that captures the definition-time topology:

FeatureService/FeatureView → SavedDataset job.

2. A **DatasetEvent** with SymlinksDatasetFacet that links the

logical SavedDataset name to its physical storage URI, plus

OwnershipDatasetFacet and LifecycleStateChangeDatasetFacet

when the information is available.

Args:

saved_dataset: The SavedDataset object

project: Project name

registered_data_sources: Optional list of registered DataSource objects

for storage-based matching

Returns:

True if successful, False otherwise

"""

if not self.is_enabled:

return False

try:

from openlineage.client.facet_v2 import schema_dataset

from feast.openlineage.facets import (

FeastProjectFacet,

FeastSavedDatasetFacet,

)

from feast.openlineage.identity import FeastJobKind

namespace = self._get_namespace(project)

inputs = []

if saved_dataset.feature_service_name:

inputs.append(

InputDataset(

namespace=namespace,

name=saved_dataset.feature_service_name,

)

)

# FeatureView inputs from feature refs ("view_name:feature_name")

if saved_dataset.features:

from feast.utils import _parse_feature_ref

seen_views: set = set()

for feat_ref in saved_dataset.features:

try:

view_name, _, _ = _parse_feature_ref(feat_ref)

except ValueError:

continue

if view_name and view_name not in seen_views:

seen_views.add(view_name)

inputs.append(

InputDataset(

namespace=namespace,

name=view_name,

)

)

# DataSource inputs matched by storage location

matched_ds = self._match_storage_to_datasources(

saved_dataset, registered_data_sources

)

for ds_name in matched_ds:

inputs.append(

InputDataset(

namespace=namespace,

name=ds_name,

)

)

sd_facets: Dict[str, Any] = {

"feast_savedDataset": FeastSavedDatasetFacet(

name=saved_dataset.name,

features=list(saved_dataset.features)

if saved_dataset.features

else [],

join_keys=list(saved_dataset.join_keys)

if saved_dataset.join_keys

else [],

feature_service_name=saved_dataset.feature_service_name or "",

full_feature_names=saved_dataset.full_feature_names,

description=saved_dataset.description

if hasattr(saved_dataset, "description")

and saved_dataset.description

else "",

tags=saved_dataset.tags if saved_dataset.tags else {},

)

}

if saved_dataset.features:

sd_facets["schema"] = schema_dataset.SchemaDatasetFacet(

fields=[

schema_dataset.SchemaDatasetFacetFields(name=f, type="UNKNOWN")

for f in saved_dataset.features

]

)

outputs = [

OutputDataset(

namespace=namespace,

name=saved_dataset.name,

facets=sd_facets,

)

]

job_facets = {

"feast_project": FeastProjectFacet(project_name=project),

**self._job_kind_facets(FeastJobKind.DEFINITION, project),

}

run_result = self._client.emit_run_event(

job_name=f"saved_dataset_{saved_dataset.name}",

run_id=str(uuid.uuid4()),

event_type=RunState.COMPLETE,

inputs=inputs,

outputs=outputs,

job_facets=job_facets,

namespace=namespace,

)

# Emit a DatasetEvent with symlink, ownership, and lifecycle facets

self._emit_saved_dataset_metadata(

saved_dataset, project, namespace, registered_data_sources

)

return run_result

except Exception as e:

logger.error(

f"Error emitting saved dataset lineage for {saved_dataset.name}: {e}"

)

return False

def _emit_saved_dataset_metadata(

self,

saved_dataset: Any,

project: str,

namespace: str,

registered_data_sources: Optional[List[Any]] = None,

):

"""Emit a DatasetEvent for a SavedDataset with OL-standard facets.

Includes:

- SymlinksDatasetFacet linking logical name to physical storage URI

- OwnershipDatasetFacet from tags (``owner`` key) or the object owner

- LifecycleStateChangeDatasetFacet (CREATE)

- SchemaDatasetFacet from features

"""

try:

from openlineage.client.facet_v2 import (

lifecycle_state_change_dataset,

ownership_dataset,

schema_dataset,

symlinks_dataset,

)

except ImportError:

logger.debug(

"openlineage facet_v2 modules not available; "

"skipping DatasetEvent for SavedDataset"

)

return

ds_facets: Dict[str, Any] = {}

# SymlinksDatasetFacet: link logical name → physical storage URI

storage_uri = self._get_saved_dataset_storage_uri(saved_dataset)

if storage_uri:

ds_facets["symlinks"] = symlinks_dataset.SymlinksDatasetFacet(

identifiers=[

symlinks_dataset.Identifier(

namespace=namespace,

name=storage_uri,

type="TABLE",

)

]

)

# OwnershipDatasetFacet

owner = None

if hasattr(saved_dataset, "tags") and saved_dataset.tags:

owner = saved_dataset.tags.get("owner")

if not owner and hasattr(saved_dataset, "owner") and saved_dataset.owner:

owner = saved_dataset.owner

if owner:

ds_facets["ownership"] = ownership_dataset.OwnershipDatasetFacet(

owners=[

ownership_dataset.Owner(

name=owner,

type="FEAST_TAG",

)

]

)

# LifecycleStateChangeDatasetFacet

ds_facets["lifecycleStateChange"] = (

lifecycle_state_change_dataset.LifecycleStateChangeDatasetFacet(

lifecycleStateChange=lifecycle_state_change_dataset.LifecycleStateChange.CREATE,

)

)

# SchemaDatasetFacet

if saved_dataset.features:

ds_facets["schema"] = schema_dataset.SchemaDatasetFacet(

fields=[

schema_dataset.SchemaDatasetFacetFields(name=f, type="UNKNOWN")

for f in saved_dataset.features

]

)

try:

self._client.emit_dataset_event(

dataset_name=saved_dataset.name,

namespace=namespace,

facets=ds_facets,

)

except Exception as e:

logger.warning(

f"Failed to emit DatasetEvent for SavedDataset "

f"{saved_dataset.name}: {e}"

)

@staticmethod

def _get_saved_dataset_storage_uri(saved_dataset: Any) -> Optional[str]:

"""Extract a physical storage URI from a SavedDataset's storage."""

if not hasattr(saved_dataset, "storage") or not saved_dataset.storage:

return None

try:

storage_proto = saved_dataset.storage.to_proto()

except Exception:

return None

if hasattr(storage_proto, "file_url") and storage_proto.file_url:

return storage_proto.file_url

if hasattr(storage_proto, "bigquery_path") and storage_proto.bigquery_path:

return storage_proto.bigquery_path

if hasattr(storage_proto, "redshift_path") and storage_proto.redshift_path:

return storage_proto.redshift_path

if hasattr(storage_proto, "s3_path") and storage_proto.s3_path:

return storage_proto.s3_path

if hasattr(storage_proto, "trino_path") and storage_proto.trino_path:

return storage_proto.trino_path

for attr_name in dir(storage_proto):

if attr_name.startswith("_"):

continue

if "path" in attr_name or "url" in attr_name or "uri" in attr_name:

val = getattr(storage_proto, attr_name, None)

if val and isinstance(val, str):

return val

return None

def emit_materialize_start(

self,

feature_views: List["FeatureView"],

start_date: Optional[datetime],

end_date: datetime,

project: str,

run_id: Optional[str] = None,

online_store_type: str = "online",

online_store: Any = None,

) -> Tuple[str, bool]:

"""

Emit a START event for a materialization run.

Args:

feature_views: Feature views being materialized

start_date: Start of materialization window (None for incremental)

end_date: End of materialization window

project: Project name

run_id: Optional run ID (will be generated if not provided)

online_store_type: Backend type for the online store sink (redis, …)

online_store: Optional RepoConfig.online_store; when set, type is

resolved via :func:`resolve_online_store_type` and overrides

``online_store_type``.

Returns:

Tuple of (run_id, success)

"""

if not self.is_enabled or not self._config.emit_on_materialize:

return "", False

from feast.openlineage.facets import FeastMaterializationFacet

from feast.openlineage.identity import (

FeastJobKind,

materialize_job_name,

)

from feast.openlineage.mappers import (

data_source_to_dataset,

feature_view_to_dataset,

online_store_to_dataset,

resolve_online_store_type,

)

if online_store is not None:

online_store_type = resolve_online_store_type(online_store)

run_id = run_id or str(uuid.uuid4())

try:

namespace = self.namespace_for(project)

# Build inputs (data sources) - include both batch and stream sources

inputs = []

seen_sources = set() # Track source names to avoid duplicates

for fv in feature_views:

# FeatureView as input — same dataset name as feast apply, with

# feast_featureView facet so mapping/metadata survive materialize.

if fv.name and fv.name not in seen_sources:

seen_sources.add(fv.name)

inputs.append(

feature_view_to_dataset(fv, namespace=namespace, as_input=True)

)

# Add batch source

if hasattr(fv, "batch_source") and fv.batch_source:

source_name = getattr(fv.batch_source, "name", None)

if source_name and source_name not in seen_sources:

seen_sources.add(source_name)

inputs.append(

data_source_to_dataset(

fv.batch_source,

namespace=namespace,

as_input=True,

)

)

# Add stream source (e.g., PushSource)

if hasattr(fv, "stream_source") and fv.stream_source:

source_name = getattr(fv.stream_source, "name", None)

if source_name and source_name not in seen_sources:

seen_sources.add(source_name)

inputs.append(

data_source_to_dataset(

fv.stream_source,

namespace=namespace,

as_input=True,

)

)

# Add entities as inputs with feast_entity facet for mapping

if hasattr(fv, "entities") and fv.entities:

from feast.openlineage.facets import FeastEntityFacet

for entity_name in fv.entities:

if entity_name and entity_name != "__dummy":

if entity_name not in seen_sources:

seen_sources.add(entity_name)

inputs.append(

InputDataset(

namespace=namespace,

name=entity_name,

facets={

"feast_entity": FeastEntityFacet(

name=entity_name,

join_keys=[],

value_type="STRING",

description="",

owner="",

tags={},

)

},

)

)

# Build outputs — physical online-store sinks (not FeatureView nodes)

outputs = [

online_store_to_dataset(

store_type=online_store_type,

feature_view_name=fv.name,

namespace=namespace,

)

for fv in feature_views

]

# Build run facets

run_facets = {

"feast_materialization": FeastMaterializationFacet(

feature_views=[fv.name for fv in feature_views],

start_date=start_date.isoformat() if start_date else None,

end_date=end_date.isoformat() if end_date else None,

project=project,

online_store_type=online_store_type or "",

)

}

job_facets = self._job_kind_facets(FeastJobKind.TRANSFORM, project)

success = self._client.emit_run_event(

job_name=materialize_job_name(project),

run_id=run_id,

event_type=RunState.START,

inputs=inputs,

outputs=outputs,

job_facets=job_facets,

run_facets=run_facets,

namespace=namespace,

)

return run_id, success

except Exception as e:

logger.error(f"Error emitting materialize start event: {e}")

return run_id, False

def emit_materialize_complete(

self,

run_id: str,

feature_views: List["FeatureView"],

project: str,

rows_written: Optional[int] = None,

online_store_type: str = "online",

online_store: Any = None,

) -> bool:

"""

Emit a COMPLETE event for a materialization run.

Args:

run_id: Run ID from the start event

feature_views: Feature views that were materialized

project: Project name

rows_written: Optional count of rows written

online_store_type: Backend type for the online store sink (redis, …)

online_store: Optional RepoConfig.online_store; overrides type when set.

Returns:

True if successful, False otherwise

"""

if not self.is_enabled or not self._config.emit_on_materialize:

return False

from feast.openlineage.facets import FeastMaterializationFacet