reifydb_store_column/store/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2025 ReifyDB
3
4use std::{
5	sync::{Arc, Mutex},
6	time::Duration,
7};
8
9use reifydb_core::{
10	common::CommitVersion,
11	internal_err,
12	value::column::{compressed::CompressedColumn, data::ColumnData},
13};
14use reifydb_runtime::context::clock::{Clock, Instant};
15use reifydb_type::Result;
16
17use crate::{ColumnStatistics, ColumnStore, backend::Backend, config::ColumnStoreConfig, statistics::merge};
18
19#[derive(Clone)]
20pub struct StandardColumnStore {
21	pub(crate) hot: Option<Backend>,
22	pub(crate) warm: Option<Backend>,
23	pub(crate) cold: Option<Backend>,
24	_config: ColumnStoreConfig,
25	tier_state: Arc<Mutex<TierState>>,
26	_clock: Clock,
27}
28
29struct TierState {
30	_hot_evicted_version: CommitVersion,
31	_warm_evicted_version: CommitVersion,
32	_last_eviction_time: Instant,
33}
34
35impl StandardColumnStore {
36	pub fn new(config: ColumnStoreConfig, clock: Clock) -> Result<Self> {
37		let tier_state = TierState {
38			_hot_evicted_version: CommitVersion(0),
39			_warm_evicted_version: CommitVersion(0),
40			_last_eviction_time: clock.instant(),
41		};
42		Ok(Self {
43			hot: config.hot.as_ref().map(|c| c.backend.clone()),
44			warm: config.warm.as_ref().map(|c| c.backend.clone()),
45			cold: config.cold.as_ref().map(|c| c.backend.clone()),
46			_config: config,
47			tier_state: Arc::new(Mutex::new(tier_state)),
48			_clock: clock,
49		})
50	}
51
52	/// Determine which tier to write to based on age and tier availability
53	fn select_write_tier(&self, _version: CommitVersion) -> Option<&Backend> {
54		// For now, always write to hot tier if available, fallback to warm, then cold
55		self.hot.as_ref().or(self.warm.as_ref()).or(self.cold.as_ref())
56	}
57
58	/// Search across tiers for data, starting with hot (most recent)
59	fn search_tiers<T, F>(&self, mut search_fn: F) -> Option<T>
60	where
61		F: FnMut(&Backend) -> Option<T>,
62	{
63		// Search hot tier first
64		if let Some(hot) = &self.hot {
65			if let Some(result) = search_fn(hot) {
66				return Some(result);
67			}
68		}
69
70		// Search warm tier
71		if let Some(warm) = &self.warm {
72			if let Some(result) = search_fn(warm) {
73				return Some(result);
74			}
75		}
76
77		// Search cold tier
78		if let Some(cold) = &self.cold {
79			if let Some(result) = search_fn(cold) {
80				return Some(result);
81			}
82		}
83
84		None
85	}
86
87	/// Collect statistics from all tiers and merge them
88	fn collect_statistics(&self, column_index: usize) -> Option<ColumnStatistics> {
89		let mut all_stats = Vec::new();
90
91		if let Some(hot) = &self.hot {
92			if let Some(stats) = hot.statistics(column_index) {
93				all_stats.push(stats);
94			}
95		}
96
97		if let Some(warm) = &self.warm {
98			if let Some(stats) = warm.statistics(column_index) {
99				all_stats.push(stats);
100			}
101		}
102
103		if let Some(cold) = &self.cold {
104			if let Some(stats) = cold.statistics(column_index) {
105				all_stats.push(stats);
106			}
107		}
108
109		merge(&all_stats)
110	}
111
112	/// Check if data should be evicted based on retention policies
113	pub fn should_evict(&self) -> bool {
114		let state = self.tier_state.lock().unwrap();
115		let elapsed = state._last_eviction_time.elapsed();
116
117		// Simple time-based eviction check
118		elapsed > Duration::from_secs(300) // Check every 5 minutes
119	}
120
121	/// Evict data from hot to warm, and warm to cold based on retention policies
122	pub fn evict_data(&self) -> Result<usize> {
123		// TODO: Implement actual eviction logic
124		// This would involve:
125		// 1. Identifying data older than retention periods
126		// 2. Moving data from hot to warm tier
127		// 3. Moving data from warm to cold tier
128		// 4. Updating tier state
129
130		Ok(0) // Return number of evicted partitions
131	}
132}
133
134impl ColumnStore for StandardColumnStore {
135	fn insert(&self, version: CommitVersion, columns: Vec<CompressedColumn>) -> Result<()> {
136		if let Some(backend) = self.select_write_tier(version) {
137			backend.insert(version, columns)
138		} else {
139			internal_err!("No available backend for column storage")
140		}
141	}
142
143	fn scan(&self, version: CommitVersion, column_indices: &[usize]) -> Result<Vec<ColumnData>> {
144		Ok(self.search_tiers(|backend| backend.scan(version, column_indices).ok()).unwrap_or_else(|| vec![]))
145	}
146
147	fn statistics(&self, column_index: usize) -> Option<ColumnStatistics> {
148		self.collect_statistics(column_index)
149	}
150
151	fn partition_count(&self) -> usize {
152		let mut count = 0;
153
154		if let Some(hot) = &self.hot {
155			count += hot.partition_count();
156		}
157
158		if let Some(warm) = &self.warm {
159			count += warm.partition_count();
160		}
161
162		if let Some(cold) = &self.cold {
163			count += cold.partition_count();
164		}
165
166		count
167	}
168
169	fn compressed_size(&self) -> usize {
170		let mut size = 0;
171
172		if let Some(hot) = &self.hot {
173			size += hot.compressed_size();
174		}
175
176		if let Some(warm) = &self.warm {
177			size += warm.compressed_size();
178		}
179
180		if let Some(cold) = &self.cold {
181			size += cold.compressed_size();
182		}
183
184		size
185	}
186
187	fn uncompressed_size(&self) -> usize {
188		let mut size = 0;
189
190		if let Some(hot) = &self.hot {
191			size += hot.uncompressed_size();
192		}
193
194		if let Some(warm) = &self.warm {
195			size += warm.uncompressed_size();
196		}
197
198		if let Some(cold) = &self.cold {
199			size += cold.uncompressed_size();
200		}
201
202		size
203	}
204}
reifydb_store_column/store/mod.rs

reifydb_store_column/store/
mod.rs
