fix(core): Handle empty external schema fields (#3329)

Signed-off-by: Kevin Su <[email protected]>

Author: pingsutw

flytekit/models/types.py

@@ -238,8 +238,8 @@ def from_flyte_idl(cls, proto: _types_pb2.StructuredDatasetType) -> _types_pb2.S
         return cls(
             columns=[StructuredDatasetType.DatasetColumn.from_flyte_idl(c) for c in proto.columns],
             format=proto.format,
-            external_schema_type=proto.external_schema_type,
-            external_schema_bytes=proto.external_schema_bytes,
+            external_schema_type=proto.external_schema_type if proto.external_schema_type else None,
+            external_schema_bytes=proto.external_schema_bytes if proto.external_schema_bytes else None,
         )
 
 

tests/flytekit/unit/types/structured_dataset/test_structured_dataset.py

@@ -751,3 +751,12 @@ def test_structured_dataset_pickleable():
     unpickled_input = pickle.loads(pickled_input)
 
     assert downstream_input == unpickled_input
+
+def test_optional_sd_with_serde():
+    ctx = FlyteContext.current_context()
+    lt_optional = TypeEngine.to_literal_type(typing.Optional[StructuredDataset])
+
+    lit = TypeEngine.to_literal(ctx, StructuredDataset(df), python_type=typing.Optional[StructuredDataset], expected=lt_optional)
+    lit2 = lit.from_flyte_idl(lit.to_flyte_idl())
+    pv = TypeEngine.to_python_value(ctx, lit2, typing.Optional[StructuredDataset])
+    assert pv is not None