Function _convert_arrow_odfv_to_proto

sdk/python/feast/utils.py:484–551

(
    table: Union[pyarrow.Table, pyarrow.RecordBatch],
    feature_view: "OnDemandFeatureView",
    join_keys: Dict[str, ValueType],
)

def _convert_arrow_odfv_to_proto(
    table: Union[pyarrow.Table, pyarrow.RecordBatch],
    feature_view: "OnDemandFeatureView",
    join_keys: Dict[str, ValueType],
) -> List[Tuple[EntityKeyProto, Dict[str, ValueProto], datetime, Optional[datetime]]]:
    # Avoid ChunkedArrays which guarantees `zero_copy_only` available.
    if isinstance(table, pyarrow.Table):
        table = table.to_batches()[0]

    columns = [
        (field.name, field.dtype.to_value_type()) for field in feature_view.features
    ] + list(join_keys.items())

    # Convert columns that exist in the table
    proto_values_by_column = _columns_to_proto_values(
        table, columns, allow_missing=True
    )

    # Ensure join keys are included, creating null values if missing from table
    for join_key, value_type in join_keys.items():
        if join_key not in proto_values_by_column:
            if join_key in table.column_names:
                proto_values_by_column[join_key] = python_values_to_proto_values(
                    table.column(join_key).to_numpy(zero_copy_only=False), value_type
                )
            else:
                # Create null proto values directly (no need to build a PyArrow array)
                proto_values_by_column[join_key] = python_values_to_proto_values(
                    [None] * table.num_rows, value_type
                )

    # Cache column names set to avoid recreating list in loop
    column_names = {c[0] for c in columns}

    # Adding On Demand Features that are missing from proto_values
    for feature in feature_view.features:
        if feature.name in column_names and feature.name not in proto_values_by_column:
            # Create null proto values directly (more efficient than building PyArrow array)
            proto_values_by_column[feature.name] = python_values_to_proto_values(
                [None] * table.num_rows, feature.dtype.to_value_type()
            )

    entity_keys = _build_entity_keys(table.num_rows, join_keys, proto_values_by_column)

    # Serialize the features per row
    feature_dict = {
        feature.name: proto_values_by_column[feature.name]
        for feature in feature_view.features
        if feature.name in proto_values_by_column
    }
    if feature_view.write_to_online_store:
        table_columns = {col.name for col in table.schema}
        for feature in feature_view.schema:
            if feature.name not in feature_dict and feature.name in table_columns:
                feature_dict[feature.name] = proto_values_by_column[feature.name]

    features = [dict(zip(feature_dict, vars)) for vars in zip(*feature_dict.values())]