Dask substitutes a string argument with a Future.result() #7551
Description
Describe the issue:
The main_task should receive 2 arguments: a string and a dictionary. Instead, it receives 2 dictionaries if a future's key equals the string argument.
Minimal Complete Verifiable Example:
from dask.distributed import LocalCluster, Client
from typing import Dict
def main_task(data_name: str, data: Dict):
return isinstance(data_name, dict) and isinstance(data, dict) and data_name == data
def get_data(data_name: str) -> Dict:
return {"value": data_name}
cluster = LocalCluster(n_workers=3)
client = Client(cluster.scheduler_address)
data_name = "test_data_name"
f1 = client.submit(
get_data,
data_name=data_name,
key=data_name,
)
f2 = client.submit(
main_task,
data_name=data_name,
data=f1,
)
# The `result` is `True` which is wrong. `data_name` and `data` should have different types and values
result = f2.result()
assert result is TrueAnything else we need to know?:
Environment:
- Dask version: 2022.9.2, 2023.1.1
- Python version: 3.9.16
- Operating System: ubuntu:22.04, macos:13.2.1
- Install method (conda, pip, source): pip