Implementation of cancel query functionality for some large databases

[axeisghost]

Some large size database like Hive may takes a long time to do the query and sometimes users may want to cancel it when the query is running. In caravel the query is called through the API of pandas, which will take up one thread to wait for the results. To implement the cancel query function, the cancel request will be started from another thread and some certificate like the id of query must be send with the request in order to maintain the statelessness of the request.
One way to do it is to execute the query statement asynchronically, then the return of the query execution method will be the query id (or other certificate that allows any user to retrieve the query). Then we manually halt the thread to wait for the response. The query id we received will be put into the database or cache that other thread can access, so when user sends the cancel request, caravel can retrieve the query id from database/cache and retrieve the query and send cancel request.
Basically I implemented the cancel query functionality specified for modified PyHive connector. The serialization of operationHandler is the certificate to find out the query and it will be put into the database each time the query is called. Other thread can retrieve it with the username of requester from the database, then send out the cancel request to Hive.
The code is under this branch.

Right now, caravel will do specified query call for Hive and all other database will use the old pandas query method. However, this change is too specified and I know it is not a good way to change it like this. I sincerely want to know other developers' thoughts about the implementation of this functionality. I believe this can be a useful function especially for larger size database.

[georgeke]

We've been thinking about this functionality as well.
Two thoughts:

1. Implementation: I'm not sure if we need to have a table for storing results. Is there a concurrency reason for doing this other than sharing the data? The multiprocessing library has a thread-safe Queue class for this. This way we don't need to write to db every query.
2. Generalization: We can continue to use the pandas read_sql_query. This is better for performance instead of having to convert our Result to a df. The thing is that each DBAPI (sqlite3, psychopg, pyhive, etc.) has different functions for issuing a "cancel" to the respective db (interrupt() for sqlite, cancel() for hive, etc.) So I was thinking we could have a dict that maps db type to a function which does the cancelling.

What do you think?

Currently, I have a branch with cancel working with sqlite, and I'm working to generalize it for other backends. I'll try to have something pushed by today

[axeisghost]

Sorry for the late reply.

The table I used does not store the data result. It stores the query ID/handler ID that allows anyone to retrieve the query in the backend database and then cancel it. Because we will have to server caravel with multiple gunicorn workers to allow multiple clients connection at the same time. The request from clients should be RESTful. That means the cancel request sent by clients should have a certificate or ID that enable server to retrieve the query again. Therefore, I added a table in caravel's db to look up the query ID. The information shared by server and clients regardless of the state is the username that has been logged in, so I used that as the key to find the query ID.

I am really interested in the dict that maps db to a function, if we have that function, we can call the cancel function request by any clients. And I have question that where should we put the dict? If I am serving caravel with multiple gunicorn workers, it definitely cannot be placed at only one worker because every client should have access to it in order to cancel the query correctly. I am not sure which part of the caravel is run as a single layer in gunicorn server but that part should be where we put the dict you designed. Is that correct?

Also, do you have the link to the branch you mentioned? I cannot find it in the airbnb/caravel repo. I will return to work on this function soon. Thank you.

[georgeke]

Ah ok, sorry for the misunderstanding. I have three questions:

1. Is the username a unique enough identifier? What if I opened explore in two tabs and ran two queries?
2. What db-api are you using for your branch? I can't seem to find documentation for the cancel function you're calling (specifically the argument you're passing in)
3. The new table seems specific to the db-api, I wonder how we can generalize this? For sqlite and postgres it seems we need to get the same connection object that issued the query. If we can figure out a way to share that object then the map below would work.

The map would just look something like this:

cancel_map = {
    'sqlite': lambda conn: conn.interrupt(),
    'postgresql': lambda conn: conn.cancel(),
    ...
}

Where conn is the connection object of the db-api

I was having trouble with a solution for question 3, which is why I didn't push my branch. It doesn't have anything too compelling anyways.

[axeisghost]

For question 1, yes, that is also one of my concern. By using username as unique identifier it will allow each user to only run one query at a time. However, I did not find any other better identifier for now. It is possible can actually send the query ID (or encoded identifier if there is security concern) to the clients and let them use it when the send the cancel request but that will largely change the code base.

For question 2, sorry I still have no time to do the documentation yet, but the function that will run when cancel request is received is at here. The dp-api cancel function is here. The original PyHive connector does not provide a good RESTful service for cancel.

For question 3, the cancel_map is good idea, but first we have to find a place to share it to every client. We could make it a global field in models.py, that probably will work. Assuming that we have the place to share the map, we will have to have the connection object that client used to execute the query and then find out the lambda function from the map. I am not sure how the connection objects will be saved when multiple-worker gunicorn server is used. Is it possible that the cancel function in cancel_map takes in query ID/certificate as a parameter so that the db-api can retrieve the query to cancel? Although it seems most of the db-api did not provide this kind of service. (that's why I added a new cancel method in PyHive connector).

Probably there is a way to store the connection object that client used to execute the query. If that exists, query ID will be not necessarily and the way to do it can be a lot more generalized. Do you have idea to do it, especially for the case when caravel is served with multiple-worker gunicorn server?

Thank you.

[georgeke]

I don't think sharing the map is much of an issue, it's super lightweight and doesn't change so we can define it where we need it.

In terms of sharing the connection between it gunicorn processes, I don't think that is possible since the connections are local to the process. So when the cancel request comes in we can't access the connection from the query request. Unless we can write cancel functions like the one you wrote in pyHive for each DB-API, I don't think there is a clean way to implement a general Cancel button.

However, as a simpler task, perhaps we could just let the server timeout long requests and make sure we clean up the connection associated with the query (using the cancel_map)? That could all be done in the same request so we don't need to share a connection. Then in the UI we can allow users to continue running other queries while the other one times out. This of course comes with the problem of a user clicking query many many times in a row and opening a lot of connections at once. Perhaps we could limit the number of open connections they have.

I'm just thinking out loud here, I'd love to hear your thoughts.

[joshwalters]

@georgeke Interesting idea about the background queries. It could impact slower query systems though (Hive launching multiple queries would eat up cluster resources). Having global timeouts might also be tricky as some users will want to run some slower queries.

Maybe we could have a query timeout per slice, similar to cache timeout? This would be the number of seconds the query has to run before it will be terminated. I think the connect_args allows for timeout, so it would be sent with the query. Maybe this timeout could also be provided in the slice editor? Again, probably not the best solution, just brainstorming.

[georgeke]

@joshwalters Apologies for the slow response. I like the idea of having a timeout per slice for further customization if needed, but I'm not sure what the implementation of this would look like? The timeout on the gunicorn workers are set when the server starts so I'm unsure how the incoming requests can affect the timeout.

I'm curious how you guys are doing it right now with long hive queries, do you just have a large timeout on the gunicorn workers?

[mistercrunch]

Notice: this issue has been closed because it has been inactive for 633 days. Feel free to comment and request for this issue to be reopened.