Concurrent requests

[nineinchnick]

Introduce two goroutines for making HTTP requests and decoding the JSON payload concurrently. This brings only a minor improvement (see benchmarks below) unless the server can send the nextUri as an HTTP header, so the next request can be made immediately, without waiting to decode the payload.

Before:

% go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup
goos: darwin
goarch: arm64
pkg: github.com/trinodb/trino-go-client/trino
BenchmarkQuery
BenchmarkQuery-10    	      1	55366793167 ns/op
PASS
ok  	github.com/trinodb/trino-go-client/trino	67.287s
go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup  51.59s user 0.88s system 77% cpu 1:07.69 total

After:

% go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup
goos: darwin
goarch: arm64
pkg: github.com/trinodb/trino-go-client/trino
BenchmarkQuery
BenchmarkQuery-10    	      1	48234415166 ns/op
PASS
ok  	github.com/trinodb/trino-go-client/trino	48.676s
go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup  51.94s user 0.89s system 107% cpu 49.096 total

[nineinchnick]

Here's the remaining work required to handle the nextUri send in the header: nineinchnick/trino-go-client@concurrent-requests...nineinchnick:trino-go-client:concurrent-requests-header

This brings the following improvement:

% go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup -trino_server_dsn=http://bench@localhost:8084
goos: darwin
goarch: arm64
pkg: github.com/trinodb/trino-go-client/trino
BenchmarkQuery
BenchmarkQuery-10    	      1	44339327667 ns/op
PASS
ok  	github.com/trinodb/trino-go-client/trino	44.505s
go test -v -race -timeout 1m ./... -run=X -bench=. -no_cleanup   51.89s user 0.87s system 117% cpu 44.908 total

The only change required in the server is:

diff --git a/core/trino-main/src/main/java/io/trino/server/protocol/ExecutingStatementResource.java b/core/trino-main/src/main/java/io/trino/server/protocol/ExecutingStatementResource.java
index 23112ad29e4..d0409b6b587 100644
--- a/core/trino-main/src/main/java/io/trino/server/protocol/ExecutingStatementResource.java
+++ b/core/trino-main/src/main/java/io/trino/server/protocol/ExecutingStatementResource.java
@@ -240,6 +240,7 @@ public class ExecutingStatementResource
         query.getSetCatalog().ifPresent(catalog -> response.header(protocolHeaders.responseSetCatalog(), catalog));
         query.getSetSchema().ifPresent(schema -> response.header(protocolHeaders.responseSetSchema(), schema));
         query.getSetPath().ifPresent(path -> response.header(protocolHeaders.responseSetPath(), path));
+        response.header("nextUri", queryResults.getNextUri());
 
         // add set session properties
         query.getSetSessionProperties()

EDIT: I had correctness issues. I updated the results above. The gain is small, ~4s.

[losipiuk]

what is it for?

[nineinchnick]

I'm not sure, I copied this config from other tests :-)

[losipiuk]

I do not quite get it? What would happen if we did not run active drain loop here? Would go routines writing to st.error block?
Then what would happen if we want to write to st.errors and Close() was not called? Do we assume that then fetch() is running and the erros channels is being read from?

[nineinchnick]

1. If an error happens in any of the two goroutines, they write errors to errors.
2. fetch() consumes the first error and returns it to the user, who's responsible for closing the whole statement.
3. Close() closes up all channels, which will allow goroutines to finish, but before they do that, more errors can happen. These are ignored because the errors channel is drained.

Both goroutines exit after the first error, but there could be a very rare case when we receive a malformed response AND the next HTTP request will fail. The HTTP request can fail first, but there can be an error in the other goroutine when trying to decode the previous response. Draining this errors channel handles this.

If the user doesn't call Close(), the goroutines can remain blocked on trying to write to full channels, either the regular one for results or the errors channel.

Dealing with all of this is forced by running tests with the race detector (go test -race), which fails if there's any possible race. This requires closing all the channels in the right order.

[nineinchnick]

I need to add a test for a query with a larger number of result rows to make sure we're not dropping any rows.

[losipiuk]

LGTM - are there any outstanding changes you want to do here @nineinchnick ?

[nineinchnick]

@losipiuk it's ready to go, thanks!