Problem
PrunableStreamingTable::scan() only pushes the projection down to the Python partition factory when the projection contains at least one data variable. A projection made purely of dimension coordinates falls through to the non-pushdown branch, the factory is called with None, and every data variable in every scanned partition is loaded from storage before ProjectionExec throws it away.
src/lib.rs (main @ 5d7633e, lines 607 to 614):
// - projection contains only dimension columns (ds.to_dataframe()
// needs at least one data variable; dimensions are always loaded)
let push_projection = match projection {
Some(indices) if !indices.is_empty() => indices
.iter()
.any(|&i| !self.dimension_columns.contains(self.schema.field(i).name())),
_ => false,
};
The reason given in that comment no longer holds. The factory built by read_xarray_table already handles a coordinate-only projection explicitly:
if data_vars_needed:
ds_block = ds[data_vars_needed].isel(block)
else:
# Only dimension coords requested - drop all data vars to avoid
# loading them unnecessarily (e.g. for queries like SELECT lat, lon).
ds_block = ds.drop_vars(list(ds.data_vars)).isel(block)
and iter_record_batches has a matching if ds.data_vars: ... else: ... branch that takes its shape from ds.sizes when the Dataset has no data variables. So the Python side is ready for this case; only the Rust predicate keeps it unreachable.
Reproducer
Counts real data-variable chunk reads with a dask.array.map_blocks counter, and records the projection the factory receives.
import dask, dask.array as da, numpy as np, pandas as pd, xarray as xr
from datafusion import SessionContext
from xarray_sql import read_xarray_table
dask.config.set(scheduler="synchronous")
loads = {"n": 0}
def block(block_info=None):
loads["n"] += 1 # one call == one data-var chunk read
return np.zeros(block_info[None]["chunk-shape"], dtype="float32")
ds = xr.Dataset(
{
name: (("time", "lat", "lon"),
da.map_blocks(block, dtype="float32",
chunks=((10,) * 4, (20,), (30,))))
for name in ("air", "hum")
},
coords={"time": pd.date_range("2000-01-01", periods=40),
"lat": np.linspace(-90, 90, 20),
"lon": np.linspace(0, 359, 30)},
)
seen = []
ctx = SessionContext()
ctx.register_table("t", read_xarray_table(
ds, {"time": 10}, _iteration_callback=lambda blk, proj: seen.append(proj)))
for sql in ("SELECT COUNT(*) c FROM t",
"SELECT lat, air FROM t",
"SELECT DISTINCT lat FROM t",
"SELECT lat, lon FROM t"):
loads["n"], seen[:] = 0, []
ctx.sql(sql).to_pandas()
proj = str(seen[0]) if seen else "(no scan)"
print(f"{sql:28} projection={proj:22} chunks_loaded={loads['n']}")
Output on main (4 partitions, 2 data variables):
SELECT COUNT(*) c FROM t projection=(no scan) chunks_loaded=0
SELECT lat, air FROM t projection=['lat', 'air'] chunks_loaded=4
SELECT DISTINCT lat FROM t projection=None chunks_loaded=8
SELECT lat, lon FROM t projection=None chunks_loaded=8
SELECT lat, air is correct: pushdown happens and only air is read, 4 chunks. The two coordinate-only queries read 8 chunks, both data variables across all four partitions, for an answer that needs zero data-variable bytes.
Impact
Any query answered entirely from the coordinate axes still pays for the full data payload:
SELECT DISTINCT time FROM ds or SELECT DISTINCT level FROM ds to discover what an axis contains
SELECT lat, lon FROM ds WHERE ... to fetch a grid outline
COUNT(DISTINCT lat)
- any front end populating filter widgets from the dimension coordinates
On a local NetCDF this is a wasted copy. On a remote Zarr store it is a chunk fetch per data variable per partition. The waste also scales with the number of data variables in the table, which the same-dimensions grouping in from_dataset makes the common case.
Proposed fix
Push the projection whenever it is Some and non-empty, dropping the "contains at least one data variable" condition, and update the comment. The non-pushdown branch stays for None and Some([]).
I checked the Python half directly rather than assuming it: feeding iter_record_batches a block with drop_vars(list(ds.data_vars)) and a coords-only pa.schema returns the right row count and columns, so the projected path looks ready for this. I have not built the Rust change yet, so the predicate edit still needs compiling and testing.
Worth adding a test that asserts the factory receives the coordinate-only projection and that the resulting batches match the unprojected scan.
Parent: #126
Problem
PrunableStreamingTable::scan()only pushes the projection down to the Python partition factory when the projection contains at least one data variable. A projection made purely of dimension coordinates falls through to the non-pushdown branch, the factory is called withNone, and every data variable in every scanned partition is loaded from storage beforeProjectionExecthrows it away.src/lib.rs(main @ 5d7633e, lines 607 to 614):The reason given in that comment no longer holds. The factory built by
read_xarray_tablealready handles a coordinate-only projection explicitly:and
iter_record_batcheshas a matchingif ds.data_vars: ... else: ...branch that takes its shape fromds.sizeswhen the Dataset has no data variables. So the Python side is ready for this case; only the Rust predicate keeps it unreachable.Reproducer
Counts real data-variable chunk reads with a
dask.array.map_blockscounter, and records the projection the factory receives.Output on main (4 partitions, 2 data variables):
SELECT lat, airis correct: pushdown happens and onlyairis read, 4 chunks. The two coordinate-only queries read 8 chunks, both data variables across all four partitions, for an answer that needs zero data-variable bytes.Impact
Any query answered entirely from the coordinate axes still pays for the full data payload:
SELECT DISTINCT time FROM dsorSELECT DISTINCT level FROM dsto discover what an axis containsSELECT lat, lon FROM ds WHERE ...to fetch a grid outlineCOUNT(DISTINCT lat)On a local NetCDF this is a wasted copy. On a remote Zarr store it is a chunk fetch per data variable per partition. The waste also scales with the number of data variables in the table, which the same-dimensions grouping in
from_datasetmakes the common case.Proposed fix
Push the projection whenever it is
Someand non-empty, dropping the "contains at least one data variable" condition, and update the comment. The non-pushdown branch stays forNoneandSome([]).I checked the Python half directly rather than assuming it: feeding
iter_record_batchesa block withdrop_vars(list(ds.data_vars))and a coords-onlypa.schemareturns the right row count and columns, so the projected path looks ready for this. I have not built the Rust change yet, so the predicate edit still needs compiling and testing.Worth adding a test that asserts the factory receives the coordinate-only projection and that the resulting batches match the unprojected scan.
Parent: #126