[BUG] Caching breaks when renaming files on local
- Dominant language
- Go
- Stars
- 7.5k
- Forks
- 886
- Avg merge
- 1d 12h
- Merged PRs (30d)
- 120
Description
### Describe the bug
I encountered this bug when writing a workflow where a task renames a file (changing the file extension to be precise). When executed twice, the following code will break on local but succeed on a cluster:
```python
from flytekit import task, Cache,FlyteFile, workflow, Resources
import os
import tempfile
@task(cache=Cache(version="1.0"), limits=Resources(cpu="1"))
def create_file() -> FlyteFile:
with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
tmp_file.write(b"Hello, world!")
tmp_file.flush()
return FlyteFile(path=tmp_file.name)
@task(limits=Resources(cpu="1"))
def process_file(file: FlyteFile) -> None:
path = file.download()
os.rename(path, path + ".processed") # changing the name or location or whatever
print("okay")
@workflow
def wf() -> None:
file = create_file()
process_file(file)
```
Full trace:
```python
Running Execution on local.
╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /home/mathisz/Documents/pipeline_framework_test/flyte/pythonvenv/lib/python3.12/site-packages/fl │
│ ytekit/core/type_engine.py:1530 in _literal_map_to_kwargs │
│ │
│ 1527 │ │ │ │ kwargs[k] = asyncio.create_task( │
│ 1528 │ │ │ │ │ TypeEngine.async_to_python_value(ctx, lm.literals[k], python_interfa │
│ 1529 │ │ │ │ ) │
│ ❱ 1530 │ │ │ await asyncio.gather(*kwargs.values()) │
│ 1531 │ │ except Exception as e: │
│ 1532 │ │ │ raise TypeTransformerFailedError( │
│ 1533 │ │ │ │ f"Error converting input '{k}' at position {i}:\n" │
│ │
│ /home/mathisz/Documents/pipeline_framework_test/flyte/pythonvenv/lib/python3.12/site-packages/fl │
│ ytekit/core/type_engine.py:1454 in async_to_python_value │
│ │
│ 1451 │ │ │ lv = await cls.unwrap_offloaded_literal(ctx, lv) │
│ 1452 │ │ transformer = cls.get_transformer(expected_python_type) │
│ 1453 │ │ if isinstance(transformer, AsyncTypeTransformer): │
│ ❱ 1454 │ │ │ pv = await transformer.async_to_python_value(ctx, lv, expected_python_type) │
│ 1455 │ │ else: │
│ 1456 │ │ │ pv = transformer.to_python_value(ctx, lv, expected_python_type) │
│ 1457 │ │ return pv │
│ │
│ /home/mathisz/Documents/pipeline_framework_test/flyte/pythonvenv/lib/python3.12/site-packages/fl │
│ ytekit/types/file/file.py:734 in async_to_python_value │
│ │
│ 731 │ │ │ raise TypeTransformerFailedError(f"{lv.scalar.blob.uri} is not a file.") │
│ 732 │ │ │
│ 733 │ │ if not ctx.file_access.is_remote(uri) and not os.path.isfile(uri): │
│ ❱ 734 │ │ │ raise FlyteAssertion( │
│ 735 │ │ │ │ f"Cannot convert from {lv} to {expected_python_type}. " f"Expected a fil │
│ 736 │ │ │ ) │
│ 737 │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
FlyteAssertion: USER:AssertionError: error=Cannot convert from Flyte Serialized object (Literal):
scalar:
blob:
metadata:
uri: /tmp/tmp6obr1z6e to . Expected a file, but /tmp/tmp6obr1z6e is not a file.
During handling of the above exception, another exception occurred:
╭─────────────────────────────── Traceback (most recent call last) ────────────────────────────────╮
│ /home/mathisz/Documents/pipeline_framework_test/flyte/broken_caching_pipeline.py:23 in wf │
│ │
│ 20 @workflow │
│ 21 def wf() -> None: │
│ 22 │ file = create_file() │
│ ❱ 23 │ process_file(file) │
│ 24 │
│ │
│ /usr/lib/python3.12/concurrent/futures/_base.py:456 in result │
│ │
│ 453 │ │ │ │ if self._state in [CANCELLED, CANCELLED_AND_NOTIFIED]: │
│ 454 │ │ │ │ │ raise CancelledError() │
│ 455 │ │ │ │ elif self._state == FINISHED: │
│ ❱ 456 │ │ │ │ │ return self.__get_result() │
│ 457 │ │ │ │ else: │
│ 458 │ │ │ │ │ raise TimeoutError() │
│ 459 │ │ finally: │
│ │
│ /usr/lib/python3.12/concurrent/futures/_base.py:401 in __get_result │
│ │
│ 398 │ def __get_result(self): │
│ 399 │ │ if self._exception: │
│ 400 │ │ │ try: │
│ ❱ 401 │ │ │ │ raise self._exception │
│ 402 │ │ │ finally: │
│ 403 │ │ │ │ # Break a reference cycle with the exception in self._exception │
│ 404 │ │ │ │ self = None │
╰──────────────────────────────────────────────────────────────────────────────────────────────────╯
TypeTransformerFailedError: Error encountered while converting inputs of 'broken_caching_pipeline.process_file':
Error converting input 'file' at position 0:
Literal value: Flyte Serialized object (Literal):
scalar:
blob:
metadata:
uri: /tmp/tmp6obr1z6e
Expected Python type:
Exception: USER:AssertionError: error=Cannot convert from Flyte Serialized object (Literal):
scalar:
blob:
metadata:
uri: /tmp/tmp6obr1z6e to . Expected a file, but /tmp/tmp6obr1z6e is not a file.
```
### Expected behavior
The cached file should probably be stored in a way so a task using it cannot overwrite it. At least the error message could be improved (it took me quite some time to figure out the problem) and the documentation should warn about this.
### Additional context to reproduce
_No response_
### Screenshots
_No response_
### Are you sure this issue hasn't been raised already?
- [x] Yes
### Have you read the Code of Conduct?
- [x] Yes
Contributor guide
Assessment
This issue has not been assessed yet.