Repository: oom-ai/oomstore
Pull request #996 (2022-01-17 09:55:02)

Jinghan/modify offline method `CreateTable`
This PR refactors offline method `CreateTable`: replace `IsCDC` by `TableType`, so that we could differentiate different table types clearly.

related to #994 


Patch:
diff --git a/internal/database/dbutil/schema.go b/internal/database/dbutil/schema.go
index 0999940e..06f55054 100644
--- a/internal/database/dbutil/schema.go
+++ b/internal/database/dbutil/schema.go
@@ -26,12 +26,12 @@ func (c ColumnList) Names() []string {
 func BuildTableSchema(
 	tableName string,
 	entity *types.Entity,
-	isCDC bool,
+	hasUnixMilli bool,
 	features types.FeatureList,
 	pkFields []string,
 	backend types.BackendType,
 ) string {
-	columns := parseColumns(entity, isCDC, features, backend)
+	columns := parseColumns(entity, hasUnixMilli, features, backend)
 	return createTableDDL(tableName, columns, pkFields, backend)
 }
 
@@ -82,7 +82,7 @@ func BuildIndexDDL(tableName string, indexName string, fields []string, backend
 	}
 }
 
-func parseColumns(entity *types.Entity, isCDC bool, features types.FeatureList, backend types.BackendType) (rs []Column) {
+func parseColumns(entity *types.Entity, hasUnixMilli bool, features types.FeatureList, backend types.BackendType) (rs []Column) {
 	// entity column
 	{
 		c := Column{Name: entity.Name, ValueType: types.String}
@@ -101,7 +101,7 @@ func parseColumns(entity *types.Entity, isCDC bool, features types.FeatureList,
 
 	// unix_milli column
 	{
-		if isCDC {
+		if hasUnixMilli {
 			valueType := types.Int64
 			dbType, err := DBValueType(backend, valueType)
 			if err != nil {
diff --git a/internal/database/offline/bigquery/store.go b/internal/database/offline/bigquery/store.go
index b00e933b..8e251b1c 100644
--- a/internal/database/offline/bigquery/store.go
+++ b/internal/database/offline/bigquery/store.go
@@ -83,7 +83,17 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 }
 
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
-	schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, opt.IsCDC, opt.Features, nil, Backend)
+	var hasUnixMilli bool
+	switch opt.TableType {
+	case types.TableBatchSnapshot:
+		hasUnixMilli = false
+	case types.TableStreamSnapshot, types.TableStreamCdc:
+		hasUnixMilli = true
+	default:
+		panic(fmt.Sprintf("unsupported table type %s", opt.TableType))
+	}
+
+	schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, hasUnixMilli, opt.Features, nil, Backend)
 	if _, err := db.Client.Query(schema).Run(ctx); err != nil {
 		return errdefs.WithStack(err)
 	}
diff --git a/internal/database/offline/redshift/store.go b/internal/database/offline/redshift/store.go
index cf20c5d6..41dd4a06 100644
--- a/internal/database/offline/redshift/store.go
+++ b/internal/database/offline/redshift/store.go
@@ -2,6 +2,7 @@ package redshift
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -59,7 +60,21 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 }
 
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
-	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
+	var hasUnixMilli bool
+	switch opt.TableType {
+	case types.TableBatchSnapshot:
+		hasUnixMilli = false
+	case types.TableStreamSnapshot, types.TableStreamCdc:
+		hasUnixMilli = true
+	default:
+		panic(fmt.Sprintf("unsupported table type %s", opt.TableType))
+	}
+
+	schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, hasUnixMilli, opt.Features, nil, Backend)
+	if _, err := db.ExecContext(ctx, schema); err != nil {
+		return errdefs.WithStack(err)
+	}
+	return nil
 }
 
 func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
diff --git a/internal/database/offline/sqlutil/create_table.go b/internal/database/offline/sqlutil/create_table.go
index c9528776..e1ccd8d7 100644
--- a/internal/database/offline/sqlutil/create_table.go
+++ b/internal/database/offline/sqlutil/create_table.go
@@ -2,6 +2,7 @@ package sqlutil
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -12,24 +13,34 @@ import (
 )
 
 func CreateTable(ctx context.Context, db *sqlx.DB, opt offline.CreateTableOpt, backend types.BackendType) error {
-	if opt.IsCDC {
-		// Create index (entity_key, unix_milli) on cdc table
+	switch opt.TableType {
+	case types.TableBatchSnapshot:
+		// Create primary key (entity_key) on batch snapshot table
+		pkFields := []string{opt.Entity.Name}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, false, opt.Features, pkFields, backend)
+		if _, err := db.ExecContext(ctx, schema); err != nil {
+			return errdefs.WithStack(err)
+		}
+	case types.TableStreamSnapshot:
+		// Create primary key (entity_key) on stream snapshot table
+		pkFields := []string{opt.Entity.Name}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, true, opt.Features, pkFields, backend)
+		if _, err := db.ExecContext(ctx, schema); err != nil {
+			return errdefs.WithStack(err)
+		}
+	case types.TableStreamCdc:
 		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, true, opt.Features, nil, backend)
 		if _, err := db.ExecContext(ctx, schema); err != nil {
 			return errdefs.WithStack(err)
 		}
+		// Create index (entity_key, unix_milli) on stream cdc table
 		indexFields := []string{opt.Entity.Name, "unix_milli"}
 		indexDDL := dbutil.BuildIndexDDL(opt.TableName, "idx", indexFields, backend)
 		if _, err := db.ExecContext(ctx, indexDDL); err != nil {
 			return errdefs.WithStack(err)
 		}
-	} else {
-		// Create primary key (entity_key) on snapshot table
-		pkFields := []string{opt.Entity.Name}
-		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, false, opt.Features, pkFields, backend)
-		if _, err := db.ExecContext(ctx, schema); err != nil {
-			return errdefs.WithStack(err)
-		}
+	default:
+		panic(fmt.Sprintf("unsupported table type %s", opt.TableType))
 	}
 	return nil
 }
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index d97883cd..730026b7 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -74,5 +74,5 @@ type CreateTableOpt struct {
 	TableName string
 	Entity    *types.Entity
 	Features  types.FeatureList
-	IsCDC     bool
+	TableType types.TableType
 }
diff --git a/oomagent/codegen/oomagent.pb.go b/oomagent/codegen/oomagent.pb.go
index 9cee979f..c802d1bc 100644
--- a/oomagent/codegen/oomagent.pb.go
+++ b/oomagent/codegen/oomagent.pb.go
@@ -987,11 +987,9 @@ type ChannelJoinRequest struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// only takes effect (and required) on the first request
-	JoinFeatures []string `protobuf:"bytes,1,rep,name=join_features,json=joinFeatures,proto3" json:"join_features,omitempty"`
-	// only takes effect (and required) on the first request
-	ExistedFeatures []string   `protobuf:"bytes,2,rep,name=existed_features,json=existedFeatures,proto3" json:"existed_features,omitempty"`
-	EntityRow       *EntityRow `protobuf:"bytes,3,opt,name=entity_row,json=entityRow,proto3" json:"entity_row,omitempty"`
+	JoinFeatures    []string   `protobuf:"bytes,1,rep,name=join_features,json=joinFeatures,proto3" json:"join_features,omitempty"`
+	ExistedFeatures []string   `protobuf:"bytes,3,rep,name=existed_features,json=existedFeatures,proto3" json:"existed_features,omitempty"`
+	EntityRow       *EntityRow `protobuf:"bytes,2,opt,name=entity_row,json=entityRow,proto3" json:"entity_row,omitempty"`
 }
 
 func (x *ChannelJoinRequest) Reset() {
@@ -1052,7 +1050,6 @@ type ChannelJoinResponse struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	// only appears in the first request
 	Header    []string `protobuf:"bytes,1,rep,name=header,proto3" json:"header,omitempty"`
 	JoinedRow []*Value `protobuf:"bytes,2,rep,name=joined_row,json=joinedRow,proto3" json:"joined_row,omitempty"`
 }
@@ -1554,9 +1551,9 @@ var file_oomagent_proto_rawDesc = []byte{
 	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,
+	0x18, 0x03, 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,
+	0x79, 0x5f, 0x72, 0x6f, 0x77, 0x18, 0x02, 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,
diff --git a/pkg/oomstore/cache.go b/pkg/oomstore/cache.go
index b96c2e85..563c2889 100644
--- a/pkg/oomstore/cache.go
+++ b/pkg/oomstore/cache.go
@@ -184,7 +184,7 @@ func (p *StreamPushProcessor) newRevision(ctx context.Context, s *OomStore, grou
 		TableName: dbutil.OfflineStreamCdcTableName(groupID, revision),
 		Entity:    entity,
 		Features:  features,
-		IsCDC:     true,
+		TableType: types.TableStreamCdc,
 	}); err != nil {
 		return err
 	}
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index 0f35ff4e..e3ed967b 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -47,7 +47,7 @@ func (s *OomStore) createFirstSnapshotTable(ctx context.Context, revision *types
 		TableName: snapshotTable,
 		Entity:    revision.Group.Entity,
 		Features:  features,
-		IsCDC:     true, // In order to add `unix_milli` column, set IsCDC as true temporarily
+		TableType: types.TableStreamSnapshot,
 	}); err != nil {
 		return err
 	}
diff --git a/pkg/oomstore/types/types.go b/pkg/oomstore/types/types.go
index c116caf1..1348c05f 100644
--- a/pkg/oomstore/types/types.go
+++ b/pkg/oomstore/types/types.go
@@ -7,6 +7,14 @@ const (
 	CategoryStream Category = "stream"
 )
 
+type TableType = string
+
+const (
+	TableBatchSnapshot  TableType = "batch_snapshot"
+	TableStreamSnapshot TableType = "stream_snapshot"
+	TableStreamCdc      TableType = "stream_cdc"
+)
+
 type ExportRecord []interface{}
 
 func (r ExportRecord) EntityKey() string {
diff --git a/proto/oomagent.proto b/proto/oomagent.proto
index a70b9a3d..80b924d5 100644
--- a/proto/oomagent.proto
+++ b/proto/oomagent.proto
@@ -118,16 +118,12 @@ message EntityRow {
 }
 
 message ChannelJoinRequest {
-  // only takes effect (and required) on the first request
   repeated string join_features = 1;
-  // only takes effect (and required) on the first request
-  repeated string existed_features = 2;
-
-  EntityRow entity_row = 3;
+  repeated string existed_features = 3;
+  EntityRow entity_row = 2;
 }
 
 message ChannelJoinResponse {
-  // only appears in the first request
   repeated string header = 1;
   repeated Value joined_row = 2;
 }
diff --git a/sdk/python/src/oomstore/codegen/oomagent_pb2.py b/sdk/python/src/oomstore/codegen/oomagent_pb2.py
index 5fd02ba7..29b22d12 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\"\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'
+  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\x03 \x03(\t\x12\'\n\nentity_row\x18\x02 \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,])
 
@@ -794,14 +794,14 @@ _CHANNELJOINREQUEST = _descriptor.Descriptor(
       serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key),
     _descriptor.FieldDescriptor(
       name='existed_features', full_name='oomagent.ChannelJoinRequest.existed_features', index=1,
-      number=2, type=9, cpp_type=9, label=3,
+      number=3, type=9, cpp_type=9, label=3,
       has_default_value=False, default_value=[],
       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='entity_row', full_name='oomagent.ChannelJoinRequest.entity_row', index=2,
-      number=3, type=11, cpp_type=10, label=1,
+      number=2, type=11, cpp_type=10, label=1,
       has_default_value=False, default_value=None,
       message_type=None, enum_type=None, containing_type=None,
       is_extension=False, extension_scope=None,
diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml
index 868040c4..a351ab1a 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -10,10 +10,6 @@ prost = "0.9"
 thiserror = "1.0"
 tokio = { version = "1.15", features = ["macros", "rt-multi-thread"] }
 tonic = "0.6"
-futures-core = "0.3"
 
 [build-dependencies]
 tonic-build = "0.6"
-
-[dev-dependencies]
-futures-util = "0.3"
diff --git a/sdk/rust/examples/channel_join.rs b/sdk/rust/examples/channel_join.rs
deleted file mode 100644
index d121fb6c..00000000
--- a/sdk/rust/examples/channel_join.rs
+++ /dev/null
@@ -1,48 +0,0 @@
-use futures_util::{pin_mut, StreamExt};
-use oomstore::{oomagent::EntityRow, 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 join_features = svec![
-        "driver_stats.conv_rate",
-        "driver_stats.acc_rate",
-        "driver_stats.avg_daily_trips"
-    ];
-    let existed_features = svec![];
-
-    let entity_rows = vec![
-        EntityRow {
-            entity_key: "1".into(),
-            unix_milli: 3,
-            values:     Vec::new(),
-        },
-        EntityRow {
-            entity_key: "7".into(),
-            unix_milli: 1,
-            values:     Vec::new(),
-        },
-        EntityRow {
-            entity_key: "7".into(),
-            unix_milli: 0,
-            values:     Vec::new(),
-        },
-    ];
-
-    let (header, rows) = client
-        .channel_join(join_features, existed_features, entity_rows.into_iter())
-        .await?;
-
-    println!("RESPONSE HEADER={:?}", header);
-
-    pin_mut!(rows); // needed for iteration
-
-    while let Some(row) = rows.next().await {
-        println!("RESPONSE ROWS={:?}", row);
-    }
-
-    Ok(())
-}
diff --git a/sdk/rust/src/error.rs b/sdk/rust/src/error.rs
index 193da603..31d3624b 100644
--- a/sdk/rust/src/error.rs
+++ b/sdk/rust/src/error.rs
@@ -13,5 +13,5 @@ pub enum OomError {
     IntConversionError(#[from] num::TryFromIntError),
 
     #[error("unknown error")]
-    Unknown(String),
+    Unknown,
 }
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index ade3d823..a4053d8a 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -4,14 +4,11 @@ mod util;
 use std::{collections::HashMap, path::Path};
 
 use error::OomError;
-use futures_core::stream::Stream;
 use google::protobuf::Empty;
 use oomagent::{
     oom_agent_client::OomAgentClient,
     value,
     ChannelImportRequest,
-    ChannelJoinRequest,
-    ChannelJoinResponse,
     FeatureValueMap,
     ImportRequest,
     OnlineGetRequest,
@@ -22,8 +19,6 @@ use oomagent::{
 use tonic::{codegen::StdError, transport, Request};
 use util::parse_raw_feature_values;
 
-use crate::{oomagent::EntityRow, util::parse_raw_values};
-
 pub type FValue = value::Kind;
 
 type Result<T> = std::result::Result<T, OomError>;
@@ -122,12 +117,12 @@ impl Client {
         let mut group = group.into();
         let mut description = description.into();
         let mut revision = revision.into();
-        let inbound = async_stream::stream! {
+        let outbound = async_stream::stream! {
             for row in rows {
                 yield ChannelImportRequest{group: group.take(), description: description.take(), revision: revision.take(), row};
             }
         };
-        let res = self.inner.channel_import(Request::new(inbound)).await?.into_inner();
+        let res = self.inner.channel_import(Request::new(outbound)).await?.into_inner();
         Ok(res.revision_id as u32)
     }
 
@@ -176,42 +171,4 @@ impl Client {
             .into_inner();
         Ok(())
     }
-
-    pub async fn channel_join(
-        &mut self,
-        join_features: Vec<String>,
-        existed_features: Vec<String>,
-        entity_rows: impl Iterator<Item = EntityRow> + Send + 'static,
-    ) -> Result<(Vec<String>, impl Stream<Item = Result<Vec<value::Kind>>>)> {
-        let inbound = async_stream::stream! {
-            for (i, row) in entity_rows.enumerate() {
-                let (join_features, existed_features) = match i {
-                    0 => (join_features.clone(), existed_features.clone()),
-                    _ => (Vec::new(), Vec::new()),
-                };
-                yield ChannelJoinRequest {
-                    join_features,
-                    existed_features,
-                    entity_row: Some(row),
-                };
-            }
-        };
-
-        let mut outbound = self.inner.channel_join(Request::new(inbound)).await?.into_inner();
-
-        let ChannelJoinResponse { header, joined_row } = outbound
-            .message()
-            .await?
-            .ok_or_else(|| OomError::Unknown(String::from("stream finished with no response")))?;
-
-        let row = parse_raw_values(joined_row);
-
-        let outbound = async_stream::try_stream! {
-            yield row;
-            while let Some(ChannelJoinResponse { joined_row, .. }) = outbound.message().await? {
-                yield parse_raw_values(joined_row)
-            }
-        };
-        Ok((header, outbound))
-    }
 }
diff --git a/sdk/rust/src/util.rs b/sdk/rust/src/util.rs
index 60dd97f7..8ba7f855 100644
--- a/sdk/rust/src/util.rs
+++ b/sdk/rust/src/util.rs
@@ -1,16 +1,10 @@
-use crate::oomagent::{value, FeatureValueMap, Value};
+use crate::oomagent::{value, FeatureValueMap};
 use std::collections::HashMap;
 
-impl From<Value> for value::Kind {
-    fn from(value: Value) -> Self {
-        value.kind.expect("`oneof` in protobuffer should not be none")
-    }
-}
-
 pub fn parse_raw_feature_values(input: FeatureValueMap) -> HashMap<String, value::Kind> {
-    input.map.into_iter().map(|(k, v)| (k, v.into())).collect()
-}
-
-pub fn parse_raw_values(values: Vec<Value>) -> Vec<value::Kind> {
-    values.into_iter().map(|v| v.into()).collect()
+    input
+        .map
+        .into_iter()
+        .map(|(k, v)| (k, v.kind.expect("`oneof` should not be none")))
+        .collect()
 }


Test patch:
diff --git a/internal/database/offline/bigquery/store_test.go b/internal/database/offline/bigquery/store_test.go
index f449d4d8..9e62a068 100644
--- a/internal/database/offline/bigquery/store_test.go
+++ b/internal/database/offline/bigquery/store_test.go
@@ -107,3 +107,7 @@ func TestSnapshot(t *testing.T) {
 	t.Skip()
 	test_impl.TestSnapshot(t, prepareStore, destroyStore(DATASET_ID))
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATASET_ID))
+}
diff --git a/internal/database/offline/mysql/store_test.go b/internal/database/offline/mysql/store_test.go
index 8ff1834f..3d738f69 100644
--- a/internal/database/offline/mysql/store_test.go
+++ b/internal/database/offline/mysql/store_test.go
@@ -66,3 +66,7 @@ func TestTableSchema(t *testing.T) {
 		}
 	})
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
+}
diff --git a/internal/database/offline/postgres/store_test.go b/internal/database/offline/postgres/store_test.go
index b1ed67ec..69486466 100644
--- a/internal/database/offline/postgres/store_test.go
+++ b/internal/database/offline/postgres/store_test.go
@@ -58,3 +58,7 @@ func TestTableSchema(t *testing.T) {
 		}
 	})
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
+}
diff --git a/internal/database/offline/redshift/store_test.go b/internal/database/offline/redshift/store_test.go
index 5b9d40ea..e9df60ad 100644
--- a/internal/database/offline/redshift/store_test.go
+++ b/internal/database/offline/redshift/store_test.go
@@ -102,3 +102,7 @@ func TestTableSchema(t *testing.T) {
 		}
 	})
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
+}
diff --git a/internal/database/offline/snowflake/store_test.go b/internal/database/offline/snowflake/store_test.go
index 28de74d0..8b56293e 100644
--- a/internal/database/offline/snowflake/store_test.go
+++ b/internal/database/offline/snowflake/store_test.go
@@ -105,3 +105,7 @@ func TestTableSchema(t *testing.T) {
 	}
 	require.ElementsMatch(t, expected.Fields, actual.Fields)
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
+}
diff --git a/internal/database/offline/sqlite/store_test.go b/internal/database/offline/sqlite/store_test.go
index d58a8b4a..fcade997 100644
--- a/internal/database/offline/sqlite/store_test.go
+++ b/internal/database/offline/sqlite/store_test.go
@@ -68,3 +68,7 @@ func TestTableSchema(t *testing.T) {
 		}
 	})
 }
+
+func TestCreateTable(t *testing.T) {
+	test_impl.TestCreateTable(t, prepareStore, destroyStore)
+}
diff --git a/internal/database/offline/test_impl/create_table.go b/internal/database/offline/test_impl/create_table.go
index 0058a10c..77eb4222 100644
--- a/internal/database/offline/test_impl/create_table.go
+++ b/internal/database/offline/test_impl/create_table.go
@@ -35,19 +35,28 @@ func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 		{
 			describtion: "cdc table (with unix_milli)",
 			opt: offline.CreateTableOpt{
-				TableName: tableName,
+				TableName: tableName + "_0",
 				Entity:    &entity,
 				Features:  features,
-				IsCDC:     true,
+				TableType: types.TableStreamCdc,
 			},
 		},
 		{
-			describtion: "snapshot table (without_milli)",
+			describtion: "stream snapshot table (with unix_milli)",
 			opt: offline.CreateTableOpt{
-				TableName: tableName,
+				TableName: tableName + "_1",
 				Entity:    &entity,
 				Features:  features,
-				IsCDC:     false,
+				TableType: types.TableStreamSnapshot,
+			},
+		},
+		{
+			describtion: "batch snapshot table (without unix_milli)",
+			opt: offline.CreateTableOpt{
+				TableName: tableName + "_2",
+				Entity:    &entity,
+				Features:  features,
+				TableType: types.TableBatchSnapshot,
 			},
 		},
 	}
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index ec5f1e9a..8c8437a8 100644
--- a/internal/database/offline/test_impl/export.go
+++ b/internal/database/offline/test_impl/export.go
@@ -2,7 +2,6 @@ package test_impl
 
 import (
 	"bufio"
-	"fmt"
 	"strings"
 	"testing"
 
@@ -100,9 +99,6 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 			if tc.expectedError != nil {
 				assert.EqualError(t, err, tc.expectedError.Error())
 			} else {
-				fmt.Println("expected: ", tc.expected)
-				fmt.Println("actual: ", values)
-
 				assert.ElementsMatch(t, tc.expected, values)
 				assert.NoError(t, err)
 				assert.NoError(t, result.CheckStreamError())
diff --git a/internal/database/offline/test_impl/push.go b/internal/database/offline/test_impl/push.go
index d1c7dcf3..cc6720e1 100644
--- a/internal/database/offline/test_impl/push.go
+++ b/internal/database/offline/test_impl/push.go
@@ -75,7 +75,7 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			TableName: dbutil.OfflineStreamCdcTableName(group.ID, revision),
 			Entity:    &entity,
 			Features:  features,
-			IsCDC:     true,
+			TableType: types.TableStreamCdc,
 		})
 		require.NoError(t, err)