Fix ClickHouse progress hang in some queries
@@ -27,6 +27,7 @@ use common_clickhouse_srv::types::Block as ClickHouseBlock;
use common_clickhouse_srv::CHContext;
use common_datablocks::DataBlock;
use common_datavalues::DataSchemaRef;
+use common_exception::ErrorCode;
use common_exception::Result;
use common_planners::InsertPlan;
use common_planners::PlanNode;
@@ -72,48 +73,55 @@ impl InteractiveWorkerBase {
_ => {
let start = Instant::now();
let interpreter = InterpreterFactory::get(ctx.clone(), plan)?;
- // Write start query log.
- let _ = interpreter
- .start()
- .await
- .map_err(|e| tracing::error!("interpreter.start.error: {:?}", e));
-
let name = interpreter.name().to_string();
- let async_data_stream = interpreter.execute(None);
- let mut data_stream = async_data_stream.await?;
histogram!(
super::clickhouse_metrics::METRIC_INTERPRETER_USEDTIME,
start.elapsed(),
"interpreter" => name
);
- let mut interval_stream = IntervalStream::new(interval(Duration::from_millis(30)));
- let cancel = Arc::new(AtomicBool::new(false));
- let (mut tx, rx) = mpsc::channel(20);
- let mut tx2 = tx.clone();
+ let cancel = Arc::new(AtomicBool::new(false));
let cancel_clone = cancel.clone();
+ let (tx, rx) = mpsc::channel(20);
+ let mut data_tx = tx.clone();
+ let mut progress_tx = tx;
+
let progress_ctx = ctx.clone();
+ let mut interval_stream = IntervalStream::new(interval(Duration::from_millis(30)));
tokio::spawn(async move {
while !cancel.load(Ordering::Relaxed) {
let _ = interval_stream.next().await;
let values = progress_ctx.get_and_reset_scan_progress_value();
- tx.send(BlockItem::ProgressTicker(values)).await.ok();
+ progress_tx
+ .send(BlockItem::ProgressTicker(values))
+ .await
+ .ok();
}
});
ctx.try_spawn(async move {
+ // Query log start.
+ let _ = interpreter
+ .start()
+ .await
+ .map_err(|e| tracing::error!("interpreter.start.error: {:?}", e));
+
+ // Execute and read stream data.
+ let async_data_stream = interpreter.execute(None);
+ let mut data_stream = async_data_stream.await?;
while let Some(block) = data_stream.next().await {
- tx2.send(BlockItem::Block(block)).await.ok();
+ data_tx.send(BlockItem::Block(block)).await.ok();
}
-
cancel_clone.store(true, Ordering::Relaxed);
- })?;
- // Write final query log.
+
+ // Query log finish.
let _ = interpreter
.finish()
.await
.map_err(|e| tracing::error!("interpreter.finish.error: {:?}", e));
+ Ok::<(), ErrorCode>(())
+ })?;
Ok(rx)
}
