Method GetFlightInfoStatement

duckdbservice/flight_handler.go:498–640

Flight SQL method implementations

(ctx context.Context, cmd flightsql.StatementQuery,
	desc *flight.FlightDescriptor)

			return status.Errorf(codes.Canceled, "wait for session idle: %v", err)
		case errors.Is(err, context.DeadlineExceeded):
			return status.Errorf(codes.DeadlineExceeded, "wait for session idle: %v", err)
		default:
			return status.Errorf(codes.Internal, "wait for session idle: %v", err)
		}
	}

	resp, _ := json.Marshal(map[string]bool{"ok": true})
	return sendActionResult(stream, &flight.Result{Body: resp})
}

// doSetSessionS3Cache applies the `duckgres.s3_cache` session GUC: it swaps
// the tenant S3 secret between the cache-proxy transport (enabled) and the
// org's native HTTPS transport (bypassed). Requires a live session — the swap
// is instance-global, which is session-scoped only under the one-session-per-
// worker contract, and a sessionless caller has no business toggling it.
// Errors surface to the control plane so the client's SET fails rather than
// silently not taking effect.
func (h *FlightSQLHandler) doSetSessionS3Cache(body []byte, stream flight.FlightService_DoActionServer) error {
	var req server.WorkerSetS3CachePayload
	if err := json.Unmarshal(body, &req); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid SetSessionS3Cache request: %v", err)
	}
	if err := h.pool.validateControlMetadata(req.WorkerControlMetadata); err != nil {
		return status.Errorf(codes.FailedPrecondition, "stale worker owner: %v", err)
	}
	if _, err := h.sessionFromContext(stream.Context()); err != nil {
		return err
	}
	mode := req.Mode
	if mode == "" {
		mode = "off"
		if req.Enabled {
			mode = "on"
		}
	}
	if err := h.pool.SetS3CacheMode(mode); err != nil {
		return status.Errorf(codes.Internal, "set session s3 cache: %v", err)
	}

	resp, _ := json.Marshal(map[string]bool{"ok": true})
	return sendActionResult(stream, &flight.Result{Body: resp})
}

func (h *FlightSQLHandler) doReleaseQueryHandle(body []byte, stream flight.FlightService_DoActionServer) error {
	var req server.WorkerReleaseQueryHandlePayload
	if err := json.Unmarshal(body, &req); err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid ReleaseQueryHandle request: %v", err)
	}
	if err := h.pool.validateControlMetadata(req.WorkerControlMetadata); err != nil {
		return status.Errorf(codes.FailedPrecondition, "stale worker owner: %v", err)
	}
	session, err := h.sessionFromContext(stream.Context())
	if err != nil {
		return err
	}
	ticket, err := flightsql.GetStatementQueryTicket(&flight.Ticket{Ticket: req.Ticket})
	if err != nil {
		return status.Errorf(codes.InvalidArgument, "invalid statement ticket: %v", err)