reifydb-engine 0.9.0

Query execution and processing engine for ReifyDB
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

use reifydb_catalog::catalog::{Catalog, flow::FlowToCreate, view::ViewColumnToCreate};
use reifydb_core::{
	interface::catalog::{
		column::ColumnIndex,
		flow::FlowStatus,
		view::{View, ViewSortKey},
	},
	sort::SortKey,
};
use reifydb_routine_abi::registry::Routines;
use reifydb_rql::{
	flow::{
		compiler::compile_flow,
		time_domain::{check_join_lateness_requirements, check_window_time_requirements},
	},
	query::QueryPlan,
};
use reifydb_transaction::transaction::{Transaction, admin::AdminTransaction};
use reifydb_value::fragment::Fragment;

use crate::Result;

fn outermost_sort(plan: &QueryPlan) -> Option<&Vec<SortKey>> {
	match plan {
		QueryPlan::Sort(node) => Some(&node.by),
		QueryPlan::Map(node) => node.input.as_deref().and_then(outermost_sort),
		QueryPlan::Extend(node) => node.input.as_deref().and_then(outermost_sort),
		QueryPlan::Filter(node) => outermost_sort(&node.input),
		QueryPlan::Take(node) => outermost_sort(&node.input),
		QueryPlan::Distinct(node) => outermost_sort(&node.input),
		_ => None,
	}
}

pub(crate) fn extract_view_sort(as_clause: &QueryPlan, columns: &[ViewColumnToCreate]) -> Vec<ViewSortKey> {
	let Some(by) = outermost_sort(as_clause) else {
		return Vec::new();
	};

	let mut resolved = Vec::with_capacity(by.len());
	for key in by {
		let Some(position) = columns.iter().position(|c| c.name.text() == key.column.text()) else {
			return Vec::new();
		};
		resolved.push(ViewSortKey {
			column: ColumnIndex(position as u8),
			direction: key.direction.clone(),
		});
	}
	resolved
}

pub mod authentication;
pub mod binding;
pub mod deferred;
pub mod dictionary;
pub mod event;

pub mod identity;
pub mod identity_attribute;
pub mod migration;
pub mod namespace;
pub mod policy;
pub mod primary_key;
pub mod procedure;
pub mod property;
pub mod queue;
pub mod relationship;
pub mod remote_namespace;
pub mod ringbuffer;
pub mod role;
pub mod series;
pub mod sink;
pub mod source;
pub mod subscription;
pub mod sumtype;
pub mod table;
pub mod tag;
pub mod test;
pub mod transactional;

pub(crate) fn create_deferred_view_flow(
	catalog: &Catalog,
	routines: &Routines,
	txn: &mut AdminTransaction,
	view: &View,
	plan: QueryPlan,
) -> Result<()> {
	let flow = catalog.create_flow(
		txn,
		FlowToCreate {
			name: Fragment::internal(view.name()),
			namespace: view.namespace(),
			status: FlowStatus::Active,
		},
	)?;

	let dag = compile_flow(catalog, routines, txn, plan, Some(view), flow.id)?;
	check_window_time_requirements(catalog, &mut Transaction::Admin(txn), &dag)?;
	check_join_lateness_requirements(catalog, &mut Transaction::Admin(txn), &dag)
}