feat: add ordinary table batching workers

Signed-off-by: WenyXu <wenymedia@gmail.com>

src/servers/Cargo.toml

@@ -154,6 +154,7 @@ auth = { workspace = true, features = ["testing"] }
 catalog = { workspace = true, features = ["testing"] }
 client = { workspace = true, features = ["testing"] }
 common-base.workspace = true
+common-meta = { workspace = true, features = ["testing"] }
 common-test-util.workspace = true
 criterion = "0.5"
 json5 = "0.4"
@@ -161,6 +162,7 @@ moka.workspace = true
 mysql_async = { version = "0.37", default-features = false, features = [
     "default-rustls",
 ] }
+operator = { workspace = true, features = ["testing"] }
 permutation = "0.4"
 rand.workspace = true
 serde_json.workspace = true

src/servers/src/batcher.rs

@@ -0,0 +1,22 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Ordinary-table batching and shared notification components.
+
+mod flow_notifier;
+mod flow_sender;
+pub mod table;
+
+#[cfg(test)]
+mod test_util;

src/servers/src/batcher/flow_notifier.rs

@@ -0,0 +1,172 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::num::NonZeroUsize;
+
+use common_batcher::notifier::{Notifier, run_notifier};
+use common_meta::cache::TableFlownodeSetCacheRef;
+use common_meta::node_manager::NodeManagerRef;
+use common_runtime::spawn_global;
+use common_telemetry::{error, warn};
+use prometheus::IntCounterVec;
+use tokio::sync::mpsc::Receiver;
+use tokio::sync::mpsc::error::TrySendError;
+
+use crate::batcher::flow_sender::{FlowNotification, FlowSender};
+
+/// Best-effort queue admission and diagnostics, independent of payload extraction.
+#[derive(Clone)]
+pub(in crate::batcher) struct FlowNotifier {
+    notifier: Notifier<FlowNotification>,
+    dropped: IntCounterVec,
+}
+
+impl FlowNotifier {
+    /// Each construction creates an independent queue with caller-owned metrics.
+    pub fn try_new(
+        capacity: usize,
+        dropped: IntCounterVec,
+    ) -> Option<(Self, Receiver<FlowNotification>)> {
+        let (notifier, receiver) = Notifier::try_new(capacity)?;
+        Some((Self { notifier, dropped }, receiver))
+    }
+
+    /// Never waits for queue capacity or delivery and never changes the write result.
+    pub fn try_notify(&self, notification: FlowNotification) -> bool {
+        match self.notifier.try_notify(notification) {
+            Ok(()) => true,
+            Err(TrySendError::Full(notification)) => {
+                self.dropped.with_label_values(&["full"]).inc();
+                warn!(
+                    "Dropping flow notification because queue is full, table_id: {}, queue_capacity: {}",
+                    notification.table_id,
+                    self.notifier.max_capacity()
+                );
+                false
+            }
+            Err(TrySendError::Closed(notification)) => {
+                self.dropped.with_label_values(&["closed"]).inc();
+                error!(
+                    "Dropping flow notification because queue is closed, table_id: {}, queue_capacity: {}",
+                    notification.table_id,
+                    self.notifier.max_capacity()
+                );
+                false
+            }
+        }
+    }
+}
+
+/// Composes queue consumption and delivery without giving the notifier task ownership.
+pub(in crate::batcher) fn start_flow_notification_worker(
+    receiver: Receiver<FlowNotification>,
+    cache: TableFlownodeSetCacheRef,
+    node_manager: NodeManagerRef,
+) {
+    let sender = FlowSender::new(cache, node_manager);
+    let concurrency = NonZeroUsize::new(8).unwrap();
+    spawn_global(run_notifier(receiver, concurrency, move |notification| {
+        let sender = sender.clone();
+        async move { sender.send(notification).await }
+    }));
+}
+
+#[cfg(test)]
+mod tests {
+
+    use prometheus::{IntCounterVec, Opts};
+
+    use crate::batcher::flow_notifier::FlowNotifier;
+    use crate::batcher::flow_sender::FlowNotification;
+
+    fn dropped_counter() -> IntCounterVec {
+        // Unregistered collectors remain independent even with identical names.
+        IntCounterVec::new(
+            Opts::new("test_flow_dropped", "Dropped notifications"),
+            &["reason"],
+        )
+        .unwrap()
+    }
+
+    fn notification(table_id: u32) -> FlowNotification {
+        FlowNotification {
+            table_id,
+            timestamps: vec![-1, 42, 1_700_000_000_000_000_001, 42],
+        }
+    }
+
+    fn dropped_counts(counter: &IntCounterVec) -> (u64, u64) {
+        (
+            counter.with_label_values(&["full"]).get(),
+            counter.with_label_values(&["closed"]).get(),
+        )
+    }
+
+    #[test]
+    fn test_queues_and_metrics_are_independent() {
+        let first_dropped = dropped_counter();
+        let second_dropped = dropped_counter();
+        let (first, mut first_rx) = FlowNotifier::try_new(1, first_dropped.clone()).unwrap();
+        let (second, mut second_rx) = FlowNotifier::try_new(2, second_dropped.clone()).unwrap();
+
+        assert!(first.try_notify(notification(1)));
+        assert!(!first.try_notify(notification(2)));
+        assert_eq!(dropped_counts(&first_dropped), (1, 0));
+        assert_eq!(dropped_counts(&second_dropped), (0, 0));
+        assert!(second.try_notify(notification(3)));
+        assert!(second.try_notify(notification(4)));
+        assert!(!second.try_notify(notification(5)));
+        assert_eq!(dropped_counts(&second_dropped), (1, 0));
+
+        let payload = first_rx.try_recv().unwrap();
+        assert_eq!(payload.table_id, 1);
+        assert_eq!(payload.timestamps, notification(1).timestamps);
+        assert!(first_rx.try_recv().is_err());
+        for table_id in [3, 4] {
+            let payload = second_rx.try_recv().unwrap();
+            assert_eq!(payload.table_id, table_id);
+            assert_eq!(payload.timestamps, notification(table_id).timestamps);
+        }
+        assert!(second_rx.try_recv().is_err());
+
+        drop(first_rx);
+        assert!(!first.try_notify(notification(6)));
+        assert_eq!(dropped_counts(&first_dropped), (1, 1));
+        assert_eq!(dropped_counts(&second_dropped), (1, 0));
+        assert!(second.try_notify(notification(7)));
+        assert_eq!(second_rx.try_recv().unwrap().table_id, 7);
+        drop(second_rx);
+        assert!(!second.try_notify(notification(8)));
+        assert_eq!(dropped_counts(&second_dropped), (1, 1));
+    }
+
+    #[test]
+    fn test_clones_share_queue_and_drop_metrics() {
+        let dropped = dropped_counter();
+        let (notifier, mut receiver) = FlowNotifier::try_new(1, dropped.clone()).unwrap();
+        let clone = notifier.clone();
+        assert!(notifier.try_notify(notification(1)));
+        assert!(!clone.try_notify(notification(2)));
+        assert_eq!(dropped_counts(&dropped), (1, 0));
+        assert_eq!(receiver.try_recv().unwrap().table_id, 1);
+        drop(notifier);
+        assert!(clone.try_notify(notification(3)));
+        let payload = receiver.try_recv().unwrap();
+        assert_eq!(payload.table_id, 3);
+        assert_eq!(payload.timestamps, notification(3).timestamps);
+        drop(receiver);
+        assert!(!clone.try_notify(notification(4)));
+        assert_eq!(dropped_counts(&dropped), (1, 1));
+    }
+}

src/servers/src/batcher/flow_sender.rs

@@ -0,0 +1,212 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::collections::HashSet;
+
+use api::v1::flow::{DirtyWindowRequest, DirtyWindowRequests};
+use common_meta::cache::TableFlownodeSetCacheRef;
+use common_meta::node_manager::NodeManagerRef;
+use common_telemetry::error;
+use store_api::storage::TableId;
+
+/// One source table's timestamps, expressed in its native time unit.
+pub(in crate::batcher) struct FlowNotification {
+    pub table_id: TableId,
+    pub timestamps: Vec<i64>,
+}
+
+/// Resolves Flow targets and delivers dirty-window notifications best-effort.
+/// Queue admission, delivery concurrency and write completion belong to callers.
+#[derive(Clone)]
+pub(in crate::batcher) struct FlowSender {
+    cache: TableFlownodeSetCacheRef,
+    node_manager: NodeManagerRef,
+}
+
+impl FlowSender {
+    /// Binds delivery dependencies without creating a queue or starting a task.
+    pub fn new(cache: TableFlownodeSetCacheRef, node_manager: NodeManagerRef) -> Self {
+        Self {
+            cache,
+            node_manager,
+        }
+    }
+
+    /// Sends once per distinct peer, without retrying or failing the completed write.
+    /// Peer RPCs stay within the caller's notification concurrency budget.
+    pub async fn send(&self, notification: FlowNotification) {
+        let table_id = notification.table_id;
+        let flownodes = match self.cache.get(table_id).await {
+            Ok(Some(flownodes)) => flownodes,
+            Ok(None) => return,
+            Err(e) => {
+                error!(e; "Failed to get flownodes for table id: {}", table_id);
+                return;
+            }
+        };
+        let peers = flownodes.values().cloned().collect::<HashSet<_>>();
+
+        for peer in peers {
+            if let Err(e) = self
+                .node_manager
+                .flownode(&peer)
+                .await
+                .handle_mark_window_dirty(DirtyWindowRequests {
+                    requests: vec![DirtyWindowRequest {
+                        table_id,
+                        timestamps: notification.timestamps.clone(),
+                        time_ranges: Vec::new(),
+                    }],
+                })
+                .await
+            {
+                error!(
+                    e;
+                    "Failed to mark timestamps as dirty, table_id: {}, peer_id: {}, peer_addr: {}",
+                    table_id,
+                    peer.id,
+                    peer.addr
+                );
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Mutex};
+
+    use api::v1::flow::{DirtyWindowRequest, DirtyWindowRequests, FlowRequest, FlowResponse};
+    use api::v1::region::InsertRequests;
+    use async_trait::async_trait;
+    use common_meta::error::{Result as MetaResult, UnexpectedSnafu};
+    use common_meta::node_manager::{
+        DatanodeManager, DatanodeRef, Flownode, FlownodeManager, FlownodeRef,
+    };
+    use common_meta::peer::Peer;
+
+    use crate::batcher::flow_sender::{FlowNotification, FlowSender};
+    use crate::batcher::test_util::mock_table_flownode_cache;
+
+    #[derive(Default)]
+    struct RecordingNodeManager {
+        requests: Arc<Mutex<Vec<(u64, DirtyWindowRequests)>>>,
+        fail_first: bool,
+    }
+
+    struct RecordingFlownode {
+        peer_id: u64,
+        requests: Arc<Mutex<Vec<(u64, DirtyWindowRequests)>>>,
+        fail_first: bool,
+    }
+
+    #[async_trait]
+    impl Flownode for RecordingFlownode {
+        async fn handle(&self, _: FlowRequest) -> MetaResult<FlowResponse> {
+            unreachable!("notifications must use the dirty-window RPC")
+        }
+
+        async fn handle_inserts(&self, _: InsertRequests) -> MetaResult<FlowResponse> {
+            unreachable!("notifications must not mirror row inserts")
+        }
+
+        async fn handle_mark_window_dirty(
+            &self,
+            request: DirtyWindowRequests,
+        ) -> MetaResult<FlowResponse> {
+            let mut requests = self.requests.lock().unwrap();
+            requests.push((self.peer_id, request));
+            if self.fail_first && requests.len() == 1 {
+                return UnexpectedSnafu {
+                    err_msg: "injected first notification failure",
+                }
+                .fail();
+            }
+            Ok(FlowResponse::default())
+        }
+    }
+
+    #[async_trait]
+    impl DatanodeManager for RecordingNodeManager {
+        async fn datanode(&self, _: &Peer) -> DatanodeRef {
+            unreachable!("flow notifications must not contact datanodes")
+        }
+    }
+
+    #[async_trait]
+    impl FlownodeManager for RecordingNodeManager {
+        async fn flownode(&self, peer: &Peer) -> FlownodeRef {
+            Arc::new(RecordingFlownode {
+                peer_id: peer.id,
+                requests: self.requests.clone(),
+                fail_first: self.fail_first,
+            })
+        }
+    }
+
+    #[tokio::test]
+    async fn test_deduplicated_delivery_preserves_payload_and_continues_after_error() {
+        // HashSet peer order is unspecified, so fail the first attempted RPC rather
+        // than a chosen peer. The second peer must still receive the notification.
+        for fail_first in [false, true] {
+            let first = Peer {
+                id: 1,
+                addr: "flow-1".to_string(),
+            };
+            let second = Peer {
+                id: 2,
+                addr: "flow-2".to_string(),
+            };
+            let cache =
+                mock_table_flownode_cache(42, vec![(0, first.clone()), (1, first), (2, second)])
+                    .await;
+            let manager = Arc::new(RecordingNodeManager {
+                fail_first,
+                ..Default::default()
+            });
+            let sender = FlowSender::new(cache, manager.clone());
+            let timestamps = vec![-1, 42, 1_700_000_000_000_000_001, 42];
+            sender
+                .send(FlowNotification {
+                    table_id: 42,
+                    timestamps: timestamps.clone(),
+                })
+                .await;
+            let expected = DirtyWindowRequests {
+                requests: vec![DirtyWindowRequest {
+                    table_id: 42,
+                    timestamps,
+                    time_ranges: vec![],
+                }],
+            };
+            let mut actual = manager.requests.lock().unwrap().clone();
+            actual.sort_by_key(|(peer, _)| *peer);
+            assert_eq!(actual, vec![(1, expected.clone()), (2, expected)]);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_missing_targets_do_not_send() {
+        let cache = mock_table_flownode_cache(42, vec![]).await;
+        let manager = Arc::new(RecordingNodeManager::default());
+        let sender = FlowSender::new(cache, manager.clone());
+        sender
+            .send(FlowNotification {
+                table_id: 42,
+                timestamps: vec![42],
+            })
+            .await;
+        assert!(manager.requests.lock().unwrap().is_empty());
+    }
+}

src/servers/src/batcher/table.rs

@@ -0,0 +1,803 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Table-scoped RecordBatch accumulation before bulk routing and encoding.
+
+mod batch;
+mod flow_notifier;
+mod metrics;
+mod pending_batch;
+mod pending_worker;
+
+use std::num::NonZeroUsize;
+use std::sync::Arc;
+use std::time::Duration;
+
+use arrow::record_batch::RecordBatch;
+use async_trait::async_trait;
+use common_batcher::flush_limiter::FlushLimiter;
+use common_batcher::flush_policy::timing::TimingFlushPolicy;
+use common_batcher::request_limiter::RequestLimiter;
+use common_batcher::worker_registry::WorkerRegistry;
+use operator::batcher::PendingRowsBatcher;
+use operator::error::{BatchFlushSnafu, Result, UnexpectedSnafu};
+use operator::insert::Inserter;
+use session::context::QueryContextRef;
+use snafu::ResultExt;
+use table::metadata::TableInfoRef;
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast, oneshot};
+
+use crate::batcher::table::flow_notifier::FlowNotifier;
+use crate::batcher::table::metrics::PENDING_WORKERS;
+use crate::batcher::table::pending_batch::PendingBatch;
+use crate::batcher::table::pending_worker::{PendingWorker, WorkerCommand, start_worker};
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq)]
+struct BatchKey {
+    catalog: String,
+    schema: String,
+    table_name: String,
+    skip_wal: bool,
+}
+
+// Match Prom batching: group by table name and keep WAL policies separate.
+fn batch_key_from_ctx(table_name: &str, ctx: &QueryContextRef) -> BatchKey {
+    BatchKey {
+        catalog: ctx.current_catalog().to_string(),
+        schema: ctx.current_schema().clone(),
+        table_name: table_name.to_string(),
+        skip_wal: ctx.skip_wal(),
+    }
+}
+
+type PendingWorkers = WorkerRegistry<BatchKey, WorkerCommand>;
+
+/// Admission and worker lookup for table-scoped bulk writes.
+pub struct TablePendingRowsBatcher {
+    workers: Arc<PendingWorkers>,
+    flush_policy: TimingFlushPolicy,
+    flush_limiter: FlushLimiter,
+    request_limiter: RequestLimiter,
+    worker_channel_capacity: usize,
+    worker_idle_timeout: Duration,
+    inserter: Arc<Inserter>,
+    flow_notifier: FlowNotifier,
+    shutdown: broadcast::Sender<()>,
+}
+
+impl TablePendingRowsBatcher {
+    /// Creates a shared timing batcher, rejecting disabled or unsupported limits.
+    pub fn try_new(
+        flush_interval: Duration,
+        max_batch_rows: usize,
+        max_concurrent_flushes: usize,
+        worker_channel_capacity: usize,
+        max_inflight_requests: usize,
+        flow_notification_queue_capacity: NonZeroUsize,
+        inserter: Arc<Inserter>,
+    ) -> Option<Arc<Self>> {
+        if worker_channel_capacity == 0
+            || worker_channel_capacity > Semaphore::MAX_PERMITS
+            || max_inflight_requests == 0
+            || max_inflight_requests > Semaphore::MAX_PERMITS
+        {
+            return None;
+        }
+        let flush_policy = TimingFlushPolicy::try_new(flush_interval, max_batch_rows)?;
+        let flush_limiter = FlushLimiter::try_new(max_concurrent_flushes)?;
+        let request_limiter = RequestLimiter::try_new(max_inflight_requests)?;
+        let flow_notifier = FlowNotifier::new(
+            inserter.table_flownode_set_cache().clone(),
+            inserter.node_manager().clone(),
+            flow_notification_queue_capacity,
+        )?;
+        let (shutdown, _) = broadcast::channel(1);
+        PENDING_WORKERS.set(0);
+        Some(Arc::new(Self {
+            workers: Arc::new(WorkerRegistry::new()),
+            flush_policy,
+            flush_limiter,
+            request_limiter,
+            worker_channel_capacity,
+            worker_idle_timeout: flush_interval.checked_mul(3).unwrap_or(flush_interval),
+            inserter,
+            flow_notifier,
+            shutdown,
+        }))
+    }
+
+    async fn worker(&self, key: &BatchKey) -> PendingWorker {
+        let (tx, receiver) = self
+            .workers
+            .get_or_create(key.clone(), self.worker_channel_capacity)
+            .await;
+        if let Some(rx) = receiver {
+            start_worker(
+                key.clone(),
+                tx.clone(),
+                self.workers.clone(),
+                rx,
+                self.shutdown.clone(),
+                self.flush_policy,
+                self.flush_limiter.clone(),
+                self.inserter.clone(),
+                self.flow_notifier.clone(),
+                self.worker_idle_timeout,
+            );
+            PENDING_WORKERS.set(self.workers.len().await as i64);
+        }
+        PendingWorker { tx }
+    }
+}
+
+impl Drop for TablePendingRowsBatcher {
+    fn drop(&mut self) {
+        let _ = self.shutdown.send(());
+    }
+}
+
+#[async_trait]
+impl PendingRowsBatcher for TablePendingRowsBatcher {
+    /// Acquires once per original request, before submitting any of its tables.
+    /// Retain a clone in every submission until its flush has completed.
+    async fn acquire(&self) -> Result<Arc<OwnedSemaphorePermit>> {
+        self.request_limiter.acquire().await.map_err(|_| {
+            UnexpectedSnafu {
+                violated: "batcher admission closed".to_string(),
+            }
+            .build()
+        })
+    }
+
+    /// Waits for completed bulk writes. Cancellation does not retract an admitted
+    /// submission. Combined failures affect all waiters and may be partial writes.
+    async fn submit(
+        &self,
+        table_info: TableInfoRef,
+        batch: RecordBatch,
+        ctx: QueryContextRef,
+        permit: Arc<OwnedSemaphorePermit>,
+    ) -> Result<usize> {
+        let total_rows = batch.num_rows();
+        if total_rows == 0 {
+            return Ok(0);
+        }
+        if table_info.catalog_name != ctx.current_catalog()
+            || table_info.schema_name != ctx.current_schema()
+        {
+            return UnexpectedSnafu {
+                violated: "batch table and request database differ".to_string(),
+            }
+            .fail();
+        }
+        let key = batch_key_from_ctx(&table_info.name, &ctx);
+        let (response_tx, response_rx) = oneshot::channel();
+        let pending = PendingBatch {
+            table_info,
+            batch,
+            ctx,
+            response_tx,
+            _permit: permit,
+        };
+        let mut command = Some(WorkerCommand::Submit(pending));
+        for _ in 0..2 {
+            let worker = self.worker(&key).await;
+            let Some(pending) = command.take() else { break };
+            match worker.tx.send(pending).await {
+                Ok(()) => break,
+                Err(error) => {
+                    command = Some(error.0);
+                    if self.workers.remove_if_same(&key, &worker.tx).await {
+                        PENDING_WORKERS.set(self.workers.len().await as i64);
+                    }
+                }
+            }
+        }
+        if command.is_some() {
+            return UnexpectedSnafu {
+                violated: "batch worker channel closed".to_string(),
+            }
+            .fail();
+        }
+        response_rx
+            .await
+            .map_err(|_| {
+                UnexpectedSnafu {
+                    violated: "batch worker stopped before reporting its write result".to_string(),
+                }
+                .build()
+            })?
+            .context(BatchFlushSnafu)?;
+        Ok(total_rows)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::num::NonZeroUsize;
+    use std::sync::{Arc, Mutex};
+    use std::time::Duration;
+
+    use api::region::RegionResponse;
+    use api::v1::helper::{tag_column_schema, time_index_column_schema};
+    use api::v1::region::{RegionRequest, bulk_insert_request, region_request};
+    use api::v1::value::ValueData;
+    use api::v1::{ColumnDataType, Row, Rows, Value};
+    use arrow::array::{Int32Array, TimestampMillisecondArray};
+    use arrow::datatypes::Schema as ArrowSchema;
+    use arrow::record_batch::RecordBatch;
+    use common_batcher::flush_limiter::FlushLimiter;
+    use common_batcher::flush_policy::timing::TimingFlushPolicy;
+    use common_batcher::request_limiter::RequestLimiter;
+    use common_batcher::worker_registry::WorkerRegistry;
+    use common_grpc::flight::FlightDecoder;
+    use common_meta::error::Result as MetaResult;
+    use common_meta::peer::Peer;
+    use common_meta::test_util::{MockDatanodeHandler, MockDatanodeManager};
+    use common_query::request::QueryRequest;
+    use common_recordbatch::SendableRecordBatchStream;
+    use common_telemetry::info;
+    use datatypes::schema::{ColumnDefaultConstraint, SchemaBuilder};
+    use datatypes::value::Value as DtValue;
+    use operator::batcher::PendingRowsBatcher;
+    use operator::error::Error;
+    use operator::insert::Inserter;
+    use operator::req_convert::insert::rows_to_record_batch;
+    use operator::test_util::{
+        create_partition_rule_manager, new_test_table_info, prepare_mocked_backend,
+    };
+    use session::context::{Channel, QueryContext};
+    use store_api::storage::RegionId;
+    use table::metadata::TableInfoRef;
+    use tokio::sync::{broadcast, mpsc, oneshot};
+    use tokio::time::timeout;
+
+    use crate::batcher::table::flow_notifier::FlowNotifier;
+    use crate::batcher::table::pending_batch::PendingBatch;
+    use crate::batcher::table::pending_worker::{WorkerCommand, start_worker};
+    use crate::batcher::table::{PendingWorkers, TablePendingRowsBatcher, batch_key_from_ctx};
+    use crate::batcher::test_util::mock_table_flownode_cache;
+
+    #[derive(Clone)]
+    struct BulkHandler {
+        requests: Arc<Mutex<Vec<RecordBatch>>>,
+        report_missing_row: bool,
+        expected_skip_wal: bool,
+    }
+
+    #[async_trait::async_trait]
+    impl MockDatanodeHandler for BulkHandler {
+        async fn handle(&self, peer: &Peer, request: RegionRequest) -> MetaResult<RegionResponse> {
+            assert_eq!(3, peer.id);
+            assert_eq!("public", request.header.unwrap().dbname);
+            let Some(region_request::Body::BulkInsert(request)) = request.body else {
+                panic!("expected bulk insert")
+            };
+            assert_eq!(RegionId::new(1, 3).as_u64(), request.region_id);
+            assert_eq!(self.expected_skip_wal, request.skip_wal);
+            assert!(request.partition_expr_version.is_some());
+            let Some(bulk_insert_request::Body::ArrowIpc(ipc)) = request.body else {
+                panic!("expected Arrow IPC")
+            };
+            let batch = FlightDecoder::try_from_schema_bytes(&ipc.schema)
+                .unwrap()
+                .try_decode_record_batch(&ipc.data_header, &ipc.payload)
+                .unwrap();
+            let rows = batch.num_rows();
+            self.requests.lock().unwrap().push(batch);
+            Ok(RegionResponse::new(
+                rows - usize::from(self.report_missing_row),
+            ))
+        }
+
+        async fn handle_query(
+            &self,
+            _: &Peer,
+            _: QueryRequest,
+        ) -> MetaResult<SendableRecordBatchStream> {
+            panic!("batching must not query the datanode")
+        }
+    }
+
+    fn rows(value: i32) -> Rows {
+        Rows {
+            schema: vec![
+                tag_column_schema("a", ColumnDataType::Int32),
+                time_index_column_schema("ts", ColumnDataType::TimestampMillisecond),
+            ],
+            rows: vec![Row {
+                values: vec![
+                    Value {
+                        value_data: Some(ValueData::I32Value(value)),
+                    },
+                    Value {
+                        value_data: Some(ValueData::TimestampMillisecondValue(
+                            1000 + i64::from(value),
+                        )),
+                    },
+                ],
+            }],
+        }
+    }
+
+    async fn run_bulk_case(
+        max_batch_rows: usize,
+        report_missing_row: bool,
+        shared_request: bool,
+    ) -> usize {
+        run_bulk_case_with_wal(max_batch_rows, report_missing_row, shared_request, false).await
+    }
+
+    async fn run_bulk_case_with_wal(
+        max_batch_rows: usize,
+        report_missing_row: bool,
+        shared_request: bool,
+        skip_wal: bool,
+    ) -> usize {
+        let backend = prepare_mocked_backend().await;
+        let partitions = create_partition_rule_manager(backend.clone()).await;
+        let captured = Arc::new(Mutex::new(Vec::new()));
+        let nodes = Arc::new(MockDatanodeManager::new(BulkHandler {
+            requests: captured.clone(),
+            report_missing_row,
+            expected_skip_wal: skip_wal,
+        }));
+        let inserter = Arc::new(Inserter::new(
+            catalog::memory::MemoryCatalogManager::new(),
+            partitions,
+            nodes,
+            mock_table_flownode_cache(1, vec![]).await,
+            true,
+        ));
+        let mut table = new_test_table_info(1, "table_1", [1, 2, 3].into_iter());
+        let mut columns = table.meta.schema.column_schemas().to_vec();
+        columns[2] = columns[2]
+            .clone()
+            .with_default_constraint(Some(ColumnDefaultConstraint::Value(DtValue::Int32(7))))
+            .unwrap();
+        table.meta.schema = Arc::new(
+            SchemaBuilder::try_from(columns)
+                .unwrap()
+                .version(123)
+                .build()
+                .unwrap(),
+        );
+        let table = Arc::new(table);
+        let first = rows_to_record_batch(&rows(1), &table).unwrap();
+        let second = rows_to_record_batch(&rows(2), &table).unwrap();
+        // Keep timer/concurrency/workload fixed; only the supplied row threshold
+        // determines whether these two submissions share a bulk request.
+        let batcher: Arc<dyn PendingRowsBatcher> = TablePendingRowsBatcher::try_new(
+            Duration::from_secs(3600),
+            max_batch_rows,
+            1,
+            4,
+            if shared_request { 1 } else { 4 },
+            NonZeroUsize::new(16).unwrap(),
+            inserter,
+        )
+        .unwrap();
+        let influx_ctx = Arc::new(QueryContext::with_channel(
+            "greptime",
+            "public",
+            Channel::Influx,
+        ));
+        let opentsdb_ctx = Arc::new(QueryContext::with_channel(
+            "greptime",
+            "public",
+            Channel::Opentsdb,
+        ));
+        influx_ctx.set_skip_wal(skip_wal);
+        opentsdb_ctx.set_skip_wal(skip_wal);
+        let first_permit = batcher.acquire().await.unwrap();
+        let second_permit = if shared_request {
+            first_permit.clone()
+        } else {
+            batcher.acquire().await.unwrap()
+        };
+        let (first_result, second_result) = timeout(Duration::from_secs(5), async {
+            tokio::join!(
+                batcher.submit(table.clone(), first, influx_ctx, first_permit),
+                batcher.submit(table.clone(), second, opentsdb_ctx, second_permit),
+            )
+        })
+        .await
+        .expect("the row threshold did not dispatch the combined bulk insert");
+        if report_missing_row {
+            assert!(first_result.is_err());
+            assert!(second_result.is_err());
+        } else {
+            assert_eq!(1, first_result.unwrap());
+            assert_eq!(1, second_result.unwrap());
+        }
+        let requests = captured.lock().unwrap();
+        let mut actual_rows = Vec::new();
+        for batch in requests.iter() {
+            assert_eq!(
+                vec!["a", "ts", "b"],
+                batch
+                    .schema()
+                    .fields()
+                    .iter()
+                    .map(|field| field.name().as_str())
+                    .collect::<Vec<_>>()
+            );
+            let a = batch
+                .column(0)
+                .as_any()
+                .downcast_ref::<Int32Array>()
+                .unwrap();
+            let ts = batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<TimestampMillisecondArray>()
+                .unwrap();
+            let b = batch
+                .column(2)
+                .as_any()
+                .downcast_ref::<Int32Array>()
+                .unwrap();
+            actual_rows.extend(
+                (0..batch.num_rows())
+                    .map(|index| (a.value(index), ts.value(index), b.value(index))),
+            );
+        }
+        actual_rows.sort_unstable();
+        assert_eq!(vec![(1, 1001, 7), (2, 1002, 7)], actual_rows);
+        requests.len()
+    }
+
+    #[tokio::test]
+    async fn test_submit_combines_then_routes_bulk_with_defaults() {
+        for report_missing_row in [false, true] {
+            assert_eq!(1, run_bulk_case(2, report_missing_row, false).await);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_row_threshold_ablation_reduces_bulk_requests() {
+        for repetition in 0..2 {
+            let direct_count = run_bulk_case(1, false, false).await;
+            let combined_count = run_bulk_case(2, false, false).await;
+            assert_eq!(2, direct_count);
+            assert_eq!(1, combined_count);
+            info!(
+                "repetition={repetition} rows=2 threshold=1 bulk_requests={direct_count}; threshold=2 bulk_requests={combined_count}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_original_request_shares_one_admission_slot() {
+        assert_eq!(1, run_bulk_case(2, false, true).await);
+    }
+
+    #[tokio::test]
+    async fn test_bulk_preserves_skip_wal() {
+        for skip_wal in [false, true] {
+            assert_eq!(1, run_bulk_case_with_wal(2, false, false, skip_wal).await);
+        }
+    }
+
+    #[test]
+    fn test_batch_key_uses_name_and_wal_policy() {
+        let ctx = QueryContext::arc();
+        let table = new_test_table_info(1, "table_1", [1, 2, 3].into_iter());
+        let mut replacement = table.clone();
+        replacement.ident.table_id = 2;
+        replacement.ident.version += 1;
+        let key = batch_key_from_ctx(&table.name, &ctx);
+        assert_eq!(key, batch_key_from_ctx(&replacement.name, &ctx));
+        assert_ne!(key, batch_key_from_ctx("table_2", &ctx));
+        for (catalog, schema) in [("other", "public"), ("greptime", "other")] {
+            let other = Arc::new(QueryContext::with_channel(catalog, schema, Channel::Influx));
+            assert_ne!(key, batch_key_from_ctx(&table.name, &other));
+        }
+        ctx.set_skip_wal(true);
+        assert_ne!(key, batch_key_from_ctx(&table.name, &ctx));
+    }
+
+    const WORKER_TIMEOUT: Duration = Duration::from_secs(5);
+
+    #[derive(Clone)]
+    struct GatedHandler {
+        inner: BulkHandler,
+        entered: mpsc::UnboundedSender<oneshot::Sender<()>>,
+    }
+
+    #[async_trait::async_trait]
+    impl MockDatanodeHandler for GatedHandler {
+        async fn handle(&self, peer: &Peer, request: RegionRequest) -> MetaResult<RegionResponse> {
+            let response = self.inner.handle(peer, request).await?;
+            let (release_tx, release_rx) = oneshot::channel();
+            self.entered.send(release_tx).unwrap();
+            release_rx.await.unwrap();
+            Ok(response)
+        }
+
+        async fn handle_query(
+            &self,
+            _: &Peer,
+            _: QueryRequest,
+        ) -> MetaResult<SendableRecordBatchStream> {
+            panic!("worker must not query the datanode")
+        }
+    }
+
+    struct WorkerTest {
+        tx: mpsc::Sender<WorkerCommand>,
+        shutdown: broadcast::Sender<()>,
+        workers: Arc<PendingWorkers>,
+        table: TableInfoRef,
+        entered: mpsc::UnboundedReceiver<oneshot::Sender<()>>,
+    }
+
+    impl Drop for WorkerTest {
+        fn drop(&mut self) {
+            let _ = self.shutdown.send(());
+        }
+    }
+
+    impl WorkerTest {
+        async fn new(max_rows: usize, max_flushes: usize) -> Self {
+            let backend = prepare_mocked_backend().await;
+            let partitions = create_partition_rule_manager(backend.clone()).await;
+            let (entered_tx, entered) = mpsc::unbounded_channel();
+            let nodes = Arc::new(MockDatanodeManager::new(GatedHandler {
+                inner: BulkHandler {
+                    requests: Arc::new(Mutex::new(Vec::new())),
+                    report_missing_row: false,
+                    expected_skip_wal: false,
+                },
+                entered: entered_tx,
+            }));
+            let cache = mock_table_flownode_cache(1, vec![]).await;
+            let notifier =
+                FlowNotifier::new(cache.clone(), nodes.clone(), NonZeroUsize::new(16).unwrap())
+                    .unwrap();
+            let inserter = Arc::new(Inserter::new(
+                catalog::memory::MemoryCatalogManager::new(),
+                partitions,
+                nodes,
+                cache,
+                true,
+            ));
+            let table = Arc::new(new_test_table_info(1, "table_1", [1, 2, 3].into_iter()));
+            let key = batch_key_from_ctx(&table.name, &QueryContext::arc());
+            let workers = Arc::new(WorkerRegistry::new());
+            let (tx, rx) = mpsc::channel(1);
+            workers.get_or_insert_with(key.clone(), || tx.clone()).await;
+            let (shutdown, _) = broadcast::channel(1);
+            start_worker(
+                key,
+                tx.clone(),
+                workers.clone(),
+                rx,
+                shutdown.clone(),
+                TimingFlushPolicy::try_new(Duration::from_secs(3600), max_rows).unwrap(),
+                FlushLimiter::try_new(max_flushes).unwrap(),
+                inserter,
+                notifier,
+                Duration::from_secs(10800),
+            );
+            Self {
+                tx,
+                shutdown,
+                workers,
+                table,
+                entered,
+            }
+        }
+
+        async fn submit(
+            &self,
+            count: usize,
+            changed_schema: bool,
+        ) -> oneshot::Receiver<Result<(), Arc<Error>>> {
+            self.submit_with_table(count, changed_schema, self.table.clone())
+                .await
+        }
+
+        async fn submit_with_table(
+            &self,
+            count: usize,
+            changed_schema: bool,
+            table_info: TableInfoRef,
+        ) -> oneshot::Receiver<Result<(), Arc<Error>>> {
+            let mut input = rows(1);
+            input.rows = vec![input.rows[0].clone(); count];
+            let mut batch = rows_to_record_batch(&input, &self.table).unwrap();
+            if changed_schema {
+                let schema = ArrowSchema::new_with_metadata(
+                    batch.schema().fields().clone(),
+                    [("test_version".to_string(), "2".to_string())]
+                        .into_iter()
+                        .collect(),
+                );
+                batch = RecordBatch::try_new(Arc::new(schema), batch.columns().to_vec()).unwrap();
+            }
+            let limiter = RequestLimiter::try_new(1).unwrap();
+            let (response_tx, response_rx) = oneshot::channel();
+            self.tx
+                .send(WorkerCommand::Submit(PendingBatch {
+                    table_info,
+                    batch,
+                    ctx: QueryContext::arc(),
+                    response_tx,
+                    _permit: limiter.acquire().await.unwrap(),
+                }))
+                .await
+                .unwrap();
+            // With capacity one, this proves the submitted command was dequeued.
+            drop(
+                timeout(WORKER_TIMEOUT, self.tx.reserve())
+                    .await
+                    .unwrap()
+                    .unwrap(),
+            );
+            response_rx
+        }
+
+        async fn entered(&mut self) -> oneshot::Sender<()> {
+            timeout(WORKER_TIMEOUT, self.entered.recv())
+                .await
+                .expect("flush did not reach datanode")
+                .unwrap()
+        }
+
+        async fn stopped(&self) {
+            timeout(WORKER_TIMEOUT, async {
+                while !self.workers.is_empty().await {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("worker did not remove its registry entry");
+        }
+    }
+
+    #[tokio::test]
+    async fn test_same_schema_flushes_overlap() {
+        let mut worker = WorkerTest::new(1, 2).await;
+        let first = worker.submit(1, false).await;
+        let first_release = worker.entered().await;
+        let second = worker.submit(1, false).await;
+        let second_release = worker.entered().await;
+        // Both RPCs reached the datanode before either was allowed to complete.
+        first_release.send(()).unwrap();
+        second_release.send(()).unwrap();
+        timeout(WORKER_TIMEOUT, first)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        timeout(WORKER_TIMEOUT, second)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        worker.shutdown.send(()).unwrap();
+        worker.stopped().await;
+    }
+
+    #[tokio::test]
+    async fn test_schema_change_waits_for_prior_flush() {
+        let mut worker = WorkerTest::new(1, 2).await;
+        let first = worker.submit(1, false).await;
+        let first_release = worker.entered().await;
+        let second = worker.submit(1, true).await;
+        // A second flush permit is free, but the schema barrier must withhold the RPC.
+        assert!(
+            timeout(Duration::from_millis(100), worker.entered.recv())
+                .await
+                .is_err()
+        );
+        first_release.send(()).unwrap();
+        let second_release = worker.entered().await;
+        second_release.send(()).unwrap();
+        timeout(WORKER_TIMEOUT, first)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        timeout(WORKER_TIMEOUT, second)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        worker.shutdown.send(()).unwrap();
+        worker.stopped().await;
+    }
+
+    #[tokio::test]
+    async fn test_table_identity_change_flushes_pending_rows_before_switching() {
+        for recreated in [false, true] {
+            let mut worker = WorkerTest::new(2, 2).await;
+            let first = worker.submit(1, false).await;
+            let mut changed = (*worker.table).clone();
+            if recreated {
+                changed.ident.table_id = 2;
+            } else {
+                changed.ident.version += 1;
+            }
+            let mut second = worker.submit_with_table(2, false, Arc::new(changed)).await;
+            // The first batch is below max_rows but must be flushed on identity change.
+            let first_release = worker.entered().await;
+            assert!(
+                timeout(Duration::from_millis(100), worker.entered.recv())
+                    .await
+                    .is_err()
+            );
+            assert!(
+                timeout(Duration::from_millis(100), &mut second)
+                    .await
+                    .is_err()
+            );
+            first_release.send(()).unwrap();
+            timeout(WORKER_TIMEOUT, first)
+                .await
+                .unwrap()
+                .unwrap()
+                .unwrap();
+            if recreated {
+                // The recreated table has no route in this fixture; its error belongs only to it.
+                assert!(
+                    timeout(WORKER_TIMEOUT, second)
+                        .await
+                        .unwrap()
+                        .unwrap()
+                        .is_err()
+                );
+            } else {
+                let second_release = worker.entered().await;
+                second_release.send(()).unwrap();
+                timeout(WORKER_TIMEOUT, second)
+                    .await
+                    .unwrap()
+                    .unwrap()
+                    .unwrap();
+            }
+            worker.shutdown.send(()).unwrap();
+            worker.stopped().await;
+        }
+    }
+
+    #[tokio::test]
+    async fn test_shutdown_flushes_inline_without_joining_inflight() {
+        let mut worker = WorkerTest::new(2, 1).await;
+        let mut first = worker.submit(2, false).await;
+        let first_release = worker.entered().await;
+        let pending = worker.submit(1, false).await;
+        worker.shutdown.send(()).unwrap();
+        // The normal flush owns the sole permit. Shutdown must still start pending inline.
+        let pending_release = worker.entered().await;
+        pending_release.send(()).unwrap();
+        timeout(WORKER_TIMEOUT, pending)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        worker.stopped().await;
+        assert!(matches!(
+            first.try_recv(),
+            Err(oneshot::error::TryRecvError::Empty)
+        ));
+        first_release.send(()).unwrap();
+        timeout(WORKER_TIMEOUT, first)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+    }
+}

src/servers/src/batcher/table/batch.rs

@@ -0,0 +1,150 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+use std::time::Instant;
+
+use arrow::compute::concat_batches;
+use arrow::record_batch::RecordBatch;
+use operator::error::{ComputeArrowSnafu, Result, UnexpectedSnafu};
+use operator::insert::Inserter;
+use snafu::ResultExt;
+use table::metadata::TableInfoRef;
+
+use crate::batcher::table::flow_notifier::FlowNotifier;
+use crate::batcher::table::metrics::{
+    FLUSH_DROPPED_ROWS, FLUSH_ELAPSED, FLUSH_FAILURES, FLUSH_ROWS, FLUSH_TOTAL,
+};
+use crate::batcher::table::pending_batch::{PendingBatch, notify_batches};
+
+/// A detached batch owns every submission until its completion is reported.
+pub(in crate::batcher::table) struct Batch {
+    pub submissions: Vec<PendingBatch>,
+    pub total_rows: usize,
+}
+
+pub(in crate::batcher::table) async fn flush_batch(
+    batch: Batch,
+    inserter: Arc<Inserter>,
+    notifier: FlowNotifier,
+) {
+    let started = Instant::now();
+    let result = send_batch(&batch, &inserter).await;
+    FLUSH_ELAPSED.observe(started.elapsed().as_secs_f64());
+    match result {
+        Ok((table, combined)) => {
+            FLUSH_TOTAL.inc();
+            FLUSH_ROWS.observe(batch.total_rows as f64);
+            notify_batches(batch.submissions, Ok(()));
+            // Admission is best-effort and only occurs after affected-row validation.
+            notifier.notify(table, &combined);
+        }
+        Err(error) => {
+            FLUSH_FAILURES.inc();
+            FLUSH_DROPPED_ROWS.inc_by(batch.total_rows as u64);
+            notify_batches(batch.submissions, Err(Arc::new(error)));
+        }
+    }
+}
+
+async fn send_batch(batch: &Batch, inserter: &Inserter) -> Result<(TableInfoRef, RecordBatch)> {
+    let first = batch.submissions.first().ok_or_else(|| {
+        UnexpectedSnafu {
+            violated: "cannot execute an empty flush".to_string(),
+        }
+        .build()
+    })?;
+    let combined = combine_batches(batch.submissions.iter().map(|pending| &pending.batch))?;
+    let affected_rows = inserter
+        .flush_bulk_batch(
+            first.table_info.clone(),
+            combined.clone(),
+            first.ctx.clone(),
+        )
+        .await?;
+    validate_affected_rows(batch.total_rows, affected_rows)?;
+    Ok((first.table_info.clone(), combined))
+}
+
+fn combine_batches<'a>(batches: impl IntoIterator<Item = &'a RecordBatch>) -> Result<RecordBatch> {
+    let batches = batches.into_iter().collect::<Vec<_>>();
+    let Some(first) = batches.first() else {
+        return UnexpectedSnafu {
+            violated: "cannot combine an empty flush".to_string(),
+        }
+        .fail();
+    };
+    let schema = first.schema();
+    if batches.iter().any(|batch| batch.schema() != schema) {
+        return UnexpectedSnafu {
+            violated: "cannot combine different Arrow schemas".to_string(),
+        }
+        .fail();
+    }
+    concat_batches(&schema, batches).context(ComputeArrowSnafu)
+}
+
+fn validate_affected_rows(expected: usize, actual: usize) -> Result<()> {
+    if expected != actual {
+        return UnexpectedSnafu {
+            violated: format!("batched write affected {actual} rows, expected {expected}; individual results cannot be attributed"),
+        }.fail();
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow::array::Int32Array;
+    use arrow::datatypes::{DataType, Field, Schema};
+
+    use crate::batcher::table::batch::*;
+
+    fn batch(name: &str, values: Vec<i32>) -> RecordBatch {
+        RecordBatch::try_new(
+            Arc::new(Schema::new(vec![Field::new(name, DataType::Int32, false)])),
+            vec![Arc::new(Int32Array::from(values))],
+        )
+        .unwrap()
+    }
+
+    #[test]
+    fn test_combine_preserves_rows_and_rejects_schema_changes() {
+        let first = batch("a", vec![1, 2]);
+        let second = batch("a", vec![3]);
+        assert_eq!(
+            batch("a", vec![1, 2, 3]),
+            combine_batches([&first, &second]).unwrap()
+        );
+        let different = batch("b", vec![4]);
+        assert!(combine_batches([&first, &different]).is_err());
+        let schema = first.schema();
+        let metadata = Schema::new_with_metadata(
+            schema.fields().clone(),
+            [("version".to_string(), "2".to_string())]
+                .into_iter()
+                .collect(),
+        );
+        let changed = RecordBatch::try_new(Arc::new(metadata), first.columns().to_vec()).unwrap();
+        assert!(combine_batches([&first, &changed]).is_err());
+    }
+
+    #[test]
+    fn test_affected_rows_must_match_before_attribution() {
+        for (expected, actual, valid) in [(0, 0, true), (3, 3, true), (3, 2, false), (3, 4, false)]
+        {
+            assert_eq!(valid, validate_affected_rows(expected, actual).is_ok());
+        }
+    }
+}

src/servers/src/batcher/table/flow_notifier.rs

@@ -0,0 +1,167 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::num::NonZeroUsize;
+
+use arrow::record_batch::RecordBatch;
+use common_meta::cache::TableFlownodeSetCacheRef;
+use common_meta::node_manager::NodeManagerRef;
+use common_telemetry::error;
+use lazy_static::lazy_static;
+use operator::req_convert::insert::extract_timestamps;
+use prometheus::{IntCounterVec, register_int_counter_vec};
+use table::metadata::TableInfoRef;
+
+use crate::batcher::flow_notifier::{
+    FlowNotifier as NotificationQueue, start_flow_notification_worker,
+};
+use crate::batcher::flow_sender::FlowNotification;
+
+lazy_static! {
+    static ref FLOW_NOTIFICATION_DROPPED: IntCounterVec = register_int_counter_vec!(
+        "greptime_table_batcher_flow_notification_dropped_total",
+        "Ordinary batch flow notifications dropped before delivery",
+        &["reason"]
+    )
+    .unwrap();
+}
+
+/// Enqueues best-effort flow notifications independently of write completion.
+#[derive(Clone)]
+pub(in crate::batcher::table) struct FlowNotifier {
+    notifier: NotificationQueue,
+}
+
+impl FlowNotifier {
+    /// Starts bounded delivery on the shared runtime, or rejects unsupported capacity.
+    pub fn new(
+        cache: TableFlownodeSetCacheRef,
+        node_manager: NodeManagerRef,
+        capacity: NonZeroUsize,
+    ) -> Option<Self> {
+        let (notifier, receiver) =
+            NotificationQueue::try_new(capacity.get(), FLOW_NOTIFICATION_DROPPED.clone())?;
+        start_flow_notification_worker(receiver, cache, node_manager);
+        Some(Self { notifier })
+    }
+
+    /// Never waits for queue capacity or delivery and never changes the write result.
+    pub fn notify(&self, table_info: TableInfoRef, batch: &RecordBatch) {
+        let Some(timestamp_column) = table_info.meta.schema.timestamp_column() else {
+            return;
+        };
+        let timestamps = match extract_timestamps(batch, &timestamp_column.name) {
+            Ok(timestamps) => timestamps,
+            Err(error) => {
+                error!(error; "Failed to extract flow notification timestamps, table_id: {}", table_info.table_id());
+                return;
+            }
+        };
+        if timestamps.is_empty() {
+            return;
+        }
+        self.enqueue(FlowNotification {
+            table_id: table_info.table_id(),
+            timestamps,
+        });
+    }
+
+    fn enqueue(&self, notification: FlowNotification) {
+        self.notifier.try_notify(notification);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use api::helper::ColumnDataTypeWrapper;
+    use api::v1::ColumnDataType;
+    use arrow::array::{
+        ArrayRef, TimestampMicrosecondArray, TimestampMillisecondArray, TimestampNanosecondArray,
+        TimestampSecondArray,
+    };
+    use arrow::record_batch::RecordBatch;
+    use datatypes::data_type::ConcreteDataType;
+    use datatypes::schema::{ColumnSchema, Schema};
+    use operator::test_util::new_test_table_info;
+
+    use crate::batcher::flow_notifier::FlowNotifier as NotificationQueue;
+    use crate::batcher::table::flow_notifier::{
+        FLOW_NOTIFICATION_DROPPED, FlowNotification, FlowNotifier,
+    };
+
+    #[test]
+    fn test_full_and_closed_queues_do_not_block() {
+        let (notifier, receiver) =
+            NotificationQueue::try_new(1, FLOW_NOTIFICATION_DROPPED.clone()).unwrap();
+        let notifier = FlowNotifier { notifier };
+        let full = FLOW_NOTIFICATION_DROPPED.with_label_values(&["full"]);
+        let closed = FLOW_NOTIFICATION_DROPPED.with_label_values(&["closed"]);
+        let before_full = full.get();
+        let before_closed = closed.get();
+        for _ in 0..2 {
+            notifier.enqueue(FlowNotification {
+                table_id: 1,
+                timestamps: vec![42],
+            });
+        }
+        assert!(full.get() > before_full);
+        drop(receiver);
+        notifier.enqueue(FlowNotification {
+            table_id: 1,
+            timestamps: vec![42],
+        });
+        assert!(closed.get() > before_closed);
+    }
+    #[test]
+    fn test_notification_preserves_native_timestamp_units() {
+        let cases: Vec<(ColumnDataType, ArrayRef)> = vec![
+            (
+                ColumnDataType::TimestampSecond,
+                Arc::new(TimestampSecondArray::from(vec![Some(42), None])),
+            ),
+            (
+                ColumnDataType::TimestampMillisecond,
+                Arc::new(TimestampMillisecondArray::from(vec![Some(42), None])),
+            ),
+            (
+                ColumnDataType::TimestampMicrosecond,
+                Arc::new(TimestampMicrosecondArray::from(vec![Some(42), None])),
+            ),
+            (
+                ColumnDataType::TimestampNanosecond,
+                Arc::new(TimestampNanosecondArray::from(vec![Some(42), None])),
+            ),
+        ];
+        for (datatype, array) in cases {
+            let mut table = new_test_table_info(1, "test", [0].into_iter());
+            let column = ColumnSchema::new(
+                "ts",
+                ConcreteDataType::from(ColumnDataTypeWrapper::new(datatype, None)),
+                true,
+            )
+            .with_time_index(true);
+            table.meta.schema = Arc::new(Schema::new(vec![column]));
+            let batch = RecordBatch::try_new(table.meta.schema.arrow_schema().clone(), vec![array])
+                .unwrap();
+            let (notifier, mut receiver) =
+                NotificationQueue::try_new(1, FLOW_NOTIFICATION_DROPPED.clone()).unwrap();
+            FlowNotifier { notifier }.notify(Arc::new(table), &batch);
+            let notification = receiver.try_recv().unwrap();
+            assert_eq!(notification.table_id, 1);
+            assert_eq!(notification.timestamps, vec![42]);
+        }
+    }
+}

src/servers/src/batcher/table/metrics.rs

@@ -0,0 +1,64 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use lazy_static::lazy_static;
+use prometheus::{
+    Histogram, IntCounter, IntGauge, register_histogram, register_int_counter, register_int_gauge,
+};
+
+lazy_static! {
+    pub(in crate::batcher::table) static ref PENDING_WORKERS: IntGauge = register_int_gauge!(
+        "greptime_table_batcher_pending_workers",
+        "Registered table batch workers"
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref FLUSH_DROPPED_ROWS: IntCounter =
+        register_int_counter!(
+            "greptime_table_batcher_flush_dropped_rows",
+            "Rows in failed table batch flushes"
+        )
+        .unwrap();
+    pub(in crate::batcher::table) static ref PENDING_BATCHES: IntGauge = register_int_gauge!(
+        "greptime_table_batcher_pending_batches",
+        "Pending table batches"
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref PENDING_ROWS: IntGauge = register_int_gauge!(
+        "greptime_table_batcher_pending_rows",
+        "Rows awaiting table batch flush"
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref FLUSH_TOTAL: IntCounter = register_int_counter!(
+        "greptime_table_batcher_flush_total",
+        "Successful table batch flushes"
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref FLUSH_FAILURES: IntCounter = register_int_counter!(
+        "greptime_table_batcher_flush_failures",
+        "Failed table batch flushes"
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref FLUSH_ROWS: Histogram = register_histogram!(
+        "greptime_table_batcher_flush_rows",
+        "Number of rows per successful table batch flush",
+        vec![100.0, 1000.0, 10000.0, 50000.0, 100000.0, 500000.0]
+    )
+    .unwrap();
+    pub(in crate::batcher::table) static ref FLUSH_ELAPSED: Histogram = register_histogram!(
+        "greptime_table_batcher_flush_elapsed",
+        "Table batch flush duration in seconds",
+        vec![0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 5.0, 10.0, 60.0, 300.0]
+    )
+    .unwrap();
+}

src/servers/src/batcher/table/pending_batch.rs

@@ -0,0 +1,96 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use arrow::record_batch::RecordBatch;
+use operator::error::Error;
+use session::context::QueryContextRef;
+use table::metadata::TableInfoRef;
+use tokio::sync::{OwnedSemaphorePermit, oneshot};
+
+/// One complete input and its completion/admission ownership.
+pub(in crate::batcher::table) struct PendingBatch {
+    pub table_info: TableInfoRef,
+    pub batch: RecordBatch,
+    pub ctx: QueryContextRef,
+    pub response_tx: oneshot::Sender<Result<(), Arc<Error>>>,
+    pub _permit: Arc<OwnedSemaphorePermit>,
+}
+
+pub(in crate::batcher::table) fn notify_batches(
+    batches: Vec<PendingBatch>,
+    result: Result<(), Arc<Error>>,
+) {
+    for batch in batches {
+        let _ = batch.response_tx.send(result.clone());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use arrow::array::Int32Array;
+    use arrow::datatypes::{DataType, Field, Schema};
+    use operator::error::UnexpectedSnafu;
+    use operator::test_util::new_test_table_info;
+    use session::context::QueryContext;
+    use tokio::sync::Semaphore;
+
+    use crate::batcher::table::pending_batch::*;
+
+    #[tokio::test]
+    async fn test_completion_fans_out_and_releases_admission() {
+        for fail in [false, true] {
+            let semaphore = Arc::new(Semaphore::new(3));
+            let mut batches = Vec::new();
+            let mut receivers = Vec::new();
+            for value in 0..3 {
+                let (response_tx, response_rx) = oneshot::channel();
+                let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)]));
+                batches.push(PendingBatch {
+                    table_info: Arc::new(new_test_table_info(1, "t", [0].into_iter())),
+                    batch: RecordBatch::try_new(
+                        schema,
+                        vec![Arc::new(Int32Array::from(vec![value]))],
+                    )
+                    .unwrap(),
+                    ctx: QueryContext::arc(),
+                    response_tx,
+                    _permit: Arc::new(semaphore.clone().acquire_owned().await.unwrap()),
+                });
+                receivers.push(response_rx);
+            }
+            assert_eq!(0, semaphore.available_permits());
+            // A cancelled waiter must not prevent delivery to the others.
+            drop(receivers.pop());
+            let error = Arc::new(
+                UnexpectedSnafu {
+                    violated: "test flush failure".to_string(),
+                }
+                .build(),
+            );
+            notify_batches(batches, if fail { Err(error.clone()) } else { Ok(()) });
+            assert_eq!(3, semaphore.available_permits());
+            for receiver in receivers {
+                match receiver.await.unwrap() {
+                    Ok(()) => assert!(!fail),
+                    Err(actual) => {
+                        assert!(fail);
+                        assert!(Arc::ptr_eq(&error, &actual));
+                    }
+                }
+            }
+        }
+    }
+}

src/servers/src/batcher/table/pending_worker.rs

@@ -0,0 +1,185 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use arrow::datatypes::SchemaRef;
+use common_batcher::flush_limiter::FlushLimiter;
+use common_batcher::flush_policy::FlushTrigger;
+use common_batcher::flush_policy::timing::TimingFlushPolicy;
+use common_batcher::pending_worker::PendingWorker as PendingCore;
+use common_runtime::spawn_global;
+use common_telemetry::warn;
+use operator::insert::Inserter;
+use tokio::sync::{broadcast, mpsc};
+use tokio::task::JoinHandle;
+use tokio::time::{Instant, sleep_until};
+
+use crate::batcher::table::batch::{Batch, flush_batch};
+use crate::batcher::table::flow_notifier::FlowNotifier;
+use crate::batcher::table::metrics::{PENDING_BATCHES, PENDING_ROWS, PENDING_WORKERS};
+use crate::batcher::table::pending_batch::PendingBatch;
+use crate::batcher::table::{BatchKey, PendingWorkers};
+
+pub(in crate::batcher::table) enum WorkerCommand {
+    Submit(PendingBatch),
+}
+
+/// The sender handle does not own the worker's processing state.
+pub(in crate::batcher::table) struct PendingWorker {
+    pub tx: mpsc::Sender<WorkerCommand>,
+}
+
+/// Identifies submissions that can share a batch and overlapping writes.
+#[derive(PartialEq)]
+struct BatchIdentity {
+    table_id: u32,
+    table_version: u64,
+    schema: SchemaRef,
+}
+
+impl BatchIdentity {
+    fn new(batch: &PendingBatch) -> Self {
+        Self {
+            table_id: batch.table_info.table_id(),
+            table_version: batch.table_info.ident.version,
+            schema: batch.batch.schema(),
+        }
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+pub(in crate::batcher::table) fn start_worker(
+    key: BatchKey,
+    worker_tx: mpsc::Sender<WorkerCommand>,
+    workers: Arc<PendingWorkers>,
+    mut rx: mpsc::Receiver<WorkerCommand>,
+    shutdown: broadcast::Sender<()>,
+    flush_policy: TimingFlushPolicy,
+    flush_limiter: FlushLimiter,
+    inserter: Arc<Inserter>,
+    notifier: FlowNotifier,
+    idle_timeout: Duration,
+) {
+    spawn_global(async move {
+        let mut pending = PendingCore::new(flush_policy);
+        let mut identity: Option<BatchIdentity> = None;
+        // Handles are retained only for the table identity and schema transition barrier. Dropping
+        // them on shutdown detaches writes, matching the Prom worker contract.
+        let mut flushes: Vec<JoinHandle<()>> = Vec::new();
+        let mut shutdown_rx = shutdown.subscribe();
+        let mut closing = false;
+        let idle_timer = sleep_until(Instant::now() + idle_timeout);
+        tokio::pin!(idle_timer);
+        loop {
+            let batch = tokio::select! {
+                command = rx.recv() => {
+                    match command {
+                        Some(WorkerCommand::Submit(batch)) => {
+                            idle_timer.as_mut().reset(Instant::now() + idle_timeout);
+                            let next_identity = BatchIdentity::new(&batch);
+                            if identity.as_ref().is_some_and(|identity| *identity != next_identity) {
+                                if let Some(old) = drain_batch(&mut pending, None)
+                                    && let Some(task) = spawn_flush(old, &flush_limiter, inserter.clone(), notifier.clone()).await {
+                                    flushes.push(task);
+                                }
+                                // Finish writes from the previous table definition before switching.
+                                for task in flushes.drain(..) {
+                                    if let Err(error) = task.await { warn!(error; "Failed to join old-schema batch flush"); }
+                                }
+                            }
+                            identity = Some(next_identity);
+                            let rows = batch.batch.num_rows();
+                            if pending.is_empty() { PENDING_BATCHES.inc(); }
+                            pending.submit(batch, rows);
+                            PENDING_ROWS.add(rows as i64);
+                            drain_batch(&mut pending, Some(FlushTrigger::Submission))
+                        }
+                        None => {
+                            if let Some(batch) = drain_batch(&mut pending, None) {
+                                flush_batch(batch, inserter.clone(), notifier.clone()).await;
+                            }
+                            break;
+                        }
+                    }
+                }
+                _ = pending.wait_flush() => drain_batch(&mut pending, Some(FlushTrigger::Deadline)),
+                _ = &mut idle_timer, if !closing => {
+                    if pending.is_empty() && rx.is_empty() && flushes.iter().all(JoinHandle::is_finished) {
+                        // Keep table identity while writes are in flight. Closing
+                        // still allows previously reserved sends to be received.
+                        rx.close();
+                        closing = true;
+                    }
+                    idle_timer.as_mut().reset(Instant::now() + idle_timeout);
+                    None
+                }
+                _ = shutdown_rx.recv() => {
+                    if let Some(batch) = drain_batch(&mut pending, None) {
+                        flush_batch(batch, inserter.clone(), notifier.clone()).await;
+                    }
+                    break;
+                }
+            };
+            if let Some(batch) = batch {
+                flushes.retain(|task| !task.is_finished());
+                if let Some(task) =
+                    spawn_flush(batch, &flush_limiter, inserter.clone(), notifier.clone()).await
+                {
+                    flushes.push(task);
+                }
+            }
+        }
+        if workers.remove_if_same(&key, &worker_tx).await {
+            PENDING_WORKERS.set(workers.len().await as i64);
+        }
+    });
+}
+
+fn drain_batch(
+    pending: &mut PendingCore<PendingBatch, TimingFlushPolicy>,
+    trigger: Option<FlushTrigger>,
+) -> Option<Batch> {
+    let total_rows = pending.total_rows();
+    let submissions = match trigger {
+        Some(trigger) => pending.take_ready(trigger)?,
+        None => pending.take_pending()?,
+    };
+    PENDING_ROWS.sub(total_rows as i64);
+    PENDING_BATCHES.dec();
+    Some(Batch {
+        submissions,
+        total_rows,
+    })
+}
+
+async fn spawn_flush(
+    batch: Batch,
+    limiter: &FlushLimiter,
+    inserter: Arc<Inserter>,
+    notifier: FlowNotifier,
+) -> Option<JoinHandle<()>> {
+    match limiter.acquire().await {
+        Ok(permit) => Some(spawn_global(async move {
+            let _permit = permit;
+            flush_batch(batch, inserter, notifier).await;
+        })),
+        Err(error) => {
+            warn!(error; "Flush limiter closed, flushing inline");
+            flush_batch(batch, inserter, notifier).await;
+            None
+        }
+    }
+}

src/servers/src/batcher/test_util.rs

@@ -0,0 +1,46 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::sync::Arc;
+
+use common_meta::cache::{TableFlownodeSetCacheRef, new_table_flownode_set_cache};
+use common_meta::instruction::{CacheIdent, CreateFlow};
+use common_meta::key::FlowPartitionId;
+use common_meta::kv_backend::memory::MemoryKvBackend;
+use common_meta::peer::Peer;
+use moka::future::CacheBuilder;
+use store_api::storage::TableId;
+
+/// Creates an isolated Flow target cache, preserving duplicate peers across partitions.
+pub(in crate::batcher) async fn mock_table_flownode_cache(
+    table_id: TableId,
+    partition_to_peer_mapping: Vec<(FlowPartitionId, Peer)>,
+) -> TableFlownodeSetCacheRef {
+    let cache = Arc::new(new_table_flownode_set_cache(
+        "batcher-test".to_string(),
+        CacheBuilder::new(4).build(),
+        Arc::new(MemoryKvBackend::default()),
+    ));
+    if !partition_to_peer_mapping.is_empty() {
+        cache
+            .invalidate(&[CacheIdent::CreateFlow(CreateFlow {
+                flow_id: 1,
+                source_table_ids: vec![table_id],
+                partition_to_peer_mapping,
+            })])
+            .await
+            .unwrap();
+    }
+    cache
+}

src/servers/src/lib.rs

@@ -22,6 +22,7 @@ use sql::statements::statement::Statement;
 pub use tower;
 
 pub mod addrs;
+pub mod batcher;
 pub mod configurator;
 pub(crate) mod elasticsearch;
 pub mod error;