fix(kernel): never return ErrBadConn on the execute/read path (H1)

toDriverError mapped a NetworkError/Unavailable status to driver.ErrBadConn on
every path, including execute and result-read. On the statement path that is
unsafe: a network failure surfaced *after* the SEA statement was sent may have
committed server-side, and driver.ErrBadConn makes database/sql transparently
re-run the whole statement — a silent duplicate write for a non-idempotent
INSERT/UPDATE/MERGE. This violates Go's own ErrBadConn rule ("never return
ErrBadConn if the server might have performed the operation").

Split classification by path:
- toConnError (session lifecycle: open/close/config — nothing executed) keeps the
  bad-conn mapping so the pool evicts a dead session. Safe: no statement ran.
- toStatementError (execute + result read) never returns ErrBadConn; it returns
  the KernelError unchanged (sqlstate preserved).

This restores the contract the rest of the stack already honors: the kernel itself
classifies ExecuteStatement as NonIdempotent (retried only on connect-phase
failures, per src/client/retry.rs), and the Thrift backend marks ExecuteStatement
non-retryable. The kernel has already exhausted its safe internal retries by the
time Go sees the error, so a second database/sql-level retry was pure hazard.

Tests: TestToStatementErrorNeverBadConn asserts no ErrBadConn for
network/unavailable/unauthenticated on the statement path; TestToConnError keeps
the session-path eviction behavior.

Co-authored-by: Isaac
Signed-off-by: Mani Kaustubh Mathur <mani.mathur@databricks.com>

Author: mani-mathur-arch

internal/backend/kernel/backend.go

@@ -83,7 +83,7 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 
 	var cfg *C.KernelSessionConfig
 	if err := call(func() C.KernelStatusCode { return C.kernel_session_config_new(&cfg) }); err != nil {
-		return fmt.Errorf("kernel: config_new: %w", toDriverError(err))
+		return fmt.Errorf("kernel: config_new: %w", toConnError(err))
 	}
 	// kernel_session_open consumes the config on EVERY path — success and
 	// failure alike (it reclaims the box up front). So we free the config
@@ -107,15 +107,15 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		if err := call(func() C.KernelStatusCode {
 			return C.kernel_session_config_set_warehouse(cfg, host.c, wh.c)
 		}); err != nil {
-			return fmt.Errorf("kernel: set_warehouse: %w", toDriverError(err))
+			return fmt.Errorf("kernel: set_warehouse: %w", toConnError(err))
 		}
 	} else {
 		path := newCStr(k.cfg.HTTPPath)
 		defer path.free()
 		if err := call(func() C.KernelStatusCode {
 			return C.kernel_session_config_set_http_path(cfg, host.c, path.c)
 		}); err != nil {
-			return fmt.Errorf("kernel: set_http_path: %w", toDriverError(err))
+			return fmt.Errorf("kernel: set_http_path: %w", toConnError(err))
 		}
 	}
 
@@ -124,7 +124,7 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 	if err := call(func() C.KernelStatusCode {
 		return C.kernel_session_config_set_auth_pat(cfg, tok.c)
 	}); err != nil {
-		return fmt.Errorf("kernel: set_auth_pat: %w", toDriverError(err))
+		return fmt.Errorf("kernel: set_auth_pat: %w", toConnError(err))
 	}
 
 	// TLS: crypto/tls's InsecureSkipVerify accepts any server cert, so relax both
@@ -135,12 +135,12 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		if err := call(func() C.KernelStatusCode {
 			return C.kernel_session_config_set_tls_allow_self_signed(cfg, C.bool(true))
 		}); err != nil {
-			return fmt.Errorf("kernel: set_tls_allow_self_signed: %w", toDriverError(err))
+			return fmt.Errorf("kernel: set_tls_allow_self_signed: %w", toConnError(err))
 		}
 		if err := call(func() C.KernelStatusCode {
 			return C.kernel_session_config_set_tls_skip_hostname_verification(cfg, C.bool(true))
 		}); err != nil {
-			return fmt.Errorf("kernel: set_tls_skip_hostname_verification: %w", toDriverError(err))
+			return fmt.Errorf("kernel: set_tls_skip_hostname_verification: %w", toConnError(err))
 		}
 	}
 
@@ -154,7 +154,7 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		if err := call(func() C.KernelStatusCode {
 			return C.kernel_session_config_set_proxy(cfg, url.c, nil, nil, nil)
 		}); err != nil {
-			return fmt.Errorf("kernel: set_proxy: %w", toDriverError(err))
+			return fmt.Errorf("kernel: set_proxy: %w", toConnError(err))
 		}
 	}
 
@@ -169,7 +169,7 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		ck.free()
 		cv.free()
 		if errSet != nil {
-			return fmt.Errorf("kernel: set_session_conf[%s]: %w", key, toDriverError(errSet))
+			return fmt.Errorf("kernel: set_session_conf[%s]: %w", key, toConnError(errSet))
 		}
 	}
 
@@ -180,7 +180,7 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 	err := call(func() C.KernelStatusCode { return C.kernel_session_open(cfg, &sess) })
 	consumed = true
 	if err != nil {
-		return fmt.Errorf("kernel: session_open: %w", toDriverError(err))
+		return fmt.Errorf("kernel: session_open: %w", toConnError(err))
 	}
 	k.session = sess
 	k.valid = true
@@ -201,7 +201,7 @@ func (k *KernelBackend) CloseSession(ctx context.Context) error {
 	err := call(func() C.KernelStatusCode { return C.kernel_session_close(k.session) })
 	k.session = nil
 	k.valid = false
-	return toDriverError(err)
+	return toConnError(err)
 }
 
 // SessionValid backs conn.IsValid → pool eviction. No I/O; inspects state

internal/backend/kernel/cgo.go

@@ -202,12 +202,13 @@ func isBadConnection(code int) bool {
 	}
 }
 
-// toDriverError classifies a kernel error for the pool: a status that means the
-// session is unusable is wrapped as a bad-connection error (which identifies as
-// driver.ErrBadConn, so database/sql evicts the conn), matching how the Thrift
-// backend surfaces connection loss. Other kernel errors — and plain
-// (non-KernelError) errors — are returned unchanged, carrying their sqlstate.
-func toDriverError(err error) error {
+// toConnError classifies a kernel error on a SESSION-lifecycle path (open/close/
+// config, where nothing has executed): a status that means the session is unusable
+// is wrapped as a bad-connection error, which identifies as driver.ErrBadConn so
+// database/sql evicts the conn from the pool. Safe here because no statement ran,
+// so there is nothing for database/sql to unsafely re-run. Other kernel errors —
+// and plain (non-KernelError) errors — are returned unchanged, carrying sqlstate.
+func toConnError(err error) error {
 	if err == nil {
 		return nil
 	}
@@ -221,6 +222,28 @@ func toDriverError(err error) error {
 	return ke
 }
 
+// toStatementError classifies a kernel error on the STATEMENT path (execute and
+// result read). It NEVER returns driver.ErrBadConn: once a statement has been
+// sent, a network/unavailable failure surfaced afterward may have committed
+// server-side, and driver.ErrBadConn would make database/sql transparently
+// re-run the statement — a silent duplicate write for a non-idempotent
+// INSERT/UPDATE/MERGE. This mirrors the kernel's own retry contract
+// (ExecuteStatement is NonIdempotent, retried only on connect-phase failures) and
+// the Thrift backend (ExecuteStatement is non-retryable), and honors Go's
+// driver.ErrBadConn rule ("never return ErrBadConn if the server might have
+// performed the operation"). The kernel has already exhausted its safe internal
+// retries by the time we see the error. Returns the KernelError (or plain error)
+// unchanged, carrying sqlstate.
+func toStatementError(err error) error {
+	if err == nil {
+		return nil
+	}
+	if ke, ok := err.(*KernelError); ok {
+		return ke
+	}
+	return err
+}
+
 // cStr wraps C.CString with a guaranteed free. The kernel copies strings into
 // owned Rust memory on receipt, so freeing immediately after the call is safe.
 // Use: cs := newCStr(s); defer cs.free(); ...C.fn(cs.c)...

internal/backend/kernel/kernel_test.go

@@ -28,24 +28,50 @@ func TestIsBadConnection(t *testing.T) {
 	}
 }
 
-// toDriverError wraps a session-unusable KernelError as driver.ErrBadConn (so
-// database/sql evicts the conn) and leaves other errors, and their sqlstate,
-// intact.
-func TestToDriverError(t *testing.T) {
-	if toDriverError(nil) != nil {
+// toConnError (session-lifecycle path) wraps a session-unusable KernelError as
+// driver.ErrBadConn so database/sql evicts the conn, and leaves other errors and
+// their sqlstate intact.
+func TestToConnError(t *testing.T) {
+	if toConnError(nil) != nil {
 		t.Fatal("nil should map to nil")
 	}
 
 	badConn := &KernelError{Code: statusUnavailable, Message: "gone"}
-	if !errors.Is(toDriverError(badConn), driver.ErrBadConn) {
-		t.Errorf("unavailable kernel error should identify as driver.ErrBadConn")
+	if !errors.Is(toConnError(badConn), driver.ErrBadConn) {
+		t.Errorf("unavailable kernel error on the session path should identify as driver.ErrBadConn")
 	}
 
 	sqlErr := &KernelError{Code: statusSqlError, Message: "boom", SQLState: "42703"}
-	got := toDriverError(sqlErr)
-	ke, ok := got.(*KernelError)
+	ke, ok := toConnError(sqlErr).(*KernelError)
 	if !ok {
-		t.Fatalf("sql error should remain a *KernelError, got %T", got)
+		t.Fatalf("sql error should remain a *KernelError, got %T", toConnError(sqlErr))
+	}
+	if ke.SQLState != "42703" {
+		t.Errorf("sqlstate lost: got %q", ke.SQLState)
+	}
+}
+
+// toStatementError (execute/read path) must NEVER return driver.ErrBadConn — even
+// for a network/unavailable status — so database/sql cannot transparently re-run a
+// statement that may have already executed server-side (silent duplicate write).
+func TestToStatementErrorNeverBadConn(t *testing.T) {
+	if toStatementError(nil) != nil {
+		t.Fatal("nil should map to nil")
+	}
+
+	for _, code := range []int{statusUnavailable, statusNetworkError, statusUnauthenticated} {
+		err := toStatementError(&KernelError{Code: code, Message: "post-execute failure"})
+		if errors.Is(err, driver.ErrBadConn) {
+			t.Errorf("statement-path error (code=%d) must NOT identify as driver.ErrBadConn "+
+				"(would let database/sql re-run a possibly-committed statement)", code)
+		}
+	}
+
+	// sqlstate still preserved.
+	sqlErr := &KernelError{Code: statusSqlError, Message: "boom", SQLState: "42703"}
+	ke, ok := toStatementError(sqlErr).(*KernelError)
+	if !ok {
+		t.Fatalf("sql error should remain a *KernelError, got %T", toStatementError(sqlErr))
 	}
 	if ke.SQLState != "42703" {
 		t.Errorf("sqlstate lost: got %q", ke.SQLState)

internal/backend/kernel/operation.go

@@ -40,7 +40,7 @@ func (k *KernelBackend) execute(ctx context.Context, req backend.ExecRequest) (b
 	if err := call(func() C.KernelStatusCode {
 		return C.kernel_session_new_statement(k.session, &stmt)
 	}); err != nil {
-		return &kernelOp{}, fmt.Errorf("kernel: new_statement: %w", toDriverError(err))
+		return &kernelOp{}, fmt.Errorf("kernel: new_statement: %w", toStatementError(err))
 	}
 
 	sql := newCStr(req.Query)
@@ -49,7 +49,7 @@ func (k *KernelBackend) execute(ctx context.Context, req backend.ExecRequest) (b
 	}); err != nil {
 		sql.free()
 		C.kernel_statement_close(stmt)
-		return &kernelOp{}, fmt.Errorf("kernel: set_sql: %w", toDriverError(err))
+		return &kernelOp{}, fmt.Errorf("kernel: set_sql: %w", toStatementError(err))
 	}
 	sql.free()
 
@@ -131,7 +131,7 @@ func (k *KernelBackend) execute(ctx context.Context, req backend.ExecRequest) (b
 		}
 		klog("Execute failed: %v", execErr)
 		op.close()
-		return op, fmt.Errorf("kernel: execute: %w", toDriverError(execErr))
+		return op, fmt.Errorf("kernel: execute: %w", toStatementError(execErr))
 	}
 	op.exec = exec
 	// Capture the modified-row count now, while exec is live — the operation is
@@ -187,7 +187,7 @@ func (o *kernelOp) Results(ctx context.Context, callbacks *dbsqlrows.TelemetryCa
 		// Operation.Close on a Results error — so close the handles here to avoid
 		// leaking the statement / executed handle (and its server operation).
 		o.close()
-		return nil, fmt.Errorf("kernel: get_result_stream: %w", toDriverError(err))
+		return nil, fmt.Errorf("kernel: get_result_stream: %w", toStatementError(err))
 	}
 	return newKernelRows(ctx, o, stream, callbacks)
 }
@@ -235,5 +235,5 @@ func (o *kernelOp) close() bool {
 // already carries the sqlstate (see KernelError), so this returns cause as-is
 // (nil when cause is nil), matching the neutral contract.
 func (o *kernelOp) ExecutionError(ctx context.Context, cause error) error {
-	return toDriverError(cause)
+	return toStatementError(cause)
 }

internal/backend/kernel/rows.go

@@ -59,7 +59,7 @@ func newKernelRows(ctx context.Context, op *kernelOp, stream *C.kernel_result_st
 		return C.kernel_result_stream_get_schema(stream, &csch)
 	}); err != nil {
 		r.Close()
-		return nil, fmt.Errorf("kernel: get_schema: %w", toDriverError(err))
+		return nil, fmt.Errorf("kernel: get_schema: %w", toStatementError(err))
 	}
 	sch, err := cdata.ImportCArrowSchema((*cdata.CArrowSchema)(unsafe.Pointer(&csch)))
 	if err != nil {
@@ -151,7 +151,7 @@ func (r *kernelRows) nextBatch() error {
 	if err := call(func() C.KernelStatusCode {
 		return C.kernel_result_stream_next_batch(r.stream, &carr, &csch)
 	}); err != nil {
-		return fmt.Errorf("kernel: next_batch: %w", toDriverError(err))
+		return fmt.Errorf("kernel: next_batch: %w", toStatementError(err))
 	}
 	if carr.release == nil {
 		r.eof = true