Fix issue of queries not reaching completed state

[guyco33]

Fix issue of queries not reaching completed state - see discussion on https://trinodb.slack.com/archives/CGB0QHWSW/p1658934510749439

Adds a test to demonstrate effect of trinodb/trino#13055

[ebyhr]

Could you add a test?

[guyco33]

It seems that 351 behaves differently with query state ..

[hashhar]

cc: @arhimondr @losipiuk Is there consensus on whether the new behaviour is working as expected? I think this is related to trinodb/trino#13055

[mdesmet]

This should be a non-op change as the only difference between scalar and scalar_one is that scalar_one would additionally throw if no rows are returned.

[guyco33]

SELECT version() should always return exactly one row and we want to raise exception when it doesn't.
Alternative solution might be to execute res.fetchall() after version = res.scalar() but it's cumbersome.

[mdesmet]

The issue is not limited to sqlalchemy. Following dbapi code will also result in cancelled query while the result is scalar.

cur.execute('SELECT VERSION()')
cur.fetchone()
cur.cancel()

It is indeed related to the PR on Trino that @hashhar refers to. The issue is that the python client doesn't acknowledge the reception of the data by calling the next_uri on the Trino API.

[hashhar]

I believe #220 is the way to go. Thanks @mdesmet for the thorough investigation and the fix.

[hashhar]

@mdesmet @ebyhr Can you PTAL again. This just adds a regression test now (which asserts the broken behaviour) - the fix will be part of #220.

[mdesmet]

This code can be used to simulate this using dbapi on the #220 branch. This didn't result in any unfinished queries. Now again I don't know enough about the related Trino internals what's the chance of this becoming a flaky test on github CI.

import trino

count_not_finished = 0
trino_connection = trino.dbapi.connect(
    host="localhost",
    port=8080,
    user="admin"
)
for _ in range(0, 10000):
    cur = trino_connection.cursor()
    cur.execute("SELECT VERSION()")
    cur.fetchone()
    cur.fetchone() # simulate sqlalchemy scalar_one
    if not cur._query.finished:
        count_not_finished += 1

print(str(count_not_finished) + " unfinished queries")

In other words we need to be sure that doing 2 fetchone() calls (actually fetching up to 3 requests as long next_uri is not null) is enough to fully consume all results and mark the query as finished.

[hashhar]

This has been addressed via #220 now. It has a different test though since the test here is potentially flaky (and also fails some of the existing tests regarding INSERTs).