@@ -45,7 +45,8 @@ def tracking_reader(ds, chunks=None, **kwargs):
     )
     ctx = XarrayContext()
 
-    ctx.from_dataset("forecast", ds, chunks={"lead_time": 2})
+    # A Dask-backed Dataset, so it is read lazily, and pruned.
+    ctx.from_dataset("forecast", ds.chunk(), chunks={"lead_time": 2})
     result = ctx.sql(
         "SELECT value FROM forecast "
         "WHERE lead_time >= INTERVAL '4 microseconds' ORDER BY value"
@@ -584,3 +585,65 @@ def test_single_table_all_coordless_dims(self):
         pd.testing.assert_frame_equal(
             result, expected, check_dtype=False, check_like=True
         )
+
+
+def _scans(ctx, table):
+    return ctx.sql(f"SELECT * FROM {table}").execution_plan().display_indent()
+
+
+def test_small_in_memory_dataset_is_a_native_table_with_the_same_rows():
+    """A small in-memory Dataset becomes a native in-memory table, and
+    queries over it give what a lazily read Dataset gives."""
+    ds = xr.Dataset(
+        {"air": (("time", "lat"), np.random.rand(6, 4))},
+        coords={"time": np.arange(6), "lat": np.arange(4) * 2.5},
+    )
+    ctx = XarrayContext()
+    ctx.from_dataset("eager", ds, chunks={"time": 2})
+    ctx.from_dataset("lazy", ds.chunk(), chunks={"time": 2})
+
+    assert "XarrayScanExec" not in _scans(ctx, "eager")
+    assert "XarrayScanExec" in _scans(ctx, "lazy")
+    query = "SELECT time, lat, air FROM {} WHERE lat > 2 ORDER BY time, lat"
+    pd.testing.assert_frame_equal(
+        ctx.sql(query.format("eager")).to_pandas(),
+        ctx.sql(query.format("lazy")).to_pandas(),
+    )
+    assert set(ctx.catalog().schema().names()) == {"eager", "lazy"}
+
+
+def test_mixed_dimension_in_memory_dataset_registers_native_tables():
+    """Each dimension group of a small in-memory Dataset is a native table
+    in the Dataset's schema, with nothing else left in the catalog."""
+    ds = xr.Dataset(
+        {
+            "surface": (("time", "lat"), np.random.rand(3, 2)),
+            "column": (("time", "lat", "level"), np.random.rand(3, 2, 4)),
+        },
+        coords={
+            "time": np.arange(3),
+            "lat": np.arange(2),
+            "level": np.arange(4),
+        },
+    )
+    ctx = XarrayContext()
+    ctx.from_dataset("era5", ds)
+
+    assert "XarrayScanExec" not in _scans(ctx, "era5.time_lat_level")
+    counts = ctx.sql(
+        "SELECT (SELECT COUNT(*) FROM era5.time_lat) AS surface, "
+        "(SELECT COUNT(*) FROM era5.time_lat_level) AS column_"
+    ).to_pandas()
+    assert counts.iloc[0].tolist() == [6, 24]
+    assert set(ctx.catalog().schema().names()) == set()
+
+
+def test_large_in_memory_dataset_is_read_lazily():
+    """An in-memory Dataset over 64 MiB as a table stays a lazy table, so
+    it isn't held twice."""
+    rows = 64 * 2**20 // 16 + 1  # x (int64) and value (float64): 16 bytes
+    ds = xr.Dataset({"value": (["x"], np.zeros(rows))})
+    ctx = XarrayContext()
+    ctx.from_dataset("big", ds)
+
+    assert "XarrayScanExec" in _scans(ctx, "big")
@@ -4,6 +4,10 @@
 so the exact statistics ``XarrayScanExec`` reports are visible to the query
 optimizer: num_rows (product of a chunk's dimension sizes), total byte size,
 and per dimension-column min/max bounds. These tests pin that behaviour.
+
+``XarrayScanExec`` scans a lazy table, so each test registers a Dask-backed
+Dataset (``ds.chunk()``): a small Dataset that is already in memory becomes
+a native DataFusion in-memory table instead.
 """
 
 import numpy as np
@@ -30,7 +34,7 @@ def test_exact_rows_in_scan_statistics():
         },
     )
     ctx = XarrayContext()
-    ctx.from_dataset("air", ds, chunks={"time": 50})
+    ctx.from_dataset("air", ds.chunk(), chunks={"time": 50})
     plan = _explain(ctx, "SELECT lat, lon, air FROM air")
     total = 100 * 4 * 5
     assert f"Rows=Exact({total})" in plan
@@ -47,7 +51,7 @@ def test_exact_byte_size_in_scan_statistics():
         },
     )
     ctx = XarrayContext()
-    ctx.from_dataset("air", ds, chunks={"time": 50})
+    ctx.from_dataset("air", ds.chunk(), chunks={"time": 50})
     plan = _explain(ctx, "SELECT lat, lon, air FROM air")
     # 2000 rows x (lat int64 + lon int64 + air float64) = 2000 x 24 bytes.
     assert f"Bytes=Exact({100 * 4 * 5 * 24})" in plan
@@ -69,7 +73,7 @@ def test_dimension_column_min_max_in_scan_statistics():
         },
     )
     ctx = XarrayContext()
-    ctx.from_dataset("air", ds, chunks={"time": 50})
+    ctx.from_dataset("air", ds.chunk(), chunks={"time": 50})
     plan = _explain(ctx, "SELECT lat, lon, air FROM air")
     # lat spans 0..3, lon spans 0..4, both never null.
     assert "Min=Exact(Int64(0)) Max=Exact(Int64(3)) Null=Exact(0)" in plan
@@ -101,7 +105,7 @@ def test_coordinate_statistics_preserve_logical_type(coord, expected):
         coords={"x": coord},
     )
     ctx = XarrayContext()
-    ctx.from_dataset("values", ds, chunks={"x": 2})
+    ctx.from_dataset("values", ds.chunk(), chunks={"x": 2})
 
     plan = _explain(ctx, "SELECT x, value FROM values")
 
@@ -114,7 +118,7 @@ def test_missing_partition_bound_removes_global_coordinate_statistics():
         coords={"x": np.asarray([0.0, 1.0, np.nan, 3.0])},
     )
     ctx = XarrayContext()
-    ctx.from_dataset("values", ds, chunks={"x": 2})
+    ctx.from_dataset("values", ds.chunk(), chunks={"x": 2})
 
     plan = _explain(ctx, "SELECT x, value FROM values")
 
@@ -134,7 +138,7 @@ def test_count_star_answered_from_statistics():
         },
     )
     ctx = XarrayContext()
-    ctx.from_dataset("air", ds, chunks={"time": 50})
+    ctx.from_dataset("air", ds.chunk(), chunks={"time": 50})
     n = ctx.sql("SELECT COUNT(*) AS n FROM air").to_pandas()["n"][0]
     assert int(n) == 100 * 4 * 5
 
@@ -159,8 +163,8 @@ def test_join_picks_small_build_side():
         coords={"lat": np.arange(8), "lon": np.arange(8)},
     )
     ctx = XarrayContext()
-    ctx.from_dataset("big", big, chunks={"time": 50})
-    ctx.from_dataset("small", small, chunks={"lat": 8})
+    ctx.from_dataset("big", big.chunk(), chunks={"time": 50})
+    ctx.from_dataset("small", small.chunk(), chunks={"lat": 8})
 
     plan = _explain(
         ctx,
 
@@ -193,6 +193,59 @@ def read_xarray(ds: xr.Dataset, chunks: Chunks = None) -> pa.RecordBatchReader:
     return pa.RecordBatchReader.from_stream(reader)
 
 
+# The largest Dataset, in bytes of long-form Arrow, that Context.from_dataset
+# registers as a native DataFusion in-memory table rather than a lazy one.
+_IN_MEMORY_TABLE_LIMIT: int = 64 * 2**20
+
+
+def _arrow_size(schema: pa.Schema, num_rows: int) -> int:
+    """An estimate of a table's size in bytes, counting 16 bytes for each
+    value of a variable-width column."""
+    widths: list[int] = [
+        field.type.bit_width // 8 if pa.types.is_primitive(field.type) else 16
+        for field in schema
+    ]
+    return num_rows * sum(widths)
+
+
+def _in_memory_partitions(
+    ds: xr.Dataset,
+    chunks: Chunks = None,
+    *,
+    limit: int = _IN_MEMORY_TABLE_LIMIT,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> list[list[pa.RecordBatch]] | None:
+    """A small in-memory Dataset's rows as Arrow partitions, or ``None``.
+
+    A lazy table pays a fixed cost on every scan, through DataFusion's FFI,
+    of about a millisecond and a half. That is noise for a large Dataset,
+    but a query that reads small tables many times pays it each time:
+    DataFusion inlines a CTE or a view at every place that reads it. A
+    Dataset that is already in memory gains nothing from being read lazily,
+    so when it is also at most ``limit`` bytes as long-form Arrow, its rows
+    are converted once, one partition per chunk, for a native in-memory
+    table.
+
+    Returns ``None`` for a Dataset that isn't in memory (a Dask array or a
+    lazily loaded file), is larger than ``limit``, or has no rows; it is
+    read lazily instead.
+    """
+    # `_in_memory` is xarray's own test for a variable whose values are
+    # loaded: a NumPy array, not a Dask array or a lazily indexed file.
+    if not all(var._in_memory for var in ds.variables.values()):
+        return None
+    ds = _ensure_default_indexes(ds)
+    schema = _parse_schema(ds)
+    num_rows = int(np.prod([ds.sizes[d] for d in ds.dims]))
+    if num_rows == 0 or _arrow_size(schema, num_rows) > limit:
+        return None
+    resolved = resolve_chunks(ds, chunks)
+    return [
+        list(iter_record_batches(ds.isel(block), schema, batch_size))
+        for block in _block_slices_from_resolved(ds, resolved)
+    ]
+
+
 def read_xarray_table(
     ds: xr.Dataset,
     chunks: Chunks = None,
 
@@ -1,3 +1,4 @@
+import pyarrow as pa
 import xarray as xr
 from datafusion import SessionContext
 from datafusion.catalog import Schema
@@ -12,7 +13,7 @@
     shared_coord_arrays,
 )
 from .ds import XarrayDataFrame
-from .reader import read_xarray_table
+from .reader import _in_memory_partitions, read_xarray_table
 
 _proj: ModuleType | None
 try:  # pyproj is an optional dependency (`pip install xarray-sql[geo]`).
@@ -74,6 +75,14 @@ def from_dataset(
             )
             ctx.sql('SELECT * FROM era5.surface')
 
+        A Dataset is read lazily, chunk by chunk, as queries run, unless it
+        is already in memory (NumPy arrays, not Dask or a lazily opened
+        file) and at most 64 MiB as a table. Such a Dataset is converted
+        to Arrow once, here, and registered as a native DataFusion
+        in-memory table, which is faster to scan many times, as a query
+        does when it reads a table in several places. It is held twice,
+        as arrays and as Arrow.
+
         For datasets with non-Gregorian cftime coordinates (e.g. 360_day,
         julian), a ``cftime()`` scalar UDF is automatically registered so
         you can write ergonomic SQL filters::
@@ -158,13 +167,28 @@ def _from_dataset(
         register = (
             self.register_table if schema is None else schema.register_table
         )
-        register(
-            table_name,
-            read_xarray_table(input_table, chunks, coord_arrays=coord_arrays),
-        )
+        partitions = _in_memory_partitions(input_table, chunks)
+        if partitions is None:
+            table = read_xarray_table(
+                input_table, chunks, coord_arrays=coord_arrays
+            )
+        else:
+            table = self._in_memory_table(partitions)
+        register(table_name, table)
         self._maybe_register_cftime_udf(input_table)
         return self
 
+    def _in_memory_table(self, partitions: list[list[pa.RecordBatch]]):
+        """A native DataFusion in-memory table of ``partitions``."""
+        before = set(self.catalog().schema().names())
+        view = self.create_dataframe(partitions).into_view()
+        # create_dataframe registers its in-memory table under a generated
+        # name. The view holds the table itself, not the name, so the name
+        # can go.
+        for name in set(self.catalog().schema().names()) - before:
+            self.deregister_table(name)
+        return view
+
     def _maybe_register_cftime_udf(self, ds: xr.Dataset) -> None:
         """Auto-register a cftime() UDF for non-Gregorian cftime coordinates."""
         for coord_name in ds.dims: