func NewSequenceFeatureView()

in domain/sequence_feature_view.go [24:192]
151 lines of code
35 McCabe index (conditional complexity)

func NewSequenceFeatureView(view *api.FeatureView, p *Project, entity *FeatureEntity) *SequenceFeatureView {
	sequenceFeatureView := &SequenceFeatureView{
		FeatureView:   view,
		Project:       p,
		FeatureEntity: entity,
	}
	for _, field := range view.Fields {
		if field.IsPrimaryKey {
			sequenceFeatureView.userIdField = field.Name
			break
		}
	}

	err := json.Unmarshal([]byte(view.Config), &sequenceFeatureView.sequenceConfig)
	if err != nil {
		panic("sequence featureview config unmarshal failed")
	}

	if sequenceFeatureView.sequenceConfig.RegistrationMode == "" {
		sequenceFeatureView.sequenceConfig.RegistrationMode = constants.Seq_Registration_Mode_Full_Sequence
	}

	sequenceFeatureView.offline_2_online_seq_map = make(map[string]string, len(sequenceFeatureView.sequenceConfig.SeqConfig))
	for _, field := range view.Fields {
		if field.IsPartition {
			continue
		} else {
			sequenceFeatureView.behaviorFields = append(sequenceFeatureView.behaviorFields, field.Name)
		}
	}
	if sequenceFeatureView.sequenceConfig.RegistrationMode == constants.Seq_Registration_Mode_Full_Sequence {
		for _, seqConfig := range sequenceFeatureView.sequenceConfig.SeqConfig {
			sequenceFeatureView.offline_2_online_seq_map[seqConfig.OfflineSeqName] = seqConfig.OnlineSeqName
		}

		seen := make(map[string]bool)
		var uniqueSeqConfigs []*api.SeqConfig
		for _, seqConfig := range sequenceFeatureView.sequenceConfig.SeqConfig {
			if !seen[seqConfig.OnlineSeqName] {
				uniqueSeqConfigs = append(uniqueSeqConfigs, seqConfig)
				seen[seqConfig.OnlineSeqName] = true
			}
		}
		sequenceFeatureView.sequenceConfig.SeqConfig = uniqueSeqConfigs
	}

	requiredElements1 := []string{"user_id", "item_id", "event"}
	requiredElements2 := []string{"user_id", "item_id", "event", "timestamp"}
	if len(sequenceFeatureView.sequenceConfig.DeduplicationMethod) == len(requiredElements1) {
		for i, v := range sequenceFeatureView.sequenceConfig.DeduplicationMethod {
			if v != requiredElements1[i] {
				panic("deduplication_method invalid")
			}
		}
		sequenceFeatureView.sequenceConfig.DeduplicationMethodNum = 1
	} else if len(sequenceFeatureView.sequenceConfig.DeduplicationMethod) == len(requiredElements2) {
		for i, v := range sequenceFeatureView.sequenceConfig.DeduplicationMethod {
			if v != requiredElements2[i] {
				panic("deduplication_method invalid")
			}
		}
		sequenceFeatureView.sequenceConfig.DeduplicationMethodNum = 2
	} else {
		panic("deduplication_method invalid")
	}

	daoConfig := dao.DaoConfig{
		DatasourceType:  p.OnlineDatasourceType,
		PrimaryKeyField: sequenceFeatureView.userIdField,
	}

	if view.WriteToFeatureDB || p.OnlineDatasourceType == constants.Datasource_Type_FeatureDB {
		daoConfig.DatasourceType = constants.Datasource_Type_FeatureDB
		daoConfig.FeatureDBDatabaseName = p.InstanceId
		daoConfig.FeatureDBSchemaName = p.ProjectName
		if sequenceFeatureView.sequenceConfig.ReferencedFeatureViewId == 0 {
			daoConfig.FeatureDBTableName = sequenceFeatureView.Name
		} else {
			daoConfig.FeatureDBTableName = sequenceFeatureView.sequenceConfig.ReferencedFeatureViewName
		}
		daoConfig.FeatureDBSignature = p.Signature

		fieldTypeMap := make(map[string]constants.FSType, len(view.Fields))
		for _, field := range view.Fields {
			if field.IsPartition {
				continue
			} else {
				fieldTypeMap[field.Name] = field.Type
			}
		}
		daoConfig.FieldTypeMap = fieldTypeMap
		daoConfig.Fields = sequenceFeatureView.behaviorFields
	} else {
		if sequenceFeatureView.sequenceConfig.ReferencedFeatureViewId == 0 {
			switch p.OnlineDatasourceType {
			case constants.Datasource_Type_Hologres:
				daoConfig.HologresName = p.OnlineStore.GetDatasourceName()
				daoConfig.HologresOfflineTableName = p.OnlineStore.GetSeqOfflineTableName(sequenceFeatureView)
				daoConfig.HologresOnlineTableName = p.OnlineStore.GetSeqOnlineTableName(sequenceFeatureView)
			case constants.Datasource_Type_TableStore:
				daoConfig.TableStoreName = p.OnlineStore.GetDatasourceName()
				daoConfig.TableStoreOfflineTableName = p.OnlineStore.GetSeqOfflineTableName(sequenceFeatureView)
				daoConfig.TableStoreOnlineTableName = p.OnlineStore.GetSeqOnlineTableName(sequenceFeatureView)

			case constants.Datasource_Type_IGraph:
				daoConfig.SaveOriginalField = true
				daoConfig.IGraphName = p.OnlineStore.GetDatasourceName()
				daoConfig.GroupName = p.ProjectName
				daoConfig.IgraphEdgeName = p.OnlineStore.GetSeqOnlineTableName(sequenceFeatureView)

				fieldTypeMap := make(map[string]constants.FSType, len(view.Fields))
				for _, field := range view.Fields {
					if field.IsPartition {
						continue
					} else {
						fieldTypeMap[field.Name] = field.Type
					}
				}
				daoConfig.FieldTypeMap = fieldTypeMap

			default:

			}
		} else {
			referencedFeatureView := p.GetFeatureView(sequenceFeatureView.sequenceConfig.ReferencedFeatureViewName)
			if referencedFeatureView == nil {
				panic(fmt.Sprintf("referenced feature view :%s not found", sequenceFeatureView.sequenceConfig.ReferencedFeatureViewName))
			}
			if referencedFeatureView.GetType() != constants.Feature_View_Type_Sequence {
				panic(fmt.Sprintf("referenced feature view :%s is not sequence feature view", sequenceFeatureView.sequenceConfig.ReferencedFeatureViewName))
			}
			referencedSeqFeatureView := referencedFeatureView.(*SequenceFeatureView)
			switch p.OnlineDatasourceType {
			case constants.Datasource_Type_Hologres:
				daoConfig.HologresName = p.OnlineStore.GetDatasourceName()
				daoConfig.HologresOfflineTableName = p.OnlineStore.GetSeqOfflineTableName(referencedSeqFeatureView)
				daoConfig.HologresOnlineTableName = p.OnlineStore.GetSeqOnlineTableName(referencedSeqFeatureView)
			case constants.Datasource_Type_TableStore:
				daoConfig.TableStoreName = p.OnlineStore.GetDatasourceName()
				daoConfig.TableStoreOfflineTableName = p.OnlineStore.GetSeqOfflineTableName(referencedSeqFeatureView)
				daoConfig.TableStoreOnlineTableName = p.OnlineStore.GetSeqOnlineTableName(referencedSeqFeatureView)

			case constants.Datasource_Type_IGraph:
				daoConfig.SaveOriginalField = true
				daoConfig.IGraphName = p.OnlineStore.GetDatasourceName()
				daoConfig.GroupName = p.ProjectName
				daoConfig.IgraphEdgeName = p.OnlineStore.GetSeqOnlineTableName(referencedSeqFeatureView)

				fieldTypeMap := make(map[string]constants.FSType, len(view.Fields))
				for _, field := range view.Fields {
					if field.IsPartition {
						continue
					} else {
						fieldTypeMap[field.Name] = field.Type
					}
				}
				daoConfig.FieldTypeMap = fieldTypeMap

			default:

			}
		}
	}

	featureViewDao := dao.NewFeatureViewDao(daoConfig)
	sequenceFeatureView.featureViewDao = featureViewDao

	return sequenceFeatureView
}