semi working state

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>
feast-dev · feast-ci-bot · Apr 13, 2022 · Mar 9, 2022 · Mar 11, 2022 · Mar 17, 2022
commit ce12eeee0331c4d19983558005cb98bdf747f309
diff --git a/go.mod b/go.mod
@@ -27,6 +27,7 @@ require (
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/gonuts/commander v0.1.0 // indirect
 	github.com/gonuts/flag v0.1.0 // indirect
+<<<<<<< HEAD
 	github.com/google/flatbuffers v2.0.5+incompatible // indirect
 	github.com/klauspost/asmfmt v1.3.1 // indirect
 	github.com/klauspost/compress v1.15.1 // indirect
@@ -41,15 +42,20 @@ require (
 	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
 	golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27 // indirect
+=======
+>>>>>>> 8a7ccbe5 (semi working state)
 	github.com/google/flatbuffers v2.0.0+incompatible // indirect
 	github.com/google/go-cmp v0.5.7 // indirect
 	github.com/klauspost/compress v1.15.1 // indirect
 	github.com/pierrec/lz4/v4 v4.1.9 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/exp v0.0.0-20211028214138-64b4c8e87d1a // indirect
+	golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect
 	golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f // indirect
 	golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 // indirect
 	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/tools v0.1.10 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/genproto v0.0.0-20220126215142-9970aeb2e350 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect

diff --git a/go.sum b/go.sum
@@ -145,7 +145,9 @@ github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiu
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.4 h1:yAGX7huGHXlcLOEtBnF4w7FQwA26wojNCwOYAEhLjQM=
 github.com/golang/snappy v0.0.4/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
+github.com/gonuts/commander v0.1.0 h1:EcDTiVw9oAVORFjQOEOuHQqcl6OXMyTgELocTq6zJ0I=
 github.com/gonuts/commander v0.1.0/go.mod h1:qkb5mSlcWodYgo7vs8ulLnXhfinhZsZcm6+H/z1JjgY=
+github.com/gonuts/flag v0.1.0 h1:fqMv/MZ+oNGu0i9gp0/IQ/ZaPIDoAZBOBaJoV7viCWM=
 github.com/gonuts/flag v0.1.0/go.mod h1:ZTmTGtrSPejTo/SRNhCqwLTmiAgyBdCkLYhHrAoBdz4=
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
@@ -296,6 +298,7 @@ github.com/pierrec/lz4/v4 v4.1.12 h1:44l88ehTZAUGW4VlO1QC4zkilL99M6Y9MXNwEs0uzP8
 github.com/pierrec/lz4/v4 v4.1.12/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@@ -428,6 +431,8 @@ golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57 h1:LQmS1nU0twXLA96Kt7U
 golang.org/x/mod v0.6.0-dev.0.20211013180041-c96bc1413d57/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.5.1-0.20210830214625-1b1db11ec8f4/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
+golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 h1:kQgndtyPBW/JIYERgdxfwMYh3AVStj88WQTlNDi2a+o=
+golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3/go.mod h1:3p9vT2HGsQu2K1YbXdKPJLVgG5VJdoTa1poYQBtP1AY=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=

@@ -6,6 +6,7 @@ import (
 	"log"
 	"time"
 
+	"github.com/apache/arrow/go/arrow/array"
 	"github.com/feast-dev/feast/go/internal/feast"
 	"github.com/feast-dev/feast/go/protos/feast/serving"
 	"github.com/feast-dev/feast/go/protos/feast/types"
@@ -27,8 +28,15 @@ type Log struct {
 	RequestContext  map[string]*types.RepeatedValue
 }
 
+// driver_id,
+// 1003, 1004
+// [acc rate conv rate avg_daily_trips]
+// [entityvalues, acc_rate conv_rate avg_daily_trips, acc_ratestatus, conv_rate_status]
+// [entityvalues, entity value]
+
 type MemoryBuffer struct {
-	logs []*Log
+	featureService *feast.FeatureService
+	logs           []*Log
 }
 
 type LoggingService struct {
@@ -99,6 +107,7 @@ func (s *LoggingService) flushLogsToOfflineStorage(t time.Time) error {
 		return fmt.Errorf("could not get offline storage type for config: %s", s.fs.GetRepoConfig().OfflineStore)
 	}
 	if offlineStoreType == "file" {
+
 		s.offlineLogStorage.FlushToStorage(s.memoryBuffer)
 		//Clean memory buffer
 		s.memoryBuffer.logs = s.memoryBuffer.logs[:0]
@@ -108,3 +117,30 @@ func (s *LoggingService) flushLogsToOfflineStorage(t time.Time) error {
 	}
 	return nil
 }
+
+func (s *LoggingService) getLogInArrowTable(memoryBuffer *MemoryBuffer) (*array.Table, error) {
+	// input memoryBuffer -> featureColumns
+	// map[string]*type.Value
+
+	// fields := make([]*arrow.Field, 0)
+	// columns := make([]array.Interface, 0)
+	// for idx, feature := range featureService.features {
+	// 	feature.Name
+
+	// 	[]*proto.Value = columnNameToProtoValue[feature.Name]
+	// 	arrowArray := types.ProtoValuesToArrowArray(protoValues)
+
+	// 	fields = append(fields, &arrow.Field{
+	// 		Name: feature.Name,
+	// 		Type: arrowArray.DataType(),
+	// 	})
+	// 	columns = append(columns, arrowArray)
+	// }
+
+	// table := array.NewTable(
+	// 	arrow.NewSchema(fields, nil),
+	// 	columns
+	// 	)
+
+	// pqarrow.WriteTable(table)
+}
@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"log"
 
 	"github.com/feast-dev/feast/go/cmd/server/logging"
 	"github.com/feast-dev/feast/go/internal/feast"
@@ -32,6 +33,20 @@ func (s *servingServiceServer) GetFeastServingInfo(ctx context.Context, request
 // Results contains values including the value of the feature, the event timestamp, and feature status in a columnar format.
 func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *serving.GetOnlineFeaturesRequest) (*serving.GetOnlineFeaturesResponse, error) {
 	featuresOrService, err := s.fs.ParseFeatures(request.GetKind())
+	log.Println("Swag money 123")
+	if featuresOrService.FeatureService != nil {
+		log.Println("INDSFS")
+
+		log.Println(featuresOrService.FeatureService.Projections[0].Features)
+		log.Println(featuresOrService.FeatureService.Projections[0].JoinKeyMap)
+	}
+	// if featuresOrService.FeatureService != nil {
+	// 	if schema, ok := schemaCache[featuresOrService.FeatureService]; !ok {
+	// 		generateLogSchemaFromFeatureService(featuresOrService.FeatureService)
+	// 	}
+
+	// }
+
 	if err != nil {
 		return nil, err
 	}

diff --git a/go/internal/feast/basefeatureview.go b/go/internal/feast/basefeatureview.go
@@ -0,0 +1,43 @@
+package feast
+
+import (
+	"fmt"
+
+	"github.com/feast-dev/feast/go/protos/feast/core"
+)
+
+type BaseFeatureView struct {
+	name       string
+	features   []*Feature
+	projection *FeatureViewProjection
+}
+
+func NewBaseFeatureView(name string, featureProtos []*core.FeatureSpecV2) *BaseFeatureView {
+	base := &BaseFeatureView{name: name}
+	features := make([]*Feature, len(featureProtos))
+	for index, featureSpecV2 := range featureProtos {
+		features[index] = NewFeatureFromProto(featureSpecV2)
+	}
+	base.features = features
+	base.projection = NewFeatureViewProjectionFromDefinition(base)
+	return base
+}
+
+func (fv *BaseFeatureView) withProjection(projection *FeatureViewProjection) (*BaseFeatureView, error) {
+	if projection.name != fv.name {
+		return nil, fmt.Errorf("the projection for the %s FeatureView cannot be applied because it differs "+
+			"in name; the projection is named %s and the name indicates which "+
+			"FeatureView the projection is for", fv.name, projection.name)
+	}
+	features := make(map[string]bool)
+	for _, feature := range fv.features {
+		features[feature.name] = true
+	}
+	for _, feature := range projection.Features {
+		if _, ok := features[feature.name]; !ok {
+			return nil, fmt.Errorf("the projection for %s cannot be applied because it contains %s which the "+
+				"FeatureView doesn't have", projection.name, feature.name)
+		}
+	}
+	return &BaseFeatureView{name: fv.name, features: fv.features, projection: projection}, nil
+}
@@ -34,7 +34,6 @@ type Features struct {
 // repo config (contents of feature_store.yaml converted to JSON map).
 func NewFeatureStore(config *registry.RepoConfig, callback transformation.TransformationCallback) (*FeatureStore, error) {
 	onlineStore, err := onlinestore.NewOnlineStore(config)
-	// offlineStore, err := NewOfflineStore(config)
 	if err != nil {
 		return nil, err
 	}
@@ -137,7 +136,6 @@ func (fs *FeatureStore) GetOnlineFeatures(
 
 	for _, groupRef := range groupedRefs {
 		featureData, err := fs.readFromOnlineStore(ctx, groupRef.EntityKeys, groupRef.FeatureViewNames, groupRef.FeatureNames)
-
 		if err != nil {
 			return nil, err
 		}
@@ -186,10 +184,6 @@ func (fs *FeatureStore) DestructOnlineStore() {
 	fs.onlineStore.Destruct()
 }
 
-func (fs *FeatureStore) GetRepoConfig() *RepoConfig {
-	return fs.config
-}
-
 // ParseFeatures parses the kind field of a GetOnlineFeaturesRequest protobuf message
 // and populates a Features struct with the result.
 func (fs *FeatureStore) ParseFeatures(kind interface{}) (*Features, error) {

diff --git a/go/internal/feast/featureviewprojection.go b/go/internal/feast/featureviewprojection.go
@@ -0,0 +1,41 @@
+package feast
+
+import (
+	"github.com/feast-dev/feast/go/protos/feast/core"
+)
+
+type FeatureViewProjection struct {
+	name       string
+	nameAlias  string
+	Features   []*Feature
+	JoinKeyMap map[string]string
+}
+
+func (fv *FeatureViewProjection) nameToUse() string {
+	if len(fv.nameAlias) == 0 {
+		return fv.name
+	}
+	return fv.nameAlias
+}
+
+func NewFeatureViewProjectionFromProto(proto *core.FeatureViewProjection) *FeatureViewProjection {
+	featureProjection := &FeatureViewProjection{name: proto.FeatureViewName,
+		nameAlias:  proto.FeatureViewNameAlias,
+		JoinKeyMap: proto.JoinKeyMap,
+	}
+
+	features := make([]*Feature, len(proto.FeatureColumns))
+	for index, featureSpecV2 := range proto.FeatureColumns {
+		features[index] = NewFeatureFromProto(featureSpecV2)
+	}
+	featureProjection.Features = features
+	return featureProjection
+}
+
+func NewFeatureViewProjectionFromDefinition(base *BaseFeatureView) *FeatureViewProjection {
+	return &FeatureViewProjection{name: base.name,
+		nameAlias:  "",
+		Features:   base.features,
+		JoinKeyMap: make(map[string]string),
+	}
+}
diff --git a/sdk/python/tests/integration/online_store/test_universal_online.py b/sdk/python/tests/integration/online_store/test_universal_online.py
@@ -1094,3 +1094,4 @@ def assert_feature_service_entity_mapping_correctness(
                 entity_rows=entity_rows,
                 full_feature_names=full_feature_names,
             )
+        assert(False)