Add PreparedMetadata and IsPrepared fields to ObservedQuery and ObservedBatch

ObservedQuery and ObservedBatch did not expose the keyspace/table targeted
by a query, requiring consumers (metrics collectors, loggers, tracers) to
parse the CQL statement string.

This change introduces a PreparedMetadata struct (Keyspace, Table) on
ObservedQuery and a parallel []PreparedMetadata slice on ObservedBatch,
populated from prepared statement metadata returned by Cassandra. An
IsPrepared bool on ObservedQuery and a parallel []bool on ObservedBatch
explicitly report when the metadata is valid; statements that take the
unprepared path (e.g. DDL) report IsPrepared=false and zero PreparedMetadata.

The information was already available internally via preparedMetadata
(used for routing), so no additional round-trips or CQL parsing are needed.

Updated TestObserve (including a negative case for an unprepared DDL
statement) and TestBatchObserve to verify the new fields.

Patch by Raj Ummadisetty; reviewed by João Reis, Bohdan Siryk for CASSGO-119

Author: raj14243

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Security-model discoverability (CASSANDRA-21464)
 - Improve host_source locking and ring refresh concurrency (CASSGO-121)
+- Add PreparedMetadata (Keyspace, Table) and IsPrepared fields to ObservedQuery, and parallel PreparedMetadata / IsPrepared slices to ObservedBatch, for statement-level observability without CQL parsing (CASSGO-119)
 
 ## [2.1.2]
 

cassandra_test.go

@@ -172,22 +172,31 @@ func TestObserve(t *testing.T) {
 	}
 
 	var (
-		observedErr      error
-		observedKeyspace string
-		observedStmt     string
+		observedErr              error
+		observedKeyspace         string
+		observedPreparedKeyspace string
+		observedStmt             string
+		observedPreparedTable    string
+		observedIsPrepared       bool
 	)
 
 	const keyspace = "gocql_test"
 
 	resetObserved := func() {
 		observedErr = errors.New("placeholder only") // used to distinguish err=nil cases
 		observedKeyspace = ""
+		observedPreparedKeyspace = ""
 		observedStmt = ""
+		observedPreparedTable = ""
+		observedIsPrepared = false
 	}
 
 	observer := funcQueryObserver(func(ctx context.Context, o ObservedQuery) {
 		observedKeyspace = o.Keyspace
+		observedPreparedKeyspace = o.PreparedMetadata.Keyspace
 		observedStmt = o.Statement
+		observedPreparedTable = o.PreparedMetadata.Table
+		observedIsPrepared = o.IsPrepared
 		observedErr = o.Err
 	})
 
@@ -202,6 +211,12 @@ func TestObserve(t *testing.T) {
 		t.Fatal("select: unexpected observed keyspace", observedKeyspace)
 	} else if observedStmt != `SELECT id FROM observe WHERE id = ?` {
 		t.Fatal("select: unexpected observed stmt", observedStmt)
+	} else if !observedIsPrepared {
+		t.Fatal("select: expected observed IsPrepared to be true")
+	} else if observedPreparedKeyspace != keyspace {
+		t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace)
+	} else if observedPreparedTable != "observe" {
+		t.Fatal("select: unexpected observed prepared table", observedPreparedTable)
 	}
 
 	resetObserved()
@@ -213,6 +228,12 @@ func TestObserve(t *testing.T) {
 		t.Fatal("insert: unexpected observed keyspace", observedKeyspace)
 	} else if observedStmt != `INSERT INTO observe (id) VALUES (?)` {
 		t.Fatal("insert: unexpected observed stmt", observedStmt)
+	} else if !observedIsPrepared {
+		t.Fatal("insert: expected observed IsPrepared to be true")
+	} else if observedPreparedKeyspace != keyspace {
+		t.Fatal("insert: unexpected observed prepared keyspace", observedPreparedKeyspace)
+	} else if observedPreparedTable != "observe" {
+		t.Fatal("insert: unexpected observed prepared table", observedPreparedTable)
 	}
 
 	resetObserved()
@@ -227,6 +248,12 @@ func TestObserve(t *testing.T) {
 		t.Fatal("select: unexpected observed keyspace", observedKeyspace)
 	} else if observedStmt != `SELECT id FROM observe WHERE id = ?` {
 		t.Fatal("select: unexpected observed stmt", observedStmt)
+	} else if !observedIsPrepared {
+		t.Fatal("select: expected observed IsPrepared to be true")
+	} else if observedPreparedKeyspace != keyspace {
+		t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace)
+	} else if observedPreparedTable != "observe" {
+		t.Fatal("select: unexpected observed prepared table", observedPreparedTable)
 	}
 
 	// also works from session observer
@@ -240,6 +267,12 @@ func TestObserve(t *testing.T) {
 		t.Fatal("select: unexpected observed keyspace", observedKeyspace)
 	} else if observedStmt != `SELECT id FROM observe WHERE id = ?` {
 		t.Fatal("select: unexpected observed stmt", observedStmt)
+	} else if !observedIsPrepared {
+		t.Fatal("select: expected observed IsPrepared to be true")
+	} else if observedPreparedKeyspace != keyspace {
+		t.Fatal("select: unexpected observed prepared keyspace", observedPreparedKeyspace)
+	} else if observedPreparedTable != "observe" {
+		t.Fatal("select: unexpected observed prepared table", observedPreparedTable)
 	}
 
 	// reports errors when the query is poorly formed
@@ -254,6 +287,24 @@ func TestObserve(t *testing.T) {
 	} else if observedStmt != `SELECT id FROM unknown_table WHERE id = ?` {
 		t.Fatal("select: unexpected observed stmt", observedStmt)
 	}
+
+	// statements that cannot be prepared (e.g. DDL) report IsPrepared=false and
+	// a zero PreparedMetadata, so the contract is explicit for observers.
+	resetObserved()
+	const ddlStmt = `CREATE TABLE IF NOT EXISTS gocql_test.observe_unprepared (id int primary key)`
+	if err := session.Query(ddlStmt).Observer(observer).Exec(); err != nil {
+		t.Fatal("ddl:", err)
+	} else if observedErr != nil {
+		t.Fatal("ddl:", observedErr)
+	} else if observedStmt != ddlStmt {
+		t.Fatal("ddl: unexpected observed stmt", observedStmt)
+	} else if observedIsPrepared {
+		t.Fatal("ddl: expected observed IsPrepared to be false for unprepared statement")
+	} else if observedPreparedKeyspace != "" {
+		t.Fatal("ddl: expected zero PreparedMetadata.Keyspace, got", observedPreparedKeyspace)
+	} else if observedPreparedTable != "" {
+		t.Fatal("ddl: expected zero PreparedMetadata.Table, got", observedPreparedTable)
+	}
 }
 
 func TestObserve_Pagination(t *testing.T) {
@@ -2134,10 +2185,12 @@ func TestBatchObserve(t *testing.T) {
 	}
 
 	type observation struct {
-		observedErr      error
-		observedKeyspace string
-		observedStmts    []string
-		observedValues   [][]interface{}
+		observedErr              error
+		observedKeyspace         string
+		observedPreparedMetadata []PreparedMetadata
+		observedStmts            []string
+		observedIsPrepared       []bool
+		observedValues           [][]interface{}
 	}
 
 	var observedBatch *observation
@@ -2149,10 +2202,12 @@ func TestBatchObserve(t *testing.T) {
 		}
 
 		observedBatch = &observation{
-			observedKeyspace: o.Keyspace,
-			observedStmts:    o.Statements,
-			observedErr:      o.Err,
-			observedValues:   o.Values,
+			observedKeyspace:         o.Keyspace,
+			observedPreparedMetadata: o.PreparedMetadata,
+			observedStmts:            o.Statements,
+			observedIsPrepared:       o.IsPrepared,
+			observedErr:              o.Err,
+			observedValues:           o.Values,
 		}
 	}))
 	for i := 0; i < 100; i++ {
@@ -2175,11 +2230,29 @@ func TestBatchObserve(t *testing.T) {
 	if observedBatch.observedKeyspace != "gocql_test" {
 		t.Fatalf("expecting keyspace 'gocql_test', got %q", observedBatch.observedKeyspace)
 	}
+	if len(observedBatch.observedPreparedMetadata) != 100 {
+		t.Fatal("expecting 100 observed prepared metadata entries, got", len(observedBatch.observedPreparedMetadata))
+	}
+	if len(observedBatch.observedIsPrepared) != 100 {
+		t.Fatal("expecting 100 observed IsPrepared flags, got", len(observedBatch.observedIsPrepared))
+	}
 	for i, stmt := range observedBatch.observedStmts {
 		if stmt != fmt.Sprintf(`INSERT INTO batch_observe_table (id,other) VALUES (?,%d)`, i) {
 			t.Fatal("unexpected query", stmt)
 		}
 
+		if !observedBatch.observedIsPrepared[i] {
+			t.Fatalf("expected observed IsPrepared at index %d to be true", i)
+		}
+
+		if observedBatch.observedPreparedMetadata[i].Keyspace != "gocql_test" {
+			t.Fatalf("unexpected observed prepared keyspace at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Keyspace)
+		}
+
+		if observedBatch.observedPreparedMetadata[i].Table != "batch_observe_table" {
+			t.Fatalf("unexpected observed prepared table at index %d: %q", i, observedBatch.observedPreparedMetadata[i].Table)
+		}
+
 		assertDeepEqual(t, "observed value", []interface{}{i}, observedBatch.observedValues[i])
 	}
 }

conn.go

@@ -1661,6 +1661,7 @@ func (c *Conn) executeQuery(ctx context.Context, q *internalQuery) *Iter {
 			q.routingInfo.keyspace = usedKeyspace
 		}
 		q.routingInfo.table = info.request.table
+		q.routingInfo.prepared = true
 		q.routingInfo.mu.Unlock()
 	} else {
 		frame = &writeQueryFrame{
@@ -1856,6 +1857,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
 
 	stmts := make(map[string]string, len(b.batchOpts.entries))
 
+	if b.batchOpts.observer != nil {
+		b.entryKeyspaces = make([]string, n)
+		b.entryTables = make([]string, n)
+		b.entryPrepared = make([]bool, n)
+	}
+
 	for i := 0; i < n; i++ {
 		entry := &b.batchOpts.entries[i]
 		batchStmt := &req.statements[i]
@@ -1867,6 +1874,12 @@ func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
 				return iter
 			}
 
+			if b.entryTables != nil {
+				b.entryKeyspaces[i] = info.request.keyspace
+				b.entryTables[i] = info.request.table
+				b.entryPrepared[i] = true
+			}
+
 			var values []interface{}
 			if entry.binding == nil {
 				values = entry.Args

query_executor.go

@@ -385,15 +385,20 @@ func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Ite
 		q.qryOpts.observer.ObserveQuery(q.qryOpts.context, ObservedQuery{
 			Keyspace:  keyspace,
 			Statement: q.qryOpts.stmt,
-			Values:    q.qryOpts.values,
-			Start:     start,
-			End:       end,
-			Rows:      iter.numRows,
-			Host:      host,
-			Metrics:   metricsForHost,
-			Err:       iter.err,
-			Attempt:   attempt,
-			Query:     q.originalQuery,
+			PreparedMetadata: PreparedMetadata{
+				Keyspace: q.routingInfo.getKeyspace(),
+				Table:    q.routingInfo.getTable(),
+			},
+			IsPrepared: q.routingInfo.isPrepared(),
+			Values:     q.qryOpts.values,
+			Start:      start,
+			End:        end,
+			Rows:       iter.numRows,
+			Host:       host,
+			Metrics:    metricsForHost,
+			Err:        iter.err,
+			Attempt:    attempt,
+			Query:      q.originalQuery,
 		})
 	}
 }
@@ -432,6 +437,7 @@ func (q *internalQuery) GetRoutingKey() ([]byte, error) {
 		q.routingInfo.mu.Lock()
 		q.routingInfo.keyspace = meta.Keyspace
 		q.routingInfo.table = meta.Table
+		q.routingInfo.prepared = true
 		q.routingInfo.mu.Unlock()
 	}
 	return createRoutingKey(meta, q.qryOpts.values)
@@ -562,6 +568,19 @@ type internalBatch struct {
 	session            *Session
 	metrics            *queryMetrics
 	hostMetricsManager hostMetricsManager
+
+	// entryKeyspaces holds the keyspace for each batch entry,
+	// populated from prepared statement metadata during execution.
+	entryKeyspaces []string
+
+	// entryTables holds the table name for each batch entry,
+	// populated from prepared statement metadata during execution.
+	entryTables []string
+
+	// entryPrepared[i] reports whether the i-th batch entry was prepared by
+	// the driver, and therefore whether entryKeyspaces[i]/entryTables[i] hold
+	// valid prepared statement metadata.
+	entryPrepared []bool
 }
 
 func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch {
@@ -597,20 +616,30 @@ func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Ite
 
 	metricsForHost := b.hostMetricsManager.attempt(latency, host)
 
-	statements := make([]string, len(b.batchOpts.entries))
-	values := make([][]interface{}, len(b.batchOpts.entries))
+	n := len(b.batchOpts.entries)
+	statements := make([]string, n)
+	values := make([][]interface{}, n)
+	preparedMetadata := make([]PreparedMetadata, n)
 
 	for i, entry := range b.batchOpts.entries {
 		statements[i] = entry.Stmt
 		values[i] = entry.Args
+		if i < len(b.entryKeyspaces) {
+			preparedMetadata[i] = PreparedMetadata{
+				Keyspace: b.entryKeyspaces[i],
+				Table:    b.entryTables[i],
+			}
+		}
 	}
 
 	b.batchOpts.observer.ObserveBatch(b.batchOpts.context, ObservedBatch{
-		Keyspace:   keyspace,
-		Statements: statements,
-		Values:     values,
-		Start:      start,
-		End:        end,
+		Keyspace:         keyspace,
+		Statements:       statements,
+		PreparedMetadata: preparedMetadata,
+		IsPrepared:       b.entryPrepared,
+		Values:           values,
+		Start:            start,
+		End:              end,
 		// Rows not used in batch observations // TODO - might be able to support it when using BatchCAS
 		Host:    host,
 		Metrics: metricsForHost,
@@ -652,6 +681,7 @@ func (b *internalBatch) GetRoutingKey() ([]byte, error) {
 		b.routingInfo.mu.Lock()
 		b.routingInfo.keyspace = meta.Keyspace
 		b.routingInfo.table = meta.Table
+		b.routingInfo.prepared = true
 		b.routingInfo.mu.Unlock()
 	}
 

session.go

@@ -1086,6 +1086,8 @@ type queryRoutingInfo struct {
 	keyspace string
 
 	table string
+
+	prepared bool
 }
 
 func (qr *queryRoutingInfo) getKeyspace() string {
@@ -1100,6 +1102,12 @@ func (qr *queryRoutingInfo) getTable() string {
 	return qr.table
 }
 
+func (qr *queryRoutingInfo) isPrepared() bool {
+	qr.mu.RLock()
+	defer qr.mu.RUnlock()
+	return qr.prepared
+}
+
 func (q *Query) defaultsFromSession() {
 	s := q.session
 
@@ -2370,10 +2378,28 @@ func (s *Session) GetHosts() []*HostInfo {
 	return s.ring.allHosts()
 }
 
+// PreparedMetadata holds metadata extracted from a prepared statement
+// (returned by Cassandra during statement preparation). It is exposed via
+// ObservedQuery and ObservedBatch when the driver has prepared the statement.
+type PreparedMetadata struct {
+	Keyspace string
+	Table    string
+}
+
 type ObservedQuery struct {
 	Keyspace  string
 	Statement string
 
+	// PreparedMetadata holds keyspace/table information from prepared statement
+	// metadata returned by Cassandra. Only valid when IsPrepared is true; the
+	// zero value is reported otherwise (for example, for DDL statements or
+	// statements not prepared by the driver).
+	PreparedMetadata PreparedMetadata
+
+	// IsPrepared reports whether the statement was prepared by the driver and
+	// therefore whether PreparedMetadata holds valid information.
+	IsPrepared bool
+
 	// Values holds a slice of bound values for the query.
 	// Do not modify the values here, they are shared with multiple goroutines.
 	Values []interface{}
@@ -2418,6 +2444,15 @@ type ObservedBatch struct {
 	Keyspace   string
 	Statements []string
 
+	// PreparedMetadata holds prepared statement metadata for each batch statement.
+	// PreparedMetadata[i] corresponds to Statements[i]. Each entry is only valid
+	// when IsPrepared[i] is true; the zero value is reported otherwise.
+	PreparedMetadata []PreparedMetadata
+
+	// IsPrepared reports, for each batch statement, whether the statement was
+	// prepared by the driver. IsPrepared[i] corresponds to Statements[i].
+	IsPrepared []bool
+
 	// Values holds a slice of bound values for each statement.
 	// Values[i] are bound values passed to Statements[i].
 	// Do not modify the values here, they are shared with multiple goroutines.