You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "joao-r-reis (via GitHub)" <gi...@apache.org> on 2025/03/13 15:41:08 UTC
[PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
joao-r-reis opened a new pull request, #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868
Going to edit this summary soon with a summary of the changes that I'm proposing with this PR. Until I've done that I'm keeping this in DRAFT status
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2035606353
##########
conn.go:
##########
@@ -1371,76 +1376,80 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
value := values[i]
typ := info.request.columns[i].TypeInfo
if err := marshalQueryValue(typ, value, v); err != nil {
- return &Iter{err: err}
+ iter.err = err
+ return iter
}
}
// if the metadata was not present in the response then we should not skip it
- params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qry.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
+ params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qryOpts.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
frame = &writeExecuteFrame{
preparedID: info.id,
params: params,
- customPayload: qry.customPayload,
+ customPayload: qryOpts.customPayload,
}
// Set "keyspace" and "table" property in the query if it is present in preparedMetadata
- qry.routingInfo.mu.Lock()
- qry.routingInfo.keyspace = info.request.keyspace
- qry.routingInfo.table = info.request.table
- qry.routingInfo.mu.Unlock()
+ q.routingInfo.mu.Lock()
Review Comment:
So I made some changes to the `routingInfo` part but I didn't remove the struct... I think it still serves a purpose by making it clear that the mutex is used to protect access against the two fields within the struct. Do you still prefer to remove the struct and just have the mutex and two fields in the request type itself?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044760057
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
Well the reason for this is to ensure that a query execution isn't affected if the user tries to modify the page byte slice after query has been submitted for execution
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044803406
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
Review Comment:
oops, maybe I added it back during rebase, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2749798683
> With this PR, this method is deprecated on Query and Batch since these two types no longer implement ExecutableQuery so we can keep the Query/Batch API much simpler.
Why not just remove it? Seems like any `HostSelectionPolicy` implementation would need to make major changes already.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2733808873
> I am only a bit concerned about the changes this PR provides to the driver internals, especially the conn part which overlaps with https://github.com/apache/cassandra-gocql-driver/pull/1822.
Yeah I'll have to spend some time rebasing this branch and even writing some tests, for now it's good enough if reviewers can provide feedback on the current state of the PR especially the public API changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2751405836
> > With this PR, this method is deprecated on Query and Batch since these two types no longer implement ExecutableQuery so we can keep the Query/Batch API much simpler.
>
> Why not just remove it? Seems like any `HostSelectionPolicy` implementation would need to make major changes already.
Hmm I don't think `HostSelectionPolicy` implementations have to make major changes after this PR unless they rely on casting the `ExecutableQuery` object into `Query` or `Batch` (and even this is a small change, they just need to call .Statement() before doing so).
> I don't want to bikeshed about names but ExecutableQuery is a bit confusing since it contains Query and Batch. Did that name come from somewhere? Would ExecutableStatement make more sense since it has a statement?
I kept `ExecutableQuery` name unchanged so `HostSelectionPolicy` implementations can pretty much continue to work without many changes (or even any change at all). cc @lukasz-antoniak because you also brought this up in 1 of your comments.
I'm concerned that renaming this interface will increase the chance users will have to change their code when upgrading the driver but I do agree that the current name isn't good.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "ribaraka (via GitHub)" <gi...@apache.org>.
ribaraka commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2052075937
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
I mean that `newPageState` is copied but never used. And it doesn't seem the `queryOptions` struct has such a field.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2035667504
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
Review Comment:
Something like this:
```
type queryOptions struct {
stmt string
// Paging
pageSize int
initialPageState []byte
disableAutoPage bool
// Monitoring
trace Tracer
observer QueryObserver
// Parameters
values []interface{}
binding func(q *QueryInfo) ([]interface{}, error)
// Timestamp
defaultTimestamp bool
defaultTimestampValue int64
// Consistency
initialConsistency Consistency
serialCons SerialConsistency
// Protocol flag
disableSkipMetadata bool
customPayload map[string][]byte
prefetch float64
rt RetryPolicy
spec SpeculativeExecutionPolicy
context context.Context
idempotent bool
keyspace string
skipPrepare bool
routingKey []byte
// getKeyspace is field so that it can be overriden in tests
getKeyspace func() string
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012217212
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.Context(), q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
Review Comment:
The type itself is not public and it is never exposed so it wasn't a deliberate choice, just a result of copy paste. I can make them all private for consistency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012241514
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
Review Comment:
yeah I'll do this
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044793118
##########
session.go:
##########
@@ -743,33 +736,47 @@ func (s *Session) routingKeyInfo(ctx context.Context, stmt string, keyspace stri
return routingKeyInfo, nil
}
-func (b *Batch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
-}
-
// Exec executes a batch operation and returns nil if successful
// otherwise an error is returned describing the failure.
func (b *Batch) Exec() error {
- iter := b.session.executeBatch(b)
+ iter := b.session.executeBatch(b, nil)
+ return iter.Close()
+}
+
+// ExecContext executes a batch operation with the provided context and returns nil if successful
+// otherwise an error is returned describing the failure.
+func (b *Batch) ExecContext(ctx context.Context) error {
+ iter := b.session.executeBatch(b, ctx)
return iter.Close()
}
-func (s *Session) executeBatch(batch *Batch) *Iter {
+// Iter executes a batch operation and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) Iter() *Iter { return b.session.executeBatch(b, nil) }
+
+// Iter executes a batch operation with the provided context and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) IterContext(ctx context.Context) *Iter {
+ return b.session.executeBatch(b, ctx)
+}
Review Comment:
done
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
+ if q.routingKey != nil {
+ routingKey := q.routingKey
+ newRoutingKey = make([]byte, len(routingKey))
+ copy(newRoutingKey, routingKey)
+ }
+ if ctx == nil {
+ ctx = q.Context()
+ }
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ pageSize: q.pageSize,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: ctx,
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: newRoutingKey,
+ getKeyspace: q.getKeyspace,
+ nowInSecondsValue: q.nowInSecondsValue,
+ keyspace: q.keyspace,
+ hostID: q.hostID,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query, ctx context.Context) *internalQuery {
+ var newPageState []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q, ctx),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: newPageState,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.Context(), q.qryOpts.stmt, q.qryOpts.keyspace)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+ if q.qryOpts.keyspace != "" {
+ return q.qryOpts.keyspace
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+func (q *internalQuery) GetHostID() string {
+ return q.qryOpts.hostID
+}
+
+func (q *internalQuery) getRoutingInfo() *queryRoutingInfo {
+ return q.routingInfo
+}
+
+func (q *internalQuery) getKeyspaceFunc() func() string {
+ return q.qryOpts.getKeyspace
+}
+
+type batchOptions struct {
+ trace Tracer
+ observer BatchObserver
+
+ bType BatchType
+ entries []BatchEntry
+
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ serialCons SerialConsistency
+
+ customPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+ nowInSeconds *int
+}
+
+func newBatchOptions(b *Batch, ctx context.Context) *batchOptions {
+ // make a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ var newRoutingKey []byte
+ if b.routingKey != nil {
+ routingKey := b.routingKey
+ newRoutingKey = make([]byte, len(routingKey))
+ copy(newRoutingKey, routingKey)
+ }
+ if ctx == nil {
+ ctx = b.Context()
+ }
+ return &batchOptions{
+ bType: b.Type,
+ entries: newEntries,
+ customPayload: b.CustomPayload,
+ rt: b.rt,
+ spec: b.spec,
+ trace: b.trace,
+ observer: b.observer,
+ serialCons: b.serialCons,
+ defaultTimestamp: b.defaultTimestamp,
+ defaultTimestampValue: b.defaultTimestampValue,
+ context: ctx,
+ keyspace: b.Keyspace(),
+ idempotent: b.IsIdempotent(),
+ routingKey: newRoutingKey,
+ nowInSeconds: b.nowInSeconds,
+ }
+}
+
+type internalBatch struct {
+ originalBatch *Batch
+ batchOpts *batchOptions
+ metrics *queryMetrics
+ consistency uint32
+ routingInfo *queryRoutingInfo
+ session *Session
+}
+
+func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch {
+ return &internalBatch{
+ originalBatch: batch,
+ batchOpts: newBatchOptions(batch, ctx),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ routingInfo: &queryRoutingInfo{},
+ session: batch.session,
+ }
+}
Review Comment:
nice catch, done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044794228
##########
query_executor.go:
##########
@@ -27,33 +27,51 @@ package gocql
import (
"context"
"sync"
+ "sync/atomic"
"time"
)
-type ExecutableQuery interface {
- borrowForExecution() // Used to ensure that the query stays alive for lifetime of a particular execution goroutine.
- releaseAfterExecution() // Used when a goroutine finishes its execution attempts, either with ok result or an error.
- execute(ctx context.Context, conn *Conn) *Iter
- attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo)
- retryPolicy() RetryPolicy
- speculativeExecutionPolicy() SpeculativeExecutionPolicy
+// Deprecated: Will be removed in a future major release. Also Query and Batch no longer implement this interface.
+//
+// Please use Statement (for Query / Batch objects) or ExecutableStatement (in HostSelectionPolicy implementations) instead.
+type ExecutableQuery = ExecutableStatement
+
+// ExecutableStatement is an interface that represents a query or batch statement that
+// exposes the correct functions for the HostSelectionPolicy to operate correctly.
+type ExecutableStatement interface {
GetRoutingKey() ([]byte, error)
Keyspace() string
Table() string
IsIdempotent() bool
GetHostID() string
+ Statement() Statement
+}
- withContext(context.Context) ExecutableQuery
+// Statement is an interface that represents a CQL statement that the driver can execute
+// (currently Query and Batch via Session.Query and Session.Batch)
+type Statement interface {
+ Iter() *Iter
+ Exec() error
+}
Review Comment:
good suggestion 👍 done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2792550710
Rebase is done, it was a bit more complex than I thought it would be so I'd appreciate if you guys could take a look again @jameshartig @lukasz-antoniak @worryg0d
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012234094
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
Review Comment:
I think copying `values` and `customPayload` might be a bit too much due to performance reasons. `initialPageState` should probably be copied for safety and since its length is usually pretty small. `routingKey` I'm not so sure, I think it will be `nil` the vast majority of cases anyway but I can make sure it gets copied as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012243301
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
Review Comment:
Good catch, leftover from a copy paste
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012189609
##########
session.go:
##########
@@ -727,33 +720,36 @@ func (s *Session) routingKeyInfo(ctx context.Context, stmt string) (*routingKeyI
return routingKeyInfo, nil
}
-func (b *Batch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
-}
-
// Exec executes a batch operation and returns nil if successful
// otherwise an error is returned describing the failure.
func (b *Batch) Exec() error {
iter := b.session.executeBatch(b)
return iter.Close()
}
+// Iter executes a batch operation and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) Iter() *Iter {
+ return b.session.executeBatch(b)
Review Comment:
> I feel that people may execute sequence of `Exec()` and `Iter()`. Shall we make an assertion that only one of the is invoked?
They can't execute sequence (`Exec().Iter()`) unless they do it like:
```
err := b.Exec()
iter := b.Iter()
```
Which I believe is already something that can be done with the current API and it will trigger two requests. We can try to improve documentation if this is a concern.
+1 to what James said
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2035661539
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
Review Comment:
On a second thought I'm not sure if I like this idea that much, I'd have to create a few new types when maybe I can just group them a bit more by arranging the way they are declared... Wdyt
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "lukasz-antoniak (via GitHub)" <gi...@apache.org>.
lukasz-antoniak commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2009652194
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
Review Comment:
Shall we group here all relevant options into sub-types, for example:
```
Paging
pageSize
initialPageState
disableAutoPage
Monitoring
observer
trace
Parameters
values
binding
WriteTimestamp
defaultTimestamp
defaultTimestampValue
Consistency
initialConsistency
serialConsistency
FrameOptions
customPayload
disableSkipMetadata
stmt
prefetch
rt
spec
context
idempotent
keyspace
skipPrepare
routingKey
```
Grouping is not definitive, it is just an idea.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "worryg0d (via GitHub)" <gi...@apache.org>.
worryg0d commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2000954850
##########
conn.go:
##########
@@ -1523,32 +1534,35 @@ func (c *Conn) UseKeyspace(keyspace string) error {
return nil
}
-func (c *Conn) executeBatch(ctx context.Context, batch *Batch) *Iter {
+func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
if c.version == protoVersion1 {
- return &Iter{err: ErrUnsupported}
+ return newErrIter(ErrUnsupported, b.metrics)
}
- n := len(batch.Entries)
+ iter := newIter(b.metrics)
+
+ n := len(b.batchOpts.Entries)
req := &writeBatchFrame{
- typ: batch.Type,
+ typ: b.batchOpts.Type,
statements: make([]batchStatment, n),
- consistency: batch.Cons,
- serialConsistency: batch.serialCons,
- defaultTimestamp: batch.defaultTimestamp,
- defaultTimestampValue: batch.defaultTimestampValue,
- customPayload: batch.CustomPayload,
+ consistency: b.GetConsistency(),
+ serialConsistency: b.batchOpts.serialCons,
+ defaultTimestamp: b.batchOpts.defaultTimestamp,
+ defaultTimestampValue: b.batchOpts.defaultTimestampValue,
+ customPayload: b.batchOpts.CustomPayload,
}
- stmts := make(map[string]string, len(batch.Entries))
+ stmts := make(map[string]string, len(b.batchOpts.Entries))
for i := 0; i < n; i++ {
- entry := &batch.Entries[i]
- b := &req.statements[i]
+ entry := &b.batchOpts.Entries[i]
+ batchStmt := &req.statements[i]
if len(entry.Args) > 0 || entry.binding != nil {
- info, err := c.prepareStatement(batch.Context(), entry.Stmt, batch.trace)
+ info, err := c.prepareStatement(b.batchOpts.context, entry.Stmt, b.batchOpts.trace)
Review Comment:
The only usage of `ctx` is at line 1634 and that is a recursive call to `executeBatch` on an unprepared Cassandra response so changing `b.batchOpts.context` to `ctx` will make the code less confusing and clearer
##########
query_executor.go:
##########
@@ -201,16 +210,385 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.qryOpts.context, q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
+ Entries []BatchEntry
+ CustomPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ trace Tracer
+ observer BatchObserver
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+}
+
+func newBatchOptions(b *Batch) *batchOptions {
+ // make we get a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ return &batchOptions{
+ Type: b.Type,
+ Entries: b.Entries,
Review Comment:
Here you should use created `newEntries` instead of `b.Entries`
##########
conn.go:
##########
@@ -1523,32 +1534,35 @@ func (c *Conn) UseKeyspace(keyspace string) error {
return nil
}
-func (c *Conn) executeBatch(ctx context.Context, batch *Batch) *Iter {
+func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
if c.version == protoVersion1 {
- return &Iter{err: ErrUnsupported}
+ return newErrIter(ErrUnsupported, b.metrics)
}
- n := len(batch.Entries)
+ iter := newIter(b.metrics)
+
+ n := len(b.batchOpts.Entries)
req := &writeBatchFrame{
- typ: batch.Type,
+ typ: b.batchOpts.Type,
statements: make([]batchStatment, n),
- consistency: batch.Cons,
- serialConsistency: batch.serialCons,
- defaultTimestamp: batch.defaultTimestamp,
- defaultTimestampValue: batch.defaultTimestampValue,
- customPayload: batch.CustomPayload,
+ consistency: b.GetConsistency(),
+ serialConsistency: b.batchOpts.serialCons,
+ defaultTimestamp: b.batchOpts.defaultTimestamp,
+ defaultTimestampValue: b.batchOpts.defaultTimestampValue,
+ customPayload: b.batchOpts.CustomPayload,
}
- stmts := make(map[string]string, len(batch.Entries))
+ stmts := make(map[string]string, len(b.batchOpts.Entries))
for i := 0; i < n; i++ {
- entry := &batch.Entries[i]
- b := &req.statements[i]
+ entry := &b.batchOpts.Entries[i]
+ batchStmt := &req.statements[i]
if len(entry.Args) > 0 || entry.binding != nil {
- info, err := c.prepareStatement(batch.Context(), entry.Stmt, batch.trace)
+ info, err := c.prepareStatement(b.batchOpts.context, entry.Stmt, b.batchOpts.trace)
Review Comment:
If I'm not missing anything here both `ctx` arg of the method and `b.batchOpts.context` should be the the same context value.
The `ctx` arg is being passed by `queryExecutor.executeQuery()` at line 102 by calling `qry.Context()` which should return the underlying `internalBatch.batchOpts.context`.
##########
query_executor.go:
##########
@@ -201,16 +210,385 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.qryOpts.context, q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
+ Entries []BatchEntry
+ CustomPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ trace Tracer
+ observer BatchObserver
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+}
+
+func newBatchOptions(b *Batch) *batchOptions {
+ // make we get a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ return &batchOptions{
+ Type: b.Type,
+ Entries: b.Entries,
+ CustomPayload: b.CustomPayload,
+ rt: b.rt,
+ spec: b.spec,
+ trace: b.trace,
+ observer: b.observer,
+ serialCons: b.serialCons,
+ defaultTimestamp: b.defaultTimestamp,
+ defaultTimestampValue: b.defaultTimestampValue,
+ context: b.Context(),
+ keyspace: b.Keyspace(),
+ idempotent: b.IsIdempotent(),
+ routingKey: b.routingKey,
+ }
+}
+
+type internalBatch struct {
+ originalBatch *Batch
+ batchOpts *batchOptions
+ metrics *queryMetrics
+ consistency uint32
+ routingInfo *queryRoutingInfo
+ session *Session
+}
+
+func newInternalBatch(batch *Batch) *internalBatch {
+ return &internalBatch{
+ originalBatch: batch,
+ batchOpts: newBatchOptions(batch),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ routingInfo: &queryRoutingInfo{},
+ session: batch.session,
+ }
+}
+
+// Attempts returns the number of attempts made to execute the batch.
+func (b *internalBatch) Attempts() int {
+ return b.metrics.attempts()
+}
+
+func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := b.metrics.attempt(1, latency, host, b.batchOpts.observer != nil)
+
+ if b.batchOpts.observer == nil {
+ return
+ }
+
+ statements := make([]string, len(b.batchOpts.Entries))
+ values := make([][]interface{}, len(b.batchOpts.Entries))
+
+ for i, entry := range b.batchOpts.Entries {
+ statements[i] = entry.Stmt
+ values[i] = entry.Args
+ }
+
+ b.batchOpts.observer.ObserveBatch(b.batchOpts.context, ObservedBatch{
+ Keyspace: keyspace,
+ Statements: statements,
+ 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,
+ Err: iter.err,
+ Attempt: attempt,
+ })
+}
+
+func (b *internalBatch) borrowForExecution() {
+ // empty, because Batch has no equivalent of Query.Release()
+ // that would race with speculative executions.
+}
+
+func (b *internalBatch) releaseAfterExecution() {
+ // empty, because Batch has no equivalent of Query.Release()
+ // that would race with speculative executions.
+}
Review Comment:
Since both these methods are removed from the interface you can get rid of them here too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2001393393
##########
conn.go:
##########
@@ -1523,32 +1534,35 @@ func (c *Conn) UseKeyspace(keyspace string) error {
return nil
}
-func (c *Conn) executeBatch(ctx context.Context, batch *Batch) *Iter {
+func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
if c.version == protoVersion1 {
- return &Iter{err: ErrUnsupported}
+ return newErrIter(ErrUnsupported, b.metrics)
}
- n := len(batch.Entries)
+ iter := newIter(b.metrics)
+
+ n := len(b.batchOpts.Entries)
req := &writeBatchFrame{
- typ: batch.Type,
+ typ: b.batchOpts.Type,
statements: make([]batchStatment, n),
- consistency: batch.Cons,
- serialConsistency: batch.serialCons,
- defaultTimestamp: batch.defaultTimestamp,
- defaultTimestampValue: batch.defaultTimestampValue,
- customPayload: batch.CustomPayload,
+ consistency: b.GetConsistency(),
+ serialConsistency: b.batchOpts.serialCons,
+ defaultTimestamp: b.batchOpts.defaultTimestamp,
+ defaultTimestampValue: b.batchOpts.defaultTimestampValue,
+ customPayload: b.batchOpts.CustomPayload,
}
- stmts := make(map[string]string, len(batch.Entries))
+ stmts := make(map[string]string, len(b.batchOpts.Entries))
for i := 0; i < n; i++ {
- entry := &batch.Entries[i]
- b := &req.statements[i]
+ entry := &b.batchOpts.Entries[i]
+ batchStmt := &req.statements[i]
if len(entry.Args) > 0 || entry.binding != nil {
- info, err := c.prepareStatement(batch.Context(), entry.Stmt, batch.trace)
+ info, err := c.prepareStatement(b.batchOpts.context, entry.Stmt, b.batchOpts.trace)
Review Comment:
Agree, done 👍
##########
query_executor.go:
##########
@@ -201,16 +210,385 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.qryOpts.context, q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
+ Entries []BatchEntry
+ CustomPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ trace Tracer
+ observer BatchObserver
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+}
+
+func newBatchOptions(b *Batch) *batchOptions {
+ // make we get a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ return &batchOptions{
+ Type: b.Type,
+ Entries: b.Entries,
+ CustomPayload: b.CustomPayload,
+ rt: b.rt,
+ spec: b.spec,
+ trace: b.trace,
+ observer: b.observer,
+ serialCons: b.serialCons,
+ defaultTimestamp: b.defaultTimestamp,
+ defaultTimestampValue: b.defaultTimestampValue,
+ context: b.Context(),
+ keyspace: b.Keyspace(),
+ idempotent: b.IsIdempotent(),
+ routingKey: b.routingKey,
+ }
+}
+
+type internalBatch struct {
+ originalBatch *Batch
+ batchOpts *batchOptions
+ metrics *queryMetrics
+ consistency uint32
+ routingInfo *queryRoutingInfo
+ session *Session
+}
+
+func newInternalBatch(batch *Batch) *internalBatch {
+ return &internalBatch{
+ originalBatch: batch,
+ batchOpts: newBatchOptions(batch),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ routingInfo: &queryRoutingInfo{},
+ session: batch.session,
+ }
+}
+
+// Attempts returns the number of attempts made to execute the batch.
+func (b *internalBatch) Attempts() int {
+ return b.metrics.attempts()
+}
+
+func (b *internalBatch) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := b.metrics.attempt(1, latency, host, b.batchOpts.observer != nil)
+
+ if b.batchOpts.observer == nil {
+ return
+ }
+
+ statements := make([]string, len(b.batchOpts.Entries))
+ values := make([][]interface{}, len(b.batchOpts.Entries))
+
+ for i, entry := range b.batchOpts.Entries {
+ statements[i] = entry.Stmt
+ values[i] = entry.Args
+ }
+
+ b.batchOpts.observer.ObserveBatch(b.batchOpts.context, ObservedBatch{
+ Keyspace: keyspace,
+ Statements: statements,
+ 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,
+ Err: iter.err,
+ Attempt: attempt,
+ })
+}
+
+func (b *internalBatch) borrowForExecution() {
+ // empty, because Batch has no equivalent of Query.Release()
+ // that would race with speculative executions.
+}
+
+func (b *internalBatch) releaseAfterExecution() {
+ // empty, because Batch has no equivalent of Query.Release()
+ // that would race with speculative executions.
+}
Review Comment:
Nice catch, done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "worryg0d (via GitHub)" <gi...@apache.org>.
worryg0d commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2000949093
##########
conn.go:
##########
@@ -1523,32 +1534,35 @@ func (c *Conn) UseKeyspace(keyspace string) error {
return nil
}
-func (c *Conn) executeBatch(ctx context.Context, batch *Batch) *Iter {
+func (c *Conn) executeBatch(ctx context.Context, b *internalBatch) *Iter {
if c.version == protoVersion1 {
- return &Iter{err: ErrUnsupported}
+ return newErrIter(ErrUnsupported, b.metrics)
}
- n := len(batch.Entries)
+ iter := newIter(b.metrics)
+
+ n := len(b.batchOpts.Entries)
req := &writeBatchFrame{
- typ: batch.Type,
+ typ: b.batchOpts.Type,
statements: make([]batchStatment, n),
- consistency: batch.Cons,
- serialConsistency: batch.serialCons,
- defaultTimestamp: batch.defaultTimestamp,
- defaultTimestampValue: batch.defaultTimestampValue,
- customPayload: batch.CustomPayload,
+ consistency: b.GetConsistency(),
+ serialConsistency: b.batchOpts.serialCons,
+ defaultTimestamp: b.batchOpts.defaultTimestamp,
+ defaultTimestampValue: b.batchOpts.defaultTimestampValue,
+ customPayload: b.batchOpts.CustomPayload,
}
- stmts := make(map[string]string, len(batch.Entries))
+ stmts := make(map[string]string, len(b.batchOpts.Entries))
for i := 0; i < n; i++ {
- entry := &batch.Entries[i]
- b := &req.statements[i]
+ entry := &b.batchOpts.Entries[i]
+ batchStmt := &req.statements[i]
if len(entry.Args) > 0 || entry.binding != nil {
- info, err := c.prepareStatement(batch.Context(), entry.Stmt, batch.trace)
+ info, err := c.prepareStatement(b.batchOpts.context, entry.Stmt, b.batchOpts.trace)
Review Comment:
If I'm not missing anything here both `ctx` arg of the method and `b.batchOpts.context` should be have the same context value.
The `ctx` arg is being passed by `queryExecutor.executeQuery()` at line 102 by calling `qry.Context()` which should return the underlying `internalBatch.batchOpts.context`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2751421634
> What about making the methods `Exec(context.Context)` and `Iter(context.Context) error` so there's no need to shallow copy query in `WithContext`? This is a bigger change though that we might not want to make. Taht said, I think Query would be a bit easier to re-use then. I know this differs from `http.Request` but in that case it might be a symptom of not wanting to break the existing API when context's were introduced.
I like the idea but I think it would be too much to ask of users when upgrading since it would affect every single statement on their app... I'd be down to adding an overload and deprecating `WithContext` though... I just don't know what names would fit for the new Iter/Exec overloads...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2011152913
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
Review Comment:
Should we be copying `values`, `customPayload`, `initialPageState`, `routingKey`? it might not be obvious that those are not copied and can't be mutated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2827987337
@jameshartig @worryg0d do you guys have some free time to take a look at this again during the next few days?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 CASSGO-73 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2128994199
##########
CHANGELOG.md:
##########
@@ -54,7 +54,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Move "execute batch" methods to Batch type (CASSGO-57)
-- Make `Session` immutable by removing setters and associated mutex (CASSGO-23)
+- Make Session immutable by removing setters and associated mutex (CASSGO-23)
+
+- Changes to Query and Batch to make them safely reusable (CASSGO-22)
Review Comment:
reminder to add CASSGO-73 here
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "ribaraka (via GitHub)" <gi...@apache.org>.
ribaraka commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2043815423
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
`newPageState` seems to be redundant here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 CASSGO-73 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis merged PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2749797454
> executing a batch (even though it doesn't make a lot of sense conceptually due to the name but Iter is what the driver uses to return data about a request so it's not just about "iterating").
Note that Yugabyte lets you iterate over the [row status](https://docs.yugabyte.com/preview/api/ycql/batch/#row-status) from a batch operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "lukasz-antoniak (via GitHub)" <gi...@apache.org>.
lukasz-antoniak commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2009691464
##########
session.go:
##########
@@ -727,33 +720,36 @@ func (s *Session) routingKeyInfo(ctx context.Context, stmt string) (*routingKeyI
return routingKeyInfo, nil
}
-func (b *Batch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
-}
-
// Exec executes a batch operation and returns nil if successful
// otherwise an error is returned describing the failure.
func (b *Batch) Exec() error {
iter := b.session.executeBatch(b)
return iter.Close()
}
+// Iter executes a batch operation and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) Iter() *Iter {
+ return b.session.executeBatch(b)
Review Comment:
I feel that people may execute sequence of `Exec()` and `Iter()`. Shall we make an assertion that only one of the is invoked?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "lukasz-antoniak (via GitHub)" <gi...@apache.org>.
lukasz-antoniak commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2009655505
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
Review Comment:
Unused?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2052177225
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
oh you're right, I removed this field from this type and moved it up to `internalQuery` but I forgot to remove this part, nice catch. Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2011149160
##########
conn.go:
##########
@@ -1371,76 +1376,80 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
value := values[i]
typ := info.request.columns[i].TypeInfo
if err := marshalQueryValue(typ, value, v); err != nil {
- return &Iter{err: err}
+ iter.err = err
+ return iter
}
}
// if the metadata was not present in the response then we should not skip it
- params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qry.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
+ params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qryOpts.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
frame = &writeExecuteFrame{
preparedID: info.id,
params: params,
- customPayload: qry.customPayload,
+ customPayload: qryOpts.customPayload,
}
// Set "keyspace" and "table" property in the query if it is present in preparedMetadata
- qry.routingInfo.mu.Lock()
- qry.routingInfo.keyspace = info.request.keyspace
- qry.routingInfo.table = info.request.table
- qry.routingInfo.mu.Unlock()
+ q.routingInfo.mu.Lock()
Review Comment:
Also can we just change `routingInfo` to not be a struct anymore? Seems like it only was because...
> // routingInfo is a pointer because Query can be copied and copyable struct can't hold a mutex.
Do we still allow copies of `internalQuery`? If so, maybe we should make a Clone method?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "lukasz-antoniak (via GitHub)" <gi...@apache.org>.
lukasz-antoniak commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2009610280
##########
query_executor.go:
##########
@@ -27,32 +27,43 @@ package gocql
import (
"context"
"sync"
+ "sync/atomic"
"time"
)
+// ExecutableQuery is an interface that represents a query or batch statement that
+// exposes the correct functions for the HostSelectionPolicy to operate correctly.
type ExecutableQuery interface {
Review Comment:
What do you think about renaming type to `HostPolicyQuery`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012238698
##########
conn.go:
##########
@@ -1371,76 +1376,80 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
value := values[i]
typ := info.request.columns[i].TypeInfo
if err := marshalQueryValue(typ, value, v); err != nil {
- return &Iter{err: err}
+ iter.err = err
+ return iter
}
}
// if the metadata was not present in the response then we should not skip it
- params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qry.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
+ params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qryOpts.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
frame = &writeExecuteFrame{
preparedID: info.id,
params: params,
- customPayload: qry.customPayload,
+ customPayload: qryOpts.customPayload,
}
// Set "keyspace" and "table" property in the query if it is present in preparedMetadata
- qry.routingInfo.mu.Lock()
- qry.routingInfo.keyspace = info.request.keyspace
- qry.routingInfo.table = info.request.table
- qry.routingInfo.mu.Unlock()
+ q.routingInfo.mu.Lock()
Review Comment:
> Does this still need to lock? I don't think it does unless we expect GetRoutingKey to be called from separate goroutines?
I think this is still the case due to speculative executions (and possibly retries I'm not sure).
> Also can we just change routingInfo to not be a struct anymore? Seems like it only was because...
>> // routingInfo is a pointer because Query can be copied and copyable struct can't hold a mutex.
> Do we still allow copies of internalQuery? If so, maybe we should make a Clone method?
I don't think it needs to be a struct anymore, I'll change it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2012240476
##########
query_executor.go:
##########
@@ -27,32 +27,43 @@ package gocql
import (
"context"
"sync"
+ "sync/atomic"
"time"
)
+// ExecutableQuery is an interface that represents a query or batch statement that
+// exposes the correct functions for the HostSelectionPolicy to operate correctly.
type ExecutableQuery interface {
Review Comment:
I posted a comment about the renaming of this interface below in a response to James
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "worryg0d (via GitHub)" <gi...@apache.org>.
worryg0d commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2765982318
> I just don't know what names would fit for the new Iter/Exec overloads...
Well, we can use the common **Context** suffix for those overloads, like it is done in sql package from standard lib - https://pkg.go.dev/database/sql#Conn.QueryContext
https://pkg.go.dev/database/sql#DB.ExecContext
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2001394047
##########
query_executor.go:
##########
@@ -201,16 +210,385 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.qryOpts.context, q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
+ Entries []BatchEntry
+ CustomPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ trace Tracer
+ observer BatchObserver
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+}
+
+func newBatchOptions(b *Batch) *batchOptions {
+ // make we get a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ return &batchOptions{
+ Type: b.Type,
+ Entries: b.Entries,
Review Comment:
Nice catch, done 👍
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2011153778
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query) *internalQuery {
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: nil,
+ refCount: 0,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.Context(), q.qryOpts.stmt)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+type batchOptions struct {
+ Type BatchType
Review Comment:
Any reason why some of these are public?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2011155701
##########
session.go:
##########
@@ -727,33 +720,36 @@ func (s *Session) routingKeyInfo(ctx context.Context, stmt string) (*routingKeyI
return routingKeyInfo, nil
}
-func (b *Batch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
-}
-
// Exec executes a batch operation and returns nil if successful
// otherwise an error is returned describing the failure.
func (b *Batch) Exec() error {
iter := b.session.executeBatch(b)
return iter.Close()
}
+// Iter executes a batch operation and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) Iter() *Iter {
+ return b.session.executeBatch(b)
Review Comment:
Hmm, I see what you mean but I don't think we can lock them into one or another if we want it to be reusable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2749837001
What about making the methods `Exec(context.Context)` and `Iter(context.Context) error` so there's no need to shallow copy query in `WithContext`? This is a bigger change though that we might not want to make. Taht said, I think Query would be a bit easier to re-use then. I know this differs from `http.Request` but in that case it might be a symptom of not wanting to break the existing API when context's were introduced.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2011147802
##########
conn.go:
##########
@@ -1371,76 +1376,80 @@ func (c *Conn) executeQuery(ctx context.Context, qry *Query) *Iter {
value := values[i]
typ := info.request.columns[i].TypeInfo
if err := marshalQueryValue(typ, value, v); err != nil {
- return &Iter{err: err}
+ iter.err = err
+ return iter
}
}
// if the metadata was not present in the response then we should not skip it
- params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qry.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
+ params.skipMeta = !(c.session.cfg.DisableSkipMetadata || qryOpts.disableSkipMetadata) && info != nil && info.response.flags&flagNoMetaData == 0
frame = &writeExecuteFrame{
preparedID: info.id,
params: params,
- customPayload: qry.customPayload,
+ customPayload: qryOpts.customPayload,
}
// Set "keyspace" and "table" property in the query if it is present in preparedMetadata
- qry.routingInfo.mu.Lock()
- qry.routingInfo.keyspace = info.request.keyspace
- qry.routingInfo.table = info.request.table
- qry.routingInfo.mu.Unlock()
+ q.routingInfo.mu.Lock()
Review Comment:
Does this still need to lock? I don't think it does unless we expect `GetRoutingKey` to be called from separate goroutines?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable and "threadsafe" [cassandra-gocql-driver]
Posted by "jameshartig (via GitHub)" <gi...@apache.org>.
jameshartig commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2749801848
I don't want to bikeshed about names but `ExecutableQuery` is a bit confusing since it contains `Query` and `Batch`. Did that name come from somewhere? Would `ExecutableStatement` make more sense since it has a statement?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2790216244
I've addressed all PR comments, I'll work on rebasing the branch now and then I'll work on the change mentioned [here](https://github.com/apache/cassandra-gocql-driver/pull/1870#issuecomment-2769579576).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "worryg0d (via GitHub)" <gi...@apache.org>.
worryg0d commented on PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#issuecomment-2765957642
> I'm concerned that renaming this interface will increase the chance users will have to change their code when upgrading the driver but I do agree that the current name isn't good.
We can add a type alias to leave the `ExecutableQuery` name for the migration period:
```go
// Deprecated: Will be removed in the future major release.
// Please use Statement instead.
type ExecutableQuery = Statement
type Statement interface {
...
}
```
I tested it on go 1.19.13 and it works fine
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "ribaraka (via GitHub)" <gi...@apache.org>.
ribaraka commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044123216
##########
session.go:
##########
@@ -743,33 +736,47 @@ func (s *Session) routingKeyInfo(ctx context.Context, stmt string, keyspace stri
return routingKeyInfo, nil
}
-func (b *Batch) execute(ctx context.Context, conn *Conn) *Iter {
- return conn.executeBatch(ctx, b)
-}
-
// Exec executes a batch operation and returns nil if successful
// otherwise an error is returned describing the failure.
func (b *Batch) Exec() error {
- iter := b.session.executeBatch(b)
+ iter := b.session.executeBatch(b, nil)
+ return iter.Close()
+}
+
+// ExecContext executes a batch operation with the provided context and returns nil if successful
+// otherwise an error is returned describing the failure.
+func (b *Batch) ExecContext(ctx context.Context) error {
+ iter := b.session.executeBatch(b, ctx)
return iter.Close()
}
-func (s *Session) executeBatch(batch *Batch) *Iter {
+// Iter executes a batch operation and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) Iter() *Iter { return b.session.executeBatch(b, nil) }
+
+// Iter executes a batch operation with the provided context and returns an Iter object
+// that can be used to access properties related to the execution like Iter.Attempts and Iter.Latency
+func (b *Batch) IterContext(ctx context.Context) *Iter {
+ return b.session.executeBatch(b, ctx)
+}
Review Comment:
nitpicks:
1. The `Iter` method should call the `IterContext`
2. The comment for `IterContext` should start with the method name
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
`newPageState` seems redundant here.
##########
query_executor.go:
##########
@@ -201,16 +210,375 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+ values []interface{}
+ initialConsistency Consistency
+ pageSize int
+ initialPageState []byte
+ prefetch float64
+ trace Tracer
+ observer QueryObserver
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ binding func(q *QueryInfo) ([]interface{}, error)
+ serialCons SerialConsistency
+ defaultTimestamp bool
+ defaultTimestampValue int64
+ disableSkipMetadata bool
+ context context.Context
+ idempotent bool
+ customPayload map[string][]byte
+ keyspace string
+ disableAutoPage bool
+ skipPrepare bool
+ routingKey []byte
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query) *queryOptions {
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ initialConsistency: q.initialConsistency,
+ pageSize: q.pageSize,
+ initialPageState: q.initialPageState,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: q.Context(),
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: q.routingKey,
+ getKeyspace: q.getKeyspace,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
Review Comment:
still persist. but changed its position :sweat_smile:
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
+ if q.routingKey != nil {
+ routingKey := q.routingKey
+ newRoutingKey = make([]byte, len(routingKey))
+ copy(newRoutingKey, routingKey)
+ }
+ if ctx == nil {
+ ctx = q.Context()
+ }
+ return &queryOptions{
+ stmt: q.stmt,
+ values: q.values,
+ pageSize: q.pageSize,
+ prefetch: q.prefetch,
+ trace: q.trace,
+ observer: q.observer,
+ rt: q.rt,
+ spec: q.spec,
+ binding: q.binding,
+ serialCons: q.serialCons,
+ defaultTimestamp: q.defaultTimestamp,
+ defaultTimestampValue: q.defaultTimestampValue,
+ disableSkipMetadata: q.disableSkipMetadata,
+ context: ctx,
+ idempotent: q.idempotent,
+ customPayload: q.customPayload,
+ disableAutoPage: q.disableAutoPage,
+ skipPrepare: q.skipPrepare,
+ routingKey: newRoutingKey,
+ getKeyspace: q.getKeyspace,
+ nowInSecondsValue: q.nowInSecondsValue,
+ keyspace: q.keyspace,
+ hostID: q.hostID,
+ }
+}
+
+type internalQuery struct {
+ originalQuery *Query
+ qryOpts *queryOptions
+ pageState []byte
+ metrics *queryMetrics
+ refCount uint32
+ conn *Conn
+ consistency uint32
+ session *Session
+ routingInfo *queryRoutingInfo
+}
+
+func newInternalQuery(q *Query, ctx context.Context) *internalQuery {
+ var newPageState []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
+ return &internalQuery{
+ originalQuery: q,
+ qryOpts: newQueryOptions(q, ctx),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ consistency: uint32(q.initialConsistency),
+ pageState: newPageState,
+ conn: nil,
+ session: q.session,
+ routingInfo: &queryRoutingInfo{},
+ }
+}
+
+// Attempts returns the number of times the query was executed.
+func (q *internalQuery) Attempts() int {
+ return q.metrics.attempts()
+}
+
+func (q *internalQuery) attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo) {
+ latency := end.Sub(start)
+ attempt, metricsForHost := q.metrics.attempt(1, latency, host, q.qryOpts.observer != nil)
+
+ if q.qryOpts.observer != nil {
+ 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,
+ })
+ }
+}
+
+func (q *internalQuery) execute(ctx context.Context, conn *Conn) *Iter {
+ return conn.executeQuery(ctx, q)
+}
+
+func (q *internalQuery) retryPolicy() RetryPolicy {
+ return q.qryOpts.rt
+}
+
+func (q *internalQuery) speculativeExecutionPolicy() SpeculativeExecutionPolicy {
+ return q.qryOpts.spec
+}
+
+func (q *internalQuery) GetRoutingKey() ([]byte, error) {
+ if q.qryOpts.routingKey != nil {
+ return q.qryOpts.routingKey, nil
+ }
+
+ if q.qryOpts.binding != nil && len(q.qryOpts.values) == 0 {
+ // If this query was created using session.Bind we wont have the query
+ // values yet, so we have to pass down to the next policy.
+ // TODO: Remove this and handle this case
+ return nil, nil
+ }
+
+ // try to determine the routing key
+ routingKeyInfo, err := q.session.routingKeyInfo(q.Context(), q.qryOpts.stmt, q.qryOpts.keyspace)
+ if err != nil {
+ return nil, err
+ }
+
+ if routingKeyInfo != nil {
+ q.routingInfo.mu.Lock()
+ q.routingInfo.keyspace = routingKeyInfo.keyspace
+ q.routingInfo.table = routingKeyInfo.table
+ q.routingInfo.mu.Unlock()
+ }
+ return createRoutingKey(routingKeyInfo, q.qryOpts.values)
+}
+
+func (q *internalQuery) Keyspace() string {
+ if q.qryOpts.getKeyspace != nil {
+ return q.qryOpts.getKeyspace()
+ }
+
+ qrKs := q.routingInfo.getKeyspace()
+ if qrKs != "" {
+ return qrKs
+ }
+ if q.qryOpts.keyspace != "" {
+ return q.qryOpts.keyspace
+ }
+
+ if q.session == nil {
+ return ""
+ }
+ // TODO(chbannis): this should be parsed from the query or we should let
+ // this be set by users.
+ return q.session.cfg.Keyspace
+}
+
+func (q *internalQuery) Table() string {
+ return q.routingInfo.getTable()
+}
+
+func (q *internalQuery) IsIdempotent() bool {
+ return q.qryOpts.idempotent
+}
+
+func (q *internalQuery) getQueryMetrics() *queryMetrics {
+ return q.metrics
+}
+
+func (q *internalQuery) SetConsistency(c Consistency) {
+ atomic.StoreUint32(&q.consistency, uint32(c))
+}
+
+func (q *internalQuery) GetConsistency() Consistency {
+ return Consistency(atomic.LoadUint32(&q.consistency))
+}
+
+func (q *internalQuery) Context() context.Context {
+ return q.qryOpts.context
+}
+
+func (q *internalQuery) Statement() Statement {
+ return q.originalQuery
+}
+
+func (q *internalQuery) GetHostID() string {
+ return q.qryOpts.hostID
+}
+
+func (q *internalQuery) getRoutingInfo() *queryRoutingInfo {
+ return q.routingInfo
+}
+
+func (q *internalQuery) getKeyspaceFunc() func() string {
+ return q.qryOpts.getKeyspace
+}
+
+type batchOptions struct {
+ trace Tracer
+ observer BatchObserver
+
+ bType BatchType
+ entries []BatchEntry
+
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ serialCons SerialConsistency
+
+ customPayload map[string][]byte
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ keyspace string
+ idempotent bool
+ routingKey []byte
+ nowInSeconds *int
+}
+
+func newBatchOptions(b *Batch, ctx context.Context) *batchOptions {
+ // make a new array so if user keeps appending entries on the Batch object it doesn't affect this execution
+ newEntries := make([]BatchEntry, len(b.Entries))
+ for i, e := range b.Entries {
+ newEntries[i] = e
+ }
+ var newRoutingKey []byte
+ if b.routingKey != nil {
+ routingKey := b.routingKey
+ newRoutingKey = make([]byte, len(routingKey))
+ copy(newRoutingKey, routingKey)
+ }
+ if ctx == nil {
+ ctx = b.Context()
+ }
+ return &batchOptions{
+ bType: b.Type,
+ entries: newEntries,
+ customPayload: b.CustomPayload,
+ rt: b.rt,
+ spec: b.spec,
+ trace: b.trace,
+ observer: b.observer,
+ serialCons: b.serialCons,
+ defaultTimestamp: b.defaultTimestamp,
+ defaultTimestampValue: b.defaultTimestampValue,
+ context: ctx,
+ keyspace: b.Keyspace(),
+ idempotent: b.IsIdempotent(),
+ routingKey: newRoutingKey,
+ nowInSeconds: b.nowInSeconds,
+ }
+}
+
+type internalBatch struct {
+ originalBatch *Batch
+ batchOpts *batchOptions
+ metrics *queryMetrics
+ consistency uint32
+ routingInfo *queryRoutingInfo
+ session *Session
+}
+
+func newInternalBatch(batch *Batch, ctx context.Context) *internalBatch {
+ return &internalBatch{
+ originalBatch: batch,
+ batchOpts: newBatchOptions(batch, ctx),
+ metrics: &queryMetrics{m: make(map[string]*hostMetrics)},
+ routingInfo: &queryRoutingInfo{},
+ session: batch.session,
+ }
+}
Review Comment:
`consistency` is missing
##########
query_executor.go:
##########
@@ -27,33 +27,51 @@ package gocql
import (
"context"
"sync"
+ "sync/atomic"
"time"
)
-type ExecutableQuery interface {
- borrowForExecution() // Used to ensure that the query stays alive for lifetime of a particular execution goroutine.
- releaseAfterExecution() // Used when a goroutine finishes its execution attempts, either with ok result or an error.
- execute(ctx context.Context, conn *Conn) *Iter
- attempt(keyspace string, end, start time.Time, iter *Iter, host *HostInfo)
- retryPolicy() RetryPolicy
- speculativeExecutionPolicy() SpeculativeExecutionPolicy
+// Deprecated: Will be removed in a future major release. Also Query and Batch no longer implement this interface.
+//
+// Please use Statement (for Query / Batch objects) or ExecutableStatement (in HostSelectionPolicy implementations) instead.
+type ExecutableQuery = ExecutableStatement
+
+// ExecutableStatement is an interface that represents a query or batch statement that
+// exposes the correct functions for the HostSelectionPolicy to operate correctly.
+type ExecutableStatement interface {
GetRoutingKey() ([]byte, error)
Keyspace() string
Table() string
IsIdempotent() bool
GetHostID() string
+ Statement() Statement
+}
- withContext(context.Context) ExecutableQuery
+// Statement is an interface that represents a CQL statement that the driver can execute
+// (currently Query and Batch via Session.Query and Session.Batch)
+type Statement interface {
+ Iter() *Iter
+ Exec() error
+}
Review Comment:
adding here `IterContext` and `ExecContetx`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSGO-22 Changes to Query and Batch to make them safely reusable [cassandra-gocql-driver]
Posted by "joao-r-reis (via GitHub)" <gi...@apache.org>.
joao-r-reis commented on code in PR #1868:
URL: https://github.com/apache/cassandra-gocql-driver/pull/1868#discussion_r2044797970
##########
query_executor.go:
##########
@@ -222,16 +238,452 @@ func (q *queryExecutor) do(ctx context.Context, qry ExecutableQuery, hostIter Ne
}
if lastErr != nil {
- return &Iter{err: lastErr}
+ return newErrIter(lastErr, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
- return &Iter{err: ErrNoConnections}
+ return newErrIter(ErrNoConnections, qry.getQueryMetrics(), qry.Keyspace(), qry.getRoutingInfo(), qry.getKeyspaceFunc())
}
-func (q *queryExecutor) run(ctx context.Context, qry ExecutableQuery, hostIter NextHost, results chan<- *Iter) {
+func (q *queryExecutor) run(ctx context.Context, qry internalRequest, hostIter NextHost, results chan<- *Iter) {
select {
case results <- q.do(ctx, qry, hostIter):
case <-ctx.Done():
}
- qry.releaseAfterExecution()
+}
+
+type queryOptions struct {
+ stmt string
+
+ // Paging
+ pageSize int
+ disableAutoPage bool
+
+ // Monitoring
+ trace Tracer
+ observer QueryObserver
+
+ // Parameters
+ values []interface{}
+ binding func(q *QueryInfo) ([]interface{}, error)
+
+ // Timestamp
+ defaultTimestamp bool
+ defaultTimestampValue int64
+
+ // Consistency
+ serialCons SerialConsistency
+
+ // Protocol flag
+ disableSkipMetadata bool
+
+ customPayload map[string][]byte
+ prefetch float64
+ rt RetryPolicy
+ spec SpeculativeExecutionPolicy
+ context context.Context
+ idempotent bool
+ keyspace string
+ skipPrepare bool
+ routingKey []byte
+ nowInSecondsValue *int
+ hostID string
+
+ // getKeyspace is field so that it can be overriden in tests
+ getKeyspace func() string
+}
+
+func newQueryOptions(q *Query, ctx context.Context) *queryOptions {
+ var newPageState, newRoutingKey []byte
+ if q.initialPageState != nil {
+ pageState := q.initialPageState
+ newPageState = make([]byte, len(pageState))
+ copy(newPageState, pageState)
+ }
Review Comment:
I do wonder if we're just adding a performance penalty to safeguard against something that users won't attempt to do 99% of the times
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org