servers/batcher/logical_table/
pending_worker.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::HashMap;
16use std::sync::Arc;
17use std::time::Duration;
18
19use catalog::CatalogManagerRef;
20use common_batcher::flush_limiter::FlushLimiter;
21use common_batcher::flush_policy::FlushTrigger;
22use common_batcher::flush_policy::timing::TimingFlushPolicy;
23use common_batcher::pending_worker::PendingWorker as PendingCore;
24use common_batcher::worker_registry::WorkerRegistry;
25use common_telemetry::debug;
26use operator::write_dispatcher::WriteDispatcherRef;
27use partition::manager::PartitionRuleManagerRef;
28use session::context::QueryContextRef;
29use table::metadata::TableId;
30use tokio::sync::{OwnedSemaphorePermit, broadcast, mpsc, oneshot};
31
32use crate::batcher::flow_notifier::FlowNotifier;
33use crate::batcher::logical_table::BatchKey;
34use crate::batcher::logical_table::batch::{self, Batch};
35use crate::batcher::logical_table::batch_convert::{RecordBatchWithTsIdx, TableBatch};
36use crate::error::Error;
37use crate::metrics::{PENDING_BATCHES, PENDING_ROWS, PENDING_WORKERS};
38
39pub(in crate::batcher::logical_table) struct PendingBatch {
40    pub(in crate::batcher::logical_table) tables: HashMap<TableId, TableBatch>,
41    pub(in crate::batcher::logical_table) db_string: String,
42    pub(in crate::batcher::logical_table) ctx: QueryContextRef,
43}
44
45pub(in crate::batcher::logical_table) struct FlushWaiter {
46    pub(in crate::batcher::logical_table) response_tx:
47        oneshot::Sender<std::result::Result<(), Arc<Error>>>,
48    pub(in crate::batcher::logical_table) _permit: Arc<OwnedSemaphorePermit>,
49}
50
51#[derive(Clone)]
52pub(in crate::batcher::logical_table) struct PendingWorker {
53    pub(in crate::batcher::logical_table) tx: mpsc::Sender<WorkerCommand>,
54}
55
56/// Commands accepted by a logical-table batching worker.
57pub(crate) enum WorkerCommand {
58    Submit {
59        table_batches: Vec<(String, u32, RecordBatchWithTsIdx)>,
60        total_rows: usize,
61        ctx: QueryContextRef,
62        response_tx: oneshot::Sender<std::result::Result<(), Arc<Error>>>,
63        _permit: Arc<OwnedSemaphorePermit>,
64    },
65    #[cfg(test)]
66    Ack { ack_tx: oneshot::Sender<()> },
67}
68
69impl PendingBatch {
70    pub(in crate::batcher::logical_table) fn new(ctx: QueryContextRef) -> Self {
71        let db_string = ctx.get_db_string();
72        Self {
73            tables: HashMap::new(),
74            db_string,
75            ctx,
76        }
77    }
78
79    pub(in crate::batcher::logical_table) fn add_table_batch(
80        &mut self,
81        table_name: String,
82        table_id: TableId,
83        record_batch: RecordBatchWithTsIdx,
84    ) {
85        let entry = self.tables.entry(table_id).or_insert_with(|| TableBatch {
86            table_name,
87            table_id,
88            batches: Vec::new(),
89            row_count: 0,
90        });
91        entry.row_count += record_batch.batch.num_rows();
92        entry.batches.push(record_batch);
93    }
94}
95
96/// Starts a worker that accumulates logical rows and dispatches physical flushes.
97#[allow(clippy::too_many_arguments)]
98pub(crate) fn start_worker(
99    key: BatchKey,
100    worker_tx: mpsc::Sender<WorkerCommand>,
101    workers: Arc<WorkerRegistry<BatchKey, WorkerCommand>>,
102    mut rx: mpsc::Receiver<WorkerCommand>,
103    shutdown: broadcast::Sender<()>,
104    partition_manager: PartitionRuleManagerRef,
105    write_dispatcher: WriteDispatcherRef,
106    catalog_manager: CatalogManagerRef,
107    flow_notification_tx: FlowNotifier,
108    worker_idle_timeout: Duration,
109    flush_policy: TimingFlushPolicy,
110    flush_limiter: FlushLimiter,
111) {
112    tokio::spawn(async move {
113        // The business batch and pending flush are populated and drained together.
114        let mut batch = None;
115        let mut pending_flush = PendingCore::new(flush_policy);
116        let mut shutdown_rx = shutdown.subscribe();
117        let idle_deadline = tokio::time::Instant::now() + worker_idle_timeout;
118        let idle_timer = tokio::time::sleep_until(idle_deadline);
119        tokio::pin!(idle_timer);
120
121        loop {
122            tokio::select! {
123                cmd = rx.recv() => {
124                    match cmd {
125                        Some(WorkerCommand::Submit { table_batches, total_rows, ctx, response_tx, _permit }) => {
126                            let submitted_at = tokio::time::Instant::now();
127                            idle_timer.as_mut().reset(submitted_at + worker_idle_timeout);
128
129                            pending_flush.submit(
130                                FlushWaiter { response_tx, _permit },
131                                total_rows,
132                            );
133                            let pending_batch = batch.get_or_insert_with(||{
134                                PENDING_BATCHES.inc();
135                                PendingBatch::new(ctx)
136                            });
137
138                            for (table_name, table_id, record_batch) in table_batches {
139                                pending_batch.add_table_batch(table_name, table_id, record_batch);
140                            }
141
142                            PENDING_ROWS.add(total_rows as i64);
143
144                            if let Some(flush) = drain_batch(&mut batch, &mut pending_flush, Some(FlushTrigger::Submission)) {
145                                batch::spawn_flush(
146                                    flush,
147                                    partition_manager.clone(),
148                                    write_dispatcher.clone(),
149                                    catalog_manager.clone(),
150                                    flow_notification_tx.clone(),
151                                    flush_limiter.clone(),
152                                ).await;
153                            }
154                        }
155                        None => {
156                            if let Some(flush) = drain_batch(&mut batch, &mut pending_flush, None) {
157                                batch::flush_batch_with_managers(
158                                    flush,
159                                    partition_manager.clone(),
160                                    write_dispatcher.clone(),
161                                    catalog_manager.clone(),
162                                    flow_notification_tx.clone(),
163                                ).await;
164                            }
165                            break;
166                        }
167                        #[cfg(test)]
168                        Some(WorkerCommand::Ack { ack_tx }) => {
169                            let _ = ack_tx.send(());
170                        }
171                    }
172                }
173                _ = &mut idle_timer => {
174                    if !should_close_worker_on_idle_timeout(
175                        pending_flush.total_rows(),
176                        rx.len(),
177                    ) {
178                        idle_timer
179                            .as_mut()
180                            .reset(tokio::time::Instant::now() + worker_idle_timeout);
181                        continue;
182                    }
183
184                    debug!(
185                        "Closing idle pending rows worker due to timeout: catalog={}, schema={}, physical_table={}",
186                        key.catalog,
187                        key.schema,
188                        key.physical_table
189                    );
190                    break;
191                }
192                _ = pending_flush.wait_flush() => {
193                    if let Some(flush) = drain_batch(&mut batch, &mut pending_flush, Some(FlushTrigger::Deadline)) {
194                        batch::spawn_flush(
195                            flush,
196                            partition_manager.clone(),
197                            write_dispatcher.clone(),
198                            catalog_manager.clone(),
199                            flow_notification_tx.clone(),
200                            flush_limiter.clone(),
201                        ).await;
202                    }
203                }
204                _ = shutdown_rx.recv() => {
205                    if let Some(flush) = drain_batch(&mut batch, &mut pending_flush, None) {
206                        batch::flush_batch_with_managers(
207                            flush,
208                            partition_manager.clone(),
209                            write_dispatcher.clone(),
210                            catalog_manager.clone(),
211                            flow_notification_tx.clone(),
212                        ).await;
213                    }
214                    break;
215                }
216            }
217        }
218
219        remove_worker_if_same_channel(workers.as_ref(), &key, &worker_tx).await;
220    });
221}
222
223pub(in crate::batcher::logical_table) async fn remove_worker_if_same_channel(
224    workers: &WorkerRegistry<BatchKey, WorkerCommand>,
225    key: &BatchKey,
226    worker_tx: &mpsc::Sender<WorkerCommand>,
227) -> bool {
228    if workers.remove_if_same(key, worker_tx).await {
229        PENDING_WORKERS.set(workers.len().await as i64);
230        true
231    } else {
232        false
233    }
234}
235
236pub(in crate::batcher::logical_table) fn should_close_worker_on_idle_timeout(
237    total_row_count: usize,
238    queued_requests: usize,
239) -> bool {
240    total_row_count == 0 && queued_requests == 0
241}
242
243/// Transfers the ready batch, or drains unconditionally when no trigger is given.
244/// Execution and flush permits remain owned by the caller.
245pub(in crate::batcher::logical_table) fn drain_batch(
246    batch: &mut Option<PendingBatch>,
247    pending_flush: &mut PendingCore<FlushWaiter, TimingFlushPolicy>,
248    trigger: Option<FlushTrigger>,
249) -> Option<Batch> {
250    let total_row_count = pending_flush.total_rows();
251    let waiters = match trigger {
252        Some(trigger) => pending_flush.take_ready(trigger)?,
253        None => pending_flush.take_pending()?,
254    };
255    let batch = batch.take()?;
256
257    if total_row_count == 0 {
258        return None;
259    }
260
261    let table_batches = batch.tables.into_values().collect();
262
263    PENDING_ROWS.sub(total_row_count as i64);
264    PENDING_BATCHES.dec();
265
266    Some(Batch {
267        table_batches,
268        total_row_count,
269        db_string: batch.db_string,
270        ctx: batch.ctx,
271        waiters,
272    })
273}
274
275#[cfg(test)]
276mod tests {
277    use std::collections::HashMap;
278    use std::sync::Arc;
279    use std::time::Duration;
280
281    use catalog::memory::MemoryCatalogManager;
282    use common_batcher::flush_limiter::FlushLimiter;
283    use common_batcher::flush_policy::FlushTrigger;
284    use common_batcher::flush_policy::timing::TimingFlushPolicy;
285    use common_batcher::pending_worker::PendingWorker as PendingCore;
286    use common_batcher::worker_registry::WorkerRegistry;
287    use common_meta::cache::new_table_route_cache;
288    use common_meta::kv_backend::memory::MemoryKvBackend;
289    use common_meta::node_manager::NodeManagerRef;
290    use moka::future::CacheBuilder;
291    use operator::write_dispatcher::WriteDispatcher;
292    use partition::cache::new_partition_info_cache;
293    use partition::manager::PartitionRuleManager;
294    use tokio::sync::{Semaphore, broadcast, mpsc, oneshot};
295    use tokio::time::advance;
296
297    use crate::batcher::flow_notifier::FlowNotifier;
298    use crate::batcher::logical_table::BatchKey;
299    use crate::batcher::logical_table::batch_convert::TableBatch;
300    use crate::batcher::logical_table::pending_worker::{
301        FlushWaiter, PendingBatch, WorkerCommand, drain_batch, remove_worker_if_same_channel,
302        should_close_worker_on_idle_timeout, start_worker,
303    };
304    use crate::batcher::logical_table::test_util::{
305        ConcurrentMockNodeManager, mock_aligned_tag_batch,
306    };
307    use crate::error::Error;
308    use crate::metrics::FLOW_NOTIFICATION_DROPPED;
309
310    #[tokio::test]
311    async fn test_drain_batch_takes_initialized_pending_batch_from_option() {
312        let ctx = session::context::QueryContext::arc();
313        let (response_tx, _response_rx) = oneshot::channel();
314        let permit = Arc::new(Semaphore::new(1)).try_acquire_owned().unwrap();
315        let mut pending_flush =
316            PendingCore::new(TimingFlushPolicy::try_new(Duration::from_secs(10), 1).unwrap());
317        pending_flush.submit(
318            FlushWaiter {
319                response_tx,
320                _permit: Arc::new(permit),
321            },
322            1,
323        );
324        let mut batch = Some(PendingBatch {
325            tables: HashMap::from([(
326                42,
327                TableBatch {
328                    table_name: "cpu".to_string(),
329                    table_id: 42,
330                    batches: vec![mock_aligned_tag_batch("tag1", "host-1", 1000, 1.0)],
331                    row_count: 1,
332                },
333            )]),
334            db_string: ctx.get_db_string(),
335            ctx: ctx.clone(),
336        });
337
338        let flush = drain_batch(
339            &mut batch,
340            &mut pending_flush,
341            Some(FlushTrigger::Submission),
342        )
343        .unwrap();
344
345        assert!(batch.is_none());
346        assert!(pending_flush.is_empty());
347        assert_eq!(0, pending_flush.total_rows());
348        assert_eq!(1, flush.waiters.len());
349        assert_eq!(1, flush.total_row_count);
350        assert_eq!(1, flush.table_batches.len());
351        assert_eq!(ctx.get_db_string(), flush.db_string);
352        assert_eq!(ctx.current_catalog(), flush.ctx.current_catalog());
353    }
354
355    #[tokio::test]
356    async fn test_drain_batch_preserves_unready_state_and_clears_zero_rows() {
357        for total_rows in [0, 1] {
358            let mut pending_flush =
359                PendingCore::new(TimingFlushPolicy::try_new(Duration::from_secs(10), 2).unwrap());
360            let mut batch = Some(PendingBatch::new(session::context::QueryContext::arc()));
361            let semaphore = Arc::new(Semaphore::new(1));
362            let (response_tx, mut response_rx) = oneshot::channel();
363            pending_flush.submit(
364                FlushWaiter {
365                    response_tx,
366                    _permit: Arc::new(semaphore.clone().acquire_owned().await.unwrap()),
367                },
368                total_rows,
369            );
370            assert!(
371                drain_batch(
372                    &mut batch,
373                    &mut pending_flush,
374                    Some(FlushTrigger::Submission)
375                )
376                .is_none()
377            );
378            assert!(batch.is_some());
379            assert!(!pending_flush.is_empty());
380            assert_eq!(total_rows, pending_flush.total_rows());
381            assert_eq!(0, semaphore.available_permits());
382
383            let drained = drain_batch(&mut batch, &mut pending_flush, None);
384            assert!(batch.is_none());
385            assert!(pending_flush.is_empty());
386            assert_eq!(0, pending_flush.total_rows());
387            if total_rows == 0 {
388                assert!(drained.is_none());
389                assert_eq!(1, semaphore.available_permits());
390                assert!(matches!(
391                    response_rx.try_recv(),
392                    Err(oneshot::error::TryRecvError::Closed)
393                ));
394            } else {
395                let drained = drained.unwrap();
396                assert_eq!(1, drained.total_row_count);
397                assert_eq!(1, drained.waiters.len());
398                assert_eq!(0, semaphore.available_permits());
399                drop(drained);
400                assert_eq!(1, semaphore.available_permits());
401            }
402        }
403    }
404
405    #[test]
406    fn test_pending_batch_keeps_same_name_batches_with_distinct_table_ids() {
407        let ctx = session::context::QueryContext::arc();
408        let mut pending_batch = PendingBatch::new(ctx);
409
410        pending_batch.add_table_batch(
411            "cpu".to_string(),
412            42,
413            mock_aligned_tag_batch("tag1", "host-1", 1000, 1.0),
414        );
415        pending_batch.add_table_batch(
416            "cpu".to_string(),
417            43,
418            mock_aligned_tag_batch("tag1", "host-1", 2000, 2.0),
419        );
420
421        assert_eq!(2, pending_batch.tables.len());
422        assert_eq!(42, pending_batch.tables[&42].table_id);
423        assert_eq!(43, pending_batch.tables[&43].table_id);
424        assert_eq!("cpu", pending_batch.tables[&42].table_name);
425        assert_eq!("cpu", pending_batch.tables[&43].table_name);
426    }
427
428    #[tokio::test]
429    async fn test_remove_worker_if_same_channel_removes_matching_entry() {
430        let workers = WorkerRegistry::new();
431        let key = BatchKey {
432            catalog: "greptime".to_string(),
433            schema: "public".to_string(),
434            physical_table: "phy".to_string(),
435            skip_wal: false,
436        };
437
438        let (tx, _rx) = mpsc::channel::<WorkerCommand>(1);
439        workers.get_or_insert_with(key.clone(), || tx.clone()).await;
440
441        assert!(remove_worker_if_same_channel(&workers, &key, &tx).await);
442        assert!(workers.is_empty().await);
443    }
444
445    #[tokio::test]
446    async fn test_remove_worker_if_same_channel_keeps_newer_entry() {
447        let workers = WorkerRegistry::new();
448        let key = BatchKey {
449            catalog: "greptime".to_string(),
450            schema: "public".to_string(),
451            physical_table: "phy".to_string(),
452            skip_wal: false,
453        };
454
455        let (stale_tx, _stale_rx) = mpsc::channel::<WorkerCommand>(1);
456        let (fresh_tx, _fresh_rx) = mpsc::channel::<WorkerCommand>(1);
457        workers
458            .get_or_insert_with(key.clone(), || fresh_tx.clone())
459            .await;
460
461        assert!(!remove_worker_if_same_channel(&workers, &key, &stale_tx).await);
462        assert!(workers.get(&key).await.is_some());
463        assert!(workers.get(&key).await.unwrap().same_channel(&fresh_tx));
464    }
465
466    #[test]
467    fn test_worker_idle_timeout_close_decision() {
468        assert!(should_close_worker_on_idle_timeout(0, 0));
469        assert!(!should_close_worker_on_idle_timeout(1, 0));
470        assert!(!should_close_worker_on_idle_timeout(0, 1));
471    }
472
473    #[tokio::test(start_paused = true)]
474    async fn test_worker_preserves_first_deadline_and_inline_shutdown() {
475        let flush_interval = Duration::from_secs(10);
476        let worker_idle_timeout = Duration::from_secs(30);
477        let key = BatchKey {
478            catalog: "greptime".to_string(),
479            schema: "public".to_string(),
480            physical_table: "phy".to_string(),
481            skip_wal: false,
482        };
483        let workers = Arc::new(WorkerRegistry::new());
484        let (worker_tx, worker_rx) = mpsc::channel(1);
485        workers
486            .get_or_insert_with(key.clone(), || worker_tx.clone())
487            .await;
488
489        let backend = Arc::new(MemoryKvBackend::default());
490        let table_route_cache = Arc::new(new_table_route_cache(
491            "pending-rows-flush-deadline-routes".to_string(),
492            CacheBuilder::new(1).build(),
493            backend.clone(),
494        ));
495        let partition_info_cache = Arc::new(new_partition_info_cache(
496            "pending-rows-flush-deadline-partitions".to_string(),
497            CacheBuilder::new(1).build(),
498            table_route_cache.clone(),
499        ));
500        let partition_manager = Arc::new(PartitionRuleManager::new(
501            backend,
502            table_route_cache,
503            partition_info_cache,
504        ));
505        let node_manager: NodeManagerRef = Arc::new(ConcurrentMockNodeManager {
506            datanodes: Arc::new(HashMap::new()),
507        });
508        let write_dispatcher = Arc::new(WriteDispatcher::new(
509            node_manager,
510            partition_manager.clone(),
511        ));
512        let catalog_manager = MemoryCatalogManager::with_default_setup();
513        let (flow_notification_tx, _flow_notification_rx) =
514            FlowNotifier::try_new(1, FLOW_NOTIFICATION_DROPPED.clone()).unwrap();
515        let (shutdown, _) = broadcast::channel(1);
516
517        let flush_limiter = FlushLimiter::try_new(1).unwrap();
518        start_worker(
519            key.clone(),
520            worker_tx.clone(),
521            workers.clone(),
522            worker_rx,
523            shutdown.clone(),
524            partition_manager,
525            write_dispatcher,
526            catalog_manager,
527            flow_notification_tx,
528            worker_idle_timeout,
529            TimingFlushPolicy::try_new(flush_interval, 3).unwrap(),
530            flush_limiter.clone(),
531        );
532
533        // Start the worker, then size-flush a batch halfway to the first
534        // worker-aligned interval boundary. This arms the reusable timer and
535        // drains the batch before that deadline is reached.
536        tokio::task::yield_now().await;
537        advance(flush_interval / 2).await;
538        let size_flush_rx = submit_mock_worker_batch(&worker_tx, 3, 1000).await;
539        let size_flush_result =
540            receive_mock_flush_result(size_flush_rx, "row threshold did not flush the first batch")
541                .await;
542        assert_missing_physical_table(size_flush_result);
543
544        // Submit a low-volume batch before the first batch's old timer would
545        // expire. It must receive a fresh full interval.
546        advance(flush_interval / 5).await;
547        let mut timed_flush_rx = submit_mock_worker_batch(&worker_tx, 1, 2000).await;
548
549        let first_submission = tokio::time::Instant::now();
550        advance(flush_interval / 2).await;
551        let later_flush_rx = submit_mock_worker_batch(&worker_tx, 1, 3000).await;
552        advance(flush_interval / 2 - Duration::from_millis(1)).await;
553        for _ in 0..10 {
554            tokio::task::yield_now().await;
555        }
556        assert!(matches!(
557            timed_flush_rx.try_recv(),
558            Err(oneshot::error::TryRecvError::Empty)
559        ));
560
561        advance(Duration::from_millis(1)).await;
562        let timed_flush_result = receive_mock_flush_result(
563            timed_flush_rx,
564            "batch was not flushed one interval after its creation",
565        )
566        .await;
567        assert_missing_physical_table(timed_flush_result);
568
569        assert_eq!(
570            first_submission + flush_interval,
571            tokio::time::Instant::now()
572        );
573        assert_missing_physical_table(
574            receive_mock_flush_result(
575                later_flush_rx,
576                "later submission was not included in the timed flush",
577            )
578            .await,
579        );
580
581        // Shutdown retains the existing inline path even with no flush permits.
582        let _held_permit = flush_limiter.acquire().await.unwrap();
583        let shutdown_flush_rx = submit_mock_worker_batch(&worker_tx, 1, 4000).await;
584        let shutdown_at = tokio::time::Instant::now();
585        let _ = shutdown.send(());
586        assert_missing_physical_table(
587            receive_mock_flush_result(
588                shutdown_flush_rx,
589                "shutdown incorrectly waited for a flush permit",
590            )
591            .await,
592        );
593        assert_eq!(shutdown_at, tokio::time::Instant::now());
594        for _ in 0..10 {
595            if workers.is_empty().await {
596                break;
597            }
598            tokio::task::yield_now().await;
599        }
600        assert!(
601            workers.is_empty().await,
602            "worker did not exit after shutdown"
603        );
604    }
605
606    const WORKER_TEST_TIMEOUT: Duration = Duration::from_secs(30);
607
608    async fn submit_mock_worker_batch(
609        worker_tx: &mpsc::Sender<WorkerCommand>,
610        total_rows: usize,
611        timestamp: i64,
612    ) -> oneshot::Receiver<std::result::Result<(), Arc<Error>>> {
613        let (response_tx, response_rx) = oneshot::channel();
614        let permit = Arc::new(Semaphore::new(1)).acquire_owned().await.unwrap();
615        worker_tx
616            .send(WorkerCommand::Submit {
617                table_batches: vec![(
618                    "cpu".to_string(),
619                    42,
620                    mock_aligned_tag_batch("tag1", "host-1", timestamp, 1.0),
621                )],
622                total_rows,
623                ctx: session::context::QueryContext::arc(),
624                response_tx,
625                _permit: Arc::new(permit),
626            })
627            .await
628            .unwrap();
629
630        // The channel is FIFO, so the ack proves the worker has dequeued and
631        // processed the submission (anchoring the flush deadline) before the
632        // caller advances virtual time.
633        let (ack_tx, ack_rx) = oneshot::channel();
634        worker_tx.send(WorkerCommand::Ack { ack_tx }).await.unwrap();
635        ack_rx
636            .await
637            .expect("worker exited before acking the submitted batch");
638
639        response_rx
640    }
641
642    async fn receive_mock_flush_result(
643        response_rx: oneshot::Receiver<std::result::Result<(), Arc<Error>>>,
644        context: &str,
645    ) -> std::result::Result<(), Arc<Error>> {
646        // Under paused time the timeout auto-advances the clock and fires
647        // deterministically if the flush never completes.
648        tokio::time::timeout(WORKER_TEST_TIMEOUT, response_rx)
649            .await
650            .unwrap_or_else(|_| panic!("{context}"))
651            .expect("flush result channel closed without a result")
652    }
653
654    fn assert_missing_physical_table(result: std::result::Result<(), Arc<Error>>) {
655        let err = result.expect_err("the empty catalog should make the flush fail");
656        assert!(
657            matches!(
658                err.as_ref(),
659                Error::Internal { err_msg }
660                    if err_msg.contains("not found during pending flush")
661            ),
662            "unexpected flush error: {err}"
663        );
664    }
665}
servers/batcher/logical_table/pending_worker.rs

servers/batcher/logical_table/
pending_worker.rs
