Add failOnError and return status options to apoc.cypher.timeboxed #711
Conversation
gem-neo4j
force-pushed
the
dev_add_improvements_to_timeout
branch
2 times, most recently
from
7017728 to
1252faf
Compare
gem-neo4j
force-pushed
the
dev_add_improvements_to_timeout
branch
from
1252faf to
15b97bc
Compare
| // Check the query is valid before trying to run it | ||
| if (failOnError) { | ||
| Util.validateQuery(db, cypher); | ||
| } |
There was a problem hiding this comment.
Why, it will fail anyway when we run it, right? I assume this throws an error on invalid queries, so no status row will be emitted?
There was a problem hiding this comment.
Ah yes, I will remove this again :)
| @@ -68,11 +70,24 @@ public Stream<CypherStatementMapResult> runTimeboxed( | |||
| @Name(value = "statement", description = "The Cypher statement to run.") String cypher, | |||
| @Name(value = "params", description = "The parameters for the given Cypher statement.") | |||
| Map<String, Object> params, | |||
| @Name(value = "timeout", description = "The maximum time the statement can run for.") long timeout) { | |||
| @Name(value = "timeout", description = "The maximum time the statement can run for.") long timeout, | |||
There was a problem hiding this comment.
| @Name(value = "timeout", description = "The maximum time the statement can run for.") long timeout, | |
| @Name(value = "timeout", description = "The maximum time, in milliseconds, the statement can run for.") long timeout, |
| value = "config", | ||
| defaultValue = "{}", | ||
| description = "{ failOnError = false :: BOOLEAN, appendStatusRow = false :: BOOLEAN }") | ||
| Map<String, Object> config) { | ||
|
|
||
| final BlockingQueue<Map<String, Object>> queue = new ArrayBlockingQueue<>(100); | ||
| final AtomicReference<Transaction> txAtomic = new AtomicReference<>(); |
There was a problem hiding this comment.
Not of concern for this PR, but it's not allowed to access transactions from multiple threads as we do with this one.
There was a problem hiding this comment.
...perhaps innocent enough since we only terminate it.
There was a problem hiding this comment.
Ah yeah, APOC is such a can of worms, sometimes tunnel vision (aka ignoring these other issues) is the only way to complete a task :P
| hasFinished = true; | ||
| // Wait a little bit longer and try again, waiting exactly the timeout means that | ||
| // there might be a slight timing issue with termination vs. setting the queue as terminated | ||
| nextElement = queue.poll(100, MILLISECONDS); |
There was a problem hiding this comment.
Can't we simply add 100 to the first poll instead?
There was a problem hiding this comment.
In my testing it only happened on failed queries, so I figured it was nicer to not add 100ms to all polling for those that work
There was a problem hiding this comment.
Can't we replace this whole procedure with the built in tx timeout, since it runs an inner transaction anyway?
pools.getDefaultExecutorService().submit(() -> {
try (final var tx = db.beginTx(timeout, TimeUnit.MILLISECONDS)) {
try (final var result = tx.execute(query)) {
while (result.hasNext()) {
queue.offer(result.next());
}
}
tx.commit();
} catch (TransactionTimeoutException e) {
// TODO Handle
} catch (Throwable t) {
// TODO Handle
} finally {
queue.offer(POISON);
}
});
return stream of queue;
gem-neo4j
force-pushed
the
dev_add_improvements_to_timeout
branch
from
cce109a to
22aa44c
Compare
The timeout procedure is a bit weird, we had a user request to give status codes. I have updated it so that there are 2 options, one which fails on error, and one which returns an extra row at the end, which is the status :)
https://github.com/neo4j/docs-apoc/pull/362