duckdbservice

type ActivationPayload ¶

type ActivationPayload struct {
	server.WorkerControlMetadata
	OrgID    string                `json:"org_id"`
	DuckLake server.DuckLakeConfig `json:"ducklake"`
	// Iceberg is the per-tenant Iceberg catalog (AWS S3 Tables) config. Empty
	// (Enabled=false) when the tenant has not opted in or hasn't been
	// provisioned yet — workers handle that as a no-op at attach time.
	Iceberg server.IcebergConfig `json:"iceberg"`
}

ActivationPayload carries the tenant-specific runtime that is delivered to a neutral shared warm worker over the control-plane RPC channel.

type DuckDBPair ¶

type DuckDBPair struct {
	Main    *sql.DB
	Control *sql.DB
	// contains filtered or unexported fields
}

DuckDBPair holds two *sql.DBs that share the same underlying DuckDB instance:

• Main is the client-query DB (MaxOpenConns=1, single-session isolation for the user's pgwire session — this is the existing behavior).
• Control is reserved for the control plane's non-blocking ops (CREATE OR REPLACE SECRET on credential refresh, ATTACH/DETACH) and is also MaxOpenConns=1 so a control DDL can't fight itself, but it has its own pool so it never queues behind a long-running client query.

Both *sql.DBs share one *duckdb.Connector / one DuckDB Database handle, which means catalogs, the SecretManager, and registered extensions are all visible across both. The main session running a query benefits from a fresh secret the moment Control swaps it — see RefreshS3Secret callers for the rationale.

Lives in duckdbservice (not server) so the control-plane binary, which imports server but not duckdbservice, doesn't transitively pull in the libduckdb-linked duckdb-go-v2 driver. CI's "does-not-link-libduckdb" check enforces this boundary.

The owner must call Close on the pair when shutting down. Close on the individual *sql.DBs goes through a non-closing wrapper, so the underlying DuckDB instance only goes away when DuckDBPair.Close fires.

func CreateWorkerDBPair(cfg server.Config, duckLakeSem chan struct{}, username string, serverStartTime time.Time, serverVersion string) (*DuckDBPair, error)

func OpenDuckDBPair(cfg server.Config, username string) (*DuckDBPair, error)

func PairFromMain(db *sql.DB) *DuckDBPair

func (p *DuckDBPair) Close() error

type DuckDBService ¶

type DuckDBService struct {
	// contains filtered or unexported fields
}

DuckDBService is a standalone Arrow Flight SQL service backed by DuckDB.

func NewDuckDBService(cfg ServiceConfig) *DuckDBService

func (svc *DuckDBService) BeginDrain()

func (svc *DuckDBService) CloseAll()

func (svc *DuckDBService) Serve(listener net.Listener) error

func (svc *DuckDBService) Shutdown()

func (svc *DuckDBService) WaitForDrain(ctx context.Context) bool

type FlightSQLHandler ¶

type FlightSQLHandler struct {
	flightsql.BaseServer
	// contains filtered or unexported fields
}

FlightSQLHandler implements Arrow Flight SQL for multiple sessions. Sessions are identified by the "x-duckgres-session" gRPC metadata header.

func NewFlightSQLHandler(pool *SessionPool) *FlightSQLHandler

func (h *FlightSQLHandler) BeginTransaction(ctx context.Context,
	req flightsql.ActionBeginTransactionRequest) ([]byte, error)

func (h *FlightSQLHandler) DoGetStatement(ctx context.Context, ticket flightsql.StatementQueryTicket) (*arrow.Schema,
	<-chan flight.StreamChunk, error)

func (h *FlightSQLHandler) DoPutCommandStatementUpdate(ctx context.Context,
	cmd flightsql.StatementUpdate) (int64, error)

func (h *FlightSQLHandler) EndTransaction(ctx context.Context,
	req flightsql.ActionEndTransactionRequest) error

func (h *FlightSQLHandler) GetFlightInfoStatement(ctx context.Context, cmd flightsql.StatementQuery,
	desc *flight.FlightDescriptor) (*flight.FlightInfo, error)

type QueryHandle ¶

type QueryHandle struct {
	Query  string
	Schema *arrow.Schema
	TxnID  string
	// contains filtered or unexported fields
}

QueryHandle stores an ad-hoc query awaiting its DoGet.

type ServiceConfig ¶

type ServiceConfig struct {
	// ListenAddr is the address to listen on.
	// Formats: "unix:///var/run/duckgres/duckdb.sock" or ":8816" or "0.0.0.0:8816"
	ListenAddr string

	// ListenFD is an inherited file descriptor for a pre-bound listener.
	// When > 0, the service uses this FD instead of creating a new socket.
	// This is used by the control plane to pass a pre-bound Unix socket to
	// worker processes, avoiding EROFS errors under ProtectSystem=strict.
	ListenFD int

	// ServerConfig is reused for CreateDBConnection (data_dir, extensions, DuckLake).
	ServerConfig server.Config

	// BearerToken is the authentication token for gRPC requests.
	// If empty, no authentication is required.
	BearerToken string

	// MaxSessions limits the number of concurrent sessions. 0 means unlimited.
	MaxSessions int

	// RequireActivation forces shared warm workers to receive tenant runtime
	// over the control-plane activation RPC before they may serve sessions.
	RequireActivation bool
}

ServiceConfig configures the standalone DuckDB Arrow Flight SQL service.

type Session ¶

type Session struct {
	ID        string
	DB        *sql.DB
	Conn      *sql.Conn // Dedicated connection for this session
	Username  string
	CreatedAt time.Time
	// contains filtered or unexported fields
}

Session represents a single DuckDB session.

type SessionPool ¶

type SessionPool struct {
	// contains filtered or unexported fields
}

SessionPool manages multiple DuckDB sessions keyed by session token.

func (p *SessionPool) ActiveDrainWork() int

func (p *SessionPool) ActiveSessions() int

func (p *SessionPool) BeginDrain()

func (p *SessionPool) CloseAll()

func (p *SessionPool) CreateSession(username, memoryLimit string, threads int, secretStatements []string) (*Session, []string, error)

func (p *SessionPool) DestroySession(token string) error

func (p *SessionPool) GetSession(token string) (*Session, bool)

func (p *SessionPool) IsDraining() bool

func (p *SessionPool) WaitForDrain(ctx context.Context) bool

func (p *SessionPool) Warmup() error