x/cqlbuilder: Add implementation of a query builder

backend/gocql/db.go

@@ -62,7 +62,19 @@ func (db *DB) ExecCAS(ctx context.Context, stmt string, vs ...interface{}) cql.C
 }
 
 func (db *DB) QueryRow(ctx context.Context, stmt string, vs ...interface{}) cql.Scanner {
-	return db.query(ctx, stmt, vs)
+	return scanner{db.query(ctx, stmt, vs)}
+}
+
+type scanner struct {
+	cql.Scanner
+}
+
+func (s scanner) Scan(vs ...interface{}) error {
+	if err := s.Scanner.Scan(vs...); err != gocql.ErrNotFound {
+		return err
+	}
+
+	return cql.ErrNoRows
 }
 
 type cursor struct {

cqltest/static_source.go

@@ -0,0 +1,47 @@
+package cqltest
+
+import (
+	"context"
+	"io"
+	"io/ioutil"
+	"strings"
+
+	"github.com/upfluence/cql/x/migration"
+)
+
+type StaticSource struct {
+	MigrationUp   string
+	MigrationDown string
+}
+
+func (ss StaticSource) ID() uint {
+	return 1
+}
+
+func (ss StaticSource) Up() (io.ReadCloser, error) {
+	return ioutil.NopCloser(strings.NewReader(ss.MigrationUp)), nil
+}
+
+func (ss StaticSource) Down() (io.ReadCloser, error) {
+	return ioutil.NopCloser(strings.NewReader(ss.MigrationDown)), nil
+}
+
+func (ss StaticSource) Get(_ context.Context, v uint) (migration.Migration, error) {
+	if v != 1 {
+		return nil, migration.ErrNotExist
+	}
+
+	return ss, nil
+}
+
+func (ss StaticSource) First(context.Context) (migration.Migration, error) {
+	return ss, nil
+}
+
+func (ss StaticSource) Next(context.Context, uint) (bool, uint, error) {
+	return false, 0, nil
+}
+
+func (ss StaticSource) Prev(context.Context, uint) (bool, uint, error) {
+	return false, 0, nil
+}

db.go

@@ -1,6 +1,11 @@
 package cql
 
-import "context"
+import (
+	"context"
+	"errors"
+)
+
+var ErrNoRows = errors.New("cql: No rows found")
 
 type BatchType uint8
 

integration/integration_test.go

@@ -17,9 +17,9 @@ func TestMigrationIntegration(t *testing.T) {
 		cqltest.WithMigratorFunc(func(db cql.DB) migration.Migrator {
 			return migration.NewMigrator(
 				db,
-				staticSource{
-					up:   "CREATE TABLE IF NOT EXISTS foo(uuid UUID PRIMARY KEY, data blob)",
-					down: "DROP TABLE foo",
+				cqltest.StaticSource{
+					MigrationUp:   "CREATE TABLE IF NOT EXISTS foo(uuid UUID PRIMARY KEY, data blob)",
+					MigrationDown: "DROP TABLE foo",
 				},
 			)
 		}),

integration/static_source.go

@@ -1,46 +1 @@
 package integration
-
-import (
-	"context"
-	"io"
-	"io/ioutil"
-	"strings"
-
-	"github.com/upfluence/cql/x/migration"
-)
-
-type staticSource struct {
-	up, down string
-}
-
-func (ss staticSource) ID() uint {
-	return 1
-}
-
-func (ss staticSource) Up() (io.ReadCloser, error) {
-	return ioutil.NopCloser(strings.NewReader(ss.up)), nil
-}
-
-func (ss staticSource) Down() (io.ReadCloser, error) {
-	return ioutil.NopCloser(strings.NewReader(ss.down)), nil
-}
-
-func (ss staticSource) Get(_ context.Context, v uint) (migration.Migration, error) {
-	if v != 1 {
-		return nil, migration.ErrNotExist
-	}
-
-	return ss, nil
-}
-
-func (ss staticSource) First(context.Context) (migration.Migration, error) {
-	return ss, nil
-}
-
-func (ss staticSource) Next(context.Context, uint) (bool, uint, error) {
-	return false, 0, nil
-}
-
-func (ss staticSource) Prev(context.Context, uint) (bool, uint, error) {
-	return false, 0, nil
-}

x/cqlbuilder/batch_statement.go

@@ -0,0 +1,34 @@
+package cqlbuilder
+
+import (
+	"context"
+
+	"github.com/upfluence/cql"
+)
+
+type BatchStatement struct {
+	Type cql.BatchType
+
+	Statements []CASStatement
+}
+
+type BatchExecer struct {
+	QueryBuilder *QueryBuilder
+	Statement    BatchStatement
+}
+
+func (be *BatchExecer) Exec(ctx context.Context, qvs map[string]interface{}) error {
+	var b = be.QueryBuilder.Batch(ctx, be.Statement.Type)
+
+	for _, s := range be.Statement.Statements {
+		stmt, vs, err := s.buildQuery(qvs)
+
+		if err != nil {
+			return err
+		}
+
+		b.Query(stmt, vs...)
+	}
+
+	return b.Exec()
+}

x/cqlbuilder/delete_statement.go

@@ -0,0 +1,69 @@
+package cqlbuilder
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+type LWTDeleteClause interface {
+	LWTClause
+
+	isDeleteClause()
+}
+
+type DeleteStatement struct {
+	Table string
+
+	Fields      []Marker
+	WhereClause PredicateClause
+
+	Timestamp time.Time
+	LWTClause LWTDeleteClause
+}
+
+func (ds DeleteStatement) casScanKeys() []string {
+	if lck, ok := ds.LWTClause.(interface{ keys() []string }); ok {
+		return lck.keys()
+	}
+
+	return nil
+}
+
+func (ds DeleteStatement) buildQuery(qvs map[string]interface{}) (string, []interface{}, error) {
+	var (
+		qw queryWriter
+
+		ks = make([]string, len(ds.Fields))
+	)
+
+	for i, f := range ds.Fields {
+		k := f.ToCQL()
+
+		if i == len(ds.Fields)-1 {
+			k += " "
+		}
+
+		ks[i] = k
+	}
+
+	fmt.Fprintf(&qw, "DELETE %sFROM %s ", strings.Join(ks, ", "), ds.Table)
+
+	DMLOptions{Timestamp: ds.Timestamp}.writeTo(&qw)
+
+	qw.WriteString("WHERE ")
+
+	if err := ds.WhereClause.WriteTo(&qw, qvs); err != nil {
+		return "", nil, err
+	}
+
+	if lc := ds.LWTClause; lc != nil {
+		qw.WriteRune(' ')
+
+		if err := lc.writeTo(&qw, qvs); err != nil {
+			return "", nil, err
+		}
+	}
+
+	return qw.String(), qw.args, nil
+}

x/cqlbuilder/delete_statement_test.go

@@ -0,0 +1,32 @@
+package cqlbuilder
+
+import "testing"
+
+func TestDeleteStatement(t *testing.T) {
+	for _, stc := range []statementTestCase{
+		{
+			name: "basic",
+			stmt: DeleteStatement{
+				Table:       "foo",
+				WhereClause: Eq(Column("bar")),
+			},
+			vs:       map[string]interface{}{"bar": 3},
+			wantStmt: "DELETE FROM foo WHERE bar = ?",
+			wantArgs: []interface{}{3},
+		},
+		{
+			name: "lwt field",
+			stmt: DeleteStatement{
+				Table:       "foo",
+				Fields:      []Marker{Column("fiz")},
+				WhereClause: Eq(Column("bar")),
+				LWTClause:   PredicateLWTClause{Predicate: Eq(Column("buz"))},
+			},
+			vs:       map[string]interface{}{"fiz": 1, "buz": 2, "bar": 3},
+			wantStmt: "DELETE fiz FROM foo WHERE bar = ? IF buz = ?",
+			wantArgs: []interface{}{3, 2},
+		},
+	} {
+		stc.assert(t)
+	}
+}

x/cqlbuilder/dml.go

@@ -0,0 +1,92 @@
+package cqlbuilder
+
+import (
+	"fmt"
+	"io"
+	"time"
+)
+
+type DMLOptions struct {
+	TTL       time.Duration
+	Timestamp time.Time
+}
+
+func (do DMLOptions) writeTo(w io.Writer) {
+	if do.TTL == 0 && do.Timestamp.IsZero() {
+		return
+	}
+
+	io.WriteString(w, " USING")
+
+	if do.TTL > 0 {
+		fmt.Fprintf(w, " TTL %d", int(do.TTL.Seconds()))
+
+		if !do.Timestamp.IsZero() {
+			io.WriteString(w, " AND")
+		}
+	}
+
+	if !do.Timestamp.IsZero() {
+		fmt.Fprintf(
+			w,
+			" TIMESTAMP %d",
+			do.Timestamp.Unix()*1000+do.Timestamp.UnixNano()/1000000,
+		)
+	}
+}
+
+type LWTClause interface {
+	writeTo(QueryWriter, map[string]interface{}) error
+}
+
+type notExistsClause struct{}
+
+var NotExistsClause = notExistsClause{}
+
+func (notExistsClause) writeTo(qw QueryWriter, _ map[string]interface{}) error {
+	_, err := io.WriteString(qw, "IF NOT EXISTS")
+	return err
+}
+
+func (notExistsClause) isInsertClause() {}
+func (notExistsClause) isUpdateClause() {}
+
+type existsClause struct{}
+
+var ExistsClause = existsClause{}
+
+func (existsClause) writeTo(qw QueryWriter, _ map[string]interface{}) error {
+	_, err := io.WriteString(qw, "IF EXISTS")
+	return err
+}
+
+func (existsClause) isUpdateClause() {}
+func (existsClause) isDeleteClause() {}
+
+type PredicateLWTClause struct {
+	Predicate PredicateClause
+}
+
+func (plc PredicateLWTClause) writeTo(qw QueryWriter, vs map[string]interface{}) error {
+	if _, err := io.WriteString(qw, "IF "); err != nil {
+		return err
+	}
+
+	return plc.Predicate.WriteTo(qw, vs)
+}
+
+func (plc PredicateLWTClause) keys() []string {
+	var (
+		ms = plc.Predicate.Markers()
+		ks = make([]string, len(ms))
+	)
+
+	for i, m := range ms {
+		ks[i] = m.Binding()
+	}
+
+	return ks
+}
+
+func (plc PredicateLWTClause) isUpdateClause() {}
+func (plc PredicateLWTClause) isDeleteClause() {}