diff --git a/docs/how-to-guides/feature-monitoring.md b/docs/how-to-guides/feature-monitoring.md index fc8c693ed30..5e2f8b9a37e 100644 --- a/docs/how-to-guides/feature-monitoring.md +++ b/docs/how-to-guides/feature-monitoring.md @@ -359,6 +359,7 @@ Monitoring works natively with all offline stores that serve as compute engines | BigQuery | SQL push-down | `MERGE` into BQ tables | | Redshift | SQL push-down | `MERGE` via Data API | | Spark | SparkSQL push-down | Parquet tables | +| Trino | SQL push-down | Trino tables | | Oracle | SQL via Ibis | `MERGE` from `DUAL` | | DuckDB | In-memory SQL | Parquet files | | Dask | PyArrow compute | Parquet files | diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/connectors/upload.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/connectors/upload.py index 98a2b89f07e..de2a48d05b2 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/connectors/upload.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/connectors/upload.py @@ -18,7 +18,7 @@ ``` """ -from datetime import datetime, timezone +from datetime import date, datetime, timezone from typing import Any, Dict, Iterator, Optional, Set import numpy as np @@ -117,10 +117,18 @@ def _is_nan(value: Any) -> bool: def _format_value(row: pd.Series, schema: Dict[str, Any]) -> str: formated_values = [] for row_name, row_value in row.items(): - if schema[row_name].startswith("timestamp"): + if _is_nan(row_value): + formated_values.append("NULL") + elif schema[row_name].startswith("timestamp"): if isinstance(row_value, datetime): row_value = format_datetime(row_value) formated_values.append(f"TIMESTAMP '{row_value}'") + elif schema[row_name].startswith("date"): + if isinstance(row_value, (datetime, date)): + row_value = row_value.strftime("%Y-%m-%d") + formated_values.append(f"DATE '{row_value}'") + elif isinstance(row_value, (bool, np.bool_)): + formated_values.append("TRUE" if row_value else "FALSE") elif isinstance(row_value, list): formated_values.append(f"ARRAY{row_value}") elif isinstance(row_value, np.ndarray): @@ -128,9 +136,8 @@ def _format_value(row: pd.Series, schema: Dict[str, Any]) -> str: elif isinstance(row_value, tuple): formated_values.append(f"ARRAY{list(row_value)}") elif isinstance(row_value, str): - formated_values.append(f"'{row_value}'") - elif _is_nan(row_value): - formated_values.append("NULL") + escaped = row_value.replace("'", "''") + formated_values.append(f"'{escaped}'") else: formated_values.append(f"{row_value}") diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_monitoring.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_monitoring.py new file mode 100644 index 00000000000..4a07f452411 --- /dev/null +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/tests/test_trino_monitoring.py @@ -0,0 +1,385 @@ +import json +from datetime import date, datetime, timezone +from unittest.mock import MagicMock, patch + +import pandas as pd +import pytest + +from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + TrinoOfflineStore, + TrinoOfflineStoreConfig, + _trino_monitoring_table_name, + _trino_normalize_histogram_column, + _trino_pandas_upsert, + _trino_sql_literal, + _trino_sql_numeric_histogram, + _trino_table_with_clause, +) +from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, +) +from feast.infra.offline_stores.contrib.trino_offline_store.trino_source import ( + TrinoSource, +) +from feast.monitoring.monitoring_utils import ( + MON_TABLE_FEATURE, + MON_TABLE_FEATURE_SERVICE, + MON_TABLE_FEATURE_VIEW, + MON_TABLE_JOB, +) +from feast.repo_config import RepoConfig + + +@pytest.fixture +def repo_config(): + return RepoConfig( + project="test_project", + registry="data/registry.db", + provider="local", + offline_store=TrinoOfflineStoreConfig( + host="localhost", + port=8080, + catalog="memory", + dataset="feast_test", + connector={"type": "memory"}, + user="test_user", + ), + ) + + +@pytest.fixture +def data_source(): + return TrinoSource( + name="test_source", + table="memory.feast_test.driver_stats", + timestamp_field="event_timestamp", + created_timestamp_column="created_timestamp", + ) + + +def test_sql_literal_formatting(): + assert _trino_sql_literal(None) == "NULL" + assert _trino_sql_literal(True) == "TRUE" + assert _trino_sql_literal(False) == "FALSE" + assert _trino_sql_literal(42) == "42" + assert _trino_sql_literal(3.14) == "3.14" + assert _trino_sql_literal("simple") == "'simple'" + assert _trino_sql_literal("O'Reilly") == "'O''Reilly'" + assert _trino_sql_literal(date(2025, 1, 15)) == "DATE '2025-01-15'" + dt = datetime(2025, 1, 15, 12, 0, 0) + assert _trino_sql_literal(dt) == "TIMESTAMP '2025-01-15 12:00:00.000000'" + + +def test_monitoring_table_name_and_with_clause(repo_config): + table_name = _trino_monitoring_table_name(repo_config, MON_TABLE_FEATURE) + assert table_name == f"memory.feast_test.{MON_TABLE_FEATURE}" + + with_clause = _trino_table_with_clause(repo_config) + assert with_clause == "" + + # Hive connector with parquet format + hive_config = RepoConfig( + project="test_project", + registry="data/registry.db", + provider="local", + offline_store=TrinoOfflineStoreConfig( + host="localhost", + port=8080, + catalog="hive", + dataset="default", + connector={"type": "hive", "file_format": "parquet"}, + user="test_user", + ), + ) + assert _trino_table_with_clause(hive_config) == "WITH (format = 'parquet')" + + +def test_normalize_histogram_column(): + pdf = pd.DataFrame( + [ + {"feature_name": "f1", "histogram": {"bins": [1, 2], "counts": [10]}}, + {"feature_name": "f2", "histogram": None}, + {"feature_name": "f3", "histogram": '{"already": "string"}'}, + ] + ) + normalized = _trino_normalize_histogram_column(pdf) + assert isinstance(normalized["histogram"].iloc[0], str) + assert json.loads(normalized["histogram"].iloc[0]) == { + "bins": [1, 2], + "counts": [10], + } + assert normalized["histogram"].iloc[1] is None + assert normalized["histogram"].iloc[2] == '{"already": "string"}' + + +def test_pandas_upsert(): + old_df = pd.DataFrame( + [ + {"project_id": "p1", "feature_name": "f1", "value": 10}, + {"project_id": "p1", "feature_name": "f2", "value": 20}, + ] + ) + new_df = pd.DataFrame( + [ + {"project_id": "p1", "feature_name": "f2", "value": 99}, + {"project_id": "p1", "feature_name": "f3", "value": 30}, + ] + ) + merged = _trino_pandas_upsert(old_df, new_df, ["project_id", "feature_name"]) + assert len(merged) == 3 + row_f2 = merged[merged["feature_name"] == "f2"].iloc[0] + assert row_f2["value"] == 99 + + +def test_compute_monitoring_metrics(repo_config, data_source): + mock_client = MagicMock() + executed_queries = [] + + def mock_execute(query_text): + executed_queries.append(query_text) + if "APPROX_PERCENTILE" in query_text: + return Results( + data=[[100, 90, 50.0, 10.0, 1.0, 100.0, 45.0, 75.0, 90.0, 95.0, 99.0]], + columns=[{"name": "col", "type": "double"}], + ) + elif "GROUP BY bucket" in query_text: + return Results( + data=[[1, 50], [2, 40]], + columns=[ + {"name": "bucket", "type": "bigint"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + elif "WITH filtered AS" in query_text: + return Results( + data=[[100, 0, 3, "val_a", 60], [100, 0, 3, "val_b", 40]], + columns=[ + {"name": "row_count", "type": "bigint"}, + {"name": "null_count", "type": "bigint"}, + {"name": "unique_count", "type": "bigint"}, + {"name": "value", "type": "varchar"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + return Results(data=[], columns=[]) + + mock_client.execute_query.side_effect = mock_execute + + with patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino._get_trino_client", + return_value=mock_client, + ): + results = TrinoOfflineStore.compute_monitoring_metrics( + config=repo_config, + data_source=data_source, + feature_columns=[("trip_cost", "numeric"), ("status", "categorical")], + timestamp_field="event_timestamp", + start_date=datetime(2025, 1, 1, tzinfo=timezone.utc), + end_date=datetime(2025, 1, 2, tzinfo=timezone.utc), + histogram_bins=5, + top_n=10, + ) + + assert len(results) == 2 + assert results[0]["feature_name"] == "trip_cost" + assert results[0]["feature_type"] == "numeric" + assert results[0]["mean"] == 50.0 + assert results[0]["histogram"] is not None + + assert results[1]["feature_name"] == "status" + assert results[1]["feature_type"] == "categorical" + assert results[1]["histogram"]["unique_count"] == 3 + assert len(results[1]["histogram"]["values"]) == 2 + + # Check query patterns + assert any("APPROX_PERCENTILE" in q for q in executed_queries) + assert any("STDDEV_SAMP" in q for q in executed_queries) + assert any("WITH filtered AS" in q for q in executed_queries) + + +def test_get_monitoring_max_timestamp(repo_config, data_source): + mock_client = MagicMock() + mock_client.execute_query.return_value = Results( + data=[[datetime(2025, 1, 15, 10, 30, 0)]], + columns=[{"name": "max_ts", "type": "timestamp"}], + ) + + with patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino._get_trino_client", + return_value=mock_client, + ): + max_ts = TrinoOfflineStore.get_monitoring_max_timestamp( + config=repo_config, + data_source=data_source, + timestamp_field="event_timestamp", + ) + + assert max_ts == datetime(2025, 1, 15, 10, 30, 0, tzinfo=timezone.utc) + + +def test_ensure_monitoring_tables(repo_config): + mock_client = MagicMock() + executed_queries = [] + mock_client.execute_query.side_effect = lambda q: executed_queries.append(q) + + with patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino._get_trino_client", + return_value=mock_client, + ): + TrinoOfflineStore.ensure_monitoring_tables(config=repo_config) + + assert any( + "CREATE SCHEMA IF NOT EXISTS memory.feast_test" in q for q in executed_queries + ) + assert any( + f"CREATE TABLE IF NOT EXISTS memory.feast_test.{MON_TABLE_FEATURE}" in q + for q in executed_queries + ) + assert any( + f"CREATE TABLE IF NOT EXISTS memory.feast_test.{MON_TABLE_FEATURE_VIEW}" in q + for q in executed_queries + ) + assert any( + f"CREATE TABLE IF NOT EXISTS memory.feast_test.{MON_TABLE_FEATURE_SERVICE}" in q + for q in executed_queries + ) + assert any( + f"CREATE TABLE IF NOT EXISTS memory.feast_test.{MON_TABLE_JOB}" in q + for q in executed_queries + ) + + +def test_save_and_query_monitoring_metrics(repo_config): + mock_client = MagicMock() + uploaded_dfs = [] + + def mock_upload(client, df, table, connector_args): + uploaded_dfs.append((table, df)) + + mock_client.execute_query.return_value = Results(data=[], columns=[]) + + with ( + patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino._get_trino_client", + return_value=mock_client, + ), + patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino.upload_pandas_dataframe_to_trino", + side_effect=mock_upload, + ), + ): + metrics = [ + { + "project_id": "test_project", + "feature_view_name": "fv1", + "feature_name": "f1", + "metric_date": date(2025, 1, 1), + "granularity": "daily", + "data_source_type": "batch", + "computed_at": datetime(2025, 1, 1, 12, 0, 0), + "max_event_timestamp": datetime(2025, 1, 1, 12, 0, 0), + "is_baseline": False, + "feature_type": "numeric", + "row_count": 100, + "null_count": 0, + "null_rate": 0.0, + "mean": 10.5, + "stddev": 2.1, + "min_val": 1.0, + "max_val": 20.0, + "p50": 10.0, + "p75": 15.0, + "p90": 18.0, + "p95": 19.0, + "p99": 20.0, + "histogram": {"bins": [1, 20], "counts": [100]}, + } + ] + TrinoOfflineStore.save_monitoring_metrics( + config=repo_config, + metric_type="feature", + metrics=metrics, + ) + + assert len(uploaded_dfs) == 1 + table_uploaded, df_uploaded = uploaded_dfs[0] + assert table_uploaded == f"memory.feast_test.{MON_TABLE_FEATURE}" + assert len(df_uploaded) == 1 + assert df_uploaded["feature_name"].iloc[0] == "f1" + + +def test_clear_monitoring_baseline(repo_config): + mock_client = MagicMock() + existing_df = pd.DataFrame( + [ + { + "project_id": "test_project", + "feature_view_name": "fv1", + "feature_name": "f1", + "data_source_type": "batch", + "is_baseline": True, + }, + { + "project_id": "other_project", + "feature_view_name": "fv1", + "feature_name": "f1", + "data_source_type": "batch", + "is_baseline": True, + }, + ] + ) + + mock_client.execute_query.return_value = Results( + data=existing_df.values.tolist(), + columns=[{"name": col, "type": "varchar"} for col in existing_df.columns], + ) + + uploaded_dfs = [] + + def mock_upload(client, df, table, connector_args): + uploaded_dfs.append(df) + + with ( + patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino._get_trino_client", + return_value=mock_client, + ), + patch( + "feast.infra.offline_stores.contrib.trino_offline_store.trino.upload_pandas_dataframe_to_trino", + side_effect=mock_upload, + ), + ): + TrinoOfflineStore.clear_monitoring_baseline( + config=repo_config, + project="test_project", + feature_view_name="fv1", + feature_name="f1", + ) + + assert len(uploaded_dfs) == 1 + cleared_df = uploaded_dfs[0] + test_proj_row = cleared_df[cleared_df["project_id"] == "test_project"].iloc[0] + assert test_proj_row["is_baseline"] is False or test_proj_row["is_baseline"] == 0 + other_proj_row = cleared_df[cleared_df["project_id"] == "other_project"].iloc[0] + assert other_proj_row["is_baseline"] is True or other_proj_row["is_baseline"] == 1 + + +def test_numeric_histogram_single_value(): + mock_client = MagicMock() + mock_client.execute_query.return_value = Results( + data=[[42]], + columns=[{"name": "cnt", "type": "bigint"}], + ) + + hist = _trino_sql_numeric_histogram( + client=mock_client, + from_expression="test_table", + col_name="val", + ts_clause="1=1", + bins=5, + min_val=10.0, + max_val=10.0, + ) + assert hist["bins"] == [10.0, 10.0] + assert hist["counts"] == [42] + assert hist["bin_width"] == 0.0 diff --git a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py index c9d4119f94f..536572b9cae 100644 --- a/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py +++ b/sdk/python/feast/infra/offline_stores/contrib/trino_offline_store/trino.py @@ -1,6 +1,8 @@ +import json import logging import uuid -from datetime import date, datetime +from datetime import date, datetime, timezone +from datetime import time as dt_time from typing import ( Any, Dict, @@ -42,6 +44,17 @@ ) from feast.infra.offline_stores.offline_utils import get_timestamp_filter_sql from feast.infra.registry.base_registry import BaseRegistry +from feast.monitoring.monitoring_utils import ( + MON_TABLE_FEATURE, + MON_TABLE_FEATURE_SERVICE, + MON_TABLE_FEATURE_VIEW, + MON_TABLE_JOB, + empty_categorical_metric, + empty_numeric_metric, + monitoring_table_meta, + normalize_monitoring_row, + opt_float, +) from feast.on_demand_feature_view import OnDemandFeatureView from feast.repo_config import FeastConfigBaseModel, RepoConfig from feast.saved_dataset import SavedDatasetStorage @@ -174,7 +187,7 @@ class TrinoOfflineStoreConfig(FeastConfigBaseModel): dataset: StrictStr = "feast" """ (optional) Trino Dataset name for temporary tables """ - auth: Optional[AuthConfig] + auth: Optional[AuthConfig] = None """ (optional) Authentication mechanism to use when connecting to Trino. Supported options are: - kerberos @@ -484,6 +497,592 @@ def pull_all_from_table_or_query( full_feature_names=False, ) + @staticmethod + def compute_monitoring_metrics( + config: RepoConfig, + data_source: DataSource, + feature_columns: List[Tuple[str, str]], + timestamp_field: str, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + histogram_bins: int = 20, + top_n: int = 10, + ) -> List[Dict[str, Any]]: + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + assert isinstance(data_source, TrinoSource) + + client = _get_trino_client(config=config) + from_expression = data_source.get_table_query_string() + ts_filter = get_timestamp_filter_sql( + start_date, + end_date, + timestamp_field, + tz=timezone.utc, + cast_style="timestamp", + date_time_separator=" ", + quote_fields=False, + ) + ts_clause = ts_filter if ts_filter else "1=1" + + numeric_features = [n for n, t in feature_columns if t == "numeric"] + categorical_features = [n for n, t in feature_columns if t == "categorical"] + results: List[Dict[str, Any]] = [] + + if numeric_features: + results.extend( + _trino_sql_numeric_stats( + client, + from_expression, + numeric_features, + ts_clause, + histogram_bins, + ) + ) + + for col_name in categorical_features: + results.append( + _trino_sql_categorical_stats( + client, + from_expression, + col_name, + ts_clause, + top_n, + ) + ) + + return results + + @staticmethod + def get_monitoring_max_timestamp( + config: RepoConfig, + data_source: DataSource, + timestamp_field: str, + ) -> Optional[datetime]: + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + assert isinstance(data_source, TrinoSource) + + client = _get_trino_client(config=config) + from_expression = data_source.get_table_query_string() + q_ts = f'"{timestamp_field}"' + sql = f"SELECT MAX({q_ts}) AS max_ts FROM {from_expression} AS _src" + results = client.execute_query(sql) + rows = results.data + if not rows or rows[0] is None or rows[0][0] is None: + return None + val = rows[0][0] + if isinstance(val, datetime): + return val if val.tzinfo else val.replace(tzinfo=timezone.utc) + if isinstance(val, date): + return datetime.combine(val, dt_time.min, tzinfo=timezone.utc) + return pd.to_datetime(val, utc=True).to_pydatetime() + + @staticmethod + def ensure_monitoring_tables(config: RepoConfig) -> None: + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + client = _get_trino_client(config=config) + catalog = config.offline_store.catalog + dataset = config.offline_store.dataset + + if dataset: + try: + client.execute_query(f"CREATE SCHEMA IF NOT EXISTS {catalog}.{dataset}") + except Exception: + logging.exception(f"Failed to create schema {catalog}.{dataset}") + pass + + with_clause = _trino_table_with_clause(config) + for ddl_template, tbl_name in zip( + _TRINO_MONITORING_DDL_STATEMENTS, + [ + MON_TABLE_FEATURE, + MON_TABLE_FEATURE_VIEW, + MON_TABLE_FEATURE_SERVICE, + MON_TABLE_JOB, + ], + ): + full_table = _trino_monitoring_table_name(config, tbl_name) + stmt = ddl_template.format( + table=full_table, + with_clause=with_clause, + ) + client.execute_query(stmt) + + for tbl in ( + MON_TABLE_FEATURE, + MON_TABLE_FEATURE_VIEW, + MON_TABLE_FEATURE_SERVICE, + ): + full_table = _trino_monitoring_table_name(config, tbl) + try: + client.execute_query( + f"ALTER TABLE {full_table} ADD COLUMN max_event_timestamp TIMESTAMP" + ) + except Exception: + # Column already exists on newly created tables or dialect difference + pass + + @staticmethod + def save_monitoring_metrics( + config: RepoConfig, + metric_type: str, + metrics: List[Dict[str, Any]], + ) -> None: + if not metrics: + return + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + table, columns, pk_columns = monitoring_table_meta(metric_type) + full_table_name = _trino_monitoring_table_name(config, table) + pdf_new = pd.DataFrame([{c: m.get(c) for c in columns} for m in metrics]) + pdf_new = _trino_normalize_histogram_column(pdf_new) + + client = _get_trino_client(config=config) + try: + results = client.execute_query(f"SELECT * FROM {full_table_name}") + pdf_old = results.to_dataframe() + pdf_merged = _trino_pandas_upsert(pdf_old, pdf_new, pk_columns) + except Exception: + pdf_merged = pdf_new + + try: + client.execute_query(f"DROP TABLE IF EXISTS {full_table_name}") + except Exception: + pass + + upload_pandas_dataframe_to_trino( + client=client, + df=pdf_merged, + table=full_table_name, + connector_args=config.offline_store.connector, + ) + + @staticmethod + def query_monitoring_metrics( + config: RepoConfig, + project: str, + metric_type: str, + filters: Optional[Dict[str, Any]] = None, + start_date: Optional[date] = None, + end_date: Optional[date] = None, + ) -> List[Dict[str, Any]]: + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + table, columns, _ = monitoring_table_meta(metric_type) + full_table_name = _trino_monitoring_table_name(config, table) + client = _get_trino_client(config=config) + + conditions: List[str] = [] + if project: + conditions.append(f'"project_id" = {_trino_sql_literal(project)}') + if filters: + for key, value in filters.items(): + if value is not None: + conditions.append(f'"{key}" = {_trino_sql_literal(value)}') + if start_date is not None: + conditions.append( + f"\"metric_date\" >= DATE '{start_date.strftime('%Y-%m-%d')}'" + ) + if end_date is not None: + conditions.append( + f"\"metric_date\" <= DATE '{end_date.strftime('%Y-%m-%d')}'" + ) + + where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else "" + order_col = '"metric_date"' if metric_type != "job" else '"job_id"' + cols_str = ", ".join(f'"{c}"' for c in columns) + query = f"SELECT {cols_str} FROM {full_table_name} {where_clause} ORDER BY {order_col}" + + try: + results = client.execute_query(query) + df = results.to_dataframe() + if df.empty: + return [] + return [normalize_monitoring_row(row.to_dict()) for _, row in df.iterrows()] + except Exception: + return [] + + @staticmethod + def clear_monitoring_baseline( + config: RepoConfig, + project: str, + feature_view_name: Optional[str] = None, + feature_name: Optional[str] = None, + data_source_type: Optional[str] = None, + ) -> None: + assert isinstance(config.offline_store, TrinoOfflineStoreConfig) + client = _get_trino_client(config=config) + full_table_name = _trino_monitoring_table_name(config, MON_TABLE_FEATURE) + + try: + results = client.execute_query(f"SELECT * FROM {full_table_name}") + pdf = results.to_dataframe() + except Exception: + return + + if pdf.empty: + return + + mask = (pdf["project_id"] == project) & (pdf["is_baseline"] == True) # noqa: E712 + if feature_view_name is not None: + mask &= pdf["feature_view_name"] == feature_view_name + if feature_name is not None: + mask &= pdf["feature_name"] == feature_name + if data_source_type is not None: + mask &= pdf["data_source_type"] == data_source_type + + if not mask.any(): + return + + pdf.loc[mask, "is_baseline"] = False + try: + client.execute_query(f"DROP TABLE IF EXISTS {full_table_name}") + except Exception: + pass + + upload_pandas_dataframe_to_trino( + client=client, + df=pdf, + table=full_table_name, + connector_args=config.offline_store.connector, + ) + + +def _trino_monitoring_table_name(config: RepoConfig, table: str) -> str: + catalog = config.offline_store.catalog + dataset = config.offline_store.dataset + if dataset: + return f"{catalog}.{dataset}.{table}" + return f"{catalog}.{table}" + + +def _trino_table_with_clause(config: RepoConfig) -> str: + connector_args = config.offline_store.connector or {} + connector_type = connector_args.get("type", "") + if connector_type in {"hive", "iceberg"}: + file_format = connector_args.get("file_format", "parquet") + return f"WITH (format = '{file_format}')" + return "" + + +def _trino_normalize_histogram_column(pdf: pd.DataFrame) -> pd.DataFrame: + if "histogram" not in pdf.columns: + return pdf + out = pdf.copy() + + def _ser(x: Any) -> Any: + if x is None: + return None + if isinstance(x, str): + return x + return json.dumps(x) + + out["histogram"] = out["histogram"].map(_ser) + return out + + +def _trino_pandas_upsert( + pdf_old: pd.DataFrame, + pdf_new: pd.DataFrame, + pk_columns: List[str], +) -> pd.DataFrame: + if pdf_old.empty: + return pdf_new + pk_cols_present = [ + c for c in pk_columns if c in pdf_old.columns and c in pdf_new.columns + ] + if not pk_cols_present: + return pd.concat([pdf_old, pdf_new], ignore_index=True) + old_idx = pdf_old.set_index(pk_cols_present) + new_idx = pdf_new.set_index(pk_cols_present) + kept = old_idx.loc[~old_idx.index.isin(new_idx.index)] + kept_df = kept.reset_index() + return pd.concat([kept_df, pdf_new], ignore_index=True) + + +def _trino_sql_literal(val: Any) -> str: + if val is None: + return "NULL" + if isinstance(val, (bool, np.bool_)): + return "TRUE" if val else "FALSE" + if isinstance(val, (int, float, np.integer, np.floating)): + return str(val) + if isinstance(val, (datetime, pd.Timestamp)): + return f"TIMESTAMP '{val.strftime('%Y-%m-%d %H:%M:%S.%f')}'" + if isinstance(val, date): + return f"DATE '{val.strftime('%Y-%m-%d')}'" + escaped = str(val).replace("'", "''") + return f"'{escaped}'" + + +def _trino_sql_numeric_stats( + client: Trino, + from_expression: str, + feature_names: List[str], + ts_clause: str, + histogram_bins: int, +) -> List[Dict[str, Any]]: + select_parts = ["COUNT(*)"] + for col in feature_names: + q = f'"{col}"' + c = f"CAST({q} AS DOUBLE)" + select_parts.extend( + [ + f"COUNT({q})", + f"AVG({c})", + f"STDDEV_SAMP({c})", + f"MIN({c})", + f"MAX({c})", + f"APPROX_PERCENTILE({c}, 0.50)", + f"APPROX_PERCENTILE({c}, 0.75)", + f"APPROX_PERCENTILE({c}, 0.90)", + f"APPROX_PERCENTILE({c}, 0.95)", + f"APPROX_PERCENTILE({c}, 0.99)", + ] + ) + + query = ( + f"SELECT {', '.join(select_parts)} " + f"FROM {from_expression} AS _src WHERE {ts_clause}" + ) + results = client.execute_query(query) + rows = results.data + if not rows or rows[0] is None or rows[0][0] is None: + return [empty_numeric_metric(n) for n in feature_names] + + row = rows[0] + row_count = int(row[0] or 0) + metric_results: List[Dict[str, Any]] = [] + + for i, col in enumerate(feature_names): + base = 1 + i * 10 + non_null = int(row[base] or 0) + null_count = row_count - non_null + + min_val = opt_float(row[base + 3]) + max_val = opt_float(row[base + 4]) + + result: Dict[str, Any] = { + "feature_name": col, + "feature_type": "numeric", + "row_count": row_count, + "null_count": null_count, + "null_rate": null_count / row_count if row_count > 0 else 0.0, + "mean": opt_float(row[base + 1]), + "stddev": opt_float(row[base + 2]), + "min_val": min_val, + "max_val": max_val, + "p50": opt_float(row[base + 5]), + "p75": opt_float(row[base + 6]), + "p90": opt_float(row[base + 7]), + "p95": opt_float(row[base + 8]), + "p99": opt_float(row[base + 9]), + "histogram": None, + } + + if min_val is not None and max_val is not None and non_null > 0: + result["histogram"] = _trino_sql_numeric_histogram( + client, + from_expression, + col, + ts_clause, + histogram_bins, + min_val, + max_val, + ) + + metric_results.append(result) + + return metric_results + + +def _trino_sql_numeric_histogram( + client: Trino, + from_expression: str, + col_name: str, + ts_clause: str, + bins: int, + min_val: float, + max_val: float, +) -> Dict[str, Any]: + q_col = f'"{col_name}"' + + if min_val == max_val: + sql = ( + f"SELECT COUNT(*) FROM {from_expression} AS _src " + f"WHERE {q_col} IS NOT NULL AND {ts_clause}" + ) + res = client.execute_query(sql) + cnt = int(res.data[0][0] or 0) if res.data and res.data[0] else 0 + return {"bins": [min_val, max_val], "counts": [cnt], "bin_width": 0.0} + + bin_width = (max_val - min_val) / bins + cast_col = f"CAST({q_col} AS DOUBLE)" + inner = ( + f"CASE WHEN {min_val} = {max_val} THEN CAST(1 AS BIGINT) " + f"ELSE LEAST(GREATEST(CAST(FLOOR(({cast_col} - {min_val}) / {bin_width}) + 1 AS BIGINT), CAST(1 AS BIGINT)), CAST({bins} AS BIGINT)) " + f"END AS bucket" + ) + + query = ( + f"SELECT bucket, COUNT(*) AS cnt FROM (" + f" SELECT {inner} " + f" FROM {from_expression} AS _src " + f" WHERE {q_col} IS NOT NULL AND {ts_clause}" + f") AS _b WHERE bucket IS NOT NULL " + f"GROUP BY bucket ORDER BY bucket" + ) + res = client.execute_query(query) + hrows = res.data or [] + counts = [0] * bins + for hr in hrows: + bucket = int(hr[0] or 0) + cnt = int(hr[1] or 0) + if 1 <= bucket <= bins: + counts[bucket - 1] = cnt + + bin_edges = [min_val + i * bin_width for i in range(bins + 1)] + return { + "bins": [float(b) for b in bin_edges], + "counts": counts, + "bin_width": float(bin_width), + } + + +def _trino_sql_categorical_stats( + client: Trino, + from_expression: str, + col_name: str, + ts_clause: str, + top_n: int, +) -> Dict[str, Any]: + q_col = f'"{col_name}"' + + query = ( + f"WITH filtered AS (" + f" SELECT * FROM {from_expression} AS _src WHERE {ts_clause}" + f") " + f"SELECT " + f" (SELECT COUNT(*) FROM filtered) AS row_count, " + f" (SELECT COUNT(*) - COUNT({q_col}) FROM filtered) AS null_count, " + f" (SELECT COUNT(DISTINCT {q_col}) FROM filtered " + f" WHERE {q_col} IS NOT NULL) AS unique_count, " + f" CAST({q_col} AS VARCHAR) AS value, COUNT(*) AS cnt " + f"FROM filtered WHERE {q_col} IS NOT NULL " + f"GROUP BY {q_col} ORDER BY cnt DESC LIMIT {int(top_n)}" + ) + + res = client.execute_query(query) + rows = res.data or [] + if not rows: + return empty_categorical_metric(col_name) + + row_count = int(rows[0][0] or 0) + null_count = int(rows[0][1] or 0) + unique_count = int(rows[0][2] or 0) + + top_entries = [{"value": r[3], "count": int(r[4] or 0)} for r in rows] + top_total = sum(e["count"] for e in top_entries) + other_count = (row_count - null_count) - top_total + + return { + "feature_name": col_name, + "feature_type": "categorical", + "row_count": row_count, + "null_count": null_count, + "null_rate": null_count / row_count if row_count > 0 else 0.0, + "mean": None, + "stddev": None, + "min_val": None, + "max_val": None, + "p50": None, + "p75": None, + "p90": None, + "p95": None, + "p99": None, + "histogram": { + "values": top_entries, + "other_count": max(other_count, 0), + "unique_count": unique_count, + }, + } + + +_TRINO_MONITORING_DDL_STATEMENTS = [ + """ +CREATE TABLE IF NOT EXISTS {table} ( + project_id VARCHAR, + feature_view_name VARCHAR, + feature_name VARCHAR, + metric_date DATE, + granularity VARCHAR, + data_source_type VARCHAR, + computed_at TIMESTAMP, + max_event_timestamp TIMESTAMP, + is_baseline BOOLEAN, + feature_type VARCHAR, + row_count BIGINT, + null_count BIGINT, + null_rate DOUBLE, + mean DOUBLE, + stddev DOUBLE, + min_val DOUBLE, + max_val DOUBLE, + p50 DOUBLE, + p75 DOUBLE, + p90 DOUBLE, + p95 DOUBLE, + p99 DOUBLE, + histogram VARCHAR +) {with_clause} +""", + """ +CREATE TABLE IF NOT EXISTS {table} ( + project_id VARCHAR, + feature_view_name VARCHAR, + metric_date DATE, + granularity VARCHAR, + data_source_type VARCHAR, + computed_at TIMESTAMP, + max_event_timestamp TIMESTAMP, + is_baseline BOOLEAN, + total_row_count BIGINT, + total_features INTEGER, + features_with_nulls INTEGER, + avg_null_rate DOUBLE, + max_null_rate DOUBLE +) {with_clause} +""", + """ +CREATE TABLE IF NOT EXISTS {table} ( + project_id VARCHAR, + feature_service_name VARCHAR, + metric_date DATE, + granularity VARCHAR, + data_source_type VARCHAR, + computed_at TIMESTAMP, + max_event_timestamp TIMESTAMP, + is_baseline BOOLEAN, + total_feature_views INTEGER, + total_features INTEGER, + avg_null_rate DOUBLE, + max_null_rate DOUBLE +) {with_clause} +""", + """ +CREATE TABLE IF NOT EXISTS {table} ( + job_id VARCHAR, + project_id VARCHAR, + feature_view_name VARCHAR, + job_type VARCHAR, + status VARCHAR, + parameters VARCHAR, + metric_date DATE, + started_at TIMESTAMP, + completed_at TIMESTAMP, + error_message VARCHAR, + result_summary VARCHAR +) {with_clause} +""", +] + def _get_table_reference_for_new_entity( catalog: str, diff --git a/sdk/python/tests/unit/monitoring/test_compute_correctness.py b/sdk/python/tests/unit/monitoring/test_compute_correctness.py index 9f9005380df..4ab913b4376 100644 --- a/sdk/python/tests/unit/monitoring/test_compute_correctness.py +++ b/sdk/python/tests/unit/monitoring/test_compute_correctness.py @@ -1702,6 +1702,274 @@ def mock_fetchall(con, sql): assert results[1]["null_count"] == 5 +def _trino_importable(): + try: + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_numeric_stats, # noqa: F401 + ) + + return True + except ImportError: + return False + + +@pytest.mark.skipif(not _trino_importable(), reason="Trino deps not installed") +class TestTrinoComputeCorrectness: + """Tests Trino result parsing with mocked Trino client. + + Trino execute_query returns Results with data: List[List[Any]]. + """ + + def test_numeric_stats(self): + from unittest.mock import MagicMock + + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_numeric_stats, + ) + from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, + ) + + vals = NUMERIC_VALUES + row = [ + 10, + 10, + statistics.mean(vals), + statistics.stdev(vals), + 1.0, + 10.0, + 5.5, + 7.75, + 9.1, + 9.55, + 9.91, + ] + hist_rows = [[i + 1, 2] for i in range(5)] + + mock_client = MagicMock() + call_count = [0] + + def mock_execute(query_text): + call_count[0] += 1 + if call_count[0] == 1: + return Results(data=[row], columns=[{"name": "col", "type": "double"}]) + return Results( + data=hist_rows, + columns=[ + {"name": "bucket", "type": "bigint"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + + mock_client.execute_query.side_effect = mock_execute + + results = _trino_sql_numeric_stats( + mock_client, + "test_table", + ["numeric_col"], + "1=1", + histogram_bins=5, + ) + + assert len(results) == 1 + r = results[0] + expected = _expected_numeric_stats() + assert_numeric_correctness(r, expected, "trino_numeric") + assert r["histogram"] is not None + assert sum(r["histogram"]["counts"]) == 10 + + def test_numeric_stats_with_nulls(self): + from unittest.mock import MagicMock + + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_numeric_stats, + ) + from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, + ) + + vals = NON_NULL_VALUES + row = [ + 10, + 5, + statistics.mean(vals), + statistics.stdev(vals), + 1.0, + 9.0, + 5.0, + 7.0, + 8.6, + 8.8, + 8.96, + ] + hist_rows = [[i + 1, 1] for i in range(5)] + + mock_client = MagicMock() + call_count = [0] + + def mock_execute(query_text): + call_count[0] += 1 + if call_count[0] == 1: + return Results(data=[row], columns=[{"name": "col", "type": "double"}]) + return Results( + data=hist_rows, + columns=[ + {"name": "bucket", "type": "bigint"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + + mock_client.execute_query.side_effect = mock_execute + + results = _trino_sql_numeric_stats( + mock_client, + "t", + ["col"], + "1=1", + histogram_bins=5, + ) + + r = results[0] + assert r["null_count"] == 5 + assert r["null_rate"] == pytest.approx(0.5) + assert r["mean"] == pytest.approx(5.0, abs=1e-4) + + def test_categorical_stats(self): + from unittest.mock import MagicMock + + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_categorical_stats, + ) + from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, + ) + + rows = [ + [10, 0, 4, "a", 4], + [10, 0, 4, "b", 3], + [10, 0, 4, "c", 2], + [10, 0, 4, "d", 1], + ] + mock_client = MagicMock() + mock_client.execute_query.return_value = Results( + data=rows, + columns=[ + {"name": "row_count", "type": "bigint"}, + {"name": "null_count", "type": "bigint"}, + {"name": "unique_count", "type": "bigint"}, + {"name": "value", "type": "varchar"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + + result = _trino_sql_categorical_stats( + mock_client, + "t", + "cat_col", + "1=1", + top_n=10, + ) + + expected = _expected_categorical_stats() + assert_categorical_correctness(result, expected, "trino_categorical") + + def test_empty_result(self): + from unittest.mock import MagicMock + + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_numeric_stats, + ) + from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, + ) + + mock_client = MagicMock() + mock_client.execute_query.return_value = Results( + data=[], + columns=[], + ) + + results = _trino_sql_numeric_stats( + mock_client, + "t", + ["col"], + "1=1", + histogram_bins=5, + ) + + assert len(results) == 1 + assert results[0]["mean"] is None + assert results[0]["row_count"] == 0 + + def test_multiple_features(self): + from unittest.mock import MagicMock + + from feast.infra.offline_stores.contrib.trino_offline_store.trino import ( + _trino_sql_numeric_stats, + ) + from feast.infra.offline_stores.contrib.trino_offline_store.trino_queries import ( + Results, + ) + + row = [ + 10, + # Feature 0: numeric_col + 10, + 5.5, + 3.03, + 1.0, + 10.0, + 5.5, + 7.75, + 9.1, + 9.55, + 9.91, + # Feature 1: numeric_with_nulls + 5, + 5.0, + 3.16, + 1.0, + 9.0, + 5.0, + 7.0, + 8.6, + 8.8, + 8.96, + ] + hist_rows = [[i + 1, 2] for i in range(5)] + + mock_client = MagicMock() + call_count = [0] + + def mock_execute(query_text): + call_count[0] += 1 + if call_count[0] == 1: + return Results(data=[row], columns=[{"name": "col", "type": "double"}]) + return Results( + data=hist_rows, + columns=[ + {"name": "bucket", "type": "bigint"}, + {"name": "cnt", "type": "bigint"}, + ], + ) + + mock_client.execute_query.side_effect = mock_execute + + results = _trino_sql_numeric_stats( + mock_client, + "t", + ["col_a", "col_b"], + "1=1", + histogram_bins=5, + ) + + assert len(results) == 2 + assert results[0]["mean"] == pytest.approx(5.5, abs=1e-2) + assert results[1]["mean"] == pytest.approx(5.0, abs=1e-2) + assert results[0]["null_count"] == 0 + assert results[1]["null_count"] == 5 + + # =================================================================== # Cross-backend consistency: MetricsCalculator vs DuckDB vs Dask # ===================================================================