reifydb-sdk 0.8.0

SDK for building ReifyDB operators, procedures, transforms and more
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

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

//! Windowed-aggregation authoring surface.
//!
//! An operator implements one of the windowed authoring traits over a
//! `reifydb_core::window::accumulator::WindowAccumulator`:
//! - [`tumbling::TumblingOperator`] - non-overlapping windows.
//! - [`tumbling_carry::TumblingCarryOperator`] - tumbling windows that carry a value forward into the next window
//!   (EMA-family, prev-close, Heikin-Ashi).
//! - [`rolling::RollingOperator`] / [`rolling_incremental::RollingIncrementalOperator`]
//!   - overlapping rolling buffers of the last N windows.
//! - [`multi_rolling::MultiRollingOperator`] - rolling windows that emit multiple rows per group (top-K).
//!
//! The matching driver handles diff routing uniformly (`Insert -> add`,
//! `Update -> remove(pre) + add(post)`, `Remove -> remove(pre)`), window
//! boundary math, late-event drop, and state persistence in one place, so the
//! operator only describes its accumulator and how to build an output row.
//! Coordinate machinery lives in `reifydb_core::window::span`; the reusable
//! accumulator primitives in `reifydb_core::window::accumulator`.

pub mod bridge;
pub mod multi_rolling;
pub mod rolling;
pub mod rolling_incremental;
pub mod tumbling;
pub mod tumbling_carry;

use reifydb_core::window::engine::{LatePolicy, config::WindowEngineConfig};

use crate::config::Config;

pub(crate) fn late_policy_from_config(config: &Config) -> LatePolicy {
	match config.str("__late_policy").as_deref() {
		Some("process") => LatePolicy::Process,
		_ => LatePolicy::Drop,
	}
}

pub(crate) fn window_engine_config(config: &Config) -> WindowEngineConfig {
	let mut builder = WindowEngineConfig::builder().late_policy(late_policy_from_config(config));
	if let Some(capacity) = config.usize("state_cache_size") {
		builder = builder.state_cache_capacity(capacity);
	}
	if let Some(capacity) = config.usize("internal_state_cache_size") {
		builder = builder.internal_state_cache_capacity(capacity);
	}
	builder.build()
}