Fix Flyte Types Upload Issues in Default Input

[Future-Outlier]

Tracking issue

flyteorg/flyte#5318

Why are the changes needed?

Before:
if we use flyte types in a dataclass as a workflow or task default inputs, it will fail to upload to remote (s3) storage.
After:
It should work.

What changes were proposed in this pull request?

push a new context (with remote access=True) to context stack in flytekit, so that when flyte types try to upload data, it will upload to remote.

Note:

1. for flytedirectory it has bug and I'll fix it in the next PR, with thee bug default input + different structure exist.
2. flyteschema will not work, but this is deprecated and low priority, will create an issue to let other solvei it.

How was this patch tested?

remote execution and integration test.

"""
add context where Kevin and Yee found and add unit tests
fix the filepathtransofrmer to literal expected literal type's type hint
"""
import typing
import os
from dataclasses import dataclass, fields, field
from typing import Dict, List
from flytekit.types.file import FlyteFile
from flytekit.types.structured import StructuredDataset
from flytekit.types.directory import FlyteDirectory
from flytekit import task, workflow, ImageSpec
import datetime
from enum import Enum
import pandas as pd

flytekit_hash = "6a8f45cf30531d39633abce95def22327deaec4f"
flytekit = f"git+https://github.com/flyteorg/flytekit.git@{flytekit_hash}"
image = ImageSpec(
    packages=[flytekit, "pandas", "pyarrow"],
    apt_packages=["git"],
    registry="localhost:30000",
)

@dataclass
class DC:
    ff: FlyteFile
    sd: StructuredDataset

# /Users/future-outlier/code/dev/flytekit/build/debugyt/user/FE/src/data/df.parquet
# StructuredDataset(uri="/Users/future-outlier/code/dev/flytekit/build/debugyt/user/FE/src/data/df.parquet", file_format="parquet")

@task(container_image=image)
def t1(dc: DC = DC(ff=FlyteFile(os.path.realpath(__file__)),
                    sd=StructuredDataset(uri="/Users/future-outlier/code/dev/flytekit/build/debugyt/user/FE/src/data/df.parquet", file_format="parquet"))) -> DC:
    with open(dc.ff, "r") as f:
        print("File Content: ", f.read())
    print("sd:", dc.sd.open(pd.DataFrame).all())
    return dc

@workflow
def wf(dc: DC = DC(ff=FlyteFile(os.path.realpath(__file__)),
                    sd=StructuredDataset(uri="/Users/future-outlier/code/dev/flytekit/build/debugyt/user/FE/src/data/df.parquet", file_format="parquet"))):
    t1(dc=dc)

if __name__ == "__main__":
    from flytekit.clis.sdk_in_container import pyflyte
    from click.testing import CliRunner
    import os
    # print(os.path.realpath(__file__))
    # print(os.getcwd())

    runner = CliRunner()
    path = os.path.realpath(__file__)
    #
    result = runner.invoke(pyflyte.main,
                           ["run", path, "wf"])
    print("Local Execution: ", result.output)
    #
    result = runner.invoke(pyflyte.main,
                           ["run", "--remote", path, "wf"])
    print("Remote Execution: ", result.output)
    # #
    result = runner.invoke(pyflyte.main,
                           ["run", "--remote", path, "t1"])
    print("Remote Execution: ", result.output)

Setup process

Screenshots

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

Docs link

[pingsutw]

integration tests are failing

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 74.92%. Comparing base (2fbdc63) to head (1da5f31).
Report is 7 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #2907      +/-   ##
==========================================
+ Coverage   72.80%   74.92%   +2.11%     
==========================================
  Files         199      199              
  Lines       20820    20774      -46     
  Branches     2676     2680       +4     
==========================================
+ Hits        15159    15565     +406     
+ Misses       4932     4446     -486     
- Partials      729      763      +34     

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

this works now

[Future-Outlier]

Why we need 2 files in the test folder?

if we use pyflyte register to the workflow file and use pyflyte run --remote, it will trigger plan with different structure already exists.
Since the Protobuf struct one is MAP in golang, and msgpack IDL one is Bytes in golang.

Should we change backend code in flyte or make thie behavior the same now?

why we should?
for better user experience

Reproduce the error

1. pyflyte register file.py
2. pyflyte run --remote file.py wf
3. FLYTE_USE_OLD_DC_FORMAT=true pyflyte run --remote file.py wf

cc @eapolinario
Let's chat about this in the next 10 hours