This is an automated email from the ASF dual-hosted git repository.
joao-r-reis pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-gocql-driver.git
The following commit(s) were added to refs/heads/trunk by this push:
new 9bc3b5de Add PreparedMetadata and IsPrepared fields to ObservedQuery
and ObservedBatch
9bc3b5de is described below
commit 9bc3b5dee806f1847db37262c5ec510d2f417140
Author: Raj <[email protected]>
AuthorDate: Tue May 19 17:23:45 2026 -0700
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
---
CHANGELOG.md | 1 +
cassandra_test.go | 95 ++++++++++++++++++++++++++++++++++++++++++++++++-------
conn.go | 13 ++++++++
query_executor.go | 62 ++++++++++++++++++++++++++----------
session.go | 35 ++++++++++++++++++++
5 files changed, 179 insertions(+), 27 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 3f132b5b..1e34ac1d 100644
--- a/CHANGELOG.md
+++ b/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]
diff --git a/cassandra_test.go b/cassandra_test.go
index 2f386c80..eda3eb9e 100644
--- a/cassandra_test.go
+++ b/cassandra_test.go
@@ -172,9 +172,12 @@ 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"
@@ -182,12 +185,18 @@ func TestObserve(t *testing.T) {
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])
}
}
diff --git a/conn.go b/conn.go
index 28b01577..54122445 100644
--- a/conn.go
+++ b/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
diff --git a/query_executor.go b/query_executor.go
index 2d7a6233..5d848c2e 100644
--- a/query_executor.go
+++ b/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()
}
diff --git a/session.go b/session.go
index fcbdd703..dee2a8fe 100644
--- a/session.go
+++ b/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.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
