Visitar URL original
fix: make vector length validation reachable, schema-driven and vecto… · feast-dev/feast@db6d34a · GitHub
Skip to content

Commit db6d34a

Browse files
haoxu0haoxu0
committed
fix: make vector length validation reachable, schema-driven and vectorized
Addresses three of the five items in #6907. Validation was reachable only from the online-write path. _validate_vector_features had a single call site inside _get_feature_view_and_df_for_online_write, so materialize, materialize_incremental and get_historical_features never ran it. Embedding workloads are overwhelmingly batch, so the path carrying vectors at volume was the unvalidated one. Every compute engine (local, spark, ray, flink, kubernetes, aws_lambda) and the passthrough provider funnel through utils._convert_arrow_to_proto, so a single Arrow-level check placed in _convert_arrow_fv_to_proto covers all of them. _validate_vector_field_lengths is O(1) for fixed-size lists and one vectorized pass for variable-size lists, so it is cheap enough to leave on. Null rows are tolerated; a non-list column with a declared vector_length is an error. Note the snowflake compute engine builds its rows without going through _convert_arrow_to_proto, so it is not covered by this change. The check assumed the vector was the first feature. feature_view.features[0].vector_index meant that declaring any field before the vector field silently disabled validation. Both validators now resolve the field via _get_feature_view_vector_field_metadata, which scans the schema and is already used elsewhere in feature_store.py. iterrows() did not scale. Replaced with a single pass over the column, which measured ~60x faster on 200k rows (1.51s to 0.025s) while keeping the row index and the two distinct failure messages. Not included, since #6907 raises them as behaviour decisions rather than fixes: requiring vector_length when vector_index=True (item 3), and whether Field.__eq__ should compare vector_index and vector_search_metric (item 5). Adds sdk/python/tests/unit/test_vector_length_validation.py with 16 tests covering both validators, including regression tests for the vector field not being first. Signed-off-by: hao-xu5 <hxu44@apple.com>
1 parent dd9f099 commit db6d34a

3 files changed

Lines changed: 271 additions & 14 deletions

File tree

‎sdk/python/feast/feature_store.py‎

Lines changed: 32 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -3619,27 +3619,45 @@ def _validate_vector_features(self, feature_view, df: pd.DataFrame) -> None:
36193619
"""
36203620
Validates vector features in the DataFrame against the feature view specifications.
36213621
3622+
Resolves the vector field from the feature view schema rather than assuming it is
3623+
the first feature, and checks lengths in a single pass over the column.
3624+
36223625
Args:
36233626
feature_view: The feature view containing vector feature specifications
36243627
df: The DataFrame to validate
36253628
36263629
Raises:
36273630
ValueError: If vector dimension constraints are violated
36283631
"""
3629-
if feature_view.features and feature_view.features[0].vector_index:
3630-
fv_vector_feature_name = feature_view.features[0].name
3631-
if feature_view.features[0].vector_length != 0:
3632-
for i, row in df.iterrows():
3633-
vector = row[fv_vector_feature_name]
3634-
if not hasattr(vector, "__len__"):
3635-
raise ValueError(
3636-
f"Row {i}: Vector feature '{fv_vector_feature_name}' is not a sequence. Got: {type(vector)}"
3637-
)
3638-
if len(vector) != feature_view.features[0].vector_length:
3639-
raise ValueError(
3640-
f"Row {i}: Vector length {len(vector)} does not match expected {feature_view.features[0].vector_length} "
3641-
f"for feature '{fv_vector_feature_name}' in feature view '{feature_view.name}'."
3642-
)
3632+
vector_field = _get_feature_view_vector_field_metadata(feature_view)
3633+
if vector_field is None or not vector_field.vector_length:
3634+
return
3635+
3636+
name = vector_field.name
3637+
if name not in df.columns:
3638+
return
3639+
3640+
expected = vector_field.vector_length
3641+
column = df[name]
3642+
3643+
# Single pass over the column. Sequences report their length, anything else
3644+
# reports None so the two failure modes stay distinguishable.
3645+
lengths = column.map(lambda v: len(v) if hasattr(v, "__len__") else None)
3646+
3647+
not_a_sequence = lengths.isna()
3648+
if not_a_sequence.any():
3649+
i = not_a_sequence.idxmax()
3650+
raise ValueError(
3651+
f"Row {i}: Vector feature '{name}' is not a sequence. Got: {type(column[i])}"
3652+
)
3653+
3654+
mismatched = lengths != expected
3655+
if mismatched.any():
3656+
i = mismatched.idxmax()
3657+
raise ValueError(
3658+
f"Row {i}: Vector length {lengths[i]} does not match expected {expected} "
3659+
f"for feature '{name}' in feature view '{feature_view.name}'."
3660+
)
36433661

36443662
def _get_feature_view_and_df_for_online_write(
36453663
self,

‎sdk/python/feast/utils.py‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -525,6 +525,8 @@ def _convert_arrow_fv_to_proto(
525525
f"Feature view '{feature_view.name}' has no batch_source and cannot be converted to proto."
526526
)
527527

528+
_validate_vector_field_lengths(table, feature_view)
529+
528530
# TODO: This will break if the feature view has aggregations or transformations
529531
columns = [
530532
(field.name, field.dtype.to_value_type()) for field in feature_view.features
@@ -1969,6 +1971,69 @@ def _get_feature_view_vector_field_metadata(
19691971
return vector_fields[0]
19701972

19711973

1974+
def _validate_vector_field_lengths(
1975+
table: Union[pyarrow.Table, pyarrow.RecordBatch],
1976+
feature_view,
1977+
) -> None:
1978+
"""Check an Arrow table's vector column against the declared ``vector_length``.
1979+
1980+
Called on the materialization path, where every compute engine funnels through
1981+
``_convert_arrow_to_proto``. The check is O(1) for fixed-size lists and a single
1982+
vectorized pass for variable-size lists, so it is safe to leave on.
1983+
1984+
A declared ``vector_length`` is a contract: a mismatch is an error, never a
1985+
silent truncation or a silently different response shape.
1986+
1987+
Args:
1988+
table: The Arrow table or record batch about to be converted.
1989+
feature_view: The feature view whose schema declares the vector field.
1990+
1991+
Raises:
1992+
ValueError: If the vector column's width disagrees with ``vector_length``.
1993+
"""
1994+
vector_field = _get_feature_view_vector_field_metadata(feature_view)
1995+
if vector_field is None or not vector_field.vector_length:
1996+
return
1997+
1998+
name = vector_field.name
1999+
if name not in table.schema.names:
2000+
return
2001+
2002+
expected = vector_field.vector_length
2003+
column = table.column(name)
2004+
column_type = column.type
2005+
2006+
def _fail(actual, row: Optional[int] = None) -> None:
2007+
where = f"Row {row}: " if row is not None else ""
2008+
raise ValueError(
2009+
f"{where}Vector length {actual} does not match expected {expected} "
2010+
f"for feature '{name}' in feature view '{feature_view.name}'."
2011+
)
2012+
2013+
if pyarrow.types.is_fixed_size_list(column_type):
2014+
if column_type.list_size != expected:
2015+
_fail(column_type.list_size)
2016+
return
2017+
2018+
if pyarrow.types.is_list(column_type) or pyarrow.types.is_large_list(column_type):
2019+
lengths = pyarrow.compute.list_value_length(column)
2020+
# Nulls have no length to compare; only non-null rows carry the contract.
2021+
mismatched = pyarrow.compute.and_kleene(
2022+
pyarrow.compute.is_valid(lengths),
2023+
pyarrow.compute.not_equal(lengths, expected),
2024+
)
2025+
if pyarrow.compute.any(mismatched).as_py():
2026+
row = pyarrow.compute.index(mismatched, True).as_py()
2027+
_fail(lengths[row].as_py(), row=row)
2028+
return
2029+
2030+
# Any other type cannot hold a vector of the declared width.
2031+
raise ValueError(
2032+
f"Vector feature '{name}' in feature view '{feature_view.name}' declares "
2033+
f"vector_length={expected} but has non-list Arrow type {column_type}."
2034+
)
2035+
2036+
19722037
def _distance_to_score(distance: float, metric: Optional[str] = None) -> float:
19732038
"""Convert a raw distance value into a higher-is-better relevance score.
19742039

‎sdk/python/tests/unit/test_vector_length_validation.py‎

Lines changed: 174 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,174 @@
1+
"""Tests for vector length validation on both the write and materialization paths.
2+
3+
Covers three fixes:
4+
* the vector field is resolved from the schema, not assumed to be ``features[0]``
5+
* the check is reachable from the materialization path via ``_convert_arrow_to_proto``
6+
* length checking is a single vectorized pass rather than a per-row loop
7+
"""
8+
9+
import numpy as np
10+
import pandas as pd
11+
import pyarrow
12+
import pytest
13+
14+
from feast import Entity, FeatureView, Field, FileSource
15+
from feast.feature_store import FeatureStore
16+
from feast.types import Array, Float32, String
17+
from feast.utils import _validate_vector_field_lengths
18+
from feast.value_type import ValueType
19+
20+
VECTOR_LENGTH = 4
21+
22+
23+
def _feature_view(vector_first: bool = True, vector_length: int = VECTOR_LENGTH):
24+
"""Build a feature view whose vector field is first or second in the schema."""
25+
vector = Field(
26+
name="embedding",
27+
dtype=Array(Float32),
28+
vector_index=True,
29+
vector_length=vector_length,
30+
vector_search_metric="COSINE",
31+
)
32+
label = Field(name="label", dtype=String)
33+
return FeatureView(
34+
name="fv",
35+
entities=[
36+
Entity(name="item_id", join_keys=["item_id"], value_type=ValueType.INT64)
37+
],
38+
schema=[vector, label] if vector_first else [label, vector],
39+
source=FileSource(
40+
path="/tmp/unused.parquet", timestamp_field="event_timestamp"
41+
),
42+
)
43+
44+
45+
def _fixed_size_list(dim: int, num_rows: int = 3) -> pyarrow.Array:
46+
values = pyarrow.array(np.zeros(dim * num_rows, dtype=np.float32))
47+
return pyarrow.FixedSizeListArray.from_arrays(values, dim)
48+
49+
50+
def _variable_list(lengths) -> pyarrow.Array:
51+
return pyarrow.array(
52+
[None if length is None else [0.0] * length for length in lengths],
53+
pyarrow.list_(pyarrow.float32()),
54+
)
55+
56+
57+
def _table(embedding: pyarrow.Array) -> pyarrow.Table:
58+
num_rows = len(embedding)
59+
return pyarrow.table(
60+
{
61+
"embedding": embedding,
62+
"label": pyarrow.array(["a"] * num_rows),
63+
"item_id": pyarrow.array(list(range(num_rows))),
64+
}
65+
)
66+
67+
68+
def _validate_df(feature_view, df: pd.DataFrame) -> None:
69+
"""Invoke the DataFrame validator without standing up a FeatureStore.
70+
71+
``_validate_vector_features`` does not touch instance state, so an
72+
uninitialized instance is enough.
73+
"""
74+
store = FeatureStore.__new__(FeatureStore)
75+
store._validate_vector_features(feature_view, df)
76+
77+
78+
class TestArrowVectorLengthValidation:
79+
"""``_validate_vector_field_lengths`` guards the materialization path."""
80+
81+
def test_fixed_size_list_matching_length_passes(self):
82+
_validate_vector_field_lengths(
83+
_table(_fixed_size_list(VECTOR_LENGTH)), _feature_view()
84+
)
85+
86+
def test_fixed_size_list_wrong_length_raises(self):
87+
with pytest.raises(ValueError, match="does not match expected 4"):
88+
_validate_vector_field_lengths(_table(_fixed_size_list(8)), _feature_view())
89+
90+
def test_variable_list_reports_first_offending_row(self):
91+
table = _table(_variable_list([4, 4, 7, 4]))
92+
with pytest.raises(ValueError, match="Row 2: Vector length 7"):
93+
_validate_vector_field_lengths(table, _feature_view())
94+
95+
def test_vector_field_not_first_is_still_validated(self):
96+
"""Regression: the old check read features[0] and skipped this case."""
97+
with pytest.raises(ValueError, match="does not match expected 4"):
98+
_validate_vector_field_lengths(
99+
_table(_fixed_size_list(8)), _feature_view(vector_first=False)
100+
)
101+
102+
def test_unset_vector_length_is_a_no_op(self):
103+
_validate_vector_field_lengths(
104+
_table(_fixed_size_list(8)), _feature_view(vector_length=0)
105+
)
106+
107+
def test_null_rows_are_tolerated(self):
108+
table = _table(_variable_list([4, None, 4]))
109+
_validate_vector_field_lengths(table, _feature_view())
110+
111+
def test_missing_column_is_a_no_op(self):
112+
table = pyarrow.table({"label": pyarrow.array(["a", "b"])})
113+
_validate_vector_field_lengths(table, _feature_view())
114+
115+
def test_non_list_arrow_type_raises(self):
116+
table = pyarrow.table(
117+
{
118+
"embedding": pyarrow.array([1.0, 2.0]),
119+
"label": pyarrow.array(["a", "b"]),
120+
}
121+
)
122+
with pytest.raises(ValueError, match="non-list Arrow type"):
123+
_validate_vector_field_lengths(table, _feature_view())
124+
125+
def test_record_batch_is_accepted(self):
126+
batch = _table(_fixed_size_list(8)).to_batches()[0]
127+
with pytest.raises(ValueError, match="does not match expected 4"):
128+
_validate_vector_field_lengths(batch, _feature_view())
129+
130+
131+
class TestDataFrameVectorLengthValidation:
132+
"""``_validate_vector_features`` guards the online write path."""
133+
134+
def test_matching_lengths_pass(self):
135+
df = pd.DataFrame({"embedding": [[0.0] * 4] * 3, "label": list("abc")})
136+
_validate_df(_feature_view(), df)
137+
138+
def test_wrong_length_reports_first_offending_row(self):
139+
df = pd.DataFrame(
140+
{"embedding": [[0.0] * 4, [0.0] * 9, [0.0] * 4], "label": list("abc")}
141+
)
142+
with pytest.raises(ValueError, match="Row 1: Vector length 9"):
143+
_validate_df(_feature_view(), df)
144+
145+
def test_non_sequence_value_reports_its_type(self):
146+
df = pd.DataFrame(
147+
{"embedding": [[0.0] * 4, 3.14, [0.0] * 4], "label": list("abc")}
148+
)
149+
with pytest.raises(ValueError, match="is not a sequence"):
150+
_validate_df(_feature_view(), df)
151+
152+
def test_vector_field_not_first_is_still_validated(self):
153+
"""Regression: the old check read features[0] and skipped this case."""
154+
df = pd.DataFrame(
155+
{"embedding": [[0.0] * 4, [0.0] * 9, [0.0] * 4], "label": list("abc")}
156+
)
157+
with pytest.raises(ValueError, match="Row 1: Vector length 9"):
158+
_validate_df(_feature_view(vector_first=False), df)
159+
160+
def test_unset_vector_length_is_a_no_op(self):
161+
df = pd.DataFrame({"embedding": [[0.0] * 9] * 3, "label": list("abc")})
162+
_validate_df(_feature_view(vector_length=0), df)
163+
164+
def test_missing_column_is_a_no_op(self):
165+
_validate_df(_feature_view(), pd.DataFrame({"label": list("abc")}))
166+
167+
def test_numpy_vectors_are_accepted(self):
168+
df = pd.DataFrame(
169+
{
170+
"embedding": [np.zeros(VECTOR_LENGTH, dtype=np.float32)] * 3,
171+
"label": list("abc"),
172+
}
173+
)
174+
_validate_df(_feature_view(), df)

0 commit comments

Comments
 (0)