executeQueryDirect executes a query directly against DuckDB without any transpilation. Used for passthrough users who send DuckDB-native SQL.
(query, cmdType string)
| 16 | |
| 17 | // executeQueryDirect executes a query directly against DuckDB without any transpilation. |
| 18 | // Used for passthrough users who send DuckDB-native SQL. |
| 19 | func (c *clientConn) executeQueryDirect(query, cmdType string) error { |
| 20 | if !queryReturnsResults(query) { |
| 21 | // Handle nested BEGIN |
| 22 | if cmdType == "BEGIN" && c.txStatus == txStatusTransaction { |
| 23 | c.sendNotice("WARNING", "25001", "there is already a transaction in progress") |
| 24 | _ = c.writeCommandComplete("BEGIN") |
| 25 | _ = c.writeReadyForQuery(c.txStatus) |
| 26 | _ = c.flushWriter() |
| 27 | return nil |
| 28 | } |
| 29 | |
| 30 | // Open cursors pin the session's single DuckDB connection — release |
| 31 | // them before a transaction-end statement needs it. |
| 32 | c.closeCursorsAtTxEnd(cmdType) |
| 33 | |
| 34 | ctx, cleanup := c.queryContext() |
| 35 | defer cleanup() |
| 36 | |
| 37 | workerStatement := workerStatementWithQuery(workerOriginClient, workerOperationDirectExec, query) |
| 38 | queryStart := time.Now() |
| 39 | var queryRowsAff int64 |
| 40 | var queryFinalErr error |
| 41 | c.logWorkerStatementStarted(workerStatement) |
| 42 | defer func() { |
| 43 | c.logWorkerStatementFinished(workerStatement, queryStart, queryRowsAff, queryFinalErr) |
| 44 | }() |
| 45 | |
| 46 | runExec := func() (ExecResult, error) { |
| 47 | return c.executor.ExecContext(ctx, query) |
| 48 | } |
| 49 | |
| 50 | result, err := runExec() |
| 51 | if err != nil && c.txStatus == txStatusIdle && isDuckLakeTransactionConflict(err) { |
| 52 | ducklakeConflictTotal.Inc() |
| 53 | result, err = retryOnConflict(runExec) |
| 54 | } |
| 55 | if err != nil { |
| 56 | result, err, _ = recoverAbortedTransaction( |
| 57 | err, |
| 58 | c.txStatus == txStatusIdle, |
| 59 | func() error { |
| 60 | _, rollbackErr := c.executor.ExecContext(context.Background(), "ROLLBACK") |
| 61 | return rollbackErr |
| 62 | }, |
| 63 | func() (ExecResult, error) { |
| 64 | return c.executor.ExecContext(ctx, query) |
| 65 | }, |
| 66 | ) |
| 67 | } |
| 68 | if err != nil { |
| 69 | queryFinalErr = err |
| 70 | errCode := classifyErrorCode(err) |
| 71 | errMsg := err.Error() |
| 72 | if c.isCallerCancellation(err) { |
| 73 | errMsg = "canceling statement due to user request" |
| 74 | } else { |
| 75 | c.logQueryError(query, err) |