fix: no panic on empty partition in register_record_batches (#1627)

* fix: no panic on empty partition in register_record_batches

`register_record_batches` read the schema via unchecked `partitions.0[0][0]`,
which panics when a partition contains no record batches (e.g. the batches of
an empty pyarrow table, whose `to_batches()` returns `[]`).

Take the schema from the first available batch across all partitions and return
a clear error when there is none, mirroring the empty-table handling added for
`from_arrow_table` in #613. A non-empty later partition now also works instead
of panicking on an empty leading one.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* refactor: address review feedback

- Use find_map(|p| p.first()) instead of flatten().next() to make the
  "first batch in any partition" intent clearer (per @kosiew).
- Extend regression test to also assert the empty outer partition list
  case ([]) raises the same ValueError.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/core/src/context.rs

diff --git a/crates/core/src/context.rs b/crates/core/src/context.rs
index 0613a96..6184c7c 100644
--- a/crates/core/src/context.rs
+++ b/crates/core/src/context.rs
@@ -832,7 +832,19 @@
         name: &str,
         partitions: PyArrowType<Vec<Vec<RecordBatch>>>,
     ) -> PyDataFusionResult<()> {
-        let schema = partitions.0[0][0].schema();
+        // Take the schema from the first available batch; error instead of
+        // panicking when the partitions hold no batches.
+        let schema = partitions
+            .0
+            .iter()
+            .find_map(|partition| partition.first())
+            .ok_or_else(|| {
+                PyValueError::new_err(
+                    "Cannot register record batches without a schema: the \
+                     provided partitions contain no record batches.",
+                )
+            })?
+            .schema();
         let table = MemTable::try_new(schema, partitions.0)?;
         self.ctx.register_table(name, Arc::new(table))?;
         Ok(())

python/tests/test_context.py

diff --git a/python/tests/test_context.py b/python/tests/test_context.py
index 112a6fd..2f068b0 100644
--- a/python/tests/test_context.py
+++ b/python/tests/test_context.py
@@ -116,6 +116,22 @@
     assert result[0].column(1) == pa.array([-3, -3, -3])
 
 
+def test_register_record_batches_empty(ctx):
+    # A partition list with no record batches carries no schema, so this used to
+    # panic on unchecked `[0][0]` indexing. It should now raise a clear error.
+    with pytest.raises(ValueError, match="no record batches"):
+        ctx.register_record_batches("t", [[]])
+
+    # An empty outer partition list carries no schema either, and raises the same error.
+    with pytest.raises(ValueError, match="no record batches"):
+        ctx.register_record_batches("t", [])
+
+    # The schema is still recovered from a later non-empty partition.
+    batch = pa.RecordBatch.from_arrays([pa.array([1, 2, 3])], names=["a"])
+    ctx.register_record_batches("t2", [[], [batch]])
+    assert ctx.sql("SELECT a FROM t2").collect()[0].column(0) == pa.array([1, 2, 3])
+
+
 def test_create_dataframe_registers_unique_table_name(ctx):
     # create a RecordBatch and register it as memtable
     batch = pa.RecordBatch.from_arrays(