@@ -3640,25 +3640,41 @@ def _validate_vector_features(self, feature_view, df: pd.DataFrame) -> None:
         expected = vector_field.vector_length
         column = df[name]
 
-        # Single pass over the column. Sequences report their length, anything else
-        # reports None so the two failure modes stay distinguishable.
-        lengths = column.map(lambda v: len(v) if hasattr(v, "__len__") else None)
+        # 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()
-        if not_a_sequence.any():
-            i = not_a_sequence.idxmax()
-            raise ValueError(
-                f"Row {i}: Vector feature '{name}' is not a sequence. Got: {type(column[i])}"
-            )
+        not_a_sequence = lengths.isna() & ~is_null
+        mismatched = lengths.notna() & (lengths != expected)
+        offending = not_a_sequence | mismatched
+        if not offending.any():
+            return
 
-        mismatched = lengths != expected
-        if mismatched.any():
-            i = mismatched.idxmax()
+        # 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 {i}: Vector length {lengths[i]} does not match expected {expected} "
-                f"for feature '{name}' in feature view '{feature_view.name}'."
+                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,
         feature_view_name: str,
 
@@ -502,6 +502,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
@@ -525,8 +528,6 @@ def _convert_arrow_fv_to_proto(
             f"Feature view '{feature_view.name}' has no batch_source and cannot be converted to proto."
         )
 
-    _validate_vector_field_lengths(table, feature_view)
-
     # TODO: This will break if the feature view has aggregations or transformations
     columns = [
         (field.name, field.dtype.to_value_type()) for field in feature_view.features
 
@@ -11,10 +11,10 @@
 import pyarrow
 import pytest
 
-from feast import Entity, FeatureView, Field, FileSource
+from feast import Entity, FeatureView, Field, FileSource, RequestSource
 from feast.feature_store import FeatureStore
 from feast.types import Array, Float32, String
-from feast.utils import _validate_vector_field_lengths
+from feast.utils import _convert_arrow_to_proto, _validate_vector_field_lengths
 from feast.value_type import ValueType
 
 VECTOR_LENGTH = 4
@@ -172,3 +172,81 @@ def test_numpy_vectors_are_accepted(self):
             }
         )
         _validate_df(_feature_view(), df)
+
+
+class TestReviewFeedback:
+    """Cases raised in review of #6909."""
+
+    def test_null_vectors_are_tolerated_on_the_dataframe_path(self):
+        """A genuine null has no length, and must not be called 'not a sequence'.
+
+        Matches the Arrow path, which already tolerates null rows.
+        """
+        df = pd.DataFrame(
+            {
+                "embedding": [[0.0] * VECTOR_LENGTH, None, np.nan],
+                "label": list("abc"),
+            }
+        )
+        _validate_df(_feature_view(), df)
+
+    def test_first_offending_row_wins_across_both_failure_modes(self):
+        """A bad length in row 0 outranks a non-sequence in row 1."""
+        df = pd.DataFrame(
+            {"embedding": [[0.0] * 9, 3.14, [0.0] * 4], "label": list("abc")}
+        )
+        with pytest.raises(ValueError, match="Row 0: Vector length 9"):
+            _validate_df(_feature_view(), df)
+
+    def test_non_sequence_wins_when_it_comes_first(self):
+        df = pd.DataFrame(
+            {"embedding": [3.14, [0.0] * 9, [0.0] * 4], "label": list("abc")}
+        )
+        with pytest.raises(
+            ValueError, match="Row 0: Vector feature .* is not a sequence"
+        ):
+            _validate_df(_feature_view(), df)
+
+    def test_duplicate_index_reports_a_scalar_type_not_a_series(self):
+        """Positional access keeps a duplicate index from yielding a Series."""
+        df = pd.DataFrame(
+            {"embedding": [[0.0] * 4, 3.14, [0.0] * 4], "label": list("abc")},
+            index=[7, 7, 7],
+        )
+        with pytest.raises(ValueError, match=r"Got: <class 'float'>"):
+            _validate_df(_feature_view(), df)
+
+    def test_on_demand_feature_view_output_is_validated(self):
+        """_convert_arrow_to_proto must validate before it branches to the ODFV path."""
+        from feast.on_demand_feature_view import on_demand_feature_view
+
+        request_source = RequestSource(
+            name="req",
+            schema=[Field(name="raw", dtype=String)],
+        )
+
+        @on_demand_feature_view(
+            sources=[request_source],
+            schema=[
+                Field(
+                    name="embedding",
+                    dtype=Array(Float32),
+                    vector_index=True,
+                    vector_length=VECTOR_LENGTH,
+                )
+            ],
+            mode="python",
+            write_to_online_store=True,
+        )
+        def odfv(inputs):
+            return {"embedding": inputs["raw"]}
+
+        # Width 8 against a declared 4 must be rejected on the ODFV branch too.
+        table = pyarrow.table(
+            {
+                "embedding": _fixed_size_list(8, num_rows=2),
+                "req": pyarrow.array(["a", "b"]),
+            }
+        )
+        with pytest.raises(ValueError, match="does not match expected 4"):
+            _convert_arrow_to_proto(table, odfv, {})