reifydb_engine/backfill/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::{BTreeMap, BTreeSet},
6	num::NonZeroU64,
7	sync::Arc,
8};
9
10use arrow_array::RecordBatch;
11use reifydb_core::{
12	common::CommitVersion,
13	interface::{catalog::object::ObjectId, change::Change, resolved::ResolvedSeries},
14	internal_err,
15};
16use reifydb_evaluate::stack::SymbolTable;
17#[cfg(feature = "testing")]
18use reifydb_flow::backfill::testing::{InstalledScanHooks, TestingScan};
19use reifydb_flow::backfill::{Scan, backfill_partitioned};
20use reifydb_transaction::transaction::Transaction;
21use reifydb_value::{fragment::Fragment, params::Params, value::partition::Partition};
22
23use crate::{
24	Result,
25	vm::{
26		services::Services,
27		volcano::{
28			query::{QueryContext, QueryNode, query_budget},
29			scan::{
30				ringbuffer::RingBufferScan, series::SeriesScanNode, table::TableScanNode,
31				view::ViewScanNode,
32			},
33		},
34	},
35};
36
37#[cfg(test)]
38mod tests;
39
40pub fn run<'a>(
41	services: &Arc<Services>,
42	tx: Transaction<'a>,
43	sources: &BTreeSet<ObjectId>,
44	batch_size: NonZeroU64,
45	consume: impl FnMut(&mut TransactionScan<'a>, Change) -> Result<()>,
46) -> Result<()> {
47	let sources = sources.iter().map(|&source| (source, None)).collect();
48	run_partitioned(services, tx, &sources, batch_size, consume)
49}
50
51pub fn run_partitioned<'a>(
52	services: &Arc<Services>,
53	tx: Transaction<'a>,
54	sources: &BTreeMap<ObjectId, Option<Partition>>,
55	batch_size: NonZeroU64,
56	mut consume: impl FnMut(&mut TransactionScan<'a>, Change) -> Result<()>,
57) -> Result<()> {
58	let mut scan = TransactionScan {
59		services: Arc::clone(services),
60		tx,
61		opened: None,
62	};
63	#[cfg(feature = "testing")]
64	if let Some(InstalledScanHooks(hooks)) = services.ioc.try_resolve::<InstalledScanHooks>() {
65		return backfill_partitioned(
66			&mut TestingScan::over(scan, hooks),
67			sources,
68			batch_size,
69			|testing, change| consume(testing.scan_mut(), change),
70		);
71	}
72	backfill_partitioned(&mut scan, sources, batch_size, &mut consume)
73}
74
75pub struct TransactionScan<'a> {
76	services: Arc<Services>,
77	tx: Transaction<'a>,
78	opened: Option<Opened>,
79}
80
81struct Opened {
82	node: Box<dyn QueryNode>,
83	context: QueryContext,
84	tagged_series: bool,
85}
86
87impl<'a> TransactionScan<'a> {
88	pub fn transaction(&mut self) -> &mut Transaction<'a> {
89		&mut self.tx
90	}
91
92	fn node(
93		&mut self,
94		source: ObjectId,
95		partition: Option<Partition>,
96		context: &Arc<QueryContext>,
97	) -> Result<(Box<dyn QueryNode>, bool)> {
98		let catalog = &self.services.catalog;
99		let tx = &mut self.tx;
100		match source {
101			ObjectId::Table(id) => {
102				let table = catalog.resolve_table(tx, id)?;
103				let node = TableScanNode::new(table, partition, Arc::clone(context), tx)?
104					.oldest_first()
105					.keep_dictionary_ids();
106				Ok((Box::new(node), false))
107			}
108			ObjectId::View(id) => {
109				let view = catalog.resolve_view(tx, id)?;
110				let node = ViewScanNode::new(view, partition, Arc::clone(context), tx)?
111					.oldest_first()
112					.keep_dictionary_ids();
113				Ok((Box::new(node), false))
114			}
115			ObjectId::RingBuffer(id) => {
116				let ringbuffer = catalog.resolve_ringbuffer(tx, id)?;
117				let node = RingBufferScan::new(ringbuffer, Arc::clone(context), tx)?
118					.oldest_first()
119					.keep_dictionary_ids();
120				Ok((Box::new(node), false))
121			}
122			ObjectId::Series(id) => {
123				let series = catalog.get_series(tx, id)?;
124				let namespace = catalog.resolve_namespace(tx, series.namespace)?;
125				let tagged = series.tag.is_some();
126				let series =
127					ResolvedSeries::new(Fragment::internal(series.name.clone()), namespace, series);
128				let node = SeriesScanNode::new(series, None, None, None, None, Arc::clone(context))?
129					.oldest_first()
130					.keep_dictionary_ids();
131				Ok((Box::new(node), tagged))
132			}
133			ObjectId::TableVirtual(_) | ObjectId::Dictionary(_) | ObjectId::Queue(_) => {
134				internal_err!("backfill cannot scan source {:?}: it has no row scan node", source)
135			}
136		}
137	}
138}
139
140impl Scan for TransactionScan<'_> {
141	fn version(&self) -> CommitVersion {
142		self.tx.version()
143	}
144
145	fn open(&mut self, source: ObjectId, batch_size: NonZeroU64) -> Result<()> {
146		self.open_partition(source, None, batch_size)
147	}
148
149	fn open_partition(
150		&mut self,
151		source: ObjectId,
152		partition: Option<Partition>,
153		batch_size: NonZeroU64,
154	) -> Result<()> {
155		let context = Arc::new(QueryContext {
156			services: Arc::clone(&self.services),
157			source: None,
158			batch_size: batch_size.get(),
159			params: Params::None,
160			symbols: SymbolTable::new(),
161			identity: self.tx.identity(),
162			memory: query_budget(&self.services),
163		});
164		let (mut node, tagged_series) = self.node(source, partition, &context)?;
165		node.initialize(&mut self.tx, &context)?;
166		self.opened = Some(Opened {
167			node,
168			context: (*context).clone(),
169			tagged_series,
170		});
171		Ok(())
172	}
173
174	fn next(&mut self) -> Result<Option<RecordBatch>> {
175		let Some(opened) = self.opened.as_mut() else {
176			return internal_err!("transaction scan pulled before any source was opened");
177		};
178		let Some(mut chunk) = opened.node.next(&mut self.tx, &mut opened.context)? else {
179			return Ok(None);
180		};
181		if opened.tagged_series {
182			chunk.remove_column(1);
183		}
184		Ok(Some(chunk))
185	}
186}
reifydb_engine/backfill/mod.rs

reifydb_engine/backfill/
mod.rs
