Method BeginTransaction

duckdbservice/flight_handler.go:885–926

(ctx context.Context,
	req flightsql.ActionBeginTransactionRequest)

				return
			}
		}
	}()

	return schema, ch, nil
}

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

	session, err := h.sessionFromContext(ctx)
	if err != nil {
		return 0, err
	}
	finishOperation, ok := session.beginOperation()
	if busyErr := sessionBusyStatus(ok); busyErr != nil {
		return 0, busyErr
	}
	session.setCurrentQueryID(queryIDFromContext(ctx))
	defer func() {
		session.setCurrentQueryID("")
		finishOperation()
	}()

	tx, _, ttx, err := session.getOpenTxn(cmd.GetTransactionId())
	if err != nil {
		return 0, err
	}

	query := cmd.GetQuery()
	var endConnWork func()
	if tx == nil && !isEmptyFlightQuery(query) {
		var ok bool
		endConnWork, ok = session.beginConnWork()
		if !ok {
			return 0, status.Error(codes.NotFound, "session closed")
		}
		defer endConnWork()
	}
	finishDrain, err := h.pool.beginDrainWork(session.allowsDrainContinuation(string(cmd.GetTransactionId())))
	if drainErr := workerDrainingStatus(err); drainErr != nil {
		return 0, drainErr
	}
	if err != nil {
		return 0, status.Errorf(codes.Internal, "start update drain tracking: %v", err)
	}