# 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,

MapKey,

MapList,

RepeatedValue,

ScalarMap,

ScalarMapEntry,

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)

elif val_attr == "scalar_map_val":

return _handle_scalar_map_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<string, Value> 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 _map_key_to_python_value(map_key: MapKey) -> Any:

"""Convert a MapKey proto to its Python equivalent."""

key_attr = map_key.WhichOneof("key")

if key_attr is None:

return None

val = getattr(map_key, key_attr)

if key_attr in ("int32_key", "int64_key"):

return int(val)

if key_attr in ("float_key", "double_key"):

return float(val)

if key_attr == "bool_key":

return bool(val)

if key_attr == "unix_timestamp_key":

return (

datetime.fromtimestamp(val, tz=timezone.utc)

if val != NULL_TIMESTAMP_INT_VALUE

else None

)

if key_attr == "bytes_key":

return bytes(val)

if key_attr in ("uuid_key", "time_uuid_key"):

return uuid_module.UUID(val)

if key_attr == "decimal_key":

return decimal.Decimal(val)

return val

def _handle_scalar_map_value(value_map_message: ScalarMap) -> Dict[Any, Any]:

"""Handle ScalarMap proto message (repeated ScalarMapEntry) → Python dict."""

result: Dict[Any, Any] = {}

for entry in value_map_message.val:

key = _map_key_to_python_value(entry.key)

value = feast_value_type_to_python_type(entry.value)

result[key] = 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):

# Non-string keys require ScalarMap; string keys (or empty dict) use Map

if value and not isinstance(next(iter(value)), str):

return ValueType.SCALAR_MAP

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 _is_array_like(value: Any) -> bool:

"""Return True if *value* is array-like (numpy array or any sized,

non-string, non-bytes container). Array-like values in a scalar

feature column cannot be mapped to a protobuf scalar field and are

therefore always treated as null."""

return isinstance(value, np.ndarray) or (

hasattr(value, "__len__") and not isinstance(value, (str, bytes))

)

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:

out: List[Any] = [None] * len(values)

clean_indices: List[int] = []

clean_values: List[Any] = []

for i, value in enumerate(values):

if _is_array_like(value) or value is None:

out[i] = ProtoValue()

else:

clean_indices.append(i)

clean_values.append(value)

if clean_values:

timestamps = _python_datetime_to_int_timestamp(clean_values)

for i, ts in zip(clean_indices, timestamps):

out[i] = ProtoValue(unix_timestamp_val=ts) # type: ignore

return out