feat(kernel): close deferred feature rows — OAuth M2M/U2M, metric-view, initial namespace

auth/auth.go

@@ -9,6 +9,28 @@ type Authenticator interface {
 	Authenticate(*http.Request) error
 }
 
+// M2MCredentialsProvider is implemented by the OAuth M2M authenticator to expose
+// its raw client-credentials. The SEA-via-kernel backend reads these to drive the
+// kernel's own M2M flow (the kernel owns the token exchange), rather than using the
+// authenticator's Authenticate method. This keeps cfg.Authenticator the single
+// source of truth for auth on both backends — the kernel selects M2M by asserting
+// this interface, so the last WithX option applied wins, exactly as on Thrift.
+type M2MCredentialsProvider interface {
+	// M2MCredentials returns the client id and client secret. (The kernel's C-ABI
+	// M2M setter takes no scopes — it applies its own default scope set, matching
+	// the Go authenticator's default — so scopes are not exposed here.)
+	M2MCredentials() (clientID, clientSecret string)
+}
+
+// U2MCredentialsProvider is implemented by the OAuth U2M authenticator to expose
+// the cloud-inferred client id the kernel should use for its browser/PKCE flow (so
+// the kernel path uses the same client id the Thrift path would). See
+// M2MCredentialsProvider for why this rather than a parallel config carrier.
+type U2MCredentialsProvider interface {
+	// U2MClientID returns the OAuth client id for the U2M browser flow.
+	U2MClientID() string
+}
+
 type AuthType int
 
 const (

auth/oauth/m2m/m2m.go

@@ -36,6 +36,13 @@ type authClient struct {
 	mx           sync.Mutex
 }
 
+// M2MCredentials exposes the raw client-credentials so the SEA-via-kernel backend
+// can drive the kernel's own M2M flow. Implements auth.M2MCredentialsProvider,
+// which keeps cfg.Authenticator the single source of truth for auth mode.
+func (c *authClient) M2MCredentials() (clientID, clientSecret string) {
+	return c.clientID, c.clientSecret
+}
+
 // Auth will start the OAuth Authorization Flow to authenticate the cli client
 // using the users credentials in the browser. Compatible with SSO.
 func (c *authClient) Authenticate(r *http.Request) error {

auth/oauth/u2m/authenticator.go

@@ -78,6 +78,12 @@ type u2mAuthenticator struct {
 	mx          sync.Mutex
 }
 
+// U2MClientID exposes the cloud-inferred OAuth client id so the SEA-via-kernel
+// backend uses the same client id for the kernel's browser/PKCE flow that the
+// Thrift path would. Implements auth.U2MCredentialsProvider, keeping
+// cfg.Authenticator the single source of truth for auth mode.
+func (c *u2mAuthenticator) U2MClientID() string { return c.clientID }
+
 // Auth will start the OAuth Authorization Flow to authenticate the cli client
 // using the users credentials in the browser. Compatible with SSO.
 func (c *u2mAuthenticator) Authenticate(r *http.Request) error {

doc.go

@@ -197,20 +197,24 @@ path databricks_sql_kernel, which would match nothing):
 	# kernel logs plus its HTTP stack (hyper/reqwest):
 	DBSQL_KERNEL_DEBUG=1 RUST_LOG=debug ./your_app 2>&1
 
-The kernel backend currently supports PAT authentication; reading scalar, nested,
-and complex-typed results (CloudFetch is handled transparently); context
-cancellation during execute (a cancelled ctx fires a real server-side cancel; on
-the read path cancellation is honored at result-batch boundaries, not mid-fetch);
-and the TLS, proxy, and session-conf (query tags, statement timeout, time zone)
-connection options. OAuth (M2M/U2M), initial catalog/schema, and
-WithEnableMetricViewMetadata are not yet supported and return a clear error at
-connect time rather than being silently ignored; likewise WithTimeout (a server
-query timeout the kernel C ABI can't set) and WithRetries used to disable retries
-(the kernel retries internally). Bound query parameters are likewise not yet
-supported and return a clear error at execute time (they arrive per-query, not at
-connect). None of these is silently ignored. (Metadata is issued as ordinary SQL
-— SHOW/DESCRIBE/information_schema — and runs on this backend like any other
-query.)
+The kernel backend currently supports PAT and OAuth (M2M via WithClientCredentials,
+and U2M via the authType=oauthU2M DSN param — the interactive browser/PKCE flow is
+owned by the kernel) authentication; reading scalar, nested, and complex-typed
+results (CloudFetch is handled transparently); context cancellation during execute
+(a cancelled ctx fires a real server-side cancel; on the read path cancellation is
+honored at result-batch boundaries, not mid-fetch); the initial namespace
+(WithInitialNamespace catalog/schema, applied post-connect via USE CATALOG / USE
+SCHEMA since the kernel C ABI has no namespace setter); metric-view metadata
+(WithEnableMetricViewMetadata, sent as the same server session conf the Thrift
+backend uses); and the TLS, proxy, and session-conf (query tags, statement timeout,
+time zone) connection options. WithTimeout (a server query timeout the kernel C ABI
+can't set) and WithRetries used to disable retries (the kernel retries internally)
+return a clear error at connect time rather than being silently ignored; token-
+provider, external/static, and federated authenticators are likewise not supported
+and rejected loudly. Bound query parameters are not yet supported and return a clear
+error at execute time (they arrive per-query, not at connect). None of these is
+silently ignored. (Metadata is issued as ordinary SQL — SHOW/DESCRIBE/
+information_schema — and runs on this backend like any other query.)
 
 WithMaxRows and retry tuning (WithRetries with a positive limit) are accepted but
 not applied on the kernel path: the kernel manages result fetching and retries

internal/backend/kernel/auth.go

@@ -0,0 +1,29 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag:
+// the Auth descriptor is a plain data type with no cgo, so it compiles in the
+// default build too (and OpenSession, which is tagged, maps it to the kernel's
+// set_auth_* C setters). Its consumer joinScopes lives in the tagged backend.go
+// next to setAuth so the default build's unused-func lint doesn't flag it.
+
+// AuthMode selects which kernel auth form OpenSession applies.
+type AuthMode int
+
+const (
+	AuthPAT AuthMode = iota // personal access token
+	AuthM2M                 // OAuth client-credentials (client id + secret)
+	AuthU2M                 // OAuth user-to-machine (browser/PKCE; kernel-owned flow)
+)
+
+// Auth is the resolved auth descriptor for a kernel connection. Only the fields
+// for Mode are populated. The connector fills it from the driver config (see
+// validateKernelConfig / toKernelAuth); OpenSession maps it to exactly one
+// kernel_session_config_set_auth_* call.
+type Auth struct {
+	Mode         AuthMode
+	Token        string   // PAT
+	ClientID     string   // M2M + U2M (U2M: the cloud-inferred Go client id)
+	ClientSecret string   // M2M
+	Scopes       []string // U2M (the kernel M2M setter takes no scopes)
+	RedirectPort uint16   // U2M browser-redirect port; 0 = kernel default
+}

internal/backend/kernel/backend.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/databricks/databricks-sql-go/internal/backend"
@@ -25,7 +26,7 @@ type Config struct {
 	Host        string // workspace hostname, no scheme
 	HTTPPath    string // e.g. /sql/1.0/warehouses/abc123 (carries ?o= org routing)
 	WarehouseID string // bare warehouse id; preferred over HTTPPath when set
-	Token       string // PAT (dapi...)
+	Auth        Auth   // PAT / OAuth M2M / OAuth U2M
 
 	// SessionConf carries server-bound session confs verbatim — the same map the
 	// Thrift backend forwards (STATEMENT_TIMEOUT, QUERY_TAGS, TIMEZONE, …).
@@ -46,6 +47,13 @@ type Config struct {
 	// Location is the session time zone used to render DATE / TIMESTAMP values,
 	// matching the Thrift path which returns them in this location. nil means UTC.
 	Location *time.Location
+
+	// Catalog / Schema select the initial namespace. The kernel C ABI has no
+	// catalog/schema config setter, so OpenSession applies them post-connect by
+	// running USE CATALOG / USE SCHEMA (the OSS ODBC driver's workaround). Empty
+	// leaves the session in the server default namespace.
+	Catalog string
+	Schema  string
 }
 
 // KernelBackend implements backend.Backend over the kernel C ABI. One backend
@@ -119,12 +127,8 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 		}
 	}
 
-	tok := newCStr(k.cfg.Token)
-	defer tok.free()
-	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", toConnError(err))
+	if err := k.setAuth(cfg); err != nil {
+		return err
 	}
 
 	// TLS: crypto/tls's InsecureSkipVerify accepts any server cert, so relax both
@@ -187,10 +191,118 @@ func (k *KernelBackend) OpenSession(ctx context.Context) error {
 	// The C ABI exposes no formatted session-id accessor; use the handle pointer
 	// as a stable per-conn id for logging / telemetry correlation.
 	k.sessionID = fmt.Sprintf("kernel-%p", sess)
+
+	// Initial namespace: the kernel C ABI has no catalog/schema config setter, so
+	// select it post-connect with USE CATALOG / USE SCHEMA (the OSS ODBC driver's
+	// approach). A failure here means the session is not in the requested namespace
+	// — a correctness precondition, like Thrift's InitialNamespace — so fail the
+	// connect and close the session we just opened (the connector does not call
+	// CloseSession on an OpenSession error).
+	if err := k.applyInitialNamespace(ctx); err != nil {
+		C.kernel_session_close(sess)
+		k.session = nil
+		k.valid = false
+		return err
+	}
+
 	klog("OpenSession OK session=%s", k.sessionID)
 	return nil
 }
 
+// setAuth applies the resolved auth form to the session config via exactly one
+// kernel_session_config_set_auth_* call. PAT and M2M are plain value setters; U2M
+// records the client id / redirect port / scopes and the kernel owns the browser
+// (PKCE) flow, started when the session opens. Empty string args are passed as NULL
+// so the kernel applies its own defaults (e.g. U2M's public client / default port).
+func (k *KernelBackend) setAuth(cfg *C.KernelSessionConfig) error {
+	switch k.cfg.Auth.Mode {
+	case AuthM2M:
+		clientID := newCStr(k.cfg.Auth.ClientID)
+		defer clientID.free()
+		secret := newCStr(k.cfg.Auth.ClientSecret)
+		defer secret.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_auth_m2m(cfg, clientID.c, secret.c)
+		}); err != nil {
+			return fmt.Errorf("kernel: set_auth_m2m: %w", toConnError(err))
+		}
+	case AuthU2M:
+		// client id / scopes are optional: NULL when empty lets the kernel use its
+		// public client / default scopes. We pass Go's cloud-inferred client id when
+		// set, so the kernel uses the same client id the Thrift path would.
+		clientID := newCStrOrNull(k.cfg.Auth.ClientID)
+		defer clientID.free()
+		scopes := newCStrOrNull(joinScopes(k.cfg.Auth.Scopes))
+		defer scopes.free()
+		if err := call(func() C.KernelStatusCode {
+			return C.kernel_session_config_set_auth_u2m(cfg, clientID.c, C.uint16_t(k.cfg.Auth.RedirectPort), scopes.c)
+		}); err != nil {
+			return fmt.Errorf("kernel: set_auth_u2m: %w", toConnError(err))
+		}
+	default: // AuthPAT
+		tok := newCStr(k.cfg.Auth.Token)
+		defer tok.free()
+		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", toConnError(err))
+		}
+	}
+	return nil
+}
+
+// joinScopes renders U2M scopes as the comma-separated form the kernel U2M setter
+// expects. Empty (no scopes) yields "" so setAuth passes NULL and the kernel
+// applies its default scope set.
+func joinScopes(scopes []string) string {
+	return strings.Join(scopes, ",")
+}
+
+// trySetAuth allocates a throwaway session config, applies auth to it, and frees
+// it — a test seam so TestSetAuthByMode can exercise the real cgo setter path
+// without putting cgo in a _test.go file (which Go forbids). Not used in
+// production. Returns the setter error (nil on success).
+func trySetAuth(auth Auth) error {
+	var cfg *C.KernelSessionConfig
+	if err := call(func() C.KernelStatusCode { return C.kernel_session_config_new(&cfg) }); err != nil {
+		return fmt.Errorf("config_new: %w", err)
+	}
+	defer C.kernel_session_config_free(cfg)
+	k := &KernelBackend{cfg: Config{Auth: auth}}
+	return k.setAuth(cfg)
+}
+
+// applyInitialNamespace runs USE CATALOG / USE SCHEMA to select the configured
+// initial namespace, since the kernel C ABI exposes no catalog/schema setter.
+// Identifiers are backtick-quoted (quoteIdent) so arbitrary names are safe. A
+// no-op when neither is set.
+func (k *KernelBackend) applyInitialNamespace(ctx context.Context) error {
+	if k.cfg.Catalog != "" {
+		if err := k.runNamespaceStmt(ctx, "USE CATALOG "+quoteIdent(k.cfg.Catalog)); err != nil {
+			return fmt.Errorf("kernel: set initial catalog %q: %w", k.cfg.Catalog, err)
+		}
+	}
+	if k.cfg.Schema != "" {
+		if err := k.runNamespaceStmt(ctx, "USE SCHEMA "+quoteIdent(k.cfg.Schema)); err != nil {
+			return fmt.Errorf("kernel: set initial schema %q: %w", k.cfg.Schema, err)
+		}
+	}
+	return nil
+}
+
+// runNamespaceStmt executes a single side-effecting statement (USE …) and closes
+// the operation. USE produces no rows, so the result stream is not read. execute
+// always returns a non-nil Operation (the Backend contract), so it is closed on
+// both the success and error paths; the execute error is authoritative.
+func (k *KernelBackend) runNamespaceStmt(ctx context.Context, sql string) error {
+	op, err := k.execute(ctx, backend.ExecRequest{Query: sql})
+	_, closeErr := op.Close(ctx)
+	if err != nil {
+		return err
+	}
+	return closeErr
+}
+
 // CloseSession tears down the server-side session. Best-effort: the kernel's
 // close is async (see the C header), so an error is logged, not hard-failed.
 func (k *KernelBackend) CloseSession(ctx context.Context) error {

internal/backend/kernel/cgo.go

@@ -175,6 +175,18 @@ type cStr struct{ c *C.char }
 
 func newCStr(s string) cStr { return cStr{c: C.CString(s)} }
 
+// newCStrOrNull is like newCStr but yields a NULL C pointer for an empty string,
+// for kernel args whose "unset" sentinel is NULL (e.g. the optional U2M client id /
+// scopes, where NULL selects the kernel's own default). C.CString("") would instead
+// pass a non-NULL pointer to an empty string, which the kernel treats as a real
+// (empty) value rather than "use the default".
+func newCStrOrNull(s string) cStr {
+	if s == "" {
+		return cStr{c: nil}
+	}
+	return cStr{c: C.CString(s)}
+}
+
 func (s cStr) free() {
 	if s.c != nil {
 		C.free(unsafe.Pointer(s.c))

internal/backend/kernel/kernel_test.go

@@ -11,6 +11,33 @@ import (
 	"github.com/databricks/databricks-sql-go/internal/backend"
 )
 
+// setAuth maps each Auth mode to exactly one kernel_session_config_set_auth_*
+// value-setter. These are pure config setters (no network), so we can assert the
+// call succeeds against a freshly allocated config for every mode — exercising the
+// real cgo path (arg marshaling, NULL-for-empty on the optional U2M args) end to
+// end via the trySetAuth test helper (cgo cannot be used directly in a _test.go
+// file). A failure here means the mode→setter wiring or the C signature drifted.
+func TestSetAuthByMode(t *testing.T) {
+	cases := []struct {
+		name string
+		auth Auth
+	}{
+		{"PAT", Auth{Mode: AuthPAT, Token: "dapi-x"}},
+		{"M2M", Auth{Mode: AuthM2M, ClientID: "cid", ClientSecret: "sec"}},
+		{"U2M full", Auth{Mode: AuthU2M, ClientID: "u2m-cid", Scopes: []string{"sql", "offline_access"}, RedirectPort: 8030}},
+		// U2M with everything defaulted: empty client id / no scopes / port 0 must
+		// pass NULL / 0 so the kernel applies its own defaults (exercises newCStrOrNull).
+		{"U2M defaults", Auth{Mode: AuthU2M}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			if err := trySetAuth(c.auth); err != nil {
+				t.Errorf("setAuth(%s) = %v, want nil", c.name, err)
+			}
+		})
+	}
+}
+
 // The pure error-classifier tests (TestIsBadConnection, TestIsSessionFatal,
 // TestToConnError, TestToStatementErrorNeverBadConn) live in the untagged
 // errors_classify_test.go so they run under CGO_ENABLED=0. The tests below need a

internal/backend/kernel/namespace.go

@@ -0,0 +1,17 @@
+package kernel
+
+// This file is intentionally NOT behind the `cgo && databricks_kernel` build tag.
+// quoteIdent is pure Go (no cgo) and is unit-tested in the default CGO_ENABLED=0
+// matrix; OpenSession (tagged) calls it to build the USE CATALOG / USE SCHEMA
+// statements that select the initial namespace.
+
+import "strings"
+
+// quoteIdent renders name as a backtick-quoted Databricks SQL identifier, doubling
+// any embedded backtick. The kernel C ABI exposes no catalog/schema config setter,
+// so the initial namespace is applied post-connect by running USE CATALOG / USE
+// SCHEMA (the same workaround the OSS ODBC driver uses); quoting makes those
+// statements injection-safe for arbitrary identifier text.
+func quoteIdent(name string) string {
+	return "`" + strings.ReplaceAll(name, "`", "``") + "`"
+}

internal/backend/kernel/namespace_test.go

@@ -0,0 +1,25 @@
+package kernel
+
+import "testing"
+
+func TestQuoteIdent(t *testing.T) {
+	cases := []struct {
+		in   string
+		want string
+	}{
+		{"main", "`main`"},
+		{"my_schema", "`my_schema`"},
+		{"", "``"},
+		{"has space", "`has space`"},
+		// A backtick in the identifier is doubled so it can't terminate the quote.
+		{"a`b", "`a``b`"},
+		// One backtick → doubled to two, wrapped in two more → four total.
+		{"`", "````"},
+		{"weird`;DROP", "`weird``;DROP`"},
+	}
+	for _, c := range cases {
+		if got := quoteIdent(c.in); got != c.want {
+			t.Errorf("quoteIdent(%q) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}