Repository: oom-ai/oomstore
Pull request #865 (2021-12-30 08:03:30)

Feat/online stream push
this PR does:
1. add method sqlutil.BuildPushCondition
2. refactor mysql push method with sqlutil.BuildPushCondition
3. implement postgres push method

relate #801 


Patch:
diff --git a/internal/database/dbutil/util.go b/internal/database/dbutil/util.go
index 0a7d24cb..c90081a0 100644
--- a/internal/database/dbutil/util.go
+++ b/internal/database/dbutil/util.go
@@ -29,7 +29,7 @@ func RandString(n int) string {
 	return string(b)
 }
 
-func Placeholders(size int, elem string, sep string) string {
+func Fill(size int, elem string, sep string) string {
 	r := make([]string, 0, size)
 
 	for i := 0; i < size; i++ {
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index bf45b636..e0edf785 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -3,10 +3,10 @@ package mysql
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/jmoiron/sqlx"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
@@ -50,22 +50,19 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
-	insertColumns := append([]string{opt.Entity.Name}, opt.FeatureNames...)
-	insertValues := append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
-
-	updateValues := opt.FeatureValues
-	updatePlaceholders := make([]string, 0, len(opt.FeatureNames))
-	for _, name := range opt.FeatureNames {
-		updatePlaceholders = append(updatePlaceholders, fmt.Sprintf("%s=?", name))
+	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, types.BackendMySQL)
+	if err != nil {
+		return err
 	}
 
 	query := fmt.Sprintf(`INSERT INTO %s (%s) VALUES(%s) ON DUPLICATE KEY UPDATE %s`,
 		tableName,
-		strings.Join(insertColumns, ","),
-		dbutil.Placeholders(len(insertColumns), "?", ","),
-		strings.Join(updatePlaceholders, ","),
+		inserts,
+		insertPlaceholders,
+		updatePlaceholders,
 	)
-	_, err := db.ExecContext(ctx, query, append(insertValues, updateValues...)...)
+
+	_, err = db.ExecContext(ctx, db.Rebind(query), values...)
 	return err
 }
 
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 434676af..ccd90e3e 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -2,6 +2,7 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
@@ -46,7 +47,23 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	panic("Implement me!")
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
+
+	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, types.BackendPostgres)
+	if err != nil {
+		return err
+	}
+
+	query := fmt.Sprintf(`INSERT INTO %s(%s) VALUES(%s) ON CONFLICT ("%s") DO UPDATE SET %s`,
+		tableName,
+		inserts,
+		insertPlaceholders,
+		opt.Entity.Name,
+		updatePlaceholders,
+	)
+
+	_, err = db.ExecContext(ctx, db.Rebind(query), values...)
+	return err
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/internal/database/online/sqlutil/push.go b/internal/database/online/sqlutil/push.go
new file mode 100644
index 00000000..9cc063ba
--- /dev/null
+++ b/internal/database/online/sqlutil/push.go
@@ -0,0 +1,32 @@
+package sqlutil
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+func BuildPushCondition(opt online.PushOpt, backend types.BackendType) (string, string, string, []interface{}, error) {
+	qt, err := dbutil.QuoteFn(backend)
+	if err != nil {
+		return "", "", "", nil, err
+	}
+
+	insertColumns := append([]string{opt.Entity.Name}, opt.FeatureNames...)
+	insertValues := append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
+
+	updatePlaceholders := make([]string, 0, len(opt.FeatureNames))
+	for _, name := range opt.FeatureNames {
+		updatePlaceholders = append(updatePlaceholders, fmt.Sprintf("%s=?", qt(name)))
+	}
+
+	return qt(insertColumns...),
+		dbutil.Fill(len(insertColumns), "?", ","),
+		strings.Join(updatePlaceholders, ","),
+		append(insertValues, opt.FeatureValues...),
+		nil
+
+}
diff --git a/oomcli/cmd/edit.go b/oomcli/cmd/edit.go
index c266f54a..7c8341a8 100644
--- a/oomcli/cmd/edit.go
+++ b/oomcli/cmd/edit.go
@@ -16,7 +16,7 @@ var editor = "vi"
 
 var editCmd = &cobra.Command{
 	Use:   "edit",
-	Short: "Edit a resource from the default editor",
+	Short: "Edit a resource from the default editor.",
 }
 
 func init() {
diff --git a/oomcli/cmd/snapshot.go b/oomcli/cmd/snapshot.go
deleted file mode 100644
index f1c4db67..00000000
--- a/oomcli/cmd/snapshot.go
+++ /dev/null
@@ -1,33 +0,0 @@
-package cmd
-
-import (
-	"context"
-	"log"
-
-	"github.com/spf13/cobra"
-)
-
-var snapshotGroupName string
-var snapshotCmd = &cobra.Command{
-	Use:   "snapshot",
-	Short: "Generate snapshots for the group",
-	Args:  cobra.RangeArgs(0, 1),
-	PreRun: func(cmd *cobra.Command, args []string) {
-		if len(args) == 1 {
-			snapshotGroupName = args[0]
-		}
-	},
-	Run: func(cmd *cobra.Command, args []string) {
-		ctx := context.Background()
-		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
-		defer oomStore.Close()
-
-		if err := oomStore.Snapshot(ctx, snapshotGroupName); err != nil {
-			log.Fatalf("failed to take snapshot for the group %s: %v\n", snapshotGroupName, err)
-		}
-	},
-}
-
-func init() {
-	rootCmd.AddCommand(snapshotCmd)
-}


Test patch:
diff --git a/internal/database/dbutil/util_test.go b/internal/database/dbutil/util_test.go
index 930891fb..2dd35dea 100644
--- a/internal/database/dbutil/util_test.go
+++ b/internal/database/dbutil/util_test.go
@@ -52,7 +52,7 @@ func TestPlaceholders(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.describe, func(t *testing.T) {
-			got := dbutil.Placeholders(c.size, c.elem, c.sep)
+			got := dbutil.Fill(c.size, c.elem, c.sep)
 			assert.Equal(t, c.want, got)
 		})
 	}
diff --git a/internal/database/online/postgres/store_test.go b/internal/database/online/postgres/store_test.go
index 48627621..e7ffbea0 100644
--- a/internal/database/online/postgres/store_test.go
+++ b/internal/database/online/postgres/store_test.go
@@ -57,6 +57,10 @@ func TestPrepareStreamTable(t *testing.T) {
 	test_impl.TestPrepareStreamTable(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
 
+func TestPush(t *testing.T) {
+	test_impl.TestPush(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
+}
+
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }