server

func ActivateDBConnection ¶

func ActivateDBConnection(db *sql.DB, cfg Config, duckLakeSem chan struct{}, username string) error

ActivateDBConnection applies tenant-specific DuckLake runtime to an already initialized generic DuckDB connection used by a shared warm worker.

func ApplyProfilingSettings ¶

func ApplyProfilingSettings(ctx context.Context, conn *sql.Conn)

ApplyProfilingSettings runs ProfilingSetupSQL against the given connection, writing profiles to ProfilingOutputPath. Use this for connections that bypass ConfigureMainDB — primarily fresh per-session connections in the cluster-mode worker, where eviction between sessions discards whatever settings ConfigureMainDB applied.

func AttachDeltaCatalog ¶

func AttachDeltaCatalog(db *sql.DB, dlCfg DuckLakeConfig, sem chan struct{}) error

AttachDeltaCatalog attaches the configured Delta Lake catalog/table alongside DuckLake. It reuses the DuckLake S3 secret settings so Delta scans can access the same object store credentials.

Delta is enabled by default. When no path is derivable (e.g. a plain standalone DuckDB instance with no DuckLake object_store/data_path), this is a benign no-op: there's no Delta sibling to attach.

func AttachDuckLake ¶

func AttachDuckLake(db *sql.DB, dlCfg DuckLakeConfig, sem chan struct{}, dataDir string) error

AttachDuckLake attaches a DuckLake catalog if configured (but does NOT set it as default). Call setDuckLakeDefault after creating per-connection views in memory.main. This is a standalone function so it can be reused by control plane workers. dataDir is used for writing migration backup files if a schema upgrade is needed.

func AttachIcebergCatalog ¶

func AttachIcebergCatalog(db *sql.DB, ic IcebergConfig, sem chan struct{}, keyID, secret, sessionToken string) error

AttachIcebergCatalog attaches the per-tenant Iceberg catalog. Lakekeeper REST is the only backend; the AWS S3 Tables path was removed when S3 Tables stopped being a supported product. Idempotent if the catalog is already attached; fail-soft for the "fresh tenant, no namespaces yet" case so a worker activation isn't blocked.

func BootstrapBundledExtensions ¶

func BootstrapBundledExtensions(dataDir string) error

BootstrapBundledExtensions eagerly seeds bundled extension binaries into the configured extension_directory cache once per data directory.

func BuildDuckDBCopyFromSQL ¶

func BuildDuckDBCopyFromSQL(tableName, columnList, filePath string, opts *CopyFromOptions) string

BuildDuckDBCopyFromSQL generates a DuckDB COPY FROM statement

func CancelClientConn ¶

func CancelClientConn(cc *clientConn)

CancelClientConn cancels the context of a clientConn.

func ConfigureDBConnection ¶

func ConfigureDBConnection(db *sql.DB, cfg Config, duckLakeSem chan struct{}, username string, serverStartTime time.Time, serverVersion string) error

ConfigureDBConnection initializes an existing DuckDB connection with pg_catalog, information_schema, and DuckLake catalog attachment.

func ConfigureMainDB ¶

func ConfigureMainDB(db *sql.DB, cfg Config, username string) error

ConfigureMainDB applies the per-instance DuckDB settings (threads, memory, temp dir, extensions, profiling) that the client-query DB needs. Shared between openBaseDB (single-DB path) and OpenDuckDBPair (shared-connector path) so the main DB is configured identically either way.

func CreateDBConnection ¶

func CreateDBConnection(cfg Config, duckLakeSem chan struct{}, username string, serverStartTime time.Time, serverVersion string) (*sql.DB, error)

CreateDBConnection creates a DuckDB connection for a client session. Uses in-memory database as an anchor for DuckLake attachment (actual data lives in RDS/S3). This is a standalone function so it can be reused by both the server and control plane workers. serverStartTime is the time the top-level server process started (may differ from processStartTime in process isolation mode where each child has its own processStartTime). serverVersion is the version of the top-level server/control-plane process.

func CreatePassthroughDBConnection ¶

func CreatePassthroughDBConnection(cfg Config, duckLakeSem chan struct{}, username string, serverStartTime time.Time, serverVersion string) (*sql.DB, error)

CreatePassthroughDBConnection creates a DuckDB connection without pg_catalog or information_schema initialization. DuckLake is still attached if configured so passthrough users can access the same data. This is used for passthrough users who send DuckDB-native SQL and don't need the PostgreSQL compatibility layer.

func DuckDBDSN ¶

func DuckDBDSN(cfg Config, username string) (string, error)

DuckDBDSN returns the DSN openBaseDB / the duckdbservice pair builder use for cfg/username. Exported so duckdbservice (which holds the duckdb-go-v2 import) can build a *duckdb.Connector against the same DSN that openBaseDB would have passed to sql.Open.

func InitMinimalServer ¶

func InitMinimalServer(s *Server, cfg Config, queryCancelCh <-chan struct{})

InitMinimalServer initializes a Server struct with minimal fields for use in control plane worker sessions.

func LegacySecretDirectory ¶

func LegacySecretDirectory(cfg Config) string

LegacySecretDirectory returns DuckDB's pre-pinning default persistent-secret location for a worker whose HOME is its DataDir (<DataDir>/.duckdb/stored_secrets). This is where secrets accumulated before SecretDirectory pinning existed, and the only reason it's a named helper is so the recycle wipe and this derivation stay colocated and can't drift apart. Returns "" when DataDir is unset.

func LoadExtensions ¶

func LoadExtensions(db *sql.DB, extensions []string) error

LoadExtensions installs and loads DuckDB extensions. This is a standalone function so it can be reused by control plane workers. Extension strings can include a source, e.g. "cache_httpfs FROM community". INSTALL uses the full string; LOAD uses just the extension name.

NOTE: Extension names come from trusted server config, not user input.

func NewClientConn ¶

func NewClientConn(s *Server, conn net.Conn, reader *bufio.Reader, writer *bufio.Writer,
	username, orgID, database, applicationName string, executor QueryExecutor, pid, secretKey int32, workerID int, workerPod string) *clientConn

NewClientConn creates a clientConn with pre-initialized fields for use by the control plane worker. The returned value is opaque (*clientConn) but can be used with SendInitialParams and RunMessageLoop.

func ParseStartupOptions ¶

func ParseStartupOptions(options string) map[string]string

ParseStartupOptions parses the Postgres startup-message `options` parameter, which carries `-c name=value` GUC settings (also accepted: `-cname=value` and `--name=value`), e.g. `-c search_path=iceberg.public`. libpq's `options` connection keyword, the PGOPTIONS env var, and pgjdbc's `currentSchema` all arrive here. Values may contain backslash-escaped spaces. Returns a map of setting name -> value (later settings win on duplicate names).

func ProcessVersion ¶

func ProcessVersion() string

ProcessVersion returns the version string for this process.

func ProfilingSetupSQL ¶

func ProfilingSetupSQL(outputPath string) []string

ProfilingSetupSQL returns the SQL statements that configure DuckDB profiling so the output is written to outputPath. These are session-scoped — DuckDB rejects `SET GLOBAL` for each one — so any caller that hands out fresh connections (e.g. after evictConnFromPool) must re-run them per connection.

func ReadMessage ¶

func ReadMessage(r io.Reader) (byte, []byte, error)

func ReadStartupMessage ¶

func ReadStartupMessage(r io.Reader) (map[string]string, error)

func RecoverAbortedTransaction ¶

func RecoverAbortedTransaction[T any](
	err error,
	canRollback bool,
	rollback func() error,
	retry func() (T, error),
) (T, error, bool)

RecoverAbortedTransaction rolls back and retries once when a DuckLake-backed connection is stuck in "Current transaction is aborted" state. This is only safe when the caller owns the transaction lifecycle (autocommit / no active user transaction). Callers should pass canRollback=false for explicit user transactions so the original error is surfaced unchanged.

func RefreshIcebergSecret ¶

func RefreshIcebergSecret(db *sql.DB, ic IcebergConfig, sem chan struct{}, keyID, secretKey, sessionToken string) error

RefreshIcebergSecret replaces the DuckDB iceberg-extension S3 secret (iceberg_sigv4) with updated credentials. Used when a hot-idle worker is reclaimed and the STS credentials minted by the control plane have rotated. Without this, iceberg queries on a long-lived worker would start 403'ing after the first STS rotation (~1h) while DuckLake stays fresh.

Unlike AttachIcebergCatalog this does NOT short-circuit when the iceberg catalog is already attached — the whole point is to overwrite the existing secret in place. ATTACH state on the catalog itself is unaffected; DuckDB resolves the secret at request time, so the new credentials take effect for the next iceberg query without an explicit reattach.

Same auth model as AttachIcebergCatalog: explicit credentials are the only supported path. A refresh with missing credentials is a config bug (the activator failed to populate fresh STS credentials in the payload) and surfaces as an explicit error rather than a silent fallback to credential_chain.

func RefreshS3Secret ¶

func RefreshS3Secret(db *sql.DB, dlCfg DuckLakeConfig, duckLakeSem chan struct{}) error

RefreshS3Secret replaces the DuckDB S3 secret with updated credentials. Used when a hot-idle worker is reclaimed and STS credentials have rotated. Respects the configured S3 provider (config, aws_sdk, credential_chain).

func RegisterDuckDBAppender ¶

func RegisterDuckDBAppender(f DuckDBAppendFunc)

RegisterDuckDBAppender wires a real DuckDB Appender implementation into the COPY codepath. duckdbservice's init() calls this; binaries that don't link duckdbservice get the unavailable fallback below.

func RegisterValueNormalizer ¶

func RegisterValueNormalizer(n ValueNormalizer)

RegisterValueNormalizer adds a hook consulted by normalizeDriverValue before the binary encoders dispatch on the value's type. Intended for use from init() in importers that own driver-specific value types — duckdbservice registers a normalizer that converts duckdb.Interval and duckdb.Decimal to their arrowmap equivalents.

func RunChildMode ¶

func RunChildMode()

RunChildMode is the entry point for child worker processes. It reconstructs the TCP connection from FD 3, completes TLS handshake, authenticates the user, creates a DuckDB connection, and runs the message loop.

Configuration is read from stdin as JSON (more secure than env vars for passwords).

Exit codes:

• 0: Success (clean disconnect)
• 1: Error (crash, protocol error)
• 10: Authentication failure

func RunMessageLoop ¶

func RunMessageLoop(cc *clientConn) error

RunMessageLoop runs the main message loop for a client connection. It cancels the connection context when the loop exits, ensuring in-flight query contexts (and any gRPC calls derived from them) are cancelled promptly.

func RunShell ¶

func RunShell(cfg Config)

RunShell starts an interactive SQL shell with a fully initialized DuckDB connection. It uses the same CreateDBConnection path as the PostgreSQL server, so extensions, DuckLake, and pg_catalog views are all available.

func S3ProviderForConfig ¶

func S3ProviderForConfig(dlCfg DuckLakeConfig) string

S3ProviderForConfig returns the effective S3 provider for the given DuckLake config.

func SanitizeSearchPath ¶

func SanitizeSearchPath(s string) (string, bool)

SanitizeSearchPath validates a client-supplied search_path so it can be safely embedded in `SET search_path = '<value>'`. Returns (trimmed, true) when the value is a plausible, injection-safe search_path; ("", false) otherwise (callers should then fall back to the default search_path).

func SecretDirectory ¶

func SecretDirectory(cfg Config) string

SecretDirectory returns the directory DuckDB should use for persistent secrets for this instance, or "" to leave DuckDB's default in place.

DuckDB defaults persistent secrets to $HOME/.duckdb/stored_secrets, independent of the database file. On a worker that means a CREATE PERSISTENT SECRET lands in whatever $HOME the process happens to have and survives across restarts on any non-ephemeral disk — long after the in-memory secret of the same name (recreated each activation) gets re-added, which is exactly what produces DuckDB's "secret occurs in multiple storage backends" ambiguity errors. Pinning it under DataDir makes the location deterministic and, crucially, wipeable on worker recycle (see duckdbservice.Warmup).

func SendInitialParams ¶

func SendInitialParams(cc *clientConn)

SendInitialParams sends the initial parameter status messages and backend key data.

func SetCatalogUseRewrite ¶

func SetCatalogUseRewrite(cc *clientConn, enabled bool)

SetCatalogUseRewrite records whether this session should expand a bare `USE ducklake`/`USE iceberg` into its reliable two-part target. This is not masking — the catalog names are real; it only works around DuckDB's bare-catalog `USE` resolution.

func SetConnectionIcebergConfig ¶

func SetConnectionIcebergConfig(cc *clientConn, cfg IcebergConfig)

SetConnectionIcebergConfig records the per-tenant Iceberg config for control-plane proxy connections. The proxy server's global config is not tenant-specific, so query-time metadata loading must use this override.

func SetConnectionPhysicalCatalog ¶

func SetConnectionPhysicalCatalog(cc *clientConn, catalog string)

SetConnectionPhysicalCatalog records the resolved DuckDB catalog for control-plane proxy connections. The PostgreSQL-visible database remains on clientConn.database; this value is used only for execution/transpiler policy.

func SetPassthrough ¶

func SetPassthrough(cc *clientConn, enabled bool)

SetPassthrough flips this session into passthrough mode (bypasses the SQL transpiler + pg_catalog). The control plane resolves the per-org flag from the config store after auth and calls this before the message loop starts. Single-tenant mode keeps using server.Config.PassthroughUsers and never calls this.

func SetProcessVersion ¶

func SetProcessVersion(v string)

SetProcessVersion sets the version string for this process. Called from main().

func SetProgressFn ¶

func SetProgressFn(s *Server, fn func(pid int32) (pct float64, rows, totalRows uint64, stalled bool))

SetProgressFn sets the progress lookup function on a Server. Used by the control plane to provide cached query progress from worker health checks.

func SetQueryLogger ¶

func SetQueryLogger(s *Server, ql *QueryLogger)

SetQueryLogger sets the query logger on a Server. Used by the control plane to attach a query logger to the minimal server after creation.

func SetUserSecretManager ¶

func SetUserSecretManager(s *Server, mgr UserSecretManager)

SetUserSecretManager installs the per-user persistent secret manager on a Server. Used by the multitenant control plane after the config store is up. Must be called before the server starts accepting connections.

func StartCredentialRefresh ¶

func StartCredentialRefresh(execer sqlExecer, dlCfg DuckLakeConfig, isTxActive ...func() bool) func()

StartCredentialRefresh starts a background goroutine that periodically refreshes S3 credentials for long-lived DuckDB connections using the credential_chain provider. This prevents credential expiration when running on EC2 with IAM instance roles, STS assume-role, or other temporary credential sources.

The execer parameter accepts either *sql.DB (standalone mode) or *sql.Conn (worker mode where the pool's only connection is pinned by the session).

The optional isTxActive callback reports whether the caller currently has an active user transaction on this connection. When provided and returning false, aborted transaction errors are auto-recovered by issuing ROLLBACK and retrying once. When omitted (or returning true), automatic rollback is skipped to avoid rolling back caller-owned transactions.

Note: ExecContext serializes behind any running query (pool contention for *sql.DB, internal mutex for *sql.Conn). This means credentials are refreshed between queries, not during them. A query that runs longer than the credential TTL (~6h for instance roles) could still fail if DuckDB makes S3 requests with stale cached credentials.

Returns a stop function that cancels the refresh goroutine. The caller must call the stop function when the connection is closed to prevent goroutine leaks. If credential refresh is not needed (static credentials, no S3, etc.), returns a no-op.

func WriteAuthCleartextPassword ¶

func WriteAuthCleartextPassword(w io.Writer) error

func WriteAuthOK ¶

func WriteAuthOK(w io.Writer) error

func WriteBackendKeyData ¶

func WriteBackendKeyData(w io.Writer, pid, secretKey int32) error

func WriteErrorResponse ¶

func WriteErrorResponse(w io.Writer, severity, code, message string) error

func WriteParameterStatus ¶

func WriteParameterStatus(w io.Writer, name, value string) error

func WriteReadyForQuery ¶

func WriteReadyForQuery(w io.Writer, txStatus byte) error