fix: wrap resultset stream pulling in "query-ctx" thread
to avoid sqllogic test hanging in cluster mode
@@ -114,14 +114,14 @@ async fn execute(
let format_typ = format.typ.clone();
// the reason of spawning new task to execute the interpreter:
- // (sorry, I failed to describe it in a more concise way)
+ // (FIXME describe this in a more concise way)
//
// - there are executions of interpreters that will block the caller (NOT async wait)
// e.g. PipelineCompleteExecutor::execute, will spawn thread that executes the pipeline,
// and then, join the thread handle.
// - async mutex (tokio::sync::Mutex) are used while executing the queries/statements
// An async task may yield while holding the lock of an async mutex. e.g. embedded meta store
- // - this method(execute) is running with default tokio runtime
+ // - this method(execute) is running with default tokio runtime (the "tokio-runtime-worker" thread)
//
// if executes the interpreter "directly" (by using current thread), the following deadlock may happen:
//
@@ -145,18 +145,10 @@ async fn execute(
//
// P.S. I think it will be better/more reasonable if we could avoid using pthread_join inside an async stack.
- let mut data_stream = ctx
- .try_spawn({
+ ctx.try_spawn({
let ctx = ctx.clone();
- async move { interpreter.execute(ctx.clone()).await }
- })?
- .await
- .map_err(|err| {
- ErrorCode::from_string(format!(
- "clickhouse handler failed to join interpreter thread: {err:?}"
- ))
- })??;
-
+ async move {
+ let mut data_stream = interpreter.execute(ctx.clone()).await?;
let table_schema = infer_table_schema(&schema)?;
let mut output_format = FileFormatOptionsExt::get_output_format_from_clickhouse_format(
format,
@@ -219,6 +211,14 @@ async fn execute(
Ok(Body::from_bytes_stream(stream).with_content_type(format_typ.get_content_type()))
}
+ })?
+ .await
+ .map_err(|err| {
+ ErrorCode::from_string(format!(
+ "clickhouse handler failed to join interpreter thread: {err:?}"
+ ))
+ })?
+}
#[poem::handler]
pub async fn clickhouse_handler_get(
