Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
43 changes: 15 additions & 28 deletions crates/liboxen/src/core/v_latest/workspaces/data_frames.rs
Original file line number Diff line number Diff line change
Expand Up @@ -222,13 +222,22 @@ pub async fn index(workspace: &Workspace, path: &Path) -> Result<(), OxenError>
let path = path.to_path_buf();
spawn_blocking(move || -> Result<_, OxenError> {
let path = path.as_path();
// Is tabular just looks at the file extensions
let file_node = repositories::tree::get_file_by_path(
let file_node = match repositories::tree::get_file_by_path(
&workspace.base_repo,
&workspace.commit,
path,
)?
.ok_or_else(|| OxenError::path_does_not_exist(path))?;
)? {
Some(file_node) => file_node,
None => {
let staged = get_staged_db_manager(&workspace.workspace_repo)?
.read_from_staged_db(path)?
.ok_or_else(|| OxenError::path_does_not_exist(path))?;
let EMerkleTreeNode::File(file_node) = staged.node.node else {
return Err(OxenError::NotAFile(path.to_path_buf().into()));
};
file_node
}
};
if *file_node.data_type() != EntryDataType::Tabular {
return Err(OxenError::basic_str(
"File format not supported, must be tabular.",
Expand All @@ -237,22 +246,7 @@ pub async fn index(workspace: &Workspace, path: &Path) -> Result<(), OxenError>

log::debug!("core::v_latest::workspaces::data_frames::index({path:?})");

let repo = &workspace.base_repo;
let commit = &workspace.commit;

log::debug!(
"core::v_latest::workspaces::data_frames::index({path:?}) got commit {commit:?}"
);

let Ok(Some(commit_merkle_tree)) =
repositories::tree::get_node_by_path_with_children(repo, commit, path)
else {
return Err(OxenError::basic_str(format!(
"Merkle tree for commit {commit} not found"
)));
};

let file_hash = commit_merkle_tree.hash;
let file_hash = *file_node.hash();

log::debug!(
"core::v_latest::workspaces::data_frames::index({path:?}) got file hash {file_hash:?}"
Expand All @@ -267,14 +261,7 @@ pub async fn index(workspace: &Workspace, path: &Path) -> Result<(), OxenError>
};
util::fs::create_dir_all(&parent)?;

let extension = match &commit_merkle_tree.node {
EMerkleTreeNode::File(file_node) => file_node.extension().to_string(),
_ => {
return Err(OxenError::basic_str("File node is not a file node"));
}
};

Ok((file_hash, extension, db_path, parent))
Ok((file_hash, file_node.extension().to_string(), db_path, parent))
})
.await??
};
Expand Down
51 changes: 26 additions & 25 deletions crates/liboxen/src/core/v_latest/workspaces/files.rs
Original file line number Diff line number Diff line change
Expand Up @@ -381,8 +381,7 @@ pub fn track_modified_data_frame(
let base_repo = &workspace.base_repo;

// Stage the file using the repositories::add method
let commit = workspace.commit.clone();
p_modify_file(base_repo, workspace_repo, &Some(commit), filepath)?;
p_modify_file(base_repo, workspace_repo, &workspace.commit, filepath)?;

// Return the relative path of the file in the workspace
let relative_path = util::fs::path_relative_to_dir(filepath, &workspace_repo.path)?;
Expand Down Expand Up @@ -1282,38 +1281,40 @@ fn unindex_if_staged_table(workspace: &Workspace, path: &Path) -> Result<(), Oxe
fn p_modify_file(
base_repo: &LocalRepository,
workspace_repo: &LocalRepository,
maybe_head_commit: &Option<Commit>,
commit: &Commit,
path: &Path,
) -> Result<(), OxenError> {
let mut maybe_file_node = None;
if let Some(head_commit) = maybe_head_commit {
maybe_file_node = repositories::tree::get_file_by_path(base_repo, head_commit, path)?;
}
let Some(mut file_node) = maybe_file_node else {
return Err(OxenError::basic_str("file not found in head commit"));
};
file_node.set_name(path.to_str().unwrap());
let committed = repositories::tree::get_file_by_path(base_repo, commit, path)?;
// Stamping each edit's time changes the staged node, so a commit that read it before this edit
// leaves the path staged for the next commit.
let edited_at = FileTime::now();
file_node.set_last_modified(edited_at.unix_seconds(), edited_at.nanoseconds());
log::debug!("p_modify_file file_node: {file_node}");

let staged_db_manager = get_staged_db_manager(workspace_repo)?;
staged_db_manager.edit_staged_node(path, |staged| {
// The committed node knows nothing of workspace metadata edits. When
// the staged node holds the same content with different metadata,
// carry that metadata over so re-staging doesn't discard it.
if let Some(staged) = staged
&& let Ok(staged_file) = staged.node.file()
&& staged_file.hash() == file_node.hash()
&& let Some(metadata) = staged_file.metadata()
{
file_node.set_metadata(Some(metadata));
file_node.recompute_metadata_hashes()?;
}
let (mut file_node, status) = match (committed, staged) {
(Some(mut file_node), staged) => {
file_node.set_name(path.to_str().unwrap());
// The committed node knows nothing of workspace metadata edits. When
// the staged node holds the same content with different metadata,
// carry that metadata over so re-staging doesn't discard it.
if let Some(staged) = staged
&& let Ok(staged_file) = staged.node.file()
&& staged_file.hash() == file_node.hash()
&& let Some(metadata) = staged_file.metadata()
{
file_node.set_metadata(Some(metadata));
file_node.recompute_metadata_hashes()?;
}
(file_node, StagedEntryStatus::Modified)
}
// A file only the workspace has, such as a copy, stays staged as it was added.
(None, Some(staged)) => (staged.node.file()?, staged.status),
(None, None) => return Err(OxenError::basic_str("file not found in head commit")),
};
file_node.set_last_modified(edited_at.unix_seconds(), edited_at.nanoseconds());
log::debug!("p_modify_file file_node: {file_node}");
Ok(StagedMerkleTreeNode {
status: StagedEntryStatus::Modified,
status,
node: MerkleTreeNode::from_file(file_node),
})
})?;
Expand Down
21 changes: 21 additions & 0 deletions crates/liboxen/src/repositories/workspaces/files.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,13 +47,15 @@ mod tests {
use crate::config::UserConfig;
use crate::core::staged::staged_db_manager::get_staged_db_manager;
use crate::error::OxenError;
use crate::model::metadata::generic_metadata::GenericMetadata;
use crate::model::{
Commit, EntryDataType, LocalRepository, NewCommitBody, StagedEntryStatus, Workspace,
};
use crate::repositories::size;
use crate::repositories::{self, workspaces};
use crate::test;
use crate::util::fs;
use serde_json::json;

#[tokio::test]
async fn test_mv_file_in_workspace() -> Result<(), OxenError> {
Expand Down Expand Up @@ -351,6 +353,16 @@ mod tests {
None,
"the source stays as committed"
);
// A copy indexes from its staged entry, and a row added to it commits to the copy.
let edited = Path::new("edited.csv");
workspaces::files::copy(&workspace, Path::new("table.csv"), edited).await?;
workspaces::data_frames::index(&repo, &workspace, edited).await?;
workspaces::data_frames::rows::add(
&repo,
&workspace,
edited,
&json!({"x": 3, "y": 4}),
)?;
stage_new_file(&workspace, "data/a.txt").await?;
workspaces::files::rm(&workspace, &[PathBuf::from("data/sub/b.txt")]).await?;
for (source, dest, refusal) in [
Expand Down Expand Up @@ -395,6 +407,15 @@ mod tests {
"{copy} is typed by its own name"
);
}
let rows = |path: &str| match node(path).metadata() {
Some(GenericMetadata::MetadataTabular(m)) => m.tabular.height,
other => panic!("{path} should have tabular metadata, got {other:?}"),
};
assert_eq!(
(rows("table.csv"), rows("edited.csv")),
(1, 2),
"the row added to the copy commits to the copy alone"
);
Comment thread
CleanCut marked this conversation as resolved.
Resolved

// A committed directory moves with its nested subdirectory, and retrying the move
// finds its source staged for removal.
Expand Down
43 changes: 37 additions & 6 deletions crates/oxen-server/src/controllers/workspaces/data_frames.rs
Original file line number Diff line number Diff line change
Expand Up @@ -157,9 +157,17 @@ pub async fn get(
.map_err(OxenError::from)??
};

let Some(df_schema) = get_by_path_async(&repo, &workspace.commit, &file_path).await? else {
log::warn!("Failed to get schema for data frame {file_path:?}");
return Err(OxenHttpError::NotFound);
let new_schema =
get_staged_schema_with_staged_db_manager_async(&workspace.workspace_repo, &file_path)
.await?;
let df_schema = match get_by_path_async(&repo, &workspace.commit, &file_path).await {
Ok(Some(schema)) => schema,
// A data frame only the workspace has, such as a copy, has no committed schema.
Ok(None) | Err(OxenError::PathDoesNotExist(_)) => new_schema.clone().ok_or_else(|| {
log::warn!("Failed to get schema for data frame {file_path:?}");
OxenHttpError::NotFound
})?,
Err(err) => return Err(err.into()),
};

let resource = ResourceVersion {
Expand All @@ -170,9 +178,6 @@ pub async fn get(
let mut df_views =
JsonDataFrameViews::from_df_and_opts_unpaginated(df, df_schema, count, &opts).await?;

let new_schema =
get_staged_schema_with_staged_db_manager_async(&workspace.workspace_repo, &file_path)
.await?;
repositories::workspaces::data_frames::columns::update_column_schemas(
new_schema,
&mut df_views,
Expand Down Expand Up @@ -986,6 +991,32 @@ mod tests {
assert_eq!(pagination.total_pages, 2);
}

// A copy that exists only in the workspace indexes and pages like the frame it copies.
let copy_path = "data/history_copy.csv";
repositories::workspaces::files::copy(
&workspace,
std::path::Path::new(file_path),
std::path::Path::new(copy_path),
)
.await?;
repositories::workspaces::data_frames::index(
&repo,
&workspace,
std::path::Path::new(copy_path),
)
.await?;
let copy_page = get_data_frame_page(
&sync_dir,
namespace,
repo_name,
&workspace_id,
copy_path,
page_query(3, 10, None),
)
.await;
assert!(copy_page.is_indexed);
assert_eq!(page_ids(&copy_page), (20..25).collect::<Vec<i64>>());

drop((app, workspace));
test::cleanup_repo_and_sync_dir(repo, &sync_dir)?;
Ok(())
Expand Down
Loading