How do I find failed statements in a batch so I can take them out and the rest can be successful on retry?

[ns-gzhang]

I'm using a batch to insert/update a table with dynamically generated statements. When there is one statement that fails, the entire batch will fail, and there seems no status for individual success or failure in the batch (as tag, err := res.Exec() returns an error on the first call while the offending statement is in the middle of a batch). Is there a way to tell which statements caused failure in a batch? Otherwise, how to achieve the same? (Not using a batch?) Thanks!

[jackc]

Is there a way to tell which statements caused failure in a batch?

I think you should be able to find which statement failed by calling Query on https://pkg.go.dev/github.com/jackc/pgx/v5#QueuedQuery before sending the batch.

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	"github.com/jackc/pgx/v5"
)

func main() {
	ctx := context.Background()

	conn, err := pgx.Connect(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close(ctx)

	batch := &pgx.Batch{}

	batch.Queue(`select 1`).Query(func(rows pgx.Rows) error {
		rows.Close()
		if err := rows.Err(); err != nil {
			return fmt.Errorf("first query: %w", err)
		}

		return nil
	})

	batch.Queue(`select 1 / n from generate_series(-5, 5) n`).Query(func(rows pgx.Rows) error {
		rows.Close()
		if err := rows.Err(); err != nil {
			return fmt.Errorf("second query: %w", err)
		}

		return nil
	})

	batch.Queue(`select 1`).Query(func(rows pgx.Rows) error {
		rows.Close()
		if err := rows.Err(); err != nil {
			return fmt.Errorf("third query: %w", err)
		}

		return nil
	})

	err = conn.SendBatch(ctx, batch).Close()
	fmt.Println("err:", err)
}

However, this will only work if the queries are executed. If there is a problem such as a syntax error no QueuedQuery callbacks will be called. In that case there is not presently a way to determine which was the problem.

Presumably, the batch system could be modified to wrap errors with something that included the query index but nobody has built that yet.

---

Otherwise, how to achieve the same? (Not using a batch?)

Well, not using a batch is the simplest way of course. If the network round trips are a problem then another option is pipeline mode down in the pgconn layer. But that is much lower level and difficult to use.

[ns-gzhang]

Thanks Jack for the answer.
My scenario is that batches contain all insert/update/delete, not select. The problem is I cannot guarantee the input data (from Kafka) would 100% compatible with the table column types (without checking beforehand). I try to catch them during execution (during testing at least). I believe I can only use Exec() not Query() on these statements in a batch. So my current plan is to fall back to individual statement execution if SendBatch encounters an error. I try not to use the lower level API. Let me know if there is a better way. Thanks!

[jackc]

You actually can use Query instead of Exec -- and not just for batches. You can use it anywhere. Both Exec and QueryRow can be implemented in terms of Query. For example, here is some code from https://github.com/jackc/pgxrecord, an experimental library of mine.

type DB interface {
	Query(ctx context.Context, sql string, optionsAndArgs ...interface{}) (pgx.Rows, error)
}

func exec(ctx context.Context, db DB, sql string, args []any) (pgconn.CommandTag, error) {
	rows, err := db.Query(ctx, sql, args...)
	if err != nil {
		return pgconn.CommandTag{}, err
	}
	rows.Close()

	return rows.CommandTag(), rows.Err()
}

---

However, if the problem is in the structure of the query then it will fail before the QueuedQuery callbacks are called.

But in that case, you could call Prepare for each query before creating the batch. Though that would incur a network round trip for each statement so it would negate much of the benefit of batching in the first place (unless you dropped down to pipeline mode).

---

But the only benefit of batching is reducing network latency (and implicit transactional mode). If you don't actually need that the simplest thing in this case would be to not use a batch.

[ns-gzhang]

Thanks Jack again for the follow-on explanation of the calls. Good to know Query() is applicable to I/U/D also.

I did a quick test to simulate the potential problems I may encounter. Here is a snippet of the code:

        batch := &pgx.Batch{}
        count := int64(0)
        for _, r := range b.Rows {
                batch.Queue(r.Sqltext, r.Args...).Exec(func(ct pgconn.CommandTag) error {
                        count += ct.RowsAffected()
                        return nil
                })
        }

I didn't check each and every argument in r.Args whether their types match that of the parameters in r.Sqltext. Looks like binding-in the arguments to parameters are performed before any statements are executed (on the client side?). I used Query() as you showed in my quick test, it seemed to only return an error for the statement with error in the middle of the batch, without executing any statement.
For now, I think I'm fine with falling back to individual statements on batch errors. Thanks again!