use super::mapping::NativeReadonlyMapping;
use super::source::{PAGE_MEMO_BYTES, memo_bit};
use crate::lifecycle::native_graph::NativeReadLease;
use crate::property_graph::catalog::{
CatalogError, CatalogImage, GraphInterpretation, Symbol, SymbolEntry, SymbolHighWaters,
SymbolKind,
};
use crate::property_graph::query::runtime::RuntimeError;
use crate::property_graph::staging::BaseIdentity;
use crate::property_graph::storage::artifact::{
self, ArtifactControlError, ArtifactId, BlockKind, ContainerKind, FramedBlock, PhysicalRef,
ValidatedArtifact,
};
use crate::property_graph::storage::memory::{StorageBuffer, StorageMemory, StorageReservation};
use crate::property_graph::storage::participant::PreparationCatalog;
use crate::property_graph::storage::records::RecordCatalog;
use crate::property_graph::storage::stream::PayloadSlice;
use crate::property_graph::storage::tree::directory::{
BlockSource, RESERVED_PINNED_SLOTS, TreeError, TreeResources,
};
use std::{
cell::{Cell, OnceCell, RefCell},
ffi::OsString,
fmt::Write as _,
path::PathBuf,
};
struct PreparationMappedArtifact {
artifact: ArtifactId,
mapping: NativeReadonlyMapping,
validation: ValidatedArtifact,
validated_pages: [Cell<u8>; PAGE_MEMO_BYTES],
}
pub(crate) struct NativeArtifactWindow<'source, 'lease, 'm> {
source: &'source NativePreparationSource<'lease, 'm>,
artifact: ArtifactId,
mapping: &'source NativeReadonlyMapping,
validation: ValidatedArtifact,
}
pub(crate) fn charged_artifact_path<'m>(
memory: &'m StorageMemory<'m>,
directory: &std::path::Path,
artifact: ArtifactId,
) -> Result<(PathBuf, StorageReservation<'m>), TreeError> {
const FILENAME_BYTES: usize = b"graph-00000000000000000000000000000000.zgraph".len();
let path_upper = directory
.as_os_str()
.len()
.checked_add(1)
.and_then(|bytes| bytes.checked_add(FILENAME_BYTES))
.ok_or(TreeError::Memory)?;
let mut charge = memory.reserve(
path_upper
.checked_add(FILENAME_BYTES)
.ok_or(TreeError::Memory)?,
)?;
let mut filename = String::new();
filename
.try_reserve_exact(FILENAME_BYTES)
.map_err(|_| TreeError::Memory)?;
write!(&mut filename, "graph-{:032x}.zgraph", artifact.get()).map_err(|_| TreeError::Memory)?;
let mut raw = OsString::with_capacity(path_upper);
raw.push(directory.as_os_str());
let mut path = PathBuf::from(raw);
path.push(&filename);
let raw = path.into_os_string();
let path_capacity = raw.capacity();
charge.resize(
filename
.capacity()
.checked_add(path_capacity)
.ok_or(TreeError::Memory)?,
)?;
drop(filename);
charge.resize(path_capacity)?;
Ok((PathBuf::from(raw), charge))
}
pub(crate) struct NativePreparationSource<'lease, 'm> {
lease: &'lease NativeReadLease,
memory: &'m StorageMemory<'m>,
slots: StorageBuffer<'m, OnceCell<PreparationMappedArtifact>>,
scoped: Cell<bool>,
window: RefCell<Option<PreparationMappedArtifact>>,
retain_window: Cell<bool>,
filled: Cell<usize>,
#[cfg(any(test, feature = "test-seams"))]
slot_observation: crate::property_graph::storage::mapping_slot_capture::Observation,
}
impl<'lease, 'm> NativePreparationSource<'lease, 'm> {
pub(crate) fn charged_path(
memory: &'m StorageMemory<'m>,
directory: &std::path::Path,
artifact: ArtifactId,
) -> Result<(PathBuf, StorageReservation<'m>), TreeError> {
charged_artifact_path(memory, directory, artifact)
}
pub(crate) fn charged_candidate_path(
memory: &'m StorageMemory<'m>,
directory: &std::path::Path,
descriptor: crate::property_graph::wal::ArtifactDescriptor,
) -> Result<(PathBuf, StorageReservation<'m>), TreeError> {
charged_artifact_path(memory, directory, descriptor.artifact)
}
pub(crate) fn new(
lease: &'lease NativeReadLease,
memory: &'m StorageMemory<'m>,
capacity: usize,
) -> Result<Self, TreeError> {
lease
.check_active()
.map_err(RuntimeError::Value)
.map_err(TreeError::Runtime)?;
if capacity == 0 || !lease.belongs_to(memory.resources()) {
return Err(TreeError::Invalid(
"native preparation source accounting owner mismatch",
));
}
#[cfg(any(test, feature = "test-seams"))]
let capacity = crate::property_graph::storage::mapping_slot_capture::capacity(capacity);
let mut slots = StorageBuffer::new(memory, capacity)?;
for _ in 0..capacity {
slots.push(OnceCell::new())?;
}
Ok(Self {
lease,
memory,
slots,
scoped: Cell::new(false),
window: RefCell::new(None),
retain_window: Cell::new(false),
filled: Cell::new(0),
#[cfg(any(test, feature = "test-seams"))]
slot_observation:
crate::property_graph::storage::mapping_slot_capture::Observation::new(
crate::property_graph::storage::mapping_slot_capture::Kind::Preparation,
lease.bundle().base().generation.get(),
capacity,
),
})
}
pub(crate) fn new_scoped(
lease: &'lease NativeReadLease,
memory: &'m StorageMemory<'m>,
capacity: usize,
) -> Result<Self, TreeError> {
let source = Self::new(lease, memory, capacity)?;
source.scoped.set(true);
source.retain_window.set(true);
Ok(source)
}
pub(crate) fn new_windowed(
lease: &'lease NativeReadLease,
memory: &'m StorageMemory<'m>,
capacity: usize,
) -> Result<Self, TreeError> {
let source = Self::new_scoped(lease, memory, capacity)?;
source.scoped.set(false);
Ok(source)
}
pub(crate) fn resources(&self, work_limit: u64) -> Result<TreeResources<'m>, TreeError> {
TreeResources::for_prepare(self.memory, work_limit)
}
pub(crate) const fn lease(&self) -> &NativeReadLease {
self.lease
}
pub(crate) const fn memory(&self) -> &'m StorageMemory<'m> {
self.memory
}
fn check_owner(&self, resources: &mut TreeResources<'_>) -> Result<(), TreeError> {
self.lease
.check_active()
.map_err(RuntimeError::Value)
.map_err(TreeError::Runtime)?;
resources.require_preparation(self.memory)?;
resources.step(0)
}
fn page_bit(&self, reference: PhysicalRef) -> Result<Option<(&Cell<u8>, u8)>, TreeError> {
if self.scoped.get() {
return Ok(None);
}
let mapped = crate::property_graph::storage::mapping_slot(
self.slots.as_slice(),
reference.artifact,
|mapped| mapped.artifact,
|| Ok(()),
)?
.and_then(OnceCell::get);
mapped
.map(|mapped| memo_bit(&mapped.validated_pages, reference))
.transpose()
}
fn decode<'b>(
&self,
mapped: &'b PreparationMappedArtifact,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<FramedBlock<'b>, TreeError> {
resources.step(1)?;
let block = mapped
.validation
.framed_block(mapped.mapping.as_bytes(), reference)
.map_err(TreeError::Format)?;
let block = self.check_required(block)?;
Ok(block)
}
fn admit_mapping(
&self,
mapping: &NativeReadonlyMapping,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<ValidatedArtifact, TreeError> {
let publication = self.lease.publication().ok_or(TreeError::Invalid(
"native preparation publication is absent",
))?;
if let Some(validation) = publication
.validated_proof(reference.artifact)
.map_err(|_| TreeError::Invalid("native graph publication poisoned"))?
{
let identity = validation.identity();
let bundle = self.lease.bundle();
if identity.generation > bundle.base().generation
|| identity.creation_serial > bundle.high_waters().creation_serial
{
return Err(TreeError::Invalid(
"native preparation artifact is newer than admitted cutoff",
));
}
let block = validation
.checked_framed_block(mapping.as_bytes(), reference)
.map_err(TreeError::Format)?;
let _ = self.check_required(block)?;
return Ok(validation);
}
#[cfg(all(test, feature = "graph-cypher"))]
crate::property_graph::storage::preparation_work_capture::artifact(
reference.artifact,
mapping.as_bytes().len() as u64,
);
let bundle = self.lease.bundle();
let frame = artifact::decode_with_control(
ContainerKind::Object,
Some((bundle.base().store, reference.artifact)),
mapping.as_bytes(),
&mut |bytes| resources.step(bytes as u64),
)
.map_err(|error| match error {
ArtifactControlError::Format(error) => TreeError::Format(error),
ArtifactControlError::Control(error) => error,
})?;
let identity = frame.identity();
if identity.generation > bundle.base().generation
|| identity.creation_serial > bundle.high_waters().creation_serial
{
return Err(TreeError::Invalid(
"native preparation artifact is newer than admitted cutoff",
));
}
let block = frame.framed_block(reference).map_err(TreeError::Format)?;
let validation = frame.validation();
let _ = self.check_required(block)?;
publication
.remember_validated_proof(reference.artifact, validation)
.map_err(|_| TreeError::Invalid("native graph publication poisoned"))?;
Ok(validation)
}
fn check_required<'a>(&self, block: FramedBlock<'a>) -> Result<FramedBlock<'a>, TreeError> {
let bundle = self.lease.bundle();
let identity = block.identity();
if identity.store != bundle.base().store
|| identity.generation > bundle.base().generation
|| identity.creation_serial > bundle.high_waters().creation_serial
{
return Err(TreeError::Invalid("native preparation artifact cutoff"));
}
if let Some(required) = bundle.required_object(block.reference()) {
let expected = required.object;
let identity = block.identity();
if identity.store != expected.store
|| identity.artifact != expected.artifact
|| identity.generation != expected.generation
|| identity.creation_serial != expected.serial
|| block.file_length() != expected.bytes as usize
|| block.file_checksum() != expected.checksum
{
return Err(TreeError::Invalid(
"native preparation required descriptor mismatch",
));
}
}
Ok(block)
}
pub(crate) fn with_artifact_window<R>(
&self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
callback: impl for<'a, 'r> FnOnce(
&'a NativeArtifactWindow<'_, 'lease, 'm>,
&'r mut TreeResources<'_>,
) -> Result<R, TreeError>,
) -> Result<R, TreeError> {
self.check_owner(resources)?;
let slot = crate::property_graph::storage::mapping_slot(
self.slots.as_slice(),
reference.artifact,
|mapped| mapped.artifact,
|| resources.step(1),
)?;
if !self.scoped_blocks()
&& let Some(cell) = slot
&& cell.get().is_none()
{
let _ = self.slot_block(reference, resources)?;
}
if let Some(mapped) = slot.and_then(OnceCell::get) {
let window = NativeArtifactWindow {
source: self,
artifact: reference.artifact,
mapping: &mapped.mapping,
validation: mapped.validation,
};
let _ = window.resolve(reference, resources)?;
return callback(&window, resources);
}
if let Ok(retained) = self.window.try_borrow()
&& let Some(mapped) = retained.as_ref()
&& mapped.artifact == reference.artifact
{
let window = NativeArtifactWindow {
source: self,
artifact: reference.artifact,
mapping: &mapped.mapping,
validation: mapped.validation,
};
let _ = window.resolve(reference, resources)?;
return callback(&window, resources);
}
let (path, path_charge) = charged_artifact_path(
self.memory,
self.lease.bundle().directory(),
reference.artifact,
)?;
let file = self
.lease
.bundle()
.vfs()
.open_for_map(&path)
.map_err(TreeError::Io)?;
let mapping = NativeReadonlyMapping::open(file, &path, self.lease)?;
let validation = self.admit_mapping(&mapping, reference, resources)?;
let window = NativeArtifactWindow {
source: self,
artifact: reference.artifact,
mapping: &mapping,
validation,
};
let result = callback(&window, resources);
drop(mapping);
drop(path);
drop(path_charge);
result
}
pub(crate) fn copy_spill_page(
&self,
required: crate::property_graph::wal::RequiredRef,
target_generation: crate::property_graph::GraphGeneration,
output: &mut [u8],
resources: &mut TreeResources<'_>,
) -> Result<usize, TreeError> {
self.check_owner(resources)?;
let bundle = self.lease.bundle();
if required.object.store != bundle.base().store
|| required.object.generation != target_generation
|| required.object.family != crate::format::FormatFamily::NativeGraphObject.id()
|| required.object.version != 1
|| required.object.artifact != required.block.artifact
|| required.block.kind != BlockKind::CommitParticipant
|| required.block.version != 1
{
return Err(TreeError::Invalid("maintenance spill required reference"));
}
let slot = crate::property_graph::storage::mapping_slot(
self.slots.as_slice(),
required.object.artifact,
|mapped| mapped.artifact,
|| resources.step(1),
)?;
if let Some(mapped) = slot.and_then(OnceCell::get) {
return self.copy_validated_spill(mapped, required, output, resources);
}
let (path, path_charge) =
charged_artifact_path(self.memory, bundle.directory(), required.object.artifact)?;
let file = bundle.vfs().open_for_map(&path).map_err(TreeError::Io)?;
let mapping = NativeReadonlyMapping::open(file, &path, self.lease)?;
#[cfg(all(test, feature = "graph-cypher"))]
crate::property_graph::storage::preparation_work_capture::artifact(
required.object.artifact,
mapping.as_bytes().len() as u64,
);
let frame = artifact::decode_with_control(
ContainerKind::Object,
Some((required.object.store, required.object.artifact)),
mapping.as_bytes(),
&mut |bytes| resources.step(bytes as u64),
)
.map_err(|error| match error {
ArtifactControlError::Format(error) => TreeError::Format(error),
ArtifactControlError::Control(error) => error,
})?;
let mapped = PreparationMappedArtifact {
artifact: required.object.artifact,
validation: frame.validation(),
validated_pages: [const { Cell::new(0) }; PAGE_MEMO_BYTES],
mapping,
};
let length = self.copy_validated_spill(&mapped, required, output, resources)?;
if !self.slots_exhausted()
&& let Some(cell) = slot
{
cell.set(mapped)
.map_err(|_| TreeError::Invalid("spill slot initialized twice"))?;
self.filled.set(self.filled.get().saturating_add(1));
#[cfg(any(test, feature = "test-seams"))]
self.slot_observation.filled(self.filled.get());
}
drop(path);
drop(path_charge);
Ok(length)
}
fn copy_validated_spill(
&self,
mapped: &PreparationMappedArtifact,
required: crate::property_graph::wal::RequiredRef,
output: &mut [u8],
resources: &mut TreeResources<'_>,
) -> Result<usize, TreeError> {
let block = mapped
.validation
.framed_block(mapped.mapping.as_bytes(), required.block)
.map_err(TreeError::Format)?;
let identity = block.identity();
if identity.store != required.object.store
|| identity.artifact != required.object.artifact
|| identity.generation != required.object.generation
|| identity.creation_serial != required.object.serial
|| block.file_length() != required.object.bytes as usize
|| block.file_checksum() != required.object.checksum
|| block.reference() != required.block
{
return Err(TreeError::Invalid("maintenance spill descriptor mismatch"));
}
let payload = block.payload();
let target = output
.get_mut(..payload.len())
.ok_or(TreeError::Invalid("maintenance spill page exceeds scratch"))?;
resources.step(payload.len() as u64)?;
target.copy_from_slice(payload);
Ok(payload.len())
}
pub(crate) fn validate_object_descriptor(
&self,
descriptor: crate::property_graph::wal::ArtifactDescriptor,
resources: &mut TreeResources<'_>,
) -> Result<(), TreeError> {
self.check_owner(resources)?;
let bundle = self.lease.bundle();
if descriptor.store != bundle.base().store
|| descriptor.family != 17
|| descriptor.version != 1
{
return Err(TreeError::Invalid("inventory object descriptor domain"));
}
let (path, path_charge) =
Self::charged_candidate_path(self.memory, bundle.directory(), descriptor)?;
let file = bundle.vfs().open_for_map(&path).map_err(TreeError::Io)?;
let mapping = NativeReadonlyMapping::open(file, &path, self.lease)?;
#[cfg(all(test, feature = "graph-cypher"))]
crate::property_graph::storage::preparation_work_capture::artifact(
descriptor.artifact,
mapping.as_bytes().len() as u64,
);
let frame = artifact::decode_with_control(
ContainerKind::Object,
Some((descriptor.store, descriptor.artifact)),
mapping.as_bytes(),
&mut |bytes| resources.step(bytes as u64),
)
.map_err(|error| match error {
ArtifactControlError::Format(error) => TreeError::Format(error),
ArtifactControlError::Control(error) => error,
})?;
let identity = frame.identity();
let checksum = u64::from_le_bytes(
*mapping
.as_bytes()
.last_chunk::<8>()
.ok_or(TreeError::Invalid("inventory object checksum"))?,
);
if identity.store != descriptor.store
|| identity.artifact != descriptor.artifact
|| identity.generation != descriptor.generation
|| identity.creation_serial != descriptor.serial
|| mapping.as_bytes().len() != descriptor.bytes as usize
|| checksum != descriptor.checksum
{
return Err(TreeError::Invalid("inventory object descriptor mismatch"));
}
drop(mapping);
drop(path);
drop(path_charge);
Ok(())
}
}
impl NativePreparationSource<'_, '_> {
fn open_mapping(
&self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<PreparationMappedArtifact, TreeError> {
let (path, path_charge) = charged_artifact_path(
self.memory,
self.lease.bundle().directory(),
reference.artifact,
)?;
#[cfg(any(test, feature = "test-seams"))]
self.slot_observation.open(self.scoped_blocks());
let file = self
.lease
.bundle()
.vfs()
.open_for_map(&path)
.map_err(TreeError::Io)?;
let mapping = NativeReadonlyMapping::open(file, &path, self.lease)?;
drop(path);
drop(path_charge);
let validation = self.admit_mapping(&mapping, reference, resources)?;
Ok(PreparationMappedArtifact {
artifact: reference.artifact,
mapping,
validation,
validated_pages: [const { Cell::new(0) }; PAGE_MEMO_BYTES],
})
}
fn slot_hit<'a>(
&'a self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<Option<FramedBlock<'a>>, TreeError> {
self.check_owner(resources)?;
let slot = crate::property_graph::storage::mapping_slot(
self.slots.as_slice(),
reference.artifact,
|mapped| mapped.artifact,
|| resources.step(1),
)?;
if let Some(mapped) = slot.and_then(OnceCell::get) {
return self.decode(mapped, reference, resources).map(Some);
}
Ok(None)
}
fn slot_block<'a>(
&'a self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<Option<FramedBlock<'a>>, TreeError> {
if let Some(block) = self.slot_hit(reference, resources)? {
return Ok(Some(block));
}
let Some(cell) = crate::property_graph::storage::mapping_slot(
self.slots.as_slice(),
reference.artifact,
|mapped| mapped.artifact,
|| resources.step(1),
)?
else {
return Ok(None);
};
cell.set(self.open_mapping(reference, resources)?)
.map_err(|_| TreeError::Invalid("native preparation source slot initialized twice"))?;
self.filled.set(self.filled.get().saturating_add(1));
#[cfg(any(test, feature = "test-seams"))]
self.slot_observation.filled(self.filled.get());
let mapped = cell.get().ok_or(TreeError::Invalid(
"native preparation source slot remained empty",
))?;
self.decode(mapped, reference, resources).map(Some)
}
fn slots_exhausted(&self) -> bool {
self.slots
.as_slice()
.len()
.saturating_sub(self.filled.get())
<= RESERVED_PINNED_SLOTS
}
}
impl BlockSource for NativePreparationSource<'_, '_> {
fn page_validated(&self, reference: PhysicalRef) -> Result<bool, TreeError> {
Ok(self
.page_bit(reference)?
.is_some_and(|(byte, mask)| byte.get() & mask != 0))
}
fn mark_page_validated(&self, reference: PhysicalRef) -> Result<(), TreeError> {
if let Some((byte, mask)) = self.page_bit(reference)? {
byte.set(byte.get() | mask);
}
Ok(())
}
fn resolve<'a>(
&'a self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<FramedBlock<'a>, TreeError> {
#[cfg(any(test, feature = "test-seams"))]
self.slot_observation.resolve(self.slots_exhausted());
let block = self.slot_block(reference, resources)?;
block.ok_or(TreeError::Memory)
}
fn with_block<R>(
&self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
callback: impl for<'a, 'r> FnOnce(
FramedBlock<'a>,
&'r mut TreeResources<'_>,
) -> Result<R, TreeError>,
) -> Result<R, TreeError> {
if !self.scoped_blocks() {
let block = self.resolve(reference, resources)?;
return callback(block, resources);
}
self.check_owner(resources)?;
if self.retain_window.get() {
if let Some(block) = self.slot_hit(reference, resources)? {
return callback(block, resources);
}
if let Ok(window) = self.window.try_borrow()
&& let Some(mapped) = window.as_ref()
&& mapped.artifact == reference.artifact
{
let block = self.decode(mapped, reference, resources)?;
return callback(block, resources);
}
if let Ok(mut window) = self.window.try_borrow_mut() {
*window = Some(self.open_mapping(reference, resources)?);
drop(window);
let window = self
.window
.try_borrow()
.map_err(|_| TreeError::Invalid("scoped window is already borrowed"))?;
let mapped = window
.as_ref()
.ok_or(TreeError::Invalid("scoped window remained empty"))?;
let block = self.decode(mapped, reference, resources)?;
return callback(block, resources);
}
}
let mapped = self.open_mapping(reference, resources)?;
let block = self.decode(&mapped, reference, resources)?;
let result = callback(block, resources);
drop(mapped);
result
}
fn scoped_blocks(&self) -> bool {
self.scoped.get() || (self.retain_window.get() && self.slots_exhausted())
}
fn with_scoped_reads<R>(&self, body: impl FnOnce() -> R) -> R {
let previous_retain = self.retain_window.replace(true);
let result = body();
self.retain_window.set(previous_retain);
if !previous_retain && let Ok(mut window) = self.window.try_borrow_mut() {
*window = None;
}
result
}
}
impl BlockSource for NativeArtifactWindow<'_, '_, '_> {
fn resolve<'a>(
&'a self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
) -> Result<FramedBlock<'a>, TreeError> {
self.source.check_owner(resources)?;
if reference.artifact != self.artifact {
return Err(TreeError::Invalid("artifact window reference owner"));
}
resources.step(1)?;
let block = self
.validation
.framed_block(self.mapping.as_bytes(), reference)
.map_err(TreeError::Format)?;
self.source.check_required(block)
}
fn with_block<R>(
&self,
reference: PhysicalRef,
resources: &mut TreeResources<'_>,
callback: impl for<'a, 'r> FnOnce(
FramedBlock<'a>,
&'r mut TreeResources<'_>,
) -> Result<R, TreeError>,
) -> Result<R, TreeError> {
if reference.artifact == self.artifact {
let block = self.resolve(reference, resources)?;
callback(block, resources)
} else {
self.source.with_block(reference, resources, callback)
}
}
fn scoped_blocks(&self) -> bool {
true
}
}
pub(crate) struct NativePreparationCatalog<'source, 'lease, 'm> {
image: CatalogImage<'source>,
base: BaseIdentity,
source: &'source NativePreparationSource<'lease, 'm>,
_descriptors: StorageReservation<'m>,
}
impl<'source, 'lease, 'm> NativePreparationCatalog<'source, 'lease, 'm> {
pub(crate) fn open(
source: &'source NativePreparationSource<'lease, 'm>,
resources: &mut TreeResources<'_>,
) -> Result<Self, TreeError> {
let bundle = source.lease().bundle();
let required = bundle.catalog();
let block = source.resolve(required.block, resources)?;
let identity = block.identity();
if identity.store != required.object.store
|| identity.artifact != required.object.artifact
|| identity.generation != required.object.generation
|| identity.creation_serial != required.object.serial
|| block.reference() != required.block
|| block.file_length() != required.object.bytes as usize
|| block.file_checksum() != required.object.checksum
|| block.reference().kind != BlockKind::CommitParticipant
{
return Err(TreeError::Invalid("catalog required descriptor mismatch"));
}
let payload = block.payload();
let header = payload
.get(..8)
.ok_or(TreeError::Invalid("catalog participant header"))?;
if header.get(..4) != Some(b"ZGCP".as_slice())
|| u16::from_le_bytes(
*header
.get(4..6)
.and_then(|bytes| bytes.first_chunk::<2>())
.ok_or(TreeError::Invalid("catalog participant role"))?,
) != 1
|| u16::from_le_bytes(
*header
.get(6..8)
.and_then(|bytes| bytes.first_chunk::<2>())
.ok_or(TreeError::Invalid("catalog participant version"))?,
) != 1
{
return Err(TreeError::Invalid("catalog participant role or version"));
}
let encoded = payload
.get(8..)
.ok_or(TreeError::Invalid("catalog participant payload"))?;
let count = usize::try_from(u64::from_le_bytes(
*encoded
.get(104..112)
.and_then(|bytes| bytes.first_chunk::<8>())
.ok_or(TreeError::Invalid("catalog symbol count"))?,
))
.map_err(|_| TreeError::Memory)?;
let allowance = count
.checked_mul(std::mem::size_of::<SymbolEntry<'_>>())
.ok_or(TreeError::Memory)?;
let mut descriptors = source.memory().reserve(allowance)?;
let mut callback_error = None;
let image = CatalogImage::decode(encoded, allowance, &mut || {
resources.step(1).map_err(|error| {
if callback_error.is_none() {
callback_error = Some(error);
}
CatalogError::Cancelled
})
})
.map_err(|error| match error {
CatalogError::Cancelled => callback_error
.take()
.unwrap_or(TreeError::Invalid("catalog cancelled")),
CatalogError::Capacity | CatalogError::Allocation => TreeError::Memory,
_ => TreeError::Invalid("invalid native graph catalog"),
})?;
descriptors.resize(image.symbols.allocated_bytes())?;
let expected = GraphInterpretation::new(bundle.lexical(), bundle.document())
.map_err(|_| TreeError::Invalid("invalid admitted catalog interpretation"))?;
let mut callback_error = None;
image
.declaration
.validate_for(bundle.base().store, expected, &mut || {
resources.step(1).map_err(|error| {
if callback_error.is_none() {
callback_error = Some(error);
}
CatalogError::Cancelled
})
})
.map_err(|error| match error {
CatalogError::Cancelled => callback_error
.take()
.unwrap_or(TreeError::Invalid("catalog cancelled")),
CatalogError::Capacity | CatalogError::Allocation => TreeError::Memory,
_ => TreeError::Invalid("invalid native graph catalog"),
})?;
let high = bundle.high_waters();
let [label, relationship_type, property, namespace] = high.symbols;
if image.declaration.node_high_water != high.node
|| image.declaration.relationship_high_water != high.relationship
|| image.symbols.high_waters()
!= (SymbolHighWaters {
label,
relationship_type,
property,
namespace,
})
{
return Err(TreeError::Invalid("catalog allocator high-water mismatch"));
}
Ok(Self {
image,
base: bundle.base(),
source,
_descriptors: descriptors,
})
}
pub(crate) fn owns(&self, source: &NativePreparationSource<'_, '_>) -> bool {
std::ptr::eq(self.source, source)
}
pub(crate) fn relationship_rule(
&self,
id: crate::property_graph::catalog::RelTypeId,
resources: &mut TreeResources<'_>,
) -> Result<Option<crate::property_graph::catalog::OnDelete>, TreeError> {
for entry in self.image.symbols.entries() {
resources.step(1)?;
if entry.symbol == Symbol::RelationshipType(id) {
let mut failure = None;
let result = self.image.relationship_rules.lookup(entry.name, &mut || {
resources.step(1).map_err(|error| {
failure = Some(error);
CatalogError::Cancelled
})
});
if let Some(error) = failure {
return Err(error);
}
return result.map_err(|_| TreeError::Invalid("invalid relationship policy"));
}
}
Err(TreeError::Invalid("relationship type absent from catalog"))
}
pub(crate) fn relationship_rules(
&self,
) -> crate::property_graph::catalog::RelationshipRules<'source> {
self.image.relationship_rules
}
pub(crate) fn symbol_entries(&self) -> &[SymbolEntry<'source>] {
self.image.symbols.entries()
}
pub(crate) fn lookup_symbol(
&self,
kind: SymbolKind,
name: crate::property_graph::GraphName<'_>,
resources: &mut TreeResources<'_>,
) -> Result<Option<Symbol>, TreeError> {
self.image
.symbols
.lookup(kind, name, &mut || {
resources.step(1).map_err(|_| CatalogError::Cancelled)
})
.map_err(|error| catalog_error(error, resources))
}
}
fn catalog_error(error: CatalogError, resources: &mut TreeResources<'_>) -> TreeError {
if error == CatalogError::Cancelled {
resources
.step(0)
.err()
.unwrap_or(TreeError::Invalid("catalog cancelled"))
} else {
TreeError::Invalid("invalid native graph catalog")
}
}
impl<S: BlockSource> RecordCatalog<S> for NativePreparationCatalog<'_, '_, '_> {
fn resolve(
&self,
kind: SymbolKind,
name: PayloadSlice<'_, S>,
resources: &mut TreeResources<'_>,
) -> Result<Symbol, TreeError> {
name.validate_utf8(resources)?;
for entry in self.image.symbols.entries() {
resources.step(1)?;
if entry.symbol.kind() == kind
&& name.compare_bytes(entry.name.as_str().as_bytes(), resources)?
== std::cmp::Ordering::Equal
{
return Ok(entry.symbol);
}
}
Err(TreeError::Invalid(
"record name is absent from admitted catalog",
))
}
}
impl<S: BlockSource> PreparationCatalog<S> for NativePreparationCatalog<'_, '_, '_> {
fn namespace_id(
&self,
name: crate::property_graph::GraphName<'_>,
r: &mut TreeResources<'_>,
) -> Result<crate::property_graph::catalog::NamespaceId, TreeError> {
match self.lookup_symbol(SymbolKind::Namespace, name, r)? {
Some(Symbol::Namespace(id)) => Ok(id),
_ => Err(TreeError::Invalid("structured namespace absent")),
}
}
fn relationship_on_delete(
&self,
id: crate::property_graph::catalog::RelTypeId,
resources: &mut TreeResources<'_>,
) -> Result<Option<crate::property_graph::catalog::OnDelete>, TreeError> {
self.relationship_rule(id, resources)
}
fn base_identity(&self) -> BaseIdentity {
self.base
}
}
#[cfg(any(test, feature = "test-seams"))]
impl NativePreparationSource<'_, '_> {
pub(crate) fn mapping_slot_report(
&self,
) -> crate::property_graph::storage::mapping_slot_capture::Report {
self.slot_observation.report()
}
}