# Copyright 2021 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 itertools

import logging

import os

import sqlite3

import sys

import time

from datetime import date, datetime, timezone

from pathlib import Path

from typing import (

Any,

Callable,

Dict,

List,

Literal,

Optional,

Sequence,

Tuple,

Union,

cast,

)

from pydantic import StrictStr

from feast import Entity

from feast.feature_view import FeatureView

from feast.field import Field

from feast.filter_models import (

ComparisonFilter,

CompoundFilter,

FilterTranslator,

FilterType,

filters_contain_numeric_comparison,

)

from feast.infra.infra_object import SQLITE_INFRA_OBJECT_CLASS_TYPE, InfraObject

from feast.infra.key_encoding_utils import (

deserialize_entity_key,

serialize_entity_key,

serialize_f32,

)

from feast.infra.online_stores.helpers import compute_table_id, extract_text_and_num

from feast.infra.online_stores.online_store import OnlineStore

from feast.infra.online_stores.vector_store import VectorStoreConfig

from feast.labeling.label_view import LabelView

from feast.protos.feast.core.InfraObject_pb2 import InfraObject as InfraObjectProto

from feast.protos.feast.core.Registry_pb2 import Registry as RegistryProto

from feast.protos.feast.core.SqliteTable_pb2 import SqliteTable as SqliteTableProto

from feast.protos.feast.types.EntityKey_pb2 import EntityKey as EntityKeyProto

from feast.protos.feast.types.Value_pb2 import Value as ValueProto

from feast.repo_config import FeastConfigBaseModel, RepoConfig

from feast.stream_feature_view import StreamFeatureView

from feast.type_map import feast_value_type_to_python_type

from feast.types import FEAST_VECTOR_TYPES, PrimitiveFeastType

from feast.utils import (

_build_retrieve_online_document_record,

_get_feature_view_vector_field_metadata,

_serialize_vector_to_float_list,

to_naive_utc,

)

def adapt_date_iso(val: date):

"""Adapt datetime.date to ISO 8601 date."""

return val.isoformat()

def adapt_datetime_iso(val: datetime):

"""Adapt datetime.datetime to timezone-naive ISO 8601 date."""

return val.isoformat()

def adapt_datetime_epoch(val: datetime):

"""Adapt datetime.datetime to Unix timestamp."""

return int(val.timestamp())

sqlite3.register_adapter(date, adapt_date_iso)

sqlite3.register_adapter(datetime, adapt_datetime_iso)

sqlite3.register_adapter(datetime, adapt_datetime_epoch)

def convert_date(val: bytes):

"""Convert ISO 8601 date to datetime.date object."""

return date.fromisoformat(val.decode())

def convert_datetime(val: bytes):

"""Convert ISO 8601 datetime to datetime.datetime object."""

return datetime.fromisoformat(val.decode())

def convert_timestamp(val: bytes):

"""Convert Unix epoch timestamp to datetime.datetime object."""

return datetime.fromtimestamp(int(val))

sqlite3.register_converter("date", convert_date)

sqlite3.register_converter("datetime", convert_datetime)

sqlite3.register_converter("timestamp", convert_timestamp)

_SQLITE_COMPARISON_OPS: Dict[str, str] = {

"eq": "=",

"ne": "!=",

"gt": ">",

"gte": ">=",

"lt": "<",

"lte": "<=",

}

class SqliteFilterTranslator(FilterTranslator):

"""Translates Feast filters into SQLite WHERE clause fragments."""

def __init__(self, table_name: str, alias: Optional[str] = None):

self.table_name = table_name

self.alias = alias

def translate(self, filters: FilterType) -> Tuple[str, List[Any]]:

if filters is None:

return "", []

return self._dispatch(filters)

def translate_comparison(self, f: ComparisonFilter) -> Tuple[str, List[Any]]:

key, value, op_type = f.key, f.value, f.type

ek_col = f"{self.alias}.entity_key" if self.alias else "entity_key"

if op_type in _SQLITE_COMPARISON_OPS:

col, db_value = _sqlite_filter_col_and_val(value)

clause = (

f"{ek_col} IN (SELECT entity_key FROM {_quote_id(self.table_name)} "

f"WHERE feature_name = ? AND {col} {_SQLITE_COMPARISON_OPS[op_type]} ?)"

)

return clause, [key, db_value]

if op_type == "in":

if not isinstance(value, list):

raise ValueError(

f"'in' filter requires a list value, got {type(value)}"

)

col, _ = (

_sqlite_filter_col_and_val(value[0]) if value else ("value_text", None)

)

db_values = [_sqlite_filter_col_and_val(v)[1] for v in value]

placeholders = ", ".join(["?"] * len(value))

clause = (

f"{ek_col} IN (SELECT entity_key FROM {_quote_id(self.table_name)} "

f"WHERE feature_name = ? AND {col} IN ({placeholders}))"

)

return clause, [key] + db_values

if op_type == "nin":

if not isinstance(value, list):

raise ValueError(

f"'nin' filter requires a list value, got {type(value)}"

)

col, _ = (

_sqlite_filter_col_and_val(value[0]) if value else ("value_text", None)

)

db_values = [_sqlite_filter_col_and_val(v)[1] for v in value]

placeholders = ", ".join(["?"] * len(value))

clause = (

f"{ek_col} IN (SELECT entity_key FROM {_quote_id(self.table_name)} "

f"WHERE feature_name = ? AND {col} NOT IN ({placeholders}))"

)

return clause, [key] + db_values

raise ValueError(f"Unknown comparison operator: {op_type}")

def translate_compound(self, f: CompoundFilter) -> Tuple[str, List[Any]]:

if not f.filters:

return "", []

parts: List[str] = []

all_params: List[Any] = []

for sub in f.filters:

sub_clause, sub_params = self._dispatch(sub)

parts.append(sub_clause)

all_params.extend(sub_params)

joiner = " AND " if f.type == "and" else " OR "

combined = "(" + joiner.join(parts) + ")"

return combined, all_params

def _sqlite_filter_col_and_val(value: Any) -> Tuple[str, Any]:

"""Return the appropriate column name and DB-ready value for a filter value."""

if isinstance(value, bool):

return "value_num", 1.0 if value else 0.0

if isinstance(value, (int, float)):

return "value_num", float(value)

return "value_text", str(value)

class SqliteOnlineStoreConfig(FeastConfigBaseModel, VectorStoreConfig):

"""Online store config for local (SQLite-based) store"""

type: Literal["sqlite", "feast.infra.online_stores.sqlite.SqliteOnlineStore"] = (

"sqlite"

)

""" Online store type selector"""

path: StrictStr = "data/online.db"

""" (optional) Path to sqlite db """

text_search_enabled: bool = False

enable_openai_compatible_store: bool = False

class SqliteOnlineStore(OnlineStore):

"""

SQLite implementation of the online store interface. Not recommended for production usage.

Attributes:

_conn: SQLite connection.

"""

_conn: Optional[sqlite3.Connection] = None

_table_has_value_num: Optional[Dict[str, bool]] = None

@staticmethod

def _get_db_path(config: RepoConfig) -> str:

assert (

config.online_store.type == "sqlite"

or config.online_store.type.endswith("SqliteOnlineStore")

)

if config.repo_path and not Path(config.online_store.path).is_absolute():

db_path = str(config.repo_path / config.online_store.path)

else:

db_path = config.online_store.path

return db_path

def _get_conn(self, config: RepoConfig):

enable_sqlite_vec = (

sys.version_info[0:2] == (3, 10) and config.online_store.vector_enabled

)

if not self._conn:

db_path = self._get_db_path(config)

self._conn = _initialize_conn(db_path, enable_sqlite_vec)

return self._conn

def _check_table_has_value_num(

self, conn: sqlite3.Connection, table_name: str

) -> bool:

"""Check if the value_num column exists in the given table, with caching."""

if self._table_has_value_num is None:

self._table_has_value_num = {}

if table_name in self._table_has_value_num:

return self._table_has_value_num[table_name]

cur = conn.execute(f"PRAGMA table_info({_quote_id(table_name)})")

columns = {row[1] for row in cur.fetchall()}

exists = "value_num" in columns

self._table_has_value_num[table_name] = exists

return exists

@staticmethod

def _filters_need_value_num(

filters: Union[ComparisonFilter, CompoundFilter],

) -> bool:

return filters_contain_numeric_comparison(filters)

def online_write_batch(

self,

config: RepoConfig,

table: FeatureView,

data: List[

Tuple[

EntityKeyProto,

Dict[str, ValueProto],

datetime,

Optional[datetime],

]

],

progress: Optional[Callable[[int], Any]],

) -> None:

conn = self._get_conn(config)

project = config.project

feature_type_dict = {f.name: f.dtype for f in table.features}

table_name = _table_id(

project, table, config.registry.enable_online_feature_view_versioning

)

enable_value_num = getattr(

config.online_store, "enable_openai_compatible_store", False

)

has_value_num_col = False

if enable_value_num:

has_value_num_col = self._check_table_has_value_num(conn, table_name)

if not has_value_num_col:

logging.warning(

"enable_openai_compatible_store is True but value_num column "

"not found in table '%s'. Run `feast apply` to add it. "

"Writing without value_num.",

table_name,

)

compute_value_num = has_value_num_col

columns = ["entity_key", "feature_name", "value", "value_text"]

if has_value_num_col:

columns.append("value_num")

if config.online_store.vector_enabled:

columns.append("vector_value")

columns.extend(["event_ts", "created_ts"])

pk_cols = {"entity_key", "feature_name"}

col_csv = ", ".join(columns)

placeholders = ", ".join(["?"] * len(columns))

update_set = ", ".join(

f"{c} = excluded.{c}" for c in columns if c not in pk_cols

)

upsert_sql = (

f"INSERT INTO {_quote_id(table_name)} ({col_csv}) "

f"VALUES ({placeholders}) "

f"ON CONFLICT(entity_key, feature_name) DO UPDATE SET {update_set};"

)

with conn:

for entity_key, values, timestamp, created_ts in data:

entity_key_bin = serialize_entity_key(

entity_key,

entity_key_serialization_version=config.entity_key_serialization_version,

)

timestamp = to_naive_utc(timestamp)

if created_ts is not None:

created_ts = to_naive_utc(created_ts)

for feature_name, val in values.items():

value_text, value_num = extract_text_and_num(val, compute_value_num)

row: List[Any] = [

entity_key_bin,

feature_name,

val.SerializeToString(),

value_text,

]

if has_value_num_col:

row.append(value_num)

if config.online_store.vector_enabled:

if (

feature_type_dict.get(feature_name, None)

in FEAST_VECTOR_TYPES

):

vector_field_length = getattr(

_get_feature_view_vector_field_metadata(table),

"vector_length",

512,

)

val_bin = serialize_f32(

val.float_list_val.val, vector_field_length

) # type: ignore

else:

val_bin = feast_value_type_to_python_type(val)

row.append(val_bin)

row.extend([timestamp, created_ts])

conn.execute(upsert_sql, tuple(row))

if progress:

progress(1)

def online_read(

self,

config: RepoConfig,

table: FeatureView,

entity_keys: List[EntityKeyProto],

requested_features: Optional[List[str]] = None,

) -> List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]]:

conn = self._get_conn(config)

cur = conn.cursor()

result: List[Tuple[Optional[datetime], Optional[Dict[str, ValueProto]]]] = []

serialized_entity_keys = [

serialize_entity_key(

entity_key,

entity_key_serialization_version=config.entity_key_serialization_version,

)

for entity_key in entity_keys

]

# Fetch all entities in one go

cur.execute(

f"SELECT entity_key, feature_name, value, event_ts "

f"FROM {_quote_id(_table_id(config.project, table, config.registry.enable_online_feature_view_versioning))} "

f"WHERE entity_key IN ({','.join('?' * len(entity_keys))}) "

f"ORDER BY entity_key",

serialized_entity_keys,

)

rows = cur.fetchall()

rows = {

k: list(group) for k, group in itertools.groupby(rows, key=lambda r: r[0])

}

for entity_key_bin in serialized_entity_keys:

res = {}

res_ts = None

for _, feature_name, val_bin, ts in rows.get(entity_key_bin, []):

val = ValueProto()

val.ParseFromString(val_bin)

res[feature_name] = val

ts = cast(datetime, ts)

if ts.tzinfo is not None:

res_ts = ts.astimezone(timezone.utc)

else:

res_ts = ts.replace(tzinfo=timezone.utc)

if not res:

result.append((None, None))

else:

result.append((res_ts, res))

return result

def update(

self,

config: RepoConfig,

tables_to_delete: Sequence[FeatureView],

tables_to_keep: Sequence[FeatureView],

entities_to_delete: Sequence[Entity],

entities_to_keep: Sequence[Entity],

partial: bool,

):

conn = self._get_conn(config)

project = config.project

include_value_num = getattr(

config.online_store, "enable_openai_compatible_store", False

)

versioning = config.registry.enable_online_feature_view_versioning

for table in tables_to_keep:

tbl = _table_id(project, table, versioning)

value_num_col = "value_num REAL," if include_value_num else ""

conn.execute(

f"CREATE TABLE IF NOT EXISTS {_quote_id(tbl)} (entity_key BLOB, feature_name TEXT, value BLOB, value_text TEXT, {value_num_col} vector_value BLOB, event_ts timestamp, created_ts timestamp, PRIMARY KEY(entity_key, feature_name))"

)

conn.execute(

f"CREATE INDEX IF NOT EXISTS {_quote_id(tbl + '_ek')} ON {_quote_id(tbl)} (entity_key);"

)

_alter_table_add_column_if_missing(conn, tbl, "value_text", "TEXT")

if include_value_num:

_alter_table_add_column_if_missing(conn, tbl, "value_num", "REAL")

if self._table_has_value_num is None:

self._table_has_value_num = {}

self._table_has_value_num[tbl] = True

for table in tables_to_delete:

conn.execute(

f"DROP TABLE IF EXISTS {_quote_id(_table_id(project, table, versioning))}"

)

def plan(

self, config: RepoConfig, desired_registry_proto: RegistryProto

) -> List[InfraObject]:

project = config.project

versioning = config.registry.enable_online_feature_view_versioning

include_value_num = getattr(

config.online_store, "enable_openai_compatible_store", False

)

# FeatureView.from_proto() is @typechecked and only accepts a

# FeatureViewProto, so it can't be applied to stream_feature_views

# (StreamFeatureViewProto) too -- each list needs its matching class.

views = [

FeatureView.from_proto(view)

for view in desired_registry_proto.feature_views

] + [

StreamFeatureView.from_proto(view)

for view in desired_registry_proto.stream_feature_views

]

infra_objects: List[InfraObject] = [

SqliteTable(

path=self._get_db_path(config),

name=_table_id(project, view, versioning),

include_value_num=include_value_num,

)

for view in views

]

for lv_proto in desired_registry_proto.label_views:

if lv_proto.spec.online:

lv = LabelView.from_proto(lv_proto)

infra_objects.append(

SqliteTable(

path=self._get_db_path(config),

name=_table_id(project, lv, versioning),

)

)

return infra_objects

def teardown(

self,

config: RepoConfig,

tables: Sequence[FeatureView],

entities: Sequence[Entity],

):

if self._conn is not None:

try:

self._conn.close()

finally:

self._conn = None

db_path = self._get_db_path(config)

for attempt in range(10):

try:

os.unlink(db_path)

return

except FileNotFoundError:

return

except PermissionError:

if attempt == 9:

raise

time.sleep(0.25)

def retrieve_online_documents(

self,

config: RepoConfig,

table: FeatureView,

requested_features: List[str],

embedding: List[float],

top_k: int,

distance_metric: Optional[str] = None,

) -> List[

Tuple[

Optional[datetime],

Optional[EntityKeyProto],

Optional[ValueProto],

Optional[ValueProto],

Optional[ValueProto],

]

]:

"""

Args:

config: Feast configuration object

table: FeatureView object as the table to search

requested_features: The list of requested features to retrieve

embedding: The query embedding to search for

top_k: The number of items to return

Returns:

List of tuples containing the event timestamp, the document feature, the vector value, and the distance

"""

project = config.project

if not config.online_store.vector_enabled:

raise ValueError("sqlite-vss is not enabled in the online store config")

conn = self._get_conn(config)

cur = conn.cursor()

vector_field_length = getattr(

_get_feature_view_vector_field_metadata(table), "vector_length", 512

)

# Convert the embedding to a binary format instead of using SerializeToString()

query_embedding_bin = serialize_f32(embedding, vector_field_length)

table_name = _table_id(

project, table, config.registry.enable_online_feature_view_versioning

)

vector_field = _get_vector_field(table)

cur.execute(

f"""

CREATE VIRTUAL TABLE vec_table using vec0(

vector_value float[{vector_field_length}]

);

"""

)

# Currently I can only insert the embedding value without crashing SQLite, will report a bug

cur.execute(

f"""

INSERT INTO vec_table(rowid, vector_value)

select rowid, vector_value from {_quote_id(table_name)}

where feature_name = ?

""",

(vector_field,),

)

cur.execute(

f"""

CREATE VIRTUAL TABLE IF NOT EXISTS vec_table using vec0(

vector_value float[{vector_field_length}]

);

"""

)

# Have to join this with the main table to get the feature name and entity_key

# Also the `top_k` doesn't appear to be working for some reason

cur.execute(

f"""

select

fv.entity_key,

f.vector_value,

fv.value,

f.distance,

fv.event_ts

from (

select

rowid,

vector_value,

distance

from vec_table

where vector_value match ?

order by distance

limit ?

) f

left join {_quote_id(table_name)} fv

on f.rowid = fv.rowid

""",

(query_embedding_bin, top_k),

)

rows = cur.fetchall()

result: List[

Tuple[

Optional[datetime],

Optional[EntityKeyProto],

Optional[ValueProto],

Optional[ValueProto],

Optional[ValueProto],

]

] = []

for entity_key, _, string_value, distance, event_ts in rows:

result.append(

_build_retrieve_online_document_record(

entity_key,

string_value if string_value else b"",

# This may be a bug

embedding,

distance,

event_ts,

config.entity_key_serialization_version,

)

)

return result

def retrieve_online_documents_v2(

self,

config: RepoConfig,

table: FeatureView,

requested_features: List[str],

embedding: Optional[List[float]],

top_k: int,

distance_metric: Optional[str] = None,

query_string: Optional[str] = None,

filters: Optional[Union[ComparisonFilter, CompoundFilter]] = None,

include_feature_view_version_metadata: bool = False,

) -> List[

Tuple[

Optional[datetime],

Optional[EntityKeyProto],

Optional[Dict[str, ValueProto]],

]

]:

"""

Retrieve documents using vector similarity search.

Args:

config: Feast configuration object

table: FeatureView object as the table to search

requested_features: List of requested features to retrieve

embedding: Query embedding to search for (optional)

top_k: Number of items to return

distance_metric: Distance metric to use (optional)

query_string: The query string to search for using keyword search (bm25) (optional)

Returns:

List of tuples containing the event timestamp, entity key, and feature values

"""

online_store = config.online_store

if not isinstance(online_store, SqliteOnlineStoreConfig):

raise ValueError("online_store must be SqliteOnlineStoreConfig")

if not online_store.vector_enabled and not online_store.text_search_enabled:

raise ValueError(

"You must enable either vector search or text search in the online store config"

)

if filters is not None:

if not getattr(

config.online_store, "enable_openai_compatible_store", False

):

raise ValueError(

"Metadata filtering requires `enable_openai_compatible_store: true` "

"in your online store config. After setting it, run `feast apply` "

"to update the database schema."

)

conn = self._get_conn(config)

cur = conn.cursor()

vector_field_length = getattr(

_get_feature_view_vector_field_metadata(table), "vector_length", 512

)

table_name = _table_id(

config.project, table, config.registry.enable_online_feature_view_versioning

)

vector_field = _get_vector_field(table)

if filters is not None and self._filters_need_value_num(filters):

if not self._check_table_has_value_num(conn, table_name):

raise ValueError(

"Numerical filtering requires the `value_num` column. "

"Run `feast apply` to add it."

)

filter_clause, filter_params = SqliteFilterTranslator(

table_name, alias="fv2"

).translate(filters)

if online_store.vector_enabled:

query_embedding_bin = serialize_f32(embedding, vector_field_length) # type: ignore

cur.execute(

f"""

CREATE VIRTUAL TABLE IF NOT EXISTS vec_table using vec0(

vector_value float[{vector_field_length}]

);

"""

)

cur.execute(

f"""

INSERT INTO vec_table (rowid, vector_value)

select rowid, vector_value from {_quote_id(table_name)}

where feature_name = ?

""",

(vector_field,),

)

elif online_store.text_search_enabled:

string_field_list = [

f.name for f in table.features if f.dtype == PrimitiveFeastType.STRING

]

string_fields = ", ".join(string_field_list)

BM25_DEFAULT_WEIGHTS = ", ".join(

[

str(1.0)

for f in table.features

if f.dtype == PrimitiveFeastType.STRING

]

)

cur.execute(

f"""

CREATE VIRTUAL TABLE IF NOT EXISTS search_table using fts5(

entity_key, fv_rowid, {string_fields}, tokenize="porter unicode61"

);

"""

)

insert_query = _generate_bm25_search_insert_query(

table_name, string_field_list

)

cur.execute(insert_query)

filter_clause, filter_params = SqliteFilterTranslator(

table_name, alias="fv"

).translate(filters)

else:

raise ValueError(

"Neither vector search nor text search are enabled in the online store config"

)

if online_store.vector_enabled:

where_parts = ["fv2.feature_name != ?"]

vector_params = [vector_field]

if filter_clause:

where_parts.append(filter_clause)

where_sql = " AND ".join(where_parts)

cur.execute(

f"""

select

fv2.entity_key,

fv2.feature_name,

fv2.value,

fv.vector_value,

f.distance,

fv.event_ts,

fv.created_ts

from (

select

rowid,

vector_value,

distance

from vec_table

where vector_value match ?

order by distance

limit ?

) f

left join {_quote_id(table_name)} fv

on f.rowid = fv.rowid

left join {_quote_id(table_name)} fv2

on fv.entity_key = fv2.entity_key

where {where_sql}

""",

[query_embedding_bin, top_k] + vector_params + filter_params,

)

elif online_store.text_search_enabled:

where_parts_text = []

if filter_clause:

where_parts_text.append(filter_clause)

where_sql_text = (

"where " + " AND ".join(where_parts_text) if where_parts_text else ""

)

cur.execute(

f"""

select

fv.entity_key,

fv.feature_name,

fv.value,

fv.vector_value,

f.distance,

fv.event_ts,

fv.created_ts

from {_quote_id(table_name)} fv

inner join (

select

fv_rowid,

entity_key,

{string_fields},

bm25(search_table, {BM25_DEFAULT_WEIGHTS}) as distance

from search_table

where search_table match ? order by distance limit ?

) f

on f.entity_key = fv.entity_key

{where_sql_text}

""",

[query_string, top_k] + filter_params,

)

else:

raise ValueError(

"Neither vector search nor text search are enabled in the online store config"

)

rows = cur.fetchall()

results: List[

Tuple[

Optional[datetime],

Optional[EntityKeyProto],

Optional[Dict[str, ValueProto]],

]

] = []

entity_dict: Dict[

str, Dict[str, Union[str, ValueProto, EntityKeyProto, datetime]]

] = {}

for (

entity_key,

feature_name,

value_bin,

vector_value,

distance,

event_ts,

created_ts,

) in rows:

entity_key_proto = deserialize_entity_key(

entity_key,

entity_key_serialization_version=config.entity_key_serialization_version,

)

if entity_key not in entity_dict:

entity_dict[entity_key] = {}

feature_val = ValueProto()

feature_val.ParseFromString(value_bin)

entity_dict[entity_key]["entity_key_proto"] = entity_key_proto

entity_dict[entity_key][feature_name] = feature_val

if online_store.vector_enabled:

entity_dict[entity_key][vector_field] = _serialize_vector_to_float_list(

vector_value

)

entity_dict[entity_key]["distance"] = ValueProto(float_val=distance)

entity_dict[entity_key]["event_ts"] = event_ts

entity_dict[entity_key]["created_ts"] = created_ts

for entity_key_value in entity_dict:

res_event_ts: Optional[datetime] = None

res_entity_key_proto: Optional[EntityKeyProto] = None

if isinstance(entity_dict[entity_key_value]["event_ts"], datetime):

res_event_ts = entity_dict[entity_key_value]["event_ts"] # type: ignore[assignment]

if isinstance(

entity_dict[entity_key_value]["entity_key_proto"], EntityKeyProto

):

res_entity_key_proto = entity_dict[entity_key_value]["entity_key_proto"] # type: ignore[assignment]

res_dict: Dict[str, ValueProto] = {

k: v

for k, v in entity_dict[entity_key_value].items()

if isinstance(v, ValueProto) and isinstance(k, str)

}

results.append(

(

res_event_ts,

res_entity_key_proto,

res_dict,

)

)

return results

def _initialize_conn(

db_path: str, enable_sqlite_vec: bool = False

) -> sqlite3.Connection:

Path(db_path).parent.mkdir(exist_ok=True)

db = sqlite3.connect(

db_path,

detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES,

check_same_thread=False,

)

if enable_sqlite_vec:

try:

import sqlite_vec # noqa: F401

except ModuleNotFoundError:

logging.warning("Cannot use sqlite_vec for vector search")

db.enable_load_extension(True)

sqlite_vec.load(db)

return db

def _alter_table_add_column_if_missing(

conn: sqlite3.Connection,

table_name: str,

column_name: str,

column_type: str,

) -> None:

"""Add a column to an existing SQLite table, ignoring if it already exists.

SQLite's ALTER TABLE ADD COLUMN doesn't support IF NOT EXISTS, so we

catch the specific OperationalError for duplicate columns and re-raise

anything else (connection failures, disk errors, etc.).

"""

try:

conn.execute(

f"ALTER TABLE {_quote_id(table_name)} ADD COLUMN {_quote_id(column_name)} {column_type}"

)

except sqlite3.OperationalError as e:

if "duplicate column name" not in str(e).lower():

raise

def _quote_id(identifier: str) -> str:

"""Quote a SQLite identifier to prevent SQL injection.

Uses the standard SQL double-quote mechanism: any embedded

double-quote characters are escaped by doubling them.

"""

return '"' + identifier.replace('"', '""') + '"'

def _table_id(project: str, table: Any, enable_versioning: bool = False) -> str:

return compute_table_id(project, table, enable_versioning)

class SqliteTable(InfraObject):

"""

A Sqlite table managed by Feast.

Attributes:

path: The absolute path of the Sqlite file.

name: The name of the table.

conn: SQLite connection.

"""

path: str

conn: sqlite3.Connection

_include_value_num: bool

def __init__(self, path: str, name: str, include_value_num: bool = False):

super().__init__(name)

self.path = path

self.conn = _initialize_conn(path)

self._include_value_num = include_value_num

def to_infra_object_proto(self) -> InfraObjectProto:

sqlite_table_proto = self.to_proto()

return InfraObjectProto(

infra_object_class_type=SQLITE_INFRA_OBJECT_CLASS_TYPE,

sqlite_table=sqlite_table_proto,

)

def to_proto(self) -> Any:

sqlite_table_proto = SqliteTableProto()

sqlite_table_proto.path = self.path

sqlite_table_proto.name = self.name

return sqlite_table_proto

@staticmethod

def from_infra_object_proto(infra_object_proto: InfraObjectProto) -> Any: