[Core feature] [flytekit] Revisit StructuredDatasetDecoder interface

[wild-endeavor]

Flytekit StructuredDatasetDecoder Interface

The current interface for the StructuredDatasetDecoder looks like

    def decode(
        self,
        ctx: FlyteContext,
        flyte_value: literals.StructuredDataset,
    ) -> Union[DF, Generator[DF, None, None]]:

This served our purposes fine until we realized in #822 that there were situations where the type information (or the encapsulating Metadata object rather) was more specific than in the incoming literal.

For example, if you have tasks

def t1() -> pd.DataFrame: ...
def t2(in1: Annotated[pd.DataFrame, kwtypes(col_a=int)]): ...

where t2(in1=t1()) then when t2 runs and attempts to decode the incoming literal, it actually has more information than the incoming literal.

Currently the code will just overwrite the metadata inside the literal, but should we leave that be and add an optional parameter to the decoder interface? current_task_metadata: Optional[StructuredDatasetMetadata] = None

Are you sure this issue hasn't been raised already?

• Yes

Have you read the Code of Conduct?

• Yes