From d0c1174ecf8065a0203ebb5328e9b7248635ff54 Mon Sep 17 00:00:00 2001 From: Madhavendra Rathore Date: Tue, 11 Aug 2026 03:32:27 +0530 Subject: [PATCH 1/3] poc(kernel): pure-Go (CGO_ENABLED=0) dynamic loader for the kernel C ABI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Proof-of-concept that the closed-source SEA kernel can be driven from a PURE-Go binary — no cgo, no C compiler, no static linking — by dlopen-ing the kernel SHARED library at run time via ebitengine/purego, instead of static-linking libdatabricks_sql_kernel.a through cgo at build time. Why: the shipped kernel backend links a static .a via cgo, which forces CGO_ENABLED=1, a C toolchain on every builder, and breaks Go's free cross-compilation — the three blockers the SEA/kernel release design calls out against ever making SEA the default backend. Dynamic loading removes all three. It's the model gosnowflake uses for its own closed-source native core. Verified live against a warehouse, built CGO_ENABLED=0: dlopen -> config -> session_open -> execute -> real server query id -> teardown, all with no cgo. No kernel source change is needed — the kernel's Cargo.toml already declares crate-type cdylib and the C ABI fns are #[no_mangle] extern "C", so the .dylib already exports them. No user CUJ change: WithUseKernel and the existing build tags are untouched; this lives behind a NEW, separate tag (databricks_kernel_dynamic) and the default pure-Go build does not pull in purego. Scope: CONTROL plane only (DML/DDL happy path). The DATA plane (Arrow result batches) is deliberately excluded because arrow-go/v12's cdata importer is itself a cgo package — closing it needs a separate decision (purego C-Data importer, arrow-go v18, or a thin cgo shim). Documented in DYNAMIC_LOADER_POC.md along with the run-time lib-discovery and glibc/musl follow-ups. PoC-only shortcuts (not for merge as-is): a local `replace` for purego so the branch builds offline, dylib path via env var, and a hand-mirrored KernelError struct layout. All flagged in the doc. Co-authored-by: Isaac --- go.mod | 5 + internal/backend/kernel/DYNAMIC_LOADER_POC.md | 80 +++++++ internal/backend/kernel/dynamic_loader.go | 222 ++++++++++++++++++ .../backend/kernel/dynamic_loader_test.go | 122 ++++++++++ 4 files changed, 429 insertions(+) create mode 100644 internal/backend/kernel/DYNAMIC_LOADER_POC.md create mode 100644 internal/backend/kernel/dynamic_loader.go create mode 100644 internal/backend/kernel/dynamic_loader_test.go diff --git a/go.mod b/go.mod index b2cea939..bcc64944 100644 --- a/go.mod +++ b/go.mod @@ -46,9 +46,14 @@ require ( ) require ( + github.com/ebitengine/purego v0.10.2 github.com/hashicorp/go-retryablehttp v0.7.7 github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 github.com/pkg/errors v0.9.1 github.com/rs/zerolog v1.28.0 golang.org/x/sys v0.45.0 // indirect ) + +// PoC-only: point purego at a local checkout so this branch builds offline. +// The real PR drops this replace and lets the module proxy fetch v0.10.2. +replace github.com/ebitengine/purego => /tmp/purego-src diff --git a/internal/backend/kernel/DYNAMIC_LOADER_POC.md b/internal/backend/kernel/DYNAMIC_LOADER_POC.md new file mode 100644 index 00000000..f313dcc1 --- /dev/null +++ b/internal/backend/kernel/DYNAMIC_LOADER_POC.md @@ -0,0 +1,80 @@ +# PoC: dynamic loading of the kernel (pure-Go, CGO_ENABLED=0) + +**Status:** working proof-of-concept, control-plane only. Verified live. +**Build tag:** `databricks_kernel_dynamic` (separate from the shipped +`cgo && databricks_kernel` path — the two never build together). + +## What this proves + +The shipped kernel backend static-links `libdatabricks_sql_kernel.a` through cgo. +That forces `CGO_ENABLED=1`, a C toolchain on every builder, and it breaks Go's +free cross-compilation — the three things the SEA/kernel release design flags as +blockers to ever making SEA the default backend. + +This PoC loads the kernel **shared** library (`.dylib`/`.so`/`.dll`) at run time +with [`ebitengine/purego`](https://github.com/ebitengine/purego) and **no cgo**. +It is the model `gosnowflake` uses for its own closed-source native core. + +Verified end-to-end against a live warehouse (pecotesting), built `CGO_ENABLED=0`: + +``` +OK dlopen: bound kernel C ABI from libdatabricks_sql_kernel.dylib (CGO_ENABLED=0) +OK config: http_path + PAT set +OK session_open: connected to warehouse over SEA via the kernel +OK execute: server queryId="01f19506-ee6a-1f18-813f-4e9156bd2a4e" numModifiedRows=-1 +PROOF: pure-Go (CGO_ENABLED=0) drove the closed-source kernel control plane end-to-end. +``` + +Also verified: +- **No kernel source change.** The kernel's `Cargo.toml` already declares + `crate-type = [..., "cdylib", ...]`, so `cargo build` already emits the + `.dylib` this PoC loads. The C ABI functions are `#[no_mangle] extern "C"`, so + they are exported from the shared library as-is. +- **No user CUJ change.** `WithUseKernel` and the existing build tags are + untouched. The default pure-Go build is unchanged and does **not** pull in + purego (confirmed: `go list -deps ./...` shows no purego in the default build). + +## Scope boundary (the honest part) + +This PoC covers the **control plane** only: dlopen → config → session open → +execute → query-id / affected-rows → teardown. That is the entire happy path for +DML/DDL and any non-result statement. + +It does **not** cover the **data plane** (Arrow result batches), for one concrete +reason: the cgo backend imports result batches via `arrow-go/v12`'s `cdata` +package (`ImportCRecordBatch`), and **`cdata` is itself a cgo package** — every +non-test file in it does `import "C"`. So a fully `CGO_ENABLED=0` result-fetch +path can't just reuse `cdata`. Closing the data plane needs a separate decision: + +1. a purego-based Arrow C-Data importer (reimplement the small struct import + without cgo), or +2. move to `arrow-go` v18 and re-evaluate, or +3. accept a hybrid: purego control plane + a thin cgo shim only for C-Data import + (loses full `CGO_ENABLED=0`, keeps dynamic loading). + +That decision — plus the run-time library-discovery story (rpath / env var) and +the glibc-vs-musl matrix — is the follow-up, and is why this is a PoC PR rather +than a backend replacement. + +## Run it + +```sh +CGO_ENABLED=0 go build -tags databricks_kernel_dynamic ./internal/backend/kernel/ + +DBX_KERNEL_DYLIB=/abs/path/to/libdatabricks_sql_kernel.dylib \ +DBX_KERNEL_HOST= \ +DBX_KERNEL_HTTPATH=/sql/1.0/warehouses/ \ +DBX_KERNEL_TOKEN= \ +CGO_ENABLED=0 go test -tags databricks_kernel_dynamic \ + -run TestDynamicLoaderControlPlane ./internal/backend/kernel/ -v +``` + +## PoC-only shortcuts (not for merge as-is) + +- `go.mod` has a `replace github.com/ebitengine/purego => /tmp/purego-src` so the + branch builds offline in a sandbox. A real PR drops the replace and lets the + module proxy fetch the pinned `v0.10.2`. +- The dylib path is passed via env var. Real code resolves it next to the + executable (rpath) or a documented env var. +- `cKernelError` is a hand-mirrored struct layout. A real PR adds + `unsafe.Sizeof`/`Offsetof` assertions (the cgo path has equivalent guards). diff --git a/internal/backend/kernel/dynamic_loader.go b/internal/backend/kernel/dynamic_loader.go new file mode 100644 index 00000000..253456a2 --- /dev/null +++ b/internal/backend/kernel/dynamic_loader.go @@ -0,0 +1,222 @@ +//go:build databricks_kernel_dynamic + +// Package-level PoC: a PURE-GO (CGO_ENABLED=0) loader for the Databricks SQL +// kernel's C ABI, using ebitengine/purego to dlopen the kernel SHARED library +// (.so/.dylib/.dll) at run time instead of static-linking a .a at build time +// through cgo. +// +// Why this exists (see the driver's SEA/kernel release design). The shipped cgo +// backend (cgo.go + siblings, //go:build cgo && databricks_kernel) links a +// static libdatabricks_sql_kernel.a at build time. That forces CGO_ENABLED=1, a +// C toolchain on every builder, and it breaks Go's free cross-compilation — the +// three things that block SEA from ever becoming the default backend. This file +// proves the alternative: load the kernel as a shared library at run time with +// NO cgo, so the Go side keeps CGO_ENABLED=0 and cross-compiles freely. It is +// the model gosnowflake uses for its own closed-source native core. +// +// Scope of this PoC (deliberately narrow, so it is reviewable): +// - CONTROL plane only: dlopen -> config -> session open -> execute -> +// query-id / affected-rows -> teardown. This is the whole happy path for +// DML/DDL and any non-result statement. +// - The DATA plane (Arrow result batches) is NOT here. arrow-go/v12's cdata +// package — the zero-copy C-Data importer the cgo rows.go uses — is itself a +// cgo package (every non-test file does `import "C"`), so a fully +// CGO_ENABLED=0 result-fetch path needs a separate decision (a purego-based +// C-Data import, or arrow-go v18). Called out as the documented follow-up; +// see dynamic_loader_test.go and the PR description. +// +// Build/run this PoC (nothing static, no C compiler): +// +// CGO_ENABLED=0 go build -tags databricks_kernel_dynamic ./internal/backend/kernel/ +// DBX_KERNEL_DYLIB=/abs/path/to/libdatabricks_sql_kernel.dylib \ +// CGO_ENABLED=0 go test -tags databricks_kernel_dynamic \ +// -run TestDynamicLoaderControlPlane ./internal/backend/kernel/ -v +// +// Memory model, mirrored from the cgo path: +// - Strings handed to the kernel are copied into C memory for the call and +// freed right after; the kernel copies them into owned Rust memory on +// receipt, so freeing immediately is safe (same contract as cgo cStr). +// - Every fallible call is wrapped so the kernel's thread-local last error is +// read on the SAME OS thread (runtime.LockOSThread), closing the same +// goroutine-migration window the cgo `call` helper documents. +package kernel + +import ( + "fmt" + "runtime" + "unsafe" + + "github.com/ebitengine/purego" +) + +// kernelStatusSuccess is KernelStatusCode_Success (0). The full enum lives in +// errors_classify.go as plain ints; this loader only needs the success sentinel +// plus the classifier those constants feed. +const kernelStatusSuccess = 0 + +// dynLib holds the dlopen handle plus the kernel C ABI functions bound as Go +// func values. Only the control-plane subset needed for the PoC is bound. +// +// purego.RegisterLibFunc maps a Go signature onto a C symbol. The mapping rules +// used here: C pointer/opaque-handle types become uintptr; `const char*` +// becomes a Go string on the ARGUMENT side (purego marshals it to a C string +// for the duration of the call); KernelStatusCode (an int enum) becomes int32. +type dynLib struct { + handle uintptr + + // Lifecycle + config (KernelStatusCode kernel_*(...)) + // initLogging's file_path is a uintptr, not a string, so the caller can pass + // 0 (a real C NULL → log to stderr). purego marshals a Go "" to a non-null + // empty C string, which the kernel would treat as a filename to open — this + // is the same NULL-vs-empty distinction the cgo path handles with + // newCStrOrNull. + initLogging func(level string, filePath uintptr) int32 + configNew func(out *uintptr) int32 + configFree func(config uintptr) + configSetHTTPath func(config uintptr, host, httpPath string) int32 + configSetWH func(config uintptr, host, warehouseID string) int32 + configSetAuthPAT func(config uintptr, token string) int32 + sessionOpen func(config uintptr, out *uintptr) int32 + sessionClose func(session uintptr) int32 + newStatement func(session uintptr, out *uintptr) int32 + setSQL func(stmt uintptr, sql string) int32 + execute func(stmt uintptr, out *uintptr) int32 + statementClose func(stmt uintptr) int32 + + // Executed-statement result metadata (control plane). + execQueryID func(executed uintptr) uintptr // returns const char* (0 if none) + execNumRows func(executed uintptr) int64 + execClose func(executed uintptr) int32 + + // Error surface: KernelError is read back through an out-param struct. + getLastError func(out *cKernelError) bool +} + +// cKernelError mirrors the C `KernelError` struct byte-for-byte (64-bit ABI) so +// purego can fill it via an out-pointer. The string fields are C `char*` +// (uintptr here), valid only until the next FFI call on this thread — copied +// out immediately in readLastError, exactly as the cgo lastError does. +// +// Layout matches databricks_kernel.h exactly (offsets are for 64-bit, 8-byte +// pointer alignment): +// +// int32_t code; // 0 +// // 4 (pad: next field is an 8-byte pointer) +// const char* message; // 8 +// const char* sql_state; // 16 +// int32_t vendor_code; // 24 +// uint16_t http_status; // 28 +// bool retryable; // 30 +// // 31 (pad: next field is an 8-byte pointer) +// const char* query_id; // 32 +// // total size 40 +// +// The order here is code, message, sql_state, vendor_code, http_status, +// retryable, query_id — NOT grouped by type. A drift from the header would +// misread the struct; the cgo path guards its enum with compile-time asserts, +// and the real PR would add an unsafe.Sizeof/Offsetof layout check here. +type cKernelError struct { + code int32 + _ [4]byte // pad to 8-align message + message uintptr // const char* + sqlState uintptr // const char* + vendorCode int32 + httpStatus uint16 + retryable bool + _ [1]byte // pad to 8-align queryID + queryID uintptr // const char* +} + +// openDynLib dlopens the kernel shared library and binds the control-plane ABI. +// path is an absolute path to libdatabricks_sql_kernel.{so,dylib,dll}. In a +// real build this would be resolved next to the executable (rpath) or from a +// documented env var; the PoC takes it explicitly. +func openDynLib(path string) (*dynLib, error) { + h, err := purego.Dlopen(path, purego.RTLD_NOW|purego.RTLD_GLOBAL) + if err != nil { + return nil, fmt.Errorf("databricks: kernel dlopen(%q): %w", path, err) + } + l := &dynLib{handle: h} + // RegisterLibFunc panics if a symbol is missing, which is what we want at + // load time — a missing symbol means an ABI/rev mismatch, a hard failure. + purego.RegisterLibFunc(&l.initLogging, h, "kernel_init_logging") + purego.RegisterLibFunc(&l.configNew, h, "kernel_session_config_new") + purego.RegisterLibFunc(&l.configFree, h, "kernel_session_config_free") + purego.RegisterLibFunc(&l.configSetHTTPath, h, "kernel_session_config_set_http_path") + purego.RegisterLibFunc(&l.configSetWH, h, "kernel_session_config_set_warehouse") + purego.RegisterLibFunc(&l.configSetAuthPAT, h, "kernel_session_config_set_auth_pat") + purego.RegisterLibFunc(&l.sessionOpen, h, "kernel_session_open") + purego.RegisterLibFunc(&l.sessionClose, h, "kernel_session_close") + purego.RegisterLibFunc(&l.newStatement, h, "kernel_session_new_statement") + purego.RegisterLibFunc(&l.setSQL, h, "kernel_statement_set_sql") + purego.RegisterLibFunc(&l.execute, h, "kernel_statement_execute") + purego.RegisterLibFunc(&l.statementClose, h, "kernel_statement_close") + purego.RegisterLibFunc(&l.execQueryID, h, "kernel_executed_statement_query_id") + purego.RegisterLibFunc(&l.execNumRows, h, "kernel_executed_statement_num_modified_rows") + purego.RegisterLibFunc(&l.execClose, h, "kernel_executed_statement_close") + purego.RegisterLibFunc(&l.getLastError, h, "kernel_get_last_error") + return l, nil +} + +// callDyn runs a fallible kernel entry point on a pinned OS thread and, on a +// non-Success status, reads the kernel's thread-local last error. This is the +// purego twin of cgo.go's `call`: the LockOSThread pin is what makes the +// separate get_last_error read observe the right thread's buffer. +func (l *dynLib) callDyn(fn func() int32) error { + runtime.LockOSThread() + defer runtime.UnlockOSThread() + st := fn() + if st == kernelStatusSuccess { + return nil + } + return l.readLastError(int(st)) +} + +// readLastError copies the kernel's thread-local last error into a Go +// *KernelError. Must run on the same OS thread as the failing call (callDyn +// guarantees it). String fields are copied out of C memory immediately because +// they are invalidated by the next FFI call. +func (l *dynLib) readLastError(code int) *KernelError { + var e cKernelError + if !l.getLastError(&e) { + return &KernelError{Code: code, Message: fmt.Sprintf("kernel status %d (no detail)", code)} + } + ke := &KernelError{ + Code: int(e.code), + Message: goStringFromC(e.message), + VendorCode: e.vendorCode, + HTTPStatus: e.httpStatus, + Retryable: e.retryable, + SQLState: goStringFromC(e.sqlState), + QueryID: goStringFromC(e.queryID), + } + return ke +} + +// goStringFromC copies a NUL-terminated C string at the given address into a Go +// string. A 0 address yields "". This is the CGO_ENABLED=0 stand-in for +// C.GoString. +// +// It walks bytes off an unsafe.Pointer base with unsafe.Add (the vet-approved +// idiom — arithmetic stays on unsafe.Pointer, never on a bare uintptr, so the +// GC's pointer accounting is never fooled), finds the NUL, then copies out. The +// copy is deliberate: the source bytes live in kernel/C memory and are only +// valid until the next FFI call, so the returned string must not alias them. +func goStringFromC(p uintptr) string { + if p == 0 { + return "" + } + // p is a C address returned across the FFI boundary, not a Go pointer, so + // this uintptr->unsafe.Pointer conversion is the documented-safe FFI case + // (go/analysis flags it heuristically; purego relies on the same pattern). + base := unsafe.Pointer(p) //nolint:govet // FFI C pointer, not GC-managed + + var n int + for *(*byte)(unsafe.Add(base, n)) != 0 { + n++ + } + if n == 0 { + return "" + } + return string(unsafe.Slice((*byte)(base), n)) +} diff --git a/internal/backend/kernel/dynamic_loader_test.go b/internal/backend/kernel/dynamic_loader_test.go new file mode 100644 index 00000000..28c9c84d --- /dev/null +++ b/internal/backend/kernel/dynamic_loader_test.go @@ -0,0 +1,122 @@ +//go:build databricks_kernel_dynamic + +package kernel + +import ( + "os" + "testing" +) + +// TestDynamicLoaderControlPlane proves the whole thesis of the dynamic-loading +// approach: a PURE-GO (CGO_ENABLED=0) binary can drive the closed-source kernel +// end-to-end for the control plane — dlopen the shared library, build a config, +// open a session against a real warehouse, execute a statement, read its server +// query id, and tear everything down — with no cgo, no C compiler, no static +// linking. +// +// It is skipped unless BOTH are set: +// +// DBX_KERNEL_DYLIB absolute path to libdatabricks_sql_kernel.{so,dylib,dll} +// DBX_KERNEL_HOST warehouse hostname (no scheme) +// DBX_KERNEL_HTTPATH /sql/1.0/warehouses/ +// DBX_KERNEL_TOKEN PAT +// +// Run: +// +// DBX_KERNEL_DYLIB=$HOME/Desktop/databricks-sql-kernel/target/release/libdatabricks_sql_kernel.dylib \ +// DBX_KERNEL_HOST=$DATABRICKS_PECOTESTING_SERVER_HOSTNAME \ +// DBX_KERNEL_HTTPATH=/sql/1.0/warehouses/00adc7b6c00429b8 \ +// DBX_KERNEL_TOKEN=$DATABRICKS_PECOTESTING_TOKEN_PERSONAL \ +// CGO_ENABLED=0 go test -tags databricks_kernel_dynamic \ +// -run TestDynamicLoaderControlPlane ./internal/backend/kernel/ -v +func TestDynamicLoaderControlPlane(t *testing.T) { + dylib := os.Getenv("DBX_KERNEL_DYLIB") + host := os.Getenv("DBX_KERNEL_HOST") + httpPath := os.Getenv("DBX_KERNEL_HTTPATH") + token := os.Getenv("DBX_KERNEL_TOKEN") + if dylib == "" || host == "" || httpPath == "" || token == "" { + t.Skip("set DBX_KERNEL_DYLIB, DBX_KERNEL_HOST, DBX_KERNEL_HTTPATH, DBX_KERNEL_TOKEN to run") + } + + l, err := openDynLib(dylib) + if err != nil { + t.Fatalf("openDynLib: %v", err) + } + t.Logf("OK dlopen: bound kernel C ABI from %s (CGO_ENABLED=0)", dylib) + + // Best-effort logging init (benign if the host already installed one). + // file_path = 0 is a real C NULL → kernel logs to stderr. + _ = l.callDyn(func() int32 { return l.initLogging("warn", 0) }) + + // Build the session config. + var cfg uintptr + if err := l.callDyn(func() int32 { return l.configNew(&cfg) }); err != nil { + t.Fatalf("configNew: %v", err) + } + if cfg == 0 { + t.Fatal("configNew returned success but null config") + } + // From here, on any early failure the config must be freed unless it was + // consumed by a successful session_open. + consumed := false + defer func() { + if !consumed { + l.configFree(cfg) + } + }() + + if err := l.callDyn(func() int32 { return l.configSetHTTPath(cfg, host, httpPath) }); err != nil { + t.Fatalf("set_http_path: %v", err) + } + if err := l.callDyn(func() int32 { return l.configSetAuthPAT(cfg, token) }); err != nil { + t.Fatalf("set_auth_pat: %v", err) + } + t.Log("OK config: http_path + PAT set") + + // Open the session (consumes the config on success). + var session uintptr + if err := l.callDyn(func() int32 { return l.sessionOpen(cfg, &session) }); err != nil { + t.Fatalf("session_open: %v", err) + } + consumed = true + if session == 0 { + t.Fatal("session_open returned success but null session") + } + defer func() { + if err := l.callDyn(func() int32 { return l.sessionClose(session) }); err != nil { + t.Errorf("session_close: %v", err) + } + }() + t.Log("OK session_open: connected to warehouse over SEA via the kernel") + + // Prepare + execute a statement. + var stmt uintptr + if err := l.callDyn(func() int32 { return l.newStatement(session, &stmt) }); err != nil { + t.Fatalf("new_statement: %v", err) + } + defer func() { _ = l.callDyn(func() int32 { return l.statementClose(stmt) }) }() + + if err := l.callDyn(func() int32 { return l.setSQL(stmt, "SELECT 1 AS one") }); err != nil { + t.Fatalf("set_sql: %v", err) + } + + var executed uintptr + if err := l.callDyn(func() int32 { return l.execute(stmt, &executed) }); err != nil { + t.Fatalf("execute: %v", err) + } + if executed == 0 { + t.Fatal("execute returned success but null executed-statement") + } + defer func() { _ = l.callDyn(func() int32 { return l.execClose(executed) }) }() + + // Read control-plane result metadata. + queryID := goStringFromC(l.execQueryID(executed)) + rows := l.execNumRows(executed) + t.Logf("OK execute: server queryId=%q numModifiedRows=%d", queryID, rows) + + if queryID == "" { + t.Error("expected a non-empty server query id from the executed statement") + } + + t.Log("PROOF: pure-Go (CGO_ENABLED=0) drove the closed-source kernel control plane end-to-end.") +} From f46eee26bd6bf799954ffe9d5f45910f27918029 Mon Sep 17 00:00:00 2001 From: Madhavendra Rathore Date: Tue, 11 Aug 2026 14:27:57 +0530 Subject: [PATCH 2/3] =?UTF-8?q?poc(kernel):=20complete=20pure-Go=20dynamic?= =?UTF-8?q?=20loader=20=E2=80=94=20data=20plane=20+=20benchmarks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the CGO_ENABLED=0 purego kernel PoC from control-plane-only to a FULL working path, including result-row fetching, verified live on pecotesting with a head-to-head latency benchmark. Data plane (the previously-blocked part): arrow-go v12's cdata importer is cgo, so it can't be used from CGO_ENABLED=0. cdata_pure.go is a pure-Go port of that importer's C-Data *import* path — it reads the flat ArrowSchema/ArrowArray structs via unsafe, invokes each struct's release callback with purego.SyscallN, and builds arrow.Records zero-copy with array.NewData. dynamic_rows.go pulls batches through the dlopen'd kernel_result_stream_* and scans them with the SAME arrowscan scanner the cgo rows.go uses, so values are identical to the cgo backend by construction. Verified live on pecotesting (TestDynamicLoaderDataPlane, CGO_ENABLED=0): scalars, null, decimal-as-exact-string, high-precision DECIMAL(38,2) returning byte-exact (proves buffer import correctness), temporal, binary, nested array/map/struct as JSON, empty result set, and 100k rows across multiple batches. Latency (BenchmarkDynLargeResult vs BenchmarkCgoLargeResult, identical 500k-row query, both kernel artifacts built from the same pinned rev with tls-rustls): cgo static 7.91s/op, purego dynamic 6.17-8.59s/op — end to end dominated by warehouse+network, statistically indistinguishable, no regression. The purego per-cell path crosses no cgo boundary. Test matrix, all green: default pure-Go suite; dynamic-tagged (CGO_ENABLED=0) unit + live e2e; cgo-tagged (CGO_ENABLED=1) unit + full live TestKernelE2E* (0 failures, confirms the existing static path is not regressed). New files (all behind databricks_kernel_dynamic except the cgo bench, which is behind cgo && databricks_kernel): - cdata_pure.go / cdata_pure_finalizer.go: pure-Go C-Data importer - dynamic_rows.go: driver.Rows over the pure-Go importer - dynamic_bench_test.go / cgo_bench.go / cgo_bench_test.go: head-to-head benchmarks Still PoC (unchanged from prior commit): local purego replace for offline builds, dylib path via env, hand-mirrored structs. Documented in DYNAMIC_LOADER_POC.md. Co-authored-by: Isaac --- internal/backend/kernel/DYNAMIC_LOADER_POC.md | 59 +- internal/backend/kernel/cdata_pure.go | 654 ++++++++++++++++++ .../backend/kernel/cdata_pure_finalizer.go | 28 + internal/backend/kernel/cgo_bench.go | 143 ++++ internal/backend/kernel/cgo_bench_test.go | 58 ++ internal/backend/kernel/dynamic_bench_test.go | 120 ++++ internal/backend/kernel/dynamic_loader.go | 16 +- .../backend/kernel/dynamic_loader_test.go | 220 ++++++ internal/backend/kernel/dynamic_rows.go | 146 ++++ 9 files changed, 1440 insertions(+), 4 deletions(-) create mode 100644 internal/backend/kernel/cdata_pure.go create mode 100644 internal/backend/kernel/cdata_pure_finalizer.go create mode 100644 internal/backend/kernel/cgo_bench.go create mode 100644 internal/backend/kernel/cgo_bench_test.go create mode 100644 internal/backend/kernel/dynamic_bench_test.go create mode 100644 internal/backend/kernel/dynamic_rows.go diff --git a/internal/backend/kernel/DYNAMIC_LOADER_POC.md b/internal/backend/kernel/DYNAMIC_LOADER_POC.md index f313dcc1..c634ffdc 100644 --- a/internal/backend/kernel/DYNAMIC_LOADER_POC.md +++ b/internal/backend/kernel/DYNAMIC_LOADER_POC.md @@ -1,9 +1,66 @@ # PoC: dynamic loading of the kernel (pure-Go, CGO_ENABLED=0) -**Status:** working proof-of-concept, control-plane only. Verified live. +**Status:** working proof-of-concept — **control plane AND data plane**, verified +live on pecotesting, with a head-to-head latency benchmark showing no regression. **Build tag:** `databricks_kernel_dynamic` (separate from the shipped `cgo && databricks_kernel` path — the two never build together). +## Update: data plane now works too (pure Go) + +The earlier limitation ("control plane only, because arrow-go's `cdata` importer +is cgo") is **resolved**. `cdata_pure.go` is a pure-Go port of arrow-go v12.0.1's +C-Data *import* path: it reads the flat `ArrowSchema`/`ArrowArray` structs via +`unsafe`, invokes each struct's `release` callback with `purego.SyscallN`, and +builds `arrow.Record`s zero-copy with `array.NewData` — no cgo. `dynRows` +(dynamic_rows.go) pulls batches through the dlopen'd `kernel_result_stream_*` +and scans them with the SAME `arrowscan` scanner the cgo path uses, so values are +identical to the cgo backend by construction. + +Verified live on pecotesting, built `CGO_ENABLED=0` (TestDynamicLoaderDataPlane): + +``` +OK scalars row = [1 2 3.5 hello true 19.99 ] (int/bigint/double/string/bool/decimal/null) +OK fetched 1000 rows in order across batches +OK nested row = [[1,2,3] {"k":1} {"a":1,"b":"x"}] (array/map/struct as JSON) +OK temporal/binary/float row = [2021-07-01 … , [26 191], 3.3, -0.01, + 9999999999999999999999999999.99] (exact high-precision decimal) +OK empty result set drained cleanly +OK fetched 100000 rows (multi-batch, likely CloudFetch) +PROOF: pure-Go (CGO_ENABLED=0) fetched + scanned result rows end-to-end. +``` + +The high-precision decimal returning byte-exact (`…9999.99`, not a float +approximation) is the strongest evidence the buffer import is correct. + +## Latency: no regression (head-to-head on pecotesting) + +Identical 500k-row × 3-col query, same drain-all-rows loop, same arrowscan +scanner; the only difference is static-cgo+cgo-cdata vs dynamic-purego+pure-import +(`BenchmarkCgoLargeResult` vs `BenchmarkDynLargeResult`): + +| Path | ns/op (500k rows) | +| --- | --- | +| cgo static | 7.91 s | +| purego dyn | 6.17 s / 6.61 s / 8.59 s (repeated) | + +End-to-end time is dominated by warehouse execution + network, so the two are +statistically indistinguishable — purego is never meaningfully slower, and the +per-cell path has no cgo boundary crossing. **No latency regression.** + +Both `.a` and `.dylib` were built from the SAME pinned kernel rev +(`KERNEL_REV`, tls-rustls) for a fair comparison. + +## Test matrix (all green) + +- Default pure-Go suite (`CGO_ENABLED=0`, no tags): `go test ./...` — pass. +- Dynamic-tagged (`CGO_ENABLED=0 -tags databricks_kernel_dynamic`): pass; + e2e/data-plane pass live on pecotesting. +- cgo-tagged (`CGO_ENABLED=1 -tags databricks_kernel`): unit pass; full + `TestKernelE2E*` suite passes live on pecotesting (0 failures) — confirms the + change does not regress the existing static path. + +## Original PoC (control plane) + ## What this proves The shipped kernel backend static-links `libdatabricks_sql_kernel.a` through cgo. diff --git a/internal/backend/kernel/cdata_pure.go b/internal/backend/kernel/cdata_pure.go new file mode 100644 index 00000000..88b16fb3 --- /dev/null +++ b/internal/backend/kernel/cdata_pure.go @@ -0,0 +1,654 @@ +//go:build databricks_kernel_dynamic + +// Pure-Go (CGO_ENABLED=0) importer for the Arrow C Data Interface. +// +// This is the data-plane counterpart to dynamic_loader.go. The shipped cgo +// backend imports result batches with arrow-go/v12's `cdata` package, but that +// package is cgo (every non-test file does `import "C"`), so it cannot be used +// from a CGO_ENABLED=0 build. This file reimplements the *import* side of the C +// Data Interface in pure Go: +// +// - The flat C structs (ArrowSchema / ArrowArray) are mirrored as Go structs +// with a byte-for-byte identical 64-bit layout, read via unsafe.Pointer. +// - The `release` callback embedded in each struct (a C function pointer) is +// invoked with purego.SyscallN — the pure-Go equivalent of calling +// ArrowArrayRelease / ArrowSchemaRelease. +// - Buffers are referenced zero-copy (unsafe.Slice over the C pointers), so the +// imported arrow.Record shares the kernel's memory exactly as the cgo path +// does; a runtime finalizer calls the release callback when the Go GC +// reclaims the ArrayData. +// +// The logic is a faithful port of apache/arrow-go v12.0.1 +// (arrow/cdata/cdata.go + interface.go, Apache-2.0), which is the version this +// driver pins — kept close to the original so it can be diffed against upstream. +// Only the import path is ported (the kernel pulls batches via next_batch; the +// ArrowArrayStream reader is not needed). Export is not needed either. +package kernel + +import ( + "errors" + "fmt" + "strconv" + "strings" + "unsafe" + + "github.com/apache/arrow/go/v12/arrow" + "github.com/apache/arrow/go/v12/arrow/array" + "github.com/apache/arrow/go/v12/arrow/bitutil" + "github.com/apache/arrow/go/v12/arrow/memory" + "github.com/ebitengine/purego" +) + +// Arrow C Data Interface flags (arrow/c/abi.h). +const ( + flagDictionaryOrdered = 1 + flagNullable = 2 + flagMapKeysSorted = 4 +) + +// cArrowSchema mirrors `struct ArrowSchema` from the Arrow C Data Interface +// (also declared in databricks_kernel.h). 64-bit layout, all fields 8-byte +// aligned so there is no padding: +// +// const char* format; // 0 +// const char* name; // 8 +// const char* metadata; // 16 +// int64_t flags; // 24 +// int64_t n_children; // 32 +// ArrowSchema** children; // 40 +// ArrowSchema* dictionary; // 48 +// void (*release)(...); // 56 +// void* private_data; // 64 (total 72) +type cArrowSchema struct { + format uintptr + name uintptr + metadata uintptr + flags int64 + nChildren int64 + children uintptr + dictionary uintptr + release uintptr + privData uintptr +} + +// cArrowArray mirrors `struct ArrowArray`. 64-bit layout, all 8-byte aligned: +// +// int64_t length; // 0 +// int64_t null_count; // 8 +// int64_t offset; // 16 +// int64_t n_buffers; // 24 +// int64_t n_children; // 32 +// const void** buffers; // 40 +// ArrowArray** children; // 48 +// ArrowArray* dictionary; // 56 +// void (*release)(...); // 64 +// void* private_data; // 72 (total 80) +type cArrowArray struct { + length int64 + nullCount int64 + offset int64 + nBuffers int64 + nChildren int64 + buffers uintptr + children uintptr + dictionary uintptr + release uintptr + privData uintptr +} + +// releaseCArray invokes the array's C release callback (idempotent: the callback +// nulls its own release pointer; we also clear ours). This is the pure-Go +// ArrowArrayRelease. The callback receives &arr synchronously and does not +// retain it, so passing a Go pointer is safe (arr holds no Go pointers). +func releaseCArray(arr *cArrowArray) { + if arr != nil && arr.release != 0 { + purego.SyscallN(arr.release, uintptr(unsafe.Pointer(arr))) + arr.release = 0 + } +} + +// releaseCSchema is the pure-Go ArrowSchemaRelease. +func releaseCSchema(s *cArrowSchema) { + if s != nil && s.release != 0 { + purego.SyscallN(s.release, uintptr(unsafe.Pointer(s))) + s.release = 0 + } +} + +// arrayMove is the pure-Go ArrowArrayMove: copy the struct and mark the source +// released so only the destination owns the memory. +func arrayMove(src, dst *cArrowArray) { + *dst = *src + src.release = 0 +} + +// formatToSimpleType maps C Data format strings to param-free arrow types. +var formatToSimpleType = map[string]arrow.DataType{ + "n": arrow.Null, "b": arrow.FixedWidthTypes.Boolean, + "c": arrow.PrimitiveTypes.Int8, "C": arrow.PrimitiveTypes.Uint8, + "s": arrow.PrimitiveTypes.Int16, "S": arrow.PrimitiveTypes.Uint16, + "i": arrow.PrimitiveTypes.Int32, "I": arrow.PrimitiveTypes.Uint32, + "l": arrow.PrimitiveTypes.Int64, "L": arrow.PrimitiveTypes.Uint64, + "e": arrow.FixedWidthTypes.Float16, "f": arrow.PrimitiveTypes.Float32, + "g": arrow.PrimitiveTypes.Float64, "z": arrow.BinaryTypes.Binary, + "Z": arrow.BinaryTypes.LargeBinary, "u": arrow.BinaryTypes.String, + "U": arrow.BinaryTypes.LargeString, "tdD": arrow.FixedWidthTypes.Date32, + "tdm": arrow.FixedWidthTypes.Date64, "tts": arrow.FixedWidthTypes.Time32s, + "ttm": arrow.FixedWidthTypes.Time32ms, "ttu": arrow.FixedWidthTypes.Time64us, + "ttn": arrow.FixedWidthTypes.Time64ns, "tDs": arrow.FixedWidthTypes.Duration_s, + "tDm": arrow.FixedWidthTypes.Duration_ms, "tDu": arrow.FixedWidthTypes.Duration_us, + "tDn": arrow.FixedWidthTypes.Duration_ns, "tiM": arrow.FixedWidthTypes.MonthInterval, + "tiD": arrow.FixedWidthTypes.DayTimeInterval, "tin": arrow.FixedWidthTypes.MonthDayNanoInterval, +} + +// decodeCMetadata decodes C Data metadata (int32-length-prefixed key/value +// pairs). Faithful port of the cgo version. +func decodeCMetadata(md uintptr) arrow.Metadata { + if md == 0 { + return arrow.Metadata{} + } + pos := md + readint32 := func() int32 { + v := *(*int32)(unsafe.Pointer(pos)) + pos += 4 + return v + } + readstr := func() string { + l := readint32() + s := string(unsafe.Slice((*byte)(unsafe.Pointer(pos)), l)) + pos += uintptr(l) + return s + } + npairs := readint32() + if npairs == 0 { + return arrow.Metadata{} + } + keys := make([]string, npairs) + vals := make([]string, npairs) + for i := int32(0); i < npairs; i++ { + keys[i] = readstr() + vals[i] = readstr() + } + return arrow.NewMetadata(keys, vals) +} + +func schemaChildrenSlice(s *cArrowSchema) []*cArrowSchema { + if s.nChildren == 0 || s.children == 0 { + return nil + } + ptrs := unsafe.Slice((*uintptr)(unsafe.Pointer(s.children)), s.nChildren) + out := make([]*cArrowSchema, len(ptrs)) + for i, p := range ptrs { + out[i] = (*cArrowSchema)(unsafe.Pointer(p)) + } + return out +} + +// importSchema converts a cArrowSchema to an arrow.Field, always releasing the +// schema (even on error), matching the cgo semantics. +func importSchema(schema *cArrowSchema) (ret arrow.Field, err error) { + defer releaseCSchema(schema) + + var childFields []arrow.Field + if schema.nChildren > 0 { + kids := schemaChildrenSlice(schema) + childFields = make([]arrow.Field, len(kids)) + for i, c := range kids { + childFields[i], err = importSchema(c) + if err != nil { + return + } + } + } + + ret.Name = goStringFromC(schema.name) + ret.Nullable = (schema.flags & flagNullable) != 0 + ret.Metadata = decodeCMetadata(schema.metadata) + + f := goStringFromC(schema.format) + if dt, ok := formatToSimpleType[f]; ok { + ret.Type = dt + if schema.dictionary != 0 { + valueField, e := importSchema((*cArrowSchema)(unsafe.Pointer(schema.dictionary))) + if e != nil { + return ret, e + } + ret.Type = &arrow.DictionaryType{ + IndexType: ret.Type, + ValueType: valueField.Type, + Ordered: (*cArrowSchema)(unsafe.Pointer(schema.dictionary)).flags&flagDictionaryOrdered != 0, + } + } + return + } + + var dt arrow.DataType + typs := strings.Split(f, ":") + const defaulttz = "UTC" + switch typs[0] { + case "tss": + tz := typs[1] + if len(typs[1]) == 0 { + tz = defaulttz + } + dt = &arrow.TimestampType{Unit: arrow.Second, TimeZone: tz} + case "tsm": + tz := typs[1] + if len(typs[1]) == 0 { + tz = defaulttz + } + dt = &arrow.TimestampType{Unit: arrow.Millisecond, TimeZone: tz} + case "tsu": + tz := typs[1] + if len(typs[1]) == 0 { + tz = defaulttz + } + dt = &arrow.TimestampType{Unit: arrow.Microsecond, TimeZone: tz} + case "tsn": + tz := typs[1] + if len(typs[1]) == 0 { + tz = defaulttz + } + dt = &arrow.TimestampType{Unit: arrow.Nanosecond, TimeZone: tz} + case "w": + byteWidth, e := strconv.Atoi(typs[1]) + if e != nil { + return ret, e + } + dt = &arrow.FixedSizeBinaryType{ByteWidth: byteWidth} + case "d": + propList := strings.Split(typs[1], ",") + if len(propList) == 3 { + return ret, errors.New("only decimal128 is supported") + } + precision, _ := strconv.Atoi(propList[0]) + scale, _ := strconv.Atoi(propList[1]) + dt = &arrow.Decimal128Type{Precision: int32(precision), Scale: int32(scale)} + } + + if f[0] == '+' { // nested types + switch f[1] { + case 'l': + dt = arrow.ListOfField(childFields[0]) + case 'L': + dt = arrow.LargeListOfField(childFields[0]) + case 'w': + listSize, e := strconv.Atoi(strings.Split(f, ":")[1]) + if e != nil { + return ret, e + } + dt = arrow.FixedSizeListOfField(int32(listSize), childFields[0]) + case 's': + dt = arrow.StructOf(childFields...) + case 'm': + st := childFields[0].Type.(*arrow.StructType) + dt = arrow.MapOf(st.Field(0).Type, st.Field(1).Type) + dt.(*arrow.MapType).KeysSorted = (schema.flags & flagMapKeysSorted) != 0 + case 'u': + var mode arrow.UnionMode + switch f[2] { + case 'd': + mode = arrow.DenseMode + case 's': + mode = arrow.SparseMode + default: + return ret, fmt.Errorf("%w: invalid union type", arrow.ErrInvalid) + } + codes := strings.Split(strings.Split(f, ":")[1], ",") + typeCodes := make([]arrow.UnionTypeCode, 0, len(codes)) + for _, i := range codes { + v, e := strconv.ParseInt(i, 10, 8) + if e != nil { + return ret, fmt.Errorf("%w: invalid type code: %s", arrow.ErrInvalid, e) + } + if v < 0 { + return ret, fmt.Errorf("%w: negative type code in union: %s", arrow.ErrInvalid, f) + } + typeCodes = append(typeCodes, arrow.UnionTypeCode(v)) + } + if len(childFields) != len(typeCodes) { + return ret, fmt.Errorf("%w: children incompatible with format string", arrow.ErrInvalid) + } + dt = arrow.UnionOf(mode, childFields, typeCodes) + } + } + + if dt == nil { + err = errors.New("unimplemented type: " + f) + } else { + ret.Type = dt + } + return +} + +// cimporter tracks state while importing a cArrowArray tree. +type cimporter struct { + dt arrow.DataType + arr *cArrowArray + data arrow.ArrayData + parent *cimporter + children []cimporter + cbuffers []uintptr +} + +func (imp *cimporter) importChild(parent *cimporter, src *cArrowArray) error { + imp.parent = parent + return imp.doImport(src) +} + +func (imp *cimporter) arrayChildrenSlice() []*cArrowArray { + if imp.arr.nChildren == 0 || imp.arr.children == 0 { + return nil + } + ptrs := unsafe.Slice((*uintptr)(unsafe.Pointer(imp.arr.children)), imp.arr.nChildren) + out := make([]*cArrowArray, len(ptrs)) + for i, p := range ptrs { + out[i] = (*cArrowArray)(unsafe.Pointer(p)) + } + return out +} + +func (imp *cimporter) doImportChildren() error { + children := imp.arrayChildrenSlice() + if len(children) > 0 { + imp.children = make([]cimporter, len(children)) + } + switch imp.dt.ID() { + case arrow.LIST: + imp.children[0].dt = imp.dt.(*arrow.ListType).Elem() + if err := imp.children[0].importChild(imp, children[0]); err != nil { + return err + } + case arrow.LARGE_LIST: + imp.children[0].dt = imp.dt.(*arrow.LargeListType).Elem() + if err := imp.children[0].importChild(imp, children[0]); err != nil { + return err + } + case arrow.FIXED_SIZE_LIST: + imp.children[0].dt = imp.dt.(*arrow.FixedSizeListType).Elem() + if err := imp.children[0].importChild(imp, children[0]); err != nil { + return err + } + case arrow.STRUCT: + st := imp.dt.(*arrow.StructType) + for i, c := range children { + imp.children[i].dt = st.Field(i).Type + if err := imp.children[i].importChild(imp, c); err != nil { + return err + } + } + case arrow.MAP: + imp.children[0].dt = imp.dt.(*arrow.MapType).ValueType() + if err := imp.children[0].importChild(imp, children[0]); err != nil { + return err + } + case arrow.DENSE_UNION: + dt := imp.dt.(*arrow.DenseUnionType) + for i, c := range children { + imp.children[i].dt = dt.Fields()[i].Type + if err := imp.children[i].importChild(imp, c); err != nil { + return err + } + } + case arrow.SPARSE_UNION: + dt := imp.dt.(*arrow.SparseUnionType) + for i, c := range children { + imp.children[i].dt = dt.Fields()[i].Type + if err := imp.children[i].importChild(imp, c); err != nil { + return err + } + } + } + return nil +} + +func (imp *cimporter) doImport(src *cArrowArray) error { + imp.arr = new(cArrowArray) + // Move src into our heap struct so a finalizer on the resulting ArrayData + // releases the C memory when the GC reclaims it (mirrors the cgo path). + arrayMove(src, imp.arr) + movedArr := imp.arr + defer func() { + if imp.data != nil { + // Finalizer safety net: release the kernel buffers when the Go GC + // reclaims the ArrayData. Release is idempotent, so an explicit + // Release() elsewhere never double-frees. (arrow-go itself sets a + // finalizer on the ArrayData too; this one frees the C-side array.) + setArrayDataFinalizer(imp.data, movedArr) + } + }() + + if err := imp.doImportChildren(); err != nil { + return err + } + + if imp.arr.nBuffers > 0 { + imp.cbuffers = unsafe.Slice((*uintptr)(unsafe.Pointer(imp.arr.buffers)), imp.arr.nBuffers) + } + + switch dt := imp.dt.(type) { + case *arrow.NullType: + if err := imp.checkNoChildren(); err != nil { + return err + } + imp.data = array.NewData(dt, int(imp.arr.length), nil, nil, int(imp.arr.nullCount), int(imp.arr.offset)) + case arrow.FixedWidthDataType: + return imp.importFixedSizePrimitive() + case *arrow.StringType: + return imp.importStringLike(int64(arrow.Int32SizeBytes)) + case *arrow.BinaryType: + return imp.importStringLike(int64(arrow.Int32SizeBytes)) + case *arrow.LargeStringType: + return imp.importStringLike(int64(arrow.Int64SizeBytes)) + case *arrow.LargeBinaryType: + return imp.importStringLike(int64(arrow.Int64SizeBytes)) + case *arrow.ListType: + return imp.importListLike() + case *arrow.LargeListType: + return imp.importListLike() + case *arrow.MapType: + return imp.importListLike() + case *arrow.FixedSizeListType: + if err := imp.checkNumChildren(1); err != nil { + return err + } + if err := imp.checkNumBuffers(1); err != nil { + return err + } + nulls, err := imp.importNullBitmap(0) + if err != nil { + return err + } + imp.data = array.NewData(dt, int(imp.arr.length), []*memory.Buffer{nulls}, []arrow.ArrayData{imp.children[0].data}, int(imp.arr.nullCount), int(imp.arr.offset)) + case *arrow.StructType: + if err := imp.checkNumBuffers(1); err != nil { + return err + } + nulls, err := imp.importNullBitmap(0) + if err != nil { + return err + } + children := make([]arrow.ArrayData, len(imp.children)) + for i := range imp.children { + children[i] = imp.children[i].data + } + imp.data = array.NewData(dt, int(imp.arr.length), []*memory.Buffer{nulls}, children, int(imp.arr.nullCount), int(imp.arr.offset)) + default: + return fmt.Errorf("unimplemented type %s", dt) + } + return nil +} + +func (imp *cimporter) importStringLike(offsetByteWidth int64) (err error) { + if err = imp.checkNoChildren(); err != nil { + return + } + if err = imp.checkNumBuffers(3); err != nil { + return + } + var nulls, offsets, values *memory.Buffer + if nulls, err = imp.importNullBitmap(0); err != nil { + return + } + if offsets, err = imp.importOffsetsBuffer(1, offsetByteWidth); err != nil { + return + } + var nvals int64 + switch offsetByteWidth { + case 4: + typedOffsets := arrow.Int32Traits.CastFromBytes(offsets.Bytes()) + nvals = int64(typedOffsets[imp.arr.offset+imp.arr.length]) + case 8: + typedOffsets := arrow.Int64Traits.CastFromBytes(offsets.Bytes()) + nvals = typedOffsets[imp.arr.offset+imp.arr.length] + } + if values, err = imp.importVariableValuesBuffer(2, 1, nvals); err != nil { + return + } + imp.data = array.NewData(imp.dt, int(imp.arr.length), []*memory.Buffer{nulls, offsets, values}, nil, int(imp.arr.nullCount), int(imp.arr.offset)) + return +} + +func (imp *cimporter) importListLike() (err error) { + if err = imp.checkNumChildren(1); err != nil { + return + } + if err = imp.checkNumBuffers(2); err != nil { + return + } + var nulls, offsets *memory.Buffer + if nulls, err = imp.importNullBitmap(0); err != nil { + return + } + offsetSize := imp.dt.Layout().Buffers[1].ByteWidth + if offsets, err = imp.importOffsetsBuffer(1, int64(offsetSize)); err != nil { + return + } + imp.data = array.NewData(imp.dt, int(imp.arr.length), []*memory.Buffer{nulls, offsets}, []arrow.ArrayData{imp.children[0].data}, int(imp.arr.nullCount), int(imp.arr.offset)) + return +} + +func (imp *cimporter) importFixedSizePrimitive() error { + if err := imp.checkNoChildren(); err != nil { + return err + } + if err := imp.checkNumBuffers(2); err != nil { + return err + } + nulls, err := imp.importNullBitmap(0) + if err != nil { + return err + } + var values *memory.Buffer + fw := imp.dt.(arrow.FixedWidthDataType) + if bitutil.IsMultipleOf8(int64(fw.BitWidth())) { + values, err = imp.importFixedSizeBuffer(1, bitutil.BytesForBits(int64(fw.BitWidth()))) + } else { + if fw.BitWidth() != 1 { + return errors.New("invalid bitwidth") + } + values, err = imp.importBitsBuffer(1) + } + if err != nil { + return err + } + var dict *array.Data + if dt, ok := imp.dt.(*arrow.DictionaryType); ok { + dictImp := &cimporter{dt: dt.ValueType} + if err := dictImp.doImport((*cArrowArray)(unsafe.Pointer(imp.arr.dictionary))); err != nil { + return err + } + defer dictImp.data.Release() + dict = dictImp.data.(*array.Data) + } + imp.data = array.NewDataWithDictionary(imp.dt, int(imp.arr.length), []*memory.Buffer{nulls, values}, int(imp.arr.nullCount), int(imp.arr.offset), dict) + return nil +} + +func (imp *cimporter) checkNoChildren() error { return imp.checkNumChildren(0) } + +func (imp *cimporter) checkNumChildren(n int64) error { + if imp.arr.nChildren != n { + return fmt.Errorf("expected %d children for imported type %s, ArrowArray has %d", n, imp.dt, imp.arr.nChildren) + } + return nil +} + +func (imp *cimporter) checkNumBuffers(n int64) error { + if imp.arr.nBuffers != n { + return fmt.Errorf("expected %d buffers for imported type %s, ArrowArray has %d", n, imp.dt, imp.arr.nBuffers) + } + return nil +} + +func (imp *cimporter) importBuffer(bufferID int, sz int64) (*memory.Buffer, error) { + if imp.cbuffers[bufferID] == 0 { + if sz != 0 { + return nil, errors.New("invalid buffer") + } + return memory.NewBufferBytes([]byte{}), nil + } + data := unsafe.Slice((*byte)(unsafe.Pointer(imp.cbuffers[bufferID])), sz) + return memory.NewBufferBytes(data), nil +} + +func (imp *cimporter) importBitsBuffer(bufferID int) (*memory.Buffer, error) { + bufsize := bitutil.BytesForBits(imp.arr.length + imp.arr.offset) + return imp.importBuffer(bufferID, bufsize) +} + +func (imp *cimporter) importNullBitmap(bufferID int) (*memory.Buffer, error) { + if imp.arr.nullCount > 0 && imp.cbuffers[bufferID] == 0 { + return nil, fmt.Errorf("ArrowArray has null bitmap buffer but non-zero null_count %d", imp.arr.nullCount) + } + if imp.arr.nullCount == 0 && imp.cbuffers[bufferID] == 0 { + return nil, nil + } + return imp.importBitsBuffer(bufferID) +} + +func (imp *cimporter) importFixedSizeBuffer(bufferID int, byteWidth int64) (*memory.Buffer, error) { + return imp.importBuffer(bufferID, byteWidth*(imp.arr.length+imp.arr.offset)) +} + +func (imp *cimporter) importOffsetsBuffer(bufferID int, offsetsize int64) (*memory.Buffer, error) { + return imp.importBuffer(bufferID, offsetsize*(imp.arr.length+imp.arr.offset+1)) +} + +func (imp *cimporter) importVariableValuesBuffer(bufferID int, byteWidth, nvals int64) (*memory.Buffer, error) { + return imp.importBuffer(bufferID, byteWidth*nvals) +} + +func importCArrayAsType(arr *cArrowArray, dt arrow.DataType) (*cimporter, error) { + imp := &cimporter{dt: dt} + err := imp.doImport(arr) + return imp, err +} + +// importCArrowSchema imports a record-batch schema (top level must be a struct). +func importCArrowSchema(out *cArrowSchema) (*arrow.Schema, error) { + ret, err := importSchema(out) + if err != nil { + return nil, err + } + st, ok := ret.Type.(*arrow.StructType) + if !ok { + return nil, errors.New("recordbatch schema import must be of struct type") + } + return arrow.NewSchema(st.Fields(), &ret.Metadata), nil +} + +// importCRecordBatchWithSchema imports an array as a record batch, schema known. +func importCRecordBatchWithSchema(arr *cArrowArray, sc *arrow.Schema) (arrow.Record, error) { + imp, err := importCArrayAsType(arr, arrow.StructOf(sc.Fields()...)) + if err != nil { + return nil, err + } + st := array.NewStructData(imp.data) + defer st.Release() + cols := make([]arrow.Array, st.NumField()) + for i := 0; i < st.NumField(); i++ { + cols[i] = st.Field(i) + } + return array.NewRecord(sc, cols, int64(st.Len())), nil +} diff --git a/internal/backend/kernel/cdata_pure_finalizer.go b/internal/backend/kernel/cdata_pure_finalizer.go new file mode 100644 index 00000000..bb9a51bb --- /dev/null +++ b/internal/backend/kernel/cdata_pure_finalizer.go @@ -0,0 +1,28 @@ +//go:build databricks_kernel_dynamic + +package kernel + +import ( + "runtime" + + "github.com/apache/arrow/go/v12/arrow" +) + +// setArrayDataFinalizer attaches a finalizer to the imported ArrayData that +// releases the C-side ArrowArray (via its release callback) when the Go GC +// reclaims the data. This is the pure-Go analogue of the cgo importer's +// runtime.SetFinalizer(imp.data, ...) that calls ArrowArrayRelease + free. +// +// releaseCArray is idempotent (it clears the release pointer after firing), so +// this backstop never double-frees if the buffers were already released. The +// kernel exports self-contained batches, so releasing here touches nothing +// session-scoped — safe even if the finalizer runs after the session closes. +// +// runtime.SetFinalizer requires a pointer to an object the GC tracks; arrow-go +// implements ArrayData as *array.Data, so the interface value is a pointer and +// SetFinalizer accepts it directly. +func setArrayDataFinalizer(data arrow.ArrayData, arr *cArrowArray) { + runtime.SetFinalizer(data, func(arrow.ArrayData) { + releaseCArray(arr) + }) +} diff --git a/internal/backend/kernel/cgo_bench.go b/internal/backend/kernel/cgo_bench.go new file mode 100644 index 00000000..266a6513 --- /dev/null +++ b/internal/backend/kernel/cgo_bench.go @@ -0,0 +1,143 @@ +//go:build cgo && databricks_kernel + +// Benchmark helper for the cgo static-link path, used by cgo_bench_test.go to +// compare against the purego dynamic path (dynamic_bench_test.go). It lives in a +// non-test file because cgo (`import "C"`) is not supported directly in _test.go +// files. It is only referenced from benchmarks, so it adds nothing to a normal +// build beyond the already-tagged kernel package. +package kernel + +/* +#include +#include "databricks_kernel.h" +struct ArrowSchema; +struct ArrowArray; +*/ +import "C" + +import ( + "database/sql/driver" + "fmt" + "time" + "unsafe" + + "github.com/apache/arrow/go/v12/arrow" + "github.com/apache/arrow/go/v12/arrow/cdata" + "github.com/databricks/databricks-sql-go/internal/arrowscan" +) + +// CgoBenchSession is an open kernel session over the cgo static-link path. +type CgoBenchSession struct { + session *C.kernel_session_t +} + +// CgoBenchOpen opens a session via the cgo path (PAT auth over http path). +func CgoBenchOpen(host, httpPath, token string) (*CgoBenchSession, error) { + initKernelLogging() + var cfg *C.KernelSessionConfig + if err := call(func() C.KernelStatusCode { return C.kernel_session_config_new(&cfg) }); err != nil { + return nil, err + } + ch := newCStr(host) + defer ch.free() + cp := newCStr(httpPath) + defer cp.free() + if err := call(func() C.KernelStatusCode { return C.kernel_session_config_set_http_path(cfg, ch.c, cp.c) }); err != nil { + return nil, err + } + ct := newCStr(token) + defer ct.free() + if err := call(func() C.KernelStatusCode { return C.kernel_session_config_set_auth_pat(cfg, ct.c) }); err != nil { + return nil, err + } + var sess *C.kernel_session_t + if err := call(func() C.KernelStatusCode { return C.kernel_session_open(cfg, &sess) }); err != nil { + return nil, err + } + return &CgoBenchSession{session: sess}, nil +} + +func (s *CgoBenchSession) Close() { + if s.session != nil { + _ = call(func() C.KernelStatusCode { return C.kernel_session_close(s.session) }) + s.session = nil + } +} + +// Drain runs sql and scans every row through the same arrowscan scanner the +// dynamic path uses, returning the row count. Mirrors dynSession.drain. +func (s *CgoBenchSession) Drain(sql string) (int, error) { + var stmt *C.kernel_statement_t + if err := call(func() C.KernelStatusCode { return C.kernel_session_new_statement(s.session, &stmt) }); err != nil { + return 0, err + } + defer C.kernel_statement_close(stmt) + cs := newCStr(sql) + defer cs.free() + if err := call(func() C.KernelStatusCode { return C.kernel_statement_set_sql(stmt, cs.c) }); err != nil { + return 0, err + } + var exec *C.kernel_executed_statement_t + if err := call(func() C.KernelStatusCode { return C.kernel_statement_execute(stmt, &exec) }); err != nil { + return 0, err + } + defer C.kernel_executed_statement_close(exec) + var stream *C.kernel_result_stream_t + if err := call(func() C.KernelStatusCode { return C.kernel_executed_statement_get_result_stream(exec, &stream) }); err != nil { + return 0, err + } + defer C.kernel_result_stream_close(stream) + + var csch C.struct_ArrowSchema + if err := call(func() C.KernelStatusCode { return C.kernel_result_stream_get_schema(stream, &csch) }); err != nil { + return 0, err + } + sch, err := cdata.ImportCArrowSchema((*cdata.CArrowSchema)(unsafe.Pointer(&csch))) + if err != nil { + return 0, err + } + keyCache := arrowscan.NewStructKeyCache() + ncols := len(sch.Fields()) + dest := make([]driver.Value, ncols) + + var cur arrow.Record + rowInCur, n := 0, 0 + for { + for cur == nil || rowInCur >= int(cur.NumRows()) { + if cur != nil { + cur.Release() + cur = nil + } + var carr C.struct_ArrowArray + var cs2 C.struct_ArrowSchema + if err := call(func() C.KernelStatusCode { + return C.kernel_result_stream_next_batch(stream, &carr, &cs2) + }); err != nil { + return 0, err + } + if carr.release == nil { + if cur != nil { + cur.Release() + } + return n, nil + } + rec, err := cdata.ImportCRecordBatch( + (*cdata.CArrowArray)(unsafe.Pointer(&carr)), + (*cdata.CArrowSchema)(unsafe.Pointer(&cs2))) + if err != nil { + return 0, err + } + cur = rec + rowInCur = 0 + keyCache.Reset() + } + for c := 0; c < ncols; c++ { + if _, err := arrowscan.ScanCellCached(cur.Column(c), rowInCur, time.UTC, keyCache); err != nil { + return 0, fmt.Errorf("scan col %d: %w", c, err) + } + _ = dest + } + rowInCur++ + n++ + } +} diff --git a/internal/backend/kernel/cgo_bench_test.go b/internal/backend/kernel/cgo_bench_test.go new file mode 100644 index 00000000..e6d4c9d4 --- /dev/null +++ b/internal/backend/kernel/cgo_bench_test.go @@ -0,0 +1,58 @@ +//go:build cgo && databricks_kernel + +package kernel + +import ( + "os" + "testing" +) + +// See cgo_bench.go for the drain helper. These benchmarks mirror +// BenchmarkDyn* (purego dynamic path) so ns/op is directly comparable. +// +// DBX_KERNEL_HOST=... DBX_KERNEL_HTTPATH=... DBX_KERNEL_TOKEN=... \ +// CGO_ENABLED=1 go test -tags databricks_kernel -run x \ +// -bench 'BenchmarkCgo' -benchtime 20x ./internal/backend/kernel/ +func cgoBenchEnv(b *testing.B) (host, httpPath, token string) { + host = os.Getenv("DBX_KERNEL_HOST") + httpPath = os.Getenv("DBX_KERNEL_HTTPATH") + token = os.Getenv("DBX_KERNEL_TOKEN") + if host == "" || httpPath == "" || token == "" { + b.Skip("set DBX_KERNEL_HOST, DBX_KERNEL_HTTPATH, DBX_KERNEL_TOKEN") + } + return +} + +func BenchmarkCgoLowLatency(b *testing.B) { + host, httpPath, token := cgoBenchEnv(b) + s, err := CgoBenchOpen(host, httpPath, token) + if err != nil { + b.Fatalf("open: %v", err) + } + defer s.Close() + b.ResetTimer() + for i := 0; i < b.N; i++ { + if _, err := s.Drain("SELECT 1 AS one"); err != nil { + b.Fatal(err) + } + } +} + +func BenchmarkCgoLargeResult(b *testing.B) { + host, httpPath, token := cgoBenchEnv(b) + s, err := CgoBenchOpen(host, httpPath, token) + if err != nil { + b.Fatalf("open: %v", err) + } + defer s.Close() + b.ResetTimer() + for i := 0; i < b.N; i++ { + n, err := s.Drain("SELECT id, id*2 AS doubled, CAST(id AS STRING) AS s FROM range(0, 500000)") + if err != nil { + b.Fatal(err) + } + if n != 500000 { + b.Fatalf("got %d rows", n) + } + } +} diff --git a/internal/backend/kernel/dynamic_bench_test.go b/internal/backend/kernel/dynamic_bench_test.go new file mode 100644 index 00000000..69541052 --- /dev/null +++ b/internal/backend/kernel/dynamic_bench_test.go @@ -0,0 +1,120 @@ +//go:build databricks_kernel_dynamic + +package kernel + +import ( + "database/sql/driver" + "io" + "os" + "testing" + "time" +) + +// BenchmarkDynLowLatency and BenchmarkDynLargeResult measure the pure-Go +// (purego dynamic) path so its numbers can be compared head-to-head with the +// cgo static path (BenchmarkCgo* in cgo_bench_test.go, built with the +// databricks_kernel tag). Both benchmarks drive the identical query through the +// same loader-level path, isolating the FFI + Arrow-import cost that differs +// between the two linking models. +// +// Run: +// +// DBX_KERNEL_DYLIB=... DBX_KERNEL_HOST=... DBX_KERNEL_HTTPATH=... DBX_KERNEL_TOKEN=... \ +// CGO_ENABLED=0 go test -tags databricks_kernel_dynamic -run x \ +// -bench 'BenchmarkDyn' -benchtime 20x ./internal/backend/kernel/ +func benchEnv(b *testing.B) (dylib, host, httpPath, token string) { + dylib = os.Getenv("DBX_KERNEL_DYLIB") + host = os.Getenv("DBX_KERNEL_HOST") + httpPath = os.Getenv("DBX_KERNEL_HTTPATH") + token = os.Getenv("DBX_KERNEL_TOKEN") + if dylib == "" || host == "" || httpPath == "" || token == "" { + b.Skip("set DBX_KERNEL_DYLIB, DBX_KERNEL_HOST, DBX_KERNEL_HTTPATH, DBX_KERNEL_TOKEN") + } + return +} + +func benchOpen(b *testing.B, dylib, host, httpPath, token string) *dynSession { + b.Helper() + l, err := openDynLib(dylib) + if err != nil { + b.Fatalf("openDynLib: %v", err) + } + _ = l.callDyn(func() int32 { return l.initLogging("warn", 0) }) + var cfg uintptr + if err := l.callDyn(func() int32 { return l.configNew(&cfg) }); err != nil { + b.Fatalf("configNew: %v", err) + } + if err := l.callDyn(func() int32 { return l.configSetHTTPath(cfg, host, httpPath) }); err != nil { + b.Fatalf("set_http_path: %v", err) + } + if err := l.callDyn(func() int32 { return l.configSetAuthPAT(cfg, token) }); err != nil { + b.Fatalf("set_auth_pat: %v", err) + } + var session uintptr + if err := l.callDyn(func() int32 { return l.sessionOpen(cfg, &session) }); err != nil { + b.Fatalf("session_open: %v", err) + } + return &dynSession{l: l, session: session} +} + +// run one query and drain all rows through the pure-Go data plane. +func (s *dynSession) drain(b *testing.B, sql string) int { + l := s.l + var stmt uintptr + if err := l.callDyn(func() int32 { return l.newStatement(s.session, &stmt) }); err != nil { + b.Fatalf("new_statement: %v", err) + } + defer func() { _ = l.callDyn(func() int32 { return l.statementClose(stmt) }) }() + if err := l.callDyn(func() int32 { return l.setSQL(stmt, sql) }); err != nil { + b.Fatalf("set_sql: %v", err) + } + var executed uintptr + if err := l.callDyn(func() int32 { return l.execute(stmt, &executed) }); err != nil { + b.Fatalf("execute: %v", err) + } + defer func() { _ = l.callDyn(func() int32 { return l.execClose(executed) }) }() + var stream uintptr + if err := l.callDyn(func() int32 { return l.getResultStream(executed, &stream) }); err != nil { + b.Fatalf("get_result_stream: %v", err) + } + rows, err := newDynRows(l, stream, time.UTC) + if err != nil { + b.Fatalf("newDynRows: %v", err) + } + defer rows.Close() + ncols := len(rows.Columns()) + dest := make([]driver.Value, ncols) + n := 0 + for { + if err := rows.Next(dest); err == io.EOF { + break + } else if err != nil { + b.Fatalf("Next: %v", err) + } + n++ + } + return n +} + +func BenchmarkDynLowLatency(b *testing.B) { + dylib, host, httpPath, token := benchEnv(b) + s := benchOpen(b, dylib, host, httpPath, token) + defer s.close() + b.ResetTimer() + for i := 0; i < b.N; i++ { + s.drain(b, "SELECT 1 AS one") + } +} + +func BenchmarkDynLargeResult(b *testing.B) { + dylib, host, httpPath, token := benchEnv(b) + s := benchOpen(b, dylib, host, httpPath, token) + defer s.close() + b.ResetTimer() + for i := 0; i < b.N; i++ { + n := s.drain(b, "SELECT id, id*2 AS doubled, CAST(id AS STRING) AS s FROM range(0, 500000)") + if n != 500000 { + b.Fatalf("got %d rows", n) + } + } +} diff --git a/internal/backend/kernel/dynamic_loader.go b/internal/backend/kernel/dynamic_loader.go index 253456a2..540ae6f2 100644 --- a/internal/backend/kernel/dynamic_loader.go +++ b/internal/backend/kernel/dynamic_loader.go @@ -84,9 +84,15 @@ type dynLib struct { statementClose func(stmt uintptr) int32 // Executed-statement result metadata (control plane). - execQueryID func(executed uintptr) uintptr // returns const char* (0 if none) - execNumRows func(executed uintptr) int64 - execClose func(executed uintptr) int32 + execQueryID func(executed uintptr) uintptr // returns const char* (0 if none) + execNumRows func(executed uintptr) int64 + execClose func(executed uintptr) int32 + + // Result stream (data plane): pull Arrow C-Data batches. + getResultStream func(executed uintptr, out *uintptr) int32 + streamGetSchema func(stream uintptr, out *cArrowSchema) int32 + streamNextBatch func(stream uintptr, outArray *cArrowArray, outSchema *cArrowSchema) int32 + streamClose func(stream uintptr) int32 // Error surface: KernelError is read back through an out-param struct. getLastError func(out *cKernelError) bool @@ -154,6 +160,10 @@ func openDynLib(path string) (*dynLib, error) { purego.RegisterLibFunc(&l.execQueryID, h, "kernel_executed_statement_query_id") purego.RegisterLibFunc(&l.execNumRows, h, "kernel_executed_statement_num_modified_rows") purego.RegisterLibFunc(&l.execClose, h, "kernel_executed_statement_close") + purego.RegisterLibFunc(&l.getResultStream, h, "kernel_executed_statement_get_result_stream") + purego.RegisterLibFunc(&l.streamGetSchema, h, "kernel_result_stream_get_schema") + purego.RegisterLibFunc(&l.streamNextBatch, h, "kernel_result_stream_next_batch") + purego.RegisterLibFunc(&l.streamClose, h, "kernel_result_stream_close") purego.RegisterLibFunc(&l.getLastError, h, "kernel_get_last_error") return l, nil } diff --git a/internal/backend/kernel/dynamic_loader_test.go b/internal/backend/kernel/dynamic_loader_test.go index 28c9c84d..fe56b5ef 100644 --- a/internal/backend/kernel/dynamic_loader_test.go +++ b/internal/backend/kernel/dynamic_loader_test.go @@ -3,10 +3,105 @@ package kernel import ( + "database/sql/driver" + "fmt" + "io" "os" "testing" + "time" ) +// dynSession is a tiny test harness around the pure-Go loader: it holds an open +// kernel session and runs queries through the pure-Go data plane (dynRows). +type dynSession struct { + t *testing.T + l *dynLib + session uintptr +} + +func openDynSession(t *testing.T, dylib, host, httpPath, token string) *dynSession { + t.Helper() + l, err := openDynLib(dylib) + if err != nil { + t.Fatalf("openDynLib: %v", err) + } + _ = l.callDyn(func() int32 { return l.initLogging("warn", 0) }) + + var cfg uintptr + if err := l.callDyn(func() int32 { return l.configNew(&cfg) }); err != nil { + t.Fatalf("configNew: %v", err) + } + if err := l.callDyn(func() int32 { return l.configSetHTTPath(cfg, host, httpPath) }); err != nil { + l.configFree(cfg) + t.Fatalf("set_http_path: %v", err) + } + if err := l.callDyn(func() int32 { return l.configSetAuthPAT(cfg, token) }); err != nil { + l.configFree(cfg) + t.Fatalf("set_auth_pat: %v", err) + } + var session uintptr + if err := l.callDyn(func() int32 { return l.sessionOpen(cfg, &session) }); err != nil { + l.configFree(cfg) + t.Fatalf("session_open: %v", err) + } + return &dynSession{t: t, l: l, session: session} +} + +func (s *dynSession) close() { + if s.session != 0 { + _ = s.l.callDyn(func() int32 { return s.l.sessionClose(s.session) }) + s.session = 0 + } +} + +// queryAll runs sql and returns every row scanned through the pure-Go data +// plane. Fails the test on any error. +func (s *dynSession) queryAll(t *testing.T, sql string) [][]driver.Value { + t.Helper() + l := s.l + + var stmt uintptr + if err := l.callDyn(func() int32 { return l.newStatement(s.session, &stmt) }); err != nil { + t.Fatalf("new_statement: %v", err) + } + defer func() { _ = l.callDyn(func() int32 { return l.statementClose(stmt) }) }() + + if err := l.callDyn(func() int32 { return l.setSQL(stmt, sql) }); err != nil { + t.Fatalf("set_sql: %v", err) + } + var executed uintptr + if err := l.callDyn(func() int32 { return l.execute(stmt, &executed) }); err != nil { + t.Fatalf("execute: %v", err) + } + defer func() { _ = l.callDyn(func() int32 { return l.execClose(executed) }) }() + + var stream uintptr + if err := l.callDyn(func() int32 { return l.getResultStream(executed, &stream) }); err != nil { + t.Fatalf("get_result_stream: %v", err) + } + + rows, err := newDynRows(l, stream, time.UTC) + if err != nil { + t.Fatalf("newDynRows: %v", err) + } + defer rows.Close() + + ncols := len(rows.Columns()) + var out [][]driver.Value + for { + dest := make([]driver.Value, ncols) + err := rows.Next(dest) + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("rows.Next: %v", err) + } + out = append(out, dest) + } + return out +} + // TestDynamicLoaderControlPlane proves the whole thesis of the dynamic-loading // approach: a PURE-GO (CGO_ENABLED=0) binary can drive the closed-source kernel // end-to-end for the control plane — dlopen the shared library, build a config, @@ -120,3 +215,128 @@ func TestDynamicLoaderControlPlane(t *testing.T) { t.Log("PROOF: pure-Go (CGO_ENABLED=0) drove the closed-source kernel control plane end-to-end.") } + +// TestDynamicLoaderDataPlane proves the DATA plane: fetch actual result rows +// through the pure-Go Arrow C-Data importer (no cgo). It runs a query with a +// mix of types and asserts the scanned values, so a layout/import bug would +// surface as a wrong value, not just a non-crash. +// +// Same env gating as TestDynamicLoaderControlPlane. +func TestDynamicLoaderDataPlane(t *testing.T) { + dylib := os.Getenv("DBX_KERNEL_DYLIB") + host := os.Getenv("DBX_KERNEL_HOST") + httpPath := os.Getenv("DBX_KERNEL_HTTPATH") + token := os.Getenv("DBX_KERNEL_TOKEN") + if dylib == "" || host == "" || httpPath == "" || token == "" { + t.Skip("set DBX_KERNEL_DYLIB, DBX_KERNEL_HOST, DBX_KERNEL_HTTPATH, DBX_KERNEL_TOKEN to run") + } + + sess := openDynSession(t, dylib, host, httpPath, token) + defer sess.close() + + t.Run("scalars + null + decimal + string", func(t *testing.T) { + got := sess.queryAll(t, `SELECT + CAST(1 AS INT) AS i, + CAST(2 AS BIGINT) AS b, + CAST(3.5 AS DOUBLE) AS d, + CAST('hello' AS STRING) AS s, + CAST(true AS BOOLEAN) AS bo, + CAST(19.99 AS DECIMAL(10,2)) AS dec, + CAST(NULL AS STRING) AS n`) + if len(got) != 1 { + t.Fatalf("expected 1 row, got %d", len(got)) + } + row := got[0] + checkEq(t, "i", row[0], int32(1)) + checkEq(t, "b", row[1], int64(2)) + checkEq(t, "d", row[2], float64(3.5)) + checkEq(t, "s", row[3], "hello") + checkEq(t, "bo", row[4], true) + checkEq(t, "dec", row[5], "19.99") // decimal renders as exact string + checkEq(t, "n", row[6], nil) + t.Logf("OK scalars row = %v", row) + }) + + t.Run("multi-row range", func(t *testing.T) { + got := sess.queryAll(t, `SELECT id FROM range(0, 1000) ORDER BY id`) + if len(got) != 1000 { + t.Fatalf("expected 1000 rows, got %d", len(got)) + } + for i, row := range got { + if row[0].(int64) != int64(i) { + t.Fatalf("row %d: got %v", i, row[0]) + } + } + t.Logf("OK fetched %d rows in order across batches", len(got)) + }) + + t.Run("nested array/map/struct", func(t *testing.T) { + got := sess.queryAll(t, `SELECT + array(1,2,3) AS arr, + map('k', 1) AS m, + named_struct('a', 1, 'b', 'x') AS st`) + if len(got) != 1 { + t.Fatalf("expected 1 row, got %d", len(got)) + } + // Nested types render as JSON strings via the shared scanner. + checkEq(t, "arr", got[0][0], "[1,2,3]") + checkEq(t, "map", got[0][1], `{"k":1}`) + checkEq(t, "struct", got[0][2], `{"a":1,"b":"x"}`) + t.Logf("OK nested row = %v", got[0]) + }) + + t.Run("temporal + binary + float edge", func(t *testing.T) { + got := sess.queryAll(t, `SELECT + CAST('2021-07-01' AS DATE) AS d, + CAST('2021-07-01 05:43:28' AS TIMESTAMP) AS ts, + CAST(X'1abf' AS BINARY) AS bin, + CAST(3.3 AS FLOAT) AS f, + CAST(-0.01 AS DECIMAL(5,2)) AS negdec, + CAST(9999999999999999999999999999.99 AS DECIMAL(38,2)) AS bigdec`) + if len(got) != 1 { + t.Fatalf("expected 1 row, got %d", len(got)) + } + row := got[0] + // date/timestamp scan to time.Time; check via string form. + checkEq(t, "date", fmt.Sprintf("%v", row[0]), "2021-07-01 00:00:00 +0000 UTC") + checkEq(t, "ts", fmt.Sprintf("%v", row[1]), "2021-07-01 05:43:28 +0000 UTC") + if b, ok := row[2].([]byte); !ok || len(b) != 2 || b[0] != 0x1a || b[1] != 0xbf { + t.Errorf("binary: got %v (%T)", row[2], row[2]) + } + checkEq(t, "negdec", row[4], "-0.01") + // high-precision decimal must be exact (no float corruption). + checkEq(t, "bigdec", row[5], "9999999999999999999999999999.99") + t.Logf("OK temporal/binary/float row = %v", row) + }) + + t.Run("empty result set", func(t *testing.T) { + got := sess.queryAll(t, `SELECT 1 AS x WHERE 1=0`) + if len(got) != 0 { + t.Fatalf("expected 0 rows, got %d", len(got)) + } + t.Log("OK empty result set drained cleanly") + }) + + t.Run("large result 100k rows", func(t *testing.T) { + got := sess.queryAll(t, `SELECT id, id*2 AS doubled, CAST(id AS STRING) AS s FROM range(0, 100000)`) + if len(got) != 100000 { + t.Fatalf("expected 100000 rows, got %d", len(got)) + } + // spot-check a few rows + for _, i := range []int{0, 1, 50000, 99999} { + if got[i][0].(int64) != int64(i) || got[i][1].(int64) != int64(i*2) { + t.Fatalf("row %d mismatch: %v", i, got[i]) + } + } + t.Logf("OK fetched %d rows (multi-batch, likely CloudFetch)", len(got)) + }) + + t.Log("PROOF: pure-Go (CGO_ENABLED=0) fetched + scanned result rows end-to-end via the C-Data importer.") +} + +func checkEq(t *testing.T, name string, got, want any) { + t.Helper() + if fmt.Sprintf("%v", got) != fmt.Sprintf("%v", want) { + t.Errorf("%s: got %v (%T), want %v (%T)", name, got, got, want, want) + } +} diff --git a/internal/backend/kernel/dynamic_rows.go b/internal/backend/kernel/dynamic_rows.go new file mode 100644 index 00000000..4a8e4379 --- /dev/null +++ b/internal/backend/kernel/dynamic_rows.go @@ -0,0 +1,146 @@ +//go:build databricks_kernel_dynamic + +package kernel + +import ( + "database/sql/driver" + "fmt" + "io" + "time" + + "github.com/apache/arrow/go/v12/arrow" + "github.com/databricks/databricks-sql-go/internal/arrowscan" +) + +// dynRows is a driver.Rows over a kernel result stream, using the pure-Go +// C-Data importer (cdata_pure.go) instead of the cgo cdata package. It pulls one +// Arrow batch at a time via the dlopen'd kernel_result_stream_next_batch, imports +// it, and scans cells with the SAME arrowscan scanner the cgo rows.go uses — so +// values are identical to the cgo backend by construction. +// +// This is the data-plane proof for the dynamic-loading approach: everything from +// the result stream to a driver.Value runs with CGO_ENABLED=0. +type dynRows struct { + l *dynLib + stream uintptr + location *time.Location + + cols []string + colTypes []arrowscan.ColumnTypeInfo + schema *arrow.Schema + + cur arrow.Record + rowInCur int + keyCache *arrowscan.StructKeyCache + closed bool + eof bool +} + +var _ driver.Rows = (*dynRows)(nil) + +// newDynRows fetches the schema up front and returns the row iterator. +func newDynRows(l *dynLib, stream uintptr, loc *time.Location) (*dynRows, error) { + r := &dynRows{l: l, stream: stream, location: loc, keyCache: arrowscan.NewStructKeyCache()} + + var csch cArrowSchema + if err := l.callDyn(func() int32 { return l.streamGetSchema(stream, &csch) }); err != nil { + r.Close() + return nil, fmt.Errorf("kernel(dyn): get_schema: %w", err) + } + sch, err := importCArrowSchema(&csch) + if err != nil { + r.Close() + return nil, fmt.Errorf("kernel(dyn): import schema: %w", err) + } + r.schema = sch + fields := sch.Fields() + r.cols = make([]string, len(fields)) + r.colTypes = make([]arrowscan.ColumnTypeInfo, len(fields)) + for i, f := range fields { + r.cols[i] = f.Name + r.colTypes[i] = arrowscan.ColumnTypeInfoFor(f.Type) + } + return r, nil +} + +func (r *dynRows) Columns() []string { return r.cols } + +func (r *dynRows) Close() error { + if r.closed { + return nil + } + r.closed = true + if r.cur != nil { + r.cur.Release() + r.cur = nil + } + if r.stream != 0 { + _ = r.l.callDyn(func() int32 { return r.l.streamClose(r.stream) }) + r.stream = 0 + } + return nil +} + +func (r *dynRows) Next(dest []driver.Value) error { + if r.closed { + return io.EOF + } + for r.cur == nil || r.rowInCur >= int(r.cur.NumRows()) { + if r.eof { + return io.EOF + } + if err := r.nextBatch(); err != nil { + return err + } + } + rec := r.cur + for c := 0; c < len(dest); c++ { + v, err := arrowscan.ScanCellCached(rec.Column(c), r.rowInCur, r.location, r.keyCache) + if err != nil { + return fmt.Errorf("kernel(dyn): scan col %d (%s): %w", c, r.cols[c], err) + } + dest[c] = v + } + r.rowInCur++ + return nil +} + +// nextBatch pulls the next Arrow batch via the dlopen'd next_batch and imports +// it with the pure-Go importer. A released array (release==0) is EOF. +func (r *dynRows) nextBatch() error { + if r.cur != nil { + r.cur.Release() + r.cur = nil + } + var carr cArrowArray + var csch cArrowSchema + if err := r.l.callDyn(func() int32 { return r.l.streamNextBatch(r.stream, &carr, &csch) }); err != nil { + return fmt.Errorf("kernel(dyn): next_batch: %w", err) + } + if carr.release == 0 { + r.eof = true + return io.EOF + } + rec, err := importCRecordBatch(&carr, &csch) + if err != nil { + return fmt.Errorf("kernel(dyn): import batch: %w", err) + } + r.cur = rec + r.rowInCur = 0 + r.keyCache.Reset() + return nil +} + +// importCRecordBatch imports a batch given both array and schema (schema is a +// struct whose fields are the columns). Mirrors cdata.ImportCRecordBatch. +func importCRecordBatch(arr *cArrowArray, sc *cArrowSchema) (arrow.Record, error) { + field, err := importSchema(sc) + if err != nil { + return nil, err + } + st, ok := field.Type.(*arrow.StructType) + if !ok { + return nil, fmt.Errorf("kernel(dyn): recordbatch import must be struct type") + } + return importCRecordBatchWithSchema(arr, arrow.NewSchema(st.Fields(), &field.Metadata)) +} From 991f1c9412865dbeaa27f0994f47c5ac51765eec Mon Sep 17 00:00:00 2001 From: Madhavendra Rathore Date: Tue, 11 Aug 2026 17:02:31 +0530 Subject: [PATCH 3/3] poc(kernel): dynamic-linking (.dylib/.so) variant + release design MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to the kernel work: with cgo accepted and the Arrow import path settled on cgo + arrow-go cdata (zero-copy C-Data, the ADBC driver-manager model), this PoCs the remaining open question — packaging the closed-source kernel as a SHARED library loaded at run time instead of a static .a baked into every binary. New build tag databricks_kernel_dynlib (added alongside databricks_kernel) selects cgo_dynlib_darwin.go, which links libdatabricks_sql_kernel.dylib with -l + an rpath instead of naming the .a. The static cgo_darwin.go is guarded with !databricks_kernel_dynlib so the two never both compile. The Arrow C-Data import path (rows.go, arrow-go cdata) is unchanged — dynamic vs static is invisible above the link layer. Verified live on pecotesting (darwin/arm64): - otool -L shows the binary references @rpath/libdatabricks_sql_kernel.dylib externally (not baked in); binary size drops ~61MB (static .a) -> ~12MB. - 20/20 TestKernelE2E* subtests pass through the runtime-loaded dylib. - Negative proof: moving the dylib away fails at load with "dyld: Library not loaded: @rpath/libdatabricks_sql_kernel.dylib" and prints the rpath search order; restoring it works again. - Static .a path and default pure-Go build both still build unchanged. One real gotcha surfaced + fixed: cargo's default dylib install_name is the absolute build path (not relocatable); it must be set to @rpath/libdatabricks_sql_kernel.dylib (install_name_tool -id, or a link arg in the kernel build). DYNAMIC_LINK_RELEASE.md documents the full release plan: .so publishing on the kernel release line (serves ODBC too), soname/major + a kernel_abi_version() load check for versioning, why ODBC and Go can run different kernel versions (separate processes), and a phased static->dynamic recommendation that keeps the CGO_ENABLED=0 pure-Go Thrift fallback intact (no user CUJ change). PoC scope: darwin/arm64 only (linux $ORIGIN + windows DLL noted as the remaining per-OS work); dylib staged locally and gitignored. Co-authored-by: Isaac --- .gitignore | 1 + .../backend/kernel/DYNAMIC_LINK_RELEASE.md | 138 ++++++++++++++++++ internal/backend/kernel/cgo_darwin.go | 2 +- internal/backend/kernel/cgo_dynlib_darwin.go | 32 ++++ 4 files changed, 172 insertions(+), 1 deletion(-) create mode 100644 internal/backend/kernel/DYNAMIC_LINK_RELEASE.md create mode 100644 internal/backend/kernel/cgo_dynlib_darwin.go diff --git a/.gitignore b/.gitignore index 970720b6..93bf2199 100644 --- a/.gitignore +++ b/.gitignore @@ -71,3 +71,4 @@ __debug_bin /build/kernel-src/ /internal/backend/kernel/lib/ /internal/backend/kernel/include/ +/internal/backend/kernel/lib_dyn/ diff --git a/internal/backend/kernel/DYNAMIC_LINK_RELEASE.md b/internal/backend/kernel/DYNAMIC_LINK_RELEASE.md new file mode 100644 index 00000000..e8c51eb1 --- /dev/null +++ b/internal/backend/kernel/DYNAMIC_LINK_RELEASE.md @@ -0,0 +1,138 @@ +# Dynamic linking (.dylib/.so) for the kernel — release design + working PoC + +**Decision context:** cgo is acceptable; the Arrow import path stays cgo + +arrow-go `cdata` (zero-copy C-Data, the ADBC driver-manager model). The *only* +open question is how the closed-source kernel binary is packaged, shipped, and +linked. This documents the **dynamic-linking** answer, with a PoC verified live +on pecotesting. + +This is NOT the purego/CGO_ENABLED=0 path (that PoC — `dynamic_loader.go`, +`cdata_pure.go` — is moot now that cgo is accepted, and should be parked). + +--- + +## What dynamic linking changes (and what it doesn't) + +The kernel is compiled to a **shared library** (`.dylib`/`.so`/`.dll`) instead of +a static archive (`.a`). At build time the Go binary records only a *reference* +to the library; at run time the OS loads the library from a search path (rpath). + +- **Unchanged:** the entire Arrow C-Data import path (`rows.go`, arrow-go + `cdata`), all backend/operation logic, the public API, `WithUseKernel`, and the + `CGO_ENABLED=0` pure-Go Thrift fallback for non-kernel users. Dynamic vs static + is invisible above the link layer. +- **Changed:** one per-OS cgo `LDFLAGS` line (link the shared lib + set an rpath), + and the packaging/release flow (ship the `.so` beside the binary). + +## PoC results (darwin/arm64, live on pecotesting) + +Behind a new build tag `databricks_kernel_dynlib` (added alongside +`databricks_kernel`), `cgo_dynlib_darwin.go` links the `.dylib` instead of the +`.a`. Verified: + +- **Binary references the kernel externally**, not baked in: + `otool -L` → `@rpath/libdatabricks_sql_kernel.dylib`. +- **Size drops from ~61 MB (static .a) to ~12 MB** (kernel code is now external). +- **rpath baked in** so the loader finds the lib at run time + (`@loader_path/lib_dyn/darwin_arm64` + an absolute dev path). +- **20/20 `TestKernelE2E*` subtests pass on pecotesting** through the + runtime-loaded dylib (all scalar/decimal/temporal/nested types). +- **Genuinely external (negative proof):** moving the dylib away makes the binary + fail at load with `dyld: Library not loaded: @rpath/libdatabricks_sql_kernel.dylib` + and prints the exact rpath search order; restoring it works again. +- **Isolation:** the static `.a` path and the default pure-Go build both still + build unchanged (the dynlib tag guards `cgo_darwin.go` with + `!databricks_kernel_dynlib`). + +## Real code changes needed to ship this + +### Kernel repo (databricks-sql-kernel) +1. **Publish the shared library as a release artifact**, per-OS/arch, built with + `crate-type = ["cdylib", ...]` (already declared) and `tls-rustls`. Sign it. +2. **Set a relocatable install name / soname at build:** + - macOS: `install_name_tool -id @rpath/libdatabricks_sql_kernel.dylib` (the + cargo default is the absolute build path — NOT relocatable; the PoC had to + fix this). Better: set it at link time via + `RUSTFLAGS=-Clink-arg=-Wl,-install_name,@rpath/...`. + - Linux: build with `-Wl,-soname,libdatabricks_sql_kernel.so.`. +3. **Add an ABI-version symbol** `uint32_t kernel_abi_version(void)` so the Go + side can detect a mismatched library at load (see Versioning). Important while + the kernel is pre-1.0 ("ABI may change freely"). + +### Go driver repo (databricks-sql-go) +1. **New dynamic-link cgo files** per OS (the PoC has darwin; add linux/windows): + - darwin: `-L -ldatabricks_sql_kernel -Wl,-rpath,@loader_path/...` + - linux: `-L -ldatabricks_sql_kernel -Wl,-rpath,$ORIGIN/...` + - windows: import lib + ship the `.dll` beside the `.exe` (no rpath concept; + DLL is found via the executable directory). +2. **`make kernel-lib` fetches the `.so`/`.dylib`** (not the `.a`) from the + published release, verifies its checksum, and places it where the rpath points. +3. **Add a load-time ABI check** calling `kernel_abi_version()` and failing with a + clear error on mismatch (turns a silent crash into an actionable message). +4. **Decide static-vs-dynamic exposure:** either replace the static path, or keep + both (static default + `databricks_kernel_dynlib` opt-in, as the PoC does). + +## `.so` publishing — how it works + +- Kernel CI builds the shared lib per platform, signs it, attaches it to the + **kernel's own release line** (design-doc option **D** — correct owner, + kernel-versioned, and the SAME artifact serves ODBC, which already dlopens a + shared lib). +- The Go driver pins a kernel version (`KERNEL_REV` / release tag). `make + kernel-lib` downloads the matching signed `.so`, verifies it, drops it next to + the build output. +- At run time the OS loads it via the rpath. For distribution, the `.so` ships + **beside the application binary** (rpath `@loader_path` / `$ORIGIN`), so each + deployment carries its own copy — no system-wide install, no cross-app + interference. + +## Versioning (the runtime contract dynamic linking introduces) + +Static linking freezes the version at build (mismatch = compile error). Dynamic +linking resolves it at run time, so a wrong/missing `.so` is a **runtime** +problem that must be guarded: + +1. **soname major** (`libkernel.so.1`): bump the major on any ABI break; the OS + refuses to load a mismatched major (clean failure, not corruption). +2. **`kernel_abi_version()` check** on load: belt-and-suspenders, essential while + pre-1.0. gosnowflake does exactly this with its native core. +3. **Pin** the expected version in the driver source (as today). + +**ODBC and Go can run different kernel versions** because they are separate +processes with isolated memory — each ships/loads its own `.so`. (Only a single +process loading *both* would need matching majors + symbol care; not a normal +deployment.) + +## Recommendation (phased) + +- **Phase 1 (opt-in, now):** keep the static `.a` path (versioning is a build-time + non-problem); ship via release-asset `make kernel-lib`. +- **Phase 2 (toward SEA-default + ODBC sharing):** dynamic `.so` published on the + kernel release line, signed, soname-versioned, with the `kernel_abi_version()` + load check. This keeps the closed-source blob a separate signable/patchable + artifact out of every customer binary, and shares one lib with ODBC. +- The `CGO_ENABLED=0` pure-Go Thrift fallback stays intact throughout, so + non-kernel users are never forced into cgo — no user CUJ change. + +## PoC files (this branch) + +- `cgo_dynlib_darwin.go` — dynamic-link LDFLAGS (tag: `databricks_kernel_dynlib`) +- `cgo_darwin.go` — guarded with `!databricks_kernel_dynlib` so static/dynamic + never both compile +- `lib_dyn/darwin_arm64/` (gitignored) — the `.dylib` with install_name fixed to + `@rpath/...` + +## Build/run the PoC + +```sh +# stage the dylib + fix its install name (a packaging step; kernel CI would do this) +cp /target/release/libdatabricks_sql_kernel.dylib internal/backend/kernel/lib_dyn/darwin_arm64/ +install_name_tool -id @rpath/libdatabricks_sql_kernel.dylib \ + internal/backend/kernel/lib_dyn/darwin_arm64/libdatabricks_sql_kernel.dylib + +# build + run e2e dynamically linked +export CGO_LDFLAGS_ALLOW='-Wl,-rpath,@loader_path.*|-Wl,-rpath,/.*' +DATABRICKS_PECOTESTING_HTTP_PATH2=/sql/1.0/warehouses/ \ +CGO_ENABLED=1 go test -tags "databricks_kernel databricks_kernel_dynlib" \ + -run TestKernelE2E ./... -v +``` diff --git a/internal/backend/kernel/cgo_darwin.go b/internal/backend/kernel/cgo_darwin.go index bc59d19e..d3249e6e 100644 --- a/internal/backend/kernel/cgo_darwin.go +++ b/internal/backend/kernel/cgo_darwin.go @@ -1,4 +1,4 @@ -//go:build cgo && databricks_kernel && darwin && arm64 +//go:build cgo && databricks_kernel && !databricks_kernel_dynlib && darwin && arm64 package kernel diff --git a/internal/backend/kernel/cgo_dynlib_darwin.go b/internal/backend/kernel/cgo_dynlib_darwin.go new file mode 100644 index 00000000..ef66d7f4 --- /dev/null +++ b/internal/backend/kernel/cgo_dynlib_darwin.go @@ -0,0 +1,32 @@ +//go:build cgo && databricks_kernel && databricks_kernel_dynlib && darwin && arm64 + +package kernel + +// DYNAMIC-LINK variant of cgo_darwin.go (PoC). Instead of statically linking +// libdatabricks_sql_kernel.a into the binary, this links the kernel as a SHARED +// library (libdatabricks_sql_kernel.dylib) that is loaded at run time. Built by +// adding BOTH `-tags databricks_kernel` AND `-tags databricks_kernel_dynlib`; +// the static cgo_darwin.go carries a `!databricks_kernel_dynlib` guard so the +// two never both compile (see its build line). +// +// What differs from the static link: +// - `-L${SRCDIR}/lib_dyn/darwin_arm64 -ldatabricks_sql_kernel` links against +// the .dylib (the linker resolves `-lX` to libX.dylib) instead of naming +// the .a as a positional input. Nothing from the .dylib is copied into the +// Go binary; only a reference (a load command) is recorded. +// - `-Wl,-rpath,@loader_path/lib_dyn/darwin_arm64` tells the produced binary +// WHERE to find the .dylib at run time: @loader_path is the directory of the +// binary itself, so the .dylib is expected at +// /lib_dyn/darwin_arm64/libdatabricks_sql_kernel.dylib. A real +// release would put the .dylib right next to the binary and use +// `-rpath,@loader_path`. The dylib's install_name must be +// `@rpath/libdatabricks_sql_kernel.dylib` (set with install_name_tool -id; +// it defaults to the absolute build path, which would NOT be relocatable). +// +// The Arrow C-Data import path (rows.go, arrow-go cdata) is UNCHANGED — dynamic +// vs static linking is invisible above the link layer. + +/* +#cgo LDFLAGS: -L${SRCDIR}/lib_dyn/darwin_arm64 -ldatabricks_sql_kernel -lc++ -lm -Wl,-rpath,@loader_path/lib_dyn/darwin_arm64 -Wl,-rpath,${SRCDIR}/lib_dyn/darwin_arm64 +*/ +import "C"