Add FileDownloadConfig annotation for FlyteFile inputs

Port new BlobType fields file_extension and enable_legacy_filename to flytekit.
FlyteFile inputs can be annotated with the FileDownloadConfig annotation to
configure the file extension to use during the copilot download phase.

e.g.

```python
def t1(file: Annotated[FlyteFile, FileDownloadConfig(file_extension="csv")]):
    ... # copilot downloads the file to e.g. /inputs/file.csv

versus...

def t1(file: FlyteFile["csv"]):
    ... # copilot downloads the file to e.g. /inputs/file
```

Author: ddl-rliu

flytekit/core/type_engine.py

@@ -105,6 +105,48 @@ def get_batch_size(t: Type) -> Optional[int]:
     return None
 
 
+class FileDownloadConfig:
+    """
+    This is used to annotate a FlyteFile when we want to download the file with a specific extension. For example,
+
+    ```python
+    # ContainerTask
+    def t1(file: Annotated[FlyteFile, FileDownloadConfig(file_extension="csv")]):
+        ... # copilot downloads the file to e.g. /inputs/file.csv
+    
+    versus...
+    
+    def t1(file: FlyteFile["csv"]):
+        ... # copilot downloads the file to e.g. /inputs/file
+    ```
+
+    file_extension: (Default is "") The file extension (e.g. "csv", "parquet") to use during copilot download.
+    enable_legacy_filename: (Default is False) When true and file_extension is non-empty, the copilot download phase
+    writes the blob to both the full path (with extension) and the old path (without extension), preserving backward compatibility for
+    workflows with tasks that may read from both.
+    """
+
+    def __init__(self, file_extension: str = "", enable_legacy_filename: bool = False):
+        self._file_extension = file_extension
+        self._enable_legacy_filename = enable_legacy_filename
+    
+    @property
+    def file_extension(self) -> str:
+        return self._file_extension
+    
+    @property
+    def enable_legacy_filename(self) -> bool:
+        return self._enable_legacy_filename
+
+
+def get_file_download_config(t: Type) -> Optional[FileDownloadConfig]:
+    if is_annotated(t):
+        for arg in get_args(t):
+            if isinstance(arg, FileDownloadConfig):
+                return arg
+    return None
+
+
 def modify_literal_uris(lit: Literal):
     """
     Modifies the literal object recursively to replace the URIs with the native paths in case they are of

flytekit/models/core/types.py

@@ -38,13 +38,19 @@ class BlobDimensionality(object):
         SINGLE = _types_pb2.BlobType.SINGLE
         MULTIPART = _types_pb2.BlobType.MULTIPART
 
-    def __init__(self, format, dimensionality):
+    def __init__(self, format, dimensionality, file_extension="", enable_legacy_filename=False):
         """
         :param Text format: A string describing the format of the underlying blob data.
         :param int dimensionality: An integer from BlobType.BlobDimensionality enum
+        :param Text file_extension: The file extension (e.g. "csv", "parquet") to use
+            during copilot download, e.g. "csv", "parquet". Empty by default.
+        :param bool enable_legacy_filename: When True and file_extension is set, the copilot
+            download phase writes the blob to both the extended path and the base path.
         """
         self._format = format
         self._dimensionality = dimensionality
+        self._file_extension = file_extension
+        self._enable_legacy_filename = enable_legacy_filename
 
     @property
     def format(self):
@@ -62,16 +68,44 @@ def dimensionality(self):
         """
         return self._dimensionality
 
+    @property
+    def file_extension(self):
+        """
+        The file extension (e.g. "csv", "parquet") to use during copilot download.
+        Default is "", which means no extension is appended.
+        :rtype: Text
+        """
+        return self._file_extension
+
+    @property
+    def enable_legacy_filename(self):
+        """
+        When True and file_extension is set, the copilot download writes the blob to
+        both the full path (with extension) and the old path (without extension).
+        :rtype: bool
+        """
+        return self._enable_legacy_filename
+
     def to_flyte_idl(self):
         """
         :rtype: flyteidl.core.types_pb2.BlobType
         """
-        return _types_pb2.BlobType(format=self.format, dimensionality=self.dimensionality)
+        return _types_pb2.BlobType(
+            format=self.format,
+            dimensionality=self.dimensionality,
+            file_extension=self._file_extension,
+            enable_legacy_filename=self._enable_legacy_filename,
+        )
 
     @classmethod
     def from_flyte_idl(cls, proto):
         """
         :param flyteidl.core.types_pb2.BlobType proto:
         :rtype: BlobType
         """
-        return cls(format=proto.format, dimensionality=proto.dimensionality)
+        return cls(
+            format=proto.format,
+            dimensionality=proto.dimensionality,
+            file_extension=proto.file_extension,
+            enable_legacy_filename=proto.enable_legacy_filename,
+        )

flytekit/types/file/file.py

@@ -24,6 +24,7 @@
     AsyncTypeTransformer,
     TypeEngine,
     TypeTransformerFailedError,
+    get_file_download_config,
     get_underlying_type,
 )
 from flytekit.exceptions.user import FlyteAssertion
@@ -477,8 +478,26 @@ def get_format(t: typing.Union[typing.Type[FlyteFile], os.PathLike]) -> str:
             return ""
         return cast(FlyteFile, t).extension()
 
-    def _blob_type(self, format: str) -> BlobType:
-        return BlobType(format=format, dimensionality=BlobType.BlobDimensionality.SINGLE)
+    @staticmethod
+    def get_file_extension(t: typing.Union[typing.Type[FlyteFile], os.PathLike]) -> str:
+        if t is os.PathLike:
+            return ""
+        file_download_config = get_file_download_config(t)
+        if file_download_config is None:
+            return ""
+        return file_download_config.file_extension or ""
+
+    @staticmethod
+    def get_enable_legacy_filename(t: typing.Union[typing.Type[FlyteFile], os.PathLike]) -> str:
+        if t is os.PathLike:
+            return False
+        file_download_config = get_file_download_config(t)
+        if file_download_config is None:
+            return False
+        return file_download_config.enable_legacy_filename or False
+
+    def _blob_type(self, format: str, file_extension: str = "", enable_legacy_filename: bool = False) -> BlobType:
+        return BlobType(format=format, dimensionality=BlobType.BlobDimensionality.SINGLE, file_extension=file_extension, enable_legacy_filename=enable_legacy_filename)
 
     def assert_type(
         self, t: typing.Union[typing.Type[FlyteFile], os.PathLike], v: typing.Union[FlyteFile, os.PathLike, str]
@@ -491,7 +510,11 @@ def assert_type(
         )
 
     def get_literal_type(self, t: typing.Union[typing.Type[FlyteFile], os.PathLike]) -> LiteralType:
-        return LiteralType(blob=self._blob_type(format=FlyteFilePathTransformer.get_format(t)))
+        return LiteralType(blob=self._blob_type(
+            format=FlyteFilePathTransformer.get_format(t),
+            file_extension=FlyteFilePathTransformer.get_file_extension(t),
+            enable_legacy_filename=FlyteFilePathTransformer.get_enable_legacy_filename(t),
+        ))
 
     def get_mime_type_from_extension(self, extension: str) -> typing.Union[str, typing.Sequence[str]]:
         extension_to_mime_type = {
@@ -565,7 +588,11 @@ async def async_to_literal(
             raise ValueError(f"Incorrect type {python_type}, must be either a FlyteFile or os.PathLike")
 
         # information used by all cases
-        meta = BlobMetadata(type=self._blob_type(format=FlyteFilePathTransformer.get_format(python_type)))
+        meta = BlobMetadata(type=self._blob_type(
+            format=FlyteFilePathTransformer.get_format(python_type),
+            file_extension=FlyteFilePathTransformer.get_file_extension(python_type),
+            enable_legacy_filename=FlyteFilePathTransformer.get_enable_legacy_filename(python_type),
+        ))
 
         if isinstance(python_val, FlyteFile):
             # Cast the source path to str type to avoid error raised when the source path is used as the blob uri,

pyproject.toml

@@ -21,7 +21,7 @@ dependencies = [
     "diskcache>=5.2.1",
     "docker>=4.0.0",
     "docstring-parser>=0.9.0",
-    "flyteidl>=1.16.1,<2.0.0a0",
+    "flyteidl @ git+https://github.com/dominodatalab/flyteidl.git@af517f6",
     "fsspec>=2023.3.0",
     # Bug in 2025.5.0, 2025.5.0post1 https://github.com/fsspec/gcsfs/issues/687
     # Bug in 2024.2.0 https://github.com/fsspec/gcsfs/pull/643