Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

working Cursor Read/Write

  • Loading branch information...
commit 8ea771ca8cb21e6cdbb25638c1cc0493b31e7f3a 1 parent a6601cc
@carloscm authored
View
32 README.md
@@ -91,6 +91,7 @@ type Timeline struct {
}
row, err = Map(&Timeline{"userid", 10000000000004, "Author Name", "Hey this thing rocks!"})
+err = pool.Mutation().Insert("Timeline", row).Run()
````
### Cursors
@@ -98,30 +99,21 @@ row, err = Map(&Timeline{"userid", 10000000000004, "Author Name", "Hey this thin
As a convenient wrapper over Map/Unmap and the Query/Mutation interfaces Gossie provides the Cursor interface. This wrapper implements a classic database cursor over rows and composite row slices. Example:
```Go
-
-type Timeline struct {
- UserID string `cf:"Timeline" key:"UserID" col:"TweetID,*name" val:"*value"`
- TweetID int64
- Author string
- Body string
-}
-
-// initialize a cursor over a struct instance. we can reuse both the cursor and the struct for all operations
-tweet := &Timeline{"userid", 10000000000004, "Author Name", "Hey this thing rocks!"}
-cursor := pool.Cursor(tweet)
+// initialize a cursor
+cursor := pool.Cursor()
// write a single tweet
-cursor.Write()
+tweet := &Timeline{"userid", 10000000000004, "Author Name", "Hey this thing rocks!"}
+err = cursor.Write(tweet)
-// read a single tweet. this will implicitly use the key field for the row key, and will add a slice operation
-// if the struct has fixed composite columns
-tweet.Read(1)
+// read a single tweet. this will implicitly use the key field for the row key, and will add a slice
+// operation if the struct has fixed composite columns
+err = tweet.Read(tweet)
-// change the row key and the fixed composite filed and read a different tweet, reusing both the struct and the cursor
-tweet.UserID = "anotheruser"
-tweet.TweetID = 20000000000001
-cursor.Read(1)
-// tweet now contains the just read tweet
+// init a new struct instance with just the key and composite fields to read a different tweet
+tweet2 := &Timeline{"userid", 10000000000004}
+err = cursor.Read(tweet2)
+// tweet no
````
Comming soon: range reads for composites with buffering and paging
View
6 src/gossie/connection.go
@@ -60,7 +60,7 @@ type ConnectionPool interface {
Mutation() Mutation
// Cursor returns a high level interface for read and write operations over structs
- Cursor(interface{}) Cursor
+ Cursor() Cursor
// Close all the connections in the pool
Close()
@@ -311,8 +311,8 @@ func (cp *connectionPool) Mutation() Mutation {
return makeMutation(cp, cp.options.WriteConsistency)
}
-func (cp *connectionPool) Cursor(source interface{}) Cursor {
- return makeCursor(cp, source)
+func (cp *connectionPool) Cursor() Cursor {
+ return makeCursor(cp)
}
func (cp *connectionPool) Keyspace() string {
View
123 src/gossie/cursor.go
@@ -22,51 +22,42 @@ todo:
*/
+var (
+ ErrorNotFound = errors.New("Row (plus any composites) not found")
+)
+
// Cursor is a simple cursor-based interface for reading and writing structs from a Cassandra column family.
type Cursor interface {
- // Read rows (or slices of a row) to fill up the struct.
- //
- // For limit == 1 a single get operation will be issued. If a composite is present then this operation will
- // issue a slice with an exact match for the composite value
- //
- // For limit > 1, and for structs with no composite, a range get operation will be issued, the result buffered
- // internally in the Cursor, and the first returned row unmapped into the struct. Next() is then available for
- // paging inside the returned results and will issue new range get operations when neede. Please note that row
- // range get is unordered in most Cassandra configurations. TO DO
- //
- // If the struct has a composite column name then this operation will issue a single row get, but with a slice
- // predicate that allows for iteration over the row with Next()/Prev() TO DO
- Read(limit int) error
-
+ //Search(...)
//Next() bool
//Prev() bool
-
//First()
+ // Read a single row (or slice of a row) to fill up the struct.
+ // A single get operation will be issued. If a composite is present then this operation will issue a slice with
+ // an exact match for the composite value with the field values present in the struct
+ Read(interface{}) error
+
// Write a single row (or slice of a row) with the data currently in the struct.
- Write() error
+ Write(interface{}) error
//Delete()
}
type cursor struct {
- source interface{}
- pool *connectionPool
- //position int
- //buffer []interface{}
+ pool *connectionPool
}
-func makeCursor(cp *connectionPool, source interface{}) (c *cursor) {
+func makeCursor(cp *connectionPool) (c *cursor) {
return &cursor{
- source: source,
- pool: cp,
+ pool: cp,
}
}
-func (c *cursor) Write() error {
+func (c *cursor) Write(source interface{}) error {
- row, ms, err := internalMap(c.source)
+ row, ms, err := internalMap(source)
if err != nil {
return err
}
@@ -78,26 +69,20 @@ func (c *cursor) Write() error {
return nil
}
-func (c *cursor) Read(limit int) error {
- if limit < 0 {
- return errors.New("Limit is less than 1, nothing to read")
- }
+func (c *cursor) Read(source interface{}) error {
// deconstruct the source struct into a reflect.Value and a (cached) struct mapping
- ms, err := newMappedStruct(c.source)
+ ms, err := newMappedStruct(source)
if err != nil {
return err
}
// sanity checks
- if !ms.sm.isStarNameColumn && !ms.sm.isSliceColumn {
- return errors.New(fmt.Sprint("Struct ", ms.v.Type().Name(), " has no *name nor slice field in its col tag, nothing to read"))
- }
if ms.sm.isSliceColumn {
return errors.New(fmt.Sprint("Slice field in col tag is unsuported in Cursor for now, check back soon!"))
}
- // marshal the key field for the key to look up
+ // marshal the key field
key, err := ms.marshalKey()
if err != nil {
return err
@@ -106,41 +91,55 @@ func (c *cursor) Read(limit int) error {
// start building the query
q := c.pool.Query().Cf(ms.sm.cf)
+ // build a slice composite comparator if needed
+ if ms.sm.isCompositeColumn {
+ // iterate over the components and set an equality comparison for every simple field
+ start := make([]byte, 0)
+ end := make([]byte, 0)
+ var component int
+ for component = 0; component < len(ms.sm.columns); component++ {
+ fm := ms.sm.columns[component]
+ if fm.fieldKind != baseTypeField {
+ break
+ }
+ b, err := ms.mapColumn(baseTypeField, fm, 0)
+ if err != nil {
+ return err
+ }
+ start = packComposite(start, b, eocEquals)
+ end = packComposite(end, b, eocGreater)
+ }
+
+ /*if component < len(ms.sm.columns) {
+ // we still got one to go, this means the last one was an iterable non-fixed type (*name or go slice)
+ //fm := ms.sm.columns[component]
+ // TODO: this will only work for *name
+ b := make([]byte, 0)
+ start = packComposite(start, b, o[6], o[7], o[8])
+ end = packComposite(end, b, o[9], o[10], o[11])
+ }*/
+
+ // TODO: fix hardcoded number of columns
+ q.Slice(&Slice{Start: start, End: end, Count: 100})
+ }
+
//isCompositeColumn bool
//isSliceColumn bool
//isStarNameColumn bool
- if limit == 1 {
+ row, err := q.Get(key)
- if ms.sm.isCompositeColumn {
- return errors.New(fmt.Sprint("Cursor composite support will be implemented soon!"))
- /*
-
- // we only want a single result so issue an exact match composite comparator slice
- start := make([]byte, 0)
- start = packComposite(start, component []byte, true, true, true)
-
- packComposite(start, component []byte, true, sliceStart, inclusive bool) []byte {
+ if err != nil {
+ return err
+ }
- s := &Slice{
- Start:
- End:
- Count: 1
- }
- q.Slice(s)
- */
+ if row == nil {
+ return ErrorNotFound
+ }
- }
- row, err := q.Get(key)
- if err != nil {
- return err
- }
- err = Unmap(row, c.source)
- if err != nil {
- return err
- }
- } else {
- return errors.New(fmt.Sprint("Limit > 1 will be implemented soon!"))
+ err = Unmap(row, source)
+ if err != nil {
+ return err
}
return nil
View
44 src/gossie/cursor_test.go
@@ -5,6 +5,14 @@ import (
"testing"
)
+/*
+
+todo:
+
+ since most of the Cursor interface is still in flux the current tests are minimal
+
+*/
+
type ReasonableOne struct {
Username string `cf:"Reasonable" key:"Username" col:"TweetID,*name" val:"*value"`
TweetID int64
@@ -28,25 +36,45 @@ func TestRead(t *testing.T) {
Body: "hey this thing appears to work, nice!",
}
- cursor := cp.Cursor(ro)
- err = cursor.Write()
+ cursor := cp.Cursor()
+ err = cursor.Write(ro)
if err != nil {
t.Error("Writing struct:", err)
}
ro2 := &ReasonableOne{
Username: "testuser",
- TweetID: 100000000000002,
+ TweetID: 100000000000003,
+ Lat: 2.00002,
+ Lon: 1.11,
+ Body: "more words",
}
-
- cursor2 := cp.Cursor(ro2)
- err = cursor2.Read(1)
+ err = cursor.Write(ro2)
if err != nil {
- t.Error("Reading struct:", err)
+ t.Error("Writing struct:", err)
}
- if !reflect.DeepEqual(ro, ro2) {
+ ro3 := &ReasonableOne{
+ Username: "testuser",
+ TweetID: 100000000000002,
+ }
+ err = cursor.Read(ro3)
+ if err != nil {
+ t.Fatal("Reading struct:", err)
+ }
+ if !reflect.DeepEqual(ro, ro3) {
t.Error("Read does not match Write")
}
+ ro3 = &ReasonableOne{
+ Username: "testuser",
+ TweetID: 100000000000003,
+ }
+ err = cursor.Read(ro3)
+ if err != nil {
+ t.Fatal("Reading struct:", err)
+ }
+ if !reflect.DeepEqual(ro2, ro3) {
+ t.Error("Read does not match Write")
+ }
}
View
89 src/gossie/struct.go
@@ -9,6 +9,11 @@ import (
)
/*
+ NOTE: currently this works but it is in dire need of a cleanup/refactoring
+ the external interfaces (Map/Unmap) are final and won't change
+*/
+
+/*
todo:
@@ -167,7 +172,7 @@ func newStructMapping(t reflect.Type) (*structMapping, error) {
if name := meta["val"]; (name != "") || (name == "*value") {
if name == "*value" {
- sm.value = &fieldMapping{fieldKind: starValueField}
+ sm.value = &fieldMapping{fieldKind: starValueField, name: "*value"}
} else if sm.value, found = fields[name]; !found {
return nil, errors.New(fmt.Sprint("Referenced value field ", name, " does not exist in struct ", t.Name()))
}
@@ -186,7 +191,7 @@ func newStructMapping(t reflect.Type) (*structMapping, error) {
return nil, errors.New(fmt.Sprint("*name can only be used in the last position of a composite, error in struct ", t.Name()))
} else {
sm.isStarNameColumn = true
- fm = &fieldMapping{fieldKind: starNameField}
+ fm = &fieldMapping{fieldKind: starNameField, name: "*name"}
}
} else if fm, found = fields[name]; !found {
return nil, errors.New(fmt.Sprint("Referenced column field ", name, " does not exist in struct ", t.Name()))
@@ -234,6 +239,7 @@ func getMapping(v reflect.Value) (*structMapping, error) {
return sm, err
}
+// mappedStruct stores the reflect.Value and mapping for a particular instance of an struct
type mappedStruct struct {
source interface{}
v reflect.Value
@@ -303,8 +309,49 @@ func Map(source interface{}) (*Row, error) {
return row, err
}
-func (ms *mappedStruct) mapField(row *Row, component int, composite []byte, value []byte, valueIndex int) error {
+func (ms *mappedStruct) mapColumn(fieldKind int, fm *fieldMapping, i int) ([]byte, error) {
+ var b []byte
+ var err error
+
+ switch fieldKind {
+ case baseTypeField:
+ // single value from a single field
+ v := ms.v.Field(fm.position)
+ b, err = Marshal(v.Interface(), fm.cassandraType)
+
+ case baseTypeSliceField:
+ // the field is a slice, get the falue from position i
+ v := ms.v.Field(fm.position)
+ vi := v.Index(i)
+ b, err = Marshal(vi.Interface(), fm.cassandraType)
+
+ case starNameField:
+ // set value to the name of the field
+ b, err = Marshal(fm.cassandraName, UTF8Type)
+ }
+
+ if err != nil {
+ return nil, errors.New(fmt.Sprint("Error marshaling field ", fm.name, ":", err))
+ }
+
+ return b, nil
+}
+
+func (ms *mappedStruct) mapColumnForRow(fieldKind int, fm *fieldMapping, i int, composite []byte) ([]byte, error) {
+ b, err := ms.mapColumn(fieldKind, fm, i)
+ if err != nil {
+ return nil, err
+ }
+
+ // add to composite, if required
+ if ms.sm.isCompositeColumn {
+ b = packComposite(composite, b, eocEquals)
+ }
+ return b, nil
+}
+
+func (ms *mappedStruct) mapField(row *Row, component int, composite []byte, value []byte, valueIndex int) error {
// check if there are components left
if component < len(ms.sm.columns) {
@@ -315,16 +362,9 @@ func (ms *mappedStruct) mapField(row *Row, component int, composite []byte, valu
// base type
case baseTypeField:
- // set value of the current composite field to the field value
- v := ms.v.Field(fm.position)
- b, err := Marshal(v.Interface(), fm.cassandraType)
+ composite, err := ms.mapColumnForRow(baseTypeField, fm, 0, composite)
if err != nil {
- return errors.New(fmt.Sprint("Error marshaling field ", fm.name, ":", err))
- }
- if ms.sm.isCompositeColumn {
- composite = packComposite(composite, b, false, false, false)
- } else {
- composite = b
+ return err
}
return ms.mapField(row, component+1, composite, value, valueIndex)
@@ -334,17 +374,9 @@ func (ms *mappedStruct) mapField(row *Row, component int, composite []byte, valu
v := ms.v.Field(fm.position)
n := v.Len()
for i := 0; i < n; i++ {
- // set value of the current composite field to the field value
- vi := v.Index(i)
- b, err := Marshal(vi.Interface(), fm.cassandraType)
+ subComposite, err := ms.mapColumnForRow(baseTypeSliceField, fm, i, composite)
if err != nil {
- return errors.New(fmt.Sprint("Error marshaling field ", fm.name, ":", err))
- }
- var subComposite []byte
- if ms.sm.isCompositeColumn {
- subComposite = packComposite(composite, b, false, false, false)
- } else {
- subComposite = b
+ return err
}
err = ms.mapField(row, component+1, subComposite, value, i)
if err != nil {
@@ -356,20 +388,13 @@ func (ms *mappedStruct) mapField(row *Row, component int, composite []byte, valu
case starNameField:
// iterate over non-key/col/val-referenced struct fields and map more columns
for _, fm := range ms.sm.others {
- // set value of the current composite field to the field name (possibly overriden by name:)
- b, err := Marshal(fm.cassandraName, UTF8Type)
+ subComposite, err := ms.mapColumnForRow(starNameField, fm, 0, composite)
if err != nil {
- return errors.New(fmt.Sprint("Error marshaling field ", fm.name, ":", err))
- }
- var subComposite []byte
- if ms.sm.isCompositeColumn {
- subComposite = packComposite(composite, b, false, false, false)
- } else {
- subComposite = b
+ return err
}
// marshal field value and pass it to next field mapper in case it is *value
v := ms.v.Field(fm.position)
- b, err = Marshal(v.Interface(), fm.cassandraType)
+ b, err := Marshal(v.Interface(), fm.cassandraType)
if err != nil {
return errors.New(fmt.Sprint("Error marshaling field ", fm.name, ":", err))
}
View
33 src/gossie/struct_test.go
@@ -63,6 +63,7 @@ type noErrB struct {
A int `cf:"cfname" key:"A" col:"*name" val:"*value"`
B int `name:"Z"`
C int `type:"AsciiType"`
+ D int
}
type noErrC struct {
A int `cf:"cfname" key:"A" col:"B,*name" val:"*value"`
@@ -140,17 +141,18 @@ func TestStructMapping(t *testing.T) {
}
checkMapping(t, goodA, mapA, "mapA")
- mapB, _ := buildMappingFromPtr(&noErrB{1, 2, 3})
+ mapB, _ := buildMappingFromPtr(&noErrB{1, 2, 3, 4})
goodB := &structMapping{
cf: "cfname",
key: &fieldMapping{fieldKind: baseTypeField, position: 0, name: "A", cassandraType: LongType, cassandraName: "A"},
columns: []*fieldMapping{
- &fieldMapping{fieldKind: starNameField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ &fieldMapping{fieldKind: starNameField, position: 0, name: "*name", cassandraType: 0, cassandraName: ""},
},
- value: &fieldMapping{fieldKind: starValueField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ value: &fieldMapping{fieldKind: starValueField, position: 0, name: "*value", cassandraType: 0, cassandraName: ""},
others: map[string]*fieldMapping{
"Z": &fieldMapping{fieldKind: baseTypeField, position: 1, name: "B", cassandraType: LongType, cassandraName: "Z"},
"C": &fieldMapping{fieldKind: baseTypeField, position: 2, name: "C", cassandraType: AsciiType, cassandraName: "C"},
+ "D": &fieldMapping{fieldKind: baseTypeField, position: 3, name: "D", cassandraType: LongType, cassandraName: "D"},
},
isCompositeColumn: false,
isSliceColumn: false,
@@ -164,9 +166,9 @@ func TestStructMapping(t *testing.T) {
key: &fieldMapping{fieldKind: baseTypeField, position: 0, name: "A", cassandraType: LongType, cassandraName: "A"},
columns: []*fieldMapping{
&fieldMapping{fieldKind: baseTypeField, position: 1, name: "B", cassandraType: LongType, cassandraName: "B"},
- &fieldMapping{fieldKind: starNameField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ &fieldMapping{fieldKind: starNameField, position: 0, name: "*name", cassandraType: 0, cassandraName: ""},
},
- value: &fieldMapping{fieldKind: starValueField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ value: &fieldMapping{fieldKind: starValueField, position: 0, name: "*value", cassandraType: 0, cassandraName: ""},
others: map[string]*fieldMapping{
"C": &fieldMapping{fieldKind: baseTypeField, position: 2, name: "C", cassandraType: LongType, cassandraName: "C"},
},
@@ -224,9 +226,9 @@ func TestStructMapping(t *testing.T) {
&fieldMapping{fieldKind: baseTypeField, position: 9, name: "FFloat64", cassandraType: DoubleType, cassandraName: "FFloat64"},
&fieldMapping{fieldKind: baseTypeField, position: 10, name: "FString", cassandraType: UTF8Type, cassandraName: "FString"},
&fieldMapping{fieldKind: baseTypeField, position: 11, name: "FUUID", cassandraType: UUIDType, cassandraName: "FUUID"},
- &fieldMapping{fieldKind: starNameField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ &fieldMapping{fieldKind: starNameField, position: 0, name: "*name", cassandraType: 0, cassandraName: ""},
},
- value: &fieldMapping{fieldKind: starValueField, position: 0, name: "", cassandraType: 0, cassandraName: ""},
+ value: &fieldMapping{fieldKind: starValueField, position: 0, name: "*value", cassandraType: 0, cassandraName: ""},
others: map[string]*fieldMapping{
"Val": &fieldMapping{fieldKind: baseTypeField, position: 12, name: "Val", cassandraType: UTF8Type, cassandraName: "Val"},
},
@@ -245,13 +247,13 @@ type structTestShell struct {
name string
}
-func (shell *structTestShell) checkMap(t *testing.T, expectedStruct interface{}) {
+func (shell *structTestShell) checkMap(t *testing.T, expectedStruct interface{}, round int) {
resultRow, err := Map(expectedStruct)
if err != nil {
t.Error("Error mapping struct: ", err)
}
if !reflect.DeepEqual(resultRow, shell.expectedRow) {
- t.Error("Mapped struct ", shell.name, " does not match expected row ", shell.expectedRow, " actual ", resultRow)
+ t.Error("(Round ", round, ") Mapped struct ", shell.name, " does not match expected row ", shell.expectedRow, " actual ", resultRow)
}
}
@@ -267,13 +269,16 @@ func (shell *structTestShell) checkUnmap(t *testing.T) interface{} {
}
func (shell *structTestShell) checkFullMap(t *testing.T) {
- shell.checkMap(t, shell.expectedStruct)
+ shell.checkMap(t, shell.expectedStruct, 1)
intermediateStruct := shell.checkUnmap(t)
- shell.checkMap(t, intermediateStruct)
+ shell.checkMap(t, intermediateStruct, 2)
}
func TestMap(t *testing.T) {
+ // CAVEAT: column ordering is not deterministic for this kind of Map/Unmap roundtrip
+ // something needs to be done to test these structs with independent order over the columns
+
shells := []*structTestShell{
&structTestShell{
name: "noErrA",
@@ -292,7 +297,7 @@ func TestMap(t *testing.T) {
&structTestShell{
name: "noErrB",
- expectedStruct: &noErrB{1, 2, 3},
+ expectedStruct: &noErrB{1, 2, 3, 4},
resultStruct: &noErrB{},
expectedRow: &Row{
Key: []byte{0, 0, 0, 0, 0, 0, 0, 1},
@@ -302,6 +307,10 @@ func TestMap(t *testing.T) {
Value: []byte{51},
},
&Column{
+ Name: []byte{68},
+ Value: []byte{0, 0, 0, 0, 0, 0, 0, 4},
+ },
+ &Column{
Name: []byte{90},
Value: []byte{0, 0, 0, 0, 0, 0, 0, 2},
},
View
24 src/gossie/types.go
@@ -624,7 +624,22 @@ func parseTypeClass(cassType string) TypeClass {
return r
}
-func packComposite(current, component []byte, comparator, sliceStart, inclusive bool) []byte {
+const (
+ _ = iota
+ eocEquals byte = 0
+ eocGreater byte = 1
+ eocLower byte = 0xff
+)
+
+func packComposite(current, component []byte, eoc byte) []byte {
+ r := make([]byte, 2)
+ enc.BigEndian.PutUint16(r, uint16(len(component)))
+ r = append(current, r...)
+ r = append(r, component...)
+ return append(r, eoc)
+}
+
+/*
var eoc byte = 0
if comparator {
if inclusive {
@@ -641,12 +656,7 @@ func packComposite(current, component []byte, comparator, sliceStart, inclusive
}
}
}
- r := make([]byte, 2)
- enc.BigEndian.PutUint16(r, uint16(len(component)))
- r = append(current, r...)
- r = append(r, component...)
- return append(r, eoc)
-}
+*/
func unpackComposite(composite []byte) [][]byte {
components := make([][]byte, 0)
Please sign in to comment.
Something went wrong with that request. Please try again.