execute multiple select queries asynchronously and consolidate results

[vic0824]

I need to execute 3 select queries and put the 3 results into a JSON object to be sent in the body of an http response.
The 3 queries are independent so I thought this would be a good opportunity to exploit the db.async interface.
However, while it was clear how to use the callback methods to get the result of each query, I could not find an easy way to launch the 3 queries asynchronously and then wait for all 3 to complete.
I have then inspected the source code and saw that the DatabaseAsyncExecutor interface has a waitCompletion() method that waits for the completion of all pending tasks.
Can I use this method to achieve the behaviour I'm looking for?

[lvca]

@vic0824 you could use the waitCompletion(), but it'd wait for any other pending requests, so if you're other things running it's not the ideal.

You could use a CountDownLatch to do that. This is an example (not tested. checkout for misspelling):

CountDownLatch counter = new CountDownLatch(3);

Resultset resultset1 = database.async().query("sql", "select from X", new AsyncResultsetCallback() {
  public void onComplete() { counter.countDown();  }
});

Resultset resultset2 = database.async().query("sql", "select from Y", new AsyncResultsetCallback() {
  public void onComplete() { counter.countDown();  }
});

Resultset resultset3 = database.async().query("sql", "select from Z", new AsyncResultsetCallback() {
  public void onComplete() { counter.countDown();  }
});

counter.await();
// OR YOU CAN SPECIFY A TIMEOUT, LIKE 10 SECONDS TOP
// counter.await(10, TimeUnit.SECONDS);

[vic0824]

Great, thanks.

[vic0824]

Hold on, how do you get the query results out of the onNext method? I get the dreaded "must be final or effectively final" error when trying to assign the result to a variable defined outside the scope of the anonymous inner class.

[vic0824]

I'm afraid that's an old implementation.
The latest version of DataBaseAsyncExecutor.query returns void:

  /**
   * Schedules the execution of a query as an idempotent (read only) command by specifying the language and an optional variable array of arguments.
   *
   * @param language The language to use between the supported ones ("sql", "gremlin", "cypher", "graphql", "mongo", etc.)
   * @param query    The command to be interpreted in the specified language as a string
   * @param callback Callback to manage the query result
   * @param args     (optional) Arguments to pass to the command as a variable length array
   */
  void query(String language, String query, AsyncResultsetCallback callback, Object... args);

and onComplete() does not have the ResultSet as an argument.

[lvca]

I found the code responsible for the asynchronous query and you're right, there is no Resultset returned, but rather this:

      if (userCallback != null) {
        userCallback.onStart(resultset);
        while (resultset.hasNext())
          if (!userCallback.onNext(resultset.next()))
            return;
        userCallback.onComplete();
      }

This means on onComplete() the resultset is already browsed and unusable. By looking at this API I think it's not really usable from a user standpoint. I'd love to do some refactoring and leave just 2 methods in the AsyncResultsetCallback interface:

public interface AsyncResultsetCallback {
  void onComplete(final ResultSet resultset);
  void onError(final Exception exception);
}

[lvca]

In this way you can also just use a lambda (the onError() is optional):

database.async().query("sql", "select from vertices", resultset -> {
  // YOUR CODE HERE
});

[lvca]

Created a new issue from this: #984

[lvca]

I've created a new test case that uses the new API:

  @Test
  public void testParallelQueries() throws InterruptedException {
    database.begin();
    try {
      CountDownLatch counter = new CountDownLatch(3);

      final ResultSet[] resultSets = new ResultSet[3];
      database.async().query("sql", "select from " + TYPE_NAME, resultset -> {
        resultSets[0] = resultset;
        counter.countDown();
      });
      database.async().query("sql", "select from " + TYPE_NAME, resultset -> {
        resultSets[1] = resultset;
        counter.countDown();
      });
      database.async().query("sql", "select from " + TYPE_NAME, resultset -> {
        resultSets[2] = resultset;
        counter.countDown();
      });

      // WAIT INDEFINITELY
      counter.await();

      Assertions.assertTrue(resultSets[0].hasNext());
      Assertions.assertTrue(resultSets[1].hasNext());
      Assertions.assertTrue(resultSets[2].hasNext());

    } finally {
      database.commit();
    }
  }

It's already available on the latest main branch and will be part of v23.3.1 to be released this week.

[vic0824]

Excellent, thanks.
When you have time, could you please take another look at #839?
I have provided a more streamlined test case that uses assertions, so you should now be able to reproduce it very quickly.