use super::batch::BatchWriteCollector;
use super::boundary::is_boundary;
use super::cid::Cid;
use super::error::Error;
use super::node::Node;
use super::store::Store;
use super::Prolly;
fn reserve_node_entries(node: &mut Node, additional: usize) {
node.keys.reserve_exact(additional);
node.vals.reserve_exact(additional);
}
pub fn rebalance<S: Store>(
prolly: &Prolly<S>,
node: Node,
ancestors: &[(Node, usize)],
) -> Result<Cid, Error> {
if node.is_empty() {
return handle_empty_node(prolly, ancestors);
}
if node.len() > node.max_chunk_size() && node.len() > 1 {
return split_node(prolly, node, ancestors);
}
if !ancestors.is_empty() && node.len() < node.min_chunk_size() {
if let Some(merged) = try_merge_with_sibling(prolly, &node, ancestors)? {
return Ok(merged);
}
}
let cid = prolly.save(&node)?;
if ancestors.is_empty() {
return Ok(cid);
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
if !node.keys.is_empty() {
parent.keys[idx] = node.keys[0].clone();
}
parent.vals[idx] = cid.0.to_vec();
rebalance(prolly, parent, &ancestors[..ancestors.len() - 1])
}
fn handle_empty_node<S: Store>(
prolly: &Prolly<S>,
ancestors: &[(Node, usize)],
) -> Result<Cid, Error> {
if ancestors.is_empty() {
let empty_node = prolly.new_leaf_node();
return prolly.save(&empty_node);
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
parent.keys.remove(idx);
parent.vals.remove(idx);
if parent.is_empty() && ancestors.len() == 1 {
let empty_node = prolly.new_leaf_node();
return prolly.save(&empty_node);
}
rebalance(prolly, parent, &ancestors[..ancestors.len() - 1])
}
fn split_node<S: Store>(
prolly: &Prolly<S>,
node: Node,
ancestors: &[(Node, usize)],
) -> Result<Cid, Error> {
let max_size = node.max_chunk_size();
let mut split_idx = None;
for i in 0..node.len() {
if is_boundary(&node, i) {
let left_size = i + 1;
let right_size = node.len() - i - 1;
if left_size <= max_size && right_size > 0 && right_size <= max_size {
split_idx = Some(i);
break;
}
}
}
let split_idx = split_idx.unwrap_or_else(|| {
let min_split = node.len().saturating_sub(max_size + 1);
let max_split = max_size.saturating_sub(1).min(node.len().saturating_sub(2));
if min_split <= max_split {
(min_split + max_split) / 2
} else {
node.len() / 2
}
});
let split_idx = split_idx.min(node.len().saturating_sub(2));
let mut left = prolly.new_node_like(&node);
left.keys = node.keys[..=split_idx].to_vec();
left.vals = node.vals[..=split_idx].to_vec();
let mut right = prolly.new_node_like(&node);
right.keys = node.keys[split_idx + 1..].to_vec();
right.vals = node.vals[split_idx + 1..].to_vec();
if right.is_empty() {
let cid = prolly.save(&node)?;
if ancestors.is_empty() {
return Ok(cid);
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
parent.vals[idx] = cid.0.to_vec();
return rebalance(prolly, parent, &ancestors[..ancestors.len() - 1]);
}
let left_cid = if left.len() > max_size && left.len() > 1 {
split_and_save_oversized(prolly, &left, ancestors)?
} else {
prolly.save(&left)?
};
let right_cid = if right.len() > max_size && right.len() > 1 {
split_and_save_oversized(prolly, &right, ancestors)?
} else {
prolly.save(&right)?
};
if ancestors.is_empty() {
let mut parent = prolly.new_internal_node(node.level + 1);
reserve_node_entries(&mut parent, 2);
parent.keys.push(left.keys[0].clone());
parent.vals.push(left_cid.0.to_vec());
parent.keys.push(right.keys[0].clone());
parent.vals.push(right_cid.0.to_vec());
if parent.len() > parent.max_chunk_size() {
return split_node(prolly, parent, &[]);
}
return prolly.save(&parent);
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
parent.keys[idx] = left.keys[0].clone();
parent.vals[idx] = left_cid.0.to_vec();
reserve_node_entries(&mut parent, 1);
parent.keys.insert(idx + 1, right.keys[0].clone());
parent.vals.insert(idx + 1, right_cid.0.to_vec());
rebalance(prolly, parent, &ancestors[..ancestors.len() - 1])
}
fn split_and_save_oversized<S: Store>(
prolly: &Prolly<S>,
node: &Node,
_ancestors: &[(Node, usize)],
) -> Result<Cid, Error> {
let max_size = node.max_chunk_size();
if node.len() <= max_size {
return prolly.save(node);
}
let capacity = max_size.max(1);
let mut chunks: Vec<Node> = Vec::with_capacity(node.len().div_ceil(capacity));
let mut start = 0;
while start < node.len() {
let chunk_size = capacity.min(node.len() - start);
let end = start + chunk_size;
let mut chunk = prolly.new_node_like(node);
chunk.keys = node.keys[start..end].to_vec();
chunk.vals = node.vals[start..end].to_vec();
chunks.push(chunk);
start = end;
}
if chunks.len() == 1 {
return prolly.save(&chunks[0]);
}
let mut parent = prolly.new_internal_node(node.level + 1);
reserve_node_entries(&mut parent, chunks.len());
for chunk in &chunks {
let chunk_cid = prolly.save(chunk)?;
parent.keys.push(chunk.keys[0].clone());
parent.vals.push(chunk_cid.0.to_vec());
}
if parent.len() > max_size && parent.len() > 1 {
return split_and_save_oversized(prolly, &parent, &[]);
}
prolly.save(&parent)
}
fn try_merge_with_sibling<S: Store>(
prolly: &Prolly<S>,
node: &Node,
ancestors: &[(Node, usize)],
) -> Result<Option<Cid>, Error> {
let (parent, idx) = ancestors.last().unwrap();
let idx = *idx;
if idx > 0 {
let left_cid = Cid(parent.vals[idx - 1]
.as_slice()
.try_into()
.map_err(|_| Error::InvalidNode)?);
let left_sibling = prolly.load(&left_cid)?;
if !is_valid_boundary_between(prolly, &left_sibling, node) {
let merged = merge_nodes(prolly, &left_sibling, node)?;
let mut new_parent = parent.clone();
new_parent.keys.remove(idx - 1);
new_parent.vals.remove(idx - 1);
let new_idx = idx - 1;
if merged.len() > merged.max_chunk_size() && merged.len() > 1 {
let mut new_ancestors: Vec<(Node, usize)> =
ancestors[..ancestors.len() - 1].to_vec();
new_ancestors.push((new_parent, new_idx));
return Ok(Some(split_node(prolly, merged, &new_ancestors)?));
}
let merged_cid = prolly.save(&merged)?;
new_parent.keys[new_idx] = merged.keys[0].clone();
new_parent.vals[new_idx] = merged_cid.0.to_vec();
return Ok(Some(rebalance(
prolly,
new_parent,
&ancestors[..ancestors.len() - 1],
)?));
}
}
if idx + 1 < parent.vals.len() {
let right_cid = Cid(parent.vals[idx + 1]
.as_slice()
.try_into()
.map_err(|_| Error::InvalidNode)?);
let right_sibling = prolly.load(&right_cid)?;
if !is_valid_boundary_between(prolly, node, &right_sibling) {
let merged = merge_nodes(prolly, node, &right_sibling)?;
let mut new_parent = parent.clone();
new_parent.keys.remove(idx + 1);
new_parent.vals.remove(idx + 1);
if merged.len() > merged.max_chunk_size() && merged.len() > 1 {
let mut new_ancestors: Vec<(Node, usize)> =
ancestors[..ancestors.len() - 1].to_vec();
new_ancestors.push((new_parent, idx));
return Ok(Some(split_node(prolly, merged, &new_ancestors)?));
}
let merged_cid = prolly.save(&merged)?;
new_parent.keys[idx] = merged.keys[0].clone();
new_parent.vals[idx] = merged_cid.0.to_vec();
return Ok(Some(rebalance(
prolly,
new_parent,
&ancestors[..ancestors.len() - 1],
)?));
}
}
Ok(None)
}
fn is_valid_boundary_between<S: Store>(_prolly: &Prolly<S>, left: &Node, _right: &Node) -> bool {
if left.is_empty() {
return false;
}
let last_idx = left.len() - 1;
is_boundary(left, last_idx)
}
fn merge_nodes<S: Store>(prolly: &Prolly<S>, left: &Node, right: &Node) -> Result<Node, Error> {
let mut merged = prolly.new_node_like(left);
let merged_len = left.len() + right.len();
merged.keys = Vec::with_capacity(merged_len);
merged.keys.extend(left.keys.iter().cloned());
merged.keys.extend(right.keys.iter().cloned());
merged.vals = Vec::with_capacity(merged_len);
merged.vals.extend(left.vals.iter().cloned());
merged.vals.extend(right.vals.iter().cloned());
Ok(merged)
}
pub fn rebalance_with_collector<S: Store>(
prolly: &Prolly<S>,
node: Node,
ancestors: &[(Node, usize)],
collector: &mut BatchWriteCollector,
) -> Result<Option<Cid>, Error> {
if node.is_empty() {
if ancestors.is_empty() {
return Ok(None); }
return handle_empty_node_with_collector(prolly, ancestors, collector);
}
if node.len() > node.max_chunk_size() && node.len() > 1 {
return split_node_with_collector(prolly, node, ancestors, collector);
}
if !ancestors.is_empty() && node.len() < node.min_chunk_size() {
if let Some(merged_cid) =
try_merge_with_sibling_collector(prolly, &node, ancestors, collector)?
{
return Ok(Some(merged_cid));
}
}
let cid = collector.add(&node);
if ancestors.is_empty() {
return Ok(Some(cid));
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
if !node.keys.is_empty() {
parent.keys[idx] = node.keys[0].clone();
}
parent.vals[idx] = cid.0.to_vec();
rebalance_with_collector(prolly, parent, &ancestors[..ancestors.len() - 1], collector)
}
fn handle_empty_node_with_collector<S: Store>(
prolly: &Prolly<S>,
ancestors: &[(Node, usize)],
collector: &mut BatchWriteCollector,
) -> Result<Option<Cid>, Error> {
if ancestors.is_empty() {
return Ok(None);
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
parent.keys.remove(idx);
parent.vals.remove(idx);
if parent.is_empty() && ancestors.len() == 1 {
return Ok(None);
}
rebalance_with_collector(prolly, parent, &ancestors[..ancestors.len() - 1], collector)
}
fn split_node_with_collector<S: Store>(
prolly: &Prolly<S>,
node: Node,
ancestors: &[(Node, usize)],
collector: &mut BatchWriteCollector,
) -> Result<Option<Cid>, Error> {
let max_size = node.max_chunk_size();
let chunks = split_into_chunks(prolly, &node, max_size);
if chunks.len() == 1 {
let cid = collector.add(&chunks[0]);
if ancestors.is_empty() {
return Ok(Some(cid));
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
if !chunks[0].keys.is_empty() {
parent.keys[idx] = chunks[0].keys[0].clone();
}
parent.vals[idx] = cid.0.to_vec();
return rebalance_with_collector(
prolly,
parent,
&ancestors[..ancestors.len() - 1],
collector,
);
}
let first_keys = chunks
.iter()
.map(|chunk| chunk.keys.first().cloned().unwrap_or_default())
.collect::<Vec<_>>();
let chunk_info: Vec<(Cid, Vec<u8>)> = collector
.add_many(chunks)
.into_iter()
.zip(first_keys)
.collect();
debug_assert!(
chunk_info.windows(2).all(|w| w[0].1 < w[1].1),
"split_node_with_collector: chunk first keys must be in strictly ascending order"
);
if ancestors.is_empty() {
let mut parent = prolly.new_internal_node(node.level + 1);
reserve_node_entries(&mut parent, chunk_info.len());
for (cid, first_key) in &chunk_info {
parent.keys.push(first_key.clone());
parent.vals.push(cid.0.to_vec());
}
if parent.len() > parent.max_chunk_size() {
return split_node_with_collector(prolly, parent, &[], collector);
}
let root_cid = collector.add(&parent);
return Ok(Some(root_cid));
}
let (mut parent, idx) = ancestors.last().unwrap().clone();
parent.keys.remove(idx);
parent.vals.remove(idx);
let additional_entries = chunk_info.len().saturating_sub(1);
reserve_node_entries(&mut parent, additional_entries);
for (i, (cid, first_key)) in chunk_info.iter().enumerate() {
parent.keys.insert(idx + i, first_key.clone());
parent.vals.insert(idx + i, cid.0.to_vec());
}
debug_assert!(
parent.keys.windows(2).all(|w| w[0] < w[1]),
"split_node_with_collector: parent keys must remain sorted after inserting chunks, \
indicating chunk keys don't overlap with siblings"
);
rebalance_with_collector(prolly, parent, &ancestors[..ancestors.len() - 1], collector)
}
pub fn split_into_chunks<S: Store>(prolly: &Prolly<S>, node: &Node, max_size: usize) -> Vec<Node> {
debug_assert!(
node.keys.windows(2).all(|w| w[0] < w[1]),
"split_into_chunks: input node keys must be in strictly ascending order"
);
let capacity = max_size.max(1);
if node.len() <= capacity {
return vec![node.clone()];
}
let num_chunks = node.len().div_ceil(capacity);
let mut chunks: Vec<Node> = Vec::with_capacity(num_chunks);
let mut start = 0;
while start < node.len() {
let remaining_chunks = num_chunks - chunks.len();
let remaining_entries = node.len() - start;
let target_size = remaining_entries
.checked_div(remaining_chunks)
.unwrap_or_else(|| remaining_entries.min(capacity))
.max(1);
let target_end = start + target_size;
let max_end = (start + capacity).min(node.len());
let min_end = start + 1;
let mut end = target_end.min(max_end).max(min_end);
let search_start = (target_end.saturating_sub(50)).max(min_end);
let search_end = (target_end + 50).min(max_end);
for i in (search_start..=search_end).rev() {
if i <= max_end && i < node.len() && is_boundary(node, i - 1) {
end = i;
break;
}
}
if end - start > capacity {
end = start + capacity;
}
let remaining_after = node.len() - end;
if remaining_after > 0
&& remaining_after < capacity / 4
&& (end - start) + remaining_after <= capacity
{
end = node.len();
}
if end - start > capacity {
end = start + capacity;
}
let mut chunk = prolly.new_node_like(node);
chunk.keys = node.keys[start..end].to_vec();
chunk.vals = node.vals[start..end].to_vec();
chunks.push(chunk);
start = end;
}
chunks
}
fn try_merge_with_sibling_collector<S: Store>(
prolly: &Prolly<S>,
node: &Node,
ancestors: &[(Node, usize)],
collector: &mut BatchWriteCollector,
) -> Result<Option<Cid>, Error> {
let (parent, idx) = ancestors.last().unwrap();
let idx = *idx;
if idx > 0 {
let left_cid = Cid(parent.vals[idx - 1]
.as_slice()
.try_into()
.map_err(|_| Error::InvalidNode)?);
let left_sibling = prolly.load(&left_cid)?;
if !is_valid_boundary_between(prolly, &left_sibling, node) {
let merged = merge_nodes(prolly, &left_sibling, node)?;
let mut new_parent = parent.clone();
new_parent.keys.remove(idx - 1);
new_parent.vals.remove(idx - 1);
let new_idx = idx - 1;
if merged.len() > merged.max_chunk_size() && merged.len() > 1 {
let mut new_ancestors: Vec<(Node, usize)> =
ancestors[..ancestors.len() - 1].to_vec();
new_ancestors.push((new_parent, new_idx));
return split_node_with_collector(prolly, merged, &new_ancestors, collector);
}
let merged_cid = collector.add(&merged);
new_parent.keys[new_idx] = merged.keys[0].clone();
new_parent.vals[new_idx] = merged_cid.0.to_vec();
return Ok(Some(
rebalance_with_collector(
prolly,
new_parent,
&ancestors[..ancestors.len() - 1],
collector,
)?
.unwrap_or(merged_cid),
));
}
}
if idx + 1 < parent.vals.len() {
let right_cid = Cid(parent.vals[idx + 1]
.as_slice()
.try_into()
.map_err(|_| Error::InvalidNode)?);
let right_sibling = prolly.load(&right_cid)?;
if !is_valid_boundary_between(prolly, node, &right_sibling) {
let merged = merge_nodes(prolly, node, &right_sibling)?;
let mut new_parent = parent.clone();
new_parent.keys.remove(idx + 1);
new_parent.vals.remove(idx + 1);
if merged.len() > merged.max_chunk_size() && merged.len() > 1 {
let mut new_ancestors: Vec<(Node, usize)> =
ancestors[..ancestors.len() - 1].to_vec();
new_ancestors.push((new_parent, idx));
return split_node_with_collector(prolly, merged, &new_ancestors, collector);
}
let merged_cid = collector.add(&merged);
new_parent.keys[idx] = merged.keys[0].clone();
new_parent.vals[idx] = merged_cid.0.to_vec();
return Ok(Some(
rebalance_with_collector(
prolly,
new_parent,
&ancestors[..ancestors.len() - 1],
collector,
)?
.unwrap_or(merged_cid),
));
}
}
Ok(None)
}
#[cfg(test)]
mod tests {
use super::super::config::Config;
use super::super::store::MemStore;
use super::*;
fn leaf_with_entries<S: Store>(prolly: &Prolly<S>, start: usize, end: usize) -> Node {
let mut node = prolly.new_leaf_node();
for idx in start..end {
node.keys.push(format!("k{idx:04}").into_bytes());
node.vals.push(format!("v{idx:04}").into_bytes());
}
node
}
#[test]
fn merge_nodes_preserves_order_and_node_settings() {
let config = Config::builder()
.min_chunk_size(2)
.max_chunk_size(8)
.chunking_factor(u32::MAX)
.build();
let prolly = Prolly::new(MemStore::new(), config);
let left = leaf_with_entries(&prolly, 0, 2);
let right = leaf_with_entries(&prolly, 2, 5);
let merged = merge_nodes(&prolly, &left, &right).unwrap();
assert!(merged.leaf);
assert_eq!(merged.level, left.level);
assert_eq!(merged.min_chunk_size(), left.min_chunk_size());
assert_eq!(merged.max_chunk_size(), left.max_chunk_size());
assert_eq!(
merged.keys,
(0..5)
.map(|idx| format!("k{idx:04}").into_bytes())
.collect::<Vec<_>>()
);
assert_eq!(
merged.vals,
(0..5)
.map(|idx| format!("v{idx:04}").into_bytes())
.collect::<Vec<_>>()
);
}
#[test]
fn split_into_chunks_preserves_all_entries_in_order() {
let config = Config::builder()
.min_chunk_size(2)
.max_chunk_size(4)
.chunking_factor(u32::MAX)
.build();
let prolly = Prolly::new(MemStore::new(), config);
let node = leaf_with_entries(&prolly, 0, 11);
let chunks = split_into_chunks(&prolly, &node, 4);
assert!(chunks.len() > 1);
assert!(chunks.iter().all(|chunk| !chunk.is_empty()));
assert!(chunks.iter().all(|chunk| chunk.len() <= 4));
assert_eq!(
chunks
.iter()
.flat_map(|chunk| chunk.keys.iter().cloned())
.collect::<Vec<_>>(),
node.keys
);
assert_eq!(
chunks
.iter()
.flat_map(|chunk| chunk.vals.iter().cloned())
.collect::<Vec<_>>(),
node.vals
);
}
}