Fix remote file system (get/put)

[pingsutw]

TL;DR

Failed to serialize the flytefile due to below error:

[3/3] currentAttempt done. Last Error: SYSTEM::Traceback (most recent call last):

      File "/usr/local/lib/python3.10/site-packages/flytekit/exceptions/scopes.py", line 165, in system_entry_point
        return wrapped(*args, **kwargs)
      File "/usr/local/lib/python3.10/site-packages/flytekit/core/base_task.py", line 662, in dispatch_execute
        literals_map, native_outputs_as_map = self._output_to_literal_map(native_outputs, exec_ctx)
      File "/usr/local/lib/python3.10/site-packages/flytekit/core/base_task.py", line 561, in _output_to_literal_map
        raise TypeError(msg)

Message:

    Failed to convert outputs of task 'test.t1' at position 0:
  'list' object has no attribute 'startswith'

SYSTEM ERROR! Contact platform administrators.

That's because asyncFileSystem (s3fs) return [None] when calling file_system.put here

Type

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested
• Unit tests added
• Code documentation added
• Any pending items have an associated Issue

Complete description

from flytekit import task, ImageSpec, workflow
from flytekit.types.file import FlyteFile

new_flytekit = "git+https://github.com/flyteorg/flytekit.git@c49ceb779cb4617053004c94d4bd71b0622c63b2"
image = ImageSpec(apt_packages=["git"], packages=[new_flytekit], registry="pingsutw")


@task(container_image=image)
def t1() -> FlyteFile:
    tmp_file = "/tmp/abc"
    with open(tmp_file, "w") as f:
        f.write("wow")
    return tmp_file


@workflow()
def wf():
    t1()


if __name__ == '__main__':
    wf()

Tracking Issue

NA

Follow-up issue

NA

[codecov]

Codecov Report

Attention: 8 lines in your changes are missing coverage. Please review.

Comparison is base (38c7687) 62.70% compared to head (cd2cbc7) 62.58%.
Report is 1 commits behind head on master.

Files	Patch %	Lines
flytekit/core/data_persistence.py	0.00%	8 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #1955      +/-   ##
==========================================
- Coverage   62.70%   62.58%   -0.13%     
==========================================
  Files         313      310       -3     
  Lines       23181    23108      -73     
  Branches     3511     3513       +2     
==========================================
- Hits        14536    14462      -74     
- Misses       8223     8224       +1     
  Partials      422      422              

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[pingsutw]

cc @kumare3 @wild-endeavor @eapolinario

[wild-endeavor]

is there a unit test we can add?

[wild-endeavor]

what does dst stand for?

[wild-endeavor]

maybe add a test like

flytekit/tests/flytekit/unit/core/test_type_engine.py

Line 203 in 38c7687

class JsonTypeTransformer(TypeTransformer[T]):

, just making a dummy transformer, if that's easy to do. if not, just add the below as a comment in both places.

[eapolinario]

dst is short for destination, right?

[pingsutw]

yes, since fsspec uses dst as well

[wild-endeavor]

@eapolinario basically with the pr that went in we kinda added a new API to the api without being able to make it explicit.

Basically, when an fsspec filesystem now handles get/put, it has the option of returning a value, and that value is now also used by flytekit as the uri in most places. This was useful because the flyte fs is not a real fs, so users don't actually pick the to_path that they want to write to. instead where it was actually written is returned by the filesystem.

In this case however, the s3 filesystem was returning [None], so the replacement mechanism was failing. (the replacement mechanism is there because in oss flyte at least, even if the user opts in to write to the flyte remote fs, you don't want to store flyte:// as the uri. you still want to store the underlying s3/gcs path).

[eapolinario]

dst is short for destination, right?

[eapolinario]

IIUC, [None] is a guard value returned by fsspec, right? If that's the case, why don't we handle those explicitly?

[wild-endeavor]

oh really? didn't see that. link me?

[wild-endeavor]

and yes we should, but the action should still be to return the to_path.

[eapolinario]

I don't think it's part of the spec and I fear this is specific to s3fs. Doing what we're doing in the PR is probably safer (as it might apply to other fsspec-compliant implementations).

[pingsutw]

AsyncFileSystem.put or AsyncFileSystem.get always return a list.

• https://github.com/fsspec/filesystem_spec/blob/294248578994de67f8da00692de6872f1352ce33/fsspec/asyn.py#L574
• https://github.com/fsspec/filesystem_spec/blob/294248578994de67f8da00692de6872f1352ce33/fsspec/asyn.py#L242-L256