GeezerVison

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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:
Expand Down Expand Up		@@ -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."
		)

97 changes: 86 additions & 11 deletions 97 sdk/python/feast/infra/offline_stores/duckdb.py

Show comments

View file

Open in desktop

Original file line number	Diff line number	Diff line change
Expand Up		@@ -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,
Expand All		@@ -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)

Expand Down Expand Up		@@ -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()
Comment thread ntkathole Oct 9, 2026 Copy link Copy Markdown Member There was a problem hiding this comment. Choose a reason for hiding this comment The reason will be displayed to describe this comment to others. Learn more. `get_existing_schema()` opens the dataset for its schema, then `write_dataset` opens it again. Negligible for local, one extra round trip for a remote catalog. Fine as is. Sorry, something went wrong. Uh oh! There was an error while loading. Please reload this page. All reactions

		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,
Expand All		@@ -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
Expand Down Expand Up		@@ -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,
Expand Down

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Oops, something went wrong.

Uh oh!

There was an error while loading. Please reload this page.

Add this suggestion to a batch that can be applied as a single commit. This suggestion is invalid because no changes were made to the code. Suggestions cannot be applied while the pull request is closed. Suggestions cannot be applied while viewing a subset of changes. Only one suggestion per line can be applied in a batch. Add this suggestion to a batch that can be applied as a single commit. Applying suggestions on deleted lines is not supported. You must change the existing code in this line in order to create a valid suggestion. Outdated suggestions cannot be applied. This suggestion has been applied or marked resolved. Suggestions cannot be applied from pending reviews. Suggestions cannot be applied on multi-line comments. Suggestions cannot be applied while the pull request is queued to merge. Suggestion cannot be applied right now. Please check back later.

You are visitor 01402503 — GeezerVison does not store the pages you look up.
Best viewed with Netscape Navigator 4.7 or Internet Explorer 6 at 800×600. No cookies required.

GeezerVison

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

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

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

ntkathole Oct 9, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!