# Copyright 2019 The Feast Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import decimal
import json
import logging
import uuid as uuid_module
from collections import defaultdict
from datetime import datetime, timezone
from typing import (
TYPE_CHECKING,
Any,
Dict,
Iterator,
List,
Optional,
Sequence,
Set,
Sized,
Tuple,
Type,
Union,
cast,
)
import numpy as np
import pandas as pd
from google.protobuf.timestamp_pb2 import Timestamp
from feast.protos.feast.types.Value_pb2 import (
BoolList,
BoolSet,
BytesList,
BytesSet,
DoubleList,
DoubleSet,
FloatList,
FloatSet,
Int32List,
Int32Set,
Int64List,
Int64Set,
Map,
MapList,
RepeatedValue,
StringList,
StringSet,
)
from feast.protos.feast.types.Value_pb2 import Value as ProtoValue
from feast.value_type import ListType, SetType, ValueType
if TYPE_CHECKING:
import pyarrow
# null timestamps get converted to -9223372036854775808
NULL_TIMESTAMP_INT_VALUE: int = np.datetime64("NaT").astype(int)
logger = logging.getLogger(__name__)
def feast_value_type_to_python_type(
field_value_proto: ProtoValue,
feature_type: Optional[ValueType] = None,
) -> Any:
"""
Converts field value Proto to Dict and returns each field's Feast Value Type value
in their respective Python value.
Args:
field_value_proto: Field value Proto
Returns:
Python native type representation/version of the given field_value_proto
"""
val_attr = field_value_proto.WhichOneof("val")
if val_attr is None:
return None
val = getattr(field_value_proto, val_attr)
# Handle JSON types stored as strings but returned as parsed Python objects
if val_attr == "json_val":
try:
return json.loads(val)
except (json.JSONDecodeError, TypeError):
return val
elif val_attr == "json_list_val":
result = []
for v in val.val:
if isinstance(v, str):
try:
result.append(json.loads(v))
except (json.JSONDecodeError, TypeError):
result.append(v)
else:
result.append(v)
return result
# Handle nested collection types (list_val, set_val)
if val_attr in ("list_val", "set_val"):
return _handle_nested_collection_value(val)
# Handle Struct types stored using Map proto, returned as dicts
if val_attr == "struct_val":
return _handle_map_value(val)
elif val_attr == "struct_list_val":
return _handle_map_list_value(val)
# Handle Map and MapList types FIRST (before generic list processing)
if val_attr == "map_val":
return _handle_map_value(val)
elif val_attr == "map_list_val":
return _handle_map_list_value(val)
# If it's a _LIST or _SET type extract the values.
if hasattr(val, "val"):
val = list(val.val)
# Convert UNIX_TIMESTAMP values to `datetime`
if val_attr == "unix_timestamp_list_val":
val = [
(
datetime.fromtimestamp(v, tz=timezone.utc)
if v != NULL_TIMESTAMP_INT_VALUE
else None
)
for v in val
]
elif val_attr == "unix_timestamp_set_val":
val = set(
[
(
datetime.fromtimestamp(v, tz=timezone.utc)
if v != NULL_TIMESTAMP_INT_VALUE
else None
)
for v in val
]
)
elif val_attr == "unix_timestamp_val":
val = (
datetime.fromtimestamp(val, tz=timezone.utc)
if val != NULL_TIMESTAMP_INT_VALUE
else None
)
# Convert _SET types to Python sets
elif val_attr.endswith("_set_val") and val_attr != "unix_timestamp_set_val":
val = set(val)
# Convert UUID values to uuid.UUID objects
if val_attr in ("uuid_val", "time_uuid_val"):
return uuid_module.UUID(val) if isinstance(val, str) else val
if val_attr in ("uuid_list_val", "time_uuid_list_val"):
return [uuid_module.UUID(v) if isinstance(v, str) else v for v in val]
if val_attr in ("uuid_set_val", "time_uuid_set_val"):
return {uuid_module.UUID(v) if isinstance(v, str) else v for v in val}
# Convert DECIMAL values to decimal.Decimal objects
if val_attr == "decimal_val":
return decimal.Decimal(val) if isinstance(val, str) else val
if val_attr == "decimal_list_val":
return [decimal.Decimal(v) if isinstance(v, str) else v for v in val]
if val_attr == "decimal_set_val":
return {decimal.Decimal(v) if isinstance(v, str) else v for v in val}
# Backward compatibility: handle UUIDs stored as string_val/string_list_val with feature_type hint
if feature_type in (ValueType.UUID, ValueType.TIME_UUID) and isinstance(val, str):
return uuid_module.UUID(val)
if feature_type in (ValueType.UUID_LIST, ValueType.TIME_UUID_LIST) and isinstance(
val, list
):
return [uuid_module.UUID(v) if isinstance(v, str) else v for v in val]
if feature_type in (ValueType.UUID_SET, ValueType.TIME_UUID_SET) and isinstance(
val, set
):
return {uuid_module.UUID(v) if isinstance(v, str) else v for v in val}
return val
def _handle_map_value(map_message) -> Dict[str, Any]:
"""Handle Map proto message containing map val."""
result = {}
for key, value in map_message.val.items():
# Recursively handle the Value message
result[key] = feast_value_type_to_python_type(value)
return result
def _handle_map_list_value(map_list_message) -> List[Dict[str, Any]]:
"""Handle MapList proto message containing repeated Map val."""
result = []
for map_item in map_list_message.val:
# Handle each Map in the list
processed_map = _handle_map_value(map_item)
result.append(processed_map)
return result
def _handle_nested_collection_value(repeated_value) -> List[Any]:
"""Handle nested collection proto (RepeatedValue containing Values).
Each inner Value is itself a list/set proto. We recursively convert
each inner Value to a Python list/set via feast_value_type_to_python_type.
"""
result = []
for inner_value in repeated_value.val:
result.append(feast_value_type_to_python_type(inner_value))
return result
def feast_value_type_to_pandas_type(value_type: ValueType) -> Any:
value_type_to_pandas_type: Dict[ValueType, str] = {
ValueType.FLOAT: "float",
ValueType.INT32: "int",
ValueType.INT64: "int",
ValueType.STRING: "str",
ValueType.DOUBLE: "float",
ValueType.BYTES: "bytes",
ValueType.BOOL: "bool",
ValueType.UNIX_TIMESTAMP: "datetime64[ns]",
ValueType.UUID: "str",
ValueType.TIME_UUID: "str",
ValueType.DECIMAL: "object",
}
if (
value_type.name in ("MAP", "JSON", "STRUCT", "VALUE_LIST", "VALUE_SET")
or value_type.name.endswith("_LIST")
or value_type.name.endswith("_SET")
):
return "object"
if value_type in value_type_to_pandas_type:
return value_type_to_pandas_type[value_type]
raise TypeError(
f"Casting to pandas type for type {value_type} failed. "
f"Type {value_type} not found"
)
def python_type_to_feast_value_type(
name: str,
value: Optional[Any] = None,
recurse: bool = True,
type_name: Optional[str] = None,
) -> ValueType:
"""
Finds the equivalent Feast Value Type for a Python value. Both native
and Pandas types are supported. This function will recursively look
for nested types when arrays are detected. All types must be homogenous.
Args:
name: Name of the value or field
value: Value that will be inspected
recurse: Whether to recursively look for nested types in arrays
Returns:
Feast Value Type
"""
type_name = (type_name or type(value).__name__).lower()
type_map = {
"int": ValueType.INT64,
"str": ValueType.STRING,
"string": ValueType.STRING, # pandas.StringDtype
"float": ValueType.DOUBLE,
"bytes": ValueType.BYTES,
"float64": ValueType.DOUBLE,
"float32": ValueType.FLOAT,
"int64": ValueType.INT64,
"uint64": ValueType.INT64,
"int32": ValueType.INT32,
"uint32": ValueType.INT32,
"int16": ValueType.INT32,
"uint16": ValueType.INT32,
"uint8": ValueType.INT32,
"int8": ValueType.INT32,
"bool_": ValueType.BOOL, # np.bool_
"bool": ValueType.BOOL,
"boolean": ValueType.BOOL,
"timedelta": ValueType.UNIX_TIMESTAMP,
"timestamp": ValueType.UNIX_TIMESTAMP,
"datetime": ValueType.UNIX_TIMESTAMP,
"datetime64[ns]": ValueType.UNIX_TIMESTAMP,
"datetime64[ns, tz]": ValueType.UNIX_TIMESTAMP, # special dtype of pandas
"datetime64[ns, utc]": ValueType.UNIX_TIMESTAMP,
"date": ValueType.UNIX_TIMESTAMP,
"category": ValueType.STRING,
"uuid": ValueType.UUID,
"decimal": ValueType.DECIMAL,
}
if type_name in type_map:
return type_map[type_name]
# Handle pandas "object" dtype by inspecting the actual value
if type_name == "object" and value is not None:
# Check the actual type of the value
actual_type = type(value).__name__.lower()
if actual_type == "str":
return ValueType.STRING
# Check if it's a dictionary (could be a Map)
elif actual_type == "dict":
return ValueType.MAP
# If it's a different type wrapped in object, try to infer from the value
elif actual_type in type_map:
return type_map[actual_type]
if isinstance(value, np.ndarray) and str(value.dtype) in type_map:
item_type = type_map[str(value.dtype)]
return ValueType[item_type.name + "_LIST"]
if isinstance(value, (list, np.ndarray)):
# Check if it's a list of maps
if value and isinstance(value[0], dict):
return ValueType.MAP_LIST
# if the value's type is "ndarray" and we couldn't infer from "value.dtype"
# this is most probably array of "object",
# so we need to iterate over objects and try to infer type of each item
if not recurse:
raise ValueError(
f"Value type for field {name} is {type(value)} but "
f"recursion is not allowed. Nested collection types cannot be "
f"inferred automatically; use an explicit Field dtype instead "
f"(e.g., dtype=Array(Array(Int32)))."
)
# This is the final type which we infer from the list
common_item_value_type = None
for item in value:
if isinstance(item, ProtoValue):
current_item_value_type: ValueType = _proto_value_to_value_type(item)
else:
# Get the type from the current item, only one level deep
current_item_value_type = python_type_to_feast_value_type(
name=name, value=item, recurse=False
)
# Validate whether the type stays consistent
if (
common_item_value_type
and not common_item_value_type == current_item_value_type
):
raise ValueError(
f"List value type for field {name} is inconsistent. "
f"{common_item_value_type} different from "
f"{current_item_value_type}."
)
common_item_value_type = current_item_value_type
if common_item_value_type is None:
return ValueType.UNKNOWN
return ValueType[common_item_value_type.name + "_LIST"]
# Check if it's a set (Set type)
if isinstance(value, set):
if not recurse:
raise ValueError(
f"Value type for field {name} is {type(value)} but "
f"recursion is not allowed. Set types can only be one level "
f"deep."
)
# Infer the type from set elements
common_set_item_type = None
for item in value:
if isinstance(item, ProtoValue):
current_set_item_type: ValueType = _proto_value_to_value_type(item)
else:
# Get the type from the current item, only one level deep
current_set_item_type = python_type_to_feast_value_type(
name=name, value=item, recurse=False
)
# Validate whether the type stays consistent
if (
common_set_item_type
and not common_set_item_type == current_set_item_type
):
raise ValueError(
f"Set value type for field {name} is inconsistent. "
f"{common_set_item_type} different from "
f"{current_set_item_type}."
)
common_set_item_type = current_set_item_type
if common_set_item_type is None:
return ValueType.UNKNOWN
return ValueType[common_set_item_type.name + "_SET"]
# Check if it's a dictionary (Map type)
if isinstance(value, dict):
return ValueType.MAP
raise ValueError(
f"Value with native type {type_name} cannot be converted into Feast value type"
)
def python_values_to_feast_value_type(
name: str, values: Any, recurse: bool = True
) -> ValueType:
inferred_dtype = ValueType.UNKNOWN
for row in values:
current_dtype = python_type_to_feast_value_type(
name, value=row, recurse=recurse
)
if inferred_dtype is ValueType.UNKNOWN:
inferred_dtype = current_dtype
else:
if current_dtype != inferred_dtype and current_dtype not in (
ValueType.UNKNOWN,
ValueType.NULL,
):
raise TypeError(
f"Input entity {name} has mixed types, {current_dtype} and {inferred_dtype}. That is not allowed. "
)
if inferred_dtype in (ValueType.UNKNOWN, ValueType.NULL):
raise ValueError(
f"field {name} cannot have all null values for type inference."
)
return inferred_dtype
def _convert_value_type_str_to_value_type(type_str: str) -> ValueType:
type_map = {
"UNKNOWN": ValueType.UNKNOWN,
"BYTES": ValueType.BYTES,
"STRING": ValueType.STRING,
"INT32": ValueType.INT32,
"INT64": ValueType.INT64,
"DOUBLE": ValueType.DOUBLE,
"FLOAT": ValueType.FLOAT,
"FLOAT32": ValueType.FLOAT,
"BOOL": ValueType.BOOL,
"NULL": ValueType.NULL,
"UNIX_TIMESTAMP": ValueType.UNIX_TIMESTAMP,
"BYTES_LIST": ValueType.BYTES_LIST,
"STRING_LIST": ValueType.STRING_LIST,
"INT32_LIST ": ValueType.INT32_LIST,
"INT64_LIST": ValueType.INT64_LIST,
"DOUBLE_LIST": ValueType.DOUBLE_LIST,
"FLOAT_LIST": ValueType.FLOAT_LIST,
"BOOL_LIST": ValueType.BOOL_LIST,
"UNIX_TIMESTAMP_LIST": ValueType.UNIX_TIMESTAMP_LIST,
"MAP": ValueType.MAP,
"MAP_LIST": ValueType.MAP_LIST,
"JSON": ValueType.JSON,
"JSON_LIST": ValueType.JSON_LIST,
"STRUCT": ValueType.STRUCT,
"STRUCT_LIST": ValueType.STRUCT_LIST,
"BYTES_SET": ValueType.BYTES_SET,
"STRING_SET": ValueType.STRING_SET,
"INT32_SET": ValueType.INT32_SET,
"INT64_SET": ValueType.INT64_SET,
"DOUBLE_SET": ValueType.DOUBLE_SET,
"FLOAT_SET": ValueType.FLOAT_SET,
"BOOL_SET": ValueType.BOOL_SET,
"UNIX_TIMESTAMP_SET": ValueType.UNIX_TIMESTAMP_SET,
"UUID": ValueType.UUID,
"TIME_UUID": ValueType.TIME_UUID,
"UUID_LIST": ValueType.UUID_LIST,
"TIME_UUID_LIST": ValueType.TIME_UUID_LIST,
"UUID_SET": ValueType.UUID_SET,
"TIME_UUID_SET": ValueType.TIME_UUID_SET,
"VALUE_LIST": ValueType.VALUE_LIST,
"VALUE_SET": ValueType.VALUE_SET,
"DECIMAL": ValueType.DECIMAL,
"DECIMAL_LIST": ValueType.DECIMAL_LIST,
"DECIMAL_SET": ValueType.DECIMAL_SET,
}
return type_map.get(type_str, ValueType.STRING)
def _type_err(item, dtype):
raise TypeError(f'Value "{item}" is of type {type(item)} not of type {dtype}')
PYTHON_LIST_VALUE_TYPE_TO_PROTO_VALUE: Dict[
ValueType, Tuple[ListType, str, List[Type]]
] = {
ValueType.FLOAT_LIST: (
FloatList,
"float_list_val",
[np.float32, np.float64, float],
),
ValueType.DOUBLE_LIST: (
DoubleList,
"double_list_val",
[np.float64, np.float32, float],
),
ValueType.INT32_LIST: (Int32List, "int32_list_val", [np.int64, np.int32, int]),
ValueType.INT64_LIST: (Int64List, "int64_list_val", [np.int64, np.int32, int]),
ValueType.UNIX_TIMESTAMP_LIST: (
Int64List,
"int64_list_val",
[np.datetime64, np.int64, np.int32, int, datetime, Timestamp],
),
ValueType.STRING_LIST: (StringList, "string_list_val", [np.str_, str]),
ValueType.BOOL_LIST: (BoolList, "bool_list_val", [np.bool_, bool]),
ValueType.BYTES_LIST: (BytesList, "bytes_list_val", [np.bytes_, bytes]),
ValueType.UUID_LIST: (
StringList,
"uuid_list_val",
[np.str_, str, uuid_module.UUID],
),
ValueType.TIME_UUID_LIST: (
StringList,
"time_uuid_list_val",
[np.str_, str, uuid_module.UUID],
),
ValueType.DECIMAL_LIST: (
StringList,
"decimal_list_val",
[np.str_, str, decimal.Decimal],
),
}
PYTHON_SET_VALUE_TYPE_TO_PROTO_VALUE: Dict[
ValueType, Tuple[SetType, str, List[Type]]
] = {
ValueType.FLOAT_SET: (
FloatSet,
"float_set_val",
[np.float32, np.float64, float],
),
ValueType.DOUBLE_SET: (
DoubleSet,
"double_set_val",
[np.float64, np.float32, float],
),
ValueType.INT32_SET: (Int32Set, "int32_set_val", [np.int64, np.int32, int]),
ValueType.INT64_SET: (Int64Set, "int64_set_val", [np.int64, np.int32, int]),
ValueType.UNIX_TIMESTAMP_SET: (
Int64Set,
"unix_timestamp_set_val",
[np.datetime64, np.int64, np.int32, int, datetime, Timestamp],
),
ValueType.STRING_SET: (StringSet, "string_set_val", [np.str_, str]),
ValueType.BOOL_SET: (BoolSet, "bool_set_val", [np.bool_, bool]),
ValueType.BYTES_SET: (BytesSet, "bytes_set_val", [np.bytes_, bytes]),
ValueType.UUID_SET: (StringSet, "uuid_set_val", [np.str_, str, uuid_module.UUID]),
ValueType.TIME_UUID_SET: (
StringSet,
"time_uuid_set_val",
[np.str_, str, uuid_module.UUID],
),
ValueType.DECIMAL_SET: (
StringSet,
"decimal_set_val",
[np.str_, str, decimal.Decimal],
),
}
PYTHON_SCALAR_VALUE_TYPE_TO_PROTO_VALUE: Dict[
ValueType, Tuple[str, Any, Optional[Set[Type]]]
] = {
ValueType.INT32: ("int32_val", lambda x: int(x), None),
ValueType.INT64: (
"int64_val",
lambda x: (
int(x.timestamp())
if isinstance(x, pd._libs.tslibs.timestamps.Timestamp)
else int(x)
),
None,
),
ValueType.FLOAT: ("float_val", lambda x: float(x), None),
ValueType.DOUBLE: (
"double_val",
lambda x: x,
{float, np.float64, int, np.int_, decimal.Decimal},
),
ValueType.STRING: ("string_val", lambda x: str(x), None),
ValueType.BYTES: ("bytes_val", lambda x: x, {bytes}),
ValueType.IMAGE_BYTES: ("bytes_val", lambda x: x, {bytes}),
ValueType.BOOL: ("bool_val", lambda x: x, {bool, np.bool_, int, np.int_}),
ValueType.UUID: ("uuid_val", lambda x: str(x), {str, uuid_module.UUID}),
ValueType.TIME_UUID: ("time_uuid_val", lambda x: str(x), {str, uuid_module.UUID}),
ValueType.DECIMAL: ("decimal_val", lambda x: str(x), {decimal.Decimal, str}),
}
def _python_datetime_to_int_timestamp(
values: Sequence[Any],
) -> Sequence[Union[int, np.int_]]:
# Fast path for Numpy array.
if isinstance(values, np.ndarray) and isinstance(values.dtype, np.datetime64):
if values.ndim != 1:
raise ValueError("Only 1 dimensional arrays are supported.")
return cast(Sequence[np.int_], values.astype("datetime64[s]").astype(np.int_))
int_timestamps = []
for value in values:
if isinstance(value, datetime):
int_timestamps.append(int(value.timestamp()))
elif isinstance(value, Timestamp):
int_timestamps.append(int(value.ToSeconds()))
elif isinstance(value, np.datetime64):
int_timestamps.append(value.astype("datetime64[s]").astype(np.int_)) # type: ignore[attr-defined]
elif isinstance(value, type(np.nan)):
int_timestamps.append(NULL_TIMESTAMP_INT_VALUE)
else:
int_timestamps.append(int(value))
return int_timestamps
def _convert_timestamp_collection_to_proto(
values: List[Any],
proto_field: str,
proto_type: type,
) -> List[ProtoValue]:
"""Convert timestamp collection values (list or set) to proto.
Args:
values: List of timestamp collections to convert.
proto_field: The proto field name (e.g., 'unix_timestamp_list_val').
proto_type: The proto type class (e.g., Int64List).
Returns:
List of ProtoValue with converted timestamps.
"""
result = []
for value in values:
if value is not None:
result.append(
ProtoValue(
**{
proto_field: proto_type(
val=_python_datetime_to_int_timestamp(value)
)
} # type: ignore
)
)
else:
result.append(ProtoValue())
return result
def _convert_bool_collection_to_proto(
values: List[Any],
proto_field: str,
proto_type: type,
) -> List[ProtoValue]:
"""Convert boolean collection values (list or set) to proto.
ProtoValue does not support direct conversion of np.bool_, so we need to
explicitly convert each element to Python bool.
Args:
values: List of boolean collections to convert.
proto_field: The proto field name (e.g., 'bool_list_val').
proto_type: The proto type class (e.g., BoolList).
Returns:
List of ProtoValue with converted booleans.
"""
result = []
for value in values:
if value is not None:
result.append(
ProtoValue(**{proto_field: proto_type(val=[bool(e) for e in value])}) # type: ignore
)
else:
result.append(ProtoValue())
return result
def _validate_collection_item_types(
sample: Any,
valid_types: List[Type],
feast_value_type: ValueType,
) -> None:
"""Validate that collection items match expected types.
Args:
sample: A sample collection value to check.
valid_types: List of valid Python types for items.
feast_value_type: The Feast value type for error messages.
Raises:
TypeError: If any item in sample is not a valid type.
"""
if sample is None:
return
if all(type(item) in valid_types for item in sample):
return
# to_numpy() upcasts INT32/INT64 with NULL to Float64 automatically
int_collection_types = [
ValueType.INT32_LIST,
ValueType.INT64_LIST,
ValueType.INT32_SET,
ValueType.INT64_SET,
]
for item in sample:
if type(item) not in valid_types:
if feast_value_type in int_collection_types:
# Check if the float values are due to NULL upcast
if not any(np.isnan(i) for i in sample if isinstance(i, float)):
logger.error(
f"{feast_value_type.name} has NULL values. to_numpy() upcasts to Float64 automatically."
)
raise _type_err(item, valid_types[0])
def _python_set_to_proto_values(
feast_value_type: ValueType, values: List[Any]
) -> List[ProtoValue]:
"""
Converts Python set values to Feast Proto Values.
Args:
feast_value_type: The target set value type
values: List of set values that will be converted
Returns:
List of Feast Value Proto
"""
# Feature can be set but None is still valid
if feast_value_type not in PYTHON_SET_VALUE_TYPE_TO_PROTO_VALUE:
return []
set_proto_type, set_field_name, set_valid_types = (
PYTHON_SET_VALUE_TYPE_TO_PROTO_VALUE[feast_value_type]
)
# Convert set to list for proto (proto doesn't have native set type)
def convert_set_to_list(value: Any) -> Any:
if value is None:
return None
if isinstance(value, set):
return list(value)
if isinstance(value, (list, tuple, np.ndarray)):
return list(set(value))
return value
converted_values = [convert_set_to_list(v) for v in values]
sample = next(filter(_non_empty_value, converted_values), None)
# Bytes to array type conversion
if isinstance(sample, (bytes, bytearray)):
if feast_value_type == ValueType.BYTES_SET:
raise _type_err(sample, ValueType.BYTES_SET)
json_sample = json.loads(sample)
if isinstance(json_sample, list):
json_values = [
json.loads(value) if value is not None else None
for value in converted_values
]
if feast_value_type == ValueType.BOOL_SET:
json_values = [
[bool(item) for item in list_item]
if list_item is not None
else None
for list_item in json_values
]
return [
ProtoValue(**{set_field_name: set_proto_type(val=v)}) # type: ignore[arg-type]
if v is not None
else ProtoValue()
for v in json_values
]
raise _type_err(sample, set_valid_types[0])
# Validate item types using shared helper
_validate_collection_item_types(sample, set_valid_types, feast_value_type)
# Handle special types using shared helpers
if feast_value_type == ValueType.UNIX_TIMESTAMP_SET:
return _convert_timestamp_collection_to_proto(
converted_values, "unix_timestamp_set_val", Int64Set
)
if feast_value_type == ValueType.BOOL_SET:
return _convert_bool_collection_to_proto(
converted_values, set_field_name, set_proto_type
)
if feast_value_type in (ValueType.UUID_SET, ValueType.TIME_UUID_SET):
# uuid.UUID objects must be converted to str for StringSet proto.
return [
(
ProtoValue(
**{set_field_name: set_proto_type(val=[str(e) for e in value])} # type: ignore[arg-type, misc]
)
if value is not None
else ProtoValue()
)
for value in converted_values
]
if feast_value_type == ValueType.DECIMAL_SET:
# decimal.Decimal objects must be converted to str for StringSet proto.
return [
(
ProtoValue(
**{set_field_name: set_proto_type(val=[str(e) for e in value])} # type: ignore[arg-type, misc]
)
if value is not None
else ProtoValue()
)
for value in converted_values
]
# Generic set conversion
return [
ProtoValue(**{set_field_name: set_proto_type(val=value)}) # type: ignore[arg-type]
if value is not None
else ProtoValue()
for value in converted_values
]
def _convert_list_values_to_proto(
feast_value_type: ValueType,
values: List[Any],
sample: Any,
) -> List[ProtoValue]:
"""Convert list-type values to proto.
Args:
feast_value_type: The target list value type.
values: List of list values to convert.
sample: First non-empty value for type checking.
Returns:
List of ProtoValue.
"""
if feast_value_type not in PYTHON_LIST_VALUE_TYPE_TO_PROTO_VALUE:
raise Exception(f"Unsupported list type: {feast_value_type}")
proto_type, field_name, valid_types = PYTHON_LIST_VALUE_TYPE_TO_PROTO_VALUE[
feast_value_type
]
# Bytes to array type conversion
if isinstance(sample, (bytes, bytearray)):
if feast_value_type == ValueType.BYTES_LIST:
raise _type_err(sample, ValueType.BYTES_LIST)
json_sample = json.loads(sample)
if isinstance(json_sample, list):
json_values = [json.loads(value) for value in values]
if feast_value_type == ValueType.BOOL_LIST:
json_values = [
[bool(item) for item in list_item] for list_item in json_values
]
return [
ProtoValue(**{field_name: proto_type(val=v)}) # type: ignore[arg-type]
for v in json_values
]
raise _type_err(sample, valid_types[0])
# Validate item types using shared helper
_validate_collection_item_types(sample, valid_types, feast_value_type)
# Handle special types using shared helpers
if feast_value_type == ValueType.UNIX_TIMESTAMP_LIST:
return _convert_timestamp_collection_to_proto(
values, "unix_timestamp_list_val", Int64List
)
if feast_value_type == ValueType.BOOL_LIST:
return _convert_bool_collection_to_proto(values, field_name, proto_type)
if feast_value_type in (ValueType.UUID_LIST, ValueType.TIME_UUID_LIST):
# uuid.UUID objects must be converted to str for StringList proto.
return [
(
ProtoValue(
**{field_name: proto_type(val=[str(e) for e in value])} # type: ignore[arg-type, misc]
)
if value is not None
else ProtoValue()
)
for value in values
]
if feast_value_type == ValueType.DECIMAL_LIST:
# decimal.Decimal objects must be converted to str for StringList proto.
return [
(
ProtoValue(
**{field_name: proto_type(val=[str(e) for e in value])} # type: ignore[arg-type, misc]
)
if value is not None
else ProtoValue()
)
for value in values
]
# Generic list conversion
return [
ProtoValue(**{field_name: proto_type(val=value)}) # type: ignore[arg-type]
if value is not None
else ProtoValue()
for value in values
]
def _convert_scalar_values_to_proto(
feast_value_type: ValueType,
values: List[Any],
sample: Any,
) -> List[ProtoValue]:
"""Convert scalar-type values to proto.
Args:
feast_value_type: The target scalar value type.
values: List of scalar values to convert.
sample: First non-empty value for type checking.
Returns:
List of ProtoValue.
"""
if sample is None:
# All input values are None
return [ProtoValue()] * len(values)
if feast_value_type == ValueType.UNIX_TIMESTAMP:
int_timestamps = _python_datetime_to_int_timestamp(values)
return [ProtoValue(unix_timestamp_val=ts) for ts in int_timestamps] # type: ignore
field_name, func, valid_scalar_types = PYTHON_SCALAR_VALUE_TYPE_TO_PROTO_VALUE[
feast_value_type
]
# Validate scalar types
if valid_scalar_types:
if (sample == 0 or sample == 0.0) and feast_value_type != ValueType.BOOL:
# Numpy converts 0 to int, but column type may be float
allowed_types = {np.int64, int, np.float64, float, decimal.Decimal}
assert type(sample) in allowed_types, (
f"Type `{type(sample)}` not in {allowed_types}"
)
else:
assert type(sample) in valid_scalar_types, (
f"Type `{type(sample)}` not in {valid_scalar_types}"
)
# Handle BOOL specially due to np.bool_ conversion requirement
if feast_value_type == ValueType.BOOL:
return [
ProtoValue(
**{field_name: func(bool(value) if type(value) is np.bool_ else value)}
) # type: ignore
if not pd.isnull(value)
else ProtoValue()
for value in values
]
# Generic scalar conversion
out = []
for value in values:
if isinstance(value, ProtoValue):
out.append(value)
elif not pd.isnull(value):
out.append(ProtoValue(**{field_name: func(value)}))
else:
out.append(ProtoValue())
return out
def _python_value_to_proto_value(
feast_value_type: ValueType, values: List[Any]
) -> List[ProtoValue]:
"""
Converts a Python (native, pandas) value to a Feast Proto Value based
on a provided value type.
Args:
feast_value_type: The target value type
values: List of Values that will be converted
Returns:
List of Feast Value Proto
"""
# Handle nested collection types (VALUE_LIST, VALUE_SET)
if feast_value_type in (ValueType.VALUE_LIST, ValueType.VALUE_SET):
return _convert_nested_collection_to_proto(feast_value_type, values)
# Handle Map types
if feast_value_type == ValueType.MAP:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
if isinstance(value, str):
value = json.loads(value)
if not isinstance(value, dict):
raise TypeError(
f"Expected dict for MAP type, got {type(value).__name__}: {value!r}"
)
result.append(ProtoValue(map_val=_python_dict_to_map_proto(value)))
return result
if feast_value_type == ValueType.MAP_LIST:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
if isinstance(value, str):
value = json.loads(value)
if not isinstance(value, list):
raise TypeError(
f"Expected list for MAP_LIST type, got {type(value).__name__}: {value!r}"
)
result.append(
ProtoValue(map_list_val=_python_list_to_map_list_proto(value))
)
return result
# Handle JSON type serialize Python objects as JSON strings
if feast_value_type == ValueType.JSON:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
if isinstance(value, str):
try:
json.loads(value)
except (json.JSONDecodeError, TypeError) as e:
raise ValueError(
f"Invalid JSON string for JSON type: {e}"
) from e
json_str = value
else:
json_str = json.dumps(value)
result.append(ProtoValue(json_val=json_str))
return result
if feast_value_type == ValueType.JSON_LIST:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
json_strings = []
for v in value:
if isinstance(v, str):
try:
json.loads(v)
except (json.JSONDecodeError, TypeError) as e:
raise ValueError(
f"Invalid JSON string in JSON_LIST: {e}"
) from e
json_strings.append(v)
else:
json_strings.append(json.dumps(v))
result.append(ProtoValue(json_list_val=StringList(val=json_strings)))
return result
# Handle Struct type reuses Map proto for storage
if feast_value_type == ValueType.STRUCT:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
if isinstance(value, str):
value = json.loads(value)
if not isinstance(value, dict):
value = (
dict(value)
if hasattr(value, "items")
else {"_value": str(value)}
)
result.append(ProtoValue(struct_val=_python_dict_to_map_proto(value)))
return result
if feast_value_type == ValueType.STRUCT_LIST:
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
if isinstance(value, str):
value = json.loads(value)
result.append(
ProtoValue(struct_list_val=_python_list_to_map_list_proto(value))
)
return result
# Get sample for type checking
sample = next(filter(_non_empty_value, values), None)
# Dispatch to appropriate converter based on type category
type_name_lower = feast_value_type.name.lower()
if "list" in type_name_lower:
return _convert_list_values_to_proto(feast_value_type, values, sample)
if "set" in type_name_lower:
return _python_set_to_proto_values(feast_value_type, values)
# Scalar types
if (
feast_value_type in PYTHON_SCALAR_VALUE_TYPE_TO_PROTO_VALUE
or feast_value_type == ValueType.UNIX_TIMESTAMP
):
return _convert_scalar_values_to_proto(feast_value_type, values, sample)
raise Exception(f"Unsupported data type: {feast_value_type}")
def _convert_nested_collection_to_proto(
feast_value_type: ValueType, values: List[Any]
) -> List[ProtoValue]:
"""Convert nested collection values (list-of-lists, list-of-sets, etc.) to proto."""
val_attr = "list_val" if feast_value_type == ValueType.VALUE_LIST else "set_val"
result = []
for value in values:
if value is None:
result.append(ProtoValue())
else:
inner_values = []
for inner_collection in value:
if inner_collection is None:
inner_values.append(ProtoValue())
else:
inner_list = list(inner_collection)
if len(inner_list) == 0:
# Empty inner collection: store as empty ProtoValue
inner_values.append(ProtoValue())
elif any(
isinstance(item, (list, set, tuple, np.ndarray))
for item in inner_list
):
# Deeper nesting (3+ levels): recurse using VALUE_LIST
inner_proto = _convert_nested_collection_to_proto(
ValueType.VALUE_LIST, [inner_list]
)
inner_values.append(inner_proto[0])
else:
# Leaf level: wrap as a single list-typed Value
proto_vals = python_values_to_proto_values(
[inner_list], ValueType.UNKNOWN
)
inner_values.append(proto_vals[0])
repeated = RepeatedValue(val=inner_values)
proto = ProtoValue()
getattr(proto, val_attr).CopyFrom(repeated)
result.append(proto)
return result
def _python_dict_to_map_proto(python_dict: Dict[str, Any]) -> Map:
"""Convert a Python dictionary to a Map proto message."""
map_proto = Map()
for key, value in python_dict.items():
# Handle None values explicitly
if value is None:
map_proto.val[key].CopyFrom(
ProtoValue()
) # Empty ProtoValue represents None
continue
if isinstance(value, dict):
# Nested map
nested_map_proto = _python_dict_to_map_proto(value)
map_proto.val[key].CopyFrom(ProtoValue(map_val=nested_map_proto))
elif isinstance(value, list) and value and isinstance(value[0], dict):
# List of maps (MapList)
map_list_proto = _python_list_to_map_list_proto(value)
map_proto.val[key].CopyFrom(ProtoValue(map_list_val=map_list_proto))
else:
# Handle scalar values and regular lists
# Let python_values_to_proto_values infer the type
proto_values = python_values_to_proto_values([value], ValueType.UNKNOWN)
map_proto.val[key].CopyFrom(proto_values[0])
return map_proto
def _python_list_to_map_list_proto(python_list: List[Dict[str, Any]]) -> MapList:
"""Convert a Python list of dictionaries to a MapList proto message."""
map_list_proto = MapList()
for item in python_list:
if isinstance(item, dict):
map_proto = _python_dict_to_map_proto(item)
map_list_proto.val.append(map_proto)
else:
raise ValueError(f"MapList can only contain dictionaries, got {type(item)}")
return map_list_proto
def python_values_to_proto_values(
values: List[Any], feature_type: ValueType = ValueType.UNKNOWN
) -> List[ProtoValue]:
value_type = feature_type
sample = next(filter(_non_empty_value, values), None) # first not empty value
if sample is not None and feature_type == ValueType.UNKNOWN:
if isinstance(sample, (list, np.ndarray)):
value_type = (
feature_type
if len(sample) == 0
else python_type_to_feast_value_type("", sample)
)
else:
value_type = python_type_to_feast_value_type("", sample)
if value_type == ValueType.UNKNOWN:
raise TypeError("Couldn't infer value type from empty value")
proto_values = _python_value_to_proto_value(value_type, values)
if len(proto_values) != len(values):
raise ValueError(
f"Number of proto values {len(proto_values)} does not match number of values {len(values)}"
)
return proto_values
PROTO_VALUE_TO_VALUE_TYPE_MAP: Dict[str, ValueType] = {
"int32_val": ValueType.INT32,
"int64_val": ValueType.INT64,
"double_val": ValueType.DOUBLE,
"float_val": ValueType.FLOAT,
"string_val": ValueType.STRING,
"bytes_val": ValueType.BYTES,
"bool_val": ValueType.BOOL,
"unix_timestamp_val": ValueType.UNIX_TIMESTAMP,
"int32_list_val": ValueType.INT32_LIST,
"int64_list_val": ValueType.INT64_LIST,
"double_list_val": ValueType.DOUBLE_LIST,
"float_list_val": ValueType.FLOAT_LIST,
"string_list_val": ValueType.STRING_LIST,
"bytes_list_val": ValueType.BYTES_LIST,
"bool_list_val": ValueType.BOOL_LIST,
"unix_timestamp_list_val": ValueType.UNIX_TIMESTAMP_LIST,
"map_val": ValueType.MAP,
"map_list_val": ValueType.MAP_LIST,
"json_val": ValueType.JSON,
"json_list_val": ValueType.JSON_LIST,
"struct_val": ValueType.STRUCT,
"struct_list_val": ValueType.STRUCT_LIST,
"list_val": ValueType.VALUE_LIST,
"set_val": ValueType.VALUE_SET,
"int32_set_val": ValueType.INT32_SET,
"int64_set_val": ValueType.INT64_SET,
"double_set_val": ValueType.DOUBLE_SET,
"float_set_val": ValueType.FLOAT_SET,
"string_set_val": ValueType.STRING_SET,
"bytes_set_val": ValueType.BYTES_SET,
"bool_set_val": ValueType.BOOL_SET,
"unix_timestamp_set_val": ValueType.UNIX_TIMESTAMP_SET,
"uuid_set_val": ValueType.UUID_SET,
"time_uuid_set_val": ValueType.TIME_UUID_SET,
"uuid_val": ValueType.UUID,
"time_uuid_val": ValueType.TIME_UUID,
"uuid_list_val": ValueType.UUID_LIST,
"time_uuid_list_val": ValueType.TIME_UUID_LIST,
"decimal_val": ValueType.DECIMAL,
"decimal_list_val": ValueType.DECIMAL_LIST,
"decimal_set_val": ValueType.DECIMAL_SET,
}
VALUE_TYPE_TO_PROTO_VALUE_MAP: Dict[ValueType, str] = {
v: k for k, v in PROTO_VALUE_TO_VALUE_TYPE_MAP.items()
}
def _proto_value_to_value_type(proto_value: ProtoValue) -> ValueType:
"""
Returns Feast ValueType given Feast ValueType string.
Args:
proto_str: str
Returns:
A variant of ValueType.
"""
proto_str = proto_value.WhichOneof("val")
if proto_str is None:
return ValueType.UNKNOWN
return PROTO_VALUE_TO_VALUE_TYPE_MAP[proto_str]
def pa_to_feast_value_type(pa_type_as_str: str) -> ValueType:
is_list = False
if pa_type_as_str.startswith("list