Repository: oom-ai/oomstore
Pull request #997 (2022-01-17 10:07:09)

feat(sdk/rust): implement channel join api


Patch:
diff --git a/internal/database/offline/sqlutil/snapshot.go b/internal/database/offline/sqlutil/snapshot.go
index 7f70963b..f5c9ab90 100644
--- a/internal/database/offline/sqlutil/snapshot.go
+++ b/internal/database/offline/sqlutil/snapshot.go
@@ -15,10 +15,9 @@ import (
 )
 
 const SNAPSHOT_QUERY = `
-INSERT INTO {{ qt .CurrSnapshotTableName }} ({{ qt .EntityName }}, {{ qt .UnixMilli }}, {{ columnJoin .FeatureNames }})
+INSERT INTO {{ qt .CurrSnapshotTableName }} ({{ qt .EntityName }}, {{ columnJoin .FeatureNames }})
 SELECT
 	l.{{ qt .EntityName }},
-	{{ unixMilli .UnixMilli }},
 	{{ featureValue .FeatureNames }}
 FROM {{ qt .PrevSnapshotTableName }} AS l
 LEFT JOIN
@@ -39,7 +38,6 @@ ON l.{{ qt .EntityName }} = r.{{ qt .EntityName }}
 {{ .Union }}
 SELECT
 	r.{{ qt .EntityName }},
-	{{ unixMilli .UnixMilli }},
 	{{ featureValue .FeatureNames }}
 FROM
 (
@@ -71,7 +69,7 @@ func Snapshot(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.SnapshotOpt)
 		currCdcTableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, currCdcTableName)
 	}
 
-	schema := dbutil.BuildTableSchema(currSnapshotTableName, opt.Group.Entity, true, opt.Features, []string{opt.Group.Entity.Name}, dbOpt.Backend)
+	schema := dbutil.BuildTableSchema(currSnapshotTableName, opt.Group.Entity, false, opt.Features, []string{opt.Group.Entity.Name}, dbOpt.Backend)
 	if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 		return err
 	}
@@ -118,9 +116,6 @@ func buildSnapshotQuery(params snapshotQueryParams) (string, error) {
 		"columnJoin": func(columns []string) string {
 			return qt(columns...)
 		},
-		"unixMilli": func(unixMilli string) string {
-			return fmt.Sprintf("(CASE WHEN r.%s IS NULL THEN l.%s ELSE r.%s END) AS %s", qt(params.FeatureNames[0]), qt(unixMilli), qt(unixMilli), unixMilli)
-		},
 		"featureValue": func(features []string) string {
 			values := make([]string, 0, len(features))
 			for _, f := range features {
diff --git a/oomagent/codegen/oomagent.pb.go b/oomagent/codegen/oomagent.pb.go
index c802d1bc..9cee979f 100644
--- a/oomagent/codegen/oomagent.pb.go
+++ b/oomagent/codegen/oomagent.pb.go
@@ -987,9 +987,11 @@ type ChannelJoinRequest struct {
 	sizeCache     protoimpl.SizeCache
 	unknownFields protoimpl.UnknownFields
 
-	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"`
+	// 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"`
 }
 
 func (x *ChannelJoinRequest) Reset() {
@@ -1050,6 +1052,7 @@ 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"`
 }
@@ -1551,9 +1554,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, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x65, 0x78, 0x69, 0x73, 0x74, 0x65, 0x64, 0x46,
+	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, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x13, 0x2e, 0x6f, 0x6f,
+	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,
diff --git a/pkg/oomstore/cache.go b/pkg/oomstore/cache.go
index b96c2e85..fe936c32 100644
--- a/pkg/oomstore/cache.go
+++ b/pkg/oomstore/cache.go
@@ -14,12 +14,12 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-var defaultStreamPushProcessorCfg = types.StreamPushProcessorConfig{
-	RevisionInterval: 24 * time.Hour,
-	Period:           5 * time.Minute,
-	MinPeriod:        2 * time.Minute,
-	BufferSize:       1000,
-}
+const (
+	Day       = 24 * time.Hour
+	Capacity  = 1000
+	Period    = 5 * time.Minute
+	MinPeriod = 2 * time.Minute
+)
 
 type GroupBuffer struct {
 	groupID  int
@@ -28,30 +28,26 @@ type GroupBuffer struct {
 }
 
 type StreamPushProcessor struct {
-	bufferSize      int
-	minPeriod       time.Duration
-	revisionInteval time.Duration
-	notifyQuit      chan struct{}
-	waitQuit        chan struct{}
+	capacity   int
+	minPeriod  time.Duration
+	notifyQuit chan struct{}
+	waitQuit   chan struct{}
 
 	ticker *time.Ticker
 	ch     chan types.StreamRecord
 	buffer sync.Map // GroupID -> GroupBuffer
 }
 
-func (s *OomStore) InitStreamPushProcessor(ctx context.Context, cfg *types.StreamPushProcessorConfig) {
-	if cfg == nil {
-		cfg = &defaultStreamPushProcessorCfg
-	}
+func (s *OomStore) InitStreamPushProcessor(ctx context.Context) {
 	processor := &StreamPushProcessor{
-		bufferSize:      cfg.BufferSize,
-		minPeriod:       cfg.MinPeriod,
-		revisionInteval: cfg.RevisionInterval,
-		notifyQuit:      make(chan struct{}),
-		waitQuit:        make(chan struct{}),
+		// TODO: make Capacity, Period, MinPeriod configurable
+		capacity:   Capacity,
+		minPeriod:  MinPeriod,
+		notifyQuit: make(chan struct{}),
+		waitQuit:   make(chan struct{}),
 
 		ch:     make(chan types.StreamRecord),
-		ticker: time.NewTicker(cfg.Period),
+		ticker: time.NewTicker(Period),
 	}
 	s.streamPushProcessor = processor
 
@@ -93,7 +89,7 @@ func (s *OomStore) InitStreamPushProcessor(ctx context.Context, cfg *types.Strea
 				b.records = append(b.records, record)
 				processor.buffer.Store(groupID, b)
 
-				if len(b.records) >= processor.bufferSize {
+				if len(b.records) >= processor.capacity {
 					if err := processor.pushToOffline(ctx, s, groupID); err != nil {
 						fmt.Fprintf(os.Stderr, "Error pushing to offline store: %+v", err)
 					}
@@ -115,7 +111,7 @@ func (p *StreamPushProcessor) Push(record types.StreamRecord) {
 	if _, ok := p.buffer.Load(record.GroupID); !ok {
 		p.buffer.Store(record.GroupID, GroupBuffer{
 			groupID: record.GroupID,
-			records: make([]types.StreamRecord, 0, p.bufferSize),
+			records: make([]types.StreamRecord, 0, p.capacity),
 		})
 	}
 	p.ch <- record
@@ -135,7 +131,7 @@ func (p *StreamPushProcessor) pushToOffline(ctx context.Context, s *OomStore, gr
 
 	buckets := make(map[int64][]types.StreamRecord)
 	for _, record := range b.records {
-		revision := p.lastRevision(record.UnixMilli)
+		revision := latestRevision(record.UnixMilli)
 		if _, ok := buckets[revision]; !ok {
 			buckets[revision] = make([]types.StreamRecord, 0)
 		}
@@ -165,7 +161,7 @@ func (p *StreamPushProcessor) pushToOffline(ctx context.Context, s *OomStore, gr
 		}
 	}
 
-	b.records = make([]types.StreamRecord, 0, p.bufferSize)
+	b.records = make([]types.StreamRecord, 0, p.capacity)
 	b.lastPush = time.Now()
 	p.buffer.Store(groupID, b)
 	return err
@@ -202,6 +198,6 @@ func (p *StreamPushProcessor) newRevision(ctx context.Context, s *OomStore, grou
 	return nil
 }
 
-func (p *StreamPushProcessor) lastRevision(unixMill int64) int64 {
-	return (unixMill / p.revisionInteval.Milliseconds()) * p.revisionInteval.Milliseconds()
+func latestRevision(unixMilli int64) int64 {
+	return (unixMilli / Day.Milliseconds()) * Day.Milliseconds()
 }
diff --git a/pkg/oomstore/oomstore.go b/pkg/oomstore/oomstore.go
index fc00cda5..d0b74816 100644
--- a/pkg/oomstore/oomstore.go
+++ b/pkg/oomstore/oomstore.go
@@ -46,7 +46,7 @@ func Open(ctx context.Context, opt types.OomStoreConfig) (*OomStore, error) {
 		offline:  offlineStore,
 		metadata: metadataStore,
 	}
-	store.InitStreamPushProcessor(ctx, opt.StreamPushProcessor)
+	store.InitStreamPushProcessor(ctx)
 
 	return store, nil
 }
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index 0f35ff4e..6cc57eaa 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -47,7 +47,6 @@ 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
 	}); err != nil {
 		return err
 	}
diff --git a/pkg/oomstore/types/config.go b/pkg/oomstore/types/config.go
index e356d4f3..3b2ba188 100644
--- a/pkg/oomstore/types/config.go
+++ b/pkg/oomstore/types/config.go
@@ -26,15 +26,6 @@ type OomStoreConfig struct {
 	MetadataStore MetadataStoreConfig `yaml:"metadata-store"`
 	OfflineStore  OfflineStoreConfig  `yaml:"offline-store"`
 	OnlineStore   OnlineStoreConfig   `yaml:"online-store"`
-
-	StreamPushProcessor *StreamPushProcessorConfig `yaml:"stream-push-processor"`
-}
-
-type StreamPushProcessorConfig struct {
-	RevisionInterval time.Duration `yaml:"revision-interval"`
-	Period           time.Duration `yaml:"period"`
-	MinPeriod        time.Duration `yaml:"min-period"`
-	BufferSize       int           `yaml:"buffer-size"`
 }
 
 type OnlineStoreConfig struct {
diff --git a/proto/oomagent.proto b/proto/oomagent.proto
index 80b924d5..a70b9a3d 100644
--- a/proto/oomagent.proto
+++ b/proto/oomagent.proto
@@ -118,12 +118,16 @@ message EntityRow {
 }
 
 message ChannelJoinRequest {
+  // only takes effect (and required) on the first request
   repeated string join_features = 1;
-  repeated string existed_features = 3;
-  EntityRow entity_row = 2;
+  // only takes effect (and required) on the first request
+  repeated string existed_features = 2;
+
+  EntityRow entity_row = 3;
 }
 
 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 29b22d12..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\"\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'
+  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,])
 
@@ -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=3, type=9, cpp_type=9, label=3,
+      number=2, 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=2, type=11, cpp_type=10, label=1,
+      number=3, 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 a351ab1a..868040c4 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -10,6 +10,10 @@ 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
new file mode 100644
index 00000000..d121fb6c
--- /dev/null
+++ b/sdk/rust/examples/channel_join.rs
@@ -0,0 +1,48 @@
+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 31d3624b..193da603 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,
+    Unknown(String),
 }
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index a4053d8a..ade3d823 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -4,11 +4,14 @@ 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,
@@ -19,6 +22,8 @@ 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>;
@@ -117,12 +122,12 @@ impl Client {
         let mut group = group.into();
         let mut description = description.into();
         let mut revision = revision.into();
-        let outbound = async_stream::stream! {
+        let inbound = 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(outbound)).await?.into_inner();
+        let res = self.inner.channel_import(Request::new(inbound)).await?.into_inner();
         Ok(res.revision_id as u32)
     }
 
@@ -171,4 +176,42 @@ 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 8ba7f855..60dd97f7 100644
--- a/sdk/rust/src/util.rs
+++ b/sdk/rust/src/util.rs
@@ -1,10 +1,16 @@
-use crate::oomagent::{value, FeatureValueMap};
+use crate::oomagent::{value, FeatureValueMap, Value};
 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.kind.expect("`oneof` should not be none")))
-        .collect()
+    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()
 }


Test patch:
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index c03c98f7..933a86d7 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -147,13 +147,11 @@ func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.Feature
 	groupBasic := &types.Group{
 		ID:       1,
 		Name:     "device_basic",
-		Entity:   &types.Entity{Name: "device"},
 		Category: types.CategoryBatch,
 	}
 	groupAdvanced := &types.Group{
 		ID:       2,
 		Name:     "device_advanced",
-		Entity:   &types.Entity{Name: "device"},
 		Category: types.CategoryBatch,
 	}
 	price := &types.Feature{
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index fe15e0b3..3ef7ef93 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -20,18 +20,20 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
-	features, _ := prepareFeatures(true)
-	group := features[0].Group
+	group := &types.Group{
+		ID:     1,
+		Entity: &types.Entity{Name: "device"},
+	}
 	unixMilli := &types.Feature{
 		Name:      "unix_milli",
 		ValueType: types.Int64,
-		Group:     group,
 	}
+	features, _ := prepareFeatures(true)
 
-	buildTestSnapshotTable(ctx, t, store, append(features, unixMilli), 1, "offline_stream_snapshot_1_1", &offline.CSVSource{
-		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100,1
-1235,apple,200,5
-1236,oneplus,155,8
+	buildTestSnapshotTable(ctx, t, store, features, 1, "offline_stream_snapshot_1_1", &offline.CSVSource{
+		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
+1235,apple,200
+1236,oneplus,155
 `)),
 		Delimiter: ",",
 	})
@@ -57,7 +59,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	result, err := store.Export(ctx, offline.ExportOpt{
 		SnapshotTables: map[int]string{1: "offline_stream_snapshot_1_2"},
 		EntityName:     "device",
-		Features:       map[int]types.FeatureList{1: append(features, unixMilli)},
+		Features:       map[int]types.FeatureList{1: features},
 	})
 	values := make([][]interface{}, 0)
 	for row := range result.Data {
@@ -67,10 +69,10 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 		return cast.ToInt64(values[i][0]) < cast.ToInt64(values[j][0])
 	})
 	expected := [][]interface{}{
-		{"1234", "xiaomi-1", int64(130), int64(10)},
-		{"1235", "apple-1", int64(113), int64(15)},
-		{"1236", "oneplus", int64(155), int64(8)},
-		{"1237", "pixel", int64(200), int64(11)},
+		{"1234", "xiaomi-1", int64(130)},
+		{"1235", "apple-1", int64(113)},
+		{"1236", "oneplus", int64(155)},
+		{"1237", "pixel", int64(200)},
 	}
 	assert.Equal(t, expected, values)
 	assert.NoError(t, result.CheckStreamError())