fix: Make vector length validation reachable, schema-driven and vectorized - #6909

Three findings from review of feast-dev#6909.

Tolerate genuine nulls on the DataFrame path (@ntkathole). lengths.isna() was
true both for values that were not sequences and for rows that were genuinely
null, so a null vector was reported as "not a sequence". Nulls are now skipped,
which also makes this path consistent with the Arrow path, which already
tolerated null rows.

Report the first offending row across both failure modes (Copilot). Checking
the not-a-sequence mask fully before the length mask meant a non-sequence in a
later row was reported ahead of a wrong length in an earlier one. The masks are
now unioned and the first offending position selected before choosing the
message, which restores the row ordering the original iterrows loop had. Value
lookup is positional so a DataFrame with a duplicate index reports a scalar
type rather than a Series.

Validate the on-demand branch too (Copilot). _convert_arrow_to_proto dispatches
to _convert_arrow_odfv_to_proto or _convert_arrow_fv_to_proto, and the check sat
in the regular branch only, so a stored ODFV declaring vector_length had its
transformed output unchecked. The call moved up into the dispatcher, ahead of
the branch, so both paths are covered.

Each finding has a test that fails against the previous commit and passes here.

Signed-off-by: hao-xu5 <hxu44@apple.com>

Adds LanceSource and teaches the DuckDB offline store to read it, completing the
read half of feast-dev#6899. LanceFormat landed in feast-dev#6925 as a format descriptor; nothing
read Lance until now.

Lance already worked through SparkSource, which drives its reader generically
from table_format.format_type.value and table_format.properties. What was
missing is a path that needs no JVM, which is also the real test of whether the
DataSource abstraction is engine-agnostic rather than Spark-agnostic in name
only.

Placement: a new source read by the existing DuckDB store, rather than a Lance
offline store or an extension of FileSource.

FileSource is the wrong host. Its format axis is already taken by file_format,
so adding table_format would give one source two overlapping format axes. It is
also read by two stores with incompatible contracts: duckdb._read_data_source
dispatches on type, while dask._read_datasource has no dispatch seam and reads
file_options.uri unconditionally as Parquet, and asserts isinstance(...,
FileSource) in three places. Decisively, Lance's catalog addressing has no path
to put in FileSource.path, so the catalog-based layer would not fit the class
even if the path-based one did.

A Lance offline store would be the wrong 120 lines. duckdb.py is a binding that
injects reader and writer callbacks into the engine in ibis.py, so a Lance store
would be a near-copy of it plus a repo_config entry, and would force a choice
between Lance and Parquet instead of mixing them in one feature service.

Reading a source as ibis.memtable(arrow_table) in the DuckDB store already has
two precedents, IcebergSource and MlflowDatasetSource. Following them leaves
ibis.py untouched, so the point-in-time join, TTL handling, field mapping and
ODFVs work unchanged, and no edit to repo_config.py or data_source.py is needed
because CUSTOM_SOURCE plus data_source_class_type is self-describing.

Both addressing modes work: a uri, and catalog/namespace/table through
namespace_client and table_id.

Pin semantics follow what was argued on feast-dev#5782 and feast-dev#6925: a pin selects data,
never shape. get_table_column_names_and_types reads the pinned schema so
feast apply infers what reads will actually see; a pre-flight check fails with a
message naming the pin when a pinned version cannot satisfy the declared schema;
and vector widths go through _validate_vector_field_lengths from feast-dev#6909 rather
than a second validator.

Tests use the dir namespace implementation, which exercises the same
namespace_client and table_id code path as a remote catalog with no server
required. 43 tests, including a demonstration that a tag pin returns earlier
data after the dataset has been overwritten for the same entity and timestamp.

Read-only for now: _write_data_source is untouched, so a LanceSource is not yet
a persist target and there is no SavedDatasetLanceStorage.

Signed-off-by: hao-xu5 <hxu44@apple.com>

* feat: Add a non-JVM read path for Lance data sources

Adds LanceSource and teaches the DuckDB offline store to read it, completing the
read half of #6899. LanceFormat landed in #6925 as a format descriptor; nothing
read Lance until now.

Lance already worked through SparkSource, which drives its reader generically
from table_format.format_type.value and table_format.properties. What was
missing is a path that needs no JVM, which is also the real test of whether the
DataSource abstraction is engine-agnostic rather than Spark-agnostic in name
only.

Placement: a new source read by the existing DuckDB store, rather than a Lance
offline store or an extension of FileSource.

FileSource is the wrong host. Its format axis is already taken by file_format,
so adding table_format would give one source two overlapping format axes. It is
also read by two stores with incompatible contracts: duckdb._read_data_source
dispatches on type, while dask._read_datasource has no dispatch seam and reads
file_options.uri unconditionally as Parquet, and asserts isinstance(...,
FileSource) in three places. Decisively, Lance's catalog addressing has no path
to put in FileSource.path, so the catalog-based layer would not fit the class
even if the path-based one did.

A Lance offline store would be the wrong 120 lines. duckdb.py is a binding that
injects reader and writer callbacks into the engine in ibis.py, so a Lance store
would be a near-copy of it plus a repo_config entry, and would force a choice
between Lance and Parquet instead of mixing them in one feature service.

Reading a source as ibis.memtable(arrow_table) in the DuckDB store already has
two precedents, IcebergSource and MlflowDatasetSource. Following them leaves
ibis.py untouched, so the point-in-time join, TTL handling, field mapping and
ODFVs work unchanged, and no edit to repo_config.py or data_source.py is needed
because CUSTOM_SOURCE plus data_source_class_type is self-describing.

Both addressing modes work: a uri, and catalog/namespace/table through
namespace_client and table_id.

Pin semantics follow what was argued on #5782 and #6925: a pin selects data,
never shape. get_table_column_names_and_types reads the pinned schema so
feast apply infers what reads will actually see; a pre-flight check fails with a
message naming the pin when a pinned version cannot satisfy the declared schema;
and vector widths go through _validate_vector_field_lengths from #6909 rather
than a second validator.

Tests use the dir namespace implementation, which exercises the same
namespace_client and table_id code path as a remote catalog with no server
required. 43 tests, including a demonstration that a tag pin returns earlier
data after the dataset has been overwritten for the same entity and timestamp.

Read-only for now: _write_data_source is untouched, so a LanceSource is not yet
a persist target and there is no SavedDatasetLanceStorage.

Signed-off-by: hao-xu5 <hxu44@apple.com>

* fix: Address Lance read review feedback

Signed-off-by: HaoXuAI <sduxuhao@gmail.com>

---------

Signed-off-by: hao-xu5 <hxu44@apple.com>
Signed-off-by: HaoXuAI <sduxuhao@gmail.com>

feast-dev#6943 made a `LanceSource` readable without Spark. This makes one writable,
through the same two addressing modes: a uri, or a `namespace_client` plus
`table_id` resolved through a Lance namespace, so a catalog-addressed
dataset is committed through its catalog rather than behind its back.

`_write_data_source` gains a `LanceSource` branch, following the
`IcebergSource` branch already there, which makes the path reachable from
`DuckDBOfflineStore.offline_write_batch`. The `isinstance` narrowing the
read path open-coded is now a shared `_as_lance_source` helper used by
both, rather than a second copy of the optional-import guard.

Three things are settled before Lance is called.

A pinned source is refused. A Lance commit always produces a new version,
so `write_dataset` has no `version` argument at all; a write through a
source pinned to version 1 or to a tag would succeed and then be invisible
through the very source that performed it. Measured: after appending to a
dataset tagged `prod` at version 1, the unpinned source reads 5 rows and
the pinned source still reads 3.

An absent dataset is created rather than appended to, because Lance has no
create-or-append mode and rejects `create` on an existing dataset.

An existing dataset's vector widths are compared against the incoming
data. Lance rejects an `append` whose schema disagrees, but it accepts an
`overwrite` that replaces a 8-wide embedding column with a 32-wide one,
silently rewriting the declared shape and leaving a version history whose
vectors are not mutually comparable, with every index built on the old
width invalidated. No legitimate schema evolution changes an embedding's
dimension, so this is refused. The guard is narrow on purpose: only vector
widths are policed, and other type changes remain Lance's business.

The declared `vector_length` is checked at the store entry point rather
than in the writer callback, which is handed a `DataSource` and so cannot
see the feature view. It reuses `_validate_vector_field_lengths` from
 feast-dev#6909 rather than adding a second implementation. Without it, creating a
dataset at a width other than the declared one would succeed and fail only
on the next read.

19 tests added, all using the `dir` namespace implementation for the
catalog-based cases so no server is needed.

Signed-off-by: hao-xu5 <hxu44@apple.com>

feast-dev#6943 made a `LanceSource` readable without Spark. This makes one writable,
through the same two addressing modes: a uri, or a `namespace_client` plus
`table_id` resolved through a Lance namespace, so a catalog-addressed
dataset is committed through its catalog rather than behind its back.

`_write_data_source` gains a `LanceSource` branch, following the
`IcebergSource` branch already there, which makes the path reachable from
`DuckDBOfflineStore.offline_write_batch`. The `isinstance` narrowing the
read path open-coded is now a shared `_as_lance_source` helper used by
both, rather than a second copy of the optional-import guard.

Three things are settled before Lance is called.

A pinned source is refused. A Lance commit always produces a new version,
so `write_dataset` has no `version` argument at all; a write through a
source pinned to version 1 or to a tag would succeed and then be invisible
through the very source that performed it. Measured: after appending to a
dataset tagged `prod` at version 1, the unpinned source reads 5 rows and
the pinned source still reads 3.

An absent dataset is created rather than appended to, because Lance has no
create-or-append mode and rejects `create` on an existing dataset.

An existing dataset's vector widths are compared against the incoming
data. Lance rejects an `append` whose schema disagrees, but it accepts an
`overwrite` that replaces a 8-wide embedding column with a 32-wide one,
silently rewriting the declared shape and leaving a version history whose
vectors are not mutually comparable, with every index built on the old
width invalidated. No legitimate schema evolution changes an embedding's
dimension, so this is refused. The guard is narrow on purpose: only vector
widths are policed, and other type changes remain Lance's business.

The declared `vector_length` is checked at the store entry point rather
than in the writer callback, which is handed a `DataSource` and so cannot
see the feature view. It reuses `_validate_vector_field_lengths` from
 feast-dev#6909 rather than adding a second implementation. Without it, creating a
dataset at a width other than the declared one would succeed and fail only
on the next read.

19 tests added, all using the `dir` namespace implementation for the
catalog-based cases so no server is needed.

Signed-off-by: hao-xu5 <hxu44@apple.com>

* feat: Add a non-JVM write path for Lance data sources

#6943 made a `LanceSource` readable without Spark. This makes one writable,
through the same two addressing modes: a uri, or a `namespace_client` plus
`table_id` resolved through a Lance namespace, so a catalog-addressed
dataset is committed through its catalog rather than behind its back.

`_write_data_source` gains a `LanceSource` branch, following the
`IcebergSource` branch already there, which makes the path reachable from
`DuckDBOfflineStore.offline_write_batch`. The `isinstance` narrowing the
read path open-coded is now a shared `_as_lance_source` helper used by
both, rather than a second copy of the optional-import guard.

Three things are settled before Lance is called.

A pinned source is refused. A Lance commit always produces a new version,
so `write_dataset` has no `version` argument at all; a write through a
source pinned to version 1 or to a tag would succeed and then be invisible
through the very source that performed it. Measured: after appending to a
dataset tagged `prod` at version 1, the unpinned source reads 5 rows and
the pinned source still reads 3.

An absent dataset is created rather than appended to, because Lance has no
create-or-append mode and rejects `create` on an existing dataset.

An existing dataset's vector widths are compared against the incoming
data. Lance rejects an `append` whose schema disagrees, but it accepts an
`overwrite` that replaces a 8-wide embedding column with a 32-wide one,
silently rewriting the declared shape and leaving a version history whose
vectors are not mutually comparable, with every index built on the old
width invalidated. No legitimate schema evolution changes an embedding's
dimension, so this is refused. The guard is narrow on purpose: only vector
widths are policed, and other type changes remain Lance's business.

The declared `vector_length` is checked at the store entry point rather
than in the writer callback, which is handed a `DataSource` and so cannot
see the feature view. It reuses `_validate_vector_field_lengths` from
 #6909 rather than adding a second implementation. Without it, creating a
dataset at a width other than the declared one would succeed and fail only
on the next read.

19 tests added, all using the `dir` namespace implementation for the
catalog-based cases so no server is needed.

Signed-off-by: hao-xu5 <hxu44@apple.com>

* fix: Correct what a Lance namespace does during a write

The docstring claimed a catalog-addressed write is "committed through its
namespace". That is only true of a create. Instrumenting the namespace
client shows how far it is actually involved:

  create -> declare_table
  append -> describe_table only
  open   -> describe_table, namespace_id

So a namespace records that a table exists and where it lives, and does
not track its versions: the version an append produces is never reported
back to it. Worth stating precisely, because it is the reason the pin
contract has to be enforced in `assert_writable` rather than left to the
catalog -- no catalog is going to reject a write on a pin's behalf.

Adds a test that records the namespace calls, so the claim is pinned by a
test rather than asserted in prose.

Signed-off-by: hao-xu5 <hxu44@apple.com>

* fix: Accept an optional source in _as_lance_source

offline_write_batch passes FeatureView.batch_source, which is optional, so
the narrow DataSource annotation made mypy reject the call. The body already
handled None, since isinstance(None, LanceSource) is False, so only the
signature changes.

Signed-off-by: hao-xu5 <hxu44@apple.com>

---------

Signed-off-by: hao-xu5 <hxu44@apple.com>