apache/beam · error · RuntimeError

Database operation failed

Error message

Database operation failed: {e}

What it means

_execute_query wraps the actual database execution; when the query, transaction, or commit fails inside the inner try block it rolls back and re-raises as RuntimeError('Database operation failed: ...') chaining the original DBAPI exception.

Solutions

1. Read the chained exception (e.__cause__) to see the underlying DB error.
2. Verify the SQL and parameter names against the actual schema.
3. Check network/service-account connectivity to the Cloud SQL instance.
4. Retry transient connection errors; fix the query text for deterministic errors.

Example fix

# before: RuntimeError: Database operation failed: column "usr_id" does not exist
TableFieldsQueryConfig(table_name="users", where_clause_template="usr_id = {id}", ...)
# after
TableFieldsQueryConfig(table_name="users", where_clause_template="id = {id}", ...)

When it happens

Trigger: Any DBAPI exception during cursor execution or commit: bad SQL syntax, missing table/column, constraint violations, dropped connections mid-query, wrong parameter binding types.

Common situations: Malformed where_clause_template producing invalid SQL; table renamed in the database; transient Cloud SQL connection resets; values violating NOT NULL or unique constraints.

Understand the failure class

Background: Database query failed: Internal Server Error 500s wrapping SQL, Prisma, and connection failures — what to check first — this error's family across 16 libraries.

Related errors

• Could not execute the query. Please check if the query is…
• A transform with label
• Attempting to create database
• Attempting to create database
• Attempting to drop database

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

Appendix: source

Thrown at sdks/python/apache_beam/transforms/enrichment_handlers/cloudsql.py:405

      transaction = connection.begin()
      try:
        if params:
          result = connection.execute(text(query), params)
        else:
          result = connection.execute(text(query))
        # Materialize results while transaction is active.
        data: Union[list[dict[str, Any]], dict[str, Any]]
        if is_batch:
          data = [row._asdict() for row in result]
        else:
          result_row = result.first()
          data = result_row._asdict() if result_row else {}
        # Explicitly commit the transaction.
        transaction.commit()
        return data
      except Exception as e:
        transaction.rollback()
        raise RuntimeError(f"Database operation failed: {e}") from e
    except Exception as e:
      raise Exception(
          f'Could not execute the query. Please check if the query is properly '
          f'formatted and the table exists. {e}') from e
    finally:
      if connection:
        connection.close()

  def _build_batch_query(
      self, requests: list[beam.Row], batch_size: int) -> str:
    """Build batched query with unique parameter names for multiple requests.

    This method extracts parameter placeholders from the where_clause_template
    using regex and creates unique parameter names for each batch item. The
    parameter names in the template can be any valid identifiers (e.g., :id,
    :param_0, :user_name) and don't need to match field names exactly.

    For batch queries, placeholders are replaced with unique names like

View on GitHub (pinned to 12126d8942)