Fix

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>

go/cmd/server/logging/logging.go

@@ -40,7 +40,6 @@ type LoggingService struct {
 }
 
 func NewLoggingService(fs *feast.FeatureStore, logChannelCapacity int, featureServiceName string, enableLogging bool) (*LoggingService, error) {
-	// start handler processes?
 	var featureService *model.FeatureService = nil
 	var err error
 	if enableLogging {
@@ -94,16 +93,20 @@ func (s *LoggingService) processLogs() {
 	}
 }
 
+// Select that eitheringests new logs that are added to the logging channel, one at a time to add
+// to the in memory buffer or flushes all of them synchronously to the OfflineStorage on a time interval.
 func (s *LoggingService) ProcessMemoryBuffer(t *time.Ticker) {
 	select {
 	case t := <-t.C:
 		s.flushLogsToOfflineStorage(t)
 	case new_log := <-s.logChannel:
-		log.Printf("Pushing %s to memory.\n", new_log.FeatureValues)
+		log.Printf("Adding %s to memory.\n", new_log.FeatureValues)
 		s.memoryBuffer.logs = append(s.memoryBuffer.logs, new_log)
 	}
 }
 
+// Acquires the logging schema from the feature service, converts the memory buffer array of rows of logs and flushes
+// them to the offline storage.
 func (s *LoggingService) flushLogsToOfflineStorage(t time.Time) error {
 	log.Printf("Flushing buffer to offline storage with channel length: %d\n at time: "+t.String(), len(s.memoryBuffer.logs))
 	offlineStoreType, ok := getOfflineStoreType(s.fs.GetRepoConfig().OfflineStore)
@@ -158,9 +161,8 @@ func ConvertMemoryBufferToArrowTable(memoryBuffer *MemoryBuffer, fcoSchema *Sche
 			entityNameToEntityValues[entityName] = append(entityNameToEntityValues[entityName], l.EntityValue[idx])
 		}
 
-		// Contains both fv and odfv feature value types => add them in order of how the appear in the featureService
+		// Contains both fv and odfv feature value types => they are processed in order of how the appear in the featureService
 		for idx, featureName := range fcoSchema.Features {
-			// for featureName, idAndType := range fcoSchema.FeaturesTypes {
 			// populate the proto value arrays with values from memory buffer in separate columns one for each feature name
 			if _, ok := columnNameToProtoValueArray[featureName]; !ok {
 				columnNameToProtoValueArray[featureName] = make([]*types.Value, 0)
@@ -223,10 +225,8 @@ func ConvertMemoryBufferToArrowTable(memoryBuffer *MemoryBuffer, fcoSchema *Sche
 	)
 
 	result := array.Record(array.NewRecord(schema, columns, int64(len(memoryBuffer.logs))))
-	// create an arrow table -> write this to parquet.
 
 	tbl := array.NewTableFromRecords(schema, []array.Record{result})
-	// arrow table -> write this to parquet
 	return array.Table(tbl), nil
 }
 
@@ -245,7 +245,6 @@ func GetSchemaFromFeatureService(featureService *model.FeatureService, entities
 	for _, entity := range entities {
 		entityNames = append(entityNames, entity.JoinKey)
 	}
-	//featureViews, err := fs.listFeatureViews(hideDummyEntity)
 
 	for _, featureView := range featureViews {
 		fvs[featureView.Base.Name] = featureView
@@ -261,7 +260,6 @@ func GetSchemaFromFeatureService(featureService *model.FeatureService, entities
 	}
 
 	allFeatureTypes := make(map[string]types.ValueType_Enum)
-	//allRequestDataTypes := make(map[string]*types.ValueType_Enum)
 	features := make([]string, 0)
 	for _, featureProjection := range featureService.Projections {
 		// Create copies of FeatureView that may contains the same *FeatureView but

go/cmd/server/logging/logging_test.go

@@ -101,6 +101,7 @@ func TestSerializeToArrowTable(t *testing.T) {
 	}
 }
 
+// Initialize all dummy featureservice, entities and featureviews/on demand featureviews for testing.
 func InitializeFeatureRepoVariablesForTest() (*model.FeatureService, []*model.Entity, []*model.FeatureView, []*model.OnDemandFeatureView) {
 	f1 := model.NewFeature(
 		"int64",
@@ -153,6 +154,7 @@ func InitializeFeatureRepoVariablesForTest() (*model.FeatureService, []*model.En
 	return featureService, []*model.Entity{entity1}, []*model.FeatureView{featureView1, featureView2}, []*model.OnDemandFeatureView{}
 }
 
+// Create dummy FeatureService, Entities, and FeatureViews add them to the logger and convert the logs to Arrow table.
 func GenerateLogsAndConvertToArrowTable() (array.Table, error) {
 	featureService, entities, featureViews, odfvs := InitializeFeatureRepoVariablesForTest()
 	schema, err := GetSchemaFromFeatureService(featureService, entities, featureViews, odfvs)

go/cmd/server/server.go

@@ -54,7 +54,7 @@ func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *s
 	}
 	// Entities are currently part of the features as a value and the order that we add it to the resp MetaData
 	// Need to figure out a way to map the correct entities to the correct ordering
-	entityValues := make(map[string][]*prototypes.Value, 0)
+	entityValuesMap := make(map[string][]*prototypes.Value, 0)
 	featureNames := make([]string, len(featureVectors))
 	for idx, vector := range featureVectors {
 
@@ -65,7 +65,7 @@ func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *s
 			return nil, err
 		}
 		if _, ok := request.Entities[vector.Name]; ok {
-			entityValues[vector.Name] = values
+			entityValuesMap[vector.Name] = values
 		}
 		resp.Results = append(resp.Results, &serving.GetOnlineFeaturesResponse_FeatureVector{
 			Values:          values,
@@ -74,7 +74,7 @@ func (s *servingServiceServer) GetOnlineFeatures(ctx context.Context, request *s
 		})
 	}
 
-	go generateLogs(s, entityValues, featureNames, resp.Results, request.RequestContext)
+	go generateLogs(s, entityValuesMap, featureNames, resp.Results, request.RequestContext)
 	return resp, nil
 }
 

go/cmd/server/server_test.go

@@ -194,15 +194,18 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 		Entities: entities,
 	}
 	response, err := client.GetOnlineFeatures(ctx, request)
+	// Wait for logger to flush.
 	time.Sleep(1 * time.Second)
 	assert.Nil(t, err)
 	assert.NotNil(t, response)
-	// Wait for logger to flush.
-	// Get the featurenames without the entity order
+
+	// Get the featurenames without the entity names that are appended at the front.
 	featureNames := response.Metadata.FeatureNames.Val[len(request.Entities):]
+	// Generated expected log rows and values
+	// TODO(kevjumba): implement for timestamp and status
 	expectedLogValues, _, _ := GetExpectedLogRows(featureNames, response.Results)
 	expectedLogValues["driver_id"] = entities["driver_id"]
-	logPath, err := filepath.Abs(filepath.Join(".", "log.parquet"))
+	logPath, err := filepath.Abs(filepath.Join(dir, "log.parquet"))
 	assert.Nil(t, err)
 	w, err := logging.CreateOrOpenLogFile(logPath)
 	assert.Nil(t, err)
@@ -225,6 +228,7 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 
 		assert.Nil(t, err)
 		assert.Equal(t, len(values), len(expectedLogValues))
+		// Need to iterate through and compare because certain values in types.RepeatedValues aren't accurately being compared.
 		for name, val := range values {
 			assert.Equal(t, len(val.Val), len(expectedLogValues[name].Val))
 			for idx, featureVal := range val.Val {
@@ -236,6 +240,7 @@ func TestGetOnlineFeaturesSqliteWithLogging(t *testing.T) {
 	assert.Nil(t, err)
 }
 
+// Generate the expected log rows based on the resulting feature vector returned from GetOnlineFeatures.
 func GetExpectedLogRows(featureNames []string, results []*serving.GetOnlineFeaturesResponse_FeatureVector) (map[string]*types.RepeatedValue, [][]int32, [][]int64) {
 	numFeatures := len(featureNames)
 	numRows := len(results[0].Values)

go/internal/feast/featurestore.go

@@ -4,17 +4,14 @@ import (
 	"context"
 	"errors"
 
-	"github.com/apache/arrow/go/v8/arrow/array"
 	"github.com/apache/arrow/go/v8/arrow/memory"
 	"github.com/feast-dev/feast/go/internal/feast/model"
 	"github.com/feast-dev/feast/go/internal/feast/onlineserving"
 	"github.com/feast-dev/feast/go/internal/feast/onlinestore"
 	"github.com/feast-dev/feast/go/internal/feast/registry"
 	"github.com/feast-dev/feast/go/internal/feast/transformation"
 	"github.com/feast-dev/feast/go/protos/feast/serving"
-	"github.com/feast-dev/feast/go/protos/feast/types"
 	prototypes "github.com/feast-dev/feast/go/protos/feast/types"
-	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 type FeatureStore struct {
@@ -32,60 +29,14 @@ type Features struct {
 	FeatureService *model.FeatureService
 }
 
-/*
-	FeatureVector type represent result of retrieving single feature for multiple rows.
-	It can be imagined as a column in output dataframe / table.
-	It contains of feature name, list of values (across all rows),
-	list of statuses and list of timestamp. All these lists have equal length.
-	And this length is also equal to number of entity rows received in request.
-*/
-type FeatureVector struct {
-	Name       string
-	Values     array.Interface
-	Statuses   []serving.FieldStatus
-	Timestamps []*timestamppb.Timestamp
-}
-
-type featureViewAndRefs struct {
-	view        *model.FeatureView
-	featureRefs []string
-}
-
 func (fs *FeatureStore) Registry() *registry.Registry {
 	return fs.registry
 }
 
-func (f *featureViewAndRefs) View() *model.FeatureView {
-	return f.view
-}
-
-func (f *featureViewAndRefs) FeatureRefs() []string {
-	return f.featureRefs
-}
-
 func (fs *FeatureStore) GetRepoConfig() *registry.RepoConfig {
 	return fs.config
 }
 
-/*
-	We group all features from a single request by entities they attached to.
-	Thus, we will be able to call online retrieval per entity and not per each feature view.
-	In this struct we collect all features and views that belongs to a group.
-	We also store here projected entity keys (only ones that needed to retrieve these features)
-	and indexes to map result of retrieval into output response.
-*/
-type GroupedFeaturesPerEntitySet struct {
-	// A list of requested feature references of the form featureViewName:featureName that share this entity set
-	featureNames     []string
-	featureViewNames []string
-	// full feature references as they supposed to appear in response
-	aliasedFeatureNames []string
-	// Entity set as a list of EntityKeys to pass to OnlineRead
-	entityKeys []*prototypes.EntityKey
-	// Reversed mapping to project result of retrieval from storage to response
-	indices [][]int
-}
-
 // NewFeatureStore constructs a feature store fat client using the
 // repo config (contents of feature_store.yaml converted to JSON map).
 func NewFeatureStore(config *registry.RepoConfig, callback transformation.TransformationCallback) (*FeatureStore, error) {
@@ -325,14 +276,14 @@ func (fs *FeatureStore) GetFeatureView(featureViewName string, hideDummyEntity b
 	return fv, nil
 }
 
-func (fs *FeatureStore) readFromOnlineStore(ctx context.Context, entityRows []*types.EntityKey,
+func (fs *FeatureStore) readFromOnlineStore(ctx context.Context, entityRows []*prototypes.EntityKey,
 	requestedFeatureViewNames []string,
 	requestedFeatureNames []string,
 ) ([][]onlinestore.FeatureData, error) {
 	numRows := len(entityRows)
-	entityRowsValue := make([]*types.EntityKey, numRows)
+	entityRowsValue := make([]*prototypes.EntityKey, numRows)
 	for index, entityKey := range entityRows {
-		entityRowsValue[index] = &types.EntityKey{JoinKeys: entityKey.JoinKeys, EntityValues: entityKey.EntityValues}
+		entityRowsValue[index] = &prototypes.EntityKey{JoinKeys: entityKey.JoinKeys, EntityValues: entityKey.EntityValues}
 	}
 	return fs.onlineStore.OnlineRead(ctx, entityRowsValue, requestedFeatureViewNames, requestedFeatureNames)
 }

go/internal/feast/featurestore_test.go

@@ -41,7 +41,7 @@ func TestNewFeatureStore(t *testing.T) {
 }
 
 func TestGetOnlineFeaturesRedis(t *testing.T) {
-	//t.Skip("@todo(achals): feature_repo isn't checked in yet")
+	t.Skip("@todo(achals): feature_repo isn't checked in yet")
 	config := registry.RepoConfig{
 		Project:  "feature_repo",
 		Registry: getRegistryPath(),

go/internal/feast/registry/registry.go

@@ -106,11 +106,11 @@ func (r *Registry) getRegistryProto() (*core.Registry, error) {
 	if err != nil {
 		return registryProto, err
 	}
-	r.Load(registryProto)
+	r.load(registryProto)
 	return registryProto, nil
 }
 
-func (r *Registry) Load(registry *core.Registry) {
+func (r *Registry) load(registry *core.Registry) {
 	r.mu.Lock()
 	defer r.mu.Unlock()
 	r.cachedRegistry = registry