`ÐšÝ Criterion: """Given a field name, an expression, and a table, construct a Pypika Criterion""" # Literal value case if isinstance(expr, (str, int, float, bool)): return _where_clause( key, {cast(WhereOperator, "$eq"): expr}, metadata_q, metadata_t, embeddings_t, ) # Operator dict case operator, value = next(iter(expr.items())) return _value_criterion(key, value, operator, metadata_q, metadata_t, embeddings_t) def _value_criterion( key: str, value: Union[LiteralValue, List[LiteralValue]], op: Union[WhereOperator, InclusionExclusionOperator], metadata_q: QueryBuilder, metadata_t: Table, embeddings_t: Table, ) -> Criterion: """Creates the filter for a single operator""" def is_numeric(obj: object) -> bool: return (not isinstance(obj, bool)) and isinstance(obj, (int, float)) sub_q = metadata_q.where(metadata_t.key == ParameterValue(key)) p_val = ParameterValue(value) if is_numeric(value) or (isinstance(value, list) and is_numeric(value[0])): int_col, float_col = metadata_t.int_value, metadata_t.float_value if op in ("$eq", "$ne"): expr = (int_col == p_val) | (float_col == p_val) elif op == "$gt": expr = (int_col > p_val) | (float_col > p_val) elif op == "$gte": expr = (int_col >= p_val) | (float_col >= p_val) elif op == "$lt": expr = (int_col < p_val) | (float_col < p_val) elif op == "$lte": expr = (int_col <= p_val) | (float_col <= p_val) else: expr = int_col.isin(p_val) | float_col.isin(p_val) else: if isinstance(value, bool) or ( isinstance(value, list) and isinstance(value[0], bool) ): col = metadata_t.bool_value else: col = metadata_t.string_value if op in ("$eq", "$ne"): expr = col == p_val else: expr = col.isin(p_val) if op in ("$ne", "$nin"): return embeddings_t.id.notin(sub_q.where(expr)) else: return embeddings_t.id.isin(sub_q.where(expr)) on"!