@@ -289,6 +289,70 @@ def to_arrow(
             _validate_vector_field_lengths(table, feature_view)
         return table
 
+    # --------------------------------------------------------------- writing
+
+    def get_write_target(self) -> Dict[str, Any]:
+        """Keyword arguments addressing this source for ``lance.write_dataset``.
+
+        Deliberately omits ``version``. A Lance commit always produces a new
+        version, so there is no argument that writes *into* an existing one;
+        ``write_dataset`` has no ``version`` parameter at all. Callers must
+        reject a pinned source before reaching a write, which
+        ``assert_writable`` does.
+        """
+        kwargs: Dict[str, Any] = {}
+        if self.storage_options:
+            kwargs["storage_options"] = dict(self.storage_options)
+
+        if self.is_catalog_based:
+            kwargs["namespace_client"] = self.get_namespace_client()
+            kwargs["table_id"] = self.table_id
+        else:
+            kwargs["uri"] = self.uri
+        return kwargs
+
+    def assert_writable(self) -> None:
+        """Refuse to write through a pinned source.
+
+        A write commits a new version, which a source pinned to an older version
+        or to a tag does not read. The write would therefore succeed and then be
+        invisible through the very source that performed it, so the only honest
+        outcome is to refuse up front.
+        """
+        if self.pin is None:
+            return
+        raise ValueError(
+            f"LanceSource '{self.name}' is pinned to {self.pin_description} and "
+            f"cannot be written through. A Lance write commits a new version, so "
+            f"it would not be visible through this pin. Write through an unpinned "
+            f"LanceSource addressing the same dataset, then move the tag or re-pin "
+            f"to the version the write produced."
+        )
+
+    def get_existing_schema(self) -> Optional["pyarrow.Schema"]:
+        """Schema of the dataset this source addresses, or ``None`` if absent.
+
+        Used to choose between creating and appending, and to compare shapes
+        before handing data to Lance. A failure to open is reported as absent
+        rather than raised, because the only distinction the caller draws is
+        whether there is an existing shape to honour.
+        """
+        import lance
+
+        kwargs = self.get_write_target()
+        try:
+            if self.is_catalog_based:
+                return lance.dataset(
+                    namespace_client=kwargs["namespace_client"],
+                    table_id=kwargs["table_id"],
+                    storage_options=kwargs.get("storage_options"),
+                ).schema
+            return lance.dataset(
+                uri=kwargs["uri"], storage_options=kwargs.get("storage_options")
+            ).schema
+        except Exception:
+            return None
+
     # ----------------------------------------------------------- DataSource
 
     def source_type(self) -> DataSourceProto.SourceType.ValueType:
@@ -469,3 +533,60 @@ def validate_lance_source_schema(feature_view: "FeatureView") -> None:
     from feast.utils import _validate_vector_field_lengths
 
     _validate_vector_field_lengths(schema.empty_table(), feature_view)
+
+
+def _vector_widths(schema: "pyarrow.Schema") -> Dict[str, int]:
+    """Width of every fixed-width vector column in an Arrow schema."""
+    import pyarrow
+
+    widths: Dict[str, int] = {}
+    for field in schema:
+        if pyarrow.types.is_fixed_size_list(field.type):
+            widths[field.name] = field.type.list_size
+    return widths
+
+
+def validate_lance_write_shape(
+    table: "pyarrow.Table",
+    source: LanceSource,
+    existing_schema: Optional["pyarrow.Schema"],
+) -> None:
+    """Refuse a write that would change an existing vector column's width.
+
+    Lance rejects an ``append`` whose schema disagrees with the dataset, but it
+    accepts an ``overwrite`` that replaces a vector column with one of a
+    different width, silently rewriting the declared shape and leaving a version
+    history whose vectors are not mutually comparable. Every ANN index and every
+    downstream consumer built against the old width is invalidated by such a
+    write, and no legitimate schema evolution changes an embedding's dimension,
+    so this is reported as an error rather than carried out.
+
+    Only vector widths are policed here. Other kinds of type change remain
+    Lance's business, because for those an ``overwrite`` is a defensible
+    evolution.
+    """
+    if existing_schema is None:
+        return
+
+    existing = _vector_widths(existing_schema)
+    incoming = _vector_widths(table.schema)
+
+    changed = {
+        name: (existing[name], incoming[name])
+        for name in existing.keys() & incoming.keys()
+        if existing[name] != incoming[name]
+    }
+    if not changed:
+        return
+
+    detail = ", ".join(
+        f"'{name}' is {was} in the dataset but {now} in the data being written"
+        for name, (was, now) in sorted(changed.items())
+    )
+    raise ValueError(
+        f"Write to Lance dataset '{source.name}' would change the width of "
+        f"vector column(s): {detail}. A width change invalidates every index and "
+        f"consumer built against the current width, and leaves a version history "
+        f"whose vectors cannot be compared with each other. Write the re-embedded "
+        f"data to a new dataset instead of overwriting this one."
+    )
@@ -53,6 +53,25 @@
 from feast.repo_config import FeastConfigBaseModel, RepoConfig
 
 
+def _as_lance_source(data_source: Optional[DataSource]) -> Optional["LanceSource"]:
+    """Narrow a DataSource to a LanceSource, or ``None`` if it is not one.
+
+    ``None`` is accepted so that a caller holding an optional source, such as
+    ``FeatureView.batch_source``, can ask the question without narrowing first.
+
+    The import is guarded because the Lance extra is optional: a repo that does
+    not install it must still be able to read and write every other source
+    type. Only the ``isinstance`` check is guarded, never the caller's
+    subsequent work, so an ImportError raised by a Lance read or write itself
+    surfaces instead of being mistaken for "not a Lance source".
+    """
+    try:
+        from feast.infra.data_sources.contrib.lance.lance_source import LanceSource
+    except ImportError:
+        return None
+    return data_source if isinstance(data_source, LanceSource) else None
+
+
 def _read_data_source(data_source: DataSource, repo_path: str) -> Table:
     from feast.infra.data_sources.contrib.iceberg_catalog.iceberg_source import (
         IcebergSource,
@@ -61,17 +80,7 @@ def _read_data_source(data_source: DataSource, repo_path: str) -> Table:
     if isinstance(data_source, IcebergSource):
         return _read_iceberg_catalog_source(data_source, repo_path)
 
-    lance_source: Optional["LanceSource"] = None
-    try:
-        from feast.infra.data_sources.contrib.lance.lance_source import LanceSource
-
-        if isinstance(data_source, LanceSource):
-            lance_source = data_source
-    except ImportError:
-        pass
-    # Deliberately outside the try: an ImportError raised by the read itself (a
-    # missing pylance, say) must surface rather than fall through to the
-    # FileSource assert below and be reported as the wrong source type.
+    lance_source = _as_lance_source(data_source)
     if lance_source is not None:
         return _read_lance_source(lance_source)
 
@@ -241,6 +250,58 @@ def _validate_lance_feature_views(feature_views: List[FeatureView]) -> None:
         validate_lance_source_schema(feature_view)
 
 
+def _write_lance_data_source(
+    table: Table,
+    data_source: "LanceSource",
+    mode: str = "append",
+    allow_overwrite: bool = False,
+) -> None:
+    """Write an ibis table to a LanceSource without Spark.
+
+    Mirrors ``_read_lance_source``: the source resolves either a uri or a
+    ``namespace_client`` + ``table_id``, and the same resolution is reused for
+    the write, so a catalog-addressed dataset is written to the location its
+    namespace resolves rather than to one assembled here.
+
+    How far the namespace is involved depends on the operation. Creating a table
+    declares it through the namespace; appending only describes the table to
+    resolve its location, and the version the append produces is not reported
+    back. A Lance namespace therefore records that a table exists and where it
+    lives, and does not track its versions. That is why a pin is a Lance-level
+    concern that ``assert_writable`` has to enforce here: no catalog is going to
+    reject the write on the pin's behalf.
+
+    Three things are settled before Lance is called. A pinned source is refused,
+    because a commit produces a new version that the pin would not read. An
+    absent dataset is created rather than appended to, since Lance has no
+    create-or-append mode. An existing dataset's vector widths are compared
+    against the incoming data, which closes the one case Lance itself permits
+    silently.
+    """
+    import lance
+
+    from feast.infra.data_sources.contrib.lance.lance_source import (
+        validate_lance_write_shape,
+    )
+
+    data_source.assert_writable()
+
+    arrow_table = table.to_pyarrow()
+    existing_schema = data_source.get_existing_schema()
+
+    if existing_schema is None:
+        lance_mode = "create"
+    else:
+        if mode == "overwrite" and not allow_overwrite:
+            raise SavedDatasetLocationAlreadyExists(
+                location=data_source.uri or ".".join(data_source.table_id)
+            )
+        validate_lance_write_shape(arrow_table, data_source, existing_schema)
+        lance_mode = mode
+
+    lance.write_dataset(arrow_table, mode=lance_mode, **data_source.get_write_target())
+
+
 def _write_data_source(
     table: Table,
     data_source: DataSource,
@@ -256,6 +317,11 @@ def _write_data_source(
         _write_iceberg_data_source(table, data_source, mode)
         return
 
+    lance_source = _as_lance_source(data_source)
+    if lance_source is not None:
+        _write_lance_data_source(table, lance_source, mode, allow_overwrite)
+        return
+
     assert isinstance(data_source, FileSource)
 
     file_options = data_source.file_options
@@ -735,6 +801,15 @@ def offline_write_batch(
         table: pyarrow.Table,
         progress: Optional[Callable[[int], Any]],
     ):
+        # Checked here rather than in the writer callback, which is handed a
+        # DataSource and so cannot see the declared schema. Creating a Lance
+        # dataset whose vectors are a different width from the declared
+        # vector_length would otherwise succeed and fail only on the next read.
+        if _as_lance_source(feature_view.batch_source) is not None:
+            from feast.utils import _validate_vector_field_lengths
+
+            _validate_vector_field_lengths(table, feature_view)
+
         offline_write_batch_ibis(
             config=config,
             feature_view=feature_view,