FrankChen021 commented on code in PR #13196:
URL: https://github.com/apache/druid/pull/13196#discussion_r990736258
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidJdbcResultSet.java:
##########
@@ -55,6 +60,104 @@
*/
public class DruidJdbcResultSet implements Closeable
{
+ /**
+ * Asynchronous result fetcher. JDBC operates via REST, which is subject to
+ * timeouts if a query takes to long to respond. Fortunately, JDBC uses a
+ * batched API, and is perfectly happy to get an empty batch. This class
+ * runs in a separate thread to fetch a batch. If the fetch takes too long,
+ * the JDBC request thread will time out waiting, will return an empty batch
+ * to the client, and will remember the fetch for use in the next fetch
+ * request. The result is that the time it takes to produce results for long
+ * running queries is decoupled from the HTTP timeout.
+ */
+ public static class ResultFetcher implements Callable<Meta.Frame>
+ {
+ private final int limit;
+ private int batchSize;
+ private int offset;
+ private Yielder<Object[]> yielder;
+
+ public ResultFetcher(
+ final int limit,
+ final Yielder<Object[]> yielder
+ )
+ {
+ this.limit = limit;
+ this.yielder = yielder;
+ }
+
+ /**
+ * In an ideal world, the batch size would be a constructor parameter.
But, JDBC,
+ * oddly, allows a different batch size per request. Hence, we set the
size using
+ * this method before each fetch attempt.
+ */
+ public void setBatchSize(int batchSize)
+ {
+ this.batchSize = batchSize;
+ }
+
+ /**
+ * Result is only valid between executions, which turns out to be
+ * the only time it is called.
+ */
+ public int offset()
+ {
+ return offset;
+ }
+
+ /**
+ * Fetch the next batch up to the batch size or EOF. Return
+ * the resulting frame. Exceptions are handled by the executor
+ * framework.
+ */
+ @Override
+ public Meta.Frame call()
+ {
+ Preconditions.checkState(batchSize > 0);
+ int rowCount = 0;
+ final int batchLimit = Math.min(limit - offset, batchSize);
+ Yielder<Object[]> yielder = this.yielder;
+ final List<Object> rows = new ArrayList<>();
Review Comment:
```suggestion
final List<Object> rows = new ArrayList<>(batchLimit);
```
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidJdbcResultSet.java:
##########
@@ -73,25 +176,46 @@ public class DruidJdbcResultSet implements Closeable
* https://github.com/apache/druid/pull/4288
* https://github.com/apache/druid/pull/4415
*/
- private final ExecutorService yielderOpenCloseExecutor;
+ private final ExecutorService queryExecutor;
private final DirectStatement stmt;
private final long maxRowCount;
+ private final ResultFetcherFactory fetcherFactory;
private State state = State.NEW;
private Meta.Signature signature;
- private Yielder<Object[]> yielder;
- private int offset;
+
+ /**
+ * The fetcher to use to read batches of rows. Holds onto the yielder for a
Review Comment:
```suggestion
* The fetcher used to read batches of rows. Holds onto the yielder for a
```
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidJdbcResultSet.java:
##########
@@ -73,25 +176,46 @@ public class DruidJdbcResultSet implements Closeable
* https://github.com/apache/druid/pull/4288
* https://github.com/apache/druid/pull/4415
*/
- private final ExecutorService yielderOpenCloseExecutor;
+ private final ExecutorService queryExecutor;
private final DirectStatement stmt;
private final long maxRowCount;
+ private final ResultFetcherFactory fetcherFactory;
private State state = State.NEW;
private Meta.Signature signature;
- private Yielder<Object[]> yielder;
- private int offset;
+
+ /**
+ * The fetcher to use to read batches of rows. Holds onto the yielder for a
+ * query. Maintains the current read offset.
+ */
+ private ResultFetcher fetcher;
+
+ /**
+ * Future for a fetch that timed out waiting, and should be use again on
Review Comment:
```suggestion
* Future for a fetch that timed out waiting, and should be used again on
```
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidMeta.java:
##########
@@ -95,7 +97,12 @@ public static <T extends Throwable> T logFailure(T error,
String message, Object
*/
public static <T extends Throwable> T logFailure(T error)
{
- logFailure(error, error.getMessage());
+ if (error instanceof NoSuchConnectionException) {
Review Comment:
Any reason that we need a special case for `NoSuchConnectionException`?
Maybe some comments in the code are useful for better understanding.
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidMeta.java:
##########
@@ -139,7 +147,9 @@ public DruidMeta(
.setDaemon(true)
.build()
),
- authMapper.getAuthenticatorChain()
+ authMapper.getAuthenticatorChain(),
+ // To prevent server hammering, the timeout must be at least 1 second.
+ new ResultFetcherFactory(Math.max(1000, config.getFetchTimeoutMs()))
Review Comment:
No need to call `Math.max` because the ctor of `ResultFetcherFactory`
already guarantees the minimal timeout is 1 sec.
##########
sql/src/main/java/org/apache/druid/sql/avatica/DruidJdbcResultSet.java:
##########
@@ -55,6 +60,104 @@
*/
public class DruidJdbcResultSet implements Closeable
{
+ /**
+ * Asynchronous result fetcher. JDBC operates via REST, which is subject to
+ * timeouts if a query takes to long to respond. Fortunately, JDBC uses a
Review Comment:
```suggestion
* timeout if a query takes too long to respond. Fortunately, JDBC uses a
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
