[BUG] Handle tasks that return 0 row TableChunk messages in `fanout_node_unbounded`
- Dominant language
- C++
- Stars
- 9.8k
- Forks
- 1.1k
- Avg merge
- 3d 6m
- Merged PRs (30d)
- 278
Description
```python
In [1]: from functools import partialmethod
...: import polars
...: from cudf_polars.utils.config import Runtime, StreamingFallbackMode
...: executor = "streaming"
...: executor_options: dict[str, Any] = {}
...: executor_options["max_rows_per_partition"] = 4
...: executor_options["target_partition_size"] = 10
...: # We expect many tests to fall back, so silence the warnings
...: executor_options["fallback_mode"] = StreamingFallbackMode.SILENT
...: executor_options["runtime"] = Runtime.RAPIDSMPF
...: collect = polars.LazyFrame.collect
...: engine = polars.GPUEngine(executor=executor, executor_options=executor_options)
...: polars.LazyFrame.collect = partialmethod(collect, engine=engine)
...:
...: import polars as pl
...: query="""
...: WITH data AS (
...: SELECT 'xyz' AS id, [0,1,2] AS items
...: UNION ALL
...: SELECT 'abc', [3,4]
...: )
...: SELECT id, item
...: FROM data CROSS JOIN UNNEST(items) AS item
...: ORDER BY item
...: """
...: with pl.SQLContext(frames={}, eager=True) as ctx:
...: polars_result = ctx.execute(query=query, eager=True)
```
results in
```python
File /cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/nodes.py:484, in fanout_node_unbounded(context, ch_in, trace_ir, ir_context, *chs_out)
481 if idx < num_outputs - 1:
482 # Copy to target memory and insert into spillable buffer
--> 483 mid = sm.insert(msg.copy(memory_reservation))
484 else:
485 # Optimization: reuse the original message for last output
486 # (no copy needed)
487 mid = sm.insert(msg)
File rapidsmpf/streaming/core/message.pyx:146
--> 146 'Could not get source, probably due dynamically evaluated source code.'
ValueError: RapidsMPF fatal error at: /tmp/conda-bld-output/bld/rattler-build_librapidsmpf/work/cpp/src/streaming/cudf/table_chunk.cpp:52: packed data cannot be empty
```
Confirmed that the message is not empty though
```python
> /cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/nodes.py(483)fanout_node_unbounded()
-> breakpoint()
(Pdb) msg.empty()
False
```
Contributor guide
Assessment
This issue has not been assessed yet.