Skip to content

Commit 47d1c45

Browse files
felixwang9817felixwang9817
committed
Add SFVs to Go serving path
Signed-off-by: Felix Wang <wangfelix98@gmail.com>
1 parent 1f92e00 commit 47d1c45

3 files changed

Lines changed: 79 additions & 2 deletions

File tree

‎go/internal/feast/featurestore.go‎

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -224,6 +224,14 @@ func (fs *FeatureStore) listAllViews() (map[string]*model.FeatureView, map[strin
224224
fvs[featureView.Base.Name] = featureView
225225
}
226226

227+
streamFeatureViews, err := fs.ListStreamFeatureViews()
228+
if err != nil {
229+
return nil, nil, err
230+
}
231+
for _, streamFeatureView := range streamFeatureViews {
232+
fvs[streamFeatureView.Base.Name] = streamFeatureView
233+
}
234+
227235
onDemandFeatureViews, err := fs.registry.ListOnDemandFeatureViews(fs.config.Project)
228236
if err != nil {
229237
return nil, nil, err
@@ -242,6 +250,14 @@ func (fs *FeatureStore) ListFeatureViews() ([]*model.FeatureView, error) {
242250
return featureViews, nil
243251
}
244252

253+
func (fs *FeatureStore) ListStreamFeatureViews() ([]*model.FeatureView, error) {
254+
streamFeatureViews, err := fs.registry.ListStreamFeatureViews(fs.config.Project)
255+
if err != nil {
256+
return streamFeatureViews, err
257+
}
258+
return streamFeatureViews, nil
259+
}
260+
245261
func (fs *FeatureStore) ListEntities(hideDummyEntity bool) ([]*model.Entity, error) {
246262

247263
allEntities, err := fs.registry.ListEntities(fs.config.Project)

‎go/internal/feast/model/featureview.go‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,24 @@ type FeatureView struct {
2424

2525
func NewFeatureViewFromProto(proto *core.FeatureView) *FeatureView {
2626
featureView := &FeatureView{Base: NewBaseFeatureView(proto.Spec.Name, proto.Spec.Features),
27-
Ttl: &(*proto.Spec.Ttl),
27+
Ttl: proto.Spec.Ttl,
28+
}
29+
if len(proto.Spec.Entities) == 0 {
30+
featureView.EntityNames = []string{DUMMY_ENTITY_NAME}
31+
} else {
32+
featureView.EntityNames = proto.Spec.Entities
33+
}
34+
entityColumns := make([]*Field, len(proto.Spec.EntityColumns))
35+
for i, entityColumn := range proto.Spec.EntityColumns {
36+
entityColumns[i] = NewFieldFromProto(entityColumn)
37+
}
38+
featureView.EntityColumns = entityColumns
39+
return featureView
40+
}
41+
42+
func NewFeatureViewFromStreamFeatureViewProto(proto *core.StreamFeatureView) *FeatureView {
43+
featureView := &FeatureView{Base: NewBaseFeatureView(proto.Spec.Name, proto.Spec.Features),
44+
Ttl: proto.Spec.Ttl,
2845
}
2946
if len(proto.Spec.Entities) == 0 {
3047
featureView.EntityNames = []string{DUMMY_ENTITY_NAME}

‎go/internal/feast/registry/registry.go‎

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ type Registry struct {
3030
cachedFeatureServices map[string]map[string]*core.FeatureService
3131
cachedEntities map[string]map[string]*core.Entity
3232
cachedFeatureViews map[string]map[string]*core.FeatureView
33+
cachedStreamFeatureViews map[string]map[string]*core.StreamFeatureView
3334
cachedOnDemandFeatureViews map[string]map[string]*core.OnDemandFeatureView
3435
cachedRegistry *core.Registry
3536
cachedRegistryProtoLastUpdated time.Time
@@ -106,10 +107,12 @@ func (r *Registry) load(registry *core.Registry) {
106107
r.cachedFeatureServices = make(map[string]map[string]*core.FeatureService)
107108
r.cachedEntities = make(map[string]map[string]*core.Entity)
108109
r.cachedFeatureViews = make(map[string]map[string]*core.FeatureView)
110+
r.cachedStreamFeatureViews = make(map[string]map[string]*core.StreamFeatureView)
109111
r.cachedOnDemandFeatureViews = make(map[string]map[string]*core.OnDemandFeatureView)
110112
r.loadEntities(registry)
111113
r.loadFeatureServices(registry)
112114
r.loadFeatureViews(registry)
115+
r.loadStreamFeatureViews(registry)
113116
r.loadOnDemandFeatureViews(registry)
114117
r.cachedRegistryProtoLastUpdated = time.Now()
115118
}
@@ -144,6 +147,16 @@ func (r *Registry) loadFeatureViews(registry *core.Registry) {
144147
}
145148
}
146149

150+
func (r *Registry) loadStreamFeatureViews(registry *core.Registry) {
151+
streamFeatureViews := registry.StreamFeatureViews
152+
for _, streamFeatureView := range streamFeatureViews {
153+
if _, ok := r.cachedStreamFeatureViews[streamFeatureView.Spec.Project]; !ok {
154+
r.cachedStreamFeatureViews[streamFeatureView.Spec.Project] = make(map[string]*core.StreamFeatureView)
155+
}
156+
r.cachedStreamFeatureViews[streamFeatureView.Spec.Project][streamFeatureView.Spec.Name] = streamFeatureView
157+
}
158+
}
159+
147160
func (r *Registry) loadOnDemandFeatureViews(registry *core.Registry) {
148161
onDemandFeatureViews := registry.OnDemandFeatureViews
149162
for _, onDemandFeatureView := range onDemandFeatureViews {
@@ -193,7 +206,26 @@ func (r *Registry) ListFeatureViews(project string) ([]*model.FeatureView, error
193206
}
194207

195208
/*
196-
Look up Feature Views inside project
209+
Look up Stream Feature Views inside project
210+
Returns empty list if project not found
211+
*/
212+
213+
func (r *Registry) ListStreamFeatureViews(project string) ([]*model.FeatureView, error) {
214+
if cachedStreamFeatureViews, ok := r.cachedStreamFeatureViews[project]; !ok {
215+
return []*model.FeatureView{}, nil
216+
} else {
217+
streamFeatureViews := make([]*model.FeatureView, len(cachedStreamFeatureViews))
218+
index := 0
219+
for _, streamFeatureViewProto := range cachedStreamFeatureViews {
220+
streamFeatureViews[index] = model.NewFeatureViewFromStreamFeatureViewProto(streamFeatureViewProto)
221+
index += 1
222+
}
223+
return streamFeatureViews, nil
224+
}
225+
}
226+
227+
/*
228+
Look up Feature Services inside project
197229
Returns empty list if project not found
198230
*/
199231

@@ -254,6 +286,18 @@ func (r *Registry) GetFeatureView(project, featureViewName string) (*model.Featu
254286
}
255287
}
256288

289+
func (r *Registry) GetStreamFeatureView(project, streamFeatureViewName string) (*model.FeatureView, error) {
290+
if cachedStreamFeatureViews, ok := r.cachedStreamFeatureViews[project]; !ok {
291+
return nil, fmt.Errorf("no cached stream feature views found for project %s", project)
292+
} else {
293+
if streamFeatureViewProto, ok := cachedStreamFeatureViews[streamFeatureViewName]; !ok {
294+
return nil, fmt.Errorf("no cached stream feature view %s found for project %s", streamFeatureViewName, project)
295+
} else {
296+
return model.NewFeatureViewFromStreamFeatureViewProto(streamFeatureViewProto), nil
297+
}
298+
}
299+
}
300+
257301
func (r *Registry) GetFeatureService(project, featureServiceName string) (*model.FeatureService, error) {
258302
if cachedFeatureServices, ok := r.cachedFeatureServices[project]; !ok {
259303
return nil, fmt.Errorf("no cached feature services found for project %s", project)

0 commit comments

Comments
 (0)