add support sequence feature view get behavior table features

Author: lgqfhwy

api/model_feature_view_seq_config.go

@@ -24,5 +24,5 @@ type SeqConfig struct {
 	SeqEvent                  string `json:"seq_event"`
 	SeqLen                    int    `json:"seq_len"`
 	OnlineSeqName             string `json:"online_seq_name"`
-	OnlineBehaviorTableFields string `json:"online_behavior_table_fields"`
+	OnlineBehaviorTableFields []string `json:"online_behavior_table_fields"`
 }

dao/feature_view_dao.go

@@ -92,7 +92,7 @@ func makeSequenceFeatures(offlineSequences, onlineSequences []*sequenceInfo, seq
 
 	//produce seqeunce feature correspond to easyrec processor
 	sequencesValueMap := make(map[string][]string)
-	sequenceMap := make(map[string]bool, 0)
+	sequenceMap := make(map[string]bool)
 
 	for _, seq := range onlineSequences {
 		key := fmt.Sprintf("%s#%s", seq.itemId, seq.event)

dao/feature_view_featuredb_dao.go

@@ -24,7 +24,6 @@ import (
 	"github.com/aliyun/aliyun-pai-featurestore-go-sdk/v2/utils"
 	"github.com/expr-lang/expr"
 	"github.com/expr-lang/expr/vm"
-	flatbuffers "github.com/google/flatbuffers/go"
 )
 
 const (
@@ -54,6 +53,31 @@ type FeatureViewFeatureDBDao struct {
 	primaryKeyField string
 }
 
+func CntSkipBytes(innerReader *bytes.Reader, fieldType constants.FSType) int {
+	var skipBytes int = 0
+	switch fieldType {
+	case constants.FS_INT32:
+		skipBytes = 4
+	case constants.FS_INT64:
+		skipBytes = 8
+	case constants.FS_FLOAT:
+		skipBytes = 4
+	case constants.FS_DOUBLE:
+		skipBytes = 8
+	case constants.FS_STRING:
+		var length uint32
+		binary.Read(innerReader, binary.LittleEndian, &length)
+		skipBytes = int(length)
+	case constants.FS_BOOLEAN:
+		skipBytes = 1
+	default:
+		var length uint32
+		binary.Read(innerReader, binary.LittleEndian, &length)
+		skipBytes = int(length)
+	}
+	return skipBytes
+}
+
 func NewFeatureViewFeatureDBDao(config DaoConfig) *FeatureViewFeatureDBDao {
 	dao := FeatureViewFeatureDBDao{
 		database:        config.FeatureDBDatabaseName,
@@ -704,170 +728,32 @@ type FeatureDBBatchGetKKVRequest struct {
 	WithValue bool     `json:"with_value"`
 }
 
-type IndexPair struct {
-	index  int
-	fsType constants.FSType
-}
-
-
-func readValueFromKKVData(data *fdbserverfb.KKVData, offset int, fsType constants.FSType) (interface{}, error) {
-
-	tab := data.Table()
-
-	// ValueInt32 - 读取 int32 值
-	valueInt32 := func(i int) int32 {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+4 <= data.ValueLength() {
-			a := tab.Vector(o)
-			return tab.GetInt32(a + flatbuffers.UOffsetT(i))
-		}
-		return 0
-	}
-
-	// ValueInt64 - 读取 int64 值
-	valueInt64 := func(i int) int64 {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+8 <= data.ValueLength() {
-			a := tab.Vector(o)
-			return tab.GetInt64(a + flatbuffers.UOffsetT(i))
-		}
-		return 0
-	}
-
-	// ValueFloat32 - 读取 float32 值
-	valueFloat32 := func(i int) float32 {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+4 <= data.ValueLength() {
-			a := tab.Vector(o)
-			return tab.GetFloat32(a + flatbuffers.UOffsetT(i))
-		}
-		return 0.0
-	}
-
-	// ValueFloat64 - 读取 float64 值
-	valueFloat64 := func(i int) float64 {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+8 <= data.ValueLength() {
-			a := tab.Vector(o)
-			return tab.GetFloat64(a + flatbuffers.UOffsetT(i))
-		}
-		return 0.0
-	}
-
-	// ValueBool - 读取 bool 值
-	valueBool := func(i int) bool {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+1 <= data.ValueLength() {
-			a := tab.Vector(o)
-			return tab.GetBool(a + flatbuffers.UOffsetT(i))
-		}
-		return false
-	}
-
-	// ValueString - 读取 string 值
-	valueString := func(i int) string {
-		o := flatbuffers.UOffsetT(tab.Offset(12))
-		if o != 0 && i+4 <= data.ValueLength() {
-			a := tab.Vector(o)
-			// 读取长度（uint32，4字节）
-			length := tab.GetUint32(a + flatbuffers.UOffsetT(i))
-
-			// 检查是否有足够的字节用于字符串数据
-			if i+4+int(length) <= data.ValueLength() {
-				strBytes := make([]byte, length)
-				for j := 0; j < int(length); j++ {
-					strBytes[j] = tab.GetByte(a + flatbuffers.UOffsetT(i+4+j))
-				}
-				return string(strBytes)
-			}
-		}
-		return ""
-	}
-
-	switch fsType {
-	case constants.FS_INT32:
-		if offset+4 <= data.ValueLength() {
-			value := valueInt32(offset)
-			return value, nil
-		}
-		return int32(0), fmt.Errorf("insufficient bytes for int32")
-
-	case constants.FS_INT64:
-		if offset+8 <= data.ValueLength() {
-			value := valueInt64(offset)
-			return value, nil
-		}
-		return int64(0), fmt.Errorf("insufficient bytes for int64")
-
-	case constants.FS_FLOAT:
-		if offset+4 <= data.ValueLength() {
-			value := valueFloat32(offset)
-			return value, nil
-		}
-		return float32(0), fmt.Errorf("insufficient bytes for float32")
-
-	case constants.FS_DOUBLE:
-		if offset+8 <= data.ValueLength() {
-			value := valueFloat64(offset)
-			return value, nil
-		}
-		return float64(0), fmt.Errorf("insufficient bytes for float64")
-
-	case constants.FS_BOOLEAN:
-		if offset+1 <= data.ValueLength() {
-			value := valueBool(offset)
-			return value, nil
-		}
-		return false, fmt.Errorf("insufficient bytes for boolean")
-
-	case constants.FS_STRING:
-		if offset+4 <= data.ValueLength() {
-			value := valueString(offset)
-			return value, nil
-		}
-		return "", fmt.Errorf("insufficient bytes for string")
-
-	default:
-		// 对于未知类型，尝试按字符串处理
-		if offset+4 <= data.ValueLength() {
-			value := valueString(offset)
-			return value, nil
-		}
-		return "", fmt.Errorf("unknown type or insufficient bytes")
-	}
-}
-
 func (d *FeatureViewFeatureDBDao) GetUserSequenceFeature(keys []interface{}, userIdField string, sequenceConfig api.FeatureViewSeqConfig, onlineConfig []*api.SeqConfig) ([]map[string]interface{}, error) {
 	currTime := time.Now().Unix()
 	sequencePlayTimeMap := makePlayTimeMap(sequenceConfig.PlayTimeFilter)
 
 	seqConfigsMap := make(map[string][]*api.SeqConfig)
 
-	featureViewFieldsIndexMap := make(map[string]int)
-	featureViewFieldsTypeMap := make(map[string]constants.FSType)
-
-	seqConfigsBehaviorFieldsIndexTypeMap := make(map[string][][]IndexPair)
+	seqConfigsBehaviorFieldsMap := make(map[string][]map[string]struct{})
 
-	for i, field := range d.fields {
-		featureViewFieldsIndexMap[field] = i
-	}
-	for field, curType := range d.fieldTypeMap {
-		featureViewFieldsTypeMap[field] = curType
-	}
+	withValue := false
 	for _, seqConfig := range onlineConfig {
 		mapKey := fmt.Sprintf("%s:%d", seqConfig.SeqEvent, seqConfig.SeqLen)
 		seqConfigsMap[mapKey] = append(seqConfigsMap[mapKey], seqConfig)
-		curBehaviorFields := strings.Split(seqConfig.OnlineBehaviorTableFields, "\u0001")
-		curBehaviorFieldsIndexTypeList := make([]IndexPair, len(curBehaviorFields))
+		curBehaviorFields := seqConfig.OnlineBehaviorTableFields
+		curBehaviorFieldsMap := make(map[string]struct{})
 		for _, field := range curBehaviorFields {
-			curBehaviorFieldsIndexTypeList = append(curBehaviorFieldsIndexTypeList, IndexPair{index: featureViewFieldsIndexMap[field], fsType: featureViewFieldsTypeMap[field]})
+			curBehaviorFieldsMap[field] = struct{}{}
+		}
+		if len(curBehaviorFieldsMap) > 0 {
+			withValue = true
+			seqConfigsBehaviorFieldsMap[mapKey] = append(seqConfigsBehaviorFieldsMap[mapKey], curBehaviorFieldsMap)
 		}
-		seqConfigsBehaviorFieldsIndexTypeMap[mapKey] = append(seqConfigsBehaviorFieldsIndexTypeMap[mapKey], curBehaviorFieldsIndexTypeList)
 	}
 
 	errChan := make(chan error, len(keys)*len(onlineConfig))
 
-	fetchDataFunc := func(seqEvent string, seqLen int, key interface{}, seqConfigsBehaviorFields []IndexPair) []*sequenceInfo {
+	fetchDataFunc := func(seqEvent string, seqLen int, key interface{}, selectBehaviorFieldsSet map[string]struct{}) []*sequenceInfo {
 		sequences := []*sequenceInfo{}
 
 		events := strings.Split(seqEvent, "|")
@@ -876,8 +762,9 @@ func (d *FeatureViewFeatureDBDao) GetUserSequenceFeature(keys []interface{}, use
 			pks = append(pks, fmt.Sprintf("%v\u001D%s", key, event))
 		}
 		request := FeatureDBBatchGetKKVRequest{
-			PKs:    pks,
-			Length: seqLen,
+			PKs:       pks,
+			Length:    seqLen,
+			WithValue: withValue,
 		}
 		body, _ := json.Marshal(request)
 		url := fmt.Sprintf("%s/api/v1/tables/%s/%s/%s/batch_get_kkv", d.featureDBClient.GetCurrentAddress(false), d.database, d.schema, d.table)
@@ -978,19 +865,89 @@ func (d *FeatureViewFeatureDBDao) GetUserSequenceFeature(keys []interface{}, use
 						continue
 					}
 				}
+				dataBytes := kkv.ValueBytes()
+				if len(dataBytes) < 2 {
+					sequences = append(sequences, seq)
+					continue
+				}
+				innerReader.Reset(dataBytes)
+				// 读取版本号
+				var protocalVersion, ifNullFlagVersion uint8
+				binary.Read(innerReader, binary.LittleEndian, &protocalVersion)
+				binary.Read(innerReader, binary.LittleEndian, &ifNullFlagVersion)
+				readFeatureDBFunc_F_1 := func() (map[string]string, error) {
+					properties := make(map[string]string)
 
-				for _, field := range seqConfigsBehaviorFields {
-					// 将值转换为字符串存储在 seq.onlineBehaviourTableFieldsMap 中
-					if d.fields != nil && field.index < len(d.fields) {
-						fieldName := d.fields[field.index]
-						rawValue, err := readValueFromKKVData(kkv, field.index, field.fsType)
-						value := utils.ToString(rawValue, "")
-						if err != nil {
-							value = ""
+					for _, field := range d.fields {
+						var isNull uint8
+						if err := binary.Read(innerReader, binary.LittleEndian, &isNull); err != nil {
+							if err == io.EOF {
+								break
+							}
+							return nil, err
+						}
+						if isNull == 1 {
+							// 跳过空值
+							continue
+						}
+
+						if _, exists := selectBehaviorFieldsSet[field]; exists {
+							switch d.fieldTypeMap[field] {
+							case constants.FS_INT32:
+								var int32Value int32
+								binary.Read(innerReader, binary.LittleEndian, &int32Value)
+								properties[field] = fmt.Sprintf("%d", int32Value)
+							case constants.FS_INT64:
+								var int64Value int64
+								binary.Read(innerReader, binary.LittleEndian, &int64Value)
+								properties[field] = fmt.Sprintf("%d", int64Value)
+							case constants.FS_FLOAT:
+								var float32Value float32
+								binary.Read(innerReader, binary.LittleEndian, &float32Value)
+								properties[field] = fmt.Sprintf("%v", float32Value)
+							case constants.FS_DOUBLE:
+								var float64Value float64
+								binary.Read(innerReader, binary.LittleEndian, &float64Value)
+								properties[field] = fmt.Sprintf("%v", float64Value)
+							case constants.FS_STRING:
+								var length uint32
+								binary.Read(innerReader, binary.LittleEndian, &length)
+								strBytes := make([]byte, length)
+								binary.Read(innerReader, binary.LittleEndian, &strBytes)
+								properties[field] = string(strBytes)
+							case constants.FS_BOOLEAN:
+								var boolValue bool
+								binary.Read(innerReader, binary.LittleEndian, &boolValue)
+								properties[field] = fmt.Sprintf("%v", boolValue)
+							default:
+								var length uint32
+								binary.Read(innerReader, binary.LittleEndian, &length)
+								strBytes := make([]byte, length)
+								binary.Read(innerReader, binary.LittleEndian, &strBytes)
+								properties[field] = string(strBytes)
+							}
+						} else {
+							skipBytes := CntSkipBytes(innerReader, d.fieldTypeMap[field])
+							if skipBytes > 0 {
+								if _, err := innerReader.Seek(int64(skipBytes), io.SeekCurrent); err != nil {
+									return nil, err
+								}
+							}
 						}
-						seq.onlineBehaviourTableFieldsMap[fieldName] = value
 					}
+					return properties, nil
+				}
 
+				if protocalVersion == FeatureDB_Protocal_Version_F && ifNullFlagVersion == FeatureDB_IfNull_Flag_Version_1 {
+					readResult, err := readFeatureDBFunc_F_1()
+					if err != nil {
+						errChan <- err
+						return nil
+					}
+					seq.onlineBehaviourTableFieldsMap = readResult
+				} else {
+					errChan <- fmt.Errorf("unsupported protocal version: %d, ifNullFlagVersion: %d", protocalVersion, ifNullFlagVersion)
+					continue
 				}
 				sequences = append(sequences, seq)
 			}
@@ -1015,8 +972,8 @@ func (d *FeatureViewFeatureDBDao) GetUserSequenceFeature(keys []interface{}, use
 				if len(seqConfigs) == 0 {
 					continue
 				}
-				seqConfigsBehaviorFields := make([][]IndexPair, len(seqConfigs))
-				if curFields, exits := seqConfigsBehaviorFieldsIndexTypeMap[k]; exits {
+				seqConfigsBehaviorFields := make([]map[string]struct{}, len(seqConfigs))
+				if curFields, exits := seqConfigsBehaviorFieldsMap[k]; exits {
 					seqConfigsBehaviorFields = curFields
 				}
 				eventWg.Add(1)
@@ -1255,28 +1212,7 @@ func (d *FeatureViewFeatureDBDao) GetUserBehaviorFeature(userIds []interface{},
 								properties[field] = string(strBytes)
 							}
 						} else {
-							var skipBytes int = 0
-							switch d.fieldTypeMap[field] {
-							case constants.FS_INT32:
-								skipBytes = 4
-							case constants.FS_INT64:
-								skipBytes = 8
-							case constants.FS_FLOAT:
-								skipBytes = 4
-							case constants.FS_DOUBLE:
-								skipBytes = 8
-							case constants.FS_STRING:
-								var length uint32
-								binary.Read(innerReader, binary.LittleEndian, &length)
-								skipBytes = int(length)
-							case constants.FS_BOOLEAN:
-								skipBytes = 1
-							default:
-								var length uint32
-								binary.Read(innerReader, binary.LittleEndian, &length)
-								skipBytes = int(length)
-							}
-
+							skipBytes := CntSkipBytes(innerReader, d.fieldTypeMap[field])
 							if skipBytes > 0 {
 								if _, err := innerReader.Seek(int64(skipBytes), io.SeekCurrent); err != nil {
 									return nil, err