Unexpected DagsterInvariantViolationError When Using PipesSubprocessClient in an Asset

[martincpt]

I encountered a DagsterInvariantViolationError when trying to return a simple Python primitive (like a str) from an asset that utilizes PipesSubprocessClient. Here's the error message:

dagster._core.errors.DagsterInvariantViolationError: Compute function for op "download_from_url" yielded a value of type <class 'str'> rather than an instance of Output, AssetMaterialization, or ExpectationResult. Values yielded by ops must be wrapped in one of these types.

Here is my code:

@asset()
def download_from_url(
    context: AssetExecutionContext,
    pipes_subprocess_client: PipesSubprocessClient,
    get_releases_url: str,
) -> str:
    cmd = ["wget", get_releases_url, "-O", DOWNLOAD_PATH]
    pipes_subprocess_client.run(command=cmd, context=context)
    return DOWNLOAD_PATH

Surprisingly, when I remove the lines involving PipesSubprocessClient, everything works as expected and returns the string without errors.

What I Tried

The error message suggested that my function was behaving more like an op than an asset, so I adjusted my code to explicitly return an Output object instead of a plain string. After this change, it started working:

@asset()
def download_from_url(
    context: AssetExecutionContext,
    pipes_subprocess_client: PipesSubprocessClient,
    get_releases_url: str,
) -> Output[str]:
    cmd = ["wget", get_releases_url, "-O", DOWNLOAD_PATH]
    pipes_subprocess_client.run(command=cmd, context=context)
    return Output(DOWNLOAD_PATH)

Questions

1. Why does using PipesSubprocessClient alter the behavior of the asset function?
   Without it, returning a simple string works fine. With it, the function seems to require the use of an Output object.

2. Is this behavior intentional or a bug?
   If intentional, it would be helpful for the documentation to clarify this distinction, as the current behavior is unexpected.

3. Why does the function behave more like an op when PipesSubprocessClient is involved?
   This isn't immediately clear from the documentation or error message.

Looking forward to insights on whether this is expected behavior or if I'm missing something in the configuration.

[qboriskr]

Up