If my parquets are being read using

/**

from a mounted network drive and sometimes it fails, is there a way to set a retry on the file read? The files exist because they were globbed by the scan and they do contain data, but I consistently get random errors that some file for an arbitrary micropartition does not exist and the whole job fails.

What kind of a drive/storage is being mounted here? We should already have some retry logic in there, but to be fair our S3 reader is a lot more tuned than our local one

It's a HDFS drive mounted to a local file path to all ray nodes

Oh! Interesting I guess that’s one way of getting around HDFS support 😎 There could be many reasons. Do you have an error message we can peek at to see if maybe we can implement some retries here?

I also tried doing my own retry using

def read_with_retry(input_dataset):
            i = 0
            while True:
                try:
                    df = daft.read_parquet(input_dataset).collect()
                    return df
                except:
                    i += 1
                    print_with_context(f"Reading {input_dataset} for the {i}th time failed.")

The error is like this but the file changes each time

No such file or directory (os error 2)
FileNotFoundError: File: /tmp/ray/data/dataset/dataset_parquet/f25f70c5-918c-4386-ae82-54402432b98c-0.parquet not found
    return MicroPartition._from_pymicropartition(_PyMicroPartition.from_scan_task(scan_task))
  File "/tmp/ray/session_2024-09-29_14-48-32_270438_4696/runtime_resources/pip/3abee577c226ebb8373eed712507cf3ba0ed6938/virtualenv/lib/python3.9/site-packages/daft/table/micropartition.py", line 71, in _from_scan_task
    table = MicroPartition._from_scan_task(self.scan_task)
  File "/tmp/ray/session_2024-09-29_14-48-32_270438_4696/runtime_resources/pip/3abee577c226ebb8373eed712507cf3ba0ed6938/virtualenv/lib/python3.9/site-packages/daft/execution/execution_step.py", line 315, in _scan
    return self._scan(inputs)
  File "/tmp/ray/session_2024-09-29_14-48-32_270438_4696/runtime_resources/pip/3abee577c226ebb8373eed712507cf3ba0ed6938/virtualenv/lib/python3.9/site-packages/daft/execution/execution_step.py", line 311, in run
    partitions = instruction.run(partitions)
  File "/tmp/ray/session_2024-09-29_14-48-32_270438_4696/runtime_resources/pip/3abee577c226ebb8373eed712507cf3ba0ed6938/virtualenv/lib/python3.9/site-packages/daft/runners/ray_runner.py", line 467, in build_partitions
    return build_partitions(instruction_stack, partial_metadatas, *inputs)
  File "/tmp/ray/session_2024-09-29_14-48-32_270438_4696/runtime_resources/pip/3abee577c226ebb8373eed712507cf3ba0ed6938/virtualenv/lib/python3.9/site-packages/daft/runners/ray_runner.py", line 489, in single_partition_pipeline
2024-10-07 14:31:29,738	ERROR worker.py:409 -- Unhandled error (suppress with 'RAY_IGNORE_UNHANDLED_ERRORS=1'): [36mray::ScanWithTask [Stage:418]()[39m (pid=57574, ip=10.157.67.238)

Interesting. Thanks for the insight we’ll take a look tomorrow!

Thanks!

Cc @Desmond Cheong as well who is looking at some parquet retry stuff on S3