apache/beam · error

failed to run query

Error message

failed to run query: %v

What it means

The prepared statement is executed with statement.QueryContext(ctx); if execution fails, the error is wrapped as "failed to run query: %v" with the query text. Unlike prepare, this covers runtime execution failures: permission errors, constraint/data issues, server-side cancellation, and context timeouts during execution.

Solutions

1. Run the query manually with the same DB credentials to reproduce and read the server error.
2. Grant SELECT privileges on the referenced tables to the pipeline's database user.
3. Increase the context timeout / pipeline timeouts if the query legitimately runs long, or optimize the query (indexes, filters).
4. Check DB server logs for lock contention, crashes, or connection limits at the failure time.

rows, err := stmt.QueryContext(ctx)
if err != nil {
	if ctx.Err() != nil { return fmt.Errorf("query canceled/timed out: %w", ctx.Err()) }
	return fmt.Errorf("failed to run query: %w", err)
}

When it happens

Trigger: statement.QueryContext(ctx) returns an error: query execution rejected by the server (insufficient privileges, deadlock, lock timeout), context canceled/expired mid-query, or connection loss during execution.

Common situations: Querying tables the pipeline's DB user cannot read; long-running queries hitting context deadlines on slow networks; database failovers; row-level security blocking access.

Understand the failure class

Background: "query failed", "%w: SQL error" — wrapped database query errors in Go libraries explained — this error's family across 3 libraries.

Related errors

• failed to prepare query
• failed to open database
• failed to scan
• Attempting to create database
• Attempting to create database

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/a97dfc8e9ab38b50. Report an issue: GitHub.

Appendix: source

Thrown at sdks/go/pkg/beam/io/databaseio/database.go:85

	// Type is the encoded schema type.
	Type beam.EncodedType `json:"type"`
}

func (f *queryFn) ProcessElement(ctx context.Context, _ []byte, emit func(beam.X)) error {
	//TODO move DB Open and Close to Setup and Teardown methods or StartBundle and FinishBundle
	db, err := sql.Open(f.Driver, f.Dsn)
	if err != nil {
		return errors.Wrapf(err, "failed to open database: %v", f.Driver)
	}
	defer db.Close()
	statement, err := db.PrepareContext(ctx, f.Query)
	if err != nil {
		return errors.Wrapf(err, "failed to prepare query: %v", f.Query)
	}
	defer statement.Close()
	rows, err := statement.QueryContext(ctx)
	if err != nil {
		return errors.Wrapf(err, "failed to run query: %v", f.Query)
	}
	defer rows.Close()
	var mapper rowMapper
	var columns []string
	for rows.Next() {
		reflectRow := reflect.New(f.Type.T)
		row := reflectRow.Interface() // row : *T
		if mapper == nil {
			columns, err = rows.Columns()
			if err != nil {
				return err
			}
			columnsTypes, _ := rows.ColumnTypes()
			if mapper, err = newQueryMapper(columns, columnsTypes, f.Type.T); err != nil {
				return errors.WithContext(err, "creating rowValues mapper")
			}
		}
		rowValues, err := mapper(reflectRow)

View on GitHub (pinned to 12126d8942)