Visitar URL original
fix: Make vector length validation reachable, schema-driven and vectorized by haoxu0 · Pull Request #6909 · feast-dev/feast · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 48 additions & 14 deletions sdk/python/feast/feature_store.py
Original file line number Diff line number Diff line change
Expand Up @@ -3619,27 +3619,61 @@ def _validate_vector_features(self, feature_view, df: pd.DataFrame) -> None:
"""
Validates vector features in the DataFrame against the feature view specifications.

Resolves the vector field from the feature view schema rather than assuming it is
the first feature, and checks lengths in a single pass over the column.

Args:
feature_view: The feature view containing vector feature specifications
df: The DataFrame to validate

Raises:
ValueError: If vector dimension constraints are violated
"""
if feature_view.features and feature_view.features[0].vector_index:
fv_vector_feature_name = feature_view.features[0].name
if feature_view.features[0].vector_length != 0:
for i, row in df.iterrows():
vector = row[fv_vector_feature_name]
if not hasattr(vector, "__len__"):
raise ValueError(
f"Row {i}: Vector feature '{fv_vector_feature_name}' is not a sequence. Got: {type(vector)}"
)
if len(vector) != feature_view.features[0].vector_length:
raise ValueError(
f"Row {i}: Vector length {len(vector)} does not match expected {feature_view.features[0].vector_length} "
f"for feature '{fv_vector_feature_name}' in feature view '{feature_view.name}'."
)
vector_field = _get_feature_view_vector_field_metadata(feature_view)
if vector_field is None or not vector_field.vector_length:
return

name = vector_field.name
if name not in df.columns:
return

expected = vector_field.vector_length
column = df[name]

# Null vectors carry no length to compare. Skipping them keeps this
# consistent with the Arrow path, which also tolerates null rows, and
# stops a genuine null being reported as "not a sequence".
is_null = column.isna()

# Single pass over the column. na_action leaves nulls as NaN, so a NaN
# length that is not null means the value was not a sequence at all.
lengths = column.map(
lambda v: len(v) if hasattr(v, "__len__") else None, na_action="ignore"
)

not_a_sequence = lengths.isna() & ~is_null
mismatched = lengths.notna() & (lengths != expected)
offending = not_a_sequence | mismatched
if not offending.any():
return

# Report the first offending row across both failure modes, by position,
# so the message is right regardless of which failure comes first and
# correct even when the index has duplicates.
position = int(offending.to_numpy().argmax())
label = column.index[position]

if bool(not_a_sequence.to_numpy()[position]):
raise ValueError(
f"Row {label}: Vector feature '{name}' is not a sequence. "
f"Got: {type(column.iloc[position])}"
)

raise ValueError(
f"Row {label}: Vector length {int(lengths.iloc[position])} does not match "
f"expected {expected} for feature '{name}' in feature view "
f"'{feature_view.name}'."
)

def _get_feature_view_and_df_for_online_write(
self,
Expand Down
66 changes: 66 additions & 0 deletions sdk/python/feast/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -548,6 +548,9 @@ def _convert_arrow_to_proto(
) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]:
# This is a workaround for isinstance(feature_view, OnDemandFeatureView), which triggers a circular import
# Check for source_request_sources or source_feature_view_projections attributes to identify ODFVs
# Validate before branching so both regular and on-demand feature views are covered.
_validate_vector_field_lengths(table, feature_view)

if (
getattr(feature_view, "source_request_sources", None) is not None
or getattr(feature_view, "source_feature_view_projections", None) is not None
Expand Down Expand Up @@ -2015,6 +2018,69 @@ def _get_feature_view_vector_field_metadata(
return vector_fields[0]


def _validate_vector_field_lengths(
table: Union[pyarrow.Table, pyarrow.RecordBatch],
feature_view,
) -> None:
"""Check an Arrow table's vector column against the declared ``vector_length``.

Called on the materialization path, where every compute engine funnels through
``_convert_arrow_to_proto``. The check is O(1) for fixed-size lists and a single
vectorized pass for variable-size lists, so it is safe to leave on.

A declared ``vector_length`` is a contract: a mismatch is an error, never a
silent truncation or a silently different response shape.

Args:
table: The Arrow table or record batch about to be converted.
feature_view: The feature view whose schema declares the vector field.

Raises:
ValueError: If the vector column's width disagrees with ``vector_length``.
"""
vector_field = _get_feature_view_vector_field_metadata(feature_view)
if vector_field is None or not vector_field.vector_length:
return

name = vector_field.name
if name not in table.schema.names:
return

expected = vector_field.vector_length
column = table.column(name)
column_type = column.type

def _fail(actual, row: Optional[int] = None) -> None:
where = f"Row {row}: " if row is not None else ""
raise ValueError(
f"{where}Vector length {actual} does not match expected {expected} "
f"for feature '{name}' in feature view '{feature_view.name}'."
)

if pyarrow.types.is_fixed_size_list(column_type):
if column_type.list_size != expected:
_fail(column_type.list_size)
return

if pyarrow.types.is_list(column_type) or pyarrow.types.is_large_list(column_type):
lengths = pyarrow.compute.list_value_length(column)
# Nulls have no length to compare; only non-null rows carry the contract.
mismatched = pyarrow.compute.and_kleene(
pyarrow.compute.is_valid(lengths),
pyarrow.compute.not_equal(lengths, expected),
)
if pyarrow.compute.any(mismatched).as_py():
row = pyarrow.compute.index(mismatched, True).as_py()
_fail(lengths[row].as_py(), row=row)
return

# Any other type cannot hold a vector of the declared width.
raise ValueError(
f"Vector feature '{name}' in feature view '{feature_view.name}' declares "
f"vector_length={expected} but has non-list Arrow type {column_type}."
)


def _distance_to_score(distance: float, metric: Optional[str] = None) -> float:
"""Convert a raw distance value into a higher-is-better relevance score.

Expand Down
Loading
Loading