Merge branch 'main' into move-old-to-planner

mergify[bot] · web-flow · commit 82e1d41e092f · 2022-09-20T10:35:16.000Z
diff --git a/src/query/functions/src/scalars/logics/and.rs b/src/query/functions/src/scalars/logics/and.rs
@@ -79,7 +79,7 @@ impl Function for LogicAndFiltersFunction {
         input_rows: usize,
     ) -> Result<ColumnRef> {
         if columns.len() == 1 {
-            return Ok(columns[1].column().clone());
+            return Ok(columns[0].column().clone());
         }
 
         let mut validity = None;
diff --git a/src/query/pipeline/core/src/processors/processor.rs b/src/query/pipeline/core/src/processors/processor.rs
@@ -23,7 +23,6 @@ use futures::FutureExt;
 use petgraph::graph::node_index;
 use petgraph::prelude::NodeIndex;
 
-#[derive(Debug)]
 pub enum Event {
     NeedData,
     NeedConsume,
diff --git a/src/query/planner/src/metadata.rs b/src/query/planner/src/metadata.rs
@@ -163,12 +163,32 @@ impl Metadata {
         table_index
     }
 
-    pub fn find_smallest_column_within(&self, indices: &[usize]) -> usize {
-        let entries = indices
+    /// find_smallest_column in given indices.
+    pub fn find_smallest_column(&self, indices: &[usize]) -> usize {
+        let mut smallest_index = indices.iter().min().expect("indices must be valid");
+        let mut smallest_size = usize::MAX;
+        for idx in indices.iter() {
+            let entry = self.column(*idx);
+            if let Ok(bytes) = entry.data_type.data_type_id().numeric_byte_size() {
+                if smallest_size > bytes {
+                    smallest_size = bytes;
+                    smallest_index = &entry.column_index;
+                }
+            }
+        }
+        *smallest_index
+    }
+
+    /// find_smallest_column_by_table_index by given table_index
+    pub fn find_smallest_column_by_table_index(&self, table_index: IndexType) -> usize {
+        let indices: Vec<usize> = self
+            .columns
             .iter()
-            .map(|i| self.column(*i).clone())
-            .collect::<Vec<_>>();
-        find_smallest_column(entries.as_slice())
+            .filter(|v| v.table_index == Some(table_index))
+            .map(|v| v.column_index)
+            .collect();
+
+        self.find_smallest_column(&indices)
     }
 }
 
@@ -295,24 +315,3 @@ impl ColumnEntry {
         self.path_indices.is_some()
     }
 }
-
-/// TODO(xuanwo): migrate this as a function of metadata.
-pub fn find_smallest_column(entries: &[ColumnEntry]) -> usize {
-    debug_assert!(!entries.is_empty());
-    let mut column_indexes = entries
-        .iter()
-        .map(|entry| entry.column_index)
-        .collect::<Vec<IndexType>>();
-    column_indexes.sort();
-    let mut smallest_index = column_indexes[0];
-    let mut smallest_size = usize::MAX;
-    for (idx, column_entry) in entries.iter().enumerate() {
-        if let Ok(bytes) = column_entry.data_type.data_type_id().numeric_byte_size() {
-            if smallest_size > bytes {
-                smallest_size = bytes;
-                smallest_index = entries[idx].column_index;
-            }
-        }
-    }
-    smallest_index
-}
diff --git a/src/query/service/src/pipelines/executor/executor_graph.rs b/src/query/service/src/pipelines/executor/executor_graph.rs
@@ -242,15 +242,7 @@ impl ExecutingGraph {
                     state_guard_cache = Some(node.state.lock().unwrap());
                 }
 
-                let event = node.processor.event()?;
-                tracing::debug!(
-                    "node id:{:?}, name:{:?}, event: {:?}",
-                    node.processor.id(),
-                    node.processor.name(),
-                    event
-                );
-
-                let processor_state = match event {
+                let processor_state = match node.processor.event()? {
                     Event::Finished => State::Finished,
                     Event::NeedData | Event::NeedConsume => State::Idle,
                     Event::Sync => {
diff --git a/src/query/service/src/pipelines/executor/executor_worker_context.rs b/src/query/service/src/pipelines/executor/executor_worker_context.rs
@@ -15,7 +15,6 @@
 use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::sync::Arc;
-use std::time::Instant;
 
 use common_base::base::TrySpawn;
 use common_exception::ErrorCode;
@@ -80,19 +79,7 @@ impl ExecutorWorkerContext {
     }
 
     unsafe fn execute_sync_task(&mut self, processor: ProcessorPtr) -> Result<Option<NodeIndex>> {
-        if tracing::enabled!(tracing::Level::DEBUG) {
-            let start = Instant::now();
-            processor.process()?;
-            tracing::debug!(
-                "sync processor, node id:{:?}, name:{:?}, event: {:?}",
-                processor.id(),
-                processor.name(),
-                start.elapsed()
-            );
-        } else {
-            processor.process()?;
-        }
-
+        processor.process()?;
         Ok(Some(processor.id()))
     }
 
diff --git a/src/query/service/src/pipelines/executor/processor_async_task.rs b/src/query/service/src/pipelines/executor/processor_async_task.rs
@@ -80,12 +80,6 @@ impl ProcessorAsyncTask {
                             );
                         }
                         Either::Right((res, _)) => {
-                            tracing::debug!(
-                                "async processor, node id {:?} name: {:?}, elapsed:{:?}",
-                                wraning_processor.id(),
-                                wraning_processor.name(),
-                                start.elapsed()
-                            );
                             return res;
                         }
                     }
diff --git a/src/query/service/src/servers/http/clickhouse_handler.rs b/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()?);
diff --git a/src/query/service/src/servers/http/v1/query/execute_state.rs b/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();
                 }
diff --git a/src/query/service/src/servers/mysql/mysql_interactive_worker.rs b/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()
         })?;
diff --git a/src/query/service/src/sessions/query_ctx.rs b/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);
     }
diff --git a/src/query/service/src/sessions/query_ctx_shared.rs b/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()
     }
diff --git a/src/query/service/src/sql/planner/metadata.rs b/src/query/service/src/sql/planner/metadata.rs
@@ -14,9 +14,6 @@
 
 use common_ast::ast::Expr;
 use common_ast::ast::Literal;
-use common_datavalues::prelude::*;
-use common_planner::ColumnEntry;
-use common_planner::IndexType;
 
 pub fn optimize_remove_count_args(name: &str, distinct: bool, args: &[&Expr]) -> bool {
     name.eq_ignore_ascii_case("count")
@@ -25,23 +22,3 @@ pub fn optimize_remove_count_args(name: &str, distinct: bool, args: &[&Expr]) ->
             .iter()
             .all(|expr| matches!(expr, Expr::Literal{lit,..} if *lit!=Literal::Null))
 }
-
-pub fn find_smallest_column(entries: &[ColumnEntry]) -> usize {
-    debug_assert!(!entries.is_empty());
-    let mut column_indexes = entries
-        .iter()
-        .map(|entry| entry.index())
-        .collect::<Vec<IndexType>>();
-    column_indexes.sort();
-    let mut smallest_index = column_indexes[0];
-    let mut smallest_size = usize::MAX;
-    for (idx, column_entry) in entries.iter().enumerate() {
-        if let Ok(bytes) = column_entry.data_type().data_type_id().numeric_byte_size() {
-            if smallest_size > bytes {
-                smallest_size = bytes;
-                smallest_index = entries[idx].index();
-            }
-        }
-    }
-    smallest_index
-}
diff --git a/src/query/service/src/sql/planner/mod.rs b/src/query/service/src/sql/planner/mod.rs
@@ -39,7 +39,6 @@ pub use binder::Visibility;
 use common_catalog::catalog::CatalogManager;
 use common_planner::Metadata;
 use common_planner::MetadataRef;
-pub use metadata::find_smallest_column;
 pub use semantic::normalize_identifier;
 pub use semantic::IdentifierNormalizer;
 pub use semantic::NameResolutionContext;
diff --git a/src/query/service/src/sql/planner/optimizer/heuristic/prune_columns.rs b/src/query/service/src/sql/planner/optimizer/heuristic/prune_columns.rs
@@ -17,7 +17,6 @@ use common_exception::Result;
 use common_planner::MetadataRef;
 use itertools::Itertools;
 
-use crate::sql::find_smallest_column;
 use crate::sql::optimizer::ColumnSet;
 use crate::sql::optimizer::RelExpr;
 use crate::sql::optimizer::SExpr;
@@ -90,7 +89,7 @@ impl ColumnPruner {
                         let smallest_index = self
                             .metadata
                             .read()
-                            .find_smallest_column_within(prewhere_columns.as_slice());
+                            .find_smallest_column(prewhere_columns.as_slice());
                         used.insert(smallest_index);
                     }
                     pw.output_columns = used.clone();
@@ -100,8 +99,10 @@ impl ColumnPruner {
                 }
 
                 if used.is_empty() {
-                    let columns = self.metadata.read().columns_by_table_index(p.table_index);
-                    let smallest_index = find_smallest_column(&columns);
+                    let smallest_index = self
+                        .metadata
+                        .read()
+                        .find_smallest_column_by_table_index(p.table_index);
                     used.insert(smallest_index);
                 }
 
diff --git a/src/query/streams/src/lib.rs b/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;
diff --git a/src/query/streams/src/stream_abort.rs b/src/query/streams/src/stream_abort.rs

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ impl Function for LogicAndFiltersFunction {`
`79`	`79`	`input_rows: usize,`
`80`	`80`	`) -> Result<ColumnRef> {`
`81`	`81`	`if columns.len() == 1 {`
`82`		`- return Ok(columns[1].column().clone());`
	`82`	`+ return Ok(columns[0].column().clone());`
`83`	`83`	`}`
`84`	`84`
`85`	`85`	`let mut validity = None;`
Original file line number	Diff line number	Diff line change
`@@ -80,12 +80,6 @@ impl ProcessorAsyncTask {`
`80`	`80`	`);`
`81`	`81`	`}`
`82`	`82`	`Either::Right((res, _)) => {`
`83`		`- tracing::debug!(`
`84`		`- "async processor, node id {:?} name: {:?}, elapsed:{:?}",`
`85`		`- wraning_processor.id(),`
`86`		`- wraning_processor.name(),`
`87`		`- start.elapsed()`
`88`		`- );`
`89`	`83`	`return res;`
`90`	`84`	`}`
`91`	`85`	`}`
Original file line number	Diff line number	Diff line change
`@@ -459,10 +459,7 @@ impl<W: AsyncWrite + Send + Unpin> InteractiveWorkerBase<W> {`
`459`	`459`	`}`
`460`	`460`	`};`
`461`	`461`
`462`		`- let abortable_stream = ctx`
`463`		`- .try_create_abortable(intercepted_stream.boxed())?`
`464`		`- .boxed();`
`465`		`- Ok::<_, ErrorCode>(abortable_stream)`
	`462`	`+ Ok::<_, ErrorCode>(intercepted_stream.boxed())`
`466`	`463`	`}`
`467`	`464`	`.in_current_span()`
`468`	`465`	`})?;`