x/cqlbuider: Initial implementation of the query builder

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,89 @@
+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 = notExistsClause{}
+
+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 (pcl PredicateLWTClause) keys() []string {
+	var ks []string
+
+	for _, m := range pcl.Predicate.Markers() {
+		ks = append(ks, m.Binding())
+	}
+
+	return ks
+}
+
+func (PredicateLWTClause) isUpdateClause() {}
+func (PredicateLWTClause) isDeleteClause() {}

x/cqlbuilder/execer.go

@@ -0,0 +1,71 @@
+package cqlbuilder
+
+import (
+	"context"
+
+	"github.com/upfluence/cql"
+)
+
+type CASScanner interface {
+	ScanCAS(map[string]interface{}) (bool, error)
+}
+
+type errCASScanner struct{ error }
+
+func (ecs errCASScanner) ScanCAS(map[string]interface{}) (bool, error) {
+	return false, ecs.error
+}
+
+type casScanner struct {
+	sc cql.CASScanner
+	ks []string
+}
+
+func (cs *casScanner) ScanCAS(qvs map[string]interface{}) (bool, error) {
+	vs := make([]interface{}, len(cs.ks))
+
+	for i, k := range cs.ks {
+		v, ok := qvs[k]
+
+		if !ok {
+			return false, ErrMissingKey{Key: k}
+		}
+
+		vs[i] = v
+	}
+
+	return cs.sc.ScanCAS(vs...)
+}
+
+type Execer interface {
+	Exec(context.Context, map[string]interface{}) error
+	ExecCAS(context.Context, map[string]interface{}) CASScanner
+}
+
+type execer struct {
+	stmt CASStatement
+	db   cql.DB
+}
+
+func (e *execer) Exec(ctx context.Context, qvs map[string]interface{}) error {
+	var stmt, vs, err = e.stmt.buildQuery(qvs)
+
+	if err != nil {
+		return err
+	}
+
+	return e.db.Exec(ctx, stmt, vs...)
+}
+
+func (e *execer) ExecCAS(ctx context.Context, qvs map[string]interface{}) CASScanner {
+	var stmt, vs, err = e.stmt.buildQuery(qvs)
+
+	if err != nil {
+		return errCASScanner{err}
+	}
+
+	return &casScanner{
+		sc: e.db.ExecCAS(ctx, stmt, vs...),
+		ks: e.stmt.casScanKeys(),
+	}
+}

x/cqlbuilder/insert_statement.go

@@ -0,0 +1,77 @@
+package cqlbuilder
+
+import (
+	"fmt"
+	"strings"
+)
+
+type LWTInsertClause interface {
+	LWTClause
+
+	isInsertClause()
+}
+
+type InsertStatement struct {
+	Table string
+
+	Fields []Marker
+
+	Options   DMLOptions
+	LWTClause LWTInsertClause
+}
+
+func (is InsertStatement) casScanKeys() []string {
+	var ks = make([]string, len(is.Fields))
+
+	for i, f := range is.Fields {
+		ks[i] = f.Binding()
+	}
+
+	return ks
+}
+
+func (is InsertStatement) buildQuery(qvs map[string]interface{}) (string, []interface{}, error) {
+	var (
+		qw queryWriter
+
+		ks = make([]string, len(is.Fields))
+		qs = make([]string, len(is.Fields))
+	)
+
+	if len(is.Fields) == 0 {
+		return "", nil, errNoMarkers
+	}
+
+	for i, f := range is.Fields {
+		k := f.Binding()
+		v, ok := qvs[k]
+
+		if !ok {
+			return "", nil, ErrMissingKey{Key: k}
+		}
+
+		ks[i] = columnName(f)
+		qs[i] = "?"
+		qw.AddArg(v)
+	}
+
+	fmt.Fprintf(
+		&qw,
+		"INSERT INTO %s(%s) VALUES (%s)",
+		is.Table,
+		strings.Join(ks, ", "),
+		strings.Join(qs, ", "),
+	)
+
+	if lc := is.LWTClause; lc != nil {
+		qw.WriteRune(' ')
+
+		if err := lc.writeTo(&qw, qvs); err != nil {
+			return "", nil, err
+		}
+	}
+
+	is.Options.writeTo(&qw)
+
+	return qw.String(), qw.args, nil
+}