fix: Recognize Parquet list<element: T> types in pa_to_feast_value_type

sdk/python/feast/infra/offline_stores/offline_utils.py

@@ -20,7 +20,7 @@
 from feast.infra.offline_stores.offline_store import OfflineStore
 from feast.infra.registry.base_registry import BaseRegistry
 from feast.repo_config import RepoConfig
-from feast.type_map import feast_value_type_to_pa
+from feast.type_map import _pa_list_inner_type_str, feast_value_type_to_pa
 from feast.utils import (
     _get_requested_feature_views_to_features_dict,
     _get_requested_on_demand_feature_views,
@@ -273,8 +273,8 @@ def get_offline_store_from_config(offline_store_config: Any) -> OfflineStore:
 def _parse_pa_type_str(pa_type_str: str) -> pa.DataType:
     """Parse a PyArrow type string to preserve inner element types for nested lists."""
     pa_type_str = pa_type_str.strip()
-    if pa_type_str.startswith("list<item: ") and pa_type_str.endswith(">"):
-        inner = pa_type_str[len("list<item: ") : -1]
+    inner = _pa_list_inner_type_str(pa_type_str)
+    if inner is not None:
         return pa.list_(_parse_pa_type_str(inner))
     if pa_type_str in _PA_BASIC_TYPES:
         return _PA_BASIC_TYPES[pa_type_str]

sdk/python/feast/type_map.py

@@ -1604,6 +1604,19 @@ def _proto_value_to_value_type(proto_value: ProtoValue) -> ValueType:
 
 _FIXED_SIZE_LIST_PREFIX = "fixed_size_list<item: "
 
+# PyArrow names the list child field "item" by default, but Parquet-compliant
+# lists (e.g. files written by pandas/pyarrow) name it "element". Either may also
+# be a large_list or mark its elements "not null".
+_PA_LIST_TYPE_PATTERN = re.compile(
+    r"(?:large_)?list<(?:item|element): (.+?)(?: not null)?>"
+)
+
+
+def _pa_list_inner_type_str(pa_type_as_str: str) -> Optional[str]:
+    """Return the element type string of a PyArrow list type string, else None."""
+    match = _PA_LIST_TYPE_PATTERN.fullmatch(pa_type_as_str)
+    return match.group(1) if match else None
+
 
 def pa_to_feast_value_type(pa_type_as_str: str) -> ValueType:
     is_list = False
@@ -1618,11 +1631,11 @@ def pa_to_feast_value_type(pa_type_as_str: str) -> ValueType:
         if closing != -1:
             pa_type_as_str = f"list<item: {inner[:closing]}>"
 
-    if pa_type_as_str.startswith("list<item: "):
+    inner_str = _pa_list_inner_type_str(pa_type_as_str)
+    if inner_str is not None:
         is_list = True
-        inner_str = pa_type_as_str[len("list<item: ") : -1]
         # Check for nested list (list of lists) before stripping
-        if inner_str.startswith("list<item: "):
+        if _pa_list_inner_type_str(inner_str) is not None:
             return ValueType.VALUE_LIST
         pa_type_as_str = inner_str
 
@@ -1650,13 +1663,6 @@ def pa_to_feast_value_type(pa_type_as_str: str) -> ValueType:
             "binary": ValueType.BYTES,
             "bool": ValueType.BOOL,
             "null": ValueType.NULL,
-            "list<element: double>": ValueType.DOUBLE_LIST,
-            "list<element: int64>": ValueType.INT64_LIST,
-            "list<element: int32>": ValueType.INT32_LIST,
-            "list<element: str>": ValueType.STRING_LIST,
-            "list<element: bool>": ValueType.BOOL_LIST,
-            "list<element: bytes>": ValueType.BYTES_LIST,
-            "list<element: float>": ValueType.FLOAT_LIST,
         }
         value_type = type_map[pa_type_as_str]
 

sdk/python/tests/unit/test_type_map.py

@@ -1977,6 +1977,85 @@ def test_pa_to_feast_value_type_nested(self):
         )
 
 
+class TestPyArrowListTypeStrings:
+    """PyArrow spells list types several ways; Parquet round-trips (pyarrow >= 13)
+    name the child field ``element`` instead of ``item``. Schema inference and
+    ``write_to_offline_store`` used to raise ``KeyError: 'list<element: string>'``.
+    """
+
+    @pytest.mark.parametrize(
+        "pa_type_str, expected",
+        [
+            ("list<element: string>", ValueType.STRING_LIST),
+            ("list<element: binary>", ValueType.BYTES_LIST),
+            ("list<element: int64>", ValueType.INT64_LIST),
+            ("list<element: double>", ValueType.DOUBLE_LIST),
+            ("list<element: timestamp[us, tz=UTC]>", ValueType.UNIX_TIMESTAMP_LIST),
+            ("list<element: struct<a: int64 not null>>", ValueType.STRUCT_LIST),
+            ("large_list<item: string>", ValueType.STRING_LIST),
+            ("large_list<element: binary>", ValueType.BYTES_LIST),
+            ("list<item: int64 not null>", ValueType.INT64_LIST),
+            ("list<item: string>", ValueType.STRING_LIST),
+            ("list<item: large_string>", ValueType.JSON_LIST),
+            ("list<element: list<element: int64>>", ValueType.VALUE_LIST),
+            ("list<item: list<item: int64> not null>", ValueType.VALUE_LIST),
+            ("large_list<item: list<element: string>>", ValueType.VALUE_LIST),
+        ],
+    )
+    def test_pa_to_feast_value_type_list_spellings(self, pa_type_str, expected):
+        assert pa_to_feast_value_type(pa_type_str) == expected
+
+    @pytest.mark.parametrize(
+        "pa_type_str, expected",
+        [
+            (
+                "list<element: list<element: int64>>",
+                pyarrow.list_(pyarrow.list_(pyarrow.int64())),
+            ),
+            (
+                "list<item: list<item: string not null> not null>",
+                pyarrow.list_(pyarrow.list_(pyarrow.string())),
+            ),
+            (
+                "large_list<item: list<element: double>>",
+                pyarrow.list_(pyarrow.list_(pyarrow.float64())),
+            ),
+        ],
+    )
+    def test_parse_pa_type_str_list_spellings(self, pa_type_str, expected):
+        from feast.infra.offline_stores.offline_utils import _parse_pa_type_str
+
+        assert _parse_pa_type_str(pa_type_str) == expected
+
+    def test_file_source_parquet_list_columns(self, tmp_path):
+        from feast.infra.offline_stores.file_source import FileSource
+        from feast.infra.offline_stores.offline_utils import (
+            get_pyarrow_schema_from_batch_source,
+        )
+        from feast.repo_config import RepoConfig
+
+        path = tmp_path / "features.parquet"
+        pd.DataFrame(
+            {"tags": [["a", "b"], ["c"]], "nested": [[[1, 2]], [[3]]]}
+        ).to_parquet(path)
+        source = FileSource(path=str(path), timestamp_field="event_timestamp")
+        config = RepoConfig(
+            project="test", registry=str(tmp_path / "registry.db"), provider="local"
+        )
+
+        column_types = dict(source.get_table_column_names_and_types(config))
+        assert column_types["tags"] == "list<element: string>"
+        to_value_type = source.source_datatype_to_feast_value_type()
+        assert to_value_type(column_types["tags"]) == ValueType.STRING_LIST
+        assert to_value_type(column_types["nested"]) == ValueType.VALUE_LIST
+
+        pa_schema, _ = get_pyarrow_schema_from_batch_source(config, source)
+        assert pa_schema.field("tags").type == pyarrow.list_(pyarrow.string())
+        assert pa_schema.field("nested").type == pyarrow.list_(
+            pyarrow.list_(pyarrow.int64())
+        )
+
+
 class TestEmptyArrayAsNull:
     """Regression tests for https://github.com/feast-dev/feast/issues/6255
     Ensure that an empty numpy array in a scalar feature column is treated as