use super::*;
pub(super) struct Catalog {
pub header: f::NativePagedMetadataHeader,
pub roots: Vec<f::NativePagedCollectionRoot>,
references: Vec<f::NativePagedCrossRef>,
epoch: Option<u64>,
pub resume: Option<f::native_resume::CatalogResume>,
}
impl Catalog {
pub fn read(source: &Source) -> Result<Self> {
let page = source.raw_page(1, true)?;
ensure(
page.page_type()? == PageType::Header,
"invalid catalog page",
)?;
let mut data = page.content().to_vec();
if let Some(header) = f::decode_native_metadata_overflow_header(&data)? {
let total = header.total_payload_bytes as usize;
ensure(
total > 0 && total <= 64 * 1024 * 1024,
"catalog exceeds compaction memory budget",
)?;
let first = total.min(data.len() - 16);
let mut payload = data[16..16 + first].to_vec();
let mut next = header.next_overflow_page_id;
while next != 0 {
source.claim(next)?;
let page = source.raw_page(next, true)?;
ensure(
page.page_type()? == PageType::Overflow,
"invalid catalog continuation",
)?;
let continuation =
f::decode_native_metadata_overflow_continuation_header(page.content())?;
let length = continuation.chunk_bytes as usize;
ensure(
length > 0
&& length <= PAGE_SIZE - HEADER_SIZE - 8
&& payload.len() + length <= total,
"invalid catalog continuation length",
)?;
payload.extend_from_slice(&page.content()[8..8 + length]);
next = continuation.next_overflow_page_id;
}
ensure(payload.len() == total, "incomplete catalog")?;
data = payload;
}
let header = f::decode_native_paged_metadata_header(&data)?
.ok_or("source is not a native paged catalog")?;
ensure(
matches!(header.format_version, 11 | 12),
"unsupported native record format",
)?;
let mut position = f::METADATA_HEADER_BYTES;
ensure(
header.collection_count as usize <= data.len() / 8,
"invalid catalog count",
)?;
let mut roots = Vec::new();
let mut names = std::collections::BTreeSet::new();
for _ in 0..header.collection_count {
let root = f::decode_native_paged_collection_root(&data, &mut position)?;
ensure(
names.insert(root.collection.clone()),
"duplicate collection",
)?;
roots.push(root);
}
let count = number(&data, position)?;
position += 4;
ensure(
count as usize <= data.len().saturating_sub(position) / 21,
"invalid reference count",
)?;
let mut references = Vec::new();
for _ in 0..count {
references.push(f::decode_native_paged_cross_ref(&data, &mut position)?);
}
f::decode_native_paged_retired_trees(&data, &mut position)?;
f::decode_native_paged_deferred_pages(&data, &mut position)?;
let epoch = f::native_store::decode_native_paged_ownership_epoch(&data, &mut position)?;
let resume = f::native_resume::decode_catalog_resume(&data)?;
if let Some(resume) = &resume {
let mut canonical = Vec::new();
f::native_resume::encode_catalog_resume(&mut canonical, resume)?;
ensure(
data.get(position..position + canonical.len()) == Some(canonical.as_slice()),
"unsupported/noncanonical resume trailer",
)?;
position += canonical.len();
}
ensure(
data[position..].iter().all(|byte| *byte == 0),
"unknown catalog extension",
)?;
Ok(Self {
header,
roots,
references,
epoch,
resume,
})
}
pub fn encode(&self, roots: &BTreeMap<u32, u32>) -> Result<Vec<u8>> {
let remap = |root: u32| -> Result<u32> {
if root == 0 {
Ok(0)
} else {
roots
.get(&root)
.copied()
.ok_or_else(|| "missing rewritten tree".into())
}
};
let mut data = Vec::new();
f::encode_native_paged_metadata_header(&mut data, self.header);
for root in &self.roots {
f::encode_native_paged_collection_root(
&mut data,
&root.collection,
remap(root.root_page)?,
);
}
data.extend_from_slice(&u32::try_from(self.references.len())?.to_le_bytes());
for reference in &self.references {
f::encode_native_paged_cross_ref(
&mut data,
reference.source_id,
reference.target_id,
reference.ref_type,
&reference.target_collection,
);
}
f::encode_native_paged_retired_trees(&mut data, &[])?;
f::encode_native_paged_deferred_pages(&mut data, &[])?;
if let Some(epoch) = self.epoch {
f::native_store::encode_native_paged_ownership_epoch(&mut data, epoch)?;
}
if let Some(mut resume) = self.resume.clone() {
for state in resume.collections.values_mut() {
state.root_page = remap(state.root_page)?;
state.key_index_root = state.key_index_root.map(remap).transpose()?;
for index in state.secondary_indices.values_mut() {
index.root_page = remap(index.root_page)?;
}
}
f::native_resume::encode_catalog_resume(&mut data, &resume)?;
}
Ok(data)
}
}
pub(super) fn write_catalog(target: &Pager, data: &[u8]) -> Result<()> {
let mut root = Page::new(PageType::Header, 1);
if data.len() <= PAGE_SIZE - HEADER_SIZE {
root.content_mut()[..data.len()].copy_from_slice(data);
} else {
let first = PAGE_SIZE - HEADER_SIZE - 16;
let chunks: Vec<_> = data[first..].chunks(PAGE_SIZE - HEADER_SIZE - 8).collect();
let mut next = 0;
for chunk in chunks.into_iter().rev() {
let mut page = target.allocate_page(PageType::Overflow)?;
f::encode_native_metadata_overflow_continuation_header(
page.content_mut(),
f::NativeMetadataOverflowContinuationHeader {
next_overflow_page_id: next,
chunk_bytes: u32::try_from(chunk.len())?,
},
)?;
page.content_mut()[8..8 + chunk.len()].copy_from_slice(chunk);
next = page.page_id();
target.write_page(next, page)?;
}
f::encode_native_metadata_overflow_header(
root.content_mut(),
f::NativeMetadataOverflowHeader {
format_version: number(data, 4)?,
total_payload_bytes: u32::try_from(data.len())?,
next_overflow_page_id: next,
},
)?;
root.content_mut()[16..16 + first].copy_from_slice(&data[..first]);
}
target.sync()?;
target.write_page(1, root)?;
Ok(())
}