Repository: oom-ai/oomstore
Pull request #998 (2022-01-17 10:35:14)

Feat new field snapshotinterval
this PR adds the field `snapshot_interval` for stream group.


Patch:
diff --git a/internal/database/dbutil/data_loader.go b/internal/database/dbutil/data_loader.go
index 76912922..ddfabadf 100644
--- a/internal/database/dbutil/data_loader.go
+++ b/internal/database/dbutil/data_loader.go
@@ -1,6 +1,7 @@
 package dbutil
 
 import (
+	"bufio"
 	"context"
 	"io"
 	"strings"
@@ -12,81 +13,53 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-type LoadDataFromSourceOpt struct {
-	Source    *offline.CSVSource
-	Entity    *types.Entity
-	TableName string
-	Header    []string
-	Features  types.FeatureList
-	Backend   types.BackendType
-}
-
 // Currying
-func LoadDataFromSource(backend types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSourceOpt) error {
-	return func(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSourceOpt) error {
-		return loadDataFromSource(tx, ctx, opt, batchSize)
+func LoadDataFromSource(backendType types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
+	return func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
+		return loadDataFromSource(tx, ctx, source, tableName, header, features, backendType, batchSize)
 	}
 }
 
-func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSourceOpt, batchSize int) error {
+func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList, backendType types.BackendType, batchSize int) error {
 	records := make([]interface{}, 0, batchSize)
 	for {
-		record, err := ReadLine(ReadLineOpt{
-			Source:   opt.Source,
-			Entity:   opt.Entity,
-			Header:   opt.Header,
-			Features: opt.Features,
-			Backend:  opt.Backend,
-		})
+		record, err := ReadLine(source.Reader, source.Delimiter, features, backendType)
 		if errdefs.Cause(err) == io.EOF {
 			break
 		}
 		if err != nil {
 			return err
 		}
-		if len(record) != len(opt.Header) {
+		if len(record) != len(header) {
 			continue
 		}
 		records = append(records, record)
 		if len(records) == batchSize {
-			if err := InsertRecordsToTableTx(tx, ctx, opt.TableName, records, opt.Header, opt.Backend); err != nil {
+			if err := InsertRecordsToTableTx(tx, ctx, tableName, records, header, backendType); err != nil {
 				return err
 			}
 			records = make([]interface{}, 0, batchSize)
 		}
 	}
-	if err := InsertRecordsToTableTx(tx, ctx, opt.TableName, records, opt.Header, opt.Backend); err != nil {
+	if err := InsertRecordsToTableTx(tx, ctx, tableName, records, header, backendType); err != nil {
 		return err
 	}
 	return nil
 }
 
-type ReadLineOpt struct {
-	Source   *offline.CSVSource
-	Entity   *types.Entity
-	Header   []string
-	Features types.FeatureList
-	Backend  types.BackendType
-}
-
-func ReadLine(opt ReadLineOpt) ([]interface{}, error) {
-	row, err := opt.Source.Reader.ReadString('\n')
+func ReadLine(reader *bufio.Reader, delimiter string, features types.FeatureList, backend types.BackendType) ([]interface{}, error) {
+	row, err := reader.ReadString('\n')
 	if err != nil {
 		return nil, errdefs.WithStack(err)
 	}
-	rowSlice := strings.Split(strings.Trim(row, "\n"), opt.Source.Delimiter)
+	rowSlice := strings.Split(strings.Trim(row, "\n"), delimiter)
 	line := make([]interface{}, 0, len(rowSlice))
 	for i, ele := range rowSlice {
-		if len(opt.Header) == 0 || len(opt.Features) == 0 || opt.Header[i] == opt.Entity.Name {
+		if i == 0 || len(features) == 0 {
 			// entity_key doesn't need to change type
 			line = append(line, ele)
-		} else if opt.Header[i] == "unix_milli" {
-			line = append(line, castElement(ele, types.Int64, opt.Backend))
 		} else {
-			feature := opt.Features.Find(func(f *types.Feature) bool {
-				return f.Name == opt.Header[i]
-			})
-			line = append(line, castElement(ele, feature.ValueType, opt.Backend))
+			line = append(line, castElement(ele, features[i-1].ValueType, backend))
 		}
 	}
 	return line, nil
diff --git a/internal/database/metadata/mysql/group.go b/internal/database/metadata/mysql/group.go
index 4cc7ec24..f105bdb3 100644
--- a/internal/database/metadata/mysql/group.go
+++ b/internal/database/metadata/mysql/group.go
@@ -13,9 +13,12 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
 		return 0, errdefs.InvalidAttribute(errdefs.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
+	if opt.Category == types.CategoryStream && opt.SnapshotInterval == 0 {
+		return 0, errdefs.InvalidAttribute(errdefs.Errorf("the field SnapshotInterval of the stream group %s cannot be zero", opt.GroupName))
+	}
 
-	query := "INSERT INTO feature_group(name, entity_id, category, description) VALUES(?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, query, opt.GroupName, opt.EntityID, opt.Category, opt.Description)
+	query := "INSERT INTO feature_group(name, entity_id, category, snapshot_interval, description) VALUES(?, ?, ?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, query, opt.GroupName, opt.EntityID, opt.Category, opt.SnapshotInterval, opt.Description)
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
diff --git a/internal/database/metadata/mysql/schema.go b/internal/database/metadata/mysql/schema.go
index e2d05b5f..691465f3 100644
--- a/internal/database/metadata/mysql/schema.go
+++ b/internal/database/metadata/mysql/schema.go
@@ -19,6 +19,7 @@ var META_TABLE_SCHEMAS = map[string]string{
 			name               	VARCHAR(32) 	NOT	NULL,
 			category           	VARCHAR(16) 	NOT	NULL COMMENT "group category: batch, stream",
 			entity_id          	INT         	NOT	NULL,
+			snapshot_interval  	INT		        DEFAULT 0,
 			online_revision_id 	INT         	DEFAULT NULL,
 			description        	VARCHAR(64) 	DEFAULT '',
 			create_time        	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/internal/database/metadata/postgres/group.go b/internal/database/metadata/postgres/group.go
index c719221f..c8ef8134 100644
--- a/internal/database/metadata/postgres/group.go
+++ b/internal/database/metadata/postgres/group.go
@@ -15,9 +15,13 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
 		return 0, errdefs.InvalidAttribute(errdefs.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
+	if opt.Category == types.CategoryStream && opt.SnapshotInterval == 0 {
+		return 0, errdefs.InvalidAttribute(errdefs.Errorf("the field SnapshotInterval of the stream group %s cannot be zero", opt.GroupName))
+	}
+
 	var groupID int
-	query := "insert into feature_group(name, entity_id, category, description) values($1, $2, $3, $4) returning id"
-	err := sqlxCtx.GetContext(ctx, &groupID, query, opt.GroupName, opt.EntityID, opt.Category, opt.Description)
+	query := "insert into feature_group(name, entity_id, category, snapshot_interval, description) values($1, $2, $3, $4, $5) returning id"
+	err := sqlxCtx.GetContext(ctx, &groupID, query, opt.GroupName, opt.EntityID, opt.Category, opt.SnapshotInterval, opt.Description)
 	if err != nil {
 		if e2, ok := err.(*pq.Error); ok {
 			if e2.Code == pgerrcode.UniqueViolation {
diff --git a/internal/database/metadata/postgres/schema.go b/internal/database/metadata/postgres/schema.go
index 9e494c02..64913e85 100644
--- a/internal/database/metadata/postgres/schema.go
+++ b/internal/database/metadata/postgres/schema.go
@@ -38,6 +38,7 @@ var META_TABLE_SCHEMAS = map[string]string{
 			name               VARCHAR(32) NOT     NULL,
 			category           VARCHAR(16) NOT     NULL,
 			entity_id          INT         NOT     NULL,
+			snapshot_interval  INT         DEFAULT 0,
 			online_revision_id INT         DEFAULT NULL,
 			description        VARCHAR(64) DEFAULT '',
 			create_time        TIMESTAMP   NOT     NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/internal/database/metadata/sqlite/group.go b/internal/database/metadata/sqlite/group.go
index 3bccab85..fa3c2af1 100644
--- a/internal/database/metadata/sqlite/group.go
+++ b/internal/database/metadata/sqlite/group.go
@@ -15,9 +15,12 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
 		return 0, errdefs.InvalidAttribute(errdefs.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
+	if opt.Category == types.CategoryStream && opt.SnapshotInterval == 0 {
+		return 0, errdefs.InvalidAttribute(errdefs.Errorf("the field SnapshotInterval of the stream group %s cannot be zero", opt.GroupName))
+	}
 
-	query := "INSERT INTO feature_group(name, entity_id, category, description) VALUES(?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, query, opt.GroupName, opt.EntityID, opt.Category, opt.Description)
+	query := "INSERT INTO feature_group(name, entity_id, category, snapshot_interval, description) VALUES(?, ?, ?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, query, opt.GroupName, opt.EntityID, opt.Category, opt.SnapshotInterval, opt.Description)
 	if err != nil {
 		if sqliteErr, ok := err.(*sqlite.Error); ok {
 			if sqliteErr.Code() == sqlite3.SQLITE_CONSTRAINT_UNIQUE {
diff --git a/internal/database/metadata/sqlite/schema.go b/internal/database/metadata/sqlite/schema.go
index 4b12bcfc..5cd4498e 100644
--- a/internal/database/metadata/sqlite/schema.go
+++ b/internal/database/metadata/sqlite/schema.go
@@ -28,6 +28,7 @@ var META_TABLE_SCHEMAS = map[string]string{
 			name               	VARCHAR(32) 	NOT	NULL,
 			category           	VARCHAR(16) 	NOT	NULL,
 			entity_id          	INT         	NOT	NULL,
+			snapshot_interval  	INT         	DEFAULT 0,
 			online_revision_id 	INT         	DEFAULT NULL,
 			description        	VARCHAR(64) 	DEFAULT '',
 			create_time        	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/internal/database/metadata/sqlutil/group.go b/internal/database/metadata/sqlutil/group.go
index 2042a118..c1c03ad5 100644
--- a/internal/database/metadata/sqlutil/group.go
+++ b/internal/database/metadata/sqlutil/group.go
@@ -18,6 +18,9 @@ func UpdateGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if opt.NewDescription != nil {
 		and["description"] = *opt.NewDescription
 	}
+	if opt.NewSnapshotInterval != nil {
+		and["snapshot_interval"] = *opt.NewSnapshotInterval
+	}
 	if opt.NewOnlineRevisionID != nil {
 		and["online_revision_id"] = *opt.NewOnlineRevisionID
 	}
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index 10f52751..f4747c95 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -1,6 +1,8 @@
 package metadata
 
-import "github.com/oom-ai/oomstore/pkg/oomstore/types"
+import (
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
 
 // Create
 type CreateEntityOpt struct {
@@ -19,6 +21,8 @@ type CreateGroupOpt struct {
 	EntityID    int
 	Description string
 	Category    types.Category
+
+	SnapshotInterval int
 }
 
 type CreateRevisionOpt struct {
@@ -43,6 +47,7 @@ type UpdateFeatureOpt struct {
 
 type UpdateGroupOpt struct {
 	GroupID             int
+	NewSnapshotInterval *int
 	NewDescription      *string
 	NewOnlineRevisionID *int
 }
diff --git a/internal/database/offline/postgres/util.go b/internal/database/offline/postgres/util.go
index 3ebb6f22..67eb6f29 100644
--- a/internal/database/offline/postgres/util.go
+++ b/internal/database/offline/postgres/util.go
@@ -5,34 +5,30 @@ import (
 	"io"
 
 	"github.com/oom-ai/oomstore/pkg/errdefs"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/offline"
 )
 
-func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, opt dbutil.LoadDataFromSourceOpt) error {
-	stmt, err := tx.PreparexContext(ctx, pq.CopyIn(opt.TableName, opt.Header...))
+func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
+	stmt, err := tx.PreparexContext(ctx, pq.CopyIn(tableName, header...))
 	if err != nil {
 		return errdefs.WithStack(err)
 	}
 	defer stmt.Close()
 
 	for {
-		record, err := dbutil.ReadLine(dbutil.ReadLineOpt{
-			Source:   opt.Source,
-			Entity:   opt.Entity,
-			Header:   opt.Header,
-			Features: opt.Features,
-			Backend:  opt.Backend,
-		})
+		record, err := dbutil.ReadLine(source.Reader, source.Delimiter, features, Backend)
 		if errdefs.Cause(err) == io.EOF {
 			break
 		}
 		if err != nil {
 			return err
 		}
-		if len(record) != len(opt.Header) {
+		if len(record) != len(header) {
 			continue
 		}
 		if _, err := stmt.ExecContext(ctx, record...); err != nil {
diff --git a/internal/database/offline/sqlutil/import.go b/internal/database/offline/sqlutil/import.go
index 26a66cd4..11169088 100644
--- a/internal/database/offline/sqlutil/import.go
+++ b/internal/database/offline/sqlutil/import.go
@@ -12,9 +12,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-type LoadData func(tx *sqlx.Tx, ctx context.Context, opt dbutil.LoadDataFromSourceOpt) error
+type LoadData func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error
 
-func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData LoadData, backend types.BackendType) (int64, error) {
+func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData LoadData, backendType types.BackendType) (int64, error) {
 	var revision int64
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
@@ -22,21 +22,14 @@ func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData Lo
 		if opt.NoPK {
 			pkFields = nil
 		}
-		schema := dbutil.BuildTableSchema(opt.SnapshotTableName, opt.Entity, false, opt.Features, pkFields, backend)
+		schema := dbutil.BuildTableSchema(opt.SnapshotTableName, opt.Entity, false, opt.Features, pkFields, backendType)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
 			return errdefs.WithStack(err)
 		}
 
 		// populate the data table
-		err = loadData(tx, ctx, dbutil.LoadDataFromSourceOpt{
-			Source:    opt.Source,
-			Entity:    opt.Entity,
-			TableName: opt.SnapshotTableName,
-			Header:    opt.Header,
-			Features:  opt.Features,
-			Backend:   backend,
-		})
+		err = loadData(tx, ctx, opt.Source, opt.SnapshotTableName, opt.Header, opt.Features)
 		if err != nil {
 			return err
 		}
diff --git a/oomagent/codegen/oomagent.pb.go b/oomagent/codegen/oomagent.pb.go
index e3b1157b..9cee979f 100644
--- a/oomagent/codegen/oomagent.pb.go
+++ b/oomagent/codegen/oomagent.pb.go
@@ -651,11 +651,11 @@ type ImportRequest struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Group       string  `protobuf:"bytes,1,opt,name=group,proto3" json:"group,omitempty"`
-	Revision    *int64  `protobuf:"varint,2,opt,name=revision,proto3,oneof" json:"revision,omitempty"`
-	Description *string `protobuf:"bytes,3,opt,name=description,proto3,oneof" json:"description,omitempty"`
-	InputFile   string  `protobuf:"bytes,4,opt,name=input_file,json=inputFile,proto3" json:"input_file,omitempty"`
-	Delimiter   *string `protobuf:"bytes,5,opt,name=delimiter,proto3,oneof" json:"delimiter,omitempty"`
+	Group         string  `protobuf:"bytes,1,opt,name=group,proto3" json:"group,omitempty"`
+	Revision      *int64  `protobuf:"varint,2,opt,name=revision,proto3,oneof" json:"revision,omitempty"`
+	Description   *string `protobuf:"bytes,3,opt,name=description,proto3,oneof" json:"description,omitempty"`
+	InputFilePath string  `protobuf:"bytes,4,opt,name=input_file_path,json=inputFilePath,proto3" json:"input_file_path,omitempty"`
+	Delimiter     *string `protobuf:"bytes,5,opt,name=delimiter,proto3,oneof" json:"delimiter,omitempty"`
 }
 
 func (x *ImportRequest) Reset() {
@@ -711,9 +711,9 @@ func (x *ImportRequest) GetDescription() string {
 	return ""
 }
 
-func (x *ImportRequest) GetInputFile() string {
+func (x *ImportRequest) GetInputFilePath() string {
 	if x != nil {
-		return x.InputFile
+		return x.InputFilePath
 	}
 	return ""
 }
@@ -1108,9 +1108,9 @@ type JoinRequest struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Features   []string `protobuf:"bytes,1,rep,name=features,proto3" json:"features,omitempty"`
-	InputFile  string   `protobuf:"bytes,2,opt,name=input_file,json=inputFile,proto3" json:"input_file,omitempty"`
-	OutputFile string   `protobuf:"bytes,3,opt,name=output_file,json=outputFile,proto3" json:"output_file,omitempty"`
+	Features       []string `protobuf:"bytes,1,rep,name=features,proto3" json:"features,omitempty"`
+	InputFilePath  string   `protobuf:"bytes,2,opt,name=input_file_path,json=inputFilePath,proto3" json:"input_file_path,omitempty"`
+	OutputFilePath string   `protobuf:"bytes,3,opt,name=output_file_path,json=outputFilePath,proto3" json:"output_file_path,omitempty"`
 }
 
 func (x *JoinRequest) Reset() {
@@ -1152,16 +1152,16 @@ func (x *JoinRequest) GetFeatures() []string {
 	return nil
 }
 
-func (x *JoinRequest) GetInputFile() string {
+func (x *JoinRequest) GetInputFilePath() string {
 	if x != nil {
-		return x.InputFile
+		return x.InputFilePath
 	}
 	return ""
 }
 
-func (x *JoinRequest) GetOutputFile() string {
+func (x *JoinRequest) GetOutputFilePath() string {
 	if x != nil {
-		return x.OutputFile
+		return x.OutputFilePath
 	}
 	return ""
 }
@@ -1327,10 +1327,10 @@ type ExportRequest struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	Features   []string `protobuf:"bytes,1,rep,name=features,proto3" json:"features,omitempty"`
-	UnixMilli  int64    `protobuf:"varint,2,opt,name=unix_milli,json=unixMilli,proto3" json:"unix_milli,omitempty"`
-	Limit      *uint64  `protobuf:"varint,3,opt,name=limit,proto3,oneof" json:"limit,omitempty"`
-	OutputFile string   `protobuf:"bytes,4,opt,name=output_file,json=outputFile,proto3" json:"output_file,omitempty"`
+	Features       []string `protobuf:"bytes,1,rep,name=features,proto3" json:"features,omitempty"`
+	UnixMilli      int64    `protobuf:"varint,2,opt,name=unix_milli,json=unixMilli,proto3" json:"unix_milli,omitempty"`
+	Limit          *uint64  `protobuf:"varint,3,opt,name=limit,proto3,oneof" json:"limit,omitempty"`
+	OutputFilePath string   `protobuf:"bytes,4,opt,name=output_file_path,json=outputFilePath,proto3" json:"output_file_path,omitempty"`
 }
 
 func (x *ExportRequest) Reset() {
@@ -1386,9 +1386,9 @@ func (x *ExportRequest) GetLimit() uint64 {
 	return 0
 }
 
-func (x *ExportRequest) GetOutputFile() string {
+func (x *ExportRequest) GetOutputFilePath() string {
 	if x != nil {
-		return x.OutputFile
+		return x.OutputFilePath
 	}
 	return ""
 }
@@ -1514,142 +1514,144 @@ var file_oomagent_proto_rawDesc = []byte{
 	0x0a, 0x0e, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
 	0x12, 0x1f, 0x0a, 0x0b, 0x72, 0x65, 0x76, 0x69, 0x73, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18,
 	0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0a, 0x72, 0x65, 0x76, 0x69, 0x73, 0x69, 0x6f, 0x6e, 0x49,
-	0x64, 0x22, 0xda, 0x01, 0x0a, 0x0d, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75,
+	0x64, 0x22, 0xe3, 0x01, 0x0a, 0x0d, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75,
 	0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01,
 	0x28, 0x09, 0x52, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1f, 0x0a, 0x08, 0x72, 0x65, 0x76,
 	0x69, 0x73, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x48, 0x00, 0x52, 0x08, 0x72,
 	0x65, 0x76, 0x69, 0x73, 0x69, 0x6f, 0x6e, 0x88, 0x01, 0x01, 0x12, 0x25, 0x0a, 0x0b, 0x64, 0x65,
 	0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48,
 	0x01, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x88, 0x01,
-	0x01, 0x12, 0x1d, 0x0a, 0x0a, 0x69, 0x6e, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x69, 0x6e, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c, 0x65,
-	0x12, 0x21, 0x0a, 0x09, 0x64, 0x65, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x18, 0x05, 0x20,
-	0x01, 0x28, 0x09, 0x48, 0x02, 0x52, 0x09, 0x64, 0x65, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72,
-	0x88, 0x01, 0x01, 0x42, 0x0b, 0x0a, 0x09, 0x5f, 0x72, 0x65, 0x76, 0x69, 0x73, 0x69, 0x6f, 0x6e,
-	0x42, 0x0e, 0x0a, 0x0c, 0x5f, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e,
-	0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x64, 0x65, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x22, 0x0e,
-	0x0a, 0x0c, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x96,
-	0x01, 0x0a, 0x0b, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1d,
-	0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01,
-	0x28, 0x09, 0x52, 0x09, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x4b, 0x65, 0x79, 0x12, 0x14, 0x0a,
-	0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x67, 0x72,
-	0x6f, 0x75, 0x70, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18,
-	0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12,
-	0x36, 0x0a, 0x0e, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x5f, 0x76, 0x61, 0x6c, 0x75, 0x65,
-	0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65,
-	0x6e, 0x74, 0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x0d, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72,
-	0x65, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x22, 0x12, 0x0a, 0x10, 0x53, 0x6e, 0x61, 0x70, 0x73,
-	0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x27, 0x0a, 0x0f, 0x53,
-	0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14,
-	0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x67,
-	0x72, 0x6f, 0x75, 0x70, 0x22, 0x61, 0x0a, 0x09, 0x45, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x52, 0x6f,
-	0x77, 0x12, 0x1d, 0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x5f, 0x6b, 0x65, 0x79, 0x18,
-	0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x4b, 0x65, 0x79,
-	0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x6d, 0x69, 0x6c, 0x6c, 0x69, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x75, 0x6e, 0x69, 0x78, 0x4d, 0x69, 0x6c, 0x6c, 0x69, 0x12,
-	0x16, 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52,
-	0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x22, 0x98, 0x01, 0x0a, 0x12, 0x43, 0x68, 0x61, 0x6e,
-	0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x23,
-	0x0a, 0x0d, 0x6a, 0x6f, 0x69, 0x6e, 0x5f, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18,
-	0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0c, 0x6a, 0x6f, 0x69, 0x6e, 0x46, 0x65, 0x61, 0x74, 0x75,
-	0x72, 0x65, 0x73, 0x12, 0x29, 0x0a, 0x10, 0x65, 0x78, 0x69, 0x73, 0x74, 0x65, 0x64, 0x5f, 0x66,
-	0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x65,
-	0x78, 0x69, 0x73, 0x74, 0x65, 0x64, 0x46, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x32,
-	0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x5f, 0x72, 0x6f, 0x77, 0x18, 0x03, 0x20, 0x01,
-	0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x6e,
-	0x74, 0x69, 0x74, 0x79, 0x52, 0x6f, 0x77, 0x52, 0x09, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x52,
-	0x6f, 0x77, 0x22, 0x5d, 0x0a, 0x13, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69,
-	0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x68, 0x65, 0x61,
-	0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65,
-	0x72, 0x12, 0x2e, 0x0a, 0x0a, 0x6a, 0x6f, 0x69, 0x6e, 0x65, 0x64, 0x5f, 0x72, 0x6f, 0x77, 0x18,
-	0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74,
-	0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x09, 0x6a, 0x6f, 0x69, 0x6e, 0x65, 0x64, 0x52, 0x6f,
-	0x77, 0x22, 0x69, 0x0a, 0x0b, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03,
-	0x28, 0x09, 0x52, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x0a,
-	0x69, 0x6e, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x09, 0x69, 0x6e, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x6f,
-	0x75, 0x74, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09,
-	0x52, 0x0a, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x22, 0x0e, 0x0a, 0x0c,
-	0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x76, 0x0a, 0x14,
-	0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73,
-	0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73,
-	0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x6d, 0x69, 0x6c, 0x6c, 0x69, 0x18, 0x02,
-	0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x75, 0x6e, 0x69, 0x78, 0x4d, 0x69, 0x6c, 0x6c, 0x69, 0x12,
-	0x19, 0x0a, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x48, 0x00,
-	0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x88, 0x01, 0x01, 0x42, 0x08, 0x0a, 0x06, 0x5f, 0x6c,
-	0x69, 0x6d, 0x69, 0x74, 0x22, 0x52, 0x0a, 0x15, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45,
-	0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a,
-	0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x68,
-	0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x21, 0x0a, 0x03, 0x72, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x03,
-	0x28, 0x0b, 0x32, 0x0f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x56, 0x61,
-	0x6c, 0x75, 0x65, 0x52, 0x03, 0x72, 0x6f, 0x77, 0x22, 0x90, 0x01, 0x0a, 0x0d, 0x45, 0x78, 0x70,
-	0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65,
-	0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x66, 0x65,
-	0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x6d,
-	0x69, 0x6c, 0x6c, 0x69, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x75, 0x6e, 0x69, 0x78,
-	0x4d, 0x69, 0x6c, 0x6c, 0x69, 0x12, 0x19, 0x0a, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x18, 0x03,
-	0x20, 0x01, 0x28, 0x04, 0x48, 0x00, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x88, 0x01, 0x01,
-	0x12, 0x1f, 0x0a, 0x0b, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x18,
-	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c,
-	0x65, 0x42, 0x08, 0x0a, 0x06, 0x5f, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22, 0x10, 0x0a, 0x0e, 0x45,
-	0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x32, 0xcf, 0x06,
-	0x0a, 0x08, 0x4f, 0x6f, 0x6d, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x12, 0x46, 0x0a, 0x09, 0x4f, 0x6e,
-	0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74, 0x12, 0x1a, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65,
-	0x6e, 0x74, 0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4f,
-	0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x55, 0x0a, 0x0e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x4d, 0x75, 0x6c, 0x74,
-	0x69, 0x47, 0x65, 0x74, 0x12, 0x1f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e,
-	0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x47, 0x65, 0x74, 0x52, 0x65,
-	0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74,
-	0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x47, 0x65, 0x74, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x04, 0x53, 0x79, 0x6e,
-	0x63, 0x12, 0x15, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x53, 0x79, 0x6e,
-	0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67,
-	0x65, 0x6e, 0x74, 0x2e, 0x53, 0x79, 0x6e, 0x63, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
-	0x22, 0x00, 0x12, 0x4d, 0x0a, 0x0d, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x49, 0x6d, 0x70,
-	0x6f, 0x72, 0x74, 0x12, 0x1e, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43,
-	0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75,
-	0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x49,
-	0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28,
-	0x01, 0x12, 0x3d, 0x0a, 0x06, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x2e, 0x6f, 0x6f,
-	0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71,
-	0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e,
-	0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00,
-	0x12, 0x37, 0x0a, 0x04, 0x50, 0x75, 0x73, 0x68, 0x12, 0x15, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67,
-	0x65, 0x6e, 0x74, 0x2e, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x16, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x50, 0x75, 0x73, 0x68, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x43, 0x0a, 0x08, 0x53, 0x6e, 0x61,
-	0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x19, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74,
-	0x2e, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74,
-	0x1a, 0x1a, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x53, 0x6e, 0x61, 0x70,
-	0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x50,
-	0x0a, 0x0b, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e, 0x12, 0x1c, 0x2e,
-	0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c,
-	0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1d, 0x2e, 0x6f, 0x6f,
-	0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f,
-	0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x28, 0x01, 0x30, 0x01,
-	0x12, 0x37, 0x0a, 0x04, 0x4a, 0x6f, 0x69, 0x6e, 0x12, 0x15, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67,
-	0x65, 0x6e, 0x74, 0x2e, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
-	0x16, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4a, 0x6f, 0x69, 0x6e, 0x52,
-	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x54, 0x0a, 0x0d, 0x43, 0x68, 0x61,
-	0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x1e, 0x2e, 0x6f, 0x6f, 0x6d,
-	0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70,
-	0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f, 0x2e, 0x6f, 0x6f, 0x6d,
-	0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70,
-	0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x30, 0x01, 0x12,
-	0x3d, 0x0a, 0x06, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x2e, 0x6f, 0x6f, 0x6d, 0x61,
-	0x67, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65,
-	0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x78,
-	0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x3f,
-	0x0a, 0x0b, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63, 0x6b, 0x12, 0x16, 0x2e,
-	0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e,
-	0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70,
-	0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x22, 0x00, 0x42,
-	0x0a, 0x5a, 0x08, 0x2f, 0x63, 0x6f, 0x64, 0x65, 0x67, 0x65, 0x6e, 0x62, 0x06, 0x70, 0x72, 0x6f,
-	0x74, 0x6f, 0x33,
+	0x01, 0x12, 0x26, 0x0a, 0x0f, 0x69, 0x6e, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x5f,
+	0x70, 0x61, 0x74, 0x68, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x69, 0x6e, 0x70, 0x75,
+	0x74, 0x46, 0x69, 0x6c, 0x65, 0x50, 0x61, 0x74, 0x68, 0x12, 0x21, 0x0a, 0x09, 0x64, 0x65, 0x6c,
+	0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x48, 0x02, 0x52, 0x09,
+	0x64, 0x65, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x88, 0x01, 0x01, 0x42, 0x0b, 0x0a, 0x09,
+	0x5f, 0x72, 0x65, 0x76, 0x69, 0x73, 0x69, 0x6f, 0x6e, 0x42, 0x0e, 0x0a, 0x0c, 0x5f, 0x64, 0x65,
+	0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x42, 0x0c, 0x0a, 0x0a, 0x5f, 0x64, 0x65,
+	0x6c, 0x69, 0x6d, 0x69, 0x74, 0x65, 0x72, 0x22, 0x0e, 0x0a, 0x0c, 0x50, 0x75, 0x73, 0x68, 0x52,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x96, 0x01, 0x0a, 0x0b, 0x50, 0x75, 0x73, 0x68,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1d, 0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x69, 0x74,
+	0x79, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x65, 0x6e, 0x74,
+	0x69, 0x74, 0x79, 0x4b, 0x65, 0x79, 0x12, 0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1a, 0x0a, 0x08,
+	0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08,
+	0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x36, 0x0a, 0x0e, 0x66, 0x65, 0x61, 0x74,
+	0x75, 0x72, 0x65, 0x5f, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b,
+	0x32, 0x0f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x56, 0x61, 0x6c, 0x75,
+	0x65, 0x52, 0x0d, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x73,
+	0x22, 0x12, 0x0a, 0x10, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x27, 0x0a, 0x0f, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x14, 0x0a, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70,
+	0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x05, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x22, 0x61, 0x0a,
+	0x09, 0x45, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x52, 0x6f, 0x77, 0x12, 0x1d, 0x0a, 0x0a, 0x65, 0x6e,
+	0x74, 0x69, 0x74, 0x79, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09,
+	0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x4b, 0x65, 0x79, 0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x6e, 0x69,
+	0x78, 0x5f, 0x6d, 0x69, 0x6c, 0x6c, 0x69, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x75,
+	0x6e, 0x69, 0x78, 0x4d, 0x69, 0x6c, 0x6c, 0x69, 0x12, 0x16, 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75,
+	0x65, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73,
+	0x22, 0x98, 0x01, 0x0a, 0x12, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x23, 0x0a, 0x0d, 0x6a, 0x6f, 0x69, 0x6e, 0x5f,
+	0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0c,
+	0x6a, 0x6f, 0x69, 0x6e, 0x46, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x29, 0x0a, 0x10,
+	0x65, 0x78, 0x69, 0x73, 0x74, 0x65, 0x64, 0x5f, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73,
+	0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x65, 0x78, 0x69, 0x73, 0x74, 0x65, 0x64, 0x46,
+	0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x32, 0x0a, 0x0a, 0x65, 0x6e, 0x74, 0x69, 0x74,
+	0x79, 0x5f, 0x72, 0x6f, 0x77, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6f, 0x6f,
+	0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x52, 0x6f, 0x77,
+	0x52, 0x09, 0x65, 0x6e, 0x74, 0x69, 0x74, 0x79, 0x52, 0x6f, 0x77, 0x22, 0x5d, 0x0a, 0x13, 0x43,
+	0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x03,
+	0x28, 0x09, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x2e, 0x0a, 0x0a, 0x6a, 0x6f,
+	0x69, 0x6e, 0x65, 0x64, 0x5f, 0x72, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f,
+	0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52,
+	0x09, 0x6a, 0x6f, 0x69, 0x6e, 0x65, 0x64, 0x52, 0x6f, 0x77, 0x22, 0x7b, 0x0a, 0x0b, 0x4a, 0x6f,
+	0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61,
+	0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x66, 0x65, 0x61,
+	0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x26, 0x0a, 0x0f, 0x69, 0x6e, 0x70, 0x75, 0x74, 0x5f, 0x66,
+	0x69, 0x6c, 0x65, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d,
+	0x69, 0x6e, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c, 0x65, 0x50, 0x61, 0x74, 0x68, 0x12, 0x28, 0x0a,
+	0x10, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x70, 0x61, 0x74,
+	0x68, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x46,
+	0x69, 0x6c, 0x65, 0x50, 0x61, 0x74, 0x68, 0x22, 0x0e, 0x0a, 0x0c, 0x4a, 0x6f, 0x69, 0x6e, 0x52,
+	0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x76, 0x0a, 0x14, 0x43, 0x68, 0x61, 0x6e, 0x6e,
+	0x65, 0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12,
+	0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28,
+	0x09, 0x52, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x75,
+	0x6e, 0x69, 0x78, 0x5f, 0x6d, 0x69, 0x6c, 0x6c, 0x69, 0x18, 0x02, 0x20, 0x01, 0x28, 0x03, 0x52,
+	0x09, 0x75, 0x6e, 0x69, 0x78, 0x4d, 0x69, 0x6c, 0x6c, 0x69, 0x12, 0x19, 0x0a, 0x05, 0x6c, 0x69,
+	0x6d, 0x69, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x48, 0x00, 0x52, 0x05, 0x6c, 0x69, 0x6d,
+	0x69, 0x74, 0x88, 0x01, 0x01, 0x42, 0x08, 0x0a, 0x06, 0x5f, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22,
+	0x52, 0x0a, 0x15, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74,
+	0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x16, 0x0a, 0x06, 0x68, 0x65, 0x61, 0x64,
+	0x65, 0x72, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72,
+	0x12, 0x21, 0x0a, 0x03, 0x72, 0x6f, 0x77, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x0f, 0x2e,
+	0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x56, 0x61, 0x6c, 0x75, 0x65, 0x52, 0x03,
+	0x72, 0x6f, 0x77, 0x22, 0x99, 0x01, 0x0a, 0x0d, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65,
+	0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65,
+	0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x09, 0x52, 0x08, 0x66, 0x65, 0x61, 0x74, 0x75, 0x72, 0x65,
+	0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x75, 0x6e, 0x69, 0x78, 0x5f, 0x6d, 0x69, 0x6c, 0x6c, 0x69, 0x18,
+	0x02, 0x20, 0x01, 0x28, 0x03, 0x52, 0x09, 0x75, 0x6e, 0x69, 0x78, 0x4d, 0x69, 0x6c, 0x6c, 0x69,
+	0x12, 0x19, 0x0a, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x18, 0x03, 0x20, 0x01, 0x28, 0x04, 0x48,
+	0x00, 0x52, 0x05, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x88, 0x01, 0x01, 0x12, 0x28, 0x0a, 0x10, 0x6f,
+	0x75, 0x74, 0x70, 0x75, 0x74, 0x5f, 0x66, 0x69, 0x6c, 0x65, 0x5f, 0x70, 0x61, 0x74, 0x68, 0x18,
+	0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x6f, 0x75, 0x74, 0x70, 0x75, 0x74, 0x46, 0x69, 0x6c,
+	0x65, 0x50, 0x61, 0x74, 0x68, 0x42, 0x08, 0x0a, 0x06, 0x5f, 0x6c, 0x69, 0x6d, 0x69, 0x74, 0x22,
+	0x10, 0x0a, 0x0e, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x32, 0xcf, 0x06, 0x0a, 0x08, 0x4f, 0x6f, 0x6d, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x12, 0x46,
+	0x0a, 0x09, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74, 0x12, 0x1a, 0x2e, 0x6f, 0x6f,
+	0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1b, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65,
+	0x6e, 0x74, 0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x47, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x55, 0x0a, 0x0e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65,
+	0x4d, 0x75, 0x6c, 0x74, 0x69, 0x47, 0x65, 0x74, 0x12, 0x1f, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67,
+	0x65, 0x6e, 0x74, 0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x4d, 0x75, 0x6c, 0x74, 0x69, 0x47,
+	0x65, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x20, 0x2e, 0x6f, 0x6f, 0x6d, 0x61,
+	0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4f, 0x6e, 0x6c, 0x69, 0x6e, 0x65, 0x4d, 0x75, 0x6c, 0x74, 0x69,
+	0x47, 0x65, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a,
+	0x04, 0x53, 0x79, 0x6e, 0x63, 0x12, 0x15, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74,
+	0x2e, 0x53, 0x79, 0x6e, 0x63, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x6f,
+	0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x53, 0x79, 0x6e, 0x63, 0x52, 0x65, 0x73, 0x70,
+	0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x4d, 0x0a, 0x0d, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65,
+	0x6c, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x1e, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65,
+	0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74,
+	0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65,
+	0x6e, 0x74, 0x2e, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73,
+	0x65, 0x22, 0x00, 0x28, 0x01, 0x12, 0x3d, 0x0a, 0x06, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x12,
+	0x17, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x49, 0x6d, 0x70, 0x6f, 0x72,
+	0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67,
+	0x65, 0x6e, 0x74, 0x2e, 0x49, 0x6d, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e,
+	0x73, 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x04, 0x50, 0x75, 0x73, 0x68, 0x12, 0x15, 0x2e, 0x6f,
+	0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x50, 0x75, 0x73, 0x68, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x50,
+	0x75, 0x73, 0x68, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x43, 0x0a,
+	0x08, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x12, 0x19, 0x2e, 0x6f, 0x6f, 0x6d, 0x61,
+	0x67, 0x65, 0x6e, 0x74, 0x2e, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x71,
+	0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e,
+	0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x50, 0x0a, 0x0b, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69,
+	0x6e, 0x12, 0x1c, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61,
+	0x6e, 0x6e, 0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a,
+	0x1d, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e,
+	0x65, 0x6c, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00,
+	0x28, 0x01, 0x30, 0x01, 0x12, 0x37, 0x0a, 0x04, 0x4a, 0x6f, 0x69, 0x6e, 0x12, 0x15, 0x2e, 0x6f,
+	0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4a, 0x6f, 0x69, 0x6e, 0x52, 0x65, 0x71, 0x75,
+	0x65, 0x73, 0x74, 0x1a, 0x16, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x4a,
+	0x6f, 0x69, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x54, 0x0a,
+	0x0d, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65, 0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x1e,
+	0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65,
+	0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1f,
+	0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x68, 0x61, 0x6e, 0x6e, 0x65,
+	0x6c, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22,
+	0x00, 0x30, 0x01, 0x12, 0x3d, 0x0a, 0x06, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x17, 0x2e,
+	0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52,
+	0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x18, 0x2e, 0x6f, 0x6f, 0x6d, 0x61, 0x67, 0x65, 0x6e,
+	0x74, 0x2e, 0x45, 0x78, 0x70, 0x6f, 0x72, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65,
+	0x22, 0x00, 0x12, 0x3f, 0x0a, 0x0b, 0x48, 0x65, 0x61, 0x6c, 0x74, 0x68, 0x43, 0x68, 0x65, 0x63,
+	0x6b, 0x12, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f,
+	0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74, 0x79, 0x1a, 0x16, 0x2e, 0x67, 0x6f, 0x6f, 0x67,
+	0x6c, 0x65, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2e, 0x45, 0x6d, 0x70, 0x74,
+	0x79, 0x22, 0x00, 0x42, 0x0a, 0x5a, 0x08, 0x2f, 0x63, 0x6f, 0x64, 0x65, 0x67, 0x65, 0x6e, 0x62,
+	0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
diff --git a/oomagent/server.go b/oomagent/server.go
index 6005737b..967cbc83 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -172,7 +172,7 @@ func (s *server) Import(ctx context.Context, req *codegen.ImportRequest) (*codeg
 		Revision:       req.Revision,
 		DataSourceType: types.CSV_FILE,
 		CsvFileDataSource: &types.CsvFileDataSource{
-			InputFilePath: req.InputFile,
+			InputFilePath: req.InputFilePath,
 			Delimiter:     ",",
 		},
 	})
@@ -273,8 +273,8 @@ func (s *server) ChannelJoin(stream codegen.OomAgent_ChannelJoinServer) error {
 func (s *server) Join(ctx context.Context, req *codegen.JoinRequest) (*codegen.JoinResponse, error) {
 	err := s.oomstore.Join(ctx, types.JoinOpt{
 		FeatureNames:   req.Features,
-		InputFilePath:  req.InputFile,
-		OutputFilePath: req.OutputFile,
+		InputFilePath:  req.InputFilePath,
+		OutputFilePath: req.OutputFilePath,
 	})
 	if err != nil {
 		return nil, internalError(err.Error())
@@ -320,7 +320,7 @@ func (s *server) Export(ctx context.Context, req *codegen.ExportRequest) (*codeg
 		FeatureNames:   req.Features,
 		UnixMilli:      req.UnixMilli,
 		Limit:          req.Limit,
-		OutputFilePath: req.OutputFile,
+		OutputFilePath: req.OutputFilePath,
 	})
 	if err != nil {
 		return nil, internalError(err.Error())
diff --git a/oomcli/cmd/docgen.go b/oomcli/cmd/docgen.go
index 1a00db27..a5c2babb 100644
--- a/oomcli/cmd/docgen.go
+++ b/oomcli/cmd/docgen.go
@@ -25,7 +25,6 @@ func init() {
 
 func genDoc() {
 	var sb strings.Builder
-	sb.WriteString("# CLI\n\n")
 	genDocTree(&sb, rootCmd)
 	fmt.Println(sb.String())
 }
diff --git a/oomcli/cmd/register_group.go b/oomcli/cmd/register_group.go
index 84bde3e6..04f9f812 100644
--- a/oomcli/cmd/register_group.go
+++ b/oomcli/cmd/register_group.go
@@ -2,6 +2,7 @@ package cmd
 
 import (
 	"context"
+	"time"
 
 	"github.com/spf13/cobra"
 
@@ -18,6 +19,10 @@ var registerGroupCmd = &cobra.Command{
 			exitf("illegal category '%s', should be either 'stream' or 'batch'", registerGroupOpt.Category)
 		}
 
+		if registerGroupOpt.Category == types.CategoryBatch {
+			registerGroupOpt.SnapshotInterval = 0
+		}
+
 		registerGroupOpt.GroupName = args[0]
 	},
 	Run: func(cmd *cobra.Command, args []string) {
@@ -41,5 +46,7 @@ func init() {
 
 	flags.StringVarP(&registerGroupOpt.Category, "category", "c", "batch", "group category")
 
+	flags.IntVarP(&registerGroupOpt.SnapshotInterval, "snapshot_interval", "s", 24*int(time.Hour.Seconds()), "stream group snapshot interval")
+
 	flags.StringVarP(&registerGroupOpt.Description, "description", "d", "", "group description")
 }
diff --git a/oomcli/cmd/types.go b/oomcli/cmd/types.go
index 6b634b59..11d64446 100644
--- a/oomcli/cmd/types.go
+++ b/oomcli/cmd/types.go
@@ -17,11 +17,12 @@ type FlattenEntity struct {
 }
 
 type FlattenGroup struct {
-	ID          int            `oomcli:"ID"`
-	Name        string         `oomcli:"NAME"`
-	Entity      string         `oomcli:"ENTITY"`
-	Category    types.Category `oomcli:"CATEGORY"`
-	Description string         `oomcli:"DESCRIPTION,truncate"`
+	ID               int            `oomcli:"ID"`
+	Name             string         `oomcli:"NAME"`
+	Entity           string         `oomcli:"ENTITY"`
+	Category         types.Category `oomcli:"CATEGORY"`
+	SnapshotInterval time.Duration  `oomcli:"SNAPSHOT-INTERVAL"`
+	Description      string         `oomcli:"DESCRIPTION,truncate"`
 
 	OnlineRevisionID *int      `oomcli:"ONLINE-REVISION-ID,wide"`
 	CreateTime       time.Time `oomcli:"CREATE-TIME,wide"`
@@ -102,6 +103,7 @@ func parseTokenLists(i interface{}) (headerTokens TokenList, dataTokens []TokenL
 				Name:             e.Name,
 				Entity:           e.Entity.Name,
 				Category:         e.Category,
+				SnapshotInterval: time.Duration(e.SnapshotInterval) * time.Second,
 				Description:      e.Description,
 				OnlineRevisionID: e.OnlineRevisionID,
 				CreateTime:       e.CreateTime,
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index 6024949d..dec4fdaf 100644
--- a/pkg/oomstore/apply.go
+++ b/pkg/oomstore/apply.go
@@ -109,10 +109,11 @@ func (s *OomStore) applyGroup(ctx context.Context, tx metadata.DBStore, newGroup
 		}
 
 		id, err := tx.CreateGroup(ctx, metadata.CreateGroupOpt{
-			GroupName:   newGroup.Name,
-			EntityID:    entity.ID,
-			Category:    newGroup.Category,
-			Description: newGroup.Description,
+			GroupName:        newGroup.Name,
+			EntityID:         entity.ID,
+			Category:         newGroup.Category,
+			SnapshotInterval: int(newGroup.SnapshotInterval.Seconds()),
+			Description:      newGroup.Description,
 		})
 		if err != nil {
 			return nil, err
@@ -129,13 +130,26 @@ func (s *OomStore) applyGroup(ctx context.Context, tx metadata.DBStore, newGroup
 		return nil, nil
 	}
 
+	opt := metadata.UpdateGroupOpt{
+		GroupID: group.ID,
+	}
 	if newGroup.Description != group.Description {
+		opt.NewDescription = &newGroup.Description
+	}
+	if newGroup.Category == types.CategoryStream && newGroup.SnapshotInterval != 0 {
+		second := int(newGroup.SnapshotInterval.Seconds())
+		opt.NewSnapshotInterval = &second
+	}
+
+	if opt.NewDescription != nil || opt.NewSnapshotInterval != nil {
 		return nil, tx.UpdateGroup(ctx, metadata.UpdateGroupOpt{
-			GroupID:        group.ID,
-			NewDescription: &newGroup.Description,
+			GroupID:             group.ID,
+			NewSnapshotInterval: opt.NewSnapshotInterval,
+			NewDescription:      &newGroup.Description,
 		})
 
 	}
+
 	return nil, nil
 }
 
@@ -217,8 +231,8 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 		switch kind {
 		case "Entity":
 			var entity apply.Entity
-			if err := mapstructure.Decode(data, &entity); err != nil {
-				return nil, err
+			if err := mapstructureDecode(data, &entity); err != nil {
+				return nil, errdefs.WithStack(err)
 			}
 
 			stage.NewEntities = append(stage.NewEntities, buildApplyEntity(entity))
@@ -232,8 +246,8 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 			}
 		case "Group":
 			var group apply.Group
-			if err := mapstructure.Decode(data, &group); err != nil {
-				return nil, err
+			if err := mapstructureDecode(data, &group); err != nil {
+				return nil, errdefs.WithStack(err)
 			}
 
 			stage.NewGroups = append(stage.NewGroups, buildApplyGroup(group, group.EntityName))
@@ -244,8 +258,8 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 
 		case "Feature":
 			var feature apply.Feature
-			if err := mapstructure.Decode(data, &feature); err != nil {
-				return nil, err
+			if err := mapstructureDecode(data, &feature); err != nil {
+				return nil, errdefs.WithStack(err)
 			}
 			stage.NewFeatures = append(stage.NewFeatures, feature)
 
@@ -257,7 +271,7 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 			switch itemsKind {
 			case "Feature":
 				featureItems := apply.FeatureItems{}
-				if err := mapstructure.Decode(data, &featureItems); err != nil {
+				if err := mapstructureDecode(data, &featureItems); err != nil {
 					return nil, err
 				}
 				for _, item := range featureItems.Items {
@@ -265,7 +279,7 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 				}
 			case "Group":
 				groupItems := apply.GroupItems{}
-				if err := mapstructure.Decode(data, &groupItems); err != nil {
+				if err := mapstructureDecode(data, &groupItems); err != nil {
 					return nil, err
 				}
 				for _, group := range groupItems.Items {
@@ -277,9 +291,10 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 				}
 			case "Entity":
 				entityItems := apply.EntityItems{}
-				if err := mapstructure.Decode(data, &entityItems); err != nil {
+				if err := mapstructureDecode(data, &entityItems); err != nil {
 					return nil, err
 				}
+
 				for _, entity := range entityItems.Items {
 					stage.NewEntities = append(stage.NewEntities, buildApplyEntity(entity))
 
@@ -299,6 +314,20 @@ func buildApplyStage(ctx context.Context, opt apply.ApplyOpt) (*apply.ApplyStage
 	return stage, nil
 }
 
+func mapstructureDecode(data map[string]interface{}, result interface{}) error {
+	dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
+		DecodeHook: mapstructure.StringToTimeDurationHookFunc(),
+		Result:     result,
+	})
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+	if err := dec.Decode(data); err != nil {
+		return errdefs.WithStack(err)
+	}
+	return nil
+}
+
 func parseKind(data map[string]interface{}) (string, error) {
 	if k, ok := data["kind"]; ok {
 		return cast.ToString(k), nil
@@ -311,7 +340,7 @@ func parseKind(data map[string]interface{}) (string, error) {
 
 func parseItemsKind(data map[string]interface{}) (string, error) {
 	items := apply.Items{}
-	if err := mapstructure.Decode(data, &items); err != nil {
+	if err := mapstructureDecode(data, &items); err != nil {
 		return "", err
 	}
 	return items.Kind(), nil
@@ -331,11 +360,12 @@ func buildApplyGroup(group apply.Group, entityName string) apply.Group {
 	// We don't want group.Features to have values.
 	// The whole stage should be a flat structure, not a nested structure.
 	return apply.Group{
-		Kind:        "Group",
-		Name:        group.Name,
-		EntityName:  entityName,
-		Category:    group.Category,
-		Description: group.Description,
+		Kind:             "Group",
+		Name:             group.Name,
+		EntityName:       entityName,
+		Category:         group.Category,
+		SnapshotInterval: group.SnapshotInterval,
+		Description:      group.Description,
 	}
 }
 
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index 5e75d7b3..1b96401d 100644
--- a/pkg/oomstore/group.go
+++ b/pkg/oomstore/group.go
@@ -16,10 +16,11 @@ func (s *OomStore) CreateGroup(ctx context.Context, opt types.CreateGroupOpt) (i
 	}
 
 	id, err := s.metadata.CreateGroup(ctx, metadata.CreateGroupOpt{
-		GroupName:   opt.GroupName,
-		EntityID:    entity.ID,
-		Description: opt.Description,
-		Category:    opt.Category,
+		GroupName:        opt.GroupName,
+		EntityID:         entity.ID,
+		Description:      opt.Description,
+		Category:         opt.Category,
+		SnapshotInterval: opt.SnapshotInterval,
 	})
 	if err != nil {
 		return 0, err
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index 385c9ac0..3b2cf38d 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -46,14 +46,8 @@ func (s *OomStore) Import(ctx context.Context, opt types.ImportOpt) (int, error)
 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 does not need pass down features
-	header, err := dbutil.ReadLine(dbutil.ReadLineOpt{
-		Source: source,
-	})
+	header, err := dbutil.ReadLine(reader, dataSource.Delimiter, nil, "")
 	if err != nil {
 		return 0, err
 	}
@@ -82,7 +76,10 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 		Header:            cast.ToStringSlice(header),
 		Revision:          opt.Revision,
 		SnapshotTableName: snapshotTableName,
-		Source:            source,
+		Source: &offline.CSVSource{
+			Reader:    reader,
+			Delimiter: dataSource.Delimiter,
+		},
 	})
 	if err != nil {
 		return 0, err
diff --git a/pkg/oomstore/types/apply/apply.go b/pkg/oomstore/types/apply/apply.go
index c833cd1c..ec08b66b 100644
--- a/pkg/oomstore/types/apply/apply.go
+++ b/pkg/oomstore/types/apply/apply.go
@@ -2,6 +2,7 @@ package apply
 
 import (
 	"io"
+	"time"
 
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -72,12 +73,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"`
-	Description string         `mapstructure:"description" yaml:"description"`
-	Features    []Feature      `mapstructure:"features" yaml:"features,omitempty"`
+	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"`
 }
 
 type GroupItems struct {
@@ -107,11 +109,12 @@ func FromGroupList(groups types.GroupList, features types.FeatureList) *GroupIte
 
 	for _, group := range groups {
 		items.Items = append(items.Items, Group{
-			Kind:        "Group",
-			Name:        group.Name,
-			EntityName:  group.Entity.Name,
-			Category:    group.Category,
-			Description: group.Description,
+			Kind:             "Group",
+			Name:             group.Name,
+			EntityName:       group.Entity.Name,
+			Category:         group.Category,
+			SnapshotInterval: time.Duration(group.SnapshotInterval) * time.Second,
+			Description:      group.Description,
 			Features: FromFeatureList(features.Filter(func(f *types.Feature) bool {
 				return f.Group.Name == group.Name
 			})).Walk(func(f Feature) Feature {
diff --git a/pkg/oomstore/types/group.go b/pkg/oomstore/types/group.go
index 8d966a60..4b9a8d24 100644
--- a/pkg/oomstore/types/group.go
+++ b/pkg/oomstore/types/group.go
@@ -13,6 +13,8 @@ type Group struct {
 	CreateTime  time.Time `db:"create_time"`
 	ModifyTime  time.Time `db:"modify_time"`
 
+	SnapshotInterval int `db:"snapshot_interval"`
+
 	EntityID         int  `db:"entity_id"`
 	OnlineRevisionID *int `db:"online_revision_id"`
 
diff --git a/pkg/oomstore/types/options.go b/pkg/oomstore/types/options.go
index 54ab88c5..6701b8b1 100644
--- a/pkg/oomstore/types/options.go
+++ b/pkg/oomstore/types/options.go
@@ -24,10 +24,11 @@ type CreateEntityOpt struct {
 }
 
 type CreateGroupOpt struct {
-	GroupName   string
-	EntityName  string
-	Category    Category
-	Description string
+	GroupName        string
+	EntityName       string
+	Category         Category
+	SnapshotInterval int
+	Description      string
 }
 
 type ChannelExportOpt struct {
diff --git a/proto/oomagent.proto b/proto/oomagent.proto
index d129a577..a70b9a3d 100644
--- a/proto/oomagent.proto
+++ b/proto/oomagent.proto
@@ -92,7 +92,7 @@ message ImportRequest {
   optional int64 revision = 2;
   optional string description = 3;
 
-  string input_file = 4;
+  string input_file_path = 4;
   optional string delimiter = 5;
 }
 
@@ -134,8 +134,8 @@ message ChannelJoinResponse {
 
 message JoinRequest {
   repeated string features = 1;
-  string input_file = 2;
-  string output_file = 3;
+  string input_file_path = 2;
+  string output_file_path = 3;
 }
 
 message JoinResponse {}
@@ -155,7 +155,7 @@ message ExportRequest {
   repeated string features = 1;
   int64 unix_milli = 2;
   optional uint64 limit = 3;
-  string output_file = 4;
+  string output_file_path = 4;
 }
 
 message ExportResponse {}
diff --git a/sdk/python/src/oomstore/codegen/oomagent_pb2.py b/sdk/python/src/oomstore/codegen/oomagent_pb2.py
index afc4aa1e..5fd02ba7 100644
--- a/sdk/python/src/oomstore/codegen/oomagent_pb2.py
+++ b/sdk/python/src/oomstore/codegen/oomagent_pb2.py
@@ -21,7 +21,7 @@ DESCRIPTOR = _descriptor.FileDescriptor(
   syntax='proto3',
   serialized_options=b'Z\010/codegen',
   create_key=_descriptor._internal_create_key,
-  serialized_pb=b'\n\x0eoomagent.proto\x12\x08oomagent\x1a\x1bgoogle/protobuf/empty.proto\x1a\x1cgoogle/protobuf/struct.proto\"\xd1\x01\n\x05Value\x12\x30\n\nnull_value\x18\x01 \x01(\x0e\x32\x1a.google.protobuf.NullValueH\x00\x12\x15\n\x0bint64_value\x18\x02 \x01(\x03H\x00\x12\x16\n\x0c\x64ouble_value\x18\x03 \x01(\x01H\x00\x12\x16\n\x0cstring_value\x18\x04 \x01(\tH\x00\x12\x14\n\nbool_value\x18\x05 \x01(\x08H\x00\x12\x1a\n\x10unix_milli_value\x18\x06 \x01(\x03H\x00\x12\x15\n\x0b\x62ytes_value\x18\x07 \x01(\x0cH\x00\x42\x06\n\x04kind\"8\n\x10OnlineGetRequest\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x02 \x03(\t\"\x7f\n\x0f\x46\x65\x61tureValueMap\x12/\n\x03map\x18\x01 \x03(\x0b\x32\".oomagent.FeatureValueMap.MapEntry\x1a;\n\x08MapEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x1e\n\x05value\x18\x02 \x01(\x0b\x32\x0f.oomagent.Value:\x02\x38\x01\">\n\x11OnlineGetResponse\x12)\n\x06result\x18\x01 \x01(\x0b\x32\x19.oomagent.FeatureValueMap\">\n\x15OnlineMultiGetRequest\x12\x13\n\x0b\x65ntity_keys\x18\x01 \x03(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x02 \x03(\t\"\xa0\x01\n\x16OnlineMultiGetResponse\x12<\n\x06result\x18\x01 \x03(\x0b\x32,.oomagent.OnlineMultiGetResponse.ResultEntry\x1aH\n\x0bResultEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12(\n\x05value\x18\x02 \x01(\x0b\x32\x19.oomagent.FeatureValueMap:\x02\x38\x01\"7\n\x0bSyncRequest\x12\x13\n\x0brevision_id\x18\x01 \x01(\x05\x12\x13\n\x0bpurge_delay\x18\x02 \x01(\x05\"\x0e\n\x0cSyncResponse\"\x8f\x01\n\x14\x43hannelImportRequest\x12\x12\n\x05group\x18\x01 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x0b\x64\x65scription\x18\x02 \x01(\tH\x01\x88\x01\x01\x12\x15\n\x08revision\x18\x03 \x01(\x03H\x02\x88\x01\x01\x12\x0b\n\x03row\x18\x04 \x01(\x0c\x42\x08\n\x06_groupB\x0e\n\x0c_descriptionB\x0b\n\t_revision\"%\n\x0eImportResponse\x12\x13\n\x0brevision_id\x18\x01 \x01(\x05\"\xa6\x01\n\rImportRequest\x12\r\n\x05group\x18\x01 \x01(\t\x12\x15\n\x08revision\x18\x02 \x01(\x03H\x00\x88\x01\x01\x12\x18\n\x0b\x64\x65scription\x18\x03 \x01(\tH\x01\x88\x01\x01\x12\x12\n\ninput_file\x18\x04 \x01(\t\x12\x16\n\tdelimiter\x18\x05 \x01(\tH\x02\x88\x01\x01\x42\x0b\n\t_revisionB\x0e\n\x0c_descriptionB\x0c\n\n_delimiter\"\x0e\n\x0cPushResponse\"k\n\x0bPushRequest\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\r\n\x05group\x18\x02 \x01(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x03 \x03(\t\x12\'\n\x0e\x66\x65\x61ture_values\x18\x04 \x03(\x0b\x32\x0f.oomagent.Value\"\x12\n\x10SnapshotResponse\" \n\x0fSnapshotRequest\x12\r\n\x05group\x18\x01 \x01(\t\"C\n\tEntityRow\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x0e\n\x06values\x18\x03 \x03(\t\"n\n\x12\x43hannelJoinRequest\x12\x15\n\rjoin_features\x18\x01 \x03(\t\x12\x18\n\x10\x65xisted_features\x18\x02 \x03(\t\x12\'\n\nentity_row\x18\x03 \x01(\x0b\x32\x13.oomagent.EntityRow\"J\n\x13\x43hannelJoinResponse\x12\x0e\n\x06header\x18\x01 \x03(\t\x12#\n\njoined_row\x18\x02 \x03(\x0b\x32\x0f.oomagent.Value\"H\n\x0bJoinRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x12\n\ninput_file\x18\x02 \x01(\t\x12\x13\n\x0boutput_file\x18\x03 \x01(\t\"\x0e\n\x0cJoinResponse\"Z\n\x14\x43hannelExportRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x12\n\x05limit\x18\x03 \x01(\x04H\x00\x88\x01\x01\x42\x08\n\x06_limit\"E\n\x15\x43hannelExportResponse\x12\x0e\n\x06header\x18\x01 \x03(\t\x12\x1c\n\x03row\x18\x02 \x03(\x0b\x32\x0f.oomagent.Value\"h\n\rExportRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x12\n\x05limit\x18\x03 \x01(\x04H\x00\x88\x01\x01\x12\x13\n\x0boutput_file\x18\x04 \x01(\tB\x08\n\x06_limit\"\x10\n\x0e\x45xportResponse2\xcf\x06\n\x08OomAgent\x12\x46\n\tOnlineGet\x12\x1a.oomagent.OnlineGetRequest\x1a\x1b.oomagent.OnlineGetResponse\"\x00\x12U\n\x0eOnlineMultiGet\x12\x1f.oomagent.OnlineMultiGetRequest\x1a .oomagent.OnlineMultiGetResponse\"\x00\x12\x37\n\x04Sync\x12\x15.oomagent.SyncRequest\x1a\x16.oomagent.SyncResponse\"\x00\x12M\n\rChannelImport\x12\x1e.oomagent.ChannelImportRequest\x1a\x18.oomagent.ImportResponse\"\x00(\x01\x12=\n\x06Import\x12\x17.oomagent.ImportRequest\x1a\x18.oomagent.ImportResponse\"\x00\x12\x37\n\x04Push\x12\x15.oomagent.PushRequest\x1a\x16.oomagent.PushResponse\"\x00\x12\x43\n\x08Snapshot\x12\x19.oomagent.SnapshotRequest\x1a\x1a.oomagent.SnapshotResponse\"\x00\x12P\n\x0b\x43hannelJoin\x12\x1c.oomagent.ChannelJoinRequest\x1a\x1d.oomagent.ChannelJoinResponse\"\x00(\x01\x30\x01\x12\x37\n\x04Join\x12\x15.oomagent.JoinRequest\x1a\x16.oomagent.JoinResponse\"\x00\x12T\n\rChannelExport\x12\x1e.oomagent.ChannelExportRequest\x1a\x1f.oomagent.ChannelExportResponse\"\x00\x30\x01\x12=\n\x06\x45xport\x12\x17.oomagent.ExportRequest\x1a\x18.oomagent.ExportResponse\"\x00\x12?\n\x0bHealthCheck\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x42\nZ\x08/codegenb\x06proto3'
+  serialized_pb=b'\n\x0eoomagent.proto\x12\x08oomagent\x1a\x1bgoogle/protobuf/empty.proto\x1a\x1cgoogle/protobuf/struct.proto\"\xd1\x01\n\x05Value\x12\x30\n\nnull_value\x18\x01 \x01(\x0e\x32\x1a.google.protobuf.NullValueH\x00\x12\x15\n\x0bint64_value\x18\x02 \x01(\x03H\x00\x12\x16\n\x0c\x64ouble_value\x18\x03 \x01(\x01H\x00\x12\x16\n\x0cstring_value\x18\x04 \x01(\tH\x00\x12\x14\n\nbool_value\x18\x05 \x01(\x08H\x00\x12\x1a\n\x10unix_milli_value\x18\x06 \x01(\x03H\x00\x12\x15\n\x0b\x62ytes_value\x18\x07 \x01(\x0cH\x00\x42\x06\n\x04kind\"8\n\x10OnlineGetRequest\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x02 \x03(\t\"\x7f\n\x0f\x46\x65\x61tureValueMap\x12/\n\x03map\x18\x01 \x03(\x0b\x32\".oomagent.FeatureValueMap.MapEntry\x1a;\n\x08MapEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\x1e\n\x05value\x18\x02 \x01(\x0b\x32\x0f.oomagent.Value:\x02\x38\x01\">\n\x11OnlineGetResponse\x12)\n\x06result\x18\x01 \x01(\x0b\x32\x19.oomagent.FeatureValueMap\">\n\x15OnlineMultiGetRequest\x12\x13\n\x0b\x65ntity_keys\x18\x01 \x03(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x02 \x03(\t\"\xa0\x01\n\x16OnlineMultiGetResponse\x12<\n\x06result\x18\x01 \x03(\x0b\x32,.oomagent.OnlineMultiGetResponse.ResultEntry\x1aH\n\x0bResultEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12(\n\x05value\x18\x02 \x01(\x0b\x32\x19.oomagent.FeatureValueMap:\x02\x38\x01\"7\n\x0bSyncRequest\x12\x13\n\x0brevision_id\x18\x01 \x01(\x05\x12\x13\n\x0bpurge_delay\x18\x02 \x01(\x05\"\x0e\n\x0cSyncResponse\"\x8f\x01\n\x14\x43hannelImportRequest\x12\x12\n\x05group\x18\x01 \x01(\tH\x00\x88\x01\x01\x12\x18\n\x0b\x64\x65scription\x18\x02 \x01(\tH\x01\x88\x01\x01\x12\x15\n\x08revision\x18\x03 \x01(\x03H\x02\x88\x01\x01\x12\x0b\n\x03row\x18\x04 \x01(\x0c\x42\x08\n\x06_groupB\x0e\n\x0c_descriptionB\x0b\n\t_revision\"%\n\x0eImportResponse\x12\x13\n\x0brevision_id\x18\x01 \x01(\x05\"\xab\x01\n\rImportRequest\x12\r\n\x05group\x18\x01 \x01(\t\x12\x15\n\x08revision\x18\x02 \x01(\x03H\x00\x88\x01\x01\x12\x18\n\x0b\x64\x65scription\x18\x03 \x01(\tH\x01\x88\x01\x01\x12\x17\n\x0finput_file_path\x18\x04 \x01(\t\x12\x16\n\tdelimiter\x18\x05 \x01(\tH\x02\x88\x01\x01\x42\x0b\n\t_revisionB\x0e\n\x0c_descriptionB\x0c\n\n_delimiter\"\x0e\n\x0cPushResponse\"k\n\x0bPushRequest\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\r\n\x05group\x18\x02 \x01(\t\x12\x10\n\x08\x66\x65\x61tures\x18\x03 \x03(\t\x12\'\n\x0e\x66\x65\x61ture_values\x18\x04 \x03(\x0b\x32\x0f.oomagent.Value\"\x12\n\x10SnapshotResponse\" \n\x0fSnapshotRequest\x12\r\n\x05group\x18\x01 \x01(\t\"C\n\tEntityRow\x12\x12\n\nentity_key\x18\x01 \x01(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x0e\n\x06values\x18\x03 \x03(\t\"n\n\x12\x43hannelJoinRequest\x12\x15\n\rjoin_features\x18\x01 \x03(\t\x12\x18\n\x10\x65xisted_features\x18\x02 \x03(\t\x12\'\n\nentity_row\x18\x03 \x01(\x0b\x32\x13.oomagent.EntityRow\"J\n\x13\x43hannelJoinResponse\x12\x0e\n\x06header\x18\x01 \x03(\t\x12#\n\njoined_row\x18\x02 \x03(\x0b\x32\x0f.oomagent.Value\"R\n\x0bJoinRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x17\n\x0finput_file_path\x18\x02 \x01(\t\x12\x18\n\x10output_file_path\x18\x03 \x01(\t\"\x0e\n\x0cJoinResponse\"Z\n\x14\x43hannelExportRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x12\n\x05limit\x18\x03 \x01(\x04H\x00\x88\x01\x01\x42\x08\n\x06_limit\"E\n\x15\x43hannelExportResponse\x12\x0e\n\x06header\x18\x01 \x03(\t\x12\x1c\n\x03row\x18\x02 \x03(\x0b\x32\x0f.oomagent.Value\"m\n\rExportRequest\x12\x10\n\x08\x66\x65\x61tures\x18\x01 \x03(\t\x12\x12\n\nunix_milli\x18\x02 \x01(\x03\x12\x12\n\x05limit\x18\x03 \x01(\x04H\x00\x88\x01\x01\x12\x18\n\x10output_file_path\x18\x04 \x01(\tB\x08\n\x06_limit\"\x10\n\x0e\x45xportResponse2\xcf\x06\n\x08OomAgent\x12\x46\n\tOnlineGet\x12\x1a.oomagent.OnlineGetRequest\x1a\x1b.oomagent.OnlineGetResponse\"\x00\x12U\n\x0eOnlineMultiGet\x12\x1f.oomagent.OnlineMultiGetRequest\x1a .oomagent.OnlineMultiGetResponse\"\x00\x12\x37\n\x04Sync\x12\x15.oomagent.SyncRequest\x1a\x16.oomagent.SyncResponse\"\x00\x12M\n\rChannelImport\x12\x1e.oomagent.ChannelImportRequest\x1a\x18.oomagent.ImportResponse\"\x00(\x01\x12=\n\x06Import\x12\x17.oomagent.ImportRequest\x1a\x18.oomagent.ImportResponse\"\x00\x12\x37\n\x04Push\x12\x15.oomagent.PushRequest\x1a\x16.oomagent.PushResponse\"\x00\x12\x43\n\x08Snapshot\x12\x19.oomagent.SnapshotRequest\x1a\x1a.oomagent.SnapshotResponse\"\x00\x12P\n\x0b\x43hannelJoin\x12\x1c.oomagent.ChannelJoinRequest\x1a\x1d.oomagent.ChannelJoinResponse\"\x00(\x01\x30\x01\x12\x37\n\x04Join\x12\x15.oomagent.JoinRequest\x1a\x16.oomagent.JoinResponse\"\x00\x12T\n\rChannelExport\x12\x1e.oomagent.ChannelExportRequest\x1a\x1f.oomagent.ChannelExportResponse\"\x00\x30\x01\x12=\n\x06\x45xport\x12\x17.oomagent.ExportRequest\x1a\x18.oomagent.ExportResponse\"\x00\x12?\n\x0bHealthCheck\x12\x16.google.protobuf.Empty\x1a\x16.google.protobuf.Empty\"\x00\x42\nZ\x08/codegenb\x06proto3'
   ,
   dependencies=[google_dot_protobuf_dot_empty__pb2.DESCRIPTOR,google_dot_protobuf_dot_struct__pb2.DESCRIPTOR,])
 
@@ -551,7 +551,7 @@ _IMPORTREQUEST = _descriptor.Descriptor(
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
     _descriptor.FieldDescriptor(
-      name='input_file', full_name='oomagent.ImportRequest.input_file', index=3,
+      name='input_file_path', full_name='oomagent.ImportRequest.input_file_path', index=3,
       number=4, type=9, cpp_type=9, label=1,
       has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
@@ -592,7 +592,7 @@ _IMPORTREQUEST = _descriptor.Descriptor(
     fields=[]),
   ],
   serialized_start=1036,
-  serialized_end=1202,
+  serialized_end=1207,
 )
 
 
@@ -616,8 +616,8 @@ _PUSHRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1204,
-  serialized_end=1218,
+  serialized_start=1209,
+  serialized_end=1223,
 )
 
 
@@ -669,8 +669,8 @@ _PUSHREQUEST = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1220,
-  serialized_end=1327,
+  serialized_start=1225,
+  serialized_end=1332,
 )
 
 
@@ -694,8 +694,8 @@ _SNAPSHOTRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1329,
-  serialized_end=1347,
+  serialized_start=1334,
+  serialized_end=1352,
 )
 
 
@@ -726,8 +726,8 @@ _SNAPSHOTREQUEST = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1349,
-  serialized_end=1381,
+  serialized_start=1354,
+  serialized_end=1386,
 )
 
 
@@ -772,8 +772,8 @@ _ENTITYROW = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1383,
-  serialized_end=1450,
+  serialized_start=1388,
+  serialized_end=1455,
 )
 
 
@@ -818,8 +818,8 @@ _CHANNELJOINREQUEST = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1452,
-  serialized_end=1562,
+  serialized_start=1457,
+  serialized_end=1567,
 )
 
 
@@ -857,8 +857,8 @@ _CHANNELJOINRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1564,
-  serialized_end=1638,
+  serialized_start=1569,
+  serialized_end=1643,
 )
 
 
@@ -878,14 +878,14 @@ _JOINREQUEST = _descriptor.Descriptor(
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
     _descriptor.FieldDescriptor(
-      name='input_file', full_name='oomagent.JoinRequest.input_file', index=1,
+      name='input_file_path', full_name='oomagent.JoinRequest.input_file_path', index=1,
       number=2, type=9, cpp_type=9, label=1,
       has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
     _descriptor.FieldDescriptor(
-      name='output_file', full_name='oomagent.JoinRequest.output_file', index=2,
+      name='output_file_path', full_name='oomagent.JoinRequest.output_file_path', index=2,
       number=3, type=9, cpp_type=9, label=1,
       has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
@@ -903,8 +903,8 @@ _JOINREQUEST = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1640,
-  serialized_end=1712,
+  serialized_start=1645,
+  serialized_end=1727,
 )
 
 
@@ -928,8 +928,8 @@ _JOINRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1714,
-  serialized_end=1728,
+  serialized_start=1729,
+  serialized_end=1743,
 )
 
 
@@ -979,8 +979,8 @@ _CHANNELEXPORTREQUEST = _descriptor.Descriptor(
       create_key=_descriptor._internal_create_key,
     fields=[]),
   ],
-  serialized_start=1730,
-  serialized_end=1820,
+  serialized_start=1745,
+  serialized_end=1835,
 )
 
 
@@ -1018,8 +1018,8 @@ _CHANNELEXPORTRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1822,
-  serialized_end=1891,
+  serialized_start=1837,
+  serialized_end=1906,
 )
 
 
@@ -1053,7 +1053,7 @@ _EXPORTREQUEST = _descriptor.Descriptor(
       is_extension=False, extension_scope=None,
       serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
     _descriptor.FieldDescriptor(
-      name='output_file', full_name='oomagent.ExportRequest.output_file', index=3,
+      name='output_file_path', full_name='oomagent.ExportRequest.output_file_path', index=3,
       number=4, type=9, cpp_type=9, label=1,
       has_default_value=False, default_value=b"".decode('utf-8'),
       message_type=None, enum_type=None, containing_type=None,
@@ -1076,8 +1076,8 @@ _EXPORTREQUEST = _descriptor.Descriptor(
       create_key=_descriptor._internal_create_key,
     fields=[]),
   ],
-  serialized_start=1893,
-  serialized_end=1997,
+  serialized_start=1908,
+  serialized_end=2017,
 )
 
 
@@ -1101,8 +1101,8 @@ _EXPORTRESPONSE = _descriptor.Descriptor(
   extension_ranges=[],
   oneofs=[
   ],
-  serialized_start=1999,
-  serialized_end=2015,
+  serialized_start=2019,
+  serialized_end=2035,
 )
 
 _VALUE.fields_by_name['null_value'].enum_type = google_dot_protobuf_dot_struct__pb2._NULLVALUE
@@ -1384,8 +1384,8 @@ _OOMAGENT = _descriptor.ServiceDescriptor(
   index=0,
   serialized_options=None,
   create_key=_descriptor._internal_create_key,
-  serialized_start=2018,
-  serialized_end=2865,
+  serialized_start=2038,
+  serialized_end=2885,
   methods=[
   _descriptor.MethodDescriptor(
     name='OnlineGet',
diff --git a/sdk/rust/examples/join.rs b/sdk/rust/examples/join.rs
deleted file mode 100644
index 038adfcf..00000000
--- a/sdk/rust/examples/join.rs
+++ /dev/null
@@ -1,51 +0,0 @@
-use std::fs;
-
-use oomstore::Client;
-
-macro_rules! svec { ($($part:expr),* $(,)?) => {{ vec![$($part.to_string(),)*] }} }
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = Client::connect("http://127.0.0.1:50051").await?;
-
-    let contents = r#"
-entity_key,unix_milli,age
-1,0,20
-1,3,44
-1,4,37
-2,3,28
-3,3,48
-4,4,22
-5,2,41
-5,2,43
-5,3,47
-5,4,59
-5,5,55
-6,1,27
-6,3,46
-7,0,42
-7,4,40
-7,4,49
-8,4,39
-9,3,35
-10,4,54
-10,4,33
-"#
-    .trim_start();
-
-    let input = "/tmp/driver_stats_label.csv";
-    let output = "/tmp/joined.csv";
-    fs::write(input, contents)?;
-
-    let features = svec![
-        "driver_stats.conv_rate",
-        "driver_stats.acc_rate",
-        "driver_stats.avg_daily_trips"
-    ];
-
-    client.join(features, input, output).await?;
-
-    fs::copy(output, "/dev/stdout")?;
-
-    Ok(())
-}
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index 42e593fc..ade3d823 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -1,7 +1,6 @@
 pub mod error;
 mod util;
 
-use crate::oomagent::JoinRequest;
 use std::{collections::HashMap, path::Path};
 
 use error::OomError;
@@ -143,11 +142,11 @@ impl Client {
         let res = self
             .inner
             .import(ImportRequest {
-                group:       group.into(),
-                description: description.into(),
-                revision:    revision.into(),
-                input_file:  input_file.as_ref().display().to_string(),
-                delimiter:   delimiter.into().map(String::from),
+                group:           group.into(),
+                description:     description.into(),
+                revision:        revision.into(),
+                input_file_path: input_file.as_ref().display().to_string(),
+                delimiter:       delimiter.into().map(String::from),
             })
             .await?
             .into_inner();
@@ -215,20 +214,4 @@ impl Client {
         };
         Ok((header, outbound))
     }
-
-    pub async fn join(
-        &mut self,
-        features: Vec<String>,
-        input_file: impl AsRef<Path>,
-        output_file: impl AsRef<Path>,
-    ) -> Result<()> {
-        self.inner
-            .join(JoinRequest {
-                features,
-                input_file: input_file.as_ref().display().to_string(),
-                output_file: output_file.as_ref().display().to_string(),
-            })
-            .await?;
-        Ok(())
-    }
 }


Test patch:
diff --git a/internal/database/metadata/mysql/db_test.go b/internal/database/metadata/mysql/db_test.go
index 6d19da7b..19f7bc8e 100644
--- a/internal/database/metadata/mysql/db_test.go
+++ b/internal/database/metadata/mysql/db_test.go
@@ -101,6 +101,10 @@ func TestCreateGroup(t *testing.T) {
 	test_impl.TestCreateGroup(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
 
+func TestCreateStreamGroup(t *testing.T) {
+	test_impl.TestCreateStreamGroup(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
+}
+
 func TestUpdateGroup(t *testing.T) {
 	test_impl.TestUpdateGroup(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
diff --git a/internal/database/metadata/postgres/db_test.go b/internal/database/metadata/postgres/db_test.go
index a8225582..7f63ada1 100644
--- a/internal/database/metadata/postgres/db_test.go
+++ b/internal/database/metadata/postgres/db_test.go
@@ -134,6 +134,10 @@ func TestCreateGroup(t *testing.T) {
 	test_impl.TestCreateGroup(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
 
+func TestCreateStreamGroup(t *testing.T) {
+	test_impl.TestCreateStreamGroup(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
+}
+
 func TestUpdateGroup(t *testing.T) {
 	test_impl.TestUpdateGroup(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
diff --git a/internal/database/metadata/sqlite/db_test.go b/internal/database/metadata/sqlite/db_test.go
index 18b95765..1f3288a8 100644
--- a/internal/database/metadata/sqlite/db_test.go
+++ b/internal/database/metadata/sqlite/db_test.go
@@ -89,6 +89,10 @@ func TestCreateGroup(t *testing.T) {
 	test_impl.TestCreateGroup(t, prepareStore, destroyStore)
 }
 
+func TestCreateStreamGroup(t *testing.T) {
+	test_impl.TestCreateStreamGroup(t, prepareStore, destroyStore)
+}
+
 func TestUpdateGroup(t *testing.T) {
 	test_impl.TestUpdateGroup(t, prepareStore, destroyStore)
 }
diff --git a/internal/database/metadata/test_impl/group.go b/internal/database/metadata/test_impl/group.go
index d9bfa307..a2e3b3f1 100644
--- a/internal/database/metadata/test_impl/group.go
+++ b/internal/database/metadata/test_impl/group.go
@@ -2,7 +2,9 @@ package test_impl
 
 import (
 	"context"
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -153,6 +155,43 @@ func TestCreateGroup(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 	require.NotNil(t, err)
 }
 
+func TestCreateStreamGroup(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
+	t.Cleanup(destroyStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	entityID := prepareEntity(t, ctx, store, "user")
+
+	opt := metadata.CreateGroupOpt{
+		GroupName:        "user-click",
+		EntityID:         entityID,
+		Description:      "description",
+		Category:         types.CategoryStream,
+		SnapshotInterval: 24 * int(time.Hour.Seconds()),
+	}
+
+	// create successfully
+	groupID, err := store.CreateGroup(ctx, opt)
+	require.NoError(t, err)
+	require.NotEqual(t, 0, groupID)
+
+	// cannot create feature group with same name
+	_, err = store.CreateGroup(ctx, opt)
+	require.Equal(t, "feature group user-click already exists", err.Error())
+
+	// cannot create feature group with invalid category
+	opt.Category = "invalid-category"
+	_, err = store.CreateGroup(ctx, opt)
+	require.NotNil(t, err)
+
+	// cannot create stream group with snapshot_interval is zero
+	opt.Category = types.CategoryStream
+	opt.SnapshotInterval = 0
+	_, err = store.CreateGroup(ctx, opt)
+	require.Equal(t, fmt.Sprintf("the field SnapshotInterval of the stream group %s cannot be zero", opt.GroupName), err.Error())
+}
+
 func TestUpdateGroup(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
 	t.Cleanup(destroyStore)
 
diff --git a/oomagent/test/client.go b/oomagent/test/client.go
index 1c7f4cb8..aac2cb7f 100644
--- a/oomagent/test/client.go
+++ b/oomagent/test/client.go
@@ -62,7 +62,7 @@ func Import() {
 		log.Fatal(err)
 	}
 
-	file, err := os.Open("input_file")
+	file, err := os.Open("input_file_path")
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/oomagent/test/data/user_click.yaml b/oomagent/test/data/user_click.yaml
index 571fe09a..58ae6178 100644
--- a/oomagent/test/data/user_click.yaml
+++ b/oomagent/test/data/user_click.yaml
@@ -1,9 +1,10 @@
 kind: Entity
 name: user
-description: 'user id'
+description: "user id"
 groups:
   - name: user-click
     category: stream
+    snapshot-interval: 24h
     description: user click post feature
     features:
       - name: last_5_click_posts
diff --git a/oomagent/test/test_export_to_file.sh b/oomagent/test/test_export_to_file.sh
index 86d22f30..1feffe99 100755
--- a/oomagent/test/test_export_to_file.sh
+++ b/oomagent/test/test_export_to_file.sh
@@ -36,7 +36,7 @@ arg=$(cat <<-EOF
 {
     "features": ["account.state"],
     "unix_milli": $unix_milli,
-    "output_file": "$output"
+    "output_file_path": "$output"
 }
 EOF
 )
@@ -62,7 +62,7 @@ group="export all features"
 arg=$(cat <<-EOF
 {
     "features": ["account.state","account.credit_score","account.account_age_days","account.has_2fa_installed"],
-    "output_file": "$output",
+    "output_file_path": "$output",
     "unix_milli": $unix_milli
 }
 EOF
diff --git a/oomagent/test/test_import_from_file.sh b/oomagent/test/test_import_from_file.sh
index 067e3049..08c64e01 100755
--- a/oomagent/test/test_import_from_file.sh
+++ b/oomagent/test/test_import_from_file.sh
@@ -9,7 +9,7 @@ case="api returns ok"
 arg=$(cat <<-EOF
 {
     "group": "account",
-    "input_file": "./data/account_10.csv",
+    "input_file_path": "./data/account_10.csv",
     "delimiter": ","
 }
 EOF
diff --git a/oomagent/test/test_join_using_file.sh b/oomagent/test/test_join_using_file.sh
index 1bce9578..adad985d 100755
--- a/oomagent/test/test_join_using_file.sh
+++ b/oomagent/test/test_join_using_file.sh
@@ -25,8 +25,8 @@ arg=$(cat <<-EOF
     "driver_stats.acc_rate",
     "driver_stats.avg_daily_trips"
   ],
-  "input_file": "./data/driver_stats_label.csv",
-  "output_file": "$output"
+  "input_file_path": "./data/driver_stats_label.csv",
+  "output_file_path": "$output"
 }
 EOF
 )
diff --git a/oomcli/test/test_apply_entity.sh b/oomcli/test/test_apply_entity.sh
index 61bb744c..47d69b4e 100755
--- a/oomcli/test/test_apply_entity.sh
+++ b/oomcli/test/test_apply_entity.sh
@@ -30,6 +30,10 @@ groups:
   - name: gender
     value-type: int64
     description: 'description'
+- name: user-click
+  category: stream
+  snapshot-interval: 24h
+  description: user click post feature
 EOF
 
     entity_expected='
@@ -40,12 +44,13 @@ ID,NAME,DESCRIPTION
     assert_eq "apply_single_complex_entity: check entity" "$(sort <<< "$entity_expected")" "$(sort <<< "$entity_actual")"
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,device,user,batch,a description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
-2,user,user,batch,a description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,device,user,batch,0s,a description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+2,user,user,batch,0s,a description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+3,user-click,user,stream,24h0m0s,user click post feature
 '
     group_actual=$(oomcli get meta group -o csv --wide)
-    filter() { cut -d ',' -f 1-5 <<<"$1"; }
+    filter() { cut -d ',' -f 1-6 <<<"$1"; }
     assert_eq "apply_single_complex_entity: check group" "$(filter "$group_expected" | sort)" "$(filter "$group_actual" | sort)"
 
     feature_expected='
@@ -91,11 +96,11 @@ ID,NAME,DESCRIPTION
     assert_eq "apply_multiple_files_of_entity: oomcli get meta entity" "$entity_expected" "$entity_actual"
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,student,user,batch,student feature group,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,student,user,batch,0s,student feature group,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
 '
     group_actual=$(oomcli get meta group -o csv --wide)
-    filter() { cut -d ',' -f 1-5 <<<"$1"; }
+    filter() { cut -d ',' -f 1-6 <<<"$1"; }
     assert_eq "oapply_multiple_files_of_entity: check group" "$(filter "$group_expected"| sort)" "$(filter "$group_actual" | sort)"
 }
 
@@ -156,10 +161,10 @@ ID,NAME,DESCRIPTION
       assert_eq "apply_entity_items: oomcli apply mutiple entity: check entity" "$(sort <<< "$entity_expected")" "$(sort <<< "$entity_actual")"
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION
-1,account,user,batch,user account info
-2,transaction_stats,user,batch,user transaction statistics
-3,phone,device,batch,phone info
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION
+1,account,user,batch,0s,user account info
+2,transaction_stats,user,batch,0s,user transaction statistics
+3,phone,device,batch,0s,phone info
 '
       group_actual=$(oomcli get meta group -o csv)
       assert_eq "apply_entity_items: oomcli apply multiple entity: check group" "$group_expected" "$group_actual"
diff --git a/oomcli/test/test_apply_group.sh b/oomcli/test/test_apply_group.sh
index ca59442f..2d18359a 100755
--- a/oomcli/test/test_apply_group.sh
+++ b/oomcli/test/test_apply_group.sh
@@ -30,8 +30,8 @@ features:
 EOF
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,device,user,batch,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,device,user,batch,0s,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
 '
     group_actual=$(oomcli get meta group -o csv --wide)
     filter() { cut -d ',' -f 1-5 <<<"$1"; }
@@ -66,12 +66,20 @@ name: account
 entity-name: user
 category: batch
 description: 'description'
+---
+kind: Group
+name: user-click
+entity-name: user
+category: stream
+snapshot-interval: 24h
+description: user click post feature
 EOF
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,device,user,batch,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
-2,account,user,batch,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,device,user,batch,0s,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+2,account,user,batch,0s,description,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+3,user-click,user,stream,24h0m0s,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
 '
     group_actual=$(oomcli get meta group -o csv --wide)
     filter() { cut -d ',' -f 1-5 <<<"$1"; }
@@ -136,10 +144,10 @@ items:
 EOF
 
     group_expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION
-1,account,user,batch,user account info
-2,transaction_stats,user,batch,user transaction statistics
-3,phone,device,batch,phone info
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION
+1,account,user,batch,0s,user account info
+2,transaction_stats,user,batch,0s,user transaction statistics
+3,phone,device,batch,0s,phone info
 '
     group_actual=$(oomcli get meta group -o csv --wide)
     filter() { cut -d ',' -f 1-5 <<<"$1"; }
diff --git a/oomcli/test/test_get_meta_entity.sh b/oomcli/test/test_get_meta_entity.sh
index e1a71eb9..7e1189be 100755
--- a/oomcli/test/test_get_meta_entity.sh
+++ b/oomcli/test/test_get_meta_entity.sh
@@ -75,6 +75,7 @@ items:
               description: age
         - name: user-click
           category: stream
+          snapshot-interval: 24h0m0s
           description: user click post feature
           features:
             - name: last_5_click_posts
diff --git a/oomcli/test/test_get_meta_group.sh b/oomcli/test/test_get_meta_group.sh
index 751e85e5..c49865d8 100755
--- a/oomcli/test/test_get_meta_group.sh
+++ b/oomcli/test/test_get_meta_group.sh
@@ -7,27 +7,27 @@ init_store
 register_features
 
 case='oomcli get meta group works'
-expected='ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,phone,device,batch,phone,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
-2,student,user,batch,student
-3,user-click,user,stream,user click post feature
+expected='ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,phone,device,batch,0s,phone,<NULL>,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+2,student,user,batch,0s,student
+3,user-click,user,stream,24h0m0s,user click post feature
 '
 actual=$(oomcli get meta group -o csv --wide)
 ignore_time() { cut -d ',' -f 1-5 <<<"$1"; }
 assert_eq "$case" "$(ignore_time "$expected" | sort)" "$(ignore_time "$actual" | sort)"
 
 case='oomcli get simplified group works'
-expected='ID,NAME,ENTITY,CATEGORY,DESCRIPTION
-1,phone,device,batch,phone
-2,student,user,batch,student
-3,user-click,user,stream,user click post feature
+expected='ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION
+1,phone,device,batch,0s,phone
+2,student,user,batch,0s,student
+3,user-click,user,stream,24h0m0s,user click post feature
 '
 actual=$(oomcli get meta group -o csv)
 assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
 
 case='oomcli get one group works'
-expected='ID,NAME,ENTITY,CATEGORY,DESCRIPTION
-1,phone,device,batch,phone
+expected='ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION
+1,phone,device,batch,0s,phone
 '
 actual=$(oomcli get meta group phone -o csv)
 assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
@@ -79,6 +79,7 @@ items:
       name: user-click
       entity-name: user
       category: stream
+      snapshot-interval: 24h0m0s
       description: user click post feature
       features:
         - name: last_5_click_posts
diff --git a/oomcli/test/test_update_group.sh b/oomcli/test/test_update_group.sh
index df75681c..136ad122 100755
--- a/oomcli/test/test_update_group.sh
+++ b/oomcli/test/test_update_group.sh
@@ -9,8 +9,8 @@ import_sample > /dev/null
 case='oomcli update group works'
 oomcli update group phone --description "new description"
 expected='
-ID,NAME,ENTITY,CATEGORY,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
-1,phone,device,batch,new description,,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
+ID,NAME,ENTITY,CATEGORY,SNAPSHOT-INTERVAL,DESCRIPTION,ONLINE-REVISION-ID,CREATE-TIME,MODIFY-TIME
+1,phone,device,batch,0s,new description,,2021-11-30T07:51:03Z,2021-11-30T08:19:13Z
 '
 filter() { cut -d ',' -f 1-5 <<<"$1"; }
 actual=$(oomcli get meta group phone -o csv)
diff --git a/pkg/oomstore/apply_test.go b/pkg/oomstore/apply_test.go
index 0f6f94a0..df1af7e9 100644
--- a/pkg/oomstore/apply_test.go
+++ b/pkg/oomstore/apply_test.go
@@ -78,6 +78,13 @@ entity-name: user
 category: batch
 description: 'description'
 ---
+kind: Group
+name: user-click
+entity-name: user
+category: stream
+snapshot-interval: 86400
+description: 'description'
+---
 kind: Feature
 name: model
 group-name: device
@@ -114,6 +121,14 @@ description: 'description'
 						Category:    "batch",
 						Description: "description",
 					},
+					{
+						Kind:             "Group",
+						Name:             "user-click",
+						EntityName:       "user",
+						Category:         "stream",
+						SnapshotInterval: 86400,
+						Description:      "description",
+					},
 				},
 				NewFeatures: []apply.Feature{
 					{
@@ -208,6 +223,17 @@ groups:
   - name: gender
     value-type: int64
     description: 'description'
+- name: user-click
+  category: stream
+  snapshot-interval: 86400
+  description: description
+  features:
+  - name: last_5_click_posts
+    value-type: string
+    description: 'description'
+  - name: number_of_user_started_posts
+    value-type: int64
+    description: 'description'
 `)},
 			wantStage: &apply.ApplyStage{
 				NewEntities: []apply.Entity{
@@ -232,6 +258,14 @@ groups:
 						EntityName:  "user",
 						Description: "description",
 					},
+					{
+						Kind:             "Group",
+						Name:             "user-click",
+						EntityName:       "user",
+						Category:         "stream",
+						SnapshotInterval: 86400,
+						Description:      "description",
+					},
 				},
 				NewFeatures: []apply.Feature{
 					{
@@ -263,6 +297,20 @@ groups:
 						ValueType:   "int64",
 						Description: "description",
 					},
+					{
+						Kind:        "Feature",
+						Name:        "last_5_click_posts",
+						GroupName:   "user-click",
+						ValueType:   "string",
+						Description: "description",
+					},
+					{
+						Kind:        "Feature",
+						Name:        "number_of_user_started_posts",
+						GroupName:   "user-click",
+						ValueType:   "int64",
+						Description: "description",
+					},
 				},
 			},
 		},