Repository: oom-ai/oomstore
Pull request #1113 (2022-01-27 10:55:04)

oomstore(json-schema): add snapshot dependence and unit test


Patch:
diff --git a/oomcli/cmd/types.go b/oomcli/cmd/types.go
index ebb30278..11d64446 100644
--- a/oomcli/cmd/types.go
+++ b/oomcli/cmd/types.go
@@ -86,7 +86,7 @@ func parseTokenLists(i interface{}) (headerTokens TokenList, dataTokens []TokenL
 				Category:         e.Group.Category,
 				ValueType:        e.ValueType,
 				Description:      e.Description,
-				OnlineRevisionID: e.Group.OnlineRevisionID,
+				OnlineRevisionID: e.OnlineRevisionID(),
 				CreateTime:       e.CreateTime,
 				ModifyTime:       e.ModifyTime,
 			})
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index dd4e957d..5371df60 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -1,8 +1,10 @@
 package oomstore
 
 import (
+	"bufio"
 	"context"
 	"os"
+	"time"
 
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/spf13/cast"
@@ -13,7 +15,8 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-// Import data into the offline feature store
+// Import data into the offline feature store as a new revision.
+// In the future we want to support more diverse data sources.
 func (s *OomStore) Import(ctx context.Context, opt types.ImportOpt) (int, error) {
 	importOpt, err := s.parseImportOpt(ctx, opt)
 	if err != nil {
@@ -27,66 +30,101 @@ func (s *OomStore) Import(ctx context.Context, opt types.ImportOpt) (int, error)
 			return 0, err
 		}
 		defer file.Close()
-		source := &types.CsvReaderDataSource{
+		return s.csvReaderImport(ctx, importOpt, &types.CsvReaderDataSource{
 			Reader:    file,
 			Delimiter: src.Delimiter,
-		}
-		if importOpt.group.Category == types.CategoryStream {
-			return 0, s.csvReaderImportStream(ctx, importOpt, source)
-		} else {
-			return s.csvReaderImportBatch(ctx, importOpt, source)
-		}
+		})
 	case types.CSV_READER:
-		if importOpt.group.Category == types.CategoryStream {
-			return 0, s.csvReaderImportStream(ctx, importOpt, opt.CsvReaderDataSource)
-		} else {
-			return s.csvReaderImportBatch(ctx, importOpt, opt.CsvReaderDataSource)
-		}
+		return s.csvReaderImport(ctx, importOpt, opt.CsvReaderDataSource)
 	case types.TABLE_LINK:
-		if importOpt.group.Category == types.CategoryStream {
-			return 0, s.tableLinkImportStream(ctx, importOpt, opt.TableLinkDataSource)
-		} else {
-			return s.tableLinkImportBatch(ctx, importOpt, opt.TableLinkDataSource)
-		}
+		return s.tableLinkImport(ctx, importOpt, opt.TableLinkDataSource)
 	default:
 		return 0, errdefs.Errorf("unsupported data source: %v", opt.DataSourceType)
 	}
 }
 
-type importOpt struct {
-	*types.ImportOpt
-	entityName string
-	group      *types.Group
-	features   types.FeatureList
-}
-
-func (s *OomStore) parseImportOpt(ctx context.Context, opt types.ImportOpt) (*importOpt, error) {
-	group, err := s.metadata.GetGroupByName(ctx, opt.GroupName)
+func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSource *types.CsvReaderDataSource) (int, error) {
+	//make sure csv data source has all defined columns
+	reader := bufio.NewReader(dataSource.Reader)
+	source := &offline.CSVSource{
+		Reader:    reader,
+		Delimiter: dataSource.Delimiter,
+	}
+	// read header
+	columnNames := append([]string{opt.entityName}, opt.features.Names()...)
+	header, err := readHeader(source, columnNames)
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
 
-	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupID: &group.ID,
+	newRevisionID, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+		Revision:      time.Now().UnixMilli(),
+		GroupID:       opt.group.ID,
+		SnapshotTable: nil,
+		Description:   opt.Description,
+		Anchored:      opt.Revision != nil,
 	})
 	if err != nil {
-		return nil, err
+		return 0, err
 	}
-	if features == nil {
-		err = errdefs.Errorf("no features under group: %s", opt.GroupName)
-		return nil, err
+	snapshotTableName := dbutil.OfflineBatchSnapshotTableName(opt.group.ID, int64(newRevisionID))
+	revision, err := s.offline.Import(ctx, offline.ImportOpt{
+		EntityName:        opt.entityName,
+		Features:          opt.features,
+		Header:            cast.ToStringSlice(header),
+		Revision:          opt.Revision,
+		SnapshotTableName: snapshotTableName,
+		Source:            source,
+	})
+	if err != nil {
+		return 0, err
 	}
 
-	entity := group.Entity
-	if entity == nil {
-		return nil, errdefs.Errorf("no entity found by group: %s", opt.GroupName)
+	if opt.Revision != nil {
+		revision = *opt.Revision
+	}
+	if err = s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
+		RevisionID:       newRevisionID,
+		NewRevision:      &revision,
+		NewSnapshotTable: &snapshotTableName,
+	}); err != nil {
+		return 0, err
 	}
-	return &importOpt{
-		ImportOpt:  &opt,
-		entityName: entity.Name,
-		group:      group,
-		features:   features,
-	}, nil
+
+	// TODO: clean up revision and data_table if import failed
+
+	return newRevisionID, nil
+}
+
+func (s *OomStore) tableLinkImport(ctx context.Context, opt *importOpt, dataSource *types.TableLinkDataSource) (int, error) {
+	// Make sure all features existing with correct value type
+	tableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
+		TableName: dataSource.TableName,
+	})
+	if err != nil {
+		return 0, err
+	}
+	if err = validateTableSchema(tableSchema, opt.features); err != nil {
+		return 0, err
+	}
+	var revision int64
+	if opt.Revision == nil {
+		revision = time.Now().UnixMilli()
+	} else {
+		revision = *opt.Revision
+	}
+	newRevisionID, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+		Revision:      revision,
+		GroupID:       opt.group.ID,
+		SnapshotTable: &dataSource.TableName,
+		Description:   opt.Description,
+		Anchored:      opt.Revision != nil,
+	})
+	if err != nil {
+		return 0, err
+	}
+
+	return newRevisionID, nil
 }
 
 func validateTableSchema(schema *types.DataTableSchema, features types.FeatureList) error {
@@ -108,7 +146,6 @@ func validateTableSchema(schema *types.DataTableSchema, features types.FeatureLi
 	}
 	return nil
 }
-
 func hasDup(a []string) bool {
 	s := make(map[string]bool)
 	for _, e := range a {
@@ -156,3 +193,47 @@ func stringSliceEqual(a, b []string) bool {
 	}
 	return true
 }
+
+func (s *OomStore) parseImportOpt(ctx context.Context, opt types.ImportOpt) (*importOpt, error) {
+	entity, group, features, err := s.getGroupInfo(ctx, opt.GroupName)
+	if err != nil {
+		return nil, err
+	}
+	return &importOpt{
+		ImportOpt:  &opt,
+		entityName: entity.Name,
+		group:      group,
+		features:   features,
+	}, nil
+}
+
+func (s *OomStore) getGroupInfo(ctx context.Context, groupName string) (*types.Entity, *types.Group, types.FeatureList, error) {
+	group, err := s.metadata.GetGroupByName(ctx, groupName)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
+		GroupID: &group.ID,
+	})
+	if err != nil {
+		return nil, nil, nil, err
+	}
+	if features == nil {
+		err = errdefs.Errorf("no features under group: %s", groupName)
+		return nil, nil, nil, err
+	}
+
+	entity := group.Entity
+	if entity == nil {
+		return nil, nil, nil, errdefs.Errorf("no entity found by group: %s", groupName)
+	}
+	return entity, group, features, nil
+}
+
+type importOpt struct {
+	*types.ImportOpt
+	entityName string
+	group      *types.Group
+	features   types.FeatureList
+}
diff --git a/pkg/oomstore/import_batch.go b/pkg/oomstore/import_batch.go
deleted file mode 100644
index 654531a6..00000000
--- a/pkg/oomstore/import_batch.go
+++ /dev/null
@@ -1,97 +0,0 @@
-package oomstore
-
-import (
-	"bufio"
-	"context"
-	"time"
-
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-	"github.com/oom-ai/oomstore/internal/database/metadata"
-	"github.com/oom-ai/oomstore/internal/database/offline"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-	"github.com/spf13/cast"
-)
-
-func (s *OomStore) csvReaderImportBatch(ctx context.Context, opt *importOpt, dataSource *types.CsvReaderDataSource) (int, error) {
-	//make sure csv data source has all defined columns
-	reader := bufio.NewReader(dataSource.Reader)
-	source := &offline.CSVSource{
-		Reader:    reader,
-		Delimiter: dataSource.Delimiter,
-	}
-	// read header
-	columnNames := append([]string{opt.entityName}, opt.features.Names()...)
-	header, err := readHeader(source, columnNames)
-	if err != nil {
-		return 0, err
-	}
-
-	newRevisionID, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:      time.Now().UnixMilli(),
-		GroupID:       opt.group.ID,
-		SnapshotTable: nil,
-		Description:   opt.Description,
-		Anchored:      opt.Revision != nil,
-	})
-	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,
-		Header:            cast.ToStringSlice(header),
-		Revision:          opt.Revision,
-		SnapshotTableName: snapshotTableName,
-		Source:            source,
-	})
-	if err != nil {
-		return 0, err
-	}
-
-	if opt.Revision != nil {
-		revision = *opt.Revision
-	}
-	if err = s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
-		RevisionID:       newRevisionID,
-		NewRevision:      &revision,
-		NewSnapshotTable: &snapshotTableName,
-	}); err != nil {
-		return 0, err
-	}
-
-	// TODO: clean up revision and data_table if import failed
-
-	return newRevisionID, nil
-}
-
-func (s *OomStore) tableLinkImportBatch(ctx context.Context, opt *importOpt, dataSource *types.TableLinkDataSource) (int, error) {
-	// Make sure all features existing with correct value type
-	tableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
-		TableName: dataSource.TableName,
-	})
-	if err != nil {
-		return 0, err
-	}
-	if err = validateTableSchema(tableSchema, opt.features); err != nil {
-		return 0, err
-	}
-	var revision int64
-	if opt.Revision == nil {
-		revision = time.Now().UnixMilli()
-	} else {
-		revision = *opt.Revision
-	}
-	newRevisionID, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:      revision,
-		GroupID:       opt.group.ID,
-		SnapshotTable: &dataSource.TableName,
-		Description:   opt.Description,
-		Anchored:      opt.Revision != nil,
-	})
-	if err != nil {
-		return 0, err
-	}
-
-	return newRevisionID, nil
-}
diff --git a/pkg/oomstore/import_stream.go b/pkg/oomstore/import_stream.go
index 8ad745e8..7e70d251 100644
--- a/pkg/oomstore/import_stream.go
+++ b/pkg/oomstore/import_stream.go
@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"io"
+	"os"
 
 	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 
@@ -21,14 +22,41 @@ const (
 	ImportStreamBatchSize = 100
 )
 
-func (s *OomStore) csvReaderImportStream(ctx context.Context, opt *importOpt, dataSource *types.CsvReaderDataSource) error {
+func (s *OomStore) ImportStream(ctx context.Context, opt types.ImportStreamOpt) error {
+	switch opt.DataSourceType {
+	case types.CSV_FILE:
+		src := opt.CsvFileDataSource
+		file, err := os.Open(src.InputFilePath)
+		if err != nil {
+			return err
+		}
+		defer file.Close()
+		return s.csvReaderImportStream(ctx, opt, &types.CsvReaderDataSource{
+			Reader:    file,
+			Delimiter: src.Delimiter,
+		})
+	case types.CSV_READER:
+		return s.csvReaderImportStream(ctx, opt, opt.CsvReaderDataSource)
+	case types.TABLE_LINK:
+		return s.tableLinkImportStream(ctx, opt, opt.TableLinkDataSource)
+	default:
+		return errdefs.Errorf("unsupported data source: %v", opt.DataSourceType)
+	}
+}
+
+func (s *OomStore) csvReaderImportStream(ctx context.Context, opt types.ImportStreamOpt, dataSource *types.CsvReaderDataSource) error {
+	// get group information
+	entity, group, features, err := s.getGroupInfo(ctx, opt.GroupName)
+	if err != nil {
+		return err
+	}
 	// read header
 	reader := bufio.NewReader(dataSource.Reader)
 	source := &offline.CSVSource{
 		Reader:    reader,
 		Delimiter: dataSource.Delimiter,
 	}
-	columnNames := append([]string{opt.entityName, "unix_milli"}, opt.features.Names()...)
+	columnNames := append([]string{entity.Name, "unix_milli"}, features.Names()...)
 	header, err := readHeader(source, columnNames)
 	if err != nil {
 		return err
@@ -39,9 +67,9 @@ func (s *OomStore) csvReaderImportStream(ctx context.Context, opt *importOpt, da
 	for {
 		line, err := dbutil.ReadLine(dbutil.ReadLineOpt{
 			Source:     source,
-			EntityName: opt.entityName,
+			EntityName: entity.Name,
 			Header:     header,
-			Features:   opt.features,
+			Features:   features,
 		})
 		if errdefs.Cause(err) == io.EOF {
 			break
@@ -52,18 +80,23 @@ func (s *OomStore) csvReaderImportStream(ctx context.Context, opt *importOpt, da
 		if len(line) != len(header) {
 			continue
 		}
-		records = append(records, generateStreamRecord(line, header, opt.group, opt.features))
+		records = append(records, generateStreamRecord(line, header, group, features))
 		if len(records) == ImportStreamBatchSize {
-			if err := s.pushStreamingRecords(ctx, records, opt.entityName, opt.group, opt.features); err != nil {
+			if err := s.pushStreamingRecords(ctx, records, entity.Name, group, features); err != nil {
 				return err
 			}
 			records = make([]types.StreamRecord, 0, ImportStreamBatchSize)
 		}
 	}
-	return s.pushStreamingRecords(ctx, records, opt.entityName, opt.group, opt.features)
+	return s.pushStreamingRecords(ctx, records, entity.Name, group, features)
 }
 
-func (s *OomStore) tableLinkImportStream(ctx context.Context, opt *importOpt, dataSource *types.TableLinkDataSource) error {
+func (s *OomStore) tableLinkImportStream(ctx context.Context, opt types.ImportStreamOpt, dataSource *types.TableLinkDataSource) error {
+	// get group information
+	_, group, features, err := s.getGroupInfo(ctx, opt.GroupName)
+	if err != nil {
+		return err
+	}
 	// get linked table schema
 	tableSchema, err := s.offline.TableSchema(ctx, offline.TableSchemaOpt{
 		TableName:      dataSource.TableName,
@@ -73,16 +106,16 @@ func (s *OomStore) tableLinkImportStream(ctx context.Context, opt *importOpt, da
 		return err
 	}
 	// validation
-	if err = validateTableSchema(tableSchema, opt.features); err != nil {
+	if err = validateTableSchema(tableSchema, features); err != nil {
 		return err
 	}
-	if err = s.validateRevisions(ctx, opt.group.ID, tableSchema); err != nil {
+	if err = s.validateRevisions(ctx, group.ID, tableSchema); err != nil {
 		return err
 	}
 
 	_, err = s.createRevision(ctx, metadata.CreateRevisionOpt{
 		Revision:    *tableSchema.TimeRange.MinUnixMilli,
-		GroupID:     opt.group.ID,
+		GroupID:     group.ID,
 		CdcTable:    &dataSource.TableName,
 		Description: opt.Description,
 		Anchored:    true,
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 997aadc9..94f03817 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -46,7 +46,7 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 			EntityKey:  opt.EntityKey,
 			Group:      *featureList[0].Group,
 			Features:   featureList,
-			RevisionID: featureList[0].Group.OnlineRevisionID,
+			RevisionID: featureList[0].OnlineRevisionID(),
 		})
 		if err != nil {
 			return nil, err
@@ -112,7 +112,7 @@ func (s *OomStore) getFeatureValueMap(ctx context.Context, entityKeys []string,
 			EntityKeys: entityKeys,
 			Group:      *features[0].Group,
 			Features:   features,
-			RevisionID: features[0].Group.OnlineRevisionID,
+			RevisionID: features[0].OnlineRevisionID(),
 		})
 		if err != nil {
 			return nil, err
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index af7bce6e..bfa5e6da 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -39,7 +39,7 @@ func (s *OomStore) createRevision(ctx context.Context, opt metadata.CreateRevisi
 			if _, err = tx.CreateRevision(ctx, metadata.CreateRevisionOpt{
 				Revision:    0,
 				GroupID:     opt.GroupID,
-				Description: "dummy revision",
+				Description: "dummy revision will be used at Join and Export",
 			}); err != nil {
 				return err
 			}
diff --git a/pkg/oomstore/types/apply/apply.go b/pkg/oomstore/types/apply/apply.go
index 4e5f2dea..cb4f616f 100644
--- a/pkg/oomstore/types/apply/apply.go
+++ b/pkg/oomstore/types/apply/apply.go
@@ -66,7 +66,7 @@ type Group struct {
 	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"`
+	SnapshotInterval time.Duration  `mapstructure:"snapshot-interval" yaml:"snapshot-interval,omitempty" json:"snapshot-interval"`
 	Description      string         `mapstructure:"description" yaml:"description" json:"description"`
 	Features         []Feature      `mapstructure:"features" yaml:"features,omitempty" json:"features,omitempty"`
 }
@@ -118,7 +118,7 @@ func FromGroupList(groups types.GroupList, features types.FeatureList) *GroupIte
 }
 
 type Entity struct {
-	Kind        string `mapstructure:"kind" yaml:"kind" json:"kind"`
+	Kind        string `mapstructure:"kind" yaml:"kind" json:"kind,omitempty"`
 	Name        string `mapstructure:"name" yaml:"name" json:"name"`
 	Description string `mapstructure:"description" yaml:"description" json:"description"`
 
diff --git a/pkg/oomstore/types/apply/entity.json b/pkg/oomstore/types/apply/entity.json
index bab9a378..a4e60f3b 100644
--- a/pkg/oomstore/types/apply/entity.json
+++ b/pkg/oomstore/types/apply/entity.json
@@ -1,12 +1,13 @@
 {
-    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$schema": "http://json-schema.org/draft-07/schema#",
     "$title": "apply entity json schema",
     "description": "apply entity json schema",
     "type": "object",
     "properties": {
         "kind": {
-            "description": "must be Entity",
-            "enum": ["Entity"]
+            "description": "must equal Entity",
+            "type": "string",
+            "const": "Entity"
         },
         "name": {
             "description": "name of the entity",
diff --git a/pkg/oomstore/types/apply/feature.json b/pkg/oomstore/types/apply/feature.json
index 6e149972..12f5959a 100644
--- a/pkg/oomstore/types/apply/feature.json
+++ b/pkg/oomstore/types/apply/feature.json
@@ -1,12 +1,13 @@
 {
-    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$schema": "http://json-schema.org/draft-07/schema#",
     "$title": "apply feature json schema",
     "description": "apply feature json schema",
     "type": "object",
     "properties": {
         "kind": {
-            "description": "must be Feature",
-            "enum": ["Feature"]
+            "description": "must equal Feature",
+            "type": "string",
+            "const": "Feature"
         },
         "name": {
             "description": "name of the feature",
diff --git a/pkg/oomstore/types/apply/group.json b/pkg/oomstore/types/apply/group.json
index 1d832a0e..6f0be640 100644
--- a/pkg/oomstore/types/apply/group.json
+++ b/pkg/oomstore/types/apply/group.json
@@ -1,12 +1,13 @@
 {
-    "$schema": "https://json-schema.org/draft/2020-12/schema",
+    "$schema": "http://json-schema.org/draft-07/schema#",
     "$title": "apply group json schema",
     "description": "apply group json schema",
     "type": "object",
     "properties": {
         "kind": {
-            "description": "must be group",
-            "enum": ["Group"]
+            "description": "must equal Group",
+            "type": "string",
+            "const": "Group"
         },
         "name": {
             "description": "name of the group",
@@ -32,5 +33,14 @@
             "type": "string"
         }
     },
-    "required": ["kind", "name", "category", "entity-name"]
+    "required": ["kind", "name", "category", "entity-name"],
+    "if": {
+        "properties": { "category": { "const": "stream" } }
+    },
+    "then": {
+        "properties": { "snapshot-interval": { "exclusiveMinimum": 0 } }
+    },
+    "else": {
+        "properties": { "snapshot-interval": { "const": 0 } }
+    }
 }
diff --git a/pkg/oomstore/types/apply/validate.go b/pkg/oomstore/types/apply/validate.go
index 629713ed..9646d8f4 100644
--- a/pkg/oomstore/types/apply/validate.go
+++ b/pkg/oomstore/types/apply/validate.go
@@ -44,20 +44,7 @@ func (e *Entity) Validate() error {
 		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
-
+	return validate(fmt.Sprintf("Entity %s", e.Name), data, &entityValidate)
 }
 
 func (g *Group) Validate() error {
@@ -66,19 +53,7 @@ func (g *Group) Validate() error {
 		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
+	return validate(fmt.Sprintf("Group %s", g.Name), data, &groupValidate)
 }
 
 func (f *Feature) Validate() error {
@@ -87,13 +62,17 @@ func (f *Feature) Validate() error {
 		return err
 	}
 
-	errs, err := featureValidate.ValidateBytes(context.Background(), data)
+	return validate(fmt.Sprintf("Feature %s", f.Name), data, &featureValidate)
+}
+
+func validate(errPrefix string, data []byte, schema *jsonschema.Schema) error {
+	errs, err := schema.ValidateBytes(context.Background(), data)
 	if err != nil {
 		return errdefs.WithStack(err)
 	}
 
 	if len(errs) > 0 {
-		err = fmt.Errorf("Feature Validate: %v", errs[0])
+		err := fmt.Errorf("%s: %v", errPrefix, errs[0])
 		for i := 1; i < len(errs); i++ {
 			err = fmt.Errorf("%v\n%v", err, errs[i])
 		}
diff --git a/pkg/oomstore/types/feature.go b/pkg/oomstore/types/feature.go
index 65fcff3e..0b43d846 100644
--- a/pkg/oomstore/types/feature.go
+++ b/pkg/oomstore/types/feature.go
@@ -43,6 +43,13 @@ func (f *Feature) Entity() *Entity {
 	return f.Group.Entity
 }
 
+func (f *Feature) OnlineRevisionID() *int {
+	if f.Group.Category == CategoryStream {
+		return nil
+	}
+	return f.Group.OnlineRevisionID
+}
+
 func (f *Feature) DBFullName(backend BackendType) string {
 	if backend != BackendBigQuery {
 		return f.FullName()
diff --git a/pkg/oomstore/types/import.go b/pkg/oomstore/types/import.go
index c0f2f43b..49e721c9 100644
--- a/pkg/oomstore/types/import.go
+++ b/pkg/oomstore/types/import.go
@@ -34,3 +34,13 @@ type CsvFileDataSource struct {
 type TableLinkDataSource struct {
 	TableName string
 }
+
+type ImportStreamOpt struct {
+	GroupName   string
+	Description string
+
+	DataSourceType      DataSourceType
+	CsvFileDataSource   *CsvFileDataSource
+	CsvReaderDataSource *CsvReaderDataSource
+	TableLinkDataSource *TableLinkDataSource
+}


Test patch:
diff --git a/oomcli/test/test_get_meta_revision.sh b/oomcli/test/test_get_meta_revision.sh
index 88b3d7e4..797d4097 100755
--- a/oomcli/test/test_get_meta_revision.sh
+++ b/oomcli/test/test_get_meta_revision.sh
@@ -9,7 +9,7 @@ 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
+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
 '
 actual=$(oomcli get meta revision --group phone -o csv --wide)
diff --git a/pkg/oomstore/types/apply/validate_test.go b/pkg/oomstore/types/apply/validate_test.go
new file mode 100644
index 00000000..d99867c2
--- /dev/null
+++ b/pkg/oomstore/types/apply/validate_test.go
@@ -0,0 +1,245 @@
+package apply_test
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types/apply"
+)
+
+func TestEntityValidate(t *testing.T) {
+	cases := []struct {
+		description string
+		entity      *apply.Entity
+		wantError   error
+	}{
+		{
+			description: "invaid kind: spelling error",
+			entity: &apply.Entity{
+				Kind:        "Entit",
+				Name:        "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Entity user: /kind: "Entit" must equal "Entity"`),
+		},
+		{
+			description: "invalid kind: empty kind",
+			entity: &apply.Entity{
+				Name:        "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Entity user: /: {"Groups":null,"desc... "kind" value is required`),
+		},
+		{
+			description: "invalid kind: all lowercase",
+			entity: &apply.Entity{
+				Kind:        "entity",
+				Name:        "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Entity user: /kind: "entity" must equal "Entity"`),
+		},
+		{
+			description: "empty name",
+			entity: &apply.Entity{
+				Kind:        "Entity",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Entity : /name: "" min length of 1 characters required: `),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.description, func(t *testing.T) {
+			err := c.entity.Validate()
+			assert.Equal(t, c.wantError, err)
+		})
+	}
+}
+
+func TestGroupValidate(t *testing.T) {
+	cases := []struct {
+		description string
+		group       *apply.Group
+		wantError   error
+	}{
+		{
+			description: "invalid kind: spelling error",
+			group: &apply.Group{
+				Kind:        "Grou",
+				Name:        "student",
+				Category:    types.CategoryBatch,
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /kind: "Grou" must equal "Group"`),
+		},
+		{
+			description: "invalid kind: empty kind",
+			group: &apply.Group{
+				Name:        "student",
+				Category:    types.CategoryBatch,
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /: {"category":"batch",... "kind" value is required`),
+		},
+		{
+			description: "invalid kind: all lowercase",
+			group: &apply.Group{
+				Kind:        "group",
+				Name:        "student",
+				Category:    types.CategoryBatch,
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /kind: "group" must equal "Group"`),
+		},
+		{
+			description: "empty name",
+			group: &apply.Group{
+				Kind:        "Group",
+				Category:    types.CategoryBatch,
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group : /name: "" min length of 1 characters required: `),
+		},
+		{
+			description: "invalid category",
+			group: &apply.Group{
+				Kind:        "Group",
+				Name:        "student",
+				Category:    "batchh",
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /category: "batchh" should be one of ["batch", "stream"]`),
+		},
+		{
+			description: "empty entity-name",
+			group: &apply.Group{
+				Kind:        "Group",
+				Name:        "student",
+				Category:    types.CategoryBatch,
+				EntityName:  "",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /: {"category":"batch",... "entity-name" value is required`),
+		},
+		{
+			description: "stream group snapshot-interval must greater than 0",
+			group: &apply.Group{
+				Kind:        "Group",
+				Name:        "student",
+				Category:    types.CategoryStream,
+				EntityName:  "user",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Group student: /snapshot-interval: 0 0 must be greater than 0`),
+		},
+		{
+			description: "batch group snapshot-interval must equal 0",
+			group: &apply.Group{
+				Kind:             "Group",
+				Name:             "student",
+				Category:         types.CategoryBatch,
+				SnapshotInterval: time.Second,
+				EntityName:       "user",
+				Description:      "~",
+			},
+			wantError: fmt.Errorf(`Group student: /snapshot-interval: 1000000000 must equal 0`),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.description, func(t *testing.T) {
+			err := c.group.Validate()
+			assert.Equal(t, c.wantError, err)
+		})
+	}
+}
+
+func TestFeatureValidate(t *testing.T) {
+	cases := []struct {
+		description string
+		feature     *apply.Feature
+		wantError   error
+	}{
+		{
+			description: "invaid kind: spelling error",
+			feature: &apply.Feature{
+				Kind:        "Featur",
+				Name:        "age",
+				GroupName:   "student",
+				ValueType:   "int64",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature age: /kind: "Featur" must equal "Feature"`),
+		},
+		{
+			description: "invalid kind: empty kind ",
+			feature: &apply.Feature{
+				Name:        "age",
+				GroupName:   "student",
+				ValueType:   "int64",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature age: /: {"description":"~","... "kind" value is required`),
+		},
+		{
+			description: "invalid kind: all lowercase",
+			feature: &apply.Feature{
+				Kind:        "feature",
+				Name:        "age",
+				GroupName:   "student",
+				ValueType:   "int64",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature age: /kind: "feature" must equal "Feature"`),
+		},
+		{
+			description: "empty group-name",
+			feature: &apply.Feature{
+				Kind:        "Feature",
+				Name:        "age",
+				ValueType:   "int64",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature age: /: {"description":"~","... "group-name" value is required`),
+		},
+		{
+
+			description: "empty feature name",
+			feature: &apply.Feature{
+				Kind:        "Feature",
+				GroupName:   "student",
+				ValueType:   "int64",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature : /name: "" min length of 1 characters required: `),
+		},
+		{
+			description: "invalid value type",
+			feature: &apply.Feature{
+				Kind:        "Feature",
+				Name:        "age",
+				GroupName:   "student",
+				ValueType:   "int",
+				Description: "~",
+			},
+			wantError: fmt.Errorf(`Feature age: /value-type: "int" should be one of ["string", "int64", "float64", "bool", "time", "bytes"]`),
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.description, func(t *testing.T) {
+			err := c.feature.Validate()
+			assert.Equal(t, c.wantError, err)
+		})
+	}
+}