Repository: oom-ai/oomstore
Pull request #1105 (2022-01-27 03:26:56)

feat(json-schema): validate config to apply with jsonschema
implement #1098 

TODO:
1. time.Duration type
2. references outside the schema for nesting data structures


Patch:
diff --git a/README.md b/README.md
index d1de9fcb..4d33fa09 100644
--- a/README.md
+++ b/README.md
@@ -126,7 +126,7 @@ oomcli sync --revision-id 2
 
 ```bash
 oomcli get online \
-  --entity-key 1006 \
+  --entity-keys 1006 \
   --feature account.state,account.credit_score,account.account_age_days,account.has_2fa_installed,transaction_stats.transaction_count_7d,transaction_stats.transaction_count_30d
 ```
 
diff --git a/go.mod b/go.mod
index 00079c27..539db2c2 100644
--- a/go.mod
+++ b/go.mod
@@ -22,6 +22,7 @@ require (
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee
 	github.com/pkg/errors v0.9.1
+	github.com/qri-io/jsonschema v0.2.1
 	github.com/snowflakedb/gosnowflake v1.6.5
 	github.com/spf13/cast v1.4.1
 	github.com/spf13/cobra v1.3.0
@@ -100,6 +101,7 @@ require (
 	github.com/prometheus/client_model v0.2.0 // indirect
 	github.com/prometheus/common v0.32.1 // indirect
 	github.com/prometheus/procfs v0.7.3 // indirect
+	github.com/qri-io/jsonpointer v0.1.1 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 // indirect
 	github.com/rivo/uniseg v0.2.0 // indirect
 	github.com/sirupsen/logrus v1.8.1 // indirect
diff --git a/go.sum b/go.sum
index 81d4d458..933e6c1a 100644
--- a/go.sum
+++ b/go.sum
@@ -718,6 +718,10 @@ github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/procfs v0.7.3 h1:4jVXhlkAyzOScmCkXBTOLRLTz8EeU+eyjrwB/EPq0VU=
 github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/qri-io/jsonpointer v0.1.1 h1:prVZBZLL6TW5vsSB9fFHFAMBLI4b0ri5vribQlTJiBA=
+github.com/qri-io/jsonpointer v0.1.1/go.mod h1:DnJPaYgiKu56EuDp8TU5wFLdZIcAnb/uH9v37ZaMV64=
+github.com/qri-io/jsonschema v0.2.1 h1:NNFoKms+kut6ABPf6xiKNM5214jzxAhDBrPHCJ97Wg0=
+github.com/qri-io/jsonschema v0.2.1/go.mod h1:g7DPkiOsK1xv6T/Ao5scXRkd+yTFygcANPBaaqW+VrI=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0 h1:OdAsTTz6OkFY5QxjkYwrChwuRruF69c169dPK26NUlk=
 github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
 github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
@@ -734,7 +738,9 @@ github.com/sagikazarmark/crypt v0.3.0/go.mod h1:uD/D+6UF4SrIR1uGEv7bBNkNqLGqUr43
 github.com/sasha-s/go-deadlock v0.2.0/go.mod h1:StQn567HiB1fF2yJ44N9au7wOhrPS3iZqiDbRupzT10=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
+github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.0.1-0.20180205163309-da645544ed44/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/sergi/go-diff v1.1.0 h1:we8PVUC3FE2uYfodKH/nBHMSetSfHDR6scGdBi+erh0=
 github.com/sergi/go-diff v1.1.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
 github.com/shirou/gopsutil v3.21.2+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
diff --git a/internal/database/dbutil/util.go b/internal/database/dbutil/util.go
index a03ca5a1..ea97c2ee 100644
--- a/internal/database/dbutil/util.go
+++ b/internal/database/dbutil/util.go
@@ -39,8 +39,8 @@ func Fill(size int, elem string, sep string) string {
 	return strings.Join(r, sep)
 }
 
-func OfflineBatchSnapshotTableName(groupID int, revisionID int64) string {
-	return fmt.Sprintf("offline_batch_snapshot_%d_%d", groupID, revisionID)
+func OfflineBatchTableName(groupID int, revisionID int64) string {
+	return fmt.Sprintf("offline_batch_%d_%d", groupID, revisionID)
 }
 
 func OfflineStreamSnapshotTableName(groupID int, revision int64) string {
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 898b024e..9aef6f2c 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -6,6 +6,7 @@ import (
 	"github.com/go-sql-driver/mysql"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
 
@@ -33,5 +34,20 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		return 0, "", errdefs.WithStack(err)
 	}
 
+	if opt.SnapshotTable == nil {
+		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
+		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, revisionID)
+		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		rowsAffected, err := result.RowsAffected()
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		if rowsAffected != 1 {
+			return 0, "", errdefs.Errorf("failed to update revision %d: revision not found", revisionID)
+		}
+	}
 	return int(revisionID), snapshotTable, nil
 }
diff --git a/internal/database/metadata/postgres/revision.go b/internal/database/metadata/postgres/revision.go
index d750aa61..bf511f17 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -7,6 +7,7 @@ import (
 	"github.com/lib/pq"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
 
@@ -29,6 +30,21 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		}
 		return 0, "", errdefs.WithStack(err)
 	}
+	if opt.SnapshotTable == nil {
+		updateQuery := "UPDATE feature_group_revision SET snapshot_table = $1 WHERE id = $2"
+		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, int64(revisionID))
+		result, err := sqlxCtx.ExecContext(ctx, updateQuery, snapshotTable, revisionID)
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		rowsAffected, err := result.RowsAffected()
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		if rowsAffected != 1 {
+			return 0, "", errdefs.Errorf("failed to update revision %d: revision not found", revisionID)
+		}
+	}
 
 	return revisionID, snapshotTable, nil
 }
diff --git a/internal/database/metadata/sqlite/revision.go b/internal/database/metadata/sqlite/revision.go
index 7f506ee4..0b540648 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -7,6 +7,7 @@ import (
 	"modernc.org/sqlite"
 	sqlite3 "modernc.org/sqlite/lib"
 
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
 
@@ -34,5 +35,20 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		return 0, "", errdefs.WithStack(err)
 	}
 
+	if opt.SnapshotTable == nil {
+		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
+		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, revisionID)
+		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		rowsAffected, err := result.RowsAffected()
+		if err != nil {
+			return 0, "", errdefs.WithStack(err)
+		}
+		if rowsAffected != 1 {
+			return 0, "", errdefs.Errorf("failed to update revision %d: revision not found", revisionID)
+		}
+	}
 	return int(revisionID), snapshotTable, nil
 }
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index d03983bd..bec13239 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -37,10 +37,20 @@ type DoJoinOpt struct {
 }
 
 func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.JoinResult, error) {
-	if err := validateJoinOpt(opt); err != nil {
-		data := make(chan []interface{})
-		defer close(data)
-		return &types.JoinResult{Data: data}, nil
+	{
+		empty := true
+		for _, features := range opt.FeatureMap {
+			if len(features) > 0 {
+				empty = false
+				break
+			}
+		}
+		if empty {
+			data := make(chan []interface{})
+			defer close(data)
+			return &types.JoinResult{Data: data}, nil
+		}
+
 	}
 
 	// Step 1: prepare temporary table entity_rows
@@ -107,19 +117,6 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	})
 }
 
-func validateJoinOpt(opt DoJoinOpt) error {
-	for groupName, features := range opt.FeatureMap {
-		if len(features) == 0 {
-			delete(opt.FeatureMap, groupName)
-			delete(opt.RevisionRangeMap, groupName)
-		}
-	}
-	if len(opt.FeatureMap) == 0 || len(opt.RevisionRangeMap) == 0 {
-		return errdefs.Errorf("empty feature map")
-	}
-	return nil
-}
-
 type joinOneGroupOpt struct {
 	GroupName           string
 	Category            types.Category
diff --git a/oomcli/cmd/get_online.go b/oomcli/cmd/get_online.go
index 28ad7ddb..c2b991b3 100644
--- a/oomcli/cmd/get_online.go
+++ b/oomcli/cmd/get_online.go
@@ -41,8 +41,8 @@ func init() {
 
 	flags := getOnlineCmd.Flags()
 
-	flags.StringSliceVarP(&multiGetOpt.EntityKeys, "entity-key", "k", nil, "entity keys")
-	_ = getOnlineCmd.MarkFlagRequired("entity-key")
+	flags.StringSliceVarP(&multiGetOpt.EntityKeys, "entity-keys", "k", nil, "entity keys")
+	_ = getOnlineCmd.MarkFlagRequired("entity-keys")
 
 	flags.StringSliceVar(&multiGetOpt.FeatureNames, "feature", nil, "feature full names")
 	_ = getOnlineCmd.MarkFlagRequired("feature")
diff --git a/oomcli/cmd/push.go b/oomcli/cmd/push.go
index b65eb74a..58672bd0 100644
--- a/oomcli/cmd/push.go
+++ b/oomcli/cmd/push.go
@@ -54,6 +54,6 @@ func init() {
 	flags.StringVarP(&pushOpt.GroupName, "group", "g", "", "feature group")
 	_ = pushCmd.MarkFlagRequired("group")
 
-	flags.StringToStringVarP(&pushOpt.FeatureValues, "feature", "f", nil, "feature name-value pairs")
-	_ = pushCmd.MarkFlagRequired("feature")
+	flags.StringToStringVarP(&pushOpt.FeatureValues, "features", "f", nil, "feature name-value pairs")
+	_ = pushCmd.MarkFlagRequired("features")
 }
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index cc939dbd..38a82312 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -57,7 +57,7 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 		return 0, err
 	}
 
-	newRevisionID, _, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+	newRevisionID, snapshotTableName, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
 		Revision:      time.Now().UnixMilli(),
 		GroupID:       opt.group.ID,
 		SnapshotTable: nil,
@@ -67,7 +67,7 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 	if err != nil {
 		return 0, err
 	}
-	snapshotTableName := dbutil.OfflineBatchSnapshotTableName(opt.group.ID, int64(newRevisionID))
+
 	revision, err := s.offline.Import(ctx, offline.ImportOpt{
 		EntityName:        opt.entityName,
 		Features:          opt.features,
@@ -83,10 +83,9 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 	if opt.Revision != nil {
 		revision = *opt.Revision
 	}
-	if err = s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
-		RevisionID:       newRevisionID,
-		NewRevision:      &revision,
-		NewSnapshotTable: &snapshotTableName,
+	if err := s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
+		RevisionID:  newRevisionID,
+		NewRevision: &revision,
 	}); err != nil {
 		return 0, err
 	}
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index d4e3dee0..2ee4656a 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -30,14 +30,14 @@ func (s *OomStore) createRevision(ctx context.Context, opt metadata.CreateRevisi
 	var snapshotTable string
 	var dummyRevision *types.Revision
 
-	if err := s.metadata.WithTransaction(ctx, func(c context.Context, tx metadata.DBStore) error {
-		_, err := tx.GetRevisionBy(ctx, opt.GroupID, 0)
+	if err := s.metadata.WithTransaction(ctx, func(c context.Context, db metadata.DBStore) error {
+		_, err := s.metadata.GetRevisionBy(ctx, opt.GroupID, 0)
 		if err != nil {
 			if !errdefs.IsNotFound(err) {
 				return err
 			}
 
-			if _, _, err = tx.CreateRevision(ctx, metadata.CreateRevisionOpt{
+			if _, _, err = s.metadata.CreateRevision(ctx, metadata.CreateRevisionOpt{
 				Revision:    0,
 				GroupID:     opt.GroupID,
 				Description: "dummy revision will be used at Join and Export",
@@ -45,13 +45,12 @@ func (s *OomStore) createRevision(ctx context.Context, opt metadata.CreateRevisi
 				return err
 			}
 
-			dummyRevision, err = tx.GetRevisionBy(ctx, opt.GroupID, 0)
-			if err != nil {
+			if dummyRevision, err = s.metadata.GetRevisionBy(ctx, opt.GroupID, 0); err != nil {
 				return err
 			}
 		}
 
-		revisionID, snapshotTable, err = tx.CreateRevision(ctx, opt)
+		revisionID, snapshotTable, err = s.metadata.CreateRevision(ctx, opt)
 		return err
 	}); err != nil {
 		return 0, "", err
@@ -67,12 +66,7 @@ func (s *OomStore) createRevision(ctx context.Context, opt metadata.CreateRevisi
 }
 
 func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revision *types.Revision) error {
-	var snapshotTableName string
-	if revision.Group.Category == types.CategoryStream {
-		snapshotTableName = dbutil.OfflineStreamSnapshotTableName(revision.GroupID, revision.Revision)
-	} else {
-		snapshotTableName = dbutil.OfflineBatchSnapshotTableName(revision.GroupID, int64(revision.ID))
-	}
+	snapshotTable := dbutil.OfflineStreamSnapshotTableName(revision.GroupID, revision.Revision)
 
 	// Create snapshot table in offline store
 	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
@@ -83,7 +77,7 @@ func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revision *type
 	}
 
 	if err = s.offline.CreateTable(ctx, offline.CreateTableOpt{
-		TableName:  snapshotTableName,
+		TableName:  snapshotTable,
 		EntityName: revision.Group.Entity.Name,
 		Features:   features,
 		TableType:  types.TableStreamSnapshot,
@@ -108,7 +102,7 @@ func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revision *type
 	// Update snapshot_table in feature_group_revision table
 	return s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
 		RevisionID:       revision.ID,
-		NewSnapshotTable: &snapshotTableName,
+		NewSnapshotTable: &snapshotTable,
 		NewCdcTable:      cdcTable,
 	})
 }
diff --git a/pkg/oomstore/types/apply/apply.go b/pkg/oomstore/types/apply/apply.go
index ec08b66b..4e5f2dea 100644
--- a/pkg/oomstore/types/apply/apply.go
+++ b/pkg/oomstore/types/apply/apply.go
@@ -4,7 +4,6 @@ import (
 	"io"
 	"time"
 
-	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -27,21 +26,11 @@ func NewApplyStage() *ApplyStage {
 }
 
 type Feature struct {
-	Kind        string `mapstructure:"kind" yaml:"kind,omitempty"`
-	Name        string `mapstructure:"name" yaml:"name"`
-	GroupName   string `mapstructure:"group-name" yaml:"group-name,omitempty"`
-	ValueType   string `mapstructure:"value-type" yaml:"value-type"`
-	Description string `mapstructure:"description" yaml:"description"`
-}
-
-func (f *Feature) Validate() error {
-	if f.Name == "" {
-		return errdefs.InvalidAttribute(errdefs.Errorf("the name of feature should not be empty"))
-	}
-	if f.ValueType == "" {
-		return errdefs.InvalidAttribute(errdefs.Errorf("the value type of feature should not be empty"))
-	}
-	return nil
+	Kind        string `mapstructure:"kind" yaml:"kind,omitempty" json:"kind,omitempty"`
+	Name        string `mapstructure:"name" yaml:"name" json:"name"`
+	GroupName   string `mapstructure:"group-name" yaml:"group-name,omitempty" json:"group-name,omitempty"`
+	ValueType   string `mapstructure:"value-type" yaml:"value-type" json:"value-type"`
+	Description string `mapstructure:"description" yaml:"description" json:"description"`
 }
 
 type FeatureItems struct {
@@ -73,13 +62,13 @@ func FromFeatureList(features types.FeatureList) FeatureItems {
 }
 
 type Group struct {
-	Kind             string         `mapstructure:"kind" yaml:"kind,omitempty"`
-	Name             string         `mapstructure:"name" yaml:"name,omitempty"`
-	EntityName       string         `mapstructure:"entity-name" yaml:"entity-name,omitempty"`
-	Category         types.Category `mapstructure:"category" yaml:"category,omitempty"`
-	SnapshotInterval time.Duration  `mapstructure:"snapshot-interval" yaml:"snapshot-interval,omitempty"`
-	Description      string         `mapstructure:"description" yaml:"description"`
-	Features         []Feature      `mapstructure:"features" yaml:"features,omitempty"`
+	Kind             string         `mapstructure:"kind" yaml:"kind,omitempty" json:"kind,omitempty"`
+	Name             string         `mapstructure:"name" yaml:"name,omitempty" json:"name"`
+	EntityName       string         `mapstructure:"entity-name" yaml:"entity-name,omitempty" json:"entity-name,omitempty"`
+	Category         types.Category `mapstructure:"category" yaml:"category,omitempty" json:"category,omitempty"`
+	SnapshotInterval time.Duration  `mapstructure:"snapshot-interval" yaml:"snapshot-interval,omitempty" json:"snapshot-interval,omitempty"`
+	Description      string         `mapstructure:"description" yaml:"description" json:"description"`
+	Features         []Feature      `mapstructure:"features" yaml:"features,omitempty" json:"features,omitempty"`
 }
 
 type GroupItems struct {
@@ -128,28 +117,14 @@ func FromGroupList(groups types.GroupList, features types.FeatureList) *GroupIte
 	return items
 }
 
-func (g *Group) Validate() error {
-	if g.Name == "" {
-		return errdefs.InvalidAttribute(errdefs.Errorf("the name of group should not be empty"))
-	}
-	return nil
-}
-
 type Entity struct {
-	Kind        string `mapstructure:"kind" yaml:"kind"`
-	Name        string `mapstructure:"name" yaml:"name"`
-	Description string `mapstructure:"description" yaml:"description"`
+	Kind        string `mapstructure:"kind" yaml:"kind" json:"kind"`
+	Name        string `mapstructure:"name" yaml:"name" json:"name"`
+	Description string `mapstructure:"description" yaml:"description" json:"description"`
 
 	Groups []Group `mapstructure:"groups" yaml:"groups,omitempty"`
 }
 
-func (e *Entity) Validate() error {
-	if e.Name == "" {
-		return errdefs.InvalidAttribute(errdefs.Errorf("the name of entity should not be empty"))
-	}
-	return nil
-}
-
 type EntityItems struct {
 	Items []Entity `yaml:"items"`
 }
diff --git a/pkg/oomstore/types/apply/entity.json b/pkg/oomstore/types/apply/entity.json
new file mode 100644
index 00000000..bab9a378
--- /dev/null
+++ b/pkg/oomstore/types/apply/entity.json
@@ -0,0 +1,22 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$title": "apply entity json schema",
+    "description": "apply entity json schema",
+    "type": "object",
+    "properties": {
+        "kind": {
+            "description": "must be Entity",
+            "enum": ["Entity"]
+        },
+        "name": {
+            "description": "name of the entity",
+            "type": "string",
+            "minLength": 1
+        },
+        "description": {
+            "description": "the entity description",
+            "type": "string"
+        }
+    },
+    "required": ["kind","name"]
+}
diff --git a/pkg/oomstore/types/apply/feature.json b/pkg/oomstore/types/apply/feature.json
new file mode 100644
index 00000000..6e149972
--- /dev/null
+++ b/pkg/oomstore/types/apply/feature.json
@@ -0,0 +1,31 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$title": "apply feature json schema",
+    "description": "apply feature json schema",
+    "type": "object",
+    "properties": {
+        "kind": {
+            "description": "must be Feature",
+            "enum": ["Feature"]
+        },
+        "name": {
+            "description": "name of the feature",
+            "type": "string",
+            "minLength": 1
+        },
+        "group-name": {
+            "description": "the name of the group to which this feature belongs",
+            "type": "string",
+            "minLength": 1
+        },
+        "value-type": {
+            "description": "feature value type",
+            "enum": ["string","int64","float64","bool","time","bytes"]
+        },
+        "description": {
+            "description": "the feature description",
+            "type": "string"
+        }
+    },
+    "required": ["kind", "name", "group-name", "value-type"]
+}
diff --git a/pkg/oomstore/types/apply/group.json b/pkg/oomstore/types/apply/group.json
new file mode 100644
index 00000000..1d832a0e
--- /dev/null
+++ b/pkg/oomstore/types/apply/group.json
@@ -0,0 +1,36 @@
+{
+    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$title": "apply group json schema",
+    "description": "apply group json schema",
+    "type": "object",
+    "properties": {
+        "kind": {
+            "description": "must be group",
+            "enum": ["Group"]
+        },
+        "name": {
+            "description": "name of the group",
+            "type": "string",
+            "minLength": 1
+        },
+        "category": {
+            "description": "group category, which can only be batch or stream",
+            "enum": ["batch", "stream"]
+        },
+        "entity-name": {
+            "description": "the name of the entity to which this group belongs",
+            "type": "string",
+            "minLength": 1
+        },
+        "snapshot-interval": {
+            "description": "stream group snapshot interval",
+            "type": "integer",
+            "format": "duration"
+        },
+        "description": {
+            "description": "the group description",
+            "type": "string"
+        }
+    },
+    "required": ["kind", "name", "category", "entity-name"]
+}
diff --git a/pkg/oomstore/types/apply/validate.go b/pkg/oomstore/types/apply/validate.go
new file mode 100644
index 00000000..629713ed
--- /dev/null
+++ b/pkg/oomstore/types/apply/validate.go
@@ -0,0 +1,103 @@
+package apply
+
+import (
+	"context"
+	_ "embed"
+	"encoding/json"
+	"fmt"
+
+	"github.com/qri-io/jsonschema"
+
+	"github.com/oom-ai/oomstore/pkg/errdefs"
+)
+
+var (
+	entityValidate  jsonschema.Schema
+	groupValidate   jsonschema.Schema
+	featureValidate jsonschema.Schema
+
+	//go:embed entity.json
+	entitySchema string
+	//go:embed group.json
+	groupSchema string
+	//go:embed feature.json
+	featureSchema string
+)
+
+func init() {
+	if err := json.Unmarshal([]byte(entitySchema), &entityValidate); err != nil {
+		panic(err)
+	}
+
+	if err := json.Unmarshal([]byte(groupSchema), &groupValidate); err != nil {
+		panic(err)
+	}
+
+	if err := json.Unmarshal([]byte(featureSchema), &featureValidate); err != nil {
+		panic(err)
+	}
+}
+
+func (e *Entity) Validate() error {
+	data, err := json.Marshal(e)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+
+	errs, err := entityValidate.ValidateBytes(context.Background(), data)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+
+	if len(errs) > 0 {
+		err = fmt.Errorf("Entity Validate: %v", errs[0])
+		for i := 1; i < len(errs); i++ {
+			err = fmt.Errorf("%v\n%v", err, errs[i])
+		}
+		return errdefs.WithStack(err)
+	}
+	return nil
+
+}
+
+func (g *Group) Validate() error {
+	data, err := json.Marshal(g)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+
+	errs, err := groupValidate.ValidateBytes(context.Background(), data)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+
+	if len(errs) > 0 {
+		err = fmt.Errorf("Group Validate: %v", errs[0])
+		for i := 1; i < len(errs); i++ {
+			err = fmt.Errorf("%v\n%v", err, errs[i])
+		}
+		return errdefs.WithStack(err)
+	}
+	return nil
+}
+
+func (f *Feature) Validate() error {
+	data, err := json.Marshal(f)
+	if err != nil {
+		return err
+	}
+
+	errs, err := featureValidate.ValidateBytes(context.Background(), data)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+
+	if len(errs) > 0 {
+		err = fmt.Errorf("Feature Validate: %v", errs[0])
+		for i := 1; i < len(errs); i++ {
+			err = fmt.Errorf("%v\n%v", err, errs[i])
+		}
+		return errdefs.WithStack(err)
+	}
+	return nil
+}


Test patch:
diff --git a/internal/database/metadata/test_impl/revision.go b/internal/database/metadata/test_impl/revision.go
index 1bce4f80..1eeb6bf7 100644
--- a/internal/database/metadata/test_impl/revision.go
+++ b/internal/database/metadata/test_impl/revision.go
@@ -62,7 +62,7 @@ func TestCreateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 			expectedRevision: &types.Revision{
 				ID:            2,
 				Revision:      2000,
-				SnapshotTable: "",
+				SnapshotTable: "offline_batch_1_2",
 				Anchored:      false,
 				Description:   "description",
 				GroupID:       groupID,
diff --git a/oomagent/test/test_channel_export.sh b/oomagent/test/test_channel_export.sh
index b95a5266..18076a2d 100755
--- a/oomagent/test/test_channel_export.sh
+++ b/oomagent/test/test_channel_export.sh
@@ -6,7 +6,7 @@ prepare_store
 prepare_oomagent
 
 import_sample account "./data/account_10.csv"
-oomcli push --entity-key 1 --group user_fake_stream --feature f1=10
+oomcli push --entity-key 1 --group user_fake_stream --features f1=10
 oomcli snapshot user_fake_stream
 
 unix_milli=${3:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
diff --git a/oomagent/test/test_channel_join.sh b/oomagent/test/test_channel_join.sh
index 61752951..bf496ec3 100755
--- a/oomagent/test/test_channel_join.sh
+++ b/oomagent/test/test_channel_join.sh
@@ -11,7 +11,7 @@ for i in {1..5}; do
     import_sample driver_stats "./data/driver_stats_revision_$i.csv" "$i"
 done
 
-oomcli push --entity-key 1 --group fake_stream --feature f1=10
+oomcli push --entity-key 1 --group fake_stream --features f1=10
 t1=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
 oomcli snapshot fake_stream
 
diff --git a/oomcli/test/test_export.sh b/oomcli/test/test_export.sh
index 1d17683b..e540126e 100755
--- a/oomcli/test/test_export.sh
+++ b/oomcli/test/test_export.sh
@@ -7,16 +7,16 @@ init_store
 register_features
 
 t0=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=1,2,3,4,5 --feature number_of_user_starred_posts=10
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
 t1=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
 
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=2,3,4,5,6 --feature number_of_user_starred_posts=11
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11
 t2=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
 
 import_student_sample
 t3=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
 
-oomcli push --entity-key 2 --group user-click --feature last_5_click_posts=1,2,3,4,5 --feature number_of_user_starred_posts=10
+oomcli push --entity-key 2 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
 t4=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
 
 oomcli_export_push_feature() {
diff --git a/oomcli/test/test_get_meta_revision.sh b/oomcli/test/test_get_meta_revision.sh
index 797d4097..5944aa73 100755
--- a/oomcli/test/test_get_meta_revision.sh
+++ b/oomcli/test/test_get_meta_revision.sh
@@ -9,8 +9,8 @@ import_device_sample > /dev/null
 case='oomcli get meta revision works'
 expected='
 ID,REVISION,GROUP,SNAPSHOT-TABLE,CDC-TABLE,DESCRIPTION,ANCHORED,CREATE-TIME,MODIFY-TIME
-1,0,phone,offline_batch_snapshot_1_1,,dummy revision will be used at Join and Export
-2,1638519905556,phone,offline_batch_snapshot_1_2,,test data,true,2021-12-03T08:25:05Z,2021-12-03T08:25:05Z
+1,0,phone,offline_stream_snapshot_1_0,,dummy revision will be used at Join and Export
+2,1638519905556,phone,offline_batch_1_2,,test data,true,2021-12-03T08:25:05Z,2021-12-03T08:25:05Z
 '
 actual=$(oomcli get meta revision --group phone -o csv --wide)
 filter() { cut -d ',' -f 1,3-6 <<<"$1"; }
diff --git a/oomcli/test/test_import_table_link.sh b/oomcli/test/test_import_table_link.sh
index 101e4c06..d94ae161 100755
--- a/oomcli/test/test_import_table_link.sh
+++ b/oomcli/test/test_import_table_link.sh
@@ -9,13 +9,13 @@ import_device_sample
 case='oomcli import using table link'
 oomcli import \
     --group phone \
-    --table-link "offline_batch_snapshot_1_2" \
+    --table-link "offline_batch_1_2" \
     --description 'linked table' > /dev/null
 expected='
 ID,REVISION,GROUP,SNAPSHOT-TABLE,CDC-TABLE,DESCRIPTION
-1,0,phone,offline_batch_snapshot_1_1,
-2,1639047117470,phone,offline_batch_snapshot_1_2,,test data
-3,1639047117552,phone,offline_batch_snapshot_1_2,,linked table
+1,0,phone,offline_stream_snapshot_1_0,
+2,1639047117470,phone,offline_batch_1_2,,test data
+3,1639047117552,phone,offline_batch_1_2,,linked table
 '
 filter() { cut -d, -f 1,3,4,5; }
 actual=$(oomcli get meta revision -o csv)
diff --git a/oomcli/test/test_join.sh b/oomcli/test/test_join.sh
index 602d330b..9df1ea2d 100755
--- a/oomcli/test/test_join.sh
+++ b/oomcli/test/test_join.sh
@@ -11,7 +11,7 @@ trap 'command rm -rf entity_rows.csv entity_rows_with_values.csv' EXIT INT TERM
 # push stream feature
 t0=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=0,1,2,3,4 --feature number_of_user_starred_posts=10 >> /dev/null 2>&1
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=0,1,2,3,4 --features number_of_user_starred_posts=10 >> /dev/null 2>&1
 t1=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
 
@@ -20,15 +20,15 @@ import_student_sample
 t2=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
 
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=1,2,3,4,5 --feature number_of_user_starred_posts=10 >> /dev/null 2>&1
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10 >> /dev/null 2>&1
 t3=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
 
-oomcli push --entity-key 2 --group user-click --feature last_5_click_posts=2,3,4,5,6 --feature number_of_user_starred_posts=11 >> /dev/null 2>&1
+oomcli push --entity-key 2 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11 >> /dev/null 2>&1
 t4=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
 
-oomcli push --entity-key 3 --group user-click --feature last_5_click_posts=3,4,5,6,7 --feature number_of_user_starred_posts=12 >> /dev/null 2>&1
+oomcli push --entity-key 3 --group user-click --features last_5_click_posts=3,4,5,6,7 --features number_of_user_starred_posts=12 >> /dev/null 2>&1
 t5=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 
 oomcli snapshot user-click
diff --git a/oomcli/test/test_push.sh b/oomcli/test/test_push.sh
index 6e486f70..c3edad31 100755
--- a/oomcli/test/test_push.sh
+++ b/oomcli/test/test_push.sh
@@ -10,7 +10,7 @@ expected='
 user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
 1,"1,2,3,4,5",10
 '
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=1,2,3,4,5 --feature number_of_user_starred_posts=10
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10
 
 actual=$(oomcli get online --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts -k 1 -o csv)
 assert_eq "$case" "$expected" "$actual"
@@ -20,6 +20,6 @@ expected='
 user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
 1,"2,3,4,5,6",11
 '
-oomcli push --entity-key 1 --group user-click --feature last_5_click_posts=2,3,4,5,6 --feature number_of_user_starred_posts=11
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11
 actual=$(oomcli get online --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts -k 1 -o csv)
 assert_eq "$case" "$expected" "$actual"