reifydb-catalog 0.5.6

Database catalog and metadata management 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
117
118
119
120
121
122
123
124
125
126
127
128

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

use std::sync::Arc;

use reifydb_core::{
	encoded::row::{EncodedRow, SHAPE_HEADER_SIZE},
	interface::change::Diff,
	row::Row,
	value::column::{buffer::pool::ColumnBufferPool, columns::Columns},
};
use reifydb_type::value::row_number::RowNumber;
use tracing::warn;

use crate::catalog::Catalog;

fn decode_row(catalog: &Catalog, row_number: RowNumber, row: EncodedRow) -> Option<Row> {
	if row.len() < SHAPE_HEADER_SIZE {
		warn!("EncodedRow too short for shape fingerprint ({} < {})", row.len(), SHAPE_HEADER_SIZE);
		return None;
	}
	let fingerprint = row.fingerprint();
	let shape = catalog.find_row_shape(fingerprint);
	match shape {
		Some(shape) => Some(Row {
			number: row_number,
			encoded: row,
			shape,
		}),
		None => {
			warn!(?fingerprint, "RowShape not found in cache for row decode");
			None
		}
	}
}

pub fn build_insert_diff(catalog: &Catalog, row_number: RowNumber, post: EncodedRow) -> Option<Diff> {
	let row = decode_row(catalog, row_number, post)?;
	Some(Diff::insert(Columns::from_row(&row)))
}

pub fn build_update_diff(catalog: &Catalog, row_number: RowNumber, pre: EncodedRow, post: EncodedRow) -> Option<Diff> {
	let pre_row = decode_row(catalog, row_number, pre)?;
	let post_row = decode_row(catalog, row_number, post)?;
	Some(Diff::update(Columns::from_row(&pre_row), Columns::from_row(&post_row)))
}

pub fn build_remove_diff(catalog: &Catalog, row_number: RowNumber, pre: EncodedRow) -> Option<Diff> {
	let row = decode_row(catalog, row_number, pre)?;
	Some(Diff::remove(Columns::from_row(&row)))
}

pub fn build_insert_diff_into(
	catalog: &Catalog,
	row_number: RowNumber,
	post: EncodedRow,
	post_buf: &mut Arc<Columns>,
) -> Option<Diff> {
	let row = decode_row(catalog, row_number, post)?;
	Arc::make_mut(post_buf).reset_from_row(&row);
	Some(Diff::insert_arc(post_buf.clone()))
}

pub fn build_update_diff_into(
	catalog: &Catalog,
	row_number: RowNumber,
	pre: EncodedRow,
	post: EncodedRow,
	pre_buf: &mut Arc<Columns>,
	post_buf: &mut Arc<Columns>,
) -> Option<Diff> {
	let pre_row = decode_row(catalog, row_number, pre)?;
	let post_row = decode_row(catalog, row_number, post)?;
	Arc::make_mut(pre_buf).reset_from_row(&pre_row);
	Arc::make_mut(post_buf).reset_from_row(&post_row);
	Some(Diff::update_arc(pre_buf.clone(), post_buf.clone()))
}

pub fn build_remove_diff_into(
	catalog: &Catalog,
	row_number: RowNumber,
	pre: EncodedRow,
	pre_buf: &mut Arc<Columns>,
) -> Option<Diff> {
	let row = decode_row(catalog, row_number, pre)?;
	Arc::make_mut(pre_buf).reset_from_row(&row);
	Some(Diff::remove_arc(pre_buf.clone()))
}

pub fn build_insert_diff_into_with_pool(
	catalog: &Catalog,
	row_number: RowNumber,
	post: EncodedRow,
	post_buf: &mut Arc<Columns>,
	pool: &ColumnBufferPool,
) -> Option<Diff> {
	let row = decode_row(catalog, row_number, post)?;
	Arc::make_mut(post_buf).reset_from_row_with_pool(&row, pool);
	Some(Diff::insert_arc(post_buf.clone()))
}

pub fn build_update_diff_into_with_pool(
	catalog: &Catalog,
	row_number: RowNumber,
	pre: EncodedRow,
	post: EncodedRow,
	pre_buf: &mut Arc<Columns>,
	post_buf: &mut Arc<Columns>,
	pool: &ColumnBufferPool,
) -> Option<Diff> {
	let pre_row = decode_row(catalog, row_number, pre)?;
	let post_row = decode_row(catalog, row_number, post)?;
	Arc::make_mut(pre_buf).reset_from_row_with_pool(&pre_row, pool);
	Arc::make_mut(post_buf).reset_from_row_with_pool(&post_row, pool);
	Some(Diff::update_arc(pre_buf.clone(), post_buf.clone()))
}

pub fn build_remove_diff_into_with_pool(
	catalog: &Catalog,
	row_number: RowNumber,
	pre: EncodedRow,
	pre_buf: &mut Arc<Columns>,
	pool: &ColumnBufferPool,
) -> Option<Diff> {
	let row = decode_row(catalog, row_number, pre)?;
	Arc::make_mut(pre_buf).reset_from_row_with_pool(&row, pool);
	Some(Diff::remove_arc(pre_buf.clone()))
}