diff --git a/xarray_sql/df.py b/xarray_sql/df.py
index af25c35a..2e941b88 100644
--- a/xarray_sql/df.py
+++ b/xarray_sql/df.py
@@ -5,8 +5,9 @@
 import pandas as pd
 import pyarrow as pa
 import xarray as xr
+from datafusion.context import ArrowStreamExportable
 
-Block = t.Dict[str, slice]
+Block = t.Dict[t.Hashable, slice]
 Chunks = t.Optional[t.Dict[str, int]]
 
 
@@ -54,7 +55,42 @@ def explode(ds: xr.Dataset, chunks: Chunks = None) -> t.Iterator[xr.Dataset]:
 
 
 def _block_len(block: Block) -> int:
-  return np.prod([v.stop - v.start for v in block.values()])
+  return int(np.prod([v.stop - v.start for v in block.values()]))
+
+
+def from_map_batched(
+    func: t.Callable[[...], pd.DataFrame],
+    *iterables,
+    args: t.Optional[t.Tuple] = None,
+    schema: pa.Schema = None,
+    **kwargs,
+) -> pa.RecordBatchReader:
+  """Create a PyArrow RecordBatchReader by mapping a function over iterables.
+
+  This is equivalent to dask's from_map but returns a PyArrow
+  RecordBatchReader that can be used with DataFusion. It iterates over
+  RecordBatches which are created via the `func` one-at-a-time.
+
+  Args:
+    func: Function to apply to each element of the iterables. Currently, the function
+      must return a Pandas DataFrame.
+    *iterables: Iterable objects to map the function over.
+    schema: Optional schema needed for the RecordBatchReader.
+    args: Additional positional arguments to pass to func.
+    **kwargs: Additional keyword arguments to pass to func.
+
+  Returns:
+    A PyArrow RecordBatchReader containing the stream of RecordBatches.
+  """
+  if args is None:
+    args = ()
+
+  def map_batches():
+    for items in zip(*iterables):
+      df = func(*items, *args, **kwargs)
+      yield pa.RecordBatch.from_pandas(df, schema=schema)
+
+  return pa.RecordBatchReader.from_batches(schema, map_batches())
 
 
 def from_map(
@@ -109,7 +145,12 @@ def from_map(
   return pa.concat_tables(results)
 
 
-def read_xarray(ds: xr.Dataset, chunks: Chunks = None) -> pa.Table:
+def pivot(ds: xr.Dataset) -> pd.DataFrame:
+  """Converts an xarray Dataset to a pandas DataFrame."""
+  return ds.to_dataframe().reset_index()
+
+
+def read_xarray(ds: xr.Dataset, chunks: Chunks = None) -> ArrowStreamExportable:
   """Pivots an Xarray Dataset into a PyArrow Table, partitioned by chunks.
 
   Args:
@@ -128,7 +169,11 @@ def read_xarray(ds: xr.Dataset, chunks: Chunks = None) -> pa.Table:
 
   blocks = list(block_slices(ds, chunks))
 
-  def pivot(b: Block) -> pd.DataFrame:
-    return ds.isel(b).to_dataframe().reset_index()
+  def pivot_block(b: Block):
+    return pivot(ds.isel(b))
+
+  schema = pa.Schema.from_pandas(pivot_block(blocks[0]))
+  last_schema = pa.Schema.from_pandas(pivot_block(blocks[-1]))
+  assert schema == last_schema, "Schemas must be consistent across blocks!"
 
-  return from_map(pivot, blocks)
+  return from_map_batched(pivot_block, blocks, schema=schema)
diff --git a/xarray_sql/df_test.py b/xarray_sql/df_test.py
index fedef88c..403f8155 100644
--- a/xarray_sql/df_test.py
+++ b/xarray_sql/df_test.py
@@ -1,4 +1,5 @@
 import itertools
+import tracemalloc
 import unittest
 
 import numpy as np
@@ -6,7 +7,7 @@
 import pyarrow as pa
 import xarray as xr
 
-from .df import explode, read_xarray, block_slices, from_map
+from .df import explode, read_xarray, block_slices, from_map, pivot, from_map_batched
 
 
 def rand_wx(start: str, end: str) -> xr.Dataset:
@@ -34,6 +35,38 @@ def rand_wx(start: str, end: str) -> xr.Dataset:
   )
 
 
+def create_large_dataset(time_steps=1000, lat_points=100, lon_points=100):
+  """Create a large xarray dataset for memory testing."""
+  np.random.seed(42)
+
+  # Create coordinates
+  time = pd.date_range('2020-01-01', periods=time_steps, freq='h')
+  lat = np.linspace(-90, 90, lat_points)
+  lon = np.linspace(-180, 180, lon_points)
+
+  # Create large data arrays
+  temp_data = np.random.rand(time_steps, lat_points, lon_points) * 40 - 10
+  precip_data = np.random.rand(time_steps, lat_points, lon_points) * 100
+
+  return xr.Dataset(
+      {
+          'temperature': (['time', 'lat', 'lon'], temp_data),
+          'precipitation': (['time', 'lat', 'lon'], precip_data),
+      },
+      coords={
+          'time': time,
+          'lat': lat,
+          'lon': lon,
+      },
+  )
+
+
+def adding_function(x, y):
+  """Simple function that adds two values and returns a DataFrame."""
+  result = pd.DataFrame({'x': [x], 'y': [y], 'sum': [x + y]})
+  return result
+
+
 class DaskTestCase(unittest.TestCase):
 
   def setUp(self) -> None:
@@ -81,49 +114,6 @@ def test_data_equal__one__last(self):
     self.assertEqual(self.air.isel(iselection), ds)
 
 
-class PyArrowTableTest(DaskTestCase):
-
-  def test_sanity(self):
-    table = read_xarray(self.air_small)
-    self.assertIsNotNone(table)
-    self.assertIsInstance(table, pa.Table)
-    self.assertEqual(len(table), np.prod(list(self.air_small.sizes.values())))
-
-  def test_columns(self):
-    table = read_xarray(self.air_small)
-    cols = table.column_names
-    self.assertEqual(cols, ['lat', 'time', 'lon', 'air'])
-
-  def test_dtypes(self):
-    table = read_xarray(self.air_small)
-    # Convert to pandas to check dtypes
-    df = table.to_pandas()
-    types = list(df.dtypes)
-    self.assertEqual([self.air_small[c].dtype for c in df.columns], types)
-
-  def test_different_chunk_sizes(self):
-    default_table = read_xarray(self.air_small)
-    chunked_table = read_xarray(self.air_small, dict(time=5))
-
-    # Both should produce valid tables
-    self.assertIsInstance(default_table, pa.Table)
-    self.assertIsInstance(chunked_table, pa.Table)
-    # Should have same number of rows
-    self.assertEqual(len(default_table), len(chunked_table))
-
-  def test_chunk_perf(self):
-    table = read_xarray(self.air, chunks=dict(time=6))
-    self.assertIsNotNone(table)
-    self.assertEqual(len(table), np.prod(list(self.air.sizes.values())))
-
-  def test_column_metadata_preserved(self):
-    try:
-      table = read_xarray(self.randwx, chunks=dict(time=24))
-      self.assertIsInstance(table, pa.Table)
-    except Exception as e:
-      self.fail(f'Unexpected error: {e}')
-
-
 class FromMapTest(unittest.TestCase):
 
   def test_basic_from_map(self):
@@ -178,5 +168,220 @@ def make_arrow_table(x):
     self.assertEqual(len(result), 3)
 
 
+class TestFromMapBatchedCorrectness(DaskTestCase):
+  """Test correctness of from_map_batched function."""
+
+  def test_basic_functionality(self):
+    """Test that from_map_batched produces correct RecordBatchReader."""
+    blocks = list(
+        block_slices(self.air_small, chunks={'time': 4, 'lat': 3, 'lon': 4})
+    )
+
+    # Get expected schema
+    first_block_df = pivot(self.air_small.isel(blocks[0]))
+    expected_schema = pa.Schema.from_pandas(first_block_df)
+
+    # Create RecordBatchReader
+    reader = from_map_batched(
+        pivot,
+        [self.air_small.isel(block) for block in blocks],
+        schema=expected_schema,
+    )
+
+    # Verify it's a RecordBatchReader
+    self.assertIsInstance(reader, pa.RecordBatchReader)
+
+    # Verify schema
+    self.assertEqual(reader.schema, expected_schema)
+
+    # Read all batches and verify content
+    batches = list(reader)
+    self.assertGreater(len(batches), 0)
+
+    # Verify each batch has the correct schema
+    for batch in batches:
+      self.assertEqual(batch.schema, expected_schema)
+      self.assertGreater(len(batch), 0)
+
+  def test_multiple_iterables(self):
+    """Test from_map_batched with multiple iterables."""
+    x_values = [1, 2, 3, 4, 5]
+    y_values = [10, 20, 30, 40, 50]
+
+    expected_schema = pa.schema(
+        [('x', pa.int64()), ('y', pa.int64()), ('sum', pa.int64())]
+    )
+
+    reader = from_map_batched(
+        adding_function, x_values, y_values, schema=expected_schema
+    )
+
+    # Read all data
+    table = reader.read_all()
+    df = table.to_pandas()
+
+    # Verify results
+    expected_df = pd.DataFrame(
+        {
+            'x': x_values,
+            'y': y_values,
+            'sum': [x + y for x, y in zip(x_values, y_values)],
+        }
+    )
+
+    pd.testing.assert_frame_equal(df, expected_df)
+
+  def test_with_args_and_kwargs(self):
+    """Test from_map_batched with additional args and kwargs."""
+
+    def multiply_and_add(x, multiplier, offset=0):
+      result = pd.DataFrame({'x': [x], 'result': [x * multiplier + offset]})
+      return result
+
+    values = [1, 2, 3]
+    expected_schema = pa.schema([('x', pa.int64()), ('result', pa.int64())])
+
+    reader = from_map_batched(
+        multiply_and_add,
+        values,
+        args=(2,),  # multiplier = 2
+        offset=5,  # offset = 5
+        schema=expected_schema,
+    )
+
+    table = reader.read_all()
+    df = table.to_pandas()
+
+    # Verify results: x * 2 + 5
+    expected_df = pd.DataFrame(
+        {'x': [1, 2, 3], 'result': [7, 9, 11]}  # (1*2+5, 2*2+5, 3*2+5)
+    )
+
+    pd.testing.assert_frame_equal(df, expected_df)
+
+  def test_empty_iterables(self):
+    """Test from_map_batched with empty iterables."""
+    empty_schema = pa.schema([('value', pa.int64())])
+
+    reader = from_map_batched(
+        lambda x: pd.DataFrame({'value': [x]}), [], schema=empty_schema
+    )
+
+    batches = list(reader)
+    self.assertEqual(len(batches), 0)
+
+  def test_consistency_with_regular_map(self):
+    """Test that results are consistent with regular mapping."""
+    blocks = list(block_slices(self.air_small, chunks={'time': 4, 'lat': 3}))
+    datasets = [self.air_small.isel(block) for block in blocks]
+
+    # Get schema from first block
+    first_df = pivot(datasets[0])
+    schema = pa.Schema.from_pandas(first_df)
+
+    # Use from_map_batched
+    reader = from_map_batched(pivot, datasets, schema=schema)
+    batched_table = reader.read_all()
+
+    # Regular map approach
+    regular_dfs = [pivot(ds) for ds in datasets]
+    regular_table = pa.Table.from_pandas(
+        pd.concat(regular_dfs, ignore_index=True)
+    )
+
+    # Results should be identical
+    self.assertEqual(batched_table.schema, regular_table.schema)
+    self.assertEqual(len(batched_table), len(regular_table))
+
+    # Compare data (allowing for potential column order differences)
+    batched_df = (
+        batched_table.to_pandas()
+        .sort_values(['time', 'lat', 'lon'])
+        .reset_index(drop=True)
+    )
+    regular_df = (
+        regular_table.to_pandas()
+        .sort_values(['time', 'lat', 'lon'])
+        .reset_index(drop=True)
+    )
+
+    pd.testing.assert_frame_equal(batched_df, regular_df)
+
+  def test_integration_with_datafusion_via_read_xarray(self):
+    """Test integration with the read_xarray function that uses from_map_batched."""
+
+    # Use a small dataset
+    air = xr.tutorial.open_dataset('air_temperature')
+    air_small = air.isel(time=slice(0, 50), lat=slice(0, 10), lon=slice(0, 15))
+    air_chunked = air_small.chunk({'time': 25, 'lat': 5, 'lon': 8})
+
+    # read_xarray uses from_map_batched internally
+    arrow_stream = read_xarray(
+        air_chunked, chunks={'time': 25, 'lat': 5, 'lon': 8}
+    )
+
+    # Verify it returns a proper ArrowStreamExportable (RecordBatchReader)
+    self.assertTrue(hasattr(arrow_stream, 'schema'))
+    self.assertTrue(hasattr(arrow_stream, '__iter__'))
+
+    # Should be able to read all data
+    table = arrow_stream.read_all()
+    self.assertGreater(len(table), 0)
+
+    # Should have the expected columns
+    expected_columns = {'time', 'lat', 'lon', 'air'}
+    actual_columns = set(table.column_names)
+    self.assertTrue(expected_columns.issubset(actual_columns))
+
+
+class ReadXarrayStreamingTest(unittest.TestCase):
+
+  def setUp(self):
+    self.large_ds = create_large_dataset().chunk({'time': 25})
+
+  def test_read_xarray__loads_one_chunk_at_a_time(self):
+    tracemalloc.start()
+    iterable = read_xarray(self.large_ds)
+    first_size, first_peak = tracemalloc.get_traced_memory()
+    tracemalloc.reset_peak()
+
+    sizes, peaks = [], []
+
+    first_chunk = self.large_ds.isel(next(block_slices(self.large_ds)))
+    chunk_size = first_chunk.nbytes
+
+    # Creating the iterator should be inexpensive -- less than one chunk.
+    # We multiply by constant factors because chunks have additional overhead
+    self.assertLess(first_size, chunk_size * 3)
+    self.assertLess(first_peak, chunk_size * 6)
+
+    for it in iterable:
+      _ = it
+      cur_size, cur_peak = tracemalloc.get_traced_memory()
+      tracemalloc.reset_peak()
+      sizes.append(cur_size), peaks.append(cur_peak)
+
+    mean_size = np.mean(sizes)
+    mean_peak = np.mean(peaks)
+
+    # Provide a bound for each size and peak
+    for size in sizes:
+      self.assertGreater(mean_size * 1.1, size)
+      self.assertGreater(chunk_size * 3, size)
+      # malloc size is about 2.66x chunk_size on average
+      self.assertLess(chunk_size * 2, size)
+
+    for peak in peaks:
+      self.assertGreater(mean_peak * 1.1, peak)
+      self.assertGreater(chunk_size * 7, peak)
+      # malloc peak is about 6.89x chunk_size on average
+      self.assertLess(chunk_size * 4, peak)
+
+    # The peak malloc should never be more than the original dataset!
+    self.assertLess(max(peaks), self.large_ds.nbytes)
+
+    tracemalloc.stop()
+
+
 if __name__ == '__main__':
   unittest.main()
diff --git a/xarray_sql/sql_test.py b/xarray_sql/sql_test.py
index f550f290..041f1320 100644
--- a/xarray_sql/sql_test.py
+++ b/xarray_sql/sql_test.py
@@ -23,13 +23,13 @@ def test_agg_small(self):
 
     query = c.sql(
         """
-    SELECT
-      "lat", "lon", SUM("air") as air_total
-    FROM 
-      "air" 
-    GROUP BY
-     "lat", "lon"
-    """
+  SELECT
+    "lat", "lon", SUM("air") as air_total
+  FROM 
+    "air" 
+  GROUP BY
+   "lat", "lon"
+  """
     )
 
     result = query.to_pandas()