[BUG] `.join(on=[...], how="full")` on empty LazyFrame does not preserve 0 length result with rapidsmpf
- Dominant language
- C++
- Stars
- 9.8k
- Forks
- 1.1k
- Avg merge
- 3d 6m
- Merged PRs (30d)
- 278
Description
```python
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"] = 5
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
df = pl.LazyFrame({"a": [5, 6], "b": [1, 2]})
empty = pl.LazyFrame(schema=df.collect_schema())
df.join(empty, on=["a", "b"], how="full", coalesce=True).collect()
```
raises
```python
+ Exception Group Traceback (most recent call last):
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/site-packages/IPython/core/interactiveshell.py", line 3747, in run_code
| exec(code_obj, self.user_global_ns, self.user_ns)
| ~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
| File "", line 18, in
| df.join(empty, on=["a", "b"], how="full", coalesce=True).collect()
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/site-packages/polars/_utils/deprecation.py", line 97, in wrapper
| return function(*args, **kwargs)
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/site-packages/polars/lazyframe/opt_flags.py", line 326, in wrapper
| return function(*args, **kwargs)
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/site-packages/polars/lazyframe/frame.py", line 2440, in collect
| return wrap_df(ldf.collect(engine, callback))
| ~~~~~~~~~~~^^^^^^^^^^^^^^^^^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/site-packages/polars/_utils/scan.py", line 27, in _execute_from_rust
| return function(with_columns, *args)
| File "/cudf/python/cudf_polars/cudf_polars/callback.py", line 323, in _callback
| return evaluate_streaming(ir, config_options)
| File "/cudf/python/cudf_polars/cudf_polars/experimental/parallel.py", line 276, in evaluate_streaming
| return evaluate_rapidsmpf(ir, config_options)
| File "/cudf/python/cudf_polars/cudf_polars/experimental/parallel.py", line 247, in evaluate_rapidsmpf
| result, _ = evaluate_logical_plan(ir, config_options, collect_metadata=False)
| ~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py", line 196, in evaluate_logical_plan
| result, metadata_collector = evaluate_pipeline(
| ~~~~~~~~~~~~~~~~~^
| ir,
| ^^^
| ...<6 lines>...
| query_id=query_id,
| ^^^^^^^^^^^^^^^^^^
| )
| ^
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/core.py", line 342, in evaluate_pipeline
| run_actor_network(actors=nodes, py_executor=executor)
| ~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
| File "rapidsmpf/streaming/core/actor.pyx", line 277, in rapidsmpf.streaming.core.actor.run_actor_network
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/concurrent/futures/_base.py", line 450, in result
| return self.__get_result()
| ~~~~~~~~~~~~~~~~~^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/concurrent/futures/_base.py", line 395, in __get_result
| raise self._exception
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/concurrent/futures/thread.py", line 86, in run
| result = ctx.run(self.task)
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/concurrent/futures/thread.py", line 73, in run
| return fn(*args, **kwargs)
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/runners.py", line 204, in run
| return runner.run(main)
| ~~~~~~~~~~^^^^^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/runners.py", line 127, in run
| return self._loop.run_until_complete(task)
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/base_events.py", line 719, in run_until_complete
| return future.result()
| ~~~~~~~~~~~~~^^
| File "rapidsmpf/streaming/core/actor.pyx", line 144, in run_py_actors
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/tasks.py", line 721, in _wrap_awaitable
| return await awaitable
| ^^^^^^^^^^^^^^^
| File "rapidsmpf/streaming/core/actor.pyx", line 109, in run
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/join.py", line 1057, in join_actor
| await gather_in_task_group(*actor_tasks)
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py", line 93, in gather_in_task_group
| async with asyncio.TaskGroup() as tg:
| ~~~~~~~~~~~~~~~~~^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/taskgroups.py", line 72, in __aexit__
| return await self._aexit(et, exc)
| ^^^^^^^^^^^^^^^^^^^^^^^^^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/taskgroups.py", line 174, in _aexit
| raise BaseExceptionGroup(
| ...<2 lines>...
| ) from None
| ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
+-+---------------- 1 ----------------
| Exception Group Traceback (most recent call last):
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/join.py", line 586, in _shuffle_join
| await gather_in_task_group(*actor_tasks)
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py", line 93, in gather_in_task_group
| async with asyncio.TaskGroup() as tg:
| ~~~~~~~~~~~~~~~~~^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/taskgroups.py", line 72, in __aexit__
| return await self._aexit(et, exc)
| ^^^^^^^^^^^^^^^^^^^^^^^^^^
| File "/conda/envs/cudf-dev-polars-tests/lib/python3.14/asyncio/taskgroups.py", line 174, in _aexit
| raise BaseExceptionGroup(
| ...<2 lines>...
| ) from None
| ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
+-+---------------- 1 ----------------
| Traceback (most recent call last):
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/join.py", line 465, in _join_chunks
| chunk_to_frame(right_chunk, right),
| ~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^
| File "/cudf/python/cudf_polars/cudf_polars/experimental/rapidsmpf/utils.py", line 926, in chunk_to_frame
| return DataFrame.from_table(
| ~~~~~~~~~~~~~~~~~~~~^
| chunk.table_view(),
| ^^^^^^^^^^^^^^^^^^^
| ...<2 lines>...
| chunk.stream,
| ^^^^^^^^^^^^^
| )
| ^
| File "/cudf/python/cudf_polars/cudf_polars/containers/dataframe.py", line 240, in from_table
| raise ValueError("Mismatching name and table length.")
| ValueError: Mismatching name and table length.
+------------------------------------
```
IIRC we may be filtering out empty (0 length) tables preemptively
```
> /cudf/python/cudf_polars/cudf_polars/containers/dataframe.py(240)from_table()
-> breakpoint()
(Pdb) names
['a', 'b']
(Pdb) table.num_columns(
... )
0
(Pdb) table.columns()
[]
(Pdb) table.to_arrow()
pyarrow.Table
----
```
Contributor guide
Assessment
This issue has not been assessed yet.