func processFunctionColumn()

in cassandra-bigtable-migration-tools/cassandra-bigtable-proxy/translator/translator_select.go [383:425]
37 lines of code
12 McCabe index (conditional complexity)

func processFunctionColumn(t *Translator, columnMetadata schemaMapping.SelectedColumns, tableName string, keySpace string, columns []string) ([]string, string, error) {

	if columnMetadata.FuncColumnName == STAR && strings.ToLower(columnMetadata.FuncName) == "count" {
		columns = append(columns, "count(*)")
		return columns, "bigint", nil

	}
	colMeta, found := t.SchemaMappingConfig.TablesMetaData[keySpace][tableName][columnMetadata.FuncColumnName]
	if !found {
		// Check if the column is an alias
		if aliasMeta, aliasFound := t.SchemaMappingConfig.TablesMetaData[keySpace][tableName][columnMetadata.Alias]; aliasFound {
			colMeta = aliasMeta
		} else if columnMetadata.FuncName == "count" && columnMetadata.FuncColumnName == STAR {
			// Handle special case for count(*)
			return append(columns, "count(*)"), "bigint", nil
		} else {
			return nil, "", fmt.Errorf("column metadata not found for column '%s' in table '%s' and keyspace '%s'", columnMetadata.FuncColumnName, tableName, keySpace)
		}
	}
	colFamiliy := t.SchemaMappingConfig.SystemColumnFamily
	dataType := colMeta.CQLType
	column := ""

	if !funcAllowedInAggregate(columnMetadata.FuncName) {
		return nil, "", fmt.Errorf("unknown function '%s'", columnMetadata.FuncName)
	}
	if columnMetadata.FuncName != "count" {
		if !dtAllowedInAggregate(colMeta.ColumnType) {
			return nil, "", fmt.Errorf("column not supported for aggregate")
		}
	}
	castValue, castErr := castColumns(colMeta, colFamiliy)
	if castErr != nil {
		return nil, "", castErr
	}
	column = fmt.Sprintf("%s(%s)", columnMetadata.FuncName, castValue)

	if columnMetadata.IsAs {
		column = column + " as " + columnMetadata.Alias
	}
	columns = append(columns, column)
	return columns, dataType, nil
}