[BUG] Attribute access on dataclasses is not reproducible on remote causing cache misses
- Dominant language
- Go
- Stars
- 7.5k
- Forks
- 886
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 120
Description
### Flyte & Flytekit version
```console
flytekit==1.16.1
```
### Describe the bug
Attribute access on dataclasses is not reproducible on remote executions (local executions do not have this problem). I've included an example workflow below which has a similar structure to one of our more complex workflows that has this issue.
Execution 1 (writes to cache):

Execution 2 (does not read from cache):

Execution 3 (reads from cache):

Execution 4 (does not read from cache):

I've noticed that the keys of the input order are switching each time I run it, and if it happens to be the same as a previous execution (e.g., compare 1 and 3) it will read from cache.
### Expected behavior
Cache should get hit on every subsequent execution with identical dataclasses.
### Additional context to reproduce
Example workflow:
```python
from dataclasses import dataclass
from flytekit import task, workflow
from mashumaro.mixins.json import DataClassJSONMixin
@dataclass
class NestedInnerSpec(DataClassJSONMixin):
"""A simple dataclass to demonstrate nested dataclasses in Flyte."""
w: int
x: int
y: int
z: int
@dataclass
class InnerSpec(DataClassJSONMixin):
a: NestedInnerSpec
b: NestedInnerSpec
@dataclass
class Spec(DataClassJSONMixin):
"""Base class for all specifications in the pipeline."""
a: InnerSpec
@task
def get_spec() -> Spec:
"""Get the default specification for the pipeline."""
return Spec(
a=InnerSpec(
a=NestedInnerSpec(w=0, x=1, y=2, z=3),
b=NestedInnerSpec(w=4, x=5, y=6, z=7),
)
)
@task(cache=True, cache_version="abc")
def task1(spec: InnerSpec) -> InnerSpec:
"""A simple task that returns the input spec."""
return spec
@workflow
def subworkflow(spec: Spec) -> InnerSpec:
"""A simple workflow that runs task1 with the provided spec."""
return task1(spec=spec.a)
@workflow
def workflow1() -> InnerSpec:
"""A simple workflow that runs task1."""
spec = get_spec()
out = subworkflow(spec=spec)
return out
```
### 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.