reifydb-store-column 0.1.3

Column-oriented storage for analytical queries in 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
117
118
119
120
121
122
123
124
125
126
127
128
129

// Copyright (c) reifydb.com 2025
// This file is licensed under the AGPL-3.0-or-later, see license.md file

use std::sync::Arc;

use crossbeam_skiplist::SkipMap;
use reifydb_core::{
	CommitVersion,
	value::column::{ColumnData, Columns, CompressedColumn},
};
use reifydb_type::Result;

use crate::{ColumnStatistics, ColumnStore, Partition, PartitionKey, statistics::merge};

#[derive(Clone)]
pub struct MemoryColumnStore {
	partitions: Arc<SkipMap<PartitionKey, Partition>>,
}

impl MemoryColumnStore {
	pub fn new() -> Self {
		Self {
			partitions: Arc::new(SkipMap::new()),
		}
	}

	/// Compress columns using the appropriate compressor for each column type
	pub fn compress_columns(&self, _columns: &Columns) -> Result<Vec<CompressedColumn>> {
		// let mut result = Vec::new();
		//
		// for column in columns.columns {
		// 	let compressor = select_compressor(column.data());
		// 	result.push(compressor.compress(column.data())?);
		// }
		//
		// Ok(result)
		todo!()
	}

	/// Decompress columns
	pub fn decompress_columns(&self, _compressed: &[CompressedColumn]) -> Result<Vec<ColumnData>> {
		// compressed
		// 	.iter()
		// 	.map(|col| {
		// 		// Select appropriate decompressor based on compression type
		// 		match col.compression {
		// 			reifydb_compression::CompressionType::Dictionary => {
		// 				let decompressor = DictionaryCompressor::new();
		// 				decompressor.decompress(col)
		// 			}
		// 			reifydb_compression::CompressionType::Delta => {
		// 				let decompressor = DeltaCompressor::new();
		// 				decompressor.decompress(col)
		// 			}
		// 			reifydb_compression::CompressionType::RunLength => {
		// 				let decompressor = RleCompressor::new();
		// 				decompressor.decompress(col)
		// 			}
		// 			reifydb_compression::CompressionType::BitPacking => {
		// 				let decompressor = BitPackCompressor::new();
		// 				decompressor.decompress(col)
		// 			}
		// 			reifydb_compression::CompressionType::Zstd => {
		// 				let decompressor = reifydb_compression::zstd::ZstdCompressor::new(3);
		// 				decompressor.decompress(col)
		// 			}
		// 			_ => {
		// 				unimplemented!()
		// 			}
		// 		}
		// 	})
		// 	.collect()
		todo!()
	}
}

impl Default for MemoryColumnStore {
	fn default() -> Self {
		Self::new()
	}
}

impl ColumnStore for MemoryColumnStore {
	fn insert(&self, version: CommitVersion, columns: Vec<CompressedColumn>) -> Result<()> {
		let key = PartitionKey::new(0, version);
		let partition = Partition::new(key.clone(), columns);

		self.partitions.insert(key, partition);
		Ok(())
	}

	fn scan(&self, version: CommitVersion, column_indices: &[usize]) -> Result<Vec<ColumnData>> {
		let key = PartitionKey::new(0, version);

		if let Some(entry) = self.partitions.get(&key) {
			let partition = entry.value();

			let selected_columns: Vec<_> =
				column_indices.iter().filter_map(|&idx| partition.columns.get(idx)).cloned().collect();

			self.decompress_columns(&selected_columns)
		} else {
			// No data for this version
			Ok(vec![])
		}
	}

	fn statistics(&self, column_index: usize) -> Option<ColumnStatistics> {
		let stats: Vec<_> = self
			.partitions
			.iter()
			.filter_map(|entry| entry.value().statistics.get(column_index).cloned())
			.collect();

		merge(&stats)
	}

	fn partition_count(&self) -> usize {
		self.partitions.len()
	}

	fn compressed_size(&self) -> usize {
		self.partitions.iter().map(|entry| entry.value().compressed_size).sum()
	}

	fn uncompressed_size(&self) -> usize {
		self.partitions.iter().map(|entry| entry.value().uncompressed_size).sum()
	}
}