Repository: lincc-frameworks/nested-pandas

Issue:
.nest.to_flatten_inner fails for empty series
**Bug report**

When I tried to use it with `.map_partitions` it failed with the following error:

<details><summary>Expand</summary>

```
---------------------------------------------------------------------------
ValueError                                Traceback (most recent call last)
Cell In[5], line 6
      1 xmatch = vsx.crossmatch_nested(
      2     ztf_dr22,
      3     radius_arcsec=0.5,
      4     n_neighbors=20,
      5     nested_column_name="ztf",
----> 6 ).map_partitions(
      7     lambda df: df.assign("ztf_lc", df["ztf"].nest.to_flatten_inner("lc"))
      8 )

File ~/.virtualenvs/default/lib/python3.10/site-packages/lsdb/catalog/catalog.py:600, in Catalog.map_partitions(self, func, meta, include_pixel, *args, **kwargs)
    558 def map_partitions(
    559     self,
    560     func: Callable[..., npd.NestedFrame],
   (...)
    564     **kwargs,
    565 ) -> Catalog | dd.Series:
    566     """Applies a function to each partition in the catalog and respective margin.
    567 
    568     The ra and dec of each row is assumed to remain unchanged. If the function returns a DataFrame,
   (...)
    598         partition. If the function returns a non dataframe output, a dask Series will be returned.
    599     """
--> 600     catalog = super().map_partitions(
    601         func,
    602         *args,
    603         meta=meta,
    604         include_pixel=include_pixel,
    605         **kwargs,
    606     )
    607     if isinstance(catalog, Catalog) and self.margin is not None:
    608         catalog.margin = self.margin.map_partitions(
    609             func,
    610             *args,
   (...)
    613             **kwargs,
    614         )  # type: ignore[assignment]

File ~/.virtualenvs/default/lib/python3.10/site-packages/lsdb/catalog/dataset/healpix_dataset.py:513, in HealpixDataset.map_partitions(self, func, meta, include_pixel, *args, **kwargs)
    511     meta = func(self._ddf._meta.copy(), HealpixPixel(0, 0), *args, **kwargs)
    512 else:
--> 513     meta = func(self._ddf._meta.copy(), *args, **kwargs)
    514 if meta is None:
    515     raise ValueError(
    516         "func returned None for empty DataFrame input. The function must return a value, changing"
    517         " the partitions in place will not work. If the function does not work for empty inputs, "
    518         "please specify a `meta` argument."
    519     )

Cell In[5], line 7, in <lambda>(df)
      1 xmatch = vsx.crossmatch_nested(
      2     ztf_dr22,
      3     radius_arcsec=0.5,
      4     n_neighbors=20,
      5     nested_column_name="ztf",
      6 ).map_partitions(
----> 7     lambda df: df.assign("ztf_lc", df["ztf"].nest.to_flatten_inner("lc"))
      8 )

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/accessor.py:631, in NestSeriesAccessor.to_flatten_inner(self, field)
    629     if other_field == field:
    630         continue
--> 631     inner = inner.nest.with_filled_field(other_field, series.nest[other_field])
    632 # Repack flat inner back to nested series
    633 result = pack_flat(inner.nest.to_flat(), name=field)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/accessor.py:293, in NestSeriesAccessor.with_filled_field(self, field, value)
    258 """Set the field by repeating values and return a new series
    259 
    260 The input value array must have as many elements as the Series,
   (...)
    290 1  10.776335  67.046751    r  1
    291 """
    292 new_array = self._series.array.copy()
--> 293 new_array.fill_field_lists(field, value)
    294 return pd.Series(new_array, copy=False, index=self._series.index, name=self._series.name)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/ext_array.py:1068, in NestedExtensionArray.fill_field_lists(self, field, value, keep_dtype)
   1066 else:
   1067     value = np.repeat(value, self.list_lengths)
-> 1068 return self.set_flat_field(field, value, keep_dtype=keep_dtype)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/ext_array.py:981, in NestedExtensionArray.set_flat_field(self, field, value, keep_dtype)
    978     pa_array = pa_array.combine_chunks()
    979 field_list_array = pa.ListArray.from_arrays(values=pa_array, offsets=self.list_offsets)
--> 981 return self.set_list_field(field, field_list_array, keep_dtype=keep_dtype)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/ext_array.py:1032, in NestedExtensionArray.set_list_field(self, field, value, keep_dtype)
   1030     pa_table = self.pa_table.drop(field).add_column(field_idx, field, pa_array)
   1031 else:
-> 1032     pa_table = self.pa_table.append_column(field, pa_array)
   1033 self.pa_table = pa_table

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/ext_array.py:753, in NestedExtensionArray.pa_table(self)
    743 @property
    744 def pa_table(self) -> pa.Table:
    745     """Pyarrow table representation of the extension array.
    746 
    747     Returns
   (...)
    751         to a field of the struct array.
    752     """
--> 753     return self._table_storage.data

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/ext_array.py:716, in NestedExtensionArray._table_storage(self)
    714 @property
    715 def _table_storage(self) -> TableStorage:
--> 716     return TableStorage.from_list_struct_storage(self._list_storage)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/_storage/table_storage.py:50, in TableStorage.from_list_struct_storage(cls, list_storage)
     47 from nested_pandas.series._storage import StructListStorage
     49 struct_list_storage = StructListStorage.from_list_struct_storage(list_storage)
---> 50 return cls.from_struct_list_storage(struct_list_storage)

File ~/.virtualenvs/default/lib/python3.10/site-packages/nested_pandas/series/_storage/table_storage.py:61, in TableStorage.from_struct_list_storage(cls, struct_list_storage)
     52 @classmethod
     53 def from_struct_list_storage(cls, struct_list_storage: StructListStorage) -> Self:  # type: ignore # noqa: F821
     54     """Construct from a StructListStorage object.
     55 
     56     Parameters
   (...)
     59         StructListStorage object.
     60     """
---> 61     table = pa.Table.from_struct_array(struct_list_storage.data)
     62     return cls(table, validate=False)

File ~/.virtualenvs/default/lib/python3.10/site-packages/pyarrow/table.pxi:4901, in pyarrow.lib.Table.from_struct_array()

File ~/.virtualenvs/default/lib/python3.10/site-packages/pyarrow/table.pxi:4987, in pyarrow.lib.Table.from_batches()

ValueError: Must pass schema, or at least one RecordBatch
```

</details>

We need to fix storage-to-storage conversion to support empty data

**Before submitting**
Please check the following:

- [x] I have described the situation in which the bug arose, including what code was executed, information about my environment, and any applicable data others will need to reproduce the problem.
- [x] I have included available evidence of the unexpected behavior (including error messages, screenshots, and/or plots) as well as a description of what I expected instead.
- [x] If I have a solution in mind, I have provided an explanation and/or pseudocode and/or task list.

Discussion:

Solution patch:
diff --git a/src/nested_pandas/series/_storage/table_storage.py b/src/nested_pandas/series/_storage/table_storage.py
index 3577a07..1c9516c 100644
--- a/src/nested_pandas/series/_storage/table_storage.py
+++ b/src/nested_pandas/series/_storage/table_storage.py
@@ -4,7 +4,11 @@
 
 import pyarrow as pa
 
-from nested_pandas.series.utils import table_to_struct_array, validate_struct_list_array_for_equal_lengths
+from nested_pandas.series.utils import (
+    table_from_struct_array,
+    table_to_struct_array,
+    validate_struct_list_array_for_equal_lengths,
+)
 
 if TYPE_CHECKING:
     from nested_pandas.series._storage.list_struct_storage import ListStructStorage
@@ -58,5 +62,5 @@ def from_struct_list_storage(cls, struct_list_storage: StructListStorage) -> Sel
         struct_list_storage : StructListStorage
             StructListStorage object.
         """
-        table = pa.Table.from_struct_array(struct_list_storage.data)
+        table = table_from_struct_array(struct_list_storage.data)
         return cls(table, validate=False)
diff --git a/src/nested_pandas/series/accessor.py b/src/nested_pandas/series/accessor.py
index 47c458a..ff5df1e 100644
--- a/src/nested_pandas/series/accessor.py
+++ b/src/nested_pandas/series/accessor.py
@@ -1,7 +1,6 @@
 # Python 3.9 doesn't support "|" for types
 from __future__ import annotations
 
-from collections import defaultdict
 from collections.abc import Generator, Mapping
 from typing import cast
 
@@ -115,7 +114,7 @@ def to_flat(self, fields: list[str] | None = None) -> pd.DataFrame:
 
         index = pd.Series(self.get_flat_index(), name=self._series.index.name)
 
-        flat_chunks = defaultdict(list)
+        flat_chunks: dict[str, list[pa.Array]] = {field: [] for field in fields}
         for chunk in self._series.array.struct_array.iterchunks():
             struct_array = cast(pa.StructArray, chunk)
             for field in fields:
@@ -125,13 +124,16 @@ def to_flat(self, fields: list[str] | None = None) -> pd.DataFrame:
 
         flat_series = {}
         for field, chunks in flat_chunks.items():
+            dtype = self._series.dtype.field_dtype(field)
+            if len(chunks) == 0:
+                chunks = [pa.array([])]
             chunked_array = pa.chunked_array(chunks)
             flat_series[field] = pd.Series(
                 chunked_array,
                 index=index,
                 name=field,
                 copy=False,
-                dtype=self._series.dtype.field_dtype(field),
+                dtype=dtype,
             )
 
         return pd.DataFrame(flat_series)
diff --git a/src/nested_pandas/series/packer.py b/src/nested_pandas/series/packer.py
index 73dc58c..84aecdb 100644
--- a/src/nested_pandas/series/packer.py
+++ b/src/nested_pandas/series/packer.py
@@ -209,8 +209,8 @@ def pack_lists(df: pd.DataFrame, name: str | None = None, *, validate: bool = Tr
     chunk_lengths = pa.array([[len(chunk) for chunk in arr.chunks] for arr in pa_chunked_arrays.values()])
     if all(chunk_length == chunk_lengths[0] for chunk_length in chunk_lengths):
         chunks = []
-        numpy_chunks = next(iter(pa_chunked_arrays.values())).num_chunks
-        for i in range(numpy_chunks):
+        num_chunks = next(iter(pa_chunked_arrays.values())).num_chunks
+        for i in range(num_chunks):
             chunks.append(
                 pa.StructArray.from_arrays(
                     [arr.chunk(i) for arr in pa_chunked_arrays.values()],
diff --git a/src/nested_pandas/series/utils.py b/src/nested_pandas/series/utils.py
index 5548ed6..b8dd2a5 100644
--- a/src/nested_pandas/series/utils.py
+++ b/src/nested_pandas/series/utils.py
@@ -307,3 +307,10 @@ def table_to_struct_array(table: pa.Table) -> pa.ChunkedArray:
     if len(table) == 0:
         return pa.chunked_array([], type=pa.struct(table.schema))
     return table.to_struct_array()
+
+
+def table_from_struct_array(array: pa.ChunkedArray | pa.array) -> pa.Table:
+    """pa.Table.from_struct_array, but working with chunkless input"""
+    if isinstance(array, pa.ChunkedArray) and array.num_chunks == 0:
+        array = pa.array([], type=array.type)
+    return pa.Table.from_struct_array(array)


Test patch:
diff --git a/tests/nested_pandas/series/test_accessor.py b/tests/nested_pandas/series/test_accessor.py
index 30356b6..9f93e73 100644
--- a/tests/nested_pandas/series/test_accessor.py
+++ b/tests/nested_pandas/series/test_accessor.py
@@ -1098,3 +1098,23 @@ def test_to_flatten_outer_wrong_field():
     nf = generate_data(10, 2)
     with pytest.raises(ValueError):
         nf.nested.nest.to_flatten_inner("t")
+
+
+def test_issue266():
+    """Test .nest.to_flatten_inner() with empty series.
+
+    https://github.com/lincc-frameworks/nested-pandas/issues/266
+    """
+
+    nf = generate_data(10, 2)
+    nf = nf.assign(id=np.repeat(np.r_[0:5], 2))
+    nf = nf.rename(columns={"nested": "inner"})
+    nnf = NestedFrame.from_flat(nf, base_columns=[], on="id", name="outer")
+
+    empty_nnf = nnf.iloc[0:0]
+
+    empty_outer_flatten = empty_nnf["outer"].nest.to_flatten_inner("inner")
+
+    assert empty_outer_flatten.dtype == NestedDtype.from_fields(
+        {"t": pa.float64(), "flux": pa.float64(), "band": pa.string(), "a": pa.float64(), "b": pa.float64()}
+    )