use reifydb_codec::{
encoded::row::EncodedRow,
key::encoded::{EncodedKey, EncodedKeyRange},
};
use reifydb_value::{Result, util::cowvec::CowVec};
use crate::{
common::CommitVersion,
delta::Delta,
interface::catalog::{flow::FlowNodeId, shape::ShapeId},
key::{
EncodableKeyRange, Key, flow_node_internal_state::FlowNodeInternalStateKeyRange,
flow_node_state::FlowNodeStateKeyRange, kind::KeyKind, partitioned_row::PartitionedRowKeyRange,
row::RowKeyRange,
},
};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum Tier {
Buffer,
Persistent,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum EntryKind {
Multi,
Source(ShapeId),
PartitionedSource(ShapeId),
Operator(FlowNodeId),
OperatorInternal(FlowNodeId),
}
pub fn classify_key(key: &EncodedKey) -> EntryKind {
match Key::decode(key) {
Some(Key::Row(row_key)) => EntryKind::Source(row_key.shape),
Some(Key::PartitionedRow(partitioned_key)) => EntryKind::PartitionedSource(partitioned_key.shape),
Some(Key::FlowNodeState(state_key)) => EntryKind::Operator(state_key.node),
Some(Key::FlowNodeInternalState(internal_key)) => EntryKind::OperatorInternal(internal_key.node),
_ => EntryKind::Multi,
}
}
pub fn is_single_version_semantics_key(key: &EncodedKey) -> bool {
Key::kind(key).is_some_and(|kind| matches!(kind, KeyKind::FlowNodeState | KeyKind::FlowNodeInternalState))
}
pub fn classify_range(range: &EncodedKeyRange) -> Option<EntryKind> {
if let (Some(start), Some(_end)) = RowKeyRange::decode(range) {
return Some(EntryKind::Source(start.shape));
}
if let (Some(start), Some(_end)) = PartitionedRowKeyRange::decode(range) {
return Some(EntryKind::PartitionedSource(start.shape));
}
if let (Some(start), Some(_end)) = FlowNodeStateKeyRange::decode(range) {
return Some(EntryKind::Operator(start.node));
}
if let (Some(start), Some(_end)) = FlowNodeInternalStateKeyRange::decode(range) {
return Some(EntryKind::OperatorInternal(start.node));
}
None
}
#[derive(Debug, Clone)]
pub struct MultiVersionRow {
pub key: EncodedKey,
pub row: EncodedRow,
pub version: CommitVersion,
}
#[derive(Debug, Clone)]
pub struct SingleVersionRow {
pub key: EncodedKey,
pub row: EncodedRow,
}
#[derive(Debug, Clone)]
pub struct MultiVersionBatch {
pub items: Vec<MultiVersionRow>,
pub has_more: bool,
}
impl MultiVersionBatch {
pub fn empty() -> Self {
Self {
items: Vec::new(),
has_more: false,
}
}
pub fn is_empty(&self) -> bool {
self.items.is_empty()
}
}
pub trait MultiVersionCommit: Send + Sync {
fn commit(&self, deltas: CowVec<Delta>, version: CommitVersion) -> Result<()>;
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ReadOptions {
pub bypass_buffer: bool,
}
pub trait MultiVersionGet: Send + Sync {
fn get(&self, key: &EncodedKey, version: CommitVersion) -> Result<Option<MultiVersionRow>>;
fn get_with_options(
&self,
key: &EncodedKey,
version: CommitVersion,
_options: ReadOptions,
) -> Result<Option<MultiVersionRow>> {
self.get(key, version)
}
}
pub trait MultiVersionContains: Send + Sync {
fn contains(&self, key: &EncodedKey, version: CommitVersion) -> Result<bool>;
fn contains_with_options(
&self,
key: &EncodedKey,
version: CommitVersion,
_options: ReadOptions,
) -> Result<bool> {
self.contains(key, version)
}
}
pub trait MultiVersionGetPrevious: Send + Sync {
fn get_previous_version(
&self,
key: &EncodedKey,
before_version: CommitVersion,
) -> Result<Option<MultiVersionRow>>;
}
pub trait MultiVersionStore:
Send + Sync + Clone + MultiVersionCommit + MultiVersionGet + MultiVersionGetPrevious + MultiVersionContains + 'static
{
}
#[derive(Debug, Clone)]
pub struct SingleVersionBatch {
pub items: Vec<SingleVersionRow>,
pub has_more: bool,
}
impl SingleVersionBatch {
pub fn empty() -> Self {
Self {
items: Vec::new(),
has_more: false,
}
}
pub fn is_empty(&self) -> bool {
self.items.is_empty()
}
}
pub trait SingleVersionCommit: Send + Sync {
fn commit(&mut self, deltas: CowVec<Delta>) -> Result<()>;
}
pub trait SingleVersionGet: Send + Sync {
fn get(&self, key: &EncodedKey) -> Result<Option<SingleVersionRow>>;
}
pub trait SingleVersionContains: Send + Sync {
fn contains(&self, key: &EncodedKey) -> Result<bool>;
}
pub trait SingleVersionSet: SingleVersionCommit {
fn set(&mut self, key: &EncodedKey, row: EncodedRow) -> Result<()> {
Self::commit(
self,
CowVec::new(vec![Delta::Set {
key: key.clone(),
row: row.clone(),
}]),
)
}
}
pub trait SingleVersionRemove: SingleVersionCommit {
fn unset(&mut self, key: &EncodedKey, row: EncodedRow) -> Result<()> {
Self::commit(
self,
CowVec::new(vec![Delta::Unset {
key: key.clone(),
row,
}]),
)
}
fn remove(&mut self, key: &EncodedKey) -> Result<()> {
Self::commit(
self,
CowVec::new(vec![Delta::Remove {
key: key.clone(),
}]),
)
}
}
pub trait SingleVersionRange: Send + Sync {
fn range_batch(&self, range: EncodedKeyRange, batch_size: u64) -> Result<SingleVersionBatch>;
fn range(&self, range: EncodedKeyRange) -> Result<SingleVersionBatch> {
self.range_batch(range, 1024)
}
fn prefix(&self, prefix: &EncodedKey) -> Result<SingleVersionBatch> {
self.range(EncodedKeyRange::prefix(prefix))
}
}
pub trait SingleVersionRangeRev: Send + Sync {
fn range_rev_batch(&self, range: EncodedKeyRange, batch_size: u64) -> Result<SingleVersionBatch>;
fn range_rev(&self, range: EncodedKeyRange) -> Result<SingleVersionBatch> {
self.range_rev_batch(range, 1024)
}
fn prefix_rev(&self, prefix: &EncodedKey) -> Result<SingleVersionBatch> {
self.range_rev(EncodedKeyRange::prefix(prefix))
}
}
pub trait SingleVersionStore:
Send
+ Sync
+ Clone
+ SingleVersionCommit
+ SingleVersionGet
+ SingleVersionContains
+ SingleVersionSet
+ SingleVersionRemove
+ SingleVersionRange
+ SingleVersionRangeRev
+ 'static
{
}
#[cfg(test)]
mod tests {
use reifydb_value::value::{Value, partition::Partition, row_number::RowNumber};
use super::{EntryKind, classify_key, classify_range};
use crate::{
interface::catalog::{id::TableId, shape::ShapeId},
key::{
partitioned_row::{PartitionedRowKey, RowLocator},
row::RowKey,
},
};
fn part(v: &str) -> Partition {
Partition::of(&[Value::Utf8(v.to_string())])
}
#[test]
fn classify_key_partitioned_row_is_partitioned_source() {
let shape = ShapeId::Table(TableId(7));
let key = PartitionedRowKey::encoded(shape, part("us"), RowLocator::Row(RowNumber(1)));
assert_eq!(classify_key(&key), EntryKind::PartitionedSource(shape));
}
#[test]
fn classify_key_row_is_still_source() {
let shape = ShapeId::Table(TableId(7));
let key = RowKey::encoded(shape, RowNumber(1));
assert_eq!(classify_key(&key), EntryKind::Source(shape));
}
#[test]
fn classify_range_all_partition_forms_are_partitioned_source() {
let shape = ShapeId::Table(TableId(9));
let p = part("us");
let last = PartitionedRowKey::encoded(shape, p, RowLocator::Row(RowNumber(5)));
assert_eq!(
classify_range(&PartitionedRowKey::partition_range(shape, p)),
Some(EntryKind::PartitionedSource(shape))
);
assert_eq!(
classify_range(&PartitionedRowKey::partition_scan_range(shape, p, Some(&last))),
Some(EntryKind::PartitionedSource(shape))
);
assert_eq!(
classify_range(&PartitionedRowKey::scan_range(shape, None)),
Some(EntryKind::PartitionedSource(shape))
);
assert_eq!(
classify_range(&PartitionedRowKey::full_scan(shape)),
Some(EntryKind::PartitionedSource(shape))
);
}
#[test]
fn classify_range_row_range_is_still_source() {
let shape = ShapeId::Table(TableId(9));
assert_eq!(classify_range(&RowKey::full_scan(shape)), Some(EntryKind::Source(shape)));
}
}