diff --git a/docs/reference/feature-store-yaml.md b/docs/reference/feature-store-yaml.md index 01f586c047b..bac483a2ebc 100644 --- a/docs/reference/feature-store-yaml.md +++ b/docs/reference/feature-store-yaml.md @@ -24,7 +24,6 @@ online_store: * **path** \(a local filepath\) — Parameter for the sqlite online store. Defines the path to the SQLite database file. * **project\_id** — Optional parameter for the datastore online store. Sets the GCP project id used by Feast, if not set Feast will use the default GCP project id in the local environment. * **project** — Defines a namespace for the entire feature store. Can be used to isolate multiple deployments in a single installation of Feast. - ## Providers The `provider` field defines the environment in which Feast will execute data flows. As a result, it also determines the default values for other fields. diff --git a/go/internal/feast/integration_tests/scylladb/feature_repo/regression_repo.py b/go/internal/feast/integration_tests/scylladb/feature_repo/regression_repo.py new file mode 100644 index 00000000000..d2163d2c344 --- /dev/null +++ b/go/internal/feast/integration_tests/scylladb/feature_repo/regression_repo.py @@ -0,0 +1,89 @@ +# Regression fixture for EAPC-22316 follow-up: a SortedFeatureView whose +# UNIX_TIMESTAMP sort key is requested as a feature must retain millisecond +# precision for rows that differ by less than one second (Go read-side fix in +# InterfaceToProtoValue, go/types/typeconversion.go). + +from datetime import timedelta + +from feast import Entity, Field, FileSource, SortedFeatureView +from feast.protos.feast.core.SortedFeatureView_pb2 import SortOrder +from feast.sort_key import SortKey +from feast.types import String, UnixTimestamp +from feast.value_type import ValueType + +tags = {"team": "Feast"} +owner = "test@test.com" + +sub_second_entity: Entity = Entity( + name="sub_second_entity", + join_keys=["sub_second_entity_id"], + value_type=ValueType.STRING, + tags=tags, + owner=owner, +) + +sub_second_source: FileSource = FileSource( + path="sub_second_data.parquet", timestamp_field="event_timestamp" +) + +sub_second_sort_key_view: SortedFeatureView = SortedFeatureView( + name="sub_second_sort_key_view", + entities=[sub_second_entity], + ttl=timedelta(days=0), + source=sub_second_source, + tags=tags, + description="Regression fixture: sort key rows <1s apart for the same entity", + owner=owner, + online=True, + sort_keys=[ + SortKey( + name="event_timestamp", + value_type=ValueType.UNIX_TIMESTAMP, + default_sort_order=SortOrder.DESC, + ) + ], + schema=[ + Field(name="value", dtype=String), + Field(name="event_timestamp", dtype=UnixTimestamp), + ], +) + +# Second regression fixture: the sort key is a secondary feature with a name +# other than "event_timestamp" (mirroring the real customer schema, where the +# sort key was named "viewed_at" and was distinct from the source's own +# ingestion timestamp field). This guards against the fix accidentally being +# tied to the "event_timestamp" column name rather than working for any +# UNIX_TIMESTAMP column declared as a sort key. +custom_sortkey_entity: Entity = Entity( + name="custom_sortkey_entity", + join_keys=["custom_sortkey_entity_id"], + value_type=ValueType.STRING, + tags=tags, + owner=owner, +) + +sub_second_custom_sortkey_source: FileSource = FileSource( + path="sub_second_custom_sortkey_data.parquet", timestamp_field="event_timestamp" +) + +sub_second_custom_sortkey_view: SortedFeatureView = SortedFeatureView( + name="sub_second_custom_sortkey_view", + entities=[custom_sortkey_entity], + ttl=timedelta(days=0), + source=sub_second_custom_sortkey_source, + tags=tags, + description="Regression fixture: sort key named 'viewed_at' (not 'event_timestamp'), rows <1s apart", + owner=owner, + online=True, + sort_keys=[ + SortKey( + name="viewed_at", + value_type=ValueType.UNIX_TIMESTAMP, + default_sort_order=SortOrder.DESC, + ) + ], + schema=[ + Field(name="value", dtype=String), + Field(name="viewed_at", dtype=UnixTimestamp), + ], +) diff --git a/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_custom_sortkey_data.parquet b/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_custom_sortkey_data.parquet new file mode 100644 index 00000000000..a9193905dd7 Binary files /dev/null and b/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_custom_sortkey_data.parquet differ diff --git a/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_data.parquet b/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_data.parquet new file mode 100644 index 00000000000..a10514da6c5 Binary files /dev/null and b/go/internal/feast/integration_tests/scylladb/feature_repo/sub_second_data.parquet differ diff --git a/go/internal/feast/integration_tests/scylladb/http/http_integration_test.go b/go/internal/feast/integration_tests/scylladb/http/http_integration_test.go index 4bb1d3e3037..4dd0accfad9 100644 --- a/go/internal/feast/integration_tests/scylladb/http/http_integration_test.go +++ b/go/internal/feast/integration_tests/scylladb/http/http_integration_test.go @@ -215,7 +215,7 @@ func TestGetOnlineFeaturesRange_Http_withOnlyEqualsFilter(t *testing.T) { "sort_key_filters": [ { "sort_key_name": "event_timestamp", - "equals": 1744769171 + "equals": 1744769171919 } ], "limit": 10 diff --git a/go/internal/feast/integration_tests/scylladb/http/valid_equals_response.json b/go/internal/feast/integration_tests/scylladb/http/valid_equals_response.json index 5f837738990..be7664a120c 100644 --- a/go/internal/feast/integration_tests/scylladb/http/valid_equals_response.json +++ b/go/internal/feast/integration_tests/scylladb/http/valid_equals_response.json @@ -357,7 +357,7 @@ { "values": [ [ - "2025-04-16 02:06:11Z" + "2025-04-16 02:06:11.919Z" ] ] } diff --git a/go/internal/feast/integration_tests/scylladb/http/valid_response.json b/go/internal/feast/integration_tests/scylladb/http/valid_response.json index 55db0470247..5170162c712 100644 --- a/go/internal/feast/integration_tests/scylladb/http/valid_response.json +++ b/go/internal/feast/integration_tests/scylladb/http/valid_response.json @@ -70,6 +70,6 @@ }, { "values" : [ [ null, null, null, null, null, null, null, null, null, null ], [ null, null, null, null, null, null, null, null, null, null ], [ null, null, null, null, null, null, null, null, null, null ] ] }, { - "values" : [ [ "2025-04-16 02:06:10Z", "2025-04-16 02:06:09Z", "2025-04-16 02:06:08Z", "2025-04-16 02:06:07Z", "2025-04-16 02:06:06Z", "2025-04-16 02:06:05Z", "2025-04-16 02:06:04Z", "2025-04-16 02:06:03Z", "2025-04-16 02:06:02Z", "2025-04-16 02:06:01Z" ], [ "2025-04-16 02:06:11Z", "2025-04-16 02:06:10Z", "2025-04-16 02:06:09Z", "2025-04-16 02:06:08Z", "2025-04-16 02:06:07Z", "2025-04-16 02:06:06Z", "2025-04-16 02:06:05Z", "2025-04-16 02:06:04Z", "2025-04-16 02:06:03Z", "2025-04-16 02:06:02Z" ], [ "2025-04-16 02:06:11Z", "2025-04-16 02:06:10Z", "2025-04-16 02:06:09Z", "2025-04-16 02:06:08Z", "2025-04-16 02:06:07Z", "2025-04-16 02:06:06Z", "2025-04-16 02:06:05Z", "2025-04-16 02:06:04Z", "2025-04-16 02:06:03Z", "2025-04-16 02:06:02Z" ] ] + "values" : [ [ "2025-04-16 02:06:10.919Z", "2025-04-16 02:06:09.919Z", "2025-04-16 02:06:08.919Z", "2025-04-16 02:06:07.919Z", "2025-04-16 02:06:06.919Z", "2025-04-16 02:06:05.919Z", "2025-04-16 02:06:04.919Z", "2025-04-16 02:06:03.919Z", "2025-04-16 02:06:02.919Z", "2025-04-16 02:06:01.919Z" ], [ "2025-04-16 02:06:11.919Z", "2025-04-16 02:06:10.919Z", "2025-04-16 02:06:09.919Z", "2025-04-16 02:06:08.919Z", "2025-04-16 02:06:07.919Z", "2025-04-16 02:06:06.919Z", "2025-04-16 02:06:05.919Z", "2025-04-16 02:06:04.919Z", "2025-04-16 02:06:03.919Z", "2025-04-16 02:06:02.919Z" ], [ "2025-04-16 02:06:11.919Z", "2025-04-16 02:06:10.919Z", "2025-04-16 02:06:09.919Z", "2025-04-16 02:06:08.919Z", "2025-04-16 02:06:07.919Z", "2025-04-16 02:06:06.919Z", "2025-04-16 02:06:05.919Z", "2025-04-16 02:06:04.919Z", "2025-04-16 02:06:03.919Z", "2025-04-16 02:06:02.919Z" ] ] } ] } \ No newline at end of file diff --git a/go/internal/feast/integration_tests/scylladb/scylladb_integration_test.go b/go/internal/feast/integration_tests/scylladb/scylladb_integration_test.go index 11ea2e5bd07..7d2d7170136 100644 --- a/go/internal/feast/integration_tests/scylladb/scylladb_integration_test.go +++ b/go/internal/feast/integration_tests/scylladb/scylladb_integration_test.go @@ -129,7 +129,7 @@ func TestGetOnlineFeaturesRange_withOnlyEqualsFilter(t *testing.T) { { SortKeyName: "event_timestamp", Query: &serving.SortKeyFilter_Equals{ - Equals: &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: 1744769171}}, + Equals: &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: 1744769171919}}, }, }, }, @@ -370,6 +370,181 @@ func TestGetOnlineFeaturesRange_withFeatureViewThrowsError(t *testing.T) { "Expected error message for non-existent sorted feature view") } +// TestGetOnlineFeaturesRange_SubSecondSortKeyValuesRemainDistinct is a +// regression test for the EAPC-22316 follow-up: InterfaceToProtoValue in +// go/types/typeconversion.go was truncating UNIX_TIMESTAMP sort key values +// read back from Cassandra to whole seconds via time.Time.Unix(), causing +// rows for the same entity that differ by less than a second to appear +// identical when the sort key is requested as a feature value. The fixture +// (sub_second_sort_key_view, seeded from sub_second_data.parquet) has three +// rows for one entity: two only 18ms apart within the same second, and one +// two seconds later - mirroring the real-world repro that caught this bug. +func TestGetOnlineFeaturesRange_SubSecondSortKeyValuesRemainDistinct(t *testing.T) { + entities := make(map[string]*types.RepeatedValue) + entities["sub_second_entity_id"] = &types.RepeatedValue{ + Val: []*types.Value{ + {Val: &types.Value_StringVal{StringVal: "entity-1"}}, + }, + } + + request := &serving.GetOnlineFeaturesRangeRequest{ + Kind: &serving.GetOnlineFeaturesRangeRequest_Features{ + Features: &serving.FeatureList{ + Val: []string{ + "sub_second_sort_key_view:event_timestamp", + "sub_second_sort_key_view:value", + }, + }, + }, + Entities: entities, + SortKeyFilters: []*serving.SortKeyFilter{ + { + SortKeyName: "event_timestamp", + Query: &serving.SortKeyFilter_Range{ + Range: &serving.SortKeyFilter_RangeQuery{ + RangeStart: &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: 0}}, + }, + }, + }, + }, + Limit: 10, + IncludeMetadata: true, + } + + response, err := client.GetOnlineFeaturesRange(ctx, request) + require.NoError(t, err) + require.NotNil(t, response) + require.Len(t, response.Results, 2, "expected event_timestamp and value results") + + eventTimestampIdx := -1 + for i, name := range response.Metadata.FeatureNames.Val { + if name == "event_timestamp" { + eventTimestampIdx = i + } + } + require.NotEqual(t, -1, eventTimestampIdx, "expected to find event_timestamp in the response") + + eventTimestampVector := response.Results[eventTimestampIdx] + require.Len(t, eventTimestampVector.Values, 1, "expected results for the single requested entity") + + timestamps := eventTimestampVector.Values[0].Val + require.Len(t, timestamps, 3, "expected all 3 sub-second-apart rows to be returned") + + seen := make(map[int64]bool) + for _, v := range timestamps { + ts := v.GetUnixTimestampVal() + assert.False(t, seen[ts], "sort key value %d was returned more than once; sub-second rows collapsed", ts) + seen[ts] = true + } + assert.Len(t, seen, 3, "expected 3 distinct millisecond-precision sort key values") + + expected := map[int64]bool{ + 1717244257886: true, // 2024-06-01 12:17:37.886 + 1717244257904: true, // 2024-06-01 12:17:37.904 + 1717244259035: true, // 2024-06-01 12:17:39.035 + } + for ts := range seen { + assert.True(t, expected[ts], "unexpected sort key value %d", ts) + } + + // EventTimestamps come from the _ts: hash field (a timestamppb.Timestamp). + // grpc_server calls GetTimestampSeconds, so values must be whole-second Unix timestamps. + require.Len(t, eventTimestampVector.EventTimestamps, 1, "expected EventTimestamps for 1 entity") + require.Len(t, eventTimestampVector.EventTimestamps[0].Val, 3, "expected EventTimestamp for each of the 3 rows") + for _, tsVal := range eventTimestampVector.EventTimestamps[0].Val { + secs := tsVal.GetUnixTimestampVal() + assert.True(t, secs > 1_000_000_000 && secs < 2_000_000_000, + "EventTimestamp must be seconds-precision (~2001–2033), got %d", secs) + } +} + +// TestGetOnlineFeaturesRange_CustomNamedSortKeyValuesRemainDistinct guards +// against the EAPC-22316 fix being tied to the "event_timestamp" column name +// rather than working for any UNIX_TIMESTAMP column declared as a sort key. +// The fixture (sub_second_custom_sortkey_view, seeded from +// sub_second_custom_sortkey_data.parquet) declares "viewed_at" - a name +// distinct from the source's own "event_timestamp" watermark field - as the +// sort key, mirroring the real customer schema. Same three-row shape: two +// rows 18ms apart within the same second, one two seconds later. +func TestGetOnlineFeaturesRange_CustomNamedSortKeyValuesRemainDistinct(t *testing.T) { + entities := make(map[string]*types.RepeatedValue) + entities["custom_sortkey_entity_id"] = &types.RepeatedValue{ + Val: []*types.Value{ + {Val: &types.Value_StringVal{StringVal: "entity-1"}}, + }, + } + + request := &serving.GetOnlineFeaturesRangeRequest{ + Kind: &serving.GetOnlineFeaturesRangeRequest_Features{ + Features: &serving.FeatureList{ + Val: []string{ + "sub_second_custom_sortkey_view:viewed_at", + "sub_second_custom_sortkey_view:value", + }, + }, + }, + Entities: entities, + SortKeyFilters: []*serving.SortKeyFilter{ + { + SortKeyName: "viewed_at", + Query: &serving.SortKeyFilter_Range{ + Range: &serving.SortKeyFilter_RangeQuery{ + RangeStart: &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: 0}}, + }, + }, + }, + }, + Limit: 10, + IncludeMetadata: true, + } + + response, err := client.GetOnlineFeaturesRange(ctx, request) + require.NoError(t, err) + require.NotNil(t, response) + require.Len(t, response.Results, 2, "expected viewed_at and value results") + + viewedAtIdx := -1 + for i, name := range response.Metadata.FeatureNames.Val { + if name == "viewed_at" { + viewedAtIdx = i + } + } + require.NotEqual(t, -1, viewedAtIdx, "expected to find viewed_at in the response") + + viewedAtVector := response.Results[viewedAtIdx] + require.Len(t, viewedAtVector.Values, 1, "expected results for the single requested entity") + + timestamps := viewedAtVector.Values[0].Val + require.Len(t, timestamps, 3, "expected all 3 sub-second-apart rows to be returned") + + seen := make(map[int64]bool) + for _, v := range timestamps { + ts := v.GetUnixTimestampVal() + assert.False(t, seen[ts], "sort key value %d was returned more than once; sub-second rows collapsed", ts) + seen[ts] = true + } + assert.Len(t, seen, 3, "expected 3 distinct millisecond-precision sort key values") + + expected := map[int64]bool{ + 1717244257886: true, // 2024-06-01 12:17:37.886 + 1717244257904: true, // 2024-06-01 12:17:37.904 + 1717244259035: true, // 2024-06-01 12:17:39.035 + } + for ts := range seen { + assert.True(t, expected[ts], "unexpected sort key value %d", ts) + } + + // EventTimestamps come from the _ts: hash field (a timestamppb.Timestamp). + // grpc_server calls GetTimestampSeconds, so values must be whole-second Unix timestamps. + require.Len(t, viewedAtVector.EventTimestamps, 1, "expected EventTimestamps for 1 entity") + require.Len(t, viewedAtVector.EventTimestamps[0].Val, 3, "expected EventTimestamp for each of the 3 rows") + for _, tsVal := range viewedAtVector.EventTimestamps[0].Val { + secs := tsVal.GetUnixTimestampVal() + assert.True(t, secs > 1_000_000_000 && secs < 2_000_000_000, + "EventTimestamp must be seconds-precision (~2001–2033), got %d", secs) + } +} + func assertResponseData(t *testing.T, response *serving.GetOnlineFeaturesRangeResponse, featureNames []string, entitiesRequested int, includeMetadata bool) { assert.NotNil(t, response) assert.Equal(t, 1, len(response.Entities), "Should have 1 list of entity") diff --git a/go/internal/feast/integration_tests/valkey/feature_repo/regression_repo.py b/go/internal/feast/integration_tests/valkey/feature_repo/regression_repo.py new file mode 100644 index 00000000000..c5849ed3adc --- /dev/null +++ b/go/internal/feast/integration_tests/valkey/feature_repo/regression_repo.py @@ -0,0 +1,47 @@ +# Regression fixture for EAPC-22316: UNIX_TIMESTAMP sort keys must retain +# millisecond precision for rows that differ by less than one second. + +from datetime import timedelta + +from feast import Entity, Field, FileSource, SortedFeatureView +from feast.protos.feast.core.SortedFeatureView_pb2 import SortOrder +from feast.sort_key import SortKey +from feast.types import String, UnixTimestamp +from feast.value_type import ValueType + +tags = {"team": "Feast"} +owner = "test@test.com" + +sub_second_entity: Entity = Entity( + name="sub_second_entity", + join_keys=["sub_second_entity_id"], + value_type=ValueType.STRING, + tags=tags, + owner=owner, +) + +sub_second_source: FileSource = FileSource( + path="sub_second_data.parquet", timestamp_field="event_timestamp" +) + +sub_second_sort_key_view: SortedFeatureView = SortedFeatureView( + name="sub_second_sort_key_view", + entities=[sub_second_entity], + ttl=timedelta(days=0), + source=sub_second_source, + tags=tags, + description="Regression fixture: sort key rows <1s apart for the same entity", + owner=owner, + online=True, + sort_keys=[ + SortKey( + name="event_timestamp", + value_type=ValueType.UNIX_TIMESTAMP, + default_sort_order=SortOrder.DESC, + ) + ], + schema=[ + Field(name="value", dtype=String), + Field(name="event_timestamp", dtype=UnixTimestamp), + ], +) diff --git a/go/internal/feast/integration_tests/valkey/feature_repo/sub_second_data.parquet b/go/internal/feast/integration_tests/valkey/feature_repo/sub_second_data.parquet new file mode 100644 index 00000000000..a10514da6c5 Binary files /dev/null and b/go/internal/feast/integration_tests/valkey/feature_repo/sub_second_data.parquet differ diff --git a/go/internal/feast/integration_tests/valkey/valkey_integration_test.go b/go/internal/feast/integration_tests/valkey/valkey_integration_test.go index fed9d16b8ed..572af80aa4b 100644 --- a/go/internal/feast/integration_tests/valkey/valkey_integration_test.go +++ b/go/internal/feast/integration_tests/valkey/valkey_integration_test.go @@ -153,7 +153,13 @@ func getValueType(value interface{}, featureName string) *types.Value { case int64: // Check if featureName contains "timestamp" if strings.Contains(featureName, "timestamp") { - return &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: value.(int64)}} + val := value.(int64) + if featureName != "event_timestamp" { + // Regular timestamp features are stored as seconds in Redis; + // ReadParquetDynamically returns ms, so divide by 1000. + val = val / 1000 + } + return &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: val}} } else { if value == nil { return &types.Value{} @@ -186,10 +192,13 @@ func getValueType(value interface{}, featureName string) *types.Value { if strings.Contains(featureName, "timestamp") { for _, v := range arrayInterface { - if v.(int64) == 0 { + elem := v.(int64) + if elem == 0 { arrayValue = append(arrayValue, -9223372036854775808) } else { - arrayValue = append(arrayValue, v.(int64)) + // Array timestamps stored as seconds in Redis; + // ReadParquetDynamically returns ms, divide by 1000. + arrayValue = append(arrayValue, elem/1000) } } return &types.Value{Val: &types.Value_UnixTimestampListVal{&types.Int64List{Val: arrayValue}}} @@ -323,7 +332,9 @@ func assertRangeFeatureMatchesParquet( expectedTS := expectedRows[i]["event_timestamp"].(int64) actualTS := result.EventTimestamps[eIdx].Val[i].GetUnixTimestampVal() - assert.Equal(t, expectedTS, actualTS) + // EventTimestamps comes from _ts: hash field stored in seconds; + // expectedTS is in ms (from ReadParquetDynamically), so divide by 1000. + assert.Equal(t, expectedTS/1000, actualTS) } } } @@ -487,7 +498,8 @@ func TestGetOnlineFeaturesRangeValkey_ReverseSortOrder(t *testing.T) { expectedTS := expectedRows[i]["event_timestamp"].(int64) actualTS := result.EventTimestamps[eIdx].Val[i].GetUnixTimestampVal() - assert.Equal(t, expectedTS, actualTS, + // EventTimestamps is seconds; expectedTS is ms from parquet. + assert.Equal(t, expectedTS/1000, actualTS, "timestamp mismatch feature=%s entity=%d idx=%d", feature, entityID, i) } @@ -569,7 +581,8 @@ func TestGetOnlineFeaturesRangeValkey_RangeStartInclusive(t *testing.T) { expTS := expected[i]["event_timestamp"].(int64) actTS := result.EventTimestamps[0].Val[i].GetUnixTimestampVal() - assert.Equal(t, expTS, actTS) + // EventTimestamps is seconds; expTS is ms from parquet. + assert.Equal(t, expTS/1000, actTS) } } } @@ -658,7 +671,8 @@ func TestGetOnlineFeaturesRangeValkey_LimitAppliedAfterHMGET(t *testing.T) { expTS := expectedRows[i]["event_timestamp"].(int64) actTS := actualTS[i].GetUnixTimestampVal() - assert.Equal(t, expTS, actTS, + // EventTimestamps is seconds; expTS is ms from parquet. + assert.Equal(t, expTS/1000, actTS, "timestamp mismatch feature=%s idx=%d", feature, i) if expectedRows[i][feature] == nil { @@ -669,3 +683,86 @@ func TestGetOnlineFeaturesRangeValkey_LimitAppliedAfterHMGET(t *testing.T) { } } } + +// TestGetOnlineFeaturesRangeValkey_SubSecondSortKey is a regression test for +// EAPC-22316: UNIX_TIMESTAMP sort key values must retain millisecond precision +// so that rows for the same entity that differ by less than one second remain +// distinct. The fixture has three rows: two 18ms apart within the same second +// and one two seconds later. +func TestGetOnlineFeaturesRangeValkey_SubSecondSortKey(t *testing.T) { + entities := map[string]*types.RepeatedValue{ + "sub_second_entity_id": {Val: []*types.Value{ + {Val: &types.Value_StringVal{StringVal: "entity-1"}}, + }}, + } + + req := &serving.GetOnlineFeaturesRangeRequest{ + Kind: &serving.GetOnlineFeaturesRangeRequest_Features{ + Features: &serving.FeatureList{ + Val: []string{ + "sub_second_sort_key_view:event_timestamp", + "sub_second_sort_key_view:value", + }, + }, + }, + Entities: entities, + SortKeyFilters: []*serving.SortKeyFilter{ + { + SortKeyName: "event_timestamp", + Query: &serving.SortKeyFilter_Range{ + Range: &serving.SortKeyFilter_RangeQuery{ + RangeStart: &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: 0}}, + }, + }, + }, + }, + Limit: 10, + IncludeMetadata: true, + } + + resp, err := client.GetOnlineFeaturesRange(ctx, req) + require.NoError(t, err) + require.NotNil(t, resp) + require.Len(t, resp.Results, 2, "expected event_timestamp and value results") + + eventTimestampIdx := -1 + for i, name := range resp.Metadata.FeatureNames.Val { + if name == "event_timestamp" { + eventTimestampIdx = i + } + } + require.NotEqual(t, -1, eventTimestampIdx, "expected to find event_timestamp in the response") + + eventTimestampVector := resp.Results[eventTimestampIdx] + require.Len(t, eventTimestampVector.Values, 1, "expected results for the single requested entity") + + timestamps := eventTimestampVector.Values[0].Val + require.Len(t, timestamps, 3, "expected all 3 sub-second-apart rows to be returned") + + seen := make(map[int64]bool) + for _, v := range timestamps { + ts := v.GetUnixTimestampVal() + assert.False(t, seen[ts], "sort key value %d was returned more than once; sub-second rows collapsed", ts) + seen[ts] = true + } + assert.Len(t, seen, 3, "expected 3 distinct millisecond-precision sort key values") + + expected := map[int64]bool{ + 1717244257886: true, // 2024-06-01 12:17:37.886 + 1717244257904: true, // 2024-06-01 12:17:37.904 + 1717244259035: true, // 2024-06-01 12:17:39.035 + } + for ts := range seen { + assert.True(t, expected[ts], "unexpected sort key value %d", ts) + } + + // EventTimestamps come from the _ts: hash field (a timestamppb.Timestamp). + // grpc_server calls GetTimestampSeconds, so values must be whole-second Unix timestamps. + require.Len(t, eventTimestampVector.EventTimestamps, 1, "expected EventTimestamps for 1 entity") + require.Len(t, eventTimestampVector.EventTimestamps[0].Val, 3, "expected EventTimestamp for each of the 3 rows") + for _, tsVal := range eventTimestampVector.EventTimestamps[0].Val { + secs := tsVal.GetUnixTimestampVal() + assert.True(t, secs > 1_000_000_000 && secs < 2_000_000_000, + "EventTimestamp must be seconds-precision (~2001–2033), got %d", secs) + } +} diff --git a/go/internal/feast/onlineserving/serving_test.go b/go/internal/feast/onlineserving/serving_test.go index 863f47829cc..1d0f5230029 100644 --- a/go/internal/feast/onlineserving/serving_test.go +++ b/go/internal/feast/onlineserving/serving_test.go @@ -1881,6 +1881,71 @@ func testTransposeRangeFeatureRowsIntoColumns(t *testing.T, useArrow bool) *Rang return vector } +// TestTransposeRangeFeatureRowsIntoColumns_SubSecondSortKeyValuesRemainDistinct +// is a regression test for the EAPC-22316 follow-up: when a UNIX_TIMESTAMP +// sort key column is requested as a feature value (as opposed to just being +// used for ordering), rows whose sort key differs by less than a second must +// still be returned as distinct millisecond-precision values, not collapsed +// to the same second. The raw driver value for a Cassandra/Scylla native +// `timestamp` column arrives here as a Go time.Time (see RangeFeatureData.Values). +func TestTransposeRangeFeatureRowsIntoColumns_SubSecondSortKeyValuesRemainDistinct(t *testing.T) { + arrowAllocator := memory.NewGoAllocator() + numRows := 1 + + sortKeyName := "viewed_at" + sortKey1 := test.CreateSortKeyProto(sortKeyName, core.SortOrder_DESC, types.ValueType_UNIX_TIMESTAMP) + entity1 := test.CreateEntityProto("device_user_agent_id", types.ValueType_STRING, "device_user_agent_id") + sfv := test.CreateSortedFeatureViewModel("personalization_pdp_features_repro", []*core.Entity{entity1}, []*core.SortKey{sortKey1}, + test.CreateFeature(sortKeyName, types.ValueType_UNIX_TIMESTAMP)) + + sortedViews := []*SortedFeatureViewAndRefs{ + {View: sfv, FeatureRefs: []string{sortKeyName}}, + } + + groupRef := &model.GroupedRangeFeatureRefs{ + FeatureNames: []string{sortKeyName}, + FeatureViewNames: []string{"personalization_pdp_features_repro"}, + AliasedFeatureNames: []string{"personalization_pdp_features_repro__" + sortKeyName}, + Indices: [][]int{{0}}, + } + + // Mirrors the real-world repro: two events for the same entity 18ms apart, + // within the same second. + t1 := time.Date(2026, 7, 9, 12, 17, 37, 886_000_000, time.UTC) + t2 := time.Date(2026, 7, 9, 12, 17, 37, 904_000_000, time.UTC) + + featureData := [][]onlinestore.RangeFeatureData{ + { + { + FeatureView: "personalization_pdp_features_repro", + FeatureName: sortKeyName, + Values: []interface{}{t2, t1}, // DESC order, most recent first + Statuses: []serving.FieldStatus{serving.FieldStatus_PRESENT, serving.FieldStatus_PRESENT}, + EventTimestamps: []timestamppb.Timestamp{ + {Seconds: t2.Unix(), Nanos: int32(t2.Nanosecond())}, + {Seconds: t1.Unix(), Nanos: int32(t1.Nanosecond())}, + }, + }, + }, + } + + vectors, err := TransposeRangeFeatureRowsIntoColumns(featureData, groupRef, sortedViews, arrowAllocator, numRows, false) + assert.NoError(t, err) + assert.Len(t, vectors, 1) + + rangeValues, err := vectors[0].GetProtoValues() + assert.NoError(t, err) + assert.Len(t, rangeValues, numRows) + assert.Len(t, rangeValues[0].Val, 2) + + got1 := rangeValues[0].Val[0].GetUnixTimestampVal() + got2 := rangeValues[0].Val[1].GetUnixTimestampVal() + + assert.NotEqual(t, got1, got2, "sub-second-apart sort key values must not collapse to the same value") + assert.Equal(t, t2.UnixMilli(), got1) + assert.Equal(t, t1.UnixMilli(), got2) +} + func TestValidateFeatureRefs(t *testing.T) { t.Run("NoCollisions", func(t *testing.T) { viewA := &model.FeatureView{ diff --git a/go/internal/feast/onlinestore/cassandraonlinestore_test.go b/go/internal/feast/onlinestore/cassandraonlinestore_test.go index 89c3722d6f1..c840d4b15a2 100644 --- a/go/internal/feast/onlinestore/cassandraonlinestore_test.go +++ b/go/internal/feast/onlinestore/cassandraonlinestore_test.go @@ -7,7 +7,9 @@ import ( "fmt" "reflect" "testing" + "time" + featuretypes "github.com/feast-dev/feast/go/types" "github.com/feast-dev/feast/go/internal/feast/model" "github.com/feast-dev/feast/go/internal/feast/registry" "github.com/feast-dev/feast/go/internal/feast/utils" @@ -530,6 +532,37 @@ func TestResolveFeatureValue_SortKeyMissingFromRow(t *testing.T) { assert.Nil(t, val) } +func TestResolveFeatureValue_SortKey_TimeTime(t *testing.T) { + // gocql surfaces Cassandra/Scylla native timestamp columns as time.Time. + // resolveFeatureValue returns the raw time.Time for sort keys; the serving + // layer then calls InterfaceToProtoValue which uses .UnixMilli() — the + // ms-precision encode path added in EAPC-22316. This test verifies both: + // (1) resolveFeatureValue preserves the time.Time without truncation, and + // (2) InterfaceToProtoValue encodes it as milliseconds, not whole seconds. + msVal := int64(1717244257886) // 2024-06-01 12:17:37.886Z + tVal := time.UnixMilli(msVal).UTC() + readValues := map[string]interface{}{ + "eventsortkey": tVal, + } + + val, status, err := resolveFeatureValue(readValues, canonicalColumnName("EventSortKey"), true) + assert.NoError(t, err) + assert.Equal(t, serving.FieldStatus_PRESENT, status) + require.NotNil(t, val) + + // resolveFeatureValue returns the raw interface{} for sort keys. + // Assert it's the exact time.Time we put in (no truncation). + gotTime, ok := val.(time.Time) + require.True(t, ok, "sort key value should be returned as time.Time") + assert.Equal(t, tVal, gotTime, "time.Time must be returned unmodified by resolveFeatureValue") + + // InterfaceToProtoValue (called by the serving layer) must encode it as ms. + protoVal, err := featuretypes.InterfaceToProtoValue(val) + require.NoError(t, err) + assert.Equal(t, msVal, protoVal.GetUnixTimestampVal(), + "time.Time sort key must be encoded as UnixMilli, not Unix seconds") +} + func TestResolveFeatureValue_AlreadyLowercaseFeature(t *testing.T) { testVal := &types.Value{Val: &types.Value_StringVal{StringVal: "hello"}} serialized := mustMarshalValueProto(t, testVal) diff --git a/go/internal/feast/onlinestore/redisonlinestore_test.go b/go/internal/feast/onlinestore/redisonlinestore_test.go index 7d27fa4d0b5..263deb2a9a9 100644 --- a/go/internal/feast/onlinestore/redisonlinestore_test.go +++ b/go/internal/feast/onlinestore/redisonlinestore_test.go @@ -423,7 +423,7 @@ func writeTestRecord( fieldHash, valB, ).Err()) require.NoError(t, store.client.ZAdd(ctx, zkey, redis.Z{ - Score: float64(ts.AsTime().Unix()), + Score: float64(ts.AsTime().UnixMilli()), Member: string(sortKeyBytes), }).Err()) } @@ -461,8 +461,8 @@ func TestOnlineReadRange(t *testing.T) { SortKeyFilters: []*model.SortKeyFilter{ { SortKeyName: "ts", - RangeStart: int64(1000), - RangeEnd: int64(2500), + RangeStart: int64(1000000), + RangeEnd: int64(2500000), StartInclusive: true, EndInclusive: true, }, @@ -479,6 +479,11 @@ func TestOnlineReadRange(t *testing.T) { require.Len(t, out.Values, 2) assert.Equal(t, "one", out.Values[0].(*types.Value).GetStringVal()) assert.Equal(t, "two", out.Values[1].(*types.Value).GetStringVal()) + + // EventTimestamps come from the _ts: hash field decoded as timestamppb.Timestamp. + require.Len(t, out.EventTimestamps, 2, "each returned row should have an event timestamp") + assert.Equal(t, int64(1000), out.EventTimestamps[0].Seconds, "row 0 event ts must match ts1") + assert.Equal(t, int64(2000), out.EventTimestamps[1].Seconds, "row 1 event ts must match ts2") } func TestOnlineReadRange_Reverse(t *testing.T) { @@ -510,7 +515,7 @@ func TestOnlineReadRange_Reverse(t *testing.T) { { SortKeyName: "ts", RangeStart: int64(0), - RangeEnd: int64(30)}, + RangeEnd: int64(30000)}, }, } @@ -539,7 +544,7 @@ func TestOnlineReadRange_EmptyResult(t *testing.T) { Limit: 10, IsReverseSortOrder: false, SortKeyFilters: []*model.SortKeyFilter{ - {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(10)}, + {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(10000)}, }, } @@ -577,7 +582,7 @@ func TestOnlineReadRange_WithLimit(t *testing.T) { Limit: 3, // Only get 3 IsReverseSortOrder: false, SortKeyFilters: []*model.SortKeyFilter{ - {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(10000)}, + {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(10000000)}, }, } @@ -623,7 +628,7 @@ func TestOnlineReadRange_MultipleEntityKeys(t *testing.T) { Limit: 10, IsReverseSortOrder: false, SortKeyFilters: []*model.SortKeyFilter{ - {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(5000)}, + {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(5000000)}, }, } @@ -669,7 +674,7 @@ func TestOnlineReadRange_MultipleFeatureViews(t *testing.T) { Limit: 10, IsReverseSortOrder: false, SortKeyFilters: []*model.SortKeyFilter{ - {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(5000)}, + {SortKeyName: "ts", RangeStart: int64(0), RangeEnd: int64(5000000)}, }, } @@ -767,10 +772,10 @@ func TestOnlineReadRange_ExclusiveRangeBounds(t *testing.T) { SortKeyFilters: []*model.SortKeyFilter{ { SortKeyName: "ts", - RangeStart: int64(1000), - RangeEnd: int64(3000), - StartInclusive: false, // Exclude 1000 - EndInclusive: false, // Exclude 3000 + RangeStart: int64(1000000), + RangeEnd: int64(3000000), + StartInclusive: false, // Exclude 1000s + EndInclusive: false, // Exclude 3000s }, }, } diff --git a/go/internal/feast/utils/redis_read_range_utils.go b/go/internal/feast/utils/redis_read_range_utils.go index e9af3700b47..9468b68f0a6 100644 --- a/go/internal/feast/utils/redis_read_range_utils.go +++ b/go/internal/feast/utils/redis_read_range_utils.go @@ -205,8 +205,8 @@ func fmtInterface(v interface{}) (string, error) { return fmt.Sprintf("%g", x), nil case time.Time: - // Feast timestamps become time.Time for sort keys - return fmt.Sprintf("%d", x.Unix()), nil + // Sort key scores are stored as milliseconds; use UnixMilli to match. + return fmt.Sprintf("%d", x.UnixMilli()), nil case string: return x, nil diff --git a/go/internal/feast/utils/redis_read_range_utils_test.go b/go/internal/feast/utils/redis_read_range_utils_test.go index 21e192a6c6f..71f531aa981 100644 --- a/go/internal/feast/utils/redis_read_range_utils_test.go +++ b/go/internal/feast/utils/redis_read_range_utils_test.go @@ -89,7 +89,7 @@ func TestGetScoreRange(t *testing.T) { StartInclusive: true, EndInclusive: true, }}, - "1000", "2000", + "1000000", "2000000", }, } diff --git a/go/internal/test/go_integration_test_utils.go b/go/internal/test/go_integration_test_utils.go index b23afdfec97..707621b6277 100644 --- a/go/internal/test/go_integration_test_utils.go +++ b/go/internal/test/go_integration_test_utils.go @@ -29,7 +29,12 @@ type Row struct { Created int64 } -var feastExec = "feast" +var feastExec = func() string { + if bin := os.Getenv("FEAST_BIN"); bin != "" { + return bin + } + return "feast" +}() func ReadParquet(filePath string) ([]*Row, error) { allocator := memory.NewGoAllocator() @@ -122,7 +127,7 @@ func ReadParquetDynamically(filePath string) ([]map[string]interface{}, error) { case *array.Timestamp: nanoseconds := int64(col.Value(rowIdx).ToTime(arrow.Second).Unix()) t := time.Unix(0, nanoseconds) - row[field.Name] = t.Unix() + row[field.Name] = t.UnixMilli() case *array.List: // Handle array (list) types listValues := []interface{}{} @@ -146,7 +151,7 @@ func ReadParquetDynamically(filePath string) ([]map[string]interface{}, error) { case *array.Timestamp: nanoseconds := int64(childCol.Value(int(i)).ToTime(arrow.Second).Unix()) t := time.Unix(0, nanoseconds) - listValues = append(listValues, t.Unix()) + listValues = append(listValues, t.UnixMilli()) default: listValues = append(listValues, nil) // Handle unsupported types } diff --git a/go/types/typeconversion.go b/go/types/typeconversion.go index 15cbda67640..61b9205aae3 100644 --- a/go/types/typeconversion.go +++ b/go/types/typeconversion.go @@ -563,7 +563,13 @@ func InterfaceToProtoValue(val interface{}) (*types.Value, error) { case bool: protoVal.Val = &types.Value_BoolVal{BoolVal: v} case time.Time: - protoVal.Val = &types.Value_UnixTimestampVal{UnixTimestampVal: v.Unix()} + // Only reached for UNIX_TIMESTAMP sort key columns read back from a native + // Cassandra/Scylla `timestamp` column (gocql surfaces those as time.Time); + // regular features are stored/read as opaque proto blobs, never time.Time. + // Sort keys are always written at millisecond precision, so UnixMilli() is + // the correct, lossless conversion here (see unixTsToTime for the matching + // decode-side threshold). + protoVal.Val = &types.Value_UnixTimestampVal{UnixTimestampVal: v.UnixMilli()} case *timestamppb.Timestamp: protoVal.Val = &types.Value_UnixTimestampVal{UnixTimestampVal: GetTimestampSeconds(v)} @@ -633,7 +639,7 @@ func InterfaceToProtoValue(val interface{}) (*types.Value, error) { case []time.Time: timestamps := make([]int64, len(v)) for j, t := range v { - timestamps[j] = t.Unix() + timestamps[j] = t.UnixMilli() } timestampList := &types.Int64List{Val: timestamps} protoVal.Val = &types.Value_UnixTimestampListVal{UnixTimestampListVal: timestampList} @@ -830,20 +836,20 @@ func valueTypeToGoTypeTimestampAsString(value *types.Value, timestampAsString bo return v case *types.Value_UnixTimestampVal: if timestampAsString { - return time.Unix(x.UnixTimestampVal, 0).UTC().Format(TimestampFormat) + return unixTsToTime(x.UnixTimestampVal).Format(TimestampFormat) } - return time.Unix(x.UnixTimestampVal, 0).UTC() + return unixTsToTime(x.UnixTimestampVal) case *types.Value_UnixTimestampListVal: if timestampAsString { timestamps := make([]string, len(x.UnixTimestampListVal.Val)) for i, ts := range x.UnixTimestampListVal.Val { - timestamps[i] = time.Unix(ts, 0).UTC().Format(TimestampFormat) + timestamps[i] = unixTsToTime(ts).Format(TimestampFormat) } return timestamps } timestamps := make([]time.Time, len(x.UnixTimestampListVal.Val)) for i, ts := range x.UnixTimestampListVal.Val { - timestamps[i] = time.Unix(ts, 0).UTC() + timestamps[i] = unixTsToTime(ts) } return timestamps default: @@ -851,6 +857,16 @@ func valueTypeToGoTypeTimestampAsString(value *types.Value, timestampAsString bo } } +// unixTsToTime converts a unix_timestamp_val int64 to time.Time using a threshold to +// distinguish seconds (regular features) from milliseconds (sort key columns). +// Values > 1e11 are unambiguously milliseconds; current-era seconds ~1.7e9 < 1e11. +func unixTsToTime(val int64) time.Time { + if val > 1e11 { + return time.UnixMilli(val).UTC() + } + return time.Unix(val, 0).UTC() +} + func transformStringToBytes(str string) []byte { bytes, decodeErr := base64.StdEncoding.DecodeString(str) if decodeErr != nil { @@ -1045,3 +1061,12 @@ func GetTimestampSeconds(ts *timestamppb.Timestamp) int64 { } return ts.GetSeconds() } + +// GetTimestampMillis flattens a proto Timestamp to milliseconds since epoch, +// preserving sub-second precision that GetTimestampSeconds discards. +func GetTimestampMillis(ts *timestamppb.Timestamp) int64 { + if ts == nil { + return 0 + } + return ts.AsTime().UnixMilli() +} diff --git a/go/types/typeconversion_test.go b/go/types/typeconversion_test.go index f2479d116e4..264a336f674 100644 --- a/go/types/typeconversion_test.go +++ b/go/types/typeconversion_test.go @@ -11,6 +11,7 @@ import ( "github.com/apache/arrow/go/v17/arrow/memory" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "google.golang.org/protobuf/proto" "github.com/feast-dev/feast/go/protos/feast/types" @@ -366,7 +367,7 @@ func TestInterfaceToProtoValue(t *testing.T) { {float32(30.5), &types.Value{Val: &types.Value_FloatVal{FloatVal: 30.5}}}, {float64(40.5), &types.Value{Val: &types.Value_DoubleVal{DoubleVal: 40.5}}}, {true, &types.Value{Val: &types.Value_BoolVal{BoolVal: true}}}, - {testTime, &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: testTime.Unix()}}}, + {testTime, &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: testTime.UnixMilli()}}}, {×tamppb.Timestamp{Seconds: testTime.Unix(), Nanos: int32(testTime.Nanosecond())}, &types.Value{Val: &types.Value_UnixTimestampVal{UnixTimestampVal: testTime.Unix()}}}, {[][]byte{{1, 2}, {3, 4}}, &types.Value{Val: &types.Value_BytesListVal{BytesListVal: &types.BytesList{Val: [][]byte{{1, 2}, {3, 4}}}}}}, {[]string{"a", "b"}, &types.Value{Val: &types.Value_StringListVal{StringListVal: &types.StringList{Val: []string{"a", "b"}}}}}, @@ -376,7 +377,7 @@ func TestInterfaceToProtoValue(t *testing.T) { {[]float32{5.5, 6.6}, &types.Value{Val: &types.Value_FloatListVal{FloatListVal: &types.FloatList{Val: []float32{5.5, 6.6}}}}}, {[]float64{7.7, 8.8}, &types.Value{Val: &types.Value_DoubleListVal{DoubleListVal: &types.DoubleList{Val: []float64{7.7, 8.8}}}}}, {[]bool{true, false}, &types.Value{Val: &types.Value_BoolListVal{BoolListVal: &types.BoolList{Val: []bool{true, false}}}}}, - {[]time.Time{testTime, testTime.Add(time.Hour)}, &types.Value{Val: &types.Value_UnixTimestampListVal{UnixTimestampListVal: &types.Int64List{Val: []int64{testTime.Unix(), testTime.Add(time.Hour).Unix()}}}}}, + {[]time.Time{testTime, testTime.Add(time.Hour)}, &types.Value{Val: &types.Value_UnixTimestampListVal{UnixTimestampListVal: &types.Int64List{Val: []int64{testTime.UnixMilli(), testTime.Add(time.Hour).UnixMilli()}}}}}, {[]*timestamppb.Timestamp{{Seconds: testTime.Unix(), Nanos: int32(testTime.Nanosecond())}, {Seconds: testTime.Add(time.Hour).Unix(), Nanos: int32(testTime.Add(time.Hour).Nanosecond())}}, &types.Value{Val: &types.Value_UnixTimestampListVal{UnixTimestampListVal: &types.Int64List{Val: []int64{testTime.Unix(), testTime.Add(time.Hour).Unix()}}}}}, {&types.Value{Val: &types.Value_NullVal{NullVal: types.Null_NULL}}, &types.Value{Val: &types.Value_NullVal{NullVal: types.Null_NULL}}}, {&types.Value{Val: &types.Value_StringVal{StringVal: "test"}}, &types.Value{Val: &types.Value_StringVal{StringVal: "test"}}}, @@ -390,6 +391,92 @@ func TestInterfaceToProtoValue(t *testing.T) { } } +// TestInterfaceToProtoValue_TimeSubSecondPrecision guards against EAPC-22316 +// regressing: a time.Time with a non-zero sub-second component (as read back +// from a Cassandra/Scylla native `timestamp` sort key column) must retain +// millisecond precision, not collapse to whole seconds. +func TestInterfaceToProtoValue_TimeSubSecondPrecision(t *testing.T) { + sortKeyTime := time.Date(2026, 7, 9, 12, 17, 37, 886_000_000, time.UTC) + + result, err := InterfaceToProtoValue(sortKeyTime) + assert.NoError(t, err) + + gotMillis := result.GetUnixTimestampVal() + assert.Equal(t, sortKeyTime.UnixMilli(), gotMillis, + "expected millisecond-precision conversion") + assert.NotEqual(t, sortKeyTime.Unix()*1000, gotMillis, + "conversion must not be second-truncated then rescaled") +} + +// TestInterfaceToProtoValue_SubSecondValuesRemainDistinct is the crux +// regression test for the EAPC-22316 follow-up: two sort key timestamps that +// differ by less than a second must decode back to distinct, correct values +// via unixTsToTime, not collapse to the same second. +func TestInterfaceToProtoValue_SubSecondValuesRemainDistinct(t *testing.T) { + t1 := time.Date(2026, 7, 9, 12, 17, 37, 886_000_000, time.UTC) + t2 := time.Date(2026, 7, 9, 12, 17, 37, 904_000_000, time.UTC) + + v1, err := InterfaceToProtoValue(t1) + assert.NoError(t, err) + v2, err := InterfaceToProtoValue(t2) + assert.NoError(t, err) + + assert.NotEqual(t, v1.GetUnixTimestampVal(), v2.GetUnixTimestampVal(), + "sub-second-apart sort key values must not collapse to the same encoded value") + + decoded1 := unixTsToTime(v1.GetUnixTimestampVal()) + decoded2 := unixTsToTime(v2.GetUnixTimestampVal()) + + assert.True(t, decoded1.Equal(t1), "expected round-trip to preserve t1 exactly, got %v", decoded1) + assert.True(t, decoded2.Equal(t2), "expected round-trip to preserve t2 exactly, got %v", decoded2) + assert.False(t, decoded1.Equal(decoded2), "decoded values must remain distinct") +} + +func TestGetTimestampMillis(t *testing.T) { + assert.Equal(t, int64(0), GetTimestampMillis(nil), "nil timestamp should return 0") + + ts := ×tamppb.Timestamp{Seconds: 1783678657, Nanos: 886_000_000} + assert.Equal(t, int64(1783678657886), GetTimestampMillis(ts), + "expected millisecond precision to be retained") +} + +func TestGetTimestampSeconds_Nil(t *testing.T) { + assert.Equal(t, int64(0), GetTimestampSeconds(nil), "nil timestamp should return 0") + + ts := ×tamppb.Timestamp{Seconds: 1783678657, Nanos: 886_000_000} + assert.Equal(t, int64(1783678657), GetTimestampSeconds(ts), + "GetTimestampSeconds must discard sub-second nanos and return only whole seconds") +} + +func TestUnixTsToTime_Boundary(t *testing.T) { + // The threshold is val > 1e11 (100_000_000_000). + // Values at or below the threshold are seconds; values above are milliseconds. + // This test pins the exact boundary so a change from > to >= (or a different + // constant) is immediately caught. + const threshold = int64(100_000_000_000) // 1e11 + + below := unixTsToTime(threshold - 1) + assert.Equal(t, time.Unix(threshold-1, 0).UTC(), below, + "val == 1e11-1 must be treated as seconds") + + exact := unixTsToTime(threshold) + assert.Equal(t, time.Unix(threshold, 0).UTC(), exact, + "val == 1e11 must be treated as seconds (condition is >, not >=)") + + above := unixTsToTime(threshold + 1) + assert.Equal(t, time.UnixMilli(threshold+1).UTC(), above, + "val == 1e11+1 must be treated as milliseconds") +} + +func TestUnixTsToTime_ZeroAndNegative(t *testing.T) { + assert.Equal(t, time.Unix(0, 0).UTC(), unixTsToTime(0), + "zero must map to Unix epoch via seconds branch") + assert.Equal(t, time.Unix(-1, 0).UTC(), unixTsToTime(-1), + "negative values must take seconds branch (all negatives < 1e11)") + assert.Equal(t, time.Unix(math.MinInt64, 0).UTC(), unixTsToTime(math.MinInt64), + "MinInt64 must take seconds branch without panic") +} + func TestValueTypeToGoType(t *testing.T) { timestamp := time.Unix(1744769099, 0).UTC() testCases := []*types.Value{ @@ -473,6 +560,31 @@ func TestValueTypeToGoTypeTimestampAsString(t *testing.T) { } } +func TestValueTypeToGoTypeTimestampAsString_MsRangeList(t *testing.T) { + // These are millisecond-precision sort key values (> 1e11); the list branch of + // unixTsToTime must take the ms path, not the seconds path, for each element. + ms1 := int64(1717244257886) // 2024-06-01 12:17:37.886Z + ms2 := int64(1717244257904) // 2024-06-01 12:17:37.904Z + + val := &types.Value{Val: &types.Value_UnixTimestampListVal{ + UnixTimestampListVal: &types.Int64List{Val: []int64{ms1, ms2}}, + }} + + result := ValueTypeToGoTypeTimestampAsString(val) + strs, ok := result.([]string) + require.True(t, ok, "expected []string from UnixTimestampListVal") + require.Len(t, strs, 2) + + assert.Equal(t, time.UnixMilli(ms1).UTC().Format(TimestampFormat), strs[0], + "ms-range list element 0 must use UnixMilli, not Unix") + assert.Equal(t, time.UnixMilli(ms2).UTC().Format(TimestampFormat), strs[1], + "ms-range list element 1 must use UnixMilli, not Unix") + + // Confirm both strings carry sub-second precision (the ".886" / ".904" suffix). + assert.Contains(t, strs[0], ".886", "expected sub-second suffix .886 in formatted timestamp") + assert.Contains(t, strs[1], ".904", "expected sub-second suffix .904 in formatted timestamp") +} + func TestConvertToValueType_String(t *testing.T) { testCases := []struct { input *types.Value diff --git a/sdk/python/feast/feature_store.py b/sdk/python/feast/feature_store.py index b1b5660f35e..10bf7cc7673 100644 --- a/sdk/python/feast/feature_store.py +++ b/sdk/python/feast/feature_store.py @@ -98,6 +98,7 @@ from feast.transformation.pandas_transformation import PandasTransformation from feast.transformation.python_transformation import PythonTransformation from feast.utils import _get_feature_view_vector_field_metadata, _utc_now +from feast.version import get_installed_version, get_version warnings.simplefilter("once", DeprecationWarning) @@ -3244,6 +3245,11 @@ async def close(self) -> None: def _print_materialization_log( start_date, end_date, num_feature_views: int, online_store: str ): + logger.info( + "Materialization starting: feast=%s feature-store-materialization=%s", + get_version(), + get_installed_version("feature-store-materialization"), + ) if start_date: print( f"Materializing {Style.BRIGHT + Fore.GREEN}{num_feature_views}{Style.RESET_ALL} feature views" diff --git a/sdk/python/feast/infra/contrib/spark_kafka_processor.py b/sdk/python/feast/infra/contrib/spark_kafka_processor.py index 36bc3a6242d..1a4c7bec234 100644 --- a/sdk/python/feast/infra/contrib/spark_kafka_processor.py +++ b/sdk/python/feast/infra/contrib/spark_kafka_processor.py @@ -25,6 +25,7 @@ from feast.infra.provider import get_provider from feast.sorted_feature_view import SortedFeatureView from feast.stream_feature_view import StreamFeatureView +from feast.version import get_installed_version, get_version logger = logging.getLogger(__name__) @@ -216,6 +217,11 @@ def _create_infra_if_necessary(self): def ingest_stream_feature_view( self, to: PushMode = PushMode.ONLINE ) -> StreamingQuery: + logger.info( + "Streaming materialization starting: feast=%s feature-store-materialization=%s", + get_version(), + get_installed_version("feature-store-materialization"), + ) self._create_infra_if_necessary() ingested_stream_df = self._ingest_stream_data() transformed_df = self._construct_transformation_plan(ingested_stream_df) diff --git a/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py b/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py index 06b0f3ac4d7..631adc11abc 100644 --- a/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py +++ b/sdk/python/feast/infra/online_stores/cassandra_online_store/cassandra_online_store.py @@ -539,17 +539,15 @@ def on_failure(exc, concurrent_queue): feast_value_type = valProto.WhichOneof("val") if feast_value_type == "unix_timestamp_val": feature_value = ( - valProto.unix_timestamp_val * 1000 - ) # Convert to milliseconds + valProto.unix_timestamp_val + ) # already milliseconds elif feast_value_type is None: feature_value = None elif feast_value_type in feast_array_types: if feast_value_type == "unix_timestamp_list_val": - # Convert list of timestamps to milliseconds - feature_value = [ - ts * 1000 - for ts in valProto.unix_timestamp_list_val.val # type:ignore - ] + feature_value = list( + valProto.unix_timestamp_list_val.val # type:ignore + ) # already milliseconds else: feature_value = getattr( valProto, str(feast_value_type) diff --git a/sdk/python/feast/infra/online_stores/eg_valkey.py b/sdk/python/feast/infra/online_stores/eg_valkey.py index 93f147f51b8..01f5bdf51f4 100644 --- a/sdk/python/feast/infra/online_stores/eg_valkey.py +++ b/sdk/python/feast/infra/online_stores/eg_valkey.py @@ -774,9 +774,7 @@ def zset_score(sort_key_value: ValueProto): """ feast_value_type = sort_key_value.WhichOneof("val") if feast_value_type == "unix_timestamp_val": - feature_value = ( - sort_key_value.unix_timestamp_val * 1000 - ) # Convert to milliseconds + feature_value = sort_key_value.unix_timestamp_val # already milliseconds else: feature_value = getattr(sort_key_value, str(feast_value_type)) return feature_value diff --git a/sdk/python/feast/infra/online_stores/redis.py b/sdk/python/feast/infra/online_stores/redis.py index 8d3d3185ca5..06e90c25586 100644 --- a/sdk/python/feast/infra/online_stores/redis.py +++ b/sdk/python/feast/infra/online_stores/redis.py @@ -500,9 +500,7 @@ def zset_score(sort_key_value: ValueProto): """ feast_value_type = sort_key_value.WhichOneof("val") if feast_value_type == "unix_timestamp_val": - feature_value = ( - sort_key_value.unix_timestamp_val * 1000 - ) # Convert to milliseconds + feature_value = sort_key_value.unix_timestamp_val # already milliseconds else: feature_value = getattr(sort_key_value, str(feast_value_type)) return feature_value diff --git a/sdk/python/feast/type_map.py b/sdk/python/feast/type_map.py index ebf6f0eae19..df08e860208 100644 --- a/sdk/python/feast/type_map.py +++ b/sdk/python/feast/type_map.py @@ -55,6 +55,11 @@ # null timestamps get converted to -9223372036854775808 NULL_TIMESTAMP_INT_VALUE: int = np.datetime64("NaT").astype(int) +# Threshold for disambiguating a raw UNIX_TIMESTAMP integer as milliseconds vs. +# seconds: current-era seconds are ~1.7e9, milliseconds ~1.7e12, so anything +# above this is unambiguously milliseconds (safe until year ~5138 in seconds). +MS_TIMESTAMP_THRESHOLD = 1e11 + logger = logging.getLogger(__name__) @@ -78,11 +83,16 @@ def feast_value_type_to_python_type(field_value_proto: ProtoValue) -> Any: if hasattr(val, "val"): val = list(val.val) - # Convert UNIX_TIMESTAMP values to `datetime` + # Convert UNIX_TIMESTAMP values to `datetime`. + # unix_timestamp_val stores seconds for regular features and milliseconds for sort key + # columns; see MS_TIMESTAMP_THRESHOLD for the disambiguation rule. if val_attr == "unix_timestamp_list_val": val = [ ( - datetime.fromtimestamp(v, tz=timezone.utc) + datetime.fromtimestamp( + v / 1000.0 if v > MS_TIMESTAMP_THRESHOLD else float(v), + tz=timezone.utc, + ) if v != NULL_TIMESTAMP_INT_VALUE else None ) @@ -90,7 +100,10 @@ def feast_value_type_to_python_type(field_value_proto: ProtoValue) -> Any: ] elif val_attr == "unix_timestamp_val": val = ( - datetime.fromtimestamp(val, tz=timezone.utc) + datetime.fromtimestamp( + val / 1000.0 if val > MS_TIMESTAMP_THRESHOLD else float(val), + tz=timezone.utc, + ) if val != NULL_TIMESTAMP_INT_VALUE else None ) @@ -358,6 +371,53 @@ def _python_datetime_to_int_timestamp( return int_timestamps +def _python_datetime_to_int_ms_timestamp( + values: Sequence[Any], +) -> Sequence[Union[int, np.int_]]: + """Convert datetime values to milliseconds since epoch (used for sort key columns).""" + # Fast path for Numpy array. + if isinstance(values, np.ndarray) and isinstance(values.dtype, np.datetime64): + if values.ndim != 1: + raise ValueError("Only 1 dimensional arrays are supported.") + return cast(Sequence[np.int_], values.astype("datetime64[ms]").astype(np.int_)) + + int_timestamps = [] + for value in values: + if isinstance(value, datetime): + int_timestamps.append(int(round(value.timestamp() * 1000))) + elif isinstance(value, Timestamp): + int_timestamps.append(int(value.ToMilliseconds())) + elif isinstance(value, np.datetime64): + int_timestamps.append(value.astype("datetime64[ms]").astype(np.int_)) # type: ignore[attr-defined] + elif isinstance(value, type(np.nan)): + int_timestamps.append(NULL_TIMESTAMP_INT_VALUE) + else: + # A raw (non-datetime) value here means the sort key column + # arrived as a plain integer rather than a typed timestamp - + # e.g. a Spark LongType column instead of TimestampType, + # which can happen if the source Avro schema's field lacks a + # timestamp-millis/timestamp-micros logicalType. There is no + # reliable way to know whether that raw integer is already + # milliseconds or is seconds, so apply the same magnitude + # threshold used everywhere else in this codebase to decide + # (unixTsToTime in Go, feast_value_type_to_python_type here) + # rather than silently assuming it's already milliseconds. + raw = int(value) + if raw > int(MS_TIMESTAMP_THRESHOLD): + int_timestamps.append(raw) + else: + logger.warning( + "Sort key column received a raw integer timestamp (%d) instead of " + "a typed datetime value; assuming seconds and converting to " + "milliseconds. This usually means the upstream source (e.g. an " + "Avro schema without a timestamp-millis/timestamp-micros " + "logicalType) is not producing typed timestamps.", + raw, + ) + int_timestamps.append(raw * 1000) + return int_timestamps + + def _python_value_to_proto_value( feast_value_type: ValueType, values: List[Any] ) -> List[ProtoValue]: diff --git a/sdk/python/feast/utils.py b/sdk/python/feast/utils.py index af55266c682..48ff94a1e16 100644 --- a/sdk/python/feast/utils.py +++ b/sdk/python/feast/utils.py @@ -1,5 +1,6 @@ import copy import itertools +import logging import os import typing import warnings @@ -42,7 +43,10 @@ from feast.protos.feast.types.Value_pb2 import FloatList as FloatListProto from feast.protos.feast.types.Value_pb2 import RepeatedValue as RepeatedValueProto from feast.protos.feast.types.Value_pb2 import Value as ValueProto -from feast.type_map import python_values_to_proto_values +from feast.type_map import ( + _python_datetime_to_int_ms_timestamp, + python_values_to_proto_values, +) from feast.types import ComplexFeastType, PrimitiveFeastType, from_feast_to_pyarrow_type from feast.value_type import ValueType from feast.version import get_version @@ -54,6 +58,8 @@ from feast.infra.registry.base_registry import BaseRegistry from feast.on_demand_feature_view import OnDemandFeatureView +logger = logging.getLogger(__name__) + APPLICATION_NAME = "feast-dev/feast" USER_AGENT = "{}/{}".format(APPLICATION_NAME, get_version()) @@ -262,9 +268,17 @@ def _convert_arrow_to_proto( getattr(feature_view, "source_request_sources", None) is not None or getattr(feature_view, "source_feature_view_projections", None) is not None ): - return _convert_arrow_odfv_to_proto(table, feature_view, join_keys) # type: ignore[arg-type] + result = _convert_arrow_odfv_to_proto(table, feature_view, join_keys) # type: ignore[arg-type] else: - return _convert_arrow_fv_to_proto(table, feature_view, join_keys) # type: ignore[arg-type] + result = _convert_arrow_fv_to_proto(table, feature_view, join_keys) # type: ignore[arg-type] + + logger.info( + "[materialize] feature_view=%s rows=%d", + feature_view.name, + len(result), + ) + + return result def _convert_arrow_fv_to_proto( @@ -281,13 +295,27 @@ def _convert_arrow_fv_to_proto( (field.name, field.dtype.to_value_type()) for field in feature_view.features ] + list(join_keys.items()) - proto_values_by_column = { - column: python_values_to_proto_values( - table.column(column).to_numpy(zero_copy_only=False), value_type - ) - for column, value_type in columns + # Sort key UNIX_TIMESTAMP columns use ms precision so that timestamps differing + # by < 1 second produce distinct sort key scores in the online store. + sort_key_ts_names = { + sk.name + for sk in getattr(feature_view, "sort_keys", []) + if sk.value_type == ValueType.UNIX_TIMESTAMP } + proto_values_by_column = {} + for column, value_type in columns: + raw = table.column(column).to_numpy(zero_copy_only=False) + if column in sort_key_ts_names: + ms_vals = _python_datetime_to_int_ms_timestamp(raw) + proto_values_by_column[column] = [ + ValueProto(unix_timestamp_val=int(ts)) for ts in ms_vals + ] + else: + proto_values_by_column[column] = python_values_to_proto_values( + raw, value_type + ) + entity_keys = [ EntityKeyProto( join_keys=join_keys, diff --git a/sdk/python/feast/version.py b/sdk/python/feast/version.py index 85d8476a66d..444efbd8503 100644 --- a/sdk/python/feast/version.py +++ b/sdk/python/feast/version.py @@ -1,10 +1,27 @@ from importlib.metadata import PackageNotFoundError, version +# This repo publishes its own SDK distribution as "eg-feast" (see setup.py), not +# "feast", so the installed package name differs from upstream OSS Feast. Check +# both, upstream name first for compatibility with non-fork installs. +_SDK_DISTRIBUTION_NAMES = ("feast", "eg-feast") + def get_version(): """Returns version information of the Feast Python Package.""" + for candidate in _SDK_DISTRIBUTION_NAMES: + try: + return version(candidate) + except PackageNotFoundError: + continue + return "unknown" + + +def get_installed_version(package_name: str) -> str: + """Returns the installed version of an arbitrary package, or "not installed" + if it isn't present. Useful for opportunistically reporting the version of a + wrapper/consumer package (e.g. a materialization job image) without requiring + it to be installed for Feast to function.""" try: - sdk_version = version("feast") + return version(package_name) except PackageNotFoundError: - sdk_version = "unknown" - return sdk_version + return "not installed" diff --git a/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py b/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py index 20e23c35e03..3e68cc5d6e8 100644 --- a/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py +++ b/sdk/python/tests/unit/infra/compute_engines/local/test_nodes.py @@ -37,9 +37,10 @@ def create_context(node_outputs): # Setup execution context + repo_config = MagicMock() return ExecutionContext( project="test_proj", - repo_config=MagicMock(), + repo_config=repo_config, offline_store=MagicMock(), online_store=MagicMock(), entity_defs=MagicMock(), diff --git a/sdk/python/tests/unit/infra/online_store/test_redis.py b/sdk/python/tests/unit/infra/online_store/test_redis.py index e2aad12bd7e..372cf97923c 100644 --- a/sdk/python/tests/unit/infra/online_store/test_redis.py +++ b/sdk/python/tests/unit/infra/online_store/test_redis.py @@ -508,7 +508,9 @@ def _make_rows(n=10): { "trip_id": ValueProto(int32_val=i), "rating": ValueProto(float_val=i + 0.5), - "event_timestamp": ValueProto(unix_timestamp_val=base_ts + i * 60), + "event_timestamp": ValueProto( + unix_timestamp_val=(base_ts + i * 60) * 1000 + ), }, datetime.fromtimestamp(base_ts + i * 60, tz=timezone.utc), None, diff --git a/sdk/python/tests/unit/infra/online_store/test_valkey.py b/sdk/python/tests/unit/infra/online_store/test_valkey.py index ed911bcb92a..2af1e8060a1 100644 --- a/sdk/python/tests/unit/infra/online_store/test_valkey.py +++ b/sdk/python/tests/unit/infra/online_store/test_valkey.py @@ -402,7 +402,9 @@ def _make_rows(n=10): { "trip_id": ValueProto(int32_val=i), "rating": ValueProto(float_val=i + 0.5), - "event_timestamp": ValueProto(unix_timestamp_val=base_ts + i * 60), + "event_timestamp": ValueProto( + unix_timestamp_val=(base_ts + i * 60) * 1000 + ), }, datetime.fromtimestamp(base_ts + i * 60, tz=timezone.utc), None, diff --git a/sdk/python/tests/unit/test_type_map.py b/sdk/python/tests/unit/test_type_map.py index be8a25c1639..55406cdafe3 100644 --- a/sdk/python/tests/unit/test_type_map.py +++ b/sdk/python/tests/unit/test_type_map.py @@ -3,6 +3,8 @@ import pytest from feast.type_map import ( + MS_TIMESTAMP_THRESHOLD, + _python_datetime_to_int_ms_timestamp, feast_value_type_to_python_type, python_values_to_proto_values, ) @@ -30,6 +32,32 @@ def test_null_unix_timestamp_list(): assert converted[0] is None +def test_python_datetime_to_int_ms_timestamp_raw_int_below_threshold_is_treated_as_seconds(): + """A raw (non-datetime) integer below MS_TIMESTAMP_THRESHOLD is assumed to be + seconds and converted to milliseconds - e.g. a sort key column that arrived + as a Spark LongType rather than TimestampType (source Avro schema missing a + timestamp-millis/timestamp-micros logicalType).""" + + seconds_value = 1717244257 # 2024-06-01 12:17:37 UTC, well below the threshold + assert seconds_value < MS_TIMESTAMP_THRESHOLD + + result = _python_datetime_to_int_ms_timestamp([seconds_value]) + + assert result == [seconds_value * 1000] + + +def test_python_datetime_to_int_ms_timestamp_raw_int_above_threshold_passes_through(): + """A raw integer already above MS_TIMESTAMP_THRESHOLD is assumed to already be + milliseconds and is passed through unchanged.""" + + ms_value = 1717244257886 # 2024-06-01 12:17:37.886 UTC + assert ms_value > MS_TIMESTAMP_THRESHOLD + + result = _python_datetime_to_int_ms_timestamp([ms_value]) + + assert result == [ms_value] + + @pytest.mark.parametrize( "values", ( diff --git a/sdk/python/tests/unit/test_version.py b/sdk/python/tests/unit/test_version.py new file mode 100644 index 00000000000..ac2428c5dde --- /dev/null +++ b/sdk/python/tests/unit/test_version.py @@ -0,0 +1,80 @@ +from importlib.metadata import PackageNotFoundError + +import pytest + +from feast.feature_store import _print_materialization_log +from feast.utils import _utc_now +from feast.version import get_installed_version, get_version + + +def test_get_version_returns_real_version(): + """get_version() should find whichever of the "feast" / "eg-feast" + distributions is actually installed, not silently fall back to "unknown".""" + assert get_version() != "unknown" + + +def test_get_version_falls_back_from_feast_to_eg_feast(monkeypatch): + """Regression test: this repo publishes its SDK distribution as "eg-feast" + (see setup.py), not "feast". get_version() must fall back to "eg-feast" when + "feast" isn't installed, instead of only checking "feast" and returning + "unknown".""" + + def fake_version(name): + if name == "feast": + raise PackageNotFoundError() + if name == "eg-feast": + return "1.2.3" + raise PackageNotFoundError() + + monkeypatch.setattr("feast.version.version", fake_version) + + assert get_version() == "1.2.3" + + +def test_get_version_falls_back_from_eg_feast_to_feast(monkeypatch): + """The reverse case: upstream OSS installs where the distribution is named + "feast" rather than "eg-feast" must still resolve correctly.""" + + def fake_version(name): + if name == "feast": + return "0.9.0" + raise PackageNotFoundError() + + monkeypatch.setattr("feast.version.version", fake_version) + + assert get_version() == "0.9.0" + + +def test_get_version_returns_unknown_when_neither_installed(monkeypatch): + def fake_version(name): + raise PackageNotFoundError() + + monkeypatch.setattr("feast.version.version", fake_version) + + assert get_version() == "unknown" + + +def test_get_installed_version_for_missing_package(): + assert get_installed_version("definitely-not-a-real-package-xyz") == "not installed" + + +def test_get_installed_version_for_installed_package(): + # "feast" or "eg-feast" (whichever is actually installed) should resolve. + result = get_installed_version("pytest") + assert result != "not installed" + assert pytest.__version__ == result + + +def test_materialization_log_always_includes_version_banner(caplog): + """The version banner is always-on - it must appear regardless of any + opt-in config, unlike the row-content logging.""" + now = _utc_now() + + with caplog.at_level("INFO"): + _print_materialization_log( + start_date=now, end_date=now, num_feature_views=1, online_store="sqlite" + ) + + assert "Materialization starting" in caplog.text + assert "feast=" in caplog.text + assert "feature-store-materialization=" in caplog.text