refactor(interpreter): remove useless code

Author: zhang2014

src/query/service/src/servers/http/clickhouse_handler.rs

@@ -111,7 +111,7 @@ async fn execute(
         .start()
         .await
         .map_err(|e| error!("interpreter.start.error: {:?}", e));
-    let data_stream: SendableDataBlockStream = {
+    let mut data_stream: SendableDataBlockStream = {
         let output_port = OutputPort::create();
         let stream_source = StreamSource::create(ctx.clone(), input_stream, output_port.clone())?;
         let mut source_pipe_builder = SourcePipeBuilder::create();
@@ -122,7 +122,6 @@ async fn execute(
         interpreter.execute(ctx.clone()).await?
     };
 
-    let mut data_stream = ctx.try_create_abortable(data_stream)?;
     let format_setting = ctx.get_format_settings()?;
     let mut output_format = format.create_format(schema, format_setting);
     let prefix = Ok(output_format.serialize_prefix()?);

src/query/service/src/servers/http/v1/query/execute_state.rs

@@ -30,8 +30,6 @@ use common_exception::Result;
 use common_legacy_planners::PlanNode;
 use common_streams::DataBlockStream;
 use common_streams::SendableDataBlockStream;
-use futures::future::AbortHandle;
-use futures::future::Abortable;
 use futures::StreamExt;
 use futures_util::FutureExt;
 use serde::Deserialize;
@@ -322,8 +320,7 @@ async fn execute(
     block_buffer: Arc<BlockBuffer>,
     executor: Arc<RwLock<Executor>>,
 ) -> Result<()> {
-    let data_stream = interpreter.execute(ctx.clone()).await?;
-    let mut data_stream = ctx.try_create_abortable(data_stream)?;
+    let mut data_stream = interpreter.execute(ctx.clone()).await?;
     let use_result_cache = !ctx.get_config().query.management_mode;
 
     match data_stream.next().await {
@@ -425,17 +422,10 @@ impl HttpQueryHandle {
         };
 
         let (error_sender, mut error_receiver) = mpsc::channel::<Result<()>>(1);
-        let (abort_handle, abort_registration) = AbortHandle::new_pair();
 
         GlobalIORuntime::instance().spawn(async move {
-            let error_receiver = Abortable::new(error_receiver.recv(), abort_registration);
-            ctx.add_source_abort_handle(abort_handle);
-            match error_receiver.await {
-                Err(_) => {
-                    Executor::stop(&executor, Err(ErrorCode::AbortedQuery("")), false).await;
-                    block_buffer.stop_push().await.unwrap();
-                }
-                Ok(Some(Err(e))) => {
+            match error_receiver.recv().await {
+                Some(Err(e)) => {
                     Executor::stop(&executor, Err(e), false).await;
                     block_buffer.stop_push().await.unwrap();
                 }

src/query/service/src/servers/mysql/mysql_interactive_worker.rs

@@ -459,10 +459,7 @@ impl<W: AsyncWrite + Send + Unpin> InteractiveWorkerBase<W> {
                     }
                 };
 
-                let abortable_stream = ctx
-                    .try_create_abortable(intercepted_stream.boxed())?
-                    .boxed();
-                Ok::<_, ErrorCode>(abortable_stream)
+                Ok::<_, ErrorCode>(intercepted_stream.boxed())
             }
             .in_current_span()
         })?;

src/query/service/src/sessions/query_ctx.rs

@@ -41,9 +41,6 @@ use common_legacy_planners::SourceInfo;
 use common_legacy_planners::StageTableInfo;
 use common_meta_app::schema::TableInfo;
 use common_meta_types::UserInfo;
-use common_streams::AbortStream;
-use common_streams::SendableDataBlockStream;
-use futures::future::AbortHandle;
 use opendal::Operator;
 use parking_lot::Mutex;
 use parking_lot::RwLock;
@@ -144,16 +141,6 @@ impl QueryContext {
         DataExchangeManager::instance()
     }
 
-    pub fn try_create_abortable(&self, input: SendableDataBlockStream) -> Result<AbortStream> {
-        let (abort_handle, abort_stream) = AbortStream::try_create(input)?;
-        self.shared.add_source_abort_handle(abort_handle);
-        Ok(abort_stream)
-    }
-
-    pub fn add_source_abort_handle(&self, abort_handle: AbortHandle) {
-        self.shared.add_source_abort_handle(abort_handle);
-    }
-
     pub fn attach_http_query(&self, handle: HttpQueryHandle) {
         self.shared.attach_http_query_handle(handle);
     }

src/query/service/src/sessions/query_ctx_shared.rs

@@ -25,7 +25,6 @@ use common_exception::ErrorCode;
 use common_exception::Result;
 use common_meta_types::UserInfo;
 use common_storage::StorageOperator;
-use futures::future::AbortHandle;
 use opendal::Operator;
 use parking_lot::Mutex;
 use parking_lot::RwLock;
@@ -68,7 +67,6 @@ pub struct QueryContextShared {
     pub(in crate::sessions) runtime: Arc<RwLock<Option<Arc<Runtime>>>>,
     pub(in crate::sessions) init_query_id: Arc<RwLock<String>>,
     pub(in crate::sessions) cluster_cache: Arc<Cluster>,
-    pub(in crate::sessions) sources_abort_handle: Arc<RwLock<Vec<AbortHandle>>>,
     pub(in crate::sessions) subquery_index: Arc<AtomicUsize>,
     pub(in crate::sessions) running_query: Arc<RwLock<Option<String>>>,
     pub(in crate::sessions) http_query: Arc<RwLock<Option<HttpQueryHandle>>>,
@@ -99,7 +97,6 @@ impl QueryContextShared {
             write_progress: Arc::new(Progress::create()),
             error: Arc::new(Mutex::new(None)),
             runtime: Arc::new(RwLock::new(None)),
-            sources_abort_handle: Arc::new(RwLock::new(Vec::new())),
             subquery_index: Arc::new(AtomicUsize::new(1)),
             running_query: Arc::new(RwLock::new(None)),
             http_query: Arc::new(RwLock::new(None)),
@@ -123,11 +120,6 @@ impl QueryContextShared {
             executor.finish(Some(cause));
         }
 
-        let mut sources_abort_handle = self.sources_abort_handle.write();
-
-        while let Some(source_abort_handle) = sources_abort_handle.pop() {
-            source_abort_handle.abort();
-        }
         // TODO: Wait for the query to be processed (write out the last error)
     }
 
@@ -256,11 +248,6 @@ impl QueryContextShared {
         running_query.as_ref().unwrap_or(&"".to_string()).clone()
     }
 
-    pub fn add_source_abort_handle(&self, handle: AbortHandle) {
-        let mut sources_abort_handle = self.sources_abort_handle.write();
-        sources_abort_handle.push(handle);
-    }
-
     pub fn get_config(&self) -> Config {
         self.config.clone()
     }

src/query/streams/src/lib.rs

@@ -14,15 +14,13 @@
 
 mod sources;
 mod stream;
-mod stream_abort;
 mod stream_datablock;
 mod stream_error;
 mod stream_progress;
 mod stream_take;
 
 pub use sources::*;
 pub use stream::*;
-pub use stream_abort::AbortStream;
 pub use stream_datablock::DataBlockStream;
 pub use stream_error::ErrorStream;
 pub use stream_progress::ProgressStream;