Somewhat intermediate state

Signed-off-by: Kevin Zhang <kzhang@tecton.ai>
feast-dev · feast-ci-bot · Mar 29, 2022 · Mar 23, 2022 · Mar 23, 2022 · Mar 24, 2022
commit 0fbd8dc8590b24801ff89a7900904b7d7e1c826c
diff --git a/go.mod b/go.mod
@@ -24,6 +24,9 @@ require (
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/kr/pretty v0.1.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.9 // indirect
+	github.com/go-sql-driver/mysql v1.6.0 // indirect
+	github.com/google/go-cmp v0.5.7 // indirect
+	github.com/mattn/go-sqlite3 v1.14.12 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	golang.org/x/exp v0.0.0-20211028214138-64b4c8e87d1a // indirect
 	golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f // indirect

diff --git a/go.sum b/go.sum
@@ -235,6 +235,8 @@ github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzE
 github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
 github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+github.com/mattn/go-sqlite3 v1.14.12 h1:TJ1bhYJPV44phC+IMu1u2K/i5RriLTPe+yc68XDJ1Z0=
+github.com/mattn/go-sqlite3 v1.14.12/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=

diff --git a/go/internal/feast/foo.db b/go/internal/feast/foo.db
@@ -1,10 +1,14 @@
 package feast
 
 import (
+	"database/sql"
+	"errors"
+
 	"context"
 	"fmt"
 
 	"github.com/feast-dev/feast/go/protos/feast/types"
+	_ "github.com/mattn/go-sqlite3"
 )
 
 type SqliteOnlineStore struct {
@@ -13,6 +17,7 @@ type SqliteOnlineStore struct {
 	project string
 
 	path string
+	db   *sql.DB
 }
 
 func NewSqliteOnlineStore(project string, onlineStoreConfig map[string]interface{}) (*SqliteOnlineStore, error) {
@@ -23,25 +28,72 @@ func NewSqliteOnlineStore(project string, onlineStoreConfig map[string]interface
 		return nil, fmt.Errorf("cannot find convert sqlite path to string %s", db_path)
 	} else {
 		store.path = dbPathStr
+		db, err := initializeConnection(dbPathStr)
+		if err != nil {
+			return nil, err
+		}
+		store.db = db
 	}
-	return &SqliteOnlineStore{
-		project: project,
-	}, nil
+	return &store, nil
 }
 
-func (*SqliteOnlineStore) Destruct() {
-
+func (s *SqliteOnlineStore) Destruct() {
+	s.db.Close()
 }
 
-func (*SqliteOnlineStore) OnlineRead(ctx context.Context, entityKeys []types.EntityKey, featureViewNames []string, featureNames []string) ([][]FeatureData, error) {
+func (s *SqliteOnlineStore) OnlineRead(ctx context.Context, entityKeys []types.EntityKey, featureViewNames []string, featureNames []string) ([][]FeatureData, error) {
+	_, err := s.getConnection()
+	if err != nil {
+		return nil, err
+	}
 	return nil, nil
 }
 
 // feature views, entities, dataframe?
-func (*SqliteOnlineStore) WriteToOnlineStore(ctx context.Context, featureViewName string, data [][]FeatureData) error {
+func (s *SqliteOnlineStore) WriteToOnlineStore(ctx context.Context, featureViewName string, data [][]FeatureData) error {
+	return nil
+}
+
+func (s *SqliteOnlineStore) Update(ctx context.Context, config *RepoConfig, tables_to_delete []*FeatureView, tables_to_keep []*FeatureView) error {
+	_, err := s.getConnection()
+	if err != nil {
+		return err
+	}
+	project := config.Project
+	for _, table := range tables_to_keep {
+		s.db.Exec(
+			fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (entity_key BLOB, feature_name TEXT, value BLOB, event_ts timestamp, created_ts timestamp,  PRIMARY KEY(entity_key, feature_name))", tableId(project, table)))
+		s.db.Exec(
+			fmt.Sprintf("CREATE INDEX IF NOT EXISTS %s_ek ON %s (entity_key);", tableId(project, table), tableId(project, table)))
+	}
+	for _, table := range tables_to_delete {
+		s.db.Exec("DROP TABLE IF EXISTS %s", tableId(project, table))
+	}
 	return nil
 }
 
-func (*SqliteOnlineStore) initializeConnection(db_path string) {
+func (s *SqliteOnlineStore) getConnection() (*sql.DB, error) {
+	if s.db == nil {
+		if s.path == "" {
+			return nil, errors.New("no database path available")
+		}
+		db, err := initializeConnection(s.path)
+		s.db = db
+		if err != nil {
+			return nil, err
+		}
+	}
+	return s.db, nil
+}
 
+func tableId(project string, table *FeatureView) string {
+	return fmt.Sprintf("%s_%s", project, table.base.name)
+}
+
+func initializeConnection(db_path string) (*sql.DB, error) {
+	db, err := sql.Open("sqlite3", "./foo.db")
+	if err != nil {
+		return nil, err
+	}
+	return db, nil
 }
@@ -0,0 +1,54 @@
+package feast
+
+import (
+	"fmt"
+	"log"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSqliteSetup(t *testing.T) {
+	dir := "../test/selected_tetra"
+	config, err := NewRepoConfigFromFile(dir)
+	assert.Nil(t, err)
+	assert.Equal(t, "selected_tetra", config.Project)
+	assert.Equal(t, "data/registry.db", config.GetRegistryConfig().Path)
+	assert.Equal(t, "local", config.Provider)
+	assert.Equal(t, map[string]interface{}{
+		"path": "data/online_store.db",
+	}, config.OnlineStore)
+	assert.Empty(t, config.OfflineStore)
+	assert.Empty(t, config.FeatureServer)
+	assert.Empty(t, config.Flags)
+}
+
+func TestSqliteUpdate(t *testing.T) {
+	dir := "../test/selected_tetra"
+	config, err := NewRepoConfigFromFile(dir)
+	assert.Nil(t, err)
+	store, err := NewSqliteOnlineStore("test", config.OnlineStore)
+	defer store.Destruct()
+	assert.Nil(t, err)
+	show_tables := `SELECT
+    	*
+	FROM
+		selected_tetra_driver_hourly_stats;`
+	rows, err := store.db.Query(show_tables)
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer rows.Close()
+	for rows.Next() {
+		fmt.Println("in")
+		var id int
+		var name string
+		err = rows.Scan(&id, &name)
+		if err != nil {
+			log.Fatal(err)
+		}
+		fmt.Println(id, name)
+	}
+	assert.False(t, true)
+
+}
diff --git a/go/internal/test/selected_tetra/__init__.py b/go/internal/test/selected_tetra/__init__.py
diff --git a/go/internal/test/selected_tetra/data/driver_stats.parquet b/go/internal/test/selected_tetra/data/driver_stats.parquet
diff --git a/go/internal/test/selected_tetra/data/online_store.db b/go/internal/test/selected_tetra/data/online_store.db
diff --git a/go/internal/test/selected_tetra/example.py b/go/internal/test/selected_tetra/example.py
@@ -0,0 +1,35 @@
+# This is an example feature definition file
+
+from datetime import timedelta
+
+from feast import Entity, Feature, FeatureView, FileSource, ValueType
+
+# Read data from parquet files. Parquet is convenient for local development mode. For
+# production, you can use your favorite DWH, such as BigQuery. See Feast documentation
+# for more info.
+driver_hourly_stats = FileSource(
+    path="/Users/kevinzhang/tecton-ai/offline_store/feast/go/internal/test/selected_tetra/data/driver_stats.parquet",
+    event_timestamp_column="event_timestamp",
+    created_timestamp_column="created",
+)
+
+# Define an entity for the driver. You can think of entity as a primary key used to
+# fetch features.
+driver = Entity(name="driver_id", value_type=ValueType.INT64, description="driver id",)
+
+# Our parquet files contain sample data that includes a driver_id column, timestamps and
+# three feature column. Here we define a Feature View that will allow us to serve this
+# data to our model online.
+driver_hourly_stats_view = FeatureView(
+    name="driver_hourly_stats",
+    entities=["driver_id"],
+    ttl=timedelta(days=1),
+    features=[
+        Feature(name="conv_rate", dtype=ValueType.FLOAT),
+        Feature(name="acc_rate", dtype=ValueType.FLOAT),
+        Feature(name="avg_daily_trips", dtype=ValueType.INT64),
+    ],
+    online=True,
+    batch_source=driver_hourly_stats,
+    tags={},
+)
diff --git a/go/internal/test/selected_tetra/feature_store.yaml b/go/internal/test/selected_tetra/feature_store.yaml
@@ -0,0 +1,5 @@
+project: selected_tetra
+registry: data/registry.db
+provider: local
+online_store:
+    path: data/online_store.db
diff --git a/sdk/python/tests/integration/feature_repos/repo_configuration.py b/sdk/python/tests/integration/feature_repos/repo_configuration.py
@@ -74,28 +74,28 @@
         [
             IntegrationTestRepoConfig(online_store=REDIS_CONFIG),
             # GCP configurations
-            IntegrationTestRepoConfig(
-                provider="gcp",
-                offline_store_creator=BigQueryDataSourceCreator,
-                online_store="datastore",
-            ),
-            IntegrationTestRepoConfig(
-                provider="gcp",
-                offline_store_creator=BigQueryDataSourceCreator,
-                online_store=REDIS_CONFIG,
-            ),
+            # IntegrationTestRepoConfig(
+            #     provider="gcp",
+            #     offline_store_creator=BigQueryDataSourceCreator,
+            #     online_store="datastore",
+            # ),
+            # IntegrationTestRepoConfig(
+            #     provider="gcp",
+            #     offline_store_creator=BigQueryDataSourceCreator,
+            #     online_store=REDIS_CONFIG,
+            # ),
             # AWS configurations
-            IntegrationTestRepoConfig(
-                provider="aws",
-                offline_store_creator=RedshiftDataSourceCreator,
-                online_store=DYNAMO_CONFIG,
-                python_feature_server=True,
-            ),
-            IntegrationTestRepoConfig(
-                provider="aws",
-                offline_store_creator=RedshiftDataSourceCreator,
-                online_store=REDIS_CONFIG,
-            ),
+            # IntegrationTestRepoConfig(
+            #     provider="aws",
+            #     offline_store_creator=RedshiftDataSourceCreator,
+            #     online_store=DYNAMO_CONFIG,
+            #     python_feature_server=True,
+            # ),
+            # IntegrationTestRepoConfig(
+            #     provider="aws",
+            #     offline_store_creator=RedshiftDataSourceCreator,
+            #     online_store=REDIS_CONFIG,
+            # ),
             # Snowflake configurations
             IntegrationTestRepoConfig(
                 provider="aws",  # no list features, no feature server