// Copyright (c) 2012 The gocql Authors. All rights reserved. // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. package gocql import ( "errors" "fmt" "io" "sync" "time" ) // Session is the interface used by users to interact with the database. // // It extends the Node interface by adding a convinient query builder and // automatically sets a default consinstency level on all operations // that do not have a consistency level set. type Session struct { Node Node Cons Consistency } // NewSession wraps an existing Node. func NewSession(node Node) *Session { return &Session{Node: node, Cons: Quorum} } // Query generates a new query object for interacting with the database. // Further details of the query may be tweaked using the resulting query // value before the query is executed. func (s *Session) Query(stmt string, values ...interface{}) *Query { return &Query{stmt: stmt, values: values, cons: s.Cons, session: s} } // Close closes all connections. The session is unusable after this // operation. func (s *Session) Close() { s.Node.Close() } func (s *Session) executeQuery(qry *Query) *Iter { conn := s.Node.Pick(nil) if conn == nil { return &Iter{qry: qry, err: ErrUnavailable} } return conn.executeQuery(qry) } func (s *Session) ExecuteBatch(batch *Batch) error { conn := s.Node.Pick(nil) if conn == nil { return ErrUnavailable } return conn.executeBatch(batch) } // Query represents a CQL statement that can be executed. type Query struct { stmt string values []interface{} cons Consistency pageSize int pageState []byte trace Tracer session *Session } // Consistency sets the consistency level for this query. If no consistency // level have been set, the default consistency level of the cluster // is used. func (q *Query) Consistency(c Consistency) *Query { q.cons = c return q } // Trace enables tracing of this query. Look at the documentation of the // Tracer interface to learn more about tracing. func (q *Query) Trace(trace Tracer) *Query { q.trace = trace return q } // PageSize will tell the iterator to fetch the result in pages of size n. // This is useful for iterating over large result sets, but setting the // page size to low might decrease the performance. This feature is only // available in Cassandra 2 and onwards. func (q *Query) PageSize(n int) *Query { q.pageSize = n return q } // Exec executes the query without returning any rows. func (q *Query) Exec() error { iter := q.session.executeQuery(q) return iter.err } // Iter executes the query and returns an iterator capable of iterating // over all results. func (q *Query) Iter() *Iter { return q.session.executeQuery(q) } // Scan executes the query, copies the columns of the first selected // row into the values pointed at by dest and discards the rest. If no rows // were selected, ErrNotFound is returned. func (q *Query) Scan(dest ...interface{}) error { iter := q.Iter() iter.Scan(dest...) return iter.Close() } type Iter struct { err error pos int rows [][][]byte columns []ColumnInfo qry *Query pageState []byte } func (iter *Iter) Columns() []ColumnInfo { return iter.columns } func (iter *Iter) Scan(dest ...interface{}) bool { if iter.err != nil { return false } if iter.pos >= len(iter.rows) { if len(iter.pageState) > 0 { qry := *iter.qry qry.pageState = iter.pageState *iter = *iter.qry.session.executeQuery(&qry) return iter.Scan(dest...) } return false } if len(dest) != len(iter.columns) { iter.err = errors.New("count mismatch") return false } for i := 0; i < len(iter.columns); i++ { err := Unmarshal(iter.columns[i].TypeInfo, iter.rows[iter.pos][i], dest[i]) if err != nil { iter.err = err return false } } iter.pos++ return true } func (iter *Iter) Close() error { return iter.err } type Batch struct { Type BatchType Entries []BatchEntry Cons Consistency } func NewBatch(typ BatchType) *Batch { return &Batch{Type: typ} } func (b *Batch) Query(stmt string, args ...interface{}) { b.Entries = append(b.Entries, BatchEntry{Stmt: stmt, Args: args}) } type BatchType int const ( LoggedBatch BatchType = 0 UnloggedBatch BatchType = 1 CounterBatch BatchType = 2 ) type BatchEntry struct { Stmt string Args []interface{} } type Consistency int const ( Any Consistency = 1 + iota One Two Three Quorum All LocalQuorum EachQuorum Serial LocalSerial ) var consinstencyNames = []string{ 0: "default", Any: "any", One: "one", Two: "two", Three: "three", Quorum: "quorum", All: "all", LocalQuorum: "localquorum", EachQuorum: "eachquorum", Serial: "serial", LocalSerial: "localserial", } func (c Consistency) String() string { return consinstencyNames[c] } type ColumnInfo struct { Keyspace string Table string Name string TypeInfo *TypeInfo } // Tracer is the interface implemented by query tracers. Tracers have the // ability to obtain a detailed event log of all events that happened during // the execution of a query from Cassandra. Gathering this information might // be essential for debugging and optimizing queries, but this feature should // not be used on production systems with very high load. type Tracer interface { Trace(traceId []byte) } type traceWriter struct { session *Session w io.Writer mu sync.Mutex } // NewTraceWriter returns a simple Tracer implementation that outputs // the event log in a textual format. func NewTraceWriter(session *Session, w io.Writer) Tracer { return traceWriter{session: session, w: w} } func (t traceWriter) Trace(traceId []byte) { var ( coordinator string duration int ) t.session.Query(`SELECT coordinator, duration FROM system_traces.sessions WHERE session_id = ?`, traceId). Consistency(One).Scan(&coordinator, &duration) iter := t.session.Query(`SELECT event_id, activity, source, source_elapsed FROM system_traces.events WHERE session_id = ?`, traceId). Consistency(One).Iter() var ( timestamp time.Time activity string source string elapsed int ) t.mu.Lock() defer t.mu.Unlock() fmt.Fprintf(t.w, "Tracing session %016x (coordinator: %s, duration: %v):\n", traceId, coordinator, time.Duration(duration)*time.Microsecond) for iter.Scan(×tamp, &activity, &source, &elapsed) { fmt.Fprintf(t.w, "%s: %s (source: %s, elapsed: %d)\n", timestamp.Format("2006/01/02 15:04:05.999999"), activity, source, elapsed) } if err := iter.Close(); err != nil { fmt.Fprintln(t.w, "Error:", err) } } type Error struct { Code int Message string } func (e Error) Error() string { return e.Message } var ( ErrNotFound = errors.New("not found") ErrUnavailable = errors.New("unavailable") ErrProtocol = errors.New("protocol error") ErrUnsupported = errors.New("feature not supported") )