Skip to content

Refactor oxen's Merkle tree operations: use MerkleStore #513

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
malcolmgreaves wants to merge 1 commit into
base: mg/merkle_dyn_file_backend
Choose a base branch
from
Open

Refactor oxen's Merkle tree operations: use MerkleStore #513

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
119 changes: 65 additions & 54 deletions crates/lib/src/command/migrate/m20250111083535_add_child_counts_to_nodes.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@ use std::path::Path;
use super::Migrate;

use crate::config::RepositoryConfig;
use crate::core::db::merkle_node::MerkleNodeDB;
use crate::core::versions::MinOxenVersion;
use crate::error::OxenError;
use crate::model::merkle_tree::merkle_tree_node_cache;
use crate::model::merkle_tree::merkle_writer::MerkleWriteSession;
use crate::model::merkle_tree::node::vnode::VNodeOpts;
use crate::model::merkle_tree::node::{
CommitNode, DirNode, EMerkleTreeNode, MerkleTreeNode, VNode,
Expand Down Expand Up @@ -135,14 +135,27 @@ fn run_on_commit(repository: &LocalRepository, commit: &Commit) -> Result<(), Ox
dir_node_opts.num_entries = num_children as u64;
let dir_node = DirNode::new(&new_repo, dir_node_opts)?;

// Write a new commit db
// Write a new commit node via a session on the old repo.
let commit_node = CommitNode::from_commit(commit.clone());
let mut root_commit_db =
MerkleNodeDB::open_read_write(&old_repo.path, &commit_node, root_node.parent_id)?;
root_commit_db.add_child(&dir_node)?;
let old_store = old_repo.merkle_store();
let new_store = new_repo.merkle_store();
let old_session = old_store.begin()?;
let new_session = new_store.begin()?;
let mut root_commit_ns = old_session.create_node(&commit_node, root_node.parent_id)?;
root_commit_ns.add_child(&dir_node)?;
root_commit_ns.finish()?;

let current_path = Path::new("");
rewrite_nodes(&old_repo, &new_repo, &root_node, current_path)?;
rewrite_nodes(
&old_repo,
&new_repo,
&*old_session,
&*new_session,
&root_node,
current_path,
)?;
new_session.finish()?;
old_session.finish()?;

// println!("new tree for commit {}", commit);
// repositories::tree::print_tree(&new_repo, commit)?;
Expand All @@ -158,9 +171,12 @@ fn run_on_commit(repository: &LocalRepository, commit: &Commit) -> Result<(), Ox

// Forgive me if you are reading this for reference, we don't have great writers for the
// merkle tree yet - so there is a lot of duplicate logic with `commit_writer.rs`
#[allow(clippy::too_many_arguments)]
fn rewrite_nodes(
old_repo: &LocalRepository,
new_repo: &LocalRepository,
old_session: &dyn MerkleWriteSession,
new_session: &dyn MerkleWriteSession,
node: &MerkleTreeNode,
current_dir: &Path,
) -> Result<(), OxenError> {
Expand All @@ -169,19 +185,10 @@ fn rewrite_nodes(
EMerkleTreeNode::Directory(dir) => {
// Load all the children of children (files and folders)
// Then redistribute into buckets...
// and then just use the MerkleNodeDB to write the nodes
// to the new tree
// and then just use the session to write the nodes to the new tree
let dir_children = repositories::tree::list_files_and_folders(child)?;
let current_dir = current_dir.join(dir.name());

// log::debug!(
// "rewrite_nodes {} children on current_dir {:?} DIRECTORY {} {}",
// dir_children.len(),
// current_dir,
// dir.hash(),
// dir
// );

let total_children = dir_children.len();
let vnode_size = old_repo.vnode_size();
let num_vnodes = (total_children as f32 / vnode_size as f32).ceil() as u128;
Expand All @@ -190,31 +197,30 @@ fn rewrite_nodes(
let mut dir_node_opts = dir.get_opts();
dir_node_opts.num_entries = total_children as u64;
let dir = DirNode::new(new_repo, dir_node_opts)?;
let mut dir_db =
MerkleNodeDB::open_read_write(&old_repo.path, &dir, node.parent_id)?;
let mut dir_ns = old_session.create_node(&dir, node.parent_id)?;

// log::debug!(
// "rewrite_nodes {} VNodes for {} children in {} with vnode size {}",
// num_vnodes,
// total_children,
// dir,
// vnode_size
// );
log::trace!(
"rewrite_nodes {} VNodes for {} children in {} with vnode size {}",
num_vnodes,
total_children,
dir,
vnode_size
);

// Compute buckets
let mut buckets: Vec<Vec<MerkleTreeNode>> = vec![vec![]; num_vnodes as usize];
for dir_child in dir_children {
let path = current_dir.join(dir_child.maybe_path().unwrap());
let hash = hasher::hash_buffer_128bit(path.to_str().unwrap().as_bytes());
let bucket_idx = hash % num_vnodes;
// log::debug!(
// "\trewrite_nodes dir_child {:?} bucket {} num_vnodes {} hash {} {}",
// path,
// bucket_idx,
// num_vnodes,
// hash,
// dir_child
// );
log::trace!(
"\trewrite_nodes dir_child {:?} bucket {} num_vnodes {} hash {} {}",
path,
bucket_idx,
num_vnodes,
hash,
dir_child
);
buckets[bucket_idx as usize].push(dir_child);
}

Expand Down Expand Up @@ -243,54 +249,59 @@ fn rewrite_nodes(
vnodes.push((vnode_id, bucket.clone()));
}

// log::debug!("rewrite_nodes count vnodes: {}", vnodes.len());
log::trace!("rewrite_nodes count vnodes: {}", vnodes.len());
for (hash, entries) in vnodes.iter() {
// create a new vnode obj and add the the db
// create a new vnode obj and add to the dir session
let opts = VNodeOpts {
hash: *hash,
num_entries: entries.len() as u64,
};
let vnode_obj = VNode::new(new_repo, opts)?;
// log::debug!("rewrite_nodes adding VNode to DirNode! {:?}", vnode_obj);
dir_db.add_child(&vnode_obj)?;
dir_ns.add_child(&vnode_obj)?;

let mut vnode_db = MerkleNodeDB::open_read_write(
&new_repo.path,
&vnode_obj,
Some(dir_db.node_id),
)?;
let dir_ns_id = *dir_ns.node_id();
let mut vnode_ns = new_session.create_node(&vnode_obj, Some(dir_ns_id))?;

// log::debug!("rewrite_nodes count entries {}", entries.len());
for entry in entries {
match &entry.node {
EMerkleTreeNode::File(f_node) => {
// log::debug!("rewrite_nodes adding FileNode to VNode! {}", f_node);
vnode_db.add_child(f_node)?;
vnode_ns.add_child(f_node)?;
}
EMerkleTreeNode::Directory(d_node) => {
let mut d_node_opts = d_node.get_opts();
let d_children = repositories::tree::list_files_and_folders(entry)?;
d_node_opts.num_entries = d_children.len() as u64;
// log::debug!(
// "rewrite_nodes adding DirNode to VNode with {} num_entries {}",
// d_node_opts.num_entries,
// d_node
// );
let d_node = DirNode::new(new_repo, d_node_opts)?;
vnode_db.add_child(&d_node)?;
vnode_ns.add_child(&d_node)?;
}
_ => {
panic!("Shouldn't reach here.")
}
}
}
vnode_ns.finish()?;
}

rewrite_nodes(old_repo, new_repo, child, &current_dir)?;
dir_ns.finish()?;

rewrite_nodes(
old_repo,
new_repo,
old_session,
new_session,
child,
&current_dir,
)?;
}
EMerkleTreeNode::VNode(_) => {
// VNode just needs to traverse to the next dirnode
rewrite_nodes(old_repo, new_repo, child, current_dir)?;
rewrite_nodes(
old_repo,
new_repo,
old_session,
new_session,
child,
current_dir,
)?;
}
_ => {
// pass, FileNode was not changed, so it is on the latest version
Expand Down
2 changes: 1 addition & 1 deletion crates/lib/src/core/db/merkle_node/file_backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ pub struct FileWriteSession<'repo> {
/// not return a wrapped `MerkleDbError`.
impl<'repo> MerkleWriteSession for FileWriteSession<'repo> {
/// Creates a new session for writing a `node` and `children` file.
/// Calls [`MerkleNodeDB::open_read_write`] internally.
/// Returns a [`FileNodeSession`] that calls [`MerkleNodeDB::open_read_write`] internally.
fn create_node<'a>(
&'a self,
node: &dyn TMerkleTreeNode,
Expand Down
19 changes: 13 additions & 6 deletions crates/lib/src/core/v_latest/commits.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@ use crate::config::UserConfig;
use crate::constants::COMMIT_COUNT_DIR;
use crate::core;
use crate::core::db::key_val::{opts, str_val_db};
use crate::core::db::merkle_node::MerkleNodeDB;
use crate::core::refs::with_ref_manager;
use crate::core::v_latest::index::CommitMerkleTree;
use crate::error::OxenError;
Expand Down Expand Up @@ -291,10 +290,14 @@ pub fn create_empty_commit(
)?;

let parent_id = Some(existing_node.hash);
let mut commit_db = MerkleNodeDB::open_read_write(&repo.path, &commit_node, parent_id)?;
let store = repo.merkle_store();
let session = store.begin()?;
let mut commit_ns = session.create_node(&commit_node, parent_id)?;
// There should always be one child, the root directory
let dir_node = existing_node.children.first().unwrap().dir()?;
commit_db.add_child(&dir_node)?;
commit_ns.add_child(&dir_node)?;
commit_ns.finish()?;
session.finish()?;

// Copy the dir hashes db to the new commit
repositories::tree::cp_dir_hashes_to(repo, &existing_commit_id, commit_node.hash())?;
Expand Down Expand Up @@ -370,9 +373,13 @@ pub fn create_initial_commit(
},
)?;

// Open the commit database and add the root directory
let mut commit_db = MerkleNodeDB::open_read_write(&repo.path, &commit_node, None)?;
commit_db.add_child(&dir_node)?;
// Open the commit write session and add the root directory
let store = repo.merkle_store();
let session = store.begin()?;
let mut commit_ns = session.create_node(&commit_node, None)?;
commit_ns.add_child(&dir_node)?;
commit_ns.finish()?;
session.finish()?;

// Initialize the dir_hash_db with the root directory hash
let commit_id_string = commit_id.to_string();
Expand Down
Loading
Loading