Skip to content

Commit ec272d5

Browse files
ntkatholentkathole
committed
feat: Add Label View to init template
Signed-off-by: ntkathole <[email protected]>
1 parent fa271be commit ec272d5

14 files changed

Lines changed: 344 additions & 120 deletions

File tree

‎sdk/python/feast/infra/online_stores/sqlite.py‎

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
from feast.infra.online_stores.helpers import compute_table_id
4646
from feast.infra.online_stores.online_store import OnlineStore
4747
from feast.infra.online_stores.vector_store import VectorStoreConfig
48+
from feast.labeling.label_view import LabelView
4849
from feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto
4950
from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto
5051
from feast.protos.feast.core.SqliteTable_pb2 import SqliteTable as SqliteTableProto
@@ -317,21 +318,33 @@ def plan(
317318
self, config: RepoConfig, desired_registry_proto: RegistryProto
318319
) -> List[InfraObject]:
319320
project = config.project
321+
versioning = config.registry.enable_online_feature_view_versioning
320322

321323
infra_objects: List[InfraObject] = [
322324
SqliteTable(
323325
path=self._get_db_path(config),
324326
name=_table_id(
325327
project,
326328
FeatureView.from_proto(view),
327-
config.registry.enable_online_feature_view_versioning,
329+
versioning,
328330
),
329331
)
330332
for view in [
331333
*desired_registry_proto.feature_views,
332334
*desired_registry_proto.stream_feature_views,
333335
]
334336
]
337+
338+
for lv_proto in desired_registry_proto.label_views:
339+
if lv_proto.spec.online:
340+
lv = LabelView.from_proto(lv_proto)
341+
infra_objects.append(
342+
SqliteTable(
343+
path=self._get_db_path(config),
344+
name=_table_id(project, lv, versioning),
345+
)
346+
)
347+
335348
return infra_objects
336349

337350
def teardown(
@@ -716,7 +729,7 @@ def _initialize_conn(
716729
return db
717730

718731

719-
def _table_id(project: str, table: FeatureView, enable_versioning: bool = False) -> str:
732+
def _table_id(project: str, table: Any, enable_versioning: bool = False) -> str:
720733
return compute_table_id(project, table, enable_versioning)
721734

722735

‎sdk/python/feast/infra/registry/sql.py‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1598,8 +1598,15 @@ def _list_objects(
15981598
skip_udf: If True, call from_proto() but skip deserializing UDFs
15991599
(dill.loads). Returns Python objects suitable for filtering and
16001600
display without requiring the UDF's source module to be installed.
1601-
Only relevant for feature view types.
1601+
Only relevant for feature view types that contain UDFs.
16021602
"""
1603+
import inspect
1604+
1605+
supports_skip_udf = (
1606+
skip_udf
1607+
and "skip_udf" in inspect.signature(python_class.from_proto).parameters
1608+
)
1609+
16031610
with self.read_engine.begin() as conn:
16041611
stmt = select(table).where(table.c.project_id == project)
16051612
rows = conn.execute(stmt).all()
@@ -1611,8 +1618,8 @@ def _list_objects(
16111618
objects.append(proto)
16121619
else:
16131620
obj = (
1614-
python_class.from_proto(proto, skip_udf=skip_udf)
1615-
if skip_udf
1621+
python_class.from_proto(proto, skip_udf=True)
1622+
if supports_skip_udf
16161623
else python_class.from_proto(proto)
16171624
)
16181625
if utils.has_all_tags(obj.tags, tags):

‎sdk/python/feast/templates/local/bootstrap.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,9 @@ def bootstrap():
77
import pathlib
88
from datetime import datetime, timedelta
99

10+
import pyarrow as pa
11+
import pyarrow.parquet as pq
12+
1013
from feast.driver_test_data import create_driver_hourly_stats_df
1114

1215
repo_path = pathlib.Path(__file__).parent.absolute() / "feature_repo"
@@ -23,6 +26,25 @@ def bootstrap():
2326
driver_stats_path = data_path / "driver_stats.parquet"
2427
driver_df.to_parquet(path=str(driver_stats_path), allow_truncated_timestamps=True)
2528

29+
# Create an empty parquet file for the label view batch source
30+
# Use explicit pyarrow schema to ensure proper column types even with zero rows
31+
label_schema = pa.schema(
32+
[
33+
("driver_id", pa.int64()),
34+
("is_reliable", pa.int64()),
35+
("quality_score", pa.float32()),
36+
("reviewer_notes", pa.string()),
37+
("labeler", pa.string()),
38+
("event_timestamp", pa.timestamp("ns")),
39+
]
40+
)
41+
label_table = pa.table(
42+
{field.name: pa.array([], type=field.type) for field in label_schema},
43+
schema=label_schema,
44+
)
45+
label_data_path = data_path / "driver_quality_labels.parquet"
46+
pq.write_table(label_table, str(label_data_path))
47+
2648
example_py_file = repo_path / "feature_definitions.py"
2749
replace_str_in_file(example_py_file, "%PROJECT_NAME%", str(project_name))
2850
replace_str_in_file(
@@ -31,6 +53,11 @@ def bootstrap():
3153
replace_str_in_file(
3254
example_py_file, "%LOGGING_PATH%", str(data_path.relative_to(repo_path))
3355
)
56+
replace_str_in_file(
57+
example_py_file,
58+
"%LABEL_DATA_PATH%",
59+
str(label_data_path.relative_to(repo_path)),
60+
)
3461

3562

3663
if __name__ == "__main__":

‎sdk/python/feast/templates/local/feature_repo/feature_definitions.py‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,11 +5,13 @@
55
import pandas as pd
66

77
from feast import (
8+
ConflictPolicy,
89
Entity,
910
FeatureService,
1011
FeatureView,
1112
Field,
1213
FileSource,
14+
LabelView,
1315
Project,
1416
PushSource,
1517
RequestSource,
@@ -165,3 +167,41 @@ def transformed_conv_rate_fresh(inputs: pd.DataFrame) -> pd.DataFrame:
165167
name="driver_activity_v3",
166168
features=[driver_stats_fresh_fv, transformed_conv_rate_fresh],
167169
)
170+
171+
# --- Label Views ---
172+
# Label views manage mutable human labels for training data, RLHF, and evaluation.
173+
# They use PushSources so labels can be submitted from the UI or external tools.
174+
175+
driver_quality_labels_source = PushSource(
176+
name="driver_quality_labels_push",
177+
batch_source=FileSource(
178+
name="driver_quality_labels_batch",
179+
path="%LABEL_DATA_PATH%",
180+
timestamp_field="event_timestamp",
181+
),
182+
)
183+
184+
driver_quality_labels = LabelView(
185+
name="driver_quality_labels",
186+
entities=[driver],
187+
schema=[
188+
Field(name="is_reliable", dtype=Int64),
189+
Field(name="quality_score", dtype=Float32),
190+
Field(name="reviewer_notes", dtype=String),
191+
Field(name="labeler", dtype=String),
192+
],
193+
source=driver_quality_labels_source,
194+
labeler_field="labeler",
195+
conflict_policy=ConflictPolicy.LAST_WRITE_WINS,
196+
description="Human quality labels for drivers - used for model training and evaluation",
197+
tags={
198+
"feast.io/labeling-method": "table",
199+
"feast.io/field-role:is_reliable": "label",
200+
"feast.io/field-role:quality_score": "label",
201+
"feast.io/field-role:reviewer_notes": "metadata",
202+
"feast.io/label-values:is_reliable": "1,0",
203+
"feast.io/label-widget:is_reliable": "binary",
204+
"feast.io/label-widget:quality_score": "number",
205+
"feast.io/label-widget:reviewer_notes": "text",
206+
},
207+
)

0 commit comments

Comments
 (0)