Merge branch 'main' into fix-testing-singletion-drop

Author: mergify[bot]

Cargo.lock

@@ -60,6 +60,7 @@ members = [
     "src/query/storages/system",
     "src/query/storages/view",
     "src/query/storages/parquet",
+    "src/query/storages/pruner",
     "src/query/users",
     # databend-query
     "src/query/service",

Cargo.toml

@@ -239,8 +239,7 @@ impl MetaService for MetaServiceImpl {
     ) -> Result<Response<Self::WatchStream>, Status> {
         let (tx, rx) = mpsc::channel(4);
 
-        let meta_node = &self.meta_node;
-        meta_node.create_watcher_stream(request.into_inner(), tx);
+        self.meta_node.add_watcher(request.into_inner(), tx);
 
         let output_stream = tokio_stream::wrappers::ReceiverStream::new(rx);
         Ok(Response::new(Box::pin(output_stream) as Self::WatchStream))

src/meta/service/src/api/grpc/grpc_service.rs

@@ -21,6 +21,7 @@ use std::time::Duration;
 
 use anyerror::AnyError;
 use common_base::base::tokio;
+use common_base::base::tokio::sync::mpsc;
 use common_base::base::tokio::sync::watch;
 use common_base::base::tokio::sync::watch::error::RecvError;
 use common_base::base::tokio::sync::Mutex;
@@ -83,8 +84,10 @@ use crate::metrics::server_metrics;
 use crate::network::Network;
 use crate::store::RaftStore;
 use crate::store::RaftStoreBare;
-use crate::watcher::WatcherManager;
-use crate::watcher::WatcherStreamSender;
+use crate::watcher::DispatcherSender;
+use crate::watcher::EventDispatcher;
+use crate::watcher::WatchEvent;
+use crate::watcher::WatcherSender;
 use crate::Opened;
 
 #[derive(serde::Serialize)]
@@ -123,7 +126,7 @@ pub type MetaRaft = Raft<LogEntry, AppliedState, Network, RaftStore>;
 // MetaNode is the container of meta data related components and threads, such as storage, the raft node and a raft-state monitor.
 pub struct MetaNode {
     pub sto: Arc<RaftStore>,
-    pub watcher: WatcherManager,
+    pub dispatcher_tx: mpsc::UnboundedSender<WatchEvent>,
     pub raft: MetaRaft,
     pub running_tx: watch::Sender<()>,
     pub running_rx: watch::Receiver<()>,
@@ -168,15 +171,15 @@ impl MetaNodeBuilder {
 
         let (tx, rx) = watch::channel::<()>(());
 
-        let watcher = WatcherManager::create();
+        let dispatcher_tx = EventDispatcher::spawn();
 
         sto.get_state_machine()
             .await
-            .set_subscriber(Box::new(watcher.subscriber.clone()));
+            .set_subscriber(Box::new(DispatcherSender(dispatcher_tx.clone())));
 
         let mn = Arc::new(MetaNode {
             sto: sto.clone(),
-            watcher,
+            dispatcher_tx,
             raft,
             running_tx: tx,
             running_rx: rx,
@@ -1082,7 +1085,10 @@ impl MetaNode {
         Ok(resp)
     }
 
-    pub fn create_watcher_stream(&self, request: WatchRequest, tx: WatcherStreamSender) {
-        self.watcher.create_watcher_stream(request, tx)
+    pub fn add_watcher(&self, request: WatchRequest, tx: WatcherSender) {
+        // TODO: handle error?
+        let _ = self
+            .dispatcher_tx
+            .send(WatchEvent::AddWatcher((request, tx)));
     }
 }

src/meta/service/src/meta_service/raftmeta.rs

@@ -15,8 +15,9 @@
 mod watcher_manager;
 mod watcher_stream;
 
-pub use watcher_manager::WatcherEvent;
+pub(crate) use watcher_manager::DispatcherSender;
+pub(crate) use watcher_manager::EventDispatcher;
+pub use watcher_manager::WatchEvent;
 pub use watcher_manager::WatcherId;
-pub use watcher_manager::WatcherManager;
-pub use watcher_manager::WatcherStreamSender;
+pub use watcher_manager::WatcherSender;
 pub use watcher_stream::WatcherStream;

src/meta/service/src/watcher/mod.rs

@@ -35,76 +35,60 @@ use crate::metrics::network_metrics;
 use crate::metrics::server_metrics;
 
 pub type WatcherId = i64;
-pub type WatcherStreamSender = Sender<Result<WatchResponse, Status>>;
 
-type CreateWatcherEvent = (WatchRequest, WatcherStreamSender);
+/// A sender for dispatcher to send event to interested watchers.
+pub type WatcherSender = Sender<Result<WatchResponse, Status>>;
 
+/// A sender for event source, such as raft state machine, to send event to [`EventDispatcher`].
 #[derive(Clone, Debug)]
-pub struct WatcherStateMachineSubscriber {
-    event_tx: mpsc::UnboundedSender<WatcherEvent>,
-}
+pub(crate) struct DispatcherSender(pub(crate) mpsc::UnboundedSender<WatchEvent>);
 
 #[derive(Clone)]
-pub enum WatcherEvent {
-    CreateWatcherEvent(CreateWatcherEvent),
-    StateMachineKvDataEvent(Change<Vec<u8>, String>),
-}
-
-#[derive(Debug)]
-pub struct WatcherManager {
-    event_tx: mpsc::UnboundedSender<WatcherEvent>,
-
-    pub subscriber: WatcherStateMachineSubscriber,
+pub enum WatchEvent {
+    AddWatcher((WatchRequest, WatcherSender)),
+    KVChange(Change<Vec<u8>, String>),
 }
 
-struct WatcherManagerCore {
-    event_rx: mpsc::UnboundedReceiver<WatcherEvent>,
+/// Receives events from event sources, dispatches them to interested watchers.
+pub(crate) struct EventDispatcher {
+    event_rx: mpsc::UnboundedReceiver<WatchEvent>,
 
     /// map range to WatcherId
     watcher_range_map: RangeMap<String, WatcherId, WatcherStream>,
 
     current_watcher_id: WatcherId,
 }
 
-impl WatcherManager {
-    pub fn create() -> Self {
+impl EventDispatcher {
+    /// Spawn a dispatcher loop task.
+    pub(crate) fn spawn() -> mpsc::UnboundedSender<WatchEvent> {
         let (event_tx, event_rx) = mpsc::unbounded_channel();
 
-        let core = WatcherManagerCore {
+        let dispatcher = EventDispatcher {
             event_rx,
             watcher_range_map: RangeMap::new(),
             current_watcher_id: 1,
         };
 
-        let _h = tokio::spawn(core.watcher_manager_main());
+        let _h = tokio::spawn(dispatcher.main());
 
-        WatcherManager {
-            event_tx: event_tx.clone(),
-            subscriber: WatcherStateMachineSubscriber { event_tx },
-        }
-    }
-
-    pub fn create_watcher_stream(&self, request: WatchRequest, tx: WatcherStreamSender) {
-        let create: CreateWatcherEvent = (request, tx);
-        let _ = self.event_tx.send(WatcherEvent::CreateWatcherEvent(create));
+        event_tx
     }
-}
 
-impl WatcherManagerCore {
     #[tracing::instrument(level = "trace", skip(self))]
-    async fn watcher_manager_main(mut self) {
+    async fn main(mut self) {
         loop {
             if let Some(event) = self.event_rx.recv().await {
                 match event {
-                    WatcherEvent::CreateWatcherEvent((req, tx)) => {
-                        self.create_watcher_stream(req, tx).await;
+                    WatchEvent::AddWatcher((req, tx)) => {
+                        self.add_watcher(req, tx).await;
                     }
-                    WatcherEvent::StateMachineKvDataEvent(kv_change) => {
-                        self.notify_event(kv_change).await;
+                    WatchEvent::KVChange(kv_change) => {
+                        self.dispatch_event(kv_change).await;
                     }
                 }
             } else {
-                info!("watcher manager has been shutdown");
+                info!("all event senders are closed. quit.");
                 break;
             }
         }
@@ -117,7 +101,8 @@ impl WatcherManagerCore {
         server_metrics::incr_watchers(-1);
     }
 
-    async fn notify_event(&mut self, change: Change<Vec<u8>, String>) {
+    /// Dispatch a kv change event to interested watchers.
+    async fn dispatch_event(&mut self, change: Change<Vec<u8>, String>) {
         let k = change.ident.as_ref().unwrap();
         let set = self.watcher_range_map.get_by_point(k);
         if set.is_empty() {
@@ -165,23 +150,24 @@ impl WatcherManagerCore {
             };
         }
 
+        // TODO: when a watcher stream is dropped, send a event to remove the watcher explicitly
         for range_key in remove_range_keys {
             self.close_stream(range_key);
         }
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    pub async fn create_watcher_stream(&mut self, create: WatchRequest, tx: WatcherStreamSender) {
+    pub async fn add_watcher(&mut self, create: WatchRequest, tx: WatcherSender) {
         info!("create_watcher_stream: {:?}", create);
 
-        let range = match WatcherManagerCore::get_range_key(create.key.clone(), &create.key_end) {
+        let range = match EventDispatcher::get_range_key(create.key.clone(), &create.key_end) {
             Ok(range) => range,
             Err(_) => return,
         };
 
         self.current_watcher_id += 1;
         let watcher_id = self.current_watcher_id;
-        let filter = create.filter_type();
+        let filter: FilterType = create.filter_type();
 
         let watcher_stream = WatcherStream::new(
             watcher_id,
@@ -210,10 +196,8 @@ impl WatcherManagerCore {
     }
 }
 
-impl StateMachineSubscriber for WatcherStateMachineSubscriber {
+impl StateMachineSubscriber for DispatcherSender {
     fn kv_changed(&self, change: Change<Vec<u8>, String>) {
-        let _ = self
-            .event_tx
-            .send(WatcherEvent::StateMachineKvDataEvent(change));
+        let _ = self.0.send(WatchEvent::KVChange(change));
     }
 }

src/meta/service/src/watcher/watcher_manager.rs

@@ -18,14 +18,14 @@ use common_meta_types::protobuf::WatchResponse;
 use tonic::Status;
 
 use super::WatcherId;
-use super::WatcherStreamSender;
+use super::WatcherSender;
 
 pub struct WatcherStream {
     pub id: WatcherId,
 
     pub filter_type: FilterType,
 
-    tx: WatcherStreamSender,
+    tx: WatcherSender,
 
     pub key: String,
 
@@ -36,7 +36,7 @@ impl WatcherStream {
     pub fn new(
         id: WatcherId,
         filter_type: FilterType,
-        tx: WatcherStreamSender,
+        tx: WatcherSender,
         key: String,
         key_end: String,
     ) -> Self {

src/meta/service/src/watcher/watcher_stream.rs

@@ -21,7 +21,7 @@ use crate::plan::Projection;
 
 #[derive(serde::Serialize, serde::Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct PrewhereInfo {
-    /// columns to be ouput be prewhere scan
+    /// columns to be output by prewhere scan
     pub output_columns: Projection,
     /// columns used for prewhere
     pub prewhere_columns: Projection,

src/query/catalog/src/plan/pushdown.rs

@@ -64,14 +64,23 @@ impl StringColumn {
 
         if arrow_type == &ArrowType::Utf8 {
             let arr = array.as_any().downcast_ref::<Utf8Array<i32>>().unwrap();
-            let iter = arr.values_iter();
-            return Self::new_from_iter(iter);
+            let offsets = arr
+                .offsets()
+                .iter()
+                .map(|x| *x as i64)
+                .collect::<Buffer<_>>();
+            return Self {
+                offsets,
+                values: arr.values().clone(),
+            };
         }
 
         if arrow_type == &ArrowType::LargeUtf8 {
             let arr = array.as_any().downcast_ref::<Utf8Array<i64>>().unwrap();
-            let iter = arr.values_iter();
-            return Self::new_from_iter(iter);
+            return Self {
+                offsets: arr.offsets().clone(),
+                values: arr.values().clone(),
+            };
         }
 
         Self::new(

src/query/datavalues/src/columns/string/mod.rs

@@ -16,6 +16,8 @@ backon = "0.2"
 base64 = "0.13.0"
 bytes = "1"
 common-auth = { path = "../../common/auth" }
+common-config = { path = "../config" }
+
 http = "0.2"
 log = "0.4"
 moka = "0.9"