Documentation
¶
Index ¶
- Constants
- Variables
- type FlightExecutor
- func NewFlightExecutor(addr, bearerToken, sessionToken string) (*FlightExecutor, error)
- func NewFlightExecutorFromClient(client *flightsql.Client, sessionToken string) *FlightExecutor
- func NewFlightExecutorFromClientWithQueryLogLimiter(client *flightsql.Client, sessionToken string, limiter *QueryLogLimiter) *FlightExecutor
- func (e *FlightExecutor) Close() error
- func (e *FlightExecutor) ConnContext(ctx context.Context) (sqlcore.RawConn, error)
- func (e *FlightExecutor) CopyFromStdin(ctx context.Context, request sqlcore.CopyFromStdinRequest, r io.Reader) (int64, error)
- func (e *FlightExecutor) Exec(query string, args ...any) (sqlcore.ExecResult, error)
- func (e *FlightExecutor) ExecContext(ctx context.Context, query string, args ...any) (result sqlcore.ExecResult, err error)
- func (e *FlightExecutor) IsDead() bool
- func (e *FlightExecutor) LastProfilingOutput() string
- func (e *FlightExecutor) Log(entry wire.QueryLogEntry)
- func (e *FlightExecutor) MarkDead()
- func (e *FlightExecutor) PingContext(ctx context.Context) error
- func (e *FlightExecutor) Query(query string, args ...any) (sqlcore.RowSet, error)
- func (e *FlightExecutor) QueryContext(ctx context.Context, query string, args ...any) (rs sqlcore.RowSet, err error)
- func (e *FlightExecutor) SetControlMetadata(workerID int, cpInstanceID string, ownerEpoch int64)
- func (e *FlightExecutor) SetOwnerEpoch(ownerEpoch int64)
- func (e *FlightExecutor) SetS3CacheEnabled(ctx context.Context, enabled bool) (err error)
- type FlightRowSet
- type QueryLogLimiter
Constants ¶
const CopyFromStdinChunkSize = 1 << 20 // 1 MiB
CopyFromStdinChunkSize is the byte size of each FlightData frame sent from the control plane to the worker during a COPY upload. Large enough to amortise per-frame overhead, small enough to keep memory bounded if the worker is slow to drain.
const CopyFromStdinDescriptorPath = "duckgres-copy-from-stdin"
CopyFromStdinDescriptorPath is the FlightDescriptor path that marks a DoPut stream as a spool-and-COPY upload rather than a standard Flight SQL CommandStatementUpdate. The duckdbservice worker matches on this path and routes to its CopyFromStdin handler.
const CopyFromStdinPathPlaceholder = "__DUCKGRES_COPY_PATH__"
CopyFromStdinPathPlaceholder is the substring inside the COPY SQL that the worker replaces with the worker-local spool file path before executing the COPY. The control plane builds the COPY SQL with BuildDuckDBCopyFromSQL using this token as the file path.
const CopyFromStdinPostgresBinaryPathVersion = "postgres-binary-v1"
CopyFromStdinPostgresBinaryPathVersion marks descriptor.Cmd as a structured, versioned PostgreSQL binary COPY request. An older worker fails closed when it receives the JSON request instead of executing scanner SQL without the accompanying validation metadata.
const CopyFromStdinSizePlaceholder = "__DUCKGRES_COPY_SIZE__"
CopyFromStdinSizePlaceholder is replaced by the worker with the exact number of bytes in its completed spool file. The postgres binary table function needs this because its COPY-function wrapper currently ignores buffer_size and cannot read files larger than its 32 MiB default.
const MaxGRPCMessageSize = 1 << 30 // 1GB
MaxGRPCMessageSize is the max gRPC message size for Flight SQL communication. DuckDB query results can easily exceed the default 4MB limit.
Variables ¶
var ErrWorkerDead = errors.New("flight worker is dead")
ErrWorkerDead is returned when the backing worker process has crashed.
Functions ¶
This section is empty.
Types ¶
type FlightExecutor ¶
type FlightExecutor struct {
// contains filtered or unexported fields
}
FlightExecutor implements QueryExecutor backed by an Arrow Flight SQL client. It routes queries to a duckdb-service worker process over a Unix socket.
func NewFlightExecutor ¶
func NewFlightExecutor(addr, bearerToken, sessionToken string) (*FlightExecutor, error)
NewFlightExecutor creates a FlightExecutor connected to the given address. addr should be "unix:///path/to/socket" for Unix sockets or "host:port" for TCP. bearerToken is the authentication token for the duckdb-service. sessionToken is the session identifier for the x-duckgres-session header.
func NewFlightExecutorFromClient ¶
func NewFlightExecutorFromClient(client *flightsql.Client, sessionToken string) *FlightExecutor
NewFlightExecutorFromClient creates a FlightExecutor that shares an existing Flight SQL client. The client is NOT closed when this executor is closed. This avoids creating a new gRPC connection per session.
func NewFlightExecutorFromClientWithQueryLogLimiter ¶
func NewFlightExecutorFromClientWithQueryLogLimiter(client *flightsql.Client, sessionToken string, limiter *QueryLogLimiter) *FlightExecutor
NewFlightExecutorFromClientWithQueryLogLimiter creates a per-session executor whose asynchronous query-log calls share a worker-wide limiter.
func (*FlightExecutor) Close ¶
func (e *FlightExecutor) Close() error
func (*FlightExecutor) ConnContext ¶
func (*FlightExecutor) CopyFromStdin ¶
func (e *FlightExecutor) CopyFromStdin(ctx context.Context, request sqlcore.CopyFromStdinRequest, r io.Reader) (int64, error)
CopyFromStdin streams COPY input bytes from r to a remote worker, then runs request.SQLTemplate against a worker-local spool file. The SQL template must contain CopyFromStdinPathPlaceholder where the file path should appear. It returns the number of rows the worker reports COPY-affected.
Wire layout:
frame 0: FlightDescriptor{Type=PATH, Path=[CopyFromStdinDescriptorPath, optional version], Cmd=SQL or structured request}
frame 1..N: DataBody=<chunk of COPY input bytes>
(client closes send)
server: PutResult{AppMetadata=DoPutUpdateResult{RecordCount=N}}
func (*FlightExecutor) Exec ¶
func (e *FlightExecutor) Exec(query string, args ...any) (sqlcore.ExecResult, error)
func (*FlightExecutor) ExecContext ¶
func (e *FlightExecutor) ExecContext(ctx context.Context, query string, args ...any) (result sqlcore.ExecResult, err error)
func (*FlightExecutor) IsDead ¶
func (e *FlightExecutor) IsDead() bool
IsDead reports whether this executor has been marked dead.
func (*FlightExecutor) LastProfilingOutput ¶
func (e *FlightExecutor) LastProfilingOutput() string
func (*FlightExecutor) Log ¶
func (e *FlightExecutor) Log(entry wire.QueryLogEntry)
Log implements the server query-log forwarding hook without making query completion wait on worker RPC or DuckLake writes.
func (*FlightExecutor) MarkDead ¶
func (e *FlightExecutor) MarkDead()
MarkDead marks this executor's backing worker as dead. All subsequent RPC calls will return ErrWorkerDead without touching the (possibly closed) gRPC client.
func (*FlightExecutor) PingContext ¶
func (e *FlightExecutor) PingContext(ctx context.Context) error
func (*FlightExecutor) QueryContext ¶
func (*FlightExecutor) SetControlMetadata ¶
func (e *FlightExecutor) SetControlMetadata(workerID int, cpInstanceID string, ownerEpoch int64)
func (*FlightExecutor) SetOwnerEpoch ¶
func (e *FlightExecutor) SetOwnerEpoch(ownerEpoch int64)
func (*FlightExecutor) SetS3CacheEnabled ¶
func (e *FlightExecutor) SetS3CacheEnabled(ctx context.Context, enabled bool) (err error)
SetS3CacheEnabled asks the session's worker to route the tenant's S3 traffic through the node-local cache proxy (true, the default) or to bypass it (false) by swapping the S3 secret's transport. Implements the server package's S3CacheControl capability, backing the `duckgres.s3_cache` session GUC. Unlike the best-effort teardown actions above, errors are surfaced: a SET that did not take effect on the worker must fail, not silently leave the session in the wrong cache state. Workers running an image that predates the action reject it with Unimplemented, which also surfaces as an error.
type FlightRowSet ¶
type FlightRowSet struct {
// contains filtered or unexported fields
}
FlightRowSet wraps an Arrow Flight RecordBatch reader to implement RowSet.
func (*FlightRowSet) Close ¶
func (r *FlightRowSet) Close() error
func (*FlightRowSet) ColumnTypes ¶
func (r *FlightRowSet) ColumnTypes() ([]sqlcore.ColumnTyper, error)
func (*FlightRowSet) Columns ¶
func (r *FlightRowSet) Columns() ([]string, error)
func (*FlightRowSet) Err ¶
func (r *FlightRowSet) Err() error
func (*FlightRowSet) Next ¶
func (r *FlightRowSet) Next() bool
func (*FlightRowSet) Scan ¶
func (r *FlightRowSet) Scan(dest ...any) error
type QueryLogLimiter ¶
type QueryLogLimiter struct {
// contains filtered or unexported fields
}
QueryLogLimiter bounds concurrent control-plane query-log RPCs for one worker. It holds no entries itself; sessions sharing a worker share the same limiter so a stalled endpoint cannot accumulate unbounded goroutines.
func NewQueryLogLimiter ¶
func NewQueryLogLimiter() *QueryLogLimiter
NewQueryLogLimiter creates a limiter with the production per-worker limit.
Source Files