reifydb_engine/bulk_insert/storage/
ringbuffer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use arrow_array::RecordBatch;
5use reifydb_value::params::Params;
6
7use crate::bulk_insert::builder::{BulkInsertBuilder, ValidationMode};
8
9#[derive(Debug, Clone)]
10pub struct PendingRingBufferInsert {
11	pub namespace: String,
12	pub ringbuffer: String,
13	pub rows: Vec<Params>,
14	pub batches: Vec<(usize, RecordBatch)>,
15}
16
17impl PendingRingBufferInsert {
18	pub fn new(namespace: String, ringbuffer: String) -> Self {
19		Self {
20			namespace,
21			ringbuffer,
22			rows: Vec::new(),
23			batches: Vec::new(),
24		}
25	}
26
27	pub fn add_row(&mut self, params: Params) {
28		self.rows.push(params);
29	}
30
31	pub fn add_rows<I: IntoIterator<Item = Params>>(&mut self, iter: I) {
32		self.rows.extend(iter);
33	}
34
35	pub fn add_batch(&mut self, batch: RecordBatch) {
36		self.batches.push((self.rows.len(), batch));
37	}
38
39	pub(crate) fn row_count(&self) -> usize {
40		self.rows.len() + self.batches.iter().map(|(_, batch)| batch.num_rows()).sum::<usize>()
41	}
42}
43
44pub struct RingBufferInsertBuilder<'a, 'e, V: ValidationMode> {
45	parent: &'a mut BulkInsertBuilder<'e, V>,
46	pending: PendingRingBufferInsert,
47}
48
49impl<'a, 'e, V: ValidationMode> RingBufferInsertBuilder<'a, 'e, V> {
50	pub(crate) fn new(parent: &'a mut BulkInsertBuilder<'e, V>, namespace: String, ringbuffer: String) -> Self {
51		Self {
52			parent,
53			pending: PendingRingBufferInsert::new(namespace, ringbuffer),
54		}
55	}
56
57	pub fn row(mut self, params: Params) -> Self {
58		self.pending.add_row(params);
59		self
60	}
61
62	pub fn rows<I>(mut self, iter: I) -> Self
63	where
64		I: IntoIterator<Item = Params>,
65	{
66		self.pending.add_rows(iter);
67		self
68	}
69
70	pub fn batch(mut self, batch: RecordBatch) -> Self {
71		self.pending.add_batch(batch);
72		self
73	}
74
75	pub fn done(self) -> &'a mut BulkInsertBuilder<'e, V> {
76		self.parent.add_ringbuffer_insert(self.pending);
77		self.parent
78	}
79}
reifydb_engine/bulk_insert/storage/ringbuffer.rs

reifydb_engine/bulk_insert/storage/
ringbuffer.rs
