Skip to content

Commit

Permalink
make properties of QueuedQuery and Batch public, closes #1878
Browse files Browse the repository at this point in the history
  • Loading branch information
@pashagolub @jackc
pashagolub authored and jackc committed Jan 26, 2024
1 parent a57bb8c commit c90f82a
Show file tree
Hide file tree
Showing 2 changed files with 49 additions and 49 deletions.
42 changes: 21 additions & 21 deletions batch.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@ import (

// QueuedQuery is a query that has been queued for execution via a Batch.
type QueuedQuery struct {
query string
arguments []any
SQL string
Arguments []any
fn batchItemFunc
sd *pgconn.StatementDescription
}
Expand Down Expand Up @@ -57,24 +57,24 @@ func (qq *QueuedQuery) Exec(fn func(ct pgconn.CommandTag) error) {
// Batch queries are a way of bundling multiple queries together to avoid
// unnecessary network round trips. A Batch must only be sent once.
type Batch struct {
queuedQueries []*QueuedQuery
QueuedQueries []*QueuedQuery
}

// Queue queues a query to batch b. query can be an SQL query or the name of a prepared statement.
// The only pgx option argument that is supported is QueryRewriter. Queries are executed using the
// connection's DefaultQueryExecMode.
func (b *Batch) Queue(query string, arguments ...any) *QueuedQuery {
qq := &QueuedQuery{
query: query,
arguments: arguments,
SQL: query,
Arguments: arguments,
}
b.queuedQueries = append(b.queuedQueries, qq)
b.QueuedQueries = append(b.QueuedQueries, qq)
return qq
}

// Len returns number of queries that have been queued so far.
func (b *Batch) Len() int {
return len(b.queuedQueries)
return len(b.QueuedQueries)
}

type BatchResults interface {
Expand Down Expand Up @@ -227,9 +227,9 @@ func (br *batchResults) Close() error {
}

// Read and run fn for all remaining items
for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.queuedQueries) {
if br.b.queuedQueries[br.qqIdx].fn != nil {
err := br.b.queuedQueries[br.qqIdx].fn(br)
for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
if br.b.QueuedQueries[br.qqIdx].fn != nil {
err := br.b.QueuedQueries[br.qqIdx].fn(br)
if err != nil {
br.err = err
}
Expand All @@ -253,10 +253,10 @@ func (br *batchResults) earlyError() error {
}

func (br *batchResults) nextQueryAndArgs() (query string, args []any, ok bool) {
if br.b != nil && br.qqIdx < len(br.b.queuedQueries) {
bi := br.b.queuedQueries[br.qqIdx]
query = bi.query
args = bi.arguments
if br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
bi := br.b.QueuedQueries[br.qqIdx]
query = bi.SQL
args = bi.Arguments
ok = true
br.qqIdx++
}
Expand Down Expand Up @@ -396,9 +396,9 @@ func (br *pipelineBatchResults) Close() error {
}

// Read and run fn for all remaining items
for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.queuedQueries) {
if br.b.queuedQueries[br.qqIdx].fn != nil {
err := br.b.queuedQueries[br.qqIdx].fn(br)
for br.err == nil && !br.closed && br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
if br.b.QueuedQueries[br.qqIdx].fn != nil {
err := br.b.QueuedQueries[br.qqIdx].fn(br)
if err != nil {
br.err = err
}
Expand All @@ -422,10 +422,10 @@ func (br *pipelineBatchResults) earlyError() error {
}

func (br *pipelineBatchResults) nextQueryAndArgs() (query string, args []any, ok bool) {
if br.b != nil && br.qqIdx < len(br.b.queuedQueries) {
bi := br.b.queuedQueries[br.qqIdx]
query = bi.query
args = bi.arguments
if br.b != nil && br.qqIdx < len(br.b.QueuedQueries) {
bi := br.b.QueuedQueries[br.qqIdx]
query = bi.SQL
args = bi.Arguments
ok = true
br.qqIdx++
}
Expand Down
56 changes: 28 additions & 28 deletions conn.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -903,10 +903,10 @@ func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults) {
return &batchResults{ctx: ctx, conn: c, err: err}
}

for _, bi := range b.queuedQueries {
for _, bi := range b.QueuedQueries {
var queryRewriter QueryRewriter
sql := bi.query
arguments := bi.arguments
sql := bi.SQL
arguments := bi.Arguments

optionLoop:
for len(arguments) > 0 {
Expand All @@ -928,8 +928,8 @@ func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults) {
}
}

bi.query = sql
bi.arguments = arguments
bi.SQL = sql
bi.Arguments = arguments
}

// TODO: changing mode per batch? Update Batch.Queue function comment when implemented
Expand All @@ -939,8 +939,8 @@ func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults) {
}

// All other modes use extended protocol and thus can use prepared statements.
for _, bi := range b.queuedQueries {
if sd, ok := c.preparedStatements[bi.query]; ok {
for _, bi := range b.QueuedQueries {
if sd, ok := c.preparedStatements[bi.SQL]; ok {
bi.sd = sd
}
}
Expand All @@ -961,11 +961,11 @@ func (c *Conn) SendBatch(ctx context.Context, b *Batch) (br BatchResults) {

func (c *Conn) sendBatchQueryExecModeSimpleProtocol(ctx context.Context, b *Batch) *batchResults {
var sb strings.Builder
for i, bi := range b.queuedQueries {
for i, bi := range b.QueuedQueries {
if i > 0 {
sb.WriteByte(';')
}
sql, err := c.sanitizeForSimpleQuery(bi.query, bi.arguments...)
sql, err := c.sanitizeForSimpleQuery(bi.SQL, bi.Arguments...)
if err != nil {
return &batchResults{ctx: ctx, conn: c, err: err}
}
Expand All @@ -984,21 +984,21 @@ func (c *Conn) sendBatchQueryExecModeSimpleProtocol(ctx context.Context, b *Batc
func (c *Conn) sendBatchQueryExecModeExec(ctx context.Context, b *Batch) *batchResults {
batch := &pgconn.Batch{}

for _, bi := range b.queuedQueries {
for _, bi := range b.QueuedQueries {
sd := bi.sd
if sd != nil {
err := c.eqb.Build(c.typeMap, sd, bi.arguments)
err := c.eqb.Build(c.typeMap, sd, bi.Arguments)
if err != nil {
return &batchResults{ctx: ctx, conn: c, err: err}
}

batch.ExecPrepared(sd.Name, c.eqb.ParamValues, c.eqb.ParamFormats, c.eqb.ResultFormats)
} else {
err := c.eqb.Build(c.typeMap, nil, bi.arguments)
err := c.eqb.Build(c.typeMap, nil, bi.Arguments)
if err != nil {
return &batchResults{ctx: ctx, conn: c, err: err}
}
batch.ExecParams(bi.query, c.eqb.ParamValues, nil, c.eqb.ParamFormats, c.eqb.ResultFormats)
batch.ExecParams(bi.SQL, c.eqb.ParamValues, nil, c.eqb.ParamFormats, c.eqb.ResultFormats)
}
}

Expand All @@ -1023,18 +1023,18 @@ func (c *Conn) sendBatchQueryExecModeCacheStatement(ctx context.Context, b *Batc
distinctNewQueries := []*pgconn.StatementDescription{}
distinctNewQueriesIdxMap := make(map[string]int)

for _, bi := range b.queuedQueries {
for _, bi := range b.QueuedQueries {
if bi.sd == nil {
sd := c.statementCache.Get(bi.query)
sd := c.statementCache.Get(bi.SQL)
if sd != nil {
bi.sd = sd
} else {
if idx, present := distinctNewQueriesIdxMap[bi.query]; present {
if idx, present := distinctNewQueriesIdxMap[bi.SQL]; present {
bi.sd = distinctNewQueries[idx]
} else {
sd = &pgconn.StatementDescription{
Name: stmtcache.StatementName(bi.query),
SQL: bi.query,
Name: stmtcache.StatementName(bi.SQL),
SQL: bi.SQL,
}
distinctNewQueriesIdxMap[sd.SQL] = len(distinctNewQueries)
distinctNewQueries = append(distinctNewQueries, sd)
Expand All @@ -1055,17 +1055,17 @@ func (c *Conn) sendBatchQueryExecModeCacheDescribe(ctx context.Context, b *Batch
distinctNewQueries := []*pgconn.StatementDescription{}
distinctNewQueriesIdxMap := make(map[string]int)

for _, bi := range b.queuedQueries {
for _, bi := range b.QueuedQueries {
if bi.sd == nil {
sd := c.descriptionCache.Get(bi.query)
sd := c.descriptionCache.Get(bi.SQL)
if sd != nil {
bi.sd = sd
} else {
if idx, present := distinctNewQueriesIdxMap[bi.query]; present {
if idx, present := distinctNewQueriesIdxMap[bi.SQL]; present {
bi.sd = distinctNewQueries[idx]
} else {
sd = &pgconn.StatementDescription{
SQL: bi.query,
SQL: bi.SQL,
}
distinctNewQueriesIdxMap[sd.SQL] = len(distinctNewQueries)
distinctNewQueries = append(distinctNewQueries, sd)
Expand All @@ -1082,13 +1082,13 @@ func (c *Conn) sendBatchQueryExecModeDescribeExec(ctx context.Context, b *Batch)
distinctNewQueries := []*pgconn.StatementDescription{}
distinctNewQueriesIdxMap := make(map[string]int)

for _, bi := range b.queuedQueries {
for _, bi := range b.QueuedQueries {
if bi.sd == nil {
if idx, present := distinctNewQueriesIdxMap[bi.query]; present {
if idx, present := distinctNewQueriesIdxMap[bi.SQL]; present {
bi.sd = distinctNewQueries[idx]
} else {
sd := &pgconn.StatementDescription{
SQL: bi.query,
SQL: bi.SQL,
}
distinctNewQueriesIdxMap[sd.SQL] = len(distinctNewQueries)
distinctNewQueries = append(distinctNewQueries, sd)
Expand Down Expand Up @@ -1154,11 +1154,11 @@ func (c *Conn) sendBatchExtendedWithDescription(ctx context.Context, b *Batch, d
}

// Queue the queries.
for _, bi := range b.queuedQueries {
err := c.eqb.Build(c.typeMap, bi.sd, bi.arguments)
for _, bi := range b.QueuedQueries {
err := c.eqb.Build(c.typeMap, bi.sd, bi.Arguments)
if err != nil {
// we wrap the error so we the user can understand which query failed inside the batch
err = fmt.Errorf("error building query %s: %w", bi.query, err)
err = fmt.Errorf("error building query %s: %w", bi.SQL, err)
return &pipelineBatchResults{ctx: ctx, conn: c, err: err, closed: true}
}

Expand Down

0 comments on commit c90f82a

Please sign in to comment.