QA: Why can't the concurrent tasks of Daft be output iteratively, but instead a partition granularity must be assembled?

[Jay-ju]

The number of tasks that can run in each batch here is the number of CPU resources. However, after each batch of data is processed, the data of the iter result is not directly generated. This part of the data is placed on the ray object_memory_store, and then an out - of - memory (OOM) error occurs. Especially when I hope to use 0.5 CPU to run a task.

daft.context.set_runner_ray()
daft.context.execution_config_ctx(
    min_cpu_per_task = 1,
    flotilla= True,
)
daft_ds = daft.read_lance(url=LANCE_PATH, io_config=io_config)
daft_torch_dataset = daft_ds.to_torch_iter_dataset()

start = time.perf_counter()
i = 0
for row in daft_torch_dataset:
    i += 1
end = time.perf_counter()  
print(f"Daft dataset took {end - start:.2f} seconds to iterate over {i} images.")  

[colin-ho]

The data has to go into object store because it is produced by a ray task.

Also, flotilla= True does not go through this part of the code, it is separate. As of today you can find the code here: https://github.com/Eventual-Inc/Daft/tree/main/src/daft-distributed