| 926 | } |
| 927 | if err != nil { |
| 928 | return 0, status.Errorf(codes.Internal, "start update drain tracking: %v", err) |
| 929 | } |
| 930 | releaseOnReturn := true |
| 931 | defer func() { |
| 932 | if releaseOnReturn { |
| 933 | finishDrain() |
| 934 | } |
| 935 | }() |
| 936 | |
| 937 | // Handle empty queries (e.g., ";" from PostgreSQL client pings). |
| 938 | if isEmptyFlightQuery(query) { |
| 939 | return 0, nil |
| 940 | } |
| 941 | session.progress.queryActive.Store(true) |
| 942 | defer session.progress.queryActive.Store(false) |
| 943 | endTxnWork := ttx.beginWork() |
| 944 | defer endTxnWork() |
| 945 | execStartedAt := clearProfilingOutput() |
| 946 | |
| 947 | execFn := func() (sql.Result, error) { |
| 948 | return session.exec(ctx, tx, query) |
| 949 | } |
| 950 | |
| 951 | // Determine whether this statement is safe to retry on transient errors. |
| 952 | // |
| 953 | // Never retry when inside a transaction (Flight SQL or SQL-level): |
| 954 | // - Transaction control stmts (COMMIT/ROLLBACK/END): retrying after DuckDB |
| 955 | // internally rolls back produces "no transaction is active". |
| 956 | // - Any other stmt inside a transaction: a transient error causes DuckDB to |
| 957 | // roll back the transaction internally. Retrying would succeed in autocommit |
| 958 | // mode, masking the rollback. The client would later COMMIT and get |
| 959 | // "no transaction is active". |
| 960 | // |
| 961 | // Only retry for autocommit statements (no Flight SQL txn, no SQL-level txn). |
| 962 | inTransaction := tx != nil || session.sqlTxActive.Load() |
| 963 | isTxControl := isTransactionControlStmt(query) |
| 964 | |
| 965 | var result sql.Result |
| 966 | var execErr error |
| 967 | if inTransaction || isTxControl { |
| 968 | result, execErr = execFn() |
| 969 | } else { |
| 970 | result, execErr = retryOnTransient(session.Logger(), execFn) |
| 971 | } |
| 972 | |
| 973 | // Conflict retry for autocommit only (see GetFlightInfoStatement comment). |
| 974 | if shouldRetryDuckLakeConflict(execErr, inTransaction) { |
| 975 | ducklakeConflictTotal.Inc() |
| 976 | result, execErr = retryOnConflict(session.Logger(), func() (sql.Result, error) { |
| 977 | return session.execConn(ctx, query) |
| 978 | }) |
| 979 | } |
| 980 | if execErr != nil { |
| 981 | result, execErr, _ = recoverAbortedTransaction( |
| 982 | session.Logger(), |
| 983 | execErr, |
| 984 | !inTransaction, |
| 985 | func() error { return session.rollbackConn(context.Background()) }, |