reifydb_engine/vm/volcano/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use arrow_array::{ArrayRef, RecordBatch};
5use arrow_schema::FieldRef;
6use reifydb_core::{
7	interface::catalog::dictionary::Dictionary,
8	value::{batch::batch, column::dictionary::decode_dictionary_ids},
9};
10use reifydb_transaction::transaction::Transaction;
11use reifydb_value::{
12	fragment::Fragment,
13	value::{
14		system_columns::{is_system_field, user_columns},
15		value_type::ValueType,
16	},
17};
18use tracing::instrument;
19
20use crate::Result;
21
22#[instrument(level = "trace", skip_all, name = "volcano::scan::dictionaries")]
23pub(crate) fn decode_dictionary_columns(
24	input: RecordBatch,
25	dictionaries: &[Option<(Dictionary, ValueType)>],
26	rx: &mut Transaction,
27) -> Result<RecordBatch> {
28	if dictionaries.iter().all(Option::is_none) {
29		return Ok(input);
30	}
31	let schema = input.schema();
32	let mut columns = Vec::with_capacity(input.num_columns());
33	let mut user_index = 0usize;
34	for (field, array) in schema.fields().iter().zip(input.columns()) {
35		if is_system_field(field) {
36			columns.push((field.clone(), array.clone()));
37			continue;
38		}
39		let dictionary = dictionaries.get(user_index).and_then(Option::as_ref);
40		user_index += 1;
41		let Some((dictionary, declared)) = dictionary else {
42			columns.push((field.clone(), array.clone()));
43			continue;
44		};
45		columns.push(decode_dictionary_column(field, array, dictionary, declared.clone(), rx)?);
46	}
47	batch(columns)
48}
49
50pub(crate) fn decode_dictionary_column(
51	field: &FieldRef,
52	array: &ArrayRef,
53	dictionary: &Dictionary,
54	target: ValueType,
55	rx: &mut Transaction,
56) -> Result<(FieldRef, ArrayRef)> {
57	decode_dictionary_ids(field, array, target, |ids| rx.get_from_dictionary(dictionary, ids))
58}
59
60pub(crate) fn user_pairs(batch: &RecordBatch) -> Vec<(FieldRef, ArrayRef)> {
61	user_columns(batch).map(|(field, array)| (field.clone(), array.clone())).collect()
62}
63
64pub(crate) fn user_header_names(headers: &ColumnHeaders) -> Vec<Fragment> {
65	headers.columns.iter().filter(|name| !name.text().starts_with('#')).cloned().collect()
66}
67
68pub(crate) fn with_system_headers(user: Vec<Fragment>, from: &ColumnHeaders) -> ColumnHeaders {
69	let mut columns = user;
70	columns.extend(from.columns.iter().filter(|name| name.text().starts_with('#')).cloned());
71	ColumnHeaders {
72		columns,
73	}
74}
75
76pub(crate) fn with_user_columns(user: Vec<(FieldRef, ArrayRef)>, from: &RecordBatch) -> Result<RecordBatch> {
77	let mut columns = user;
78	for (field, array) in from.schema_ref().fields().iter().zip(from.columns()) {
79		if is_system_field(field) {
80			columns.push((field.clone(), array.clone()));
81		}
82	}
83	batch(columns)
84}
85
86use query::{QueryContext, QueryNode};
87use reifydb_core::value::column::headers::ColumnHeaders;
88
89pub(crate) struct NoopNode;
90
91impl QueryNode for NoopNode {
92	fn initialize<'a>(&mut self, _: &mut Transaction<'a>, _: &QueryContext) -> Result<()> {
93		Ok(())
94	}
95	fn next<'a>(&mut self, _: &mut Transaction<'a>, _: &mut QueryContext) -> Result<Option<RecordBatch>> {
96		Ok(None)
97	}
98	fn headers(&self) -> Option<ColumnHeaders> {
99		None
100	}
101}
102
103pub mod aggregate;
104pub mod append;
105pub mod apply_transform;
106pub mod assert;
107pub mod compile;
108pub mod distinct;
109pub mod environment;
110pub mod extend;
111pub mod filter;
112pub mod generator;
113pub mod inline;
114pub mod join;
115pub(crate) mod key_rows;
116pub mod lookup;
117pub mod map;
118pub(crate) mod partition_parameter;
119pub mod patch;
120pub mod query;
121pub(crate) mod rank;
122pub mod row_lookup;
123pub mod run_tests;
124pub mod scan;
125pub(crate) mod series_parameter;
126pub mod sort;
127pub mod take;
128pub mod top_k;
129pub(crate) mod udf;
130pub mod variable;
131pub mod window;
reifydb_engine/vm/volcano/mod.rs

reifydb_engine/vm/volcano/
mod.rs
