@@ -31,6 +31,10 @@ lance_format = LanceFormat(
 )
 ```
 
+The DuckDB offline store reads a `LanceSource` as a lazy scan, so only the
+columns a query selects are read from the dataset. Materializing a few scalar
+features does not load the embedding columns stored beside them.
+
 ### Apache Iceberg
 
 [Apache Iceberg](https://iceberg.apache.org/) is an open table format designed for huge analytic datasets. It provides:
 
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 from datetime import date, datetime, timezone
@@ -193,8 +194,24 @@ def _read_lance_source(data_source: "LanceSource") -> Table:
     is honoured before any data crosses into ibis. The declared-schema contract
     is enforced separately, by ``validate_lance_source_schema``, because the
     reader callback is handed a ``DataSource`` without its feature view.
+
+    The dataset is registered with DuckDB as a lazy scan rather than read into
+    memory. A ``LanceDataset`` is a ``pyarrow.dataset.Dataset``, so DuckDB pushes
+    the query's column projection into the Lance scanner: a retrieval that needs
+    a few scalar features never reads the embedding columns stored beside them.
+
+    The view is named after the dataset's location and resolved version. An
+    opened dataset is a snapshot, so the name always denotes the same data and
+    re-registering it is harmless, while a job built before a later write keeps
+    reading the version it was built against.
     """
-    return ibis.memtable(data_source.to_arrow())
+    dataset = data_source.get_dataset()
+    digest = hashlib.sha256(f"{dataset.uri}@{dataset.version}".encode()).hexdigest()
+    view_name = f"__feast_lance_{digest[:16]}"
+
+    backend = ibis.get_backend()
+    backend.con.register(view_name, dataset)
+    return backend.table(view_name)
 
 
 def _validate_lance_feature_views(feature_views: List[FeatureView]) -> None:
 
@@ -592,6 +592,94 @@ def test_pull_latest_from_a_lance_source(tmp_path):
     assert job.to_arrow().num_rows == 3
 
 
+def _driver_stats_with_embeddings(n_rows=3, width=8):
+    stats = _driver_stats(n_rows=n_rows)
+    embeddings = pa.FixedSizeListArray.from_arrays(
+        pa.array([0.5] * (n_rows * width), pa.float32()), width
+    )
+    return stats.append_column("embedding", embeddings)
+
+
+@pytest.fixture
+def scanned_columns(monkeypatch):
+    """Record the column projection of every scan Lance is asked to run."""
+    scans = []
+    original = lance.LanceDataset.scanner
+
+    def recording_scanner(self, *args, **kwargs):
+        scans.append(kwargs.get("columns"))
+        return original(self, *args, **kwargs)
+
+    monkeypatch.setattr(lance.LanceDataset, "scanner", recording_scanner)
+    return scans
+
+
+def test_the_duckdb_reader_is_lazy(tmp_path, scanned_columns):
+    uri = str(tmp_path / "stats.lance")
+    _write(uri, _driver_stats())
+    table = _read_data_source(
+        LanceSource(uri=uri, timestamp_field="event_timestamp"), str(tmp_path)
+    )
+    assert scanned_columns == []
+    assert table.to_pyarrow().num_rows == 3
+    assert scanned_columns
+
+
+def test_pull_latest_does_not_scan_columns_it_does_not_select(
+    tmp_path, scanned_columns
+):
+    uri = str(tmp_path / "stats.lance")
+    _write(uri, _driver_stats_with_embeddings())
+    job = DuckDBOfflineStore.pull_latest_from_table_or_query(
+        config=_repo_config(tmp_path),
+        data_source=LanceSource(uri=uri, timestamp_field="event_timestamp"),
+        join_key_columns=["driver_id"],
+        feature_name_columns=["conv_rate"],
+        timestamp_field="event_timestamp",
+        created_timestamp_column=None,
+        start_date=datetime(2026, 1, 1),
+        end_date=datetime(2026, 1, 2),
+    )
+    result = job.to_arrow()
+    assert result.num_rows == 3
+    assert "embedding" not in result.column_names
+    assert scanned_columns
+    assert all(
+        columns is not None and "embedding" not in columns
+        for columns in scanned_columns
+    )
+
+
+def test_historical_retrieval_does_not_scan_columns_it_does_not_select(
+    tmp_path, scanned_columns
+):
+    uri = str(tmp_path / "stats.lance")
+    _write(uri, _driver_stats_with_embeddings())
+    source = LanceSource(uri=uri, timestamp_field="event_timestamp")
+    df = _historical(tmp_path, source, entity_hour=13)
+    # driver 1000 is the first row of the fixture.
+    assert df["conv_rate"].iloc[0] == pytest.approx(0.1)
+    assert scanned_columns
+    assert all(
+        columns is not None and "embedding" not in columns
+        for columns in scanned_columns
+    )
+
+
+def test_a_job_reads_the_version_it_was_built_against(tmp_path):
+    """A lazy scan must not pick up a write that lands before it executes."""
+    uri = str(tmp_path / "stats.lance")
+    _write(uri, _driver_stats(n_rows=3))
+    source = LanceSource(uri=uri, timestamp_field="event_timestamp")
+    before = _read_data_source(source, str(tmp_path))
+
+    _write(uri, _driver_stats(n_rows=2, base_hour=12), mode="append")
+    after = _read_data_source(source, str(tmp_path))
+
+    assert before.to_pyarrow().num_rows == 3
+    assert after.to_pyarrow().num_rows == 5
+
+
 def _pit_fixture(tmp_path, namespace=None):
     """One driver with two feature rows, an hour apart.
Original file line number	Diff line number	Diff line change
`@@ -31,6 +31,10 @@ lance_format = LanceFormat(`
`31`	`31`	`)`
`32`	`32`	```
`33`	`33`
	`34`	+The DuckDB offline store reads a `LanceSource` as a lazy scan, so only the
	`35`	`+columns a query selects are read from the dataset. Materializing a few scalar`
	`36`	`+features does not load the embedding columns stored beside them.`
	`37`	`+`
`34`	`38`	`### Apache Iceberg`
`35`	`39`
`36`	`40`	`[Apache Iceberg](https://iceberg.apache.org/) is an open table format designed for huge analytic datasets. It provides:`