Traceback (most recent call last):
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/working_dir_files/_ray_pkg_1e73be3aae0a1261/daft_test_7.py", line 14, in <module>
    df.write_parquet(
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py", line 38, in _wrap
    return func(*args, **kwargs)
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 822, in write_parquet
    write_df.collect()
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/api_annotations.py", line 38, in _wrap
    return func(*args, **kwargs)
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3988, in collect
    self._materialize_results()
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/dataframe/dataframe.py", line 3951, in _materialize_results
    self._result_cache = get_or_create_runner().run(self._builder)
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/ray_runner.py", line 1412, in run
    return self._collect_into_cache(results_iter)
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/ray_runner.py", line 1403, in _collect_into_cache
    for i, result in enumerate(results_iter):
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/ray_runner.py", line 1390, in run_iter
    yield from self.flotilla_plan_runner.stream_plan(
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/flotilla.py", line 309, in stream_plan
    materialized_result = ray.get(self.runner.get_next_partition.remote(plan_id))
  File "/home/ray/anaconda3/lib/python3.9/site-packages/ray/_private/auto_init_hook.py", line 22, in auto_init_wrapper
    return fn(*args, **kwargs)
  File "/home/ray/anaconda3/lib/python3.9/site-packages/ray/_private/client_mode_hook.py", line 104, in wrapper
    return func(*args, **kwargs)
  File "/home/ray/anaconda3/lib/python3.9/site-packages/ray/_private/worker.py", line 2849, in get
    values, debugger_breakpoint = worker.get_objects(object_refs, timeout=timeout)
  File "/home/ray/anaconda3/lib/python3.9/site-packages/ray/_private/worker.py", line 937, in get_objects
    raise value.as_instanceof_cause()
ray.exceptions.RayTaskError(DaftCoreException): ray::RemoteFlotillaRunner.get_next_partition() (pid=2915, ip=10.224.16.126, actor_id=1a6ee62a716d7731d876037402000000, repr=<daft.runners.flotilla.RemoteFlotillaRunner object at 0x7f36d80437f0>)
  File "/home/ray/anaconda3/lib/python3.9/concurrent/futures/_base.py", line 439, in result
    return self.__get_result()
  File "/home/ray/anaconda3/lib/python3.9/concurrent/futures/_base.py", line 391, in __get_result
    raise self._exception
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/flotilla.py", line 249, in get_next_partition
    next_partition_ref = await self.curr_result_gens[plan_id].__anext__()
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/flotilla.py", line 142, in get_result
    return await self.task
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/flotilla.py", line 138, in _get_result
    raise e
  File "/tmp/ray/session_2025-10-22_07-28-59_169964_1/runtime_resources/pip/659065918d88b2363673f783d947ad796ab0d8f4/virtualenv/lib/python3.9/site-packages/daft/runners/flotilla.py", line 124, in _get_result
    metadatas = await metadatas_ref
ray.exceptions.RayTaskError(DaftCoreException): ray::PhysicalScan->Project->HashAggregate() (pid=383, ip=10.224.4.171, actor_id=d361308861cf00be4481101202000000, repr=<daft.runners.flotilla.RaySwordfishActor object at 0x7fec44704850>)
    async for partition in native_executor.run_async(plan, psets_mp, exec_cfg, None, context):
daft.exceptions.DaftCoreException: DaftError::External Cached error: Unable to open file abfs://path/to/my.parquet: Error { context: Full(Custom { kind: Io, error: reqwest::Error { kind: Decode, source: hyper::Error(Body, Os { code: 104, kind: ConnectionReset, message: "Connection reset by peer" }) } }, "error converting `reqwest` request into a byte stream") }