Method startDisconnectMonitor

server/conn.go:375–422

startDisconnectMonitor starts a goroutine that polls the client connection for disconnects using bufio.Reader.Peek. During query execution the message loop is blocked on the executor, so nobody else touches the bufio.Reader, making concurrent Peek calls safe. When the client sends a TCP FIN or RST,

(ctx context.Context)

	physicalCatalog := c.physicalCatalog
	switch {
	case physicalCatalog == physicalDuckLakeCatalog:
		backend = transpiler.BackendDuckLake
	case c.server.cfg.DuckLake.MetadataStore != "" || c.server.cfg.AlwaysDuckLake:
		backend = transpiler.BackendDuckLake
		physicalCatalog = physicalDuckLakeCatalog
	}

	logicalDatabaseName := ""
	if backend == transpiler.BackendDuckLake {
		logicalDatabaseName = c.database
	}

	return transpiler.New(transpiler.Config{
		Backend:             backend,
		LogicalDatabaseName: logicalDatabaseName,
		PhysicalCatalogName: physicalCatalog,
		ConvertPlaceholders: convertPlaceholders,
	})
}

// querySourceGUCName is the fully-qualified name of the duckgres-namespaced
// session GUC, used as the SHOW result column label.
const querySourceGUCName = "duckgres.query_source"

// defaultQuerySource is the value reported for `duckgres.query_source` when the
// session GUC has not been set. Downstream (pull-based compute billing) treats a
// missing query_source as this bucket.
const defaultQuerySource = "standard"

// QuerySource returns the current value of the `duckgres.query_source` session
// GUC, or defaultQuerySource ("standard") if it was never set / set to empty.
// This never errors on a missing value: an unset GUC is defined to mean
// "standard". It is the accessor a future compute meter reads to bucket usage.
func (c *clientConn) QuerySource() string {
	if c.querySource == "" {
		return defaultQuerySource
	}
	return c.querySource
}

// setQuerySource records a client-supplied `duckgres.query_source` value on the
// session. Callers must pass an already-validated canonical value ("standard",
// "endpoints", or "" = reset to default): the transpiler validates SET
// statements (transform.NormalizeQuerySource, rejecting anything else with
// 22023 before this is reached) and applyStartupQuerySource validates the
// startup option. The setter itself stays dumb; ConnectionBilling additionally
// clamps at the metering boundary as defense in depth.
func (c *clientConn) setQuerySource(value string) {
	c.querySource = value
}