Add support for handling tuple types

Add basic support to handling scanning of tuples. This
changes the Scan logic such that tuple columns are scanned
as if they were defined as ordinary columns.

conn.go

@@ -503,8 +503,8 @@ func (c *Conn) executeQuery(qry *Query) *Iter {
 		return &Iter{}
 	case *resultRowsFrame:
 		iter := &Iter{
-			columns: x.meta.columns,
-			rows:    x.rows,
+			meta: x.meta,
+			rows: x.rows,
 		}
 
 		if len(x.meta.pagingState) > 0 {

frame.go

@@ -588,7 +588,18 @@ func (f *framer) readTypeInfo() TypeInfo {
 
 	switch simple.typ {
 	case TypeTuple:
-		panic("not implemented")
+		n := f.readShort()
+		tuple := TupleTypeInfo{
+			NativeType: simple,
+			Elems:      make([]TypeInfo, n),
+		}
+
+		for i := 0; i < int(n); i++ {
+			tuple.Elems[i] = f.readTypeInfo()
+		}
+
+		return tuple
+
 	case TypeMap, TypeList, TypeSet:
 		collection := CollectionType{
 			NativeType: simple,
@@ -613,6 +624,11 @@ type resultMetadata struct {
 	pagingState []byte
 
 	columns []ColumnInfo
+
+	// this is a count of the total number of columns which can be scanned,
+	// it is at minimum len(columns) but may be larger, for instance when a column
+	// is a UDT or tuple.
+	actualColCount int
 }
 
 func (r resultMetadata) String() string {
@@ -625,6 +641,7 @@ func (f *framer) parseResultMetadata() resultMetadata {
 	}
 
 	colCount := f.readInt()
+	meta.actualColCount = colCount
 
 	if meta.flags&flagHasMorePages == flagHasMorePages {
 		meta.pagingState = f.readBytes()
@@ -656,6 +673,12 @@ func (f *framer) parseResultMetadata() resultMetadata {
 
 		col.Name = f.readString()
 		col.TypeInfo = f.readTypeInfo()
+		switch v := col.TypeInfo.(type) {
+		// maybe also UDT
+		case TupleTypeInfo:
+			// -1 because we already included the tuple column
+			meta.actualColCount += len(v.Elems) - 1
+		}
 	}
 
 	meta.columns = cols

helpers.go

@@ -46,6 +46,10 @@ func goType(t TypeInfo) reflect.Type {
 		return reflect.MapOf(goType(t.(CollectionType).Key), goType(t.(CollectionType).Elem))
 	case TypeVarint:
 		return reflect.TypeOf(*new(*big.Int))
+	case TypeTuple:
+		// what can we do here? all there is to do is to make a list of interface{}
+		tuple := t.(TupleTypeInfo)
+		return reflect.TypeOf(make([]interface{}, len(tuple.Elems)))
 	default:
 		return nil
 	}
@@ -56,47 +60,48 @@ func dereference(i interface{}) interface{} {
 }
 
 func getApacheCassandraType(class string) Type {
-	if strings.HasPrefix(class, apacheCassandraTypePrefix) {
-		switch strings.TrimPrefix(class, apacheCassandraTypePrefix) {
-		case "AsciiType":
-			return TypeAscii
-		case "LongType":
-			return TypeBigInt
-		case "BytesType":
-			return TypeBlob
-		case "BooleanType":
-			return TypeBoolean
-		case "CounterColumnType":
-			return TypeCounter
-		case "DecimalType":
-			return TypeDecimal
-		case "DoubleType":
-			return TypeDouble
-		case "FloatType":
-			return TypeFloat
-		case "Int32Type":
-			return TypeInt
-		case "DateType", "TimestampType":
-			return TypeTimestamp
-		case "UUIDType":
-			return TypeUUID
-		case "UTF8Type":
-			return TypeVarchar
-		case "IntegerType":
-			return TypeVarint
-		case "TimeUUIDType":
-			return TypeTimeUUID
-		case "InetAddressType":
-			return TypeInet
-		case "MapType":
-			return TypeMap
-		case "ListType":
-			return TypeList
-		case "SetType":
-			return TypeSet
-		}
+	switch strings.TrimPrefix(class, apacheCassandraTypePrefix) {
+	case "AsciiType":
+		return TypeAscii
+	case "LongType":
+		return TypeBigInt
+	case "BytesType":
+		return TypeBlob
+	case "BooleanType":
+		return TypeBoolean
+	case "CounterColumnType":
+		return TypeCounter
+	case "DecimalType":
+		return TypeDecimal
+	case "DoubleType":
+		return TypeDouble
+	case "FloatType":
+		return TypeFloat
+	case "Int32Type":
+		return TypeInt
+	case "DateType", "TimestampType":
+		return TypeTimestamp
+	case "UUIDType":
+		return TypeUUID
+	case "UTF8Type":
+		return TypeVarchar
+	case "IntegerType":
+		return TypeVarint
+	case "TimeUUIDType":
+		return TypeTimeUUID
+	case "InetAddressType":
+		return TypeInet
+	case "MapType":
+		return TypeMap
+	case "ListType":
+		return TypeList
+	case "SetType":
+		return TypeSet
+	case "TupleType":
+		return TypeTuple
+	default:
+		return TypeCustom
 	}
-	return TypeCustom
 }
 
 func (r *RowData) rowMap(m map[string]interface{}) {

marshal.go

@@ -128,6 +128,8 @@ func Unmarshal(info TypeInfo, data []byte, value interface{}) error {
 		return unmarshalUUID(info, data, value)
 	case TypeInet:
 		return unmarshalInet(info, data, value)
+	case TypeTuple:
+		return unmarshalTuple(info, data, value)
 	}
 	// TODO(tux21b): add the remaining types
 	return fmt.Errorf("can not unmarshal %s into %T", info, value)
@@ -1167,6 +1169,35 @@ func unmarshalInet(info TypeInfo, data []byte, value interface{}) error {
 	return unmarshalErrorf("cannot unmarshal %s into %T", info, value)
 }
 
+// currently only support unmarshal into a list of values, this makes it possible
+// to support tuples without changing the query API. In the future this can be extend
+// to allow unmarshalling into custom tuple types.
+func unmarshalTuple(info TypeInfo, data []byte, value interface{}) error {
+	if v, ok := value.(Unmarshaler); ok {
+		return v.UnmarshalCQL(info, data)
+	}
+
+	tuple := info.(TupleTypeInfo)
+	switch v := value.(type) {
+	case []interface{}:
+		for i, elem := range tuple.Elems {
+			// each element inside data is a [bytes]
+			size := readInt(data)
+			data = data[4:]
+
+			err := Unmarshal(elem, data[:size], v[i])
+			if err != nil {
+				return err
+			}
+			data = data[size:]
+		}
+
+		return nil
+	}
+
+	return unmarshalErrorf("cannot unmarshal %s into %T", info, value)
+}
+
 // TypeInfo describes a Cassandra specific data type.
 type TypeInfo interface {
 	Type() Type
@@ -1234,7 +1265,7 @@ func (c CollectionType) String() string {
 
 type TupleTypeInfo struct {
 	NativeType
-	Elem []TypeInfo
+	Elems []TypeInfo
 }
 
 // String returns a human readable name for the Cassandra datatype
@@ -1307,8 +1338,10 @@ func (t Type) String() string {
 		return "set"
 	case TypeVarint:
 		return "varint"
+	case TypeTuple:
+		return "tuple"
 	default:
-		return fmt.Sprintf("unkown_type_%d", t)
+		return fmt.Sprintf("unknown_type_%d", t)
 	}
 }
 

session.go

@@ -598,16 +598,16 @@ func (q *Query) MapScanCAS(dest map[string]interface{}) (applied bool, err error
 // were returned by a query. The iterator might send additional queries to the
 // database during the iteration if paging was enabled.
 type Iter struct {
-	err     error
-	pos     int
-	rows    [][][]byte
-	columns []ColumnInfo
-	next    *nextIter
+	err  error
+	pos  int
+	rows [][][]byte
+	meta resultMetadata
+	next *nextIter
 }
 
 // Columns returns the name and type of the selected columns.
 func (iter *Iter) Columns() []ColumnInfo {
-	return iter.columns
+	return iter.meta.columns
 }
 
 // Scan consumes the next row of the iterator and copies the columns of the
@@ -632,20 +632,46 @@ func (iter *Iter) Scan(dest ...interface{}) bool {
 	if iter.next != nil && iter.pos == iter.next.pos {
 		go iter.next.fetch()
 	}
-	if len(dest) != len(iter.columns) {
+
+	// currently only support scanning into an expand tuple, such that its the same
+	// as scanning in more values from a single column
+	if len(dest) != iter.meta.actualColCount {
 		iter.err = errors.New("count mismatch")
 		return false
 	}
-	for i := 0; i < len(iter.columns); i++ {
+
+	// i is the current position in dest, could posible replace it and just use
+	// slices of dest
+	i := 0
+	for c, col := range iter.meta.columns {
 		if dest[i] == nil {
+			i++
 			continue
 		}
-		err := Unmarshal(iter.columns[i].TypeInfo, iter.rows[iter.pos][i], dest[i])
-		if err != nil {
-			iter.err = err
+
+		// how can we allow users to pass in a single struct to unmarshal into
+		if col.TypeInfo.Type() == TypeTuple {
+			// this will panic, actually a bug, please report
+			tuple := col.TypeInfo.(TupleTypeInfo)
+
+			count := len(tuple.Elems)
+			// here we pass in a slice of the struct which has the number number of
+			// values as elements in the tuple
+			iter.err = Unmarshal(col.TypeInfo, iter.rows[iter.pos][c], dest[i:i+count])
+			if iter.err != nil {
+				return false
+			}
+			i += count
+			continue
+		}
+
+		iter.err = Unmarshal(col.TypeInfo, iter.rows[iter.pos][c], dest[i])
+		if iter.err != nil {
 			return false
 		}
+		i++
 	}
+
 	iter.pos++
 	return true
 }

tuple_test.go

@@ -0,0 +1,51 @@
+// +build all integration
+
+package gocql
+
+import "testing"
+
+func TestTupleSimple(t *testing.T) {
+	if *flagProto < protoVersion3 {
+		t.Skip("tuple types are only available of proto>=3")
+	}
+
+	session := createSession(t)
+	defer session.Close()
+
+	err := createTable(session, `CREATE TABLE tuple_test(
+		id int,
+		coord frozen<tuple<int, int>>,
+
+		primary key(id))`)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = session.Query("INSERT INTO tuple_test(id, coord) VALUES(?, (?, ?))", 1, 100, -100).Exec()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var (
+		id    int
+		coord struct {
+			x int
+			y int
+		}
+	)
+
+	iter := session.Query("SELECT id, coord FROM tuple_test WHERE id=?", 1)
+	if err := iter.Scan(&id, &coord.x, &coord.y); err != nil {
+		t.Fatal(err)
+	}
+
+	if id != 1 {
+		t.Errorf("expected to get id=1 got: %v", id)
+	}
+	if coord.x != 100 {
+		t.Errorf("expected to get coord.x=100 got: %v", coord.x)
+	}
+	if coord.y != -100 {
+		t.Errorf("expected to get coord.y=-100 got: %v", coord.y)
+	}
+}