@@ -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,
 
@@ -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
@@ -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.