Skip to content

Commit 2c1e439

Browse files
abergmeierabergmeier
committed
Add support for Valkey as Online Store.
1 parent 728aa2e commit 2c1e439

7 files changed

Lines changed: 452 additions & 14 deletions

File tree

‎go.mod‎

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@ require (
2929
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0
3030
go.opentelemetry.io/otel/sdk v1.43.0
3131
go.opentelemetry.io/otel/trace v1.43.0
32-
golang.org/x/sync v0.18.0
32+
golang.org/x/sync v0.19.0
3333
google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba
3434
google.golang.org/grpc v1.76.0
3535
google.golang.org/protobuf v1.36.10
@@ -105,6 +105,7 @@ require (
105105
github.com/prometheus/procfs v0.16.1 // indirect
106106
github.com/spiffe/go-spiffe/v2 v2.5.0 // indirect
107107
github.com/stretchr/objx v0.5.2 // indirect
108+
github.com/valkey-io/valkey-go v1.0.74
108109
github.com/zeebo/errs v1.4.0 // indirect
109110
github.com/zeebo/xxh3 v1.0.2 // indirect
110111
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
@@ -115,16 +116,16 @@ require (
115116
go.opentelemetry.io/otel/sdk/metric v1.43.0 // indirect
116117
go.opentelemetry.io/proto/otlp v1.7.1 // indirect
117118
go.yaml.in/yaml/v2 v2.4.2 // indirect
118-
golang.org/x/crypto v0.45.0 // indirect
119+
golang.org/x/crypto v0.46.0 // indirect
119120
golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 // indirect
120-
golang.org/x/mod v0.29.0 // indirect
121-
golang.org/x/net v0.47.0 // indirect
121+
golang.org/x/mod v0.30.0 // indirect
122+
golang.org/x/net v0.48.0 // indirect
122123
golang.org/x/oauth2 v0.33.0 // indirect
123124
golang.org/x/sys v0.42.0 // indirect
124-
golang.org/x/telemetry v0.0.0-20251008203120-078029d740a8 // indirect
125-
golang.org/x/text v0.31.0 // indirect
125+
golang.org/x/telemetry v0.0.0-20251111182119-bc8e575c7b54 // indirect
126+
golang.org/x/text v0.32.0 // indirect
126127
golang.org/x/time v0.14.0 // indirect
127-
golang.org/x/tools v0.38.0 // indirect
128+
golang.org/x/tools v0.39.0 // indirect
128129
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da // indirect
129130
google.golang.org/api v0.256.0 // indirect
130131
google.golang.org/genproto v0.0.0-20250922171735-9219d122eba9 // indirect

‎go.sum‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV
248248
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
249249
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
250250
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
251+
github.com/valkey-io/valkey-go v1.0.74 h1:NqtBHzjybz+is+c71hsyZP7hoE5lwCHQX026me0Vb08=
252+
github.com/valkey-io/valkey-go v1.0.74/go.mod h1:VGhZ6fs68Qrn2+OhH+6waZH27bjpgQOiLyUQyXuYK5k=
251253
github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
252254
github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
253255
github.com/zeebo/errs v1.4.0 h1:XNdoD/RRMKP7HD0UhJnIzUy74ISdGGxURlYG8HSWSfM=
@@ -287,19 +289,27 @@ go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
287289
golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
288290
golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
289291
golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
292+
golang.org/x/crypto v0.46.0 h1:cKRW/pmt1pKAfetfu+RCEvjvZkA9RimPbh7bhFjGVBU=
293+
golang.org/x/crypto v0.46.0/go.mod h1:Evb/oLKmMraqjZ2iQTwDwvCtJkczlDuTmdJXoZVzqU0=
290294
golang.org/x/exp v0.0.0-20240909161429-701f63a606c0 h1:e66Fs6Z+fZTbFBAxKfP3PALWBtpfqks2bwGcexMxgtk=
291295
golang.org/x/exp v0.0.0-20240909161429-701f63a606c0/go.mod h1:2TbTHSBQa924w8M6Xs1QcRcFwyucIwBGpK1p2f1YFFY=
292296
golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA=
293297
golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w=
298+
golang.org/x/mod v0.30.0 h1:fDEXFVZ/fmCKProc/yAXXUijritrDzahmwwefnjoPFk=
299+
golang.org/x/mod v0.30.0/go.mod h1:lAsf5O2EvJeSFMiBxXDki7sCgAxEUcZHXoXMKT4GJKc=
294300
golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
295301
golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY=
296302
golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU=
303+
golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
304+
golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
297305
golang.org/x/oauth2 v0.33.0 h1:4Q+qn+E5z8gPRJfmRy7C2gGG3T4jIprK6aSYgTXGRpo=
298306
golang.org/x/oauth2 v0.33.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA=
299307
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
300308
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
301309
golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I=
302310
golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
311+
golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4=
312+
golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
303313
golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
304314
golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
305315
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
@@ -310,12 +320,18 @@ golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
310320
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
311321
golang.org/x/telemetry v0.0.0-20251008203120-078029d740a8 h1:LvzTn0GQhWuvKH/kVRS3R3bVAsdQWI7hvfLHGgh9+lU=
312322
golang.org/x/telemetry v0.0.0-20251008203120-078029d740a8/go.mod h1:Pi4ztBfryZoJEkyFTI5/Ocsu2jXyDr6iSdgJiYE/uwE=
323+
golang.org/x/telemetry v0.0.0-20251111182119-bc8e575c7b54 h1:E2/AqCUMZGgd73TQkxUMcMla25GB9i/5HOdLr+uH7Vo=
324+
golang.org/x/telemetry v0.0.0-20251111182119-bc8e575c7b54/go.mod h1:hKdjCMrbv9skySur+Nek8Hd0uJ0GuxJIoIX2payrIdQ=
313325
golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM=
314326
golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM=
327+
golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
328+
golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
315329
golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
316330
golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
317331
golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ=
318332
golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs=
333+
golang.org/x/tools v0.39.0 h1:ik4ho21kwuQln40uelmciQPp9SipgNDdrafrYA4TmQQ=
334+
golang.org/x/tools v0.39.0/go.mod h1:JnefbkDPyD8UU2kI5fuf8ZX4/yUeh9W877ZeBONxUqQ=
319335
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da h1:noIWHXmPHxILtqtCOPIhSt0ABwskkZKjD3bXGnZGpNY=
320336
golang.org/x/xerrors v0.0.0-20240903120638-7835f813f4da/go.mod h1:NDW/Ps6MPRej6fsCIbMTohpP40sJ/P/vI1MoTEGwX90=
321337
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=

‎go/internal/feast/onlinestore/onlinestore.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,9 @@ func NewOnlineStore(config *registry.RepoConfig) (OnlineStore, error) {
6363
} else if onlineStoreType == "redis" {
6464
onlineStore, err := NewRedisOnlineStore(config.Project, config, config.OnlineStore)
6565
return onlineStore, err
66+
} else if onlineStoreType == "valkey" {
67+
onlineStore, err := NewValkeyOnlineStore(config.Project, config, config.OnlineStore)
68+
return onlineStore, err
6669
} else if onlineStoreType == "dynamodb" {
6770
onlineStore, err := NewDynamodbOnlineStore(config.Project, config, config.OnlineStore)
6871
return onlineStore, err

‎go/internal/feast/onlinestore/redisonlinestore.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -175,7 +175,7 @@ func (r *RedisOnlineStore) buildFeatureViewIndices(featureViewNames []string, fe
175175
return featureViewIndices, indicesFeatureView, index
176176
}
177177

178-
func (r *RedisOnlineStore) buildRedisHashSetKeys(featureViewNames []string, featureNames []string, indicesFeatureView map[int]string, index int) ([]string, []string) {
178+
func buildRedisHashSetKeys(featureViewNames []string, featureNames []string, indicesFeatureView map[int]string, index int) ([]string, []string) {
179179
featureCount := len(featureNames)
180180
var hsetKeys = make([]string, index)
181181
h := murmur3.New32()
@@ -218,7 +218,7 @@ func (r *RedisOnlineStore) OnlineRead(ctx context.Context, entityKeys []*types.E
218218

219219
featureCount := len(featureNames)
220220
featureViewIndices, indicesFeatureView, index := r.buildFeatureViewIndices(featureViewNames, featureNames)
221-
hsetKeys, featureNamesWithTimeStamps := r.buildRedisHashSetKeys(featureViewNames, featureNames, indicesFeatureView, index)
221+
hsetKeys, featureNamesWithTimeStamps := buildRedisHashSetKeys(featureViewNames, featureNames, indicesFeatureView, index)
222222
redisKeys, redisKeyToEntityIndex, err := r.buildRedisKeys(entityKeys)
223223
if err != nil {
224224
return nil, err

‎go/internal/feast/onlinestore/redisonlinestore_test.go‎

Lines changed: 3 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -107,16 +107,14 @@ func TestBuildFeatureViewIndices(t *testing.T) {
107107
}
108108

109109
func TestBuildHsetKeys(t *testing.T) {
110-
r := &RedisOnlineStore{}
111-
112110
t.Run("test with empty featureViewNames and featureNames", func(t *testing.T) {
113-
hsetKeys, featureNames := r.buildRedisHashSetKeys([]string{}, []string{}, map[int]string{}, 0)
111+
hsetKeys, featureNames := buildRedisHashSetKeys([]string{}, []string{}, map[int]string{}, 0)
114112
assert.Equal(t, 0, len(hsetKeys))
115113
assert.Equal(t, 0, len(featureNames))
116114
})
117115

118116
t.Run("test with non-empty featureViewNames and featureNames", func(t *testing.T) {
119-
hsetKeys, featureNames := r.buildRedisHashSetKeys([]string{"view1", "view2"}, []string{"feature1", "feature2"}, map[int]string{2: "view1", 3: "view2"}, 4)
117+
hsetKeys, featureNames := buildRedisHashSetKeys([]string{"view1", "view2"}, []string{"feature1", "feature2"}, map[int]string{2: "view1", 3: "view2"}, 4)
120118
assert.Equal(t, 4, len(hsetKeys))
121119
assert.Equal(t, 4, len(featureNames))
122120
assert.Equal(t, "_ts:view1", hsetKeys[2])
@@ -126,7 +124,7 @@ func TestBuildHsetKeys(t *testing.T) {
126124
})
127125

128126
t.Run("test with more featureViewNames than featureNames", func(t *testing.T) {
129-
hsetKeys, featureNames := r.buildRedisHashSetKeys([]string{"view1", "view2", "view3"}, []string{"feature1", "feature2", "feature3"}, map[int]string{3: "view1", 4: "view2", 5: "view3"}, 6)
127+
hsetKeys, featureNames := buildRedisHashSetKeys([]string{"view1", "view2", "view3"}, []string{"feature1", "feature2", "feature3"}, map[int]string{3: "view1", 4: "view2", 5: "view3"}, 6)
130128
assert.Equal(t, 6, len(hsetKeys))
131129
assert.Equal(t, 6, len(featureNames))
132130
assert.Equal(t, "_ts:view1", hsetKeys[3])

0 commit comments

Comments
 (0)