Skip to content

Commit f0bc070

Browse files
jyejarecursoragent
jyejare
and
cursoragent
authored
fix: Feature freshness logic updated (#6765)
* fix: feature freshness logic updated Signed-off-by: Jitendra Yejare <[email protected]> * fix: Avoid redundant _get_max_timestamp query in _compute_for_feature_view Move the _get_max_timestamp call to compute_metrics and pass the result into _compute_for_feature_view, eliminating a duplicate data-source query per feature view. Signed-off-by: Jitendra Yejare <[email protected]> Co-authored-by: Cursor <[email protected]> --------- Signed-off-by: Jitendra Yejare <[email protected]> Co-authored-by: Cursor <[email protected]>
1 parent 03dac6f commit f0bc070

15 files changed

Lines changed: 281 additions & 20 deletions

File tree

‎docs/how-to-guides/feature-monitoring.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,7 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta
219219
"feature_name": "conv_rate",
220220
"feature_type": "numeric",
221221
"metric_date": "2025-03-26",
222+
"max_event_timestamp": "2025-03-27T14:30:00+00:00",
222223
"granularity": "daily",
223224
"data_source_type": "batch",
224225
"row_count": 15000,
@@ -242,6 +243,8 @@ GET /monitoring/metrics/features?project=my_project&feature_view_name=driver_sta
242243
]
243244
```
244245

246+
The UI **Freshness** column uses `max_event_timestamp` — `MAX(event_timestamp)` from the source — not `metric_date` (the DQM window start). Age is `now − max_event_timestamp`.
247+
245248
### Per-feature-view aggregates
246249

247250
```

‎sdk/python/feast/infra/offline_stores/bigquery.py‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -693,7 +693,7 @@ def _bq_scalar_param_type(column: str) -> str:
693693
return "BOOL"
694694
if column == "metric_date":
695695
return "DATE"
696-
if column == "computed_at":
696+
if column in ("computed_at", "max_event_timestamp"):
697697
return "TIMESTAMP"
698698
if column in {
699699
"row_count",
@@ -889,6 +889,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
889889
granularity STRING NOT NULL,
890890
data_source_type STRING NOT NULL,
891891
computed_at TIMESTAMP NOT NULL,
892+
max_event_timestamp TIMESTAMP,
892893
is_baseline BOOL NOT NULL,
893894
feature_type STRING NOT NULL,
894895
row_count INT64,
@@ -915,6 +916,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
915916
granularity STRING NOT NULL,
916917
data_source_type STRING NOT NULL,
917918
computed_at TIMESTAMP NOT NULL,
919+
max_event_timestamp TIMESTAMP,
918920
is_baseline BOOL NOT NULL,
919921
total_row_count INT64,
920922
total_features INT64,
@@ -932,6 +934,7 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
932934
granularity STRING NOT NULL,
933935
data_source_type STRING NOT NULL,
934936
computed_at TIMESTAMP NOT NULL,
937+
max_event_timestamp TIMESTAMP,
935938
is_baseline BOOL NOT NULL,
936939
total_feature_views INT64,
937940
total_features INT64,
@@ -958,6 +961,11 @@ def _bq_ensure_monitoring_tables(config: RepoConfig) -> None:
958961
"""
959962
for ddl in (feature_ddl, view_ddl, service_ddl, job_ddl):
960963
client.query(ddl).result()
964+
for tbl in (MON_TABLE_FEATURE, MON_TABLE_FEATURE_VIEW, MON_TABLE_FEATURE_SERVICE):
965+
client.query(
966+
f"ALTER TABLE `{proj}.{ds}.{tbl}` "
967+
"ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMP"
968+
).result()
961969

962970

963971
def _bq_get_monitoring_max_timestamp(

‎sdk/python/feast/infra/offline_stores/contrib/oracle_offline_store/oracle.py‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -715,6 +715,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
715715
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
716716
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
717717
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
718+
max_event_timestamp TIMESTAMP WITH TIME ZONE,
718719
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
719720
feature_type VARCHAR2(50) NOT NULL,
720721
row_count NUMBER,
@@ -746,6 +747,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
746747
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
747748
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
748749
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
750+
max_event_timestamp TIMESTAMP WITH TIME ZONE,
749751
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
750752
total_row_count NUMBER,
751753
total_features NUMBER,
@@ -768,6 +770,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
768770
granularity VARCHAR2(20) DEFAULT 'daily' NOT NULL,
769771
data_source_type VARCHAR2(50) DEFAULT 'batch' NOT NULL,
770772
computed_at TIMESTAMP WITH TIME ZONE DEFAULT SYSTIMESTAMP NOT NULL,
773+
max_event_timestamp TIMESTAMP WITH TIME ZONE,
771774
is_baseline NUMBER(1) DEFAULT 0 NOT NULL,
772775
total_feature_views NUMBER,
773776
total_features NUMBER,
@@ -779,6 +782,19 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
779782
""",
780783
)
781784

785+
_oracle_try_execute_ddl(
786+
con,
787+
f"ALTER TABLE {MON_TABLE_FEATURE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
788+
)
789+
_oracle_try_execute_ddl(
790+
con,
791+
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
792+
)
793+
_oracle_try_execute_ddl(
794+
con,
795+
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD (max_event_timestamp TIMESTAMP WITH TIME ZONE)",
796+
)
797+
782798
_oracle_try_execute_ddl(
783799
con,
784800
f"""

‎sdk/python/feast/infra/offline_stores/contrib/postgres_offline_store/postgres.py‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -429,6 +429,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
429429
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
430430
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
431431
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
432+
max_event_timestamp TIMESTAMPTZ,
432433
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
433434
feature_type VARCHAR(50) NOT NULL,
434435
row_count BIGINT,
@@ -468,6 +469,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
468469
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
469470
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
470471
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
472+
max_event_timestamp TIMESTAMPTZ,
471473
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
472474
total_row_count BIGINT,
473475
total_features INTEGER,
@@ -487,6 +489,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
487489
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
488490
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
489491
computed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
492+
max_event_timestamp TIMESTAMPTZ,
490493
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
491494
total_feature_views INTEGER,
492495
total_features INTEGER,
@@ -497,6 +500,15 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
497500
);
498501
""")
499502

503+
cur.execute(f"""
504+
ALTER TABLE {MON_TABLE_FEATURE}
505+
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
506+
ALTER TABLE {MON_TABLE_FEATURE_VIEW}
507+
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
508+
ALTER TABLE {MON_TABLE_FEATURE_SERVICE}
509+
ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ;
510+
""")
511+
500512
cur.execute(f"""
501513
CREATE TABLE IF NOT EXISTS {MON_TABLE_JOB} (
502514
job_id VARCHAR(36) PRIMARY KEY,

‎sdk/python/feast/infra/offline_stores/contrib/spark_offline_store/spark.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -574,6 +574,18 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
574574
)
575575
for stmt in _SPARK_MONITORING_DDL_STATEMENTS:
576576
spark_session.sql(stmt)
577+
from pyspark.sql.utils import AnalysisException
578+
579+
for stmt in (
580+
f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
581+
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
582+
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMNS (max_event_timestamp TIMESTAMP)",
583+
):
584+
try:
585+
spark_session.sql(stmt)
586+
except AnalysisException:
587+
# Column already exists on newly created tables.
588+
pass
577589

578590
@staticmethod
579591
def save_monitoring_metrics(
@@ -678,6 +690,7 @@ def clear_monitoring_baseline(
678690
granularity STRING NOT NULL,
679691
data_source_type STRING NOT NULL,
680692
computed_at TIMESTAMP NOT NULL,
693+
max_event_timestamp TIMESTAMP,
681694
is_baseline BOOLEAN NOT NULL,
682695
feature_type STRING NOT NULL,
683696
row_count BIGINT,
@@ -703,6 +716,7 @@ def clear_monitoring_baseline(
703716
granularity STRING NOT NULL,
704717
data_source_type STRING NOT NULL,
705718
computed_at TIMESTAMP NOT NULL,
719+
max_event_timestamp TIMESTAMP,
706720
is_baseline BOOLEAN NOT NULL,
707721
total_row_count BIGINT,
708722
total_features INT,
@@ -719,6 +733,7 @@ def clear_monitoring_baseline(
719733
granularity STRING NOT NULL,
720734
data_source_type STRING NOT NULL,
721735
computed_at TIMESTAMP NOT NULL,
736+
max_event_timestamp TIMESTAMP,
722737
is_baseline BOOLEAN NOT NULL,
723738
total_feature_views INT,
724739
total_features INT,

‎sdk/python/feast/infra/offline_stores/dask.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,7 @@
4040
from feast.monitoring.monitoring_utils import (
4141
MONITORING_DIR,
4242
MONITORING_PARQUET_FILES,
43+
MONITORING_TIMESTAMP_FIELDS,
4344
monitoring_parquet_meta,
4445
normalize_monitoring_row,
4546
opt_float,
@@ -968,7 +969,7 @@ def _dask_parquet_query(
968969
for _, row in df.iterrows():
969970
record = {c: row.get(c) for c in columns}
970971
normalize_monitoring_row(record)
971-
for key in ("metric_date", "computed_at"):
972+
for key in MONITORING_TIMESTAMP_FIELDS:
972973
val = record.get(key)
973974
if (
974975
val is not None

‎sdk/python/feast/infra/offline_stores/duckdb.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
from feast.monitoring.monitoring_utils import (
4040
MONITORING_DIR,
4141
MONITORING_PARQUET_FILES,
42+
MONITORING_TIMESTAMP_FIELDS,
4243
empty_categorical_metric,
4344
empty_numeric_metric,
4445
monitoring_parquet_meta,
@@ -487,7 +488,7 @@ def _duckdb_parquet_query(
487488
for _, row in df.iterrows():
488489
record = {c: row.get(c) for c in columns}
489490
normalize_monitoring_row(record)
490-
for key in ("metric_date", "computed_at"):
491+
for key in MONITORING_TIMESTAMP_FIELDS:
491492
val = record.get(key)
492493
if (
493494
val is not None

‎sdk/python/feast/infra/offline_stores/redshift.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -555,6 +555,7 @@ def clear_monitoring_baseline(
555555
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
556556
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
557557
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
558+
max_event_timestamp TIMESTAMPTZ,
558559
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
559560
feature_type VARCHAR(50) NOT NULL,
560561
row_count BIGINT,
@@ -582,6 +583,7 @@ def clear_monitoring_baseline(
582583
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
583584
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
584585
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
586+
max_event_timestamp TIMESTAMPTZ,
585587
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
586588
total_row_count BIGINT,
587589
total_features INTEGER,
@@ -600,6 +602,7 @@ def clear_monitoring_baseline(
600602
granularity VARCHAR(20) NOT NULL DEFAULT 'daily',
601603
data_source_type VARCHAR(50) NOT NULL DEFAULT 'batch',
602604
computed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
605+
max_event_timestamp TIMESTAMPTZ,
603606
is_baseline BOOLEAN NOT NULL DEFAULT FALSE,
604607
total_feature_views INTEGER,
605608
total_features INTEGER,
@@ -625,6 +628,9 @@ def clear_monitoring_baseline(
625628
PRIMARY KEY (job_id)
626629
);
627630
""",
631+
f"ALTER TABLE {MON_TABLE_FEATURE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
632+
f"ALTER TABLE {MON_TABLE_FEATURE_VIEW} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
633+
f"ALTER TABLE {MON_TABLE_FEATURE_SERVICE} ADD COLUMN IF NOT EXISTS max_event_timestamp TIMESTAMPTZ",
628634
]
629635

630636

‎sdk/python/feast/infra/offline_stores/snowflake.py‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -564,6 +564,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
564564
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
565565
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
566566
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
567+
"max_event_timestamp" TIMESTAMP_TZ,
567568
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
568569
"feature_type" VARCHAR(50) NOT NULL,
569570
"row_count" BIGINT,
@@ -591,6 +592,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
591592
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
592593
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
593594
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
595+
"max_event_timestamp" TIMESTAMP_TZ,
594596
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
595597
"total_row_count" BIGINT,
596598
"total_features" INTEGER,
@@ -609,6 +611,7 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
609611
"granularity" VARCHAR(20) NOT NULL DEFAULT 'daily',
610612
"data_source_type" VARCHAR(50) NOT NULL DEFAULT 'batch',
611613
"computed_at" TIMESTAMP_TZ NOT NULL DEFAULT CURRENT_TIMESTAMP(),
614+
"max_event_timestamp" TIMESTAMP_TZ,
612615
"is_baseline" BOOLEAN NOT NULL DEFAULT FALSE,
613616
"total_feature_views" INTEGER,
614617
"total_features" INTEGER,
@@ -642,6 +645,11 @@ def ensure_monitoring_tables(config: RepoConfig) -> None:
642645
execute_snowflake_statement(conn, ddl_view)
643646
execute_snowflake_statement(conn, ddl_service)
644647
execute_snowflake_statement(conn, ddl_job)
648+
for fq in (fq_feature, fq_view, fq_service):
649+
execute_snowflake_statement(
650+
conn,
651+
f'ALTER TABLE {fq} ADD COLUMN IF NOT EXISTS "max_event_timestamp" TIMESTAMP_TZ',
652+
)
645653

646654
@staticmethod
647655
def save_monitoring_metrics(

0 commit comments

Comments
 (0)