reifydb-engine 0.9.1

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
105
106
107
108
109
110
111
112
113
114
115
116

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

use reifydb_core::{
	error::diagnostic::subscription::subscription_missing_as_clause,
	interface::catalog::{flow::FlowId, id::SubscriptionId},
	value::column::columns::Columns,
};
use reifydb_evaluate::stack::SymbolTable;
use reifydb_rql::{
	flow::compiler::compile_subscription_flow_ephemeral, nodes::CreateSubscriptionNode, query::QueryPlan,
};
use reifydb_transaction::transaction::Transaction;
use reifydb_value::{error::Error, fragment::Fragment, params::Params, value::Value};

use crate::{
	Result,
	subscription::{SubscriptionContext, SubscriptionServiceRef},
	vm::services::Services,
};

pub(crate) fn create_subscription(
	services: &Services,
	txn: &mut Transaction<'_>,
	plan: CreateSubscriptionNode,
	symbols: &SymbolTable,
	params: &Params,
) -> Result<Columns> {
	if let Some(ref as_clause) = plan.as_clause
		&& let QueryPlan::RemoteScan(ref remote) = **as_clause
	{
		let token_value = match &remote.token {
			Some(t) => Value::Utf8(t.clone()),
			None => Value::none(),
		};
		let max_rows_value = match plan.hydration.max_rows {
			Some(n) => Value::Uint8(n),
			None => Value::none(),
		};
		let throttle_value = match plan.throttle {
			Some(d) => Value::Uint8(u64::try_from(d.milliseconds()?).unwrap_or(u64::MAX)),
			None => Value::none(),
		};
		let linger_value = match plan.linger {
			Some(d) => Value::Uint8(u64::try_from(d.milliseconds()?).unwrap_or(u64::MAX)),
			None => Value::none(),
		};
		return Ok(Columns::single_row([
			("remote_address", Value::Utf8(remote.address.clone())),
			("remote_body", Value::Utf8(remote.remote_rql.clone())),
			("remote_token", token_value),
			("remote_hydration_enabled", Value::Boolean(plan.hydration.enabled)),
			("remote_hydration_max_rows", max_rows_value),
			("remote_throttle_ms", throttle_value),
			("remote_linger_ms", linger_value),
		]));
	}

	let sub_service = services.ioc.resolve::<SubscriptionServiceRef>()?;

	let subscription_id = sub_service.next_id();

	let column_names: Vec<String> = plan.columns.iter().map(|c| c.name.clone()).collect();

	let as_clause =
		plan.as_clause.ok_or_else(|| Error(Box::new(subscription_missing_as_clause(Fragment::None))))?;

	let flow_id = FlowId(subscription_id.0);
	let flow_dag = compile_subscription_flow_ephemeral(
		&services.catalog,
		&services.routines,
		txn,
		*as_clause,
		subscription_id,
		flow_id,
	)?;

	let ctx = subscription_context(subscription_id, txn, symbols, params);
	sub_service.register_subscription(flow_dag, column_names, plan.hydration.enabled, ctx, txn)?;

	let hydration_max_rows = match plan.hydration.max_rows {
		Some(n) => Value::Uint8(n),
		None => Value::none(),
	};
	let throttle_value = match plan.throttle {
		Some(d) => Value::Uint8(u64::try_from(d.milliseconds()?).unwrap_or(u64::MAX)),
		None => Value::none(),
	};
	let linger_value = match plan.linger {
		Some(d) => Value::Uint8(u64::try_from(d.milliseconds()?).unwrap_or(u64::MAX)),
		None => Value::none(),
	};

	Ok(Columns::single_row([
		("subscription_id", Value::Uint8(subscription_id.0)),
		("created", Value::Boolean(true)),
		("hydration_enabled", Value::Boolean(plan.hydration.enabled)),
		("hydration_max_rows", hydration_max_rows),
		("throttle_ms", throttle_value),
		("linger_ms", linger_value),
	]))
}

fn subscription_context(
	id: SubscriptionId,
	txn: &mut Transaction<'_>,
	symbols: &SymbolTable,
	params: &Params,
) -> SubscriptionContext {
	SubscriptionContext {
		id,
		identity: txn.identity(),
		symbols: symbols.clone(),
		params: params.clone(),
	}
}