mod decode;
mod row_encoding;
pub(super) use decode::{
bytes as decode_bytes_with_work_context, string as decode_string_with_work_context,
};
mod runtime;
pub(crate) use runtime::clone_string as clone_string_with_work_context;
pub(super) use runtime::{
rebuild_indexes as rebuild_indexes_with_work_context, row_key as row_key_with_work_context,
row_pages as row_pages_with_work_context,
validate_foreign_keys as validate_foreign_keys_with_work_context,
};
mod mount;
pub(crate) use mount::decode_relational_checkpoint_file_with_work_context;
pub(super) mod schema;
pub(crate) use schema::{decode_row_owned, decode_schema_owned};
mod validation;
pub(crate) use schema::decode_schema as decode_relational_table_schema_with_work_context;
pub(super) use validation::{
primary_key_positions_with_work_context, validate_row_with_work_context,
validate_table_schema_with_work_context,
};
use super::*;
use crate::background::{CheckpointBytes, CheckpointWorkContext, CheckpointWorkError};
pub(crate) fn encode_relational_table_schema_with_work_context(
schema: &RelationalTableSchema,
work: &CheckpointWorkContext,
) -> Result<Vec<u8>, RelationalError> {
validate_table_schema_with_work_context(schema, work)?;
let mut encoder = Encoder::default();
table_schema_with_work_context(&mut encoder, schema, work)?;
work.checkpoint().map_err(work_error)?;
Ok(encoder.finish())
}
pub(crate) fn validate_relational_table_schema_codec_shape_with_work_context(
schema: &RelationalTableSchema,
max_schema_items: usize,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
validate_table_schema_with_work_context(schema, work)?;
for (context, count) in [
("columns", schema.columns.len()),
("primary-key columns", schema.primary_key.len()),
("unique constraints", schema.unique_constraints.len()),
("foreign keys", schema.foreign_keys.len()),
("indexes", schema.indexes.len()),
] {
let unit = work.start_unit().map_err(work_error)?;
if count > max_schema_items {
return Err(RelationalError::Admission(format!(
"relational table schema contains {count} {context}, exceeding limit {max_schema_items}"
)));
}
unit.finish();
}
work.checkpoint().map_err(work_error)
}
pub(crate) fn encode_relational_row_payload_with_work_context(
row: &RelationalRow,
work: &CheckpointWorkContext,
) -> Result<CheckpointBytes, RelationalError> {
row_encoding::encode(row, work)
}
pub(super) fn work_error(error: CheckpointWorkError) -> RelationalError {
RelationalError::Admission(error.to_string())
}
fn encode_value_with_work_context(
encoder: &mut Encoder,
value: &RelationalValue,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
let unit = work.start_unit().map_err(work_error)?;
let variable = match value {
RelationalValue::Text(text) => Some((4, text.as_bytes())),
RelationalValue::Bytea(bytes) => Some((5, bytes.as_slice())),
RelationalValue::Vector(value) => Some((8, value.encoded())),
_ => None,
};
if let Some((tag, bytes)) = variable {
encoder.value_count = encoder.value_count.checked_add(1).ok_or_else(|| {
RelationalError::Admission("encoded value count overflow".to_string())
})?;
encoder.u8(tag);
encoder.u64(u64::try_from(bytes.len()).map_err(|_| {
RelationalError::Admission("durable byte string is too large".to_string())
})?);
unit.finish();
for chunk in bytes.chunks(64 * 1024) {
let unit = work.start_unit().map_err(work_error)?;
encoder.bytes.extend_from_slice(chunk);
unit.finish();
}
} else {
encoder.value(value)?;
unit.finish();
}
work.checkpoint().map_err(work_error)
}
#[derive(Clone, Copy)]
pub(crate) enum CheckpointOutputIo {
#[cfg(test)]
Memory,
File,
}
impl CheckpointOutputIo {
fn wave(
self,
work: &CheckpointWorkContext,
) -> Result<Option<Box<dyn hawdb_core::RuntimeIoWavePermit>>, RelationalError> {
match self {
#[cfg(test)]
Self::Memory => Ok(None),
Self::File => work.io_wave().map_err(work_error),
}
}
fn seek<W: Seek>(
self,
writer: &mut W,
position: SeekFrom,
operation: &str,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
let unit = work.start_unit().map_err(work_error)?;
let _wave = self.wave(work)?;
writer
.seek(position)
.map_err(|error| RelationalError::from_io(operation, error))?;
unit.finish();
work.checkpoint().map_err(work_error)
}
fn write<W: Write>(
self,
writer: &mut W,
bytes: &[u8],
operation: &str,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
for block in bytes.chunks(64 * 1024) {
let unit = work.start_unit().map_err(work_error)?;
let _wave = self.wave(work)?;
writer
.write_all(block)
.map_err(|error| RelationalError::from_io(operation, error))?;
unit.finish();
}
work.checkpoint().map_err(work_error)
}
}
fn string_with_work_context(
encoder: &mut Encoder,
value: &str,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
let unit = work.start_unit().map_err(work_error)?;
encoder.u64(
u64::try_from(value.len())
.map_err(|_| RelationalError::Admission("durable byte string is too large".into()))?,
);
unit.finish();
for block in value.as_bytes().chunks(64 * 1024) {
let unit = work.start_unit().map_err(work_error)?;
encoder.bytes.extend_from_slice(block);
unit.finish();
}
work.checkpoint().map_err(work_error)
}
fn table_schema_with_work_context(
encoder: &mut Encoder,
schema: &RelationalTableSchema,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
struct Controlled<'a> {
encoder: &'a mut Encoder,
work: &'a CheckpointWorkContext,
}
impl schema_encoding::Sink for Controlled<'_> {
fn raw(&mut self, bytes: &[u8]) -> Result<(), RelationalError> {
for block in bytes.chunks(64 * 1024) {
let unit = self.work.start_unit().map_err(work_error)?;
self.encoder.bytes.extend_from_slice(block);
unit.finish();
}
self.work.checkpoint().map_err(work_error)
}
fn value(&mut self, value: &RelationalValue) -> Result<(), RelationalError> {
encode_value_with_work_context(self.encoder, value, self.work)
}
}
schema_encoding::schema(&mut Controlled { encoder, work }, schema.into())
}
pub(super) fn validate_reachability(
state: &RelationalState,
work: &CheckpointWorkContext,
) -> Result<(), RelationalError> {
let mut reachable = BTreeSet::new();
for segment in state.segments.values() {
for row in segment.rows.values() {
for value in row.values() {
let unit = work.start_unit().map_err(work_error)?;
if let RelationalValue::Overflow(reference) = value {
reachable.insert(&reference.digest);
}
unit.finish();
}
}
work.checkpoint().map_err(work_error)?;
}
for digest in &reachable {
let unit = work.start_unit().map_err(work_error)?;
if !state.overflow_segments.contains_key(*digest) {
return Err(RelationalError::Corruption(format!(
"checkpoint row references missing overflow segment {digest}"
)));
}
unit.finish();
}
if reachable.len() != state.overflow_segments.len() {
return Err(RelationalError::Corruption(
"checkpoint contains unreachable overflow segments".into(),
));
}
work.checkpoint().map_err(work_error)
}
enum OverflowBytes<'a> {
Borrowed(&'a [u8]),
Owned(CheckpointBytes),
}
impl std::ops::Deref for OverflowBytes<'_> {
type Target = [u8];
fn deref(&self) -> &Self::Target {
match self {
Self::Borrowed(bytes) => bytes,
Self::Owned(bytes) => bytes,
}
}
}
fn overflow_with_work_context<'a>(
segment: &'a RelationalOverflowSegment,
work: &CheckpointWorkContext,
) -> Result<OverflowBytes<'a>, RelationalError> {
work.checkpoint().map_err(work_error)?;
match segment {
RelationalOverflowSegment::Inline(bytes) => Ok(OverflowBytes::Borrowed(bytes)),
RelationalOverflowSegment::CheckpointInline(bytes) => Ok(OverflowBytes::Borrowed(bytes)),
RelationalOverflowSegment::FileRange { reader, range } => reader
.checkpoint_range_with_work_context(range, work)
.map(OverflowBytes::Owned)
.map_err(|error| match error {
crate::scan::CheckpointRangeReadError::Read(error) => RelationalError::Corruption(
format!("failed to read file-backed overflow segment: {error}"),
),
crate::scan::CheckpointRangeReadError::Work(error) => work_error(error),
}),
}
}
struct Payload<'a, W> {
writer: &'a mut W,
max_payload_bytes: usize,
payload_bytes: usize,
hasher: IntegrityHasher,
io: CheckpointOutputIo,
work: &'a CheckpointWorkContext,
}
impl<W: Write> schema_encoding::Sink for Payload<'_, W> {
fn raw(&mut self, bytes: &[u8]) -> Result<(), RelationalError> {
self.write_all(bytes)
}
}
impl<'a, W: Write> Payload<'a, W> {
fn write_all(&mut self, bytes: &[u8]) -> Result<(), RelationalError> {
let unit = self.work.start_unit().map_err(work_error)?;
let next = self.payload_bytes.checked_add(bytes.len()).ok_or_else(|| {
RelationalError::Admission("relational checkpoint size overflow".into())
})?;
if next > self.max_payload_bytes {
return Err(RelationalError::Admission(format!(
"relational checkpoint payload contains {next} bytes, exceeding limit {}",
self.max_payload_bytes
)));
}
unit.finish();
for block in bytes.chunks(64 * 1024) {
let unit = self.work.start_unit().map_err(work_error)?;
let _wave = self.io.wave(self.work)?;
self.writer.write_all(block).map_err(|error| {
RelationalError::from_io("failed to stream relational checkpoint payload", error)
})?;
self.hasher.update(block);
self.payload_bytes += block.len();
unit.finish();
}
self.work.checkpoint().map_err(work_error)
}
}
pub(crate) fn encode_relational_checkpoint_with_work_context<W: Write + Seek>(
writer: &mut W,
epoch: u64,
state: &RelationalState,
max_record_bytes: usize,
io: CheckpointOutputIo,
work: &CheckpointWorkContext,
) -> Result<u64, RelationalError> {
validate_reachability(state, work)?;
if max_record_bytes < HEADER_BYTES {
return Err(RelationalError::Admission(format!(
"relational checkpoint max_record_bytes {max_record_bytes} is smaller than its header"
)));
}
io.seek(
writer,
SeekFrom::Start(HEADER_BYTES as u64),
"failed to reserve relational checkpoint header",
work,
)?;
let mut payload = Payload {
writer,
max_payload_bytes: max_record_bytes - HEADER_BYTES,
payload_bytes: 0,
hasher: IntegrityHasher::new(),
io,
work,
};
let unit = work.start_unit().map_err(work_error)?;
let mut tables = Encoder::default();
tables.count(state.schemas.len(), "checkpoint tables")?;
unit.finish();
payload.write_all(&tables.finish())?;
for (name, schema) in &state.schemas {
let segment = state.segments.get(name).ok_or_else(|| {
RelationalError::Corruption(format!("table {name} is missing its row segment"))
})?;
let mut table = Encoder::default();
string_with_work_context(&mut table, name, work)?;
table_schema_with_work_context(&mut table, schema, work)?;
let unit = work.start_unit().map_err(work_error)?;
table.count(segment.rows.len(), "checkpoint rows")?;
unit.finish();
payload.write_all(&table.finish())?;
for row in segment.rows.values() {
schema_encoding::row(&mut payload, row)?;
}
}
let unit = work.start_unit().map_err(work_error)?;
let mut count = Encoder::default();
count.count(
state.overflow_segments.len(),
"checkpoint overflow segments",
)?;
unit.finish();
payload.write_all(&count.finish())?;
for (digest, segment) in &state.overflow_segments {
let envelope = overflow_with_work_context(segment, work)?;
if work.integrity(&envelope).map_err(work_error)?.sha256 != *digest {
return Err(RelationalError::Corruption(format!(
"relational checkpoint overflow segment {digest} has an invalid digest"
)));
}
let unit = work.start_unit().map_err(work_error)?;
let mut metadata = Encoder::default();
metadata.sha256(*digest);
metadata.u64(u64::try_from(envelope.len()).map_err(|_| {
RelationalError::Admission("overflow envelope length does not fit u64".into())
})?);
unit.finish();
payload.write_all(&metadata.finish())?;
payload.write_all(&envelope)?;
}
let unit = work.start_unit().map_err(work_error)?;
let Payload {
writer,
payload_bytes: payload_len,
hasher,
..
} = payload;
let total_len = HEADER_BYTES
.checked_add(payload_len)
.ok_or_else(|| RelationalError::Admission("checkpoint size overflow".into()))?;
let mut header = Vec::with_capacity(HEADER_BYTES);
encode_envelope_header(
&mut header,
CHECKPOINT_MAGIC,
epoch,
payload_len as u64,
hasher.finish(),
);
unit.finish();
io.seek(
writer,
SeekFrom::Start(0),
"failed to seek relational checkpoint header",
work,
)?;
io.write(
writer,
&header,
"failed to write relational checkpoint header",
work,
)?;
io.seek(
writer,
SeekFrom::Start(total_len as u64),
"failed to finish relational checkpoint",
work,
)?;
work.checkpoint().map_err(work_error)?;
Ok(total_len as u64)
}
#[cfg(test)]
mod tests;