func parseAssignments()

in cassandra-bigtable-migration-tools/cassandra-bigtable-proxy/translator/translator_update.go [49:134]
83 lines of code
19 McCabe index (conditional complexity)

func parseAssignments(assignments []cql.IAssignmentElementContext, tableName string, schemaMapping *schemaMapping.SchemaMappingConfig, keyspace string, prependColumn *[]string) (*UpdateSetResponse, error) {
	if len(assignments) == 0 {
		return nil, errors.New("invalid input")
	}
	var setResp []UpdateSetValue
	var paramKeys []string
	params := make(map[string]interface{})

	for i, setVal := range assignments {
		assignStr := setVal.GetText()
		strSplit := strings.Split(assignStr, "=")
		s := strconv.Itoa(i + 1)
		placeholder := "set" + s
		colObj := setVal.OBJECT_NAME(0)
		if colObj == nil {
			return nil, errors.New("error parsing column for assignments")
		}
		columnName := colObj.GetText()
		if columnName == "" {
			return nil, errors.New("no columnName found for assignments")
		}
		columnName = strings.ReplaceAll(columnName, literalPlaceholder, "")
		var value string
		if strings.Contains(assignStr, "=?") {
			value = questionMark
		} else {
			valConst := setVal.Constant()
			if valConst == nil {
				value = strSplit[1]
				if value == "" {
					return nil, errors.New("error parsing value for assignments")
				}
			} else {
				value = valConst.GetText()
			}

		}

		value = strings.ReplaceAll(value, "'", "")
		var val interface{} //encryted val
		columnType, err := schemaMapping.GetColumnType(keyspace, tableName, columnName)
		if err != nil {
			return nil, fmt.Errorf("undefined column name %s in table %s.%s", columnName, keyspace, tableName)
		}
		if columnType.IsPrimaryKey {
			return nil, fmt.Errorf("primary key not allowed to assignments")
		}
		if value != questionMark {
			if columnType.IsCollection {
				val = value
			} else {
				val, err = formatValues(value, columnType.CQLType, 4)
				if err != nil {
					return nil, err
				}
			}
			params[placeholder] = val
		}
		paramKeys = append(paramKeys, placeholder)
		if (strings.Contains(columnType.CQLType, "map") || strings.Contains(columnType.CQLType, "list")) && IsMapKey(strSplit[0]) {
			_, extractkey := ExtractMapKey(assignStr)
			if extractkey == "" {
				return nil, fmt.Errorf("invalid format: missing map key")
			}
			newVal := fmt.Sprintf("%s+{%s:%s}", columnName, extractkey, value)
			val = newVal
		} else if strings.Contains(columnType.CQLType, "list") && strings.Contains(value, fmt.Sprintf("+%s", columnName)) {
			listAssignment := strSplit[1]
			valSplit := strings.Split(listAssignment, "+")
			newVal := fmt.Sprintf("%s+%s", valSplit[1], valSplit[0])
			val = newVal
			*prependColumn = append(*prependColumn, columnName)
		}
		setResp = append(setResp, UpdateSetValue{
			Column:    columnName,
			Value:     "@" + placeholder,
			Encrypted: val,
			CQLType:   columnType.CQLType,
		})
	}
	return &UpdateSetResponse{
		UpdateSetValues: setResp,
		ParamKeys:       paramKeys,
		Params:          params,
	}, nil
}