mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-22 03:32:28 +00:00
feat: add remote branch diff and merge client APIs (#3686)
This PR adds some support for `diff` / `merge` in the remote client as for local tables we stay `NotSupported` until https://github.com/lance-format/lance/issues/7263. This wires the two review-and-land calls against the remote REST API: - `POST /v1/table/{id}/branches/diff` - `POST /v1/table/{id}/branches/merge` Rust gets typed results (`BranchDiff`, `MergeBranchResult`). Python returns the wire JSON, same shape as the REST response. Merge here means promoting a branch's added columns onto `main`. ### Behavior - Remote only. Local raises `NotSupported`. - A rejected merge is not an exception. HTTP 409 still returns `Ok` / a dict with `status="rejected"` and blockers in `diff.mergeBlockers`. - Unknown blocker / status codes parse as `Unknown` so a newer server does not break older clients. - `MergePreview` tolerates missing fields for the same reason. - Merge requests are not retried. 409 is final and carries the body you need. ### Example ```python table = db.open_table("images") table.branches.create("exp") exp = table.branches.checkout("exp") exp.add_columns({"tag": "cast('draft' as string)"}) diff = table.branches.diff("exp") preview = table.branches.merge("exp", dry_run=True) result = table.branches.merge("exp", dry_run=False) if result["status"] == "merged": print("landed at", result["mainVersionAfter"]) elif result["status"] == "rejected": print(result["diff"]["mergeBlockers"]) ``` ### Testing cargo test -p lancedb --features remote diff_branch cargo test -p lancedb --features remote merge_branch --------- Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -18,9 +18,11 @@ use crate::index::waiter::wait_for_index;
|
||||
use crate::query::{QueryFilter, QueryRequest, Select, VectorQueryRequest};
|
||||
use crate::table::AddColumnsResult;
|
||||
use crate::table::AddResult;
|
||||
use crate::table::BranchDiff;
|
||||
use crate::table::DeleteResult;
|
||||
use crate::table::DropColumnsResult;
|
||||
use crate::table::LsmWriteSpec;
|
||||
use crate::table::MergeBranchResult;
|
||||
use crate::table::MergeResult;
|
||||
use crate::table::Tags;
|
||||
use crate::table::UpdateResult;
|
||||
@@ -1817,6 +1819,79 @@ impl<S: HttpSend> BaseTable for RemoteTable<S> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn diff_branch(&self, from_branch: &str) -> Result<BranchDiff> {
|
||||
if from_branch.trim().is_empty() {
|
||||
return Err(Error::InvalidInput {
|
||||
message: "from_branch must be a non-empty string".into(),
|
||||
});
|
||||
}
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!("/v1/table/{}/branches/diff/", self.identifier))
|
||||
.json(&serde_json::json!({ "from_branch": from_branch }));
|
||||
let (request_id, response) = self.send(request, true).await?;
|
||||
if response.status() == StatusCode::NOT_FOUND {
|
||||
return Err(Error::TableNotFound {
|
||||
name: format!("{} (branch: {})", self.name, from_branch),
|
||||
source: format!("branch '{}' does not exist", from_branch).into(),
|
||||
});
|
||||
}
|
||||
let response = self.check_table_response(&request_id, response).await?;
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
serde_json::from_str(&body).map_err(|err| Error::Http {
|
||||
source: format!(
|
||||
"Failed to parse diff_branch response: {}, body: {}",
|
||||
err, body
|
||||
)
|
||||
.into(),
|
||||
request_id,
|
||||
status_code: None,
|
||||
})
|
||||
}
|
||||
|
||||
async fn merge_branch(&self, from_branch: &str, dry_run: bool) -> Result<MergeBranchResult> {
|
||||
if from_branch.trim().is_empty() {
|
||||
return Err(Error::InvalidInput {
|
||||
message: "from_branch must be a non-empty string".into(),
|
||||
});
|
||||
}
|
||||
let request = self
|
||||
.client
|
||||
.post(&format!("/v1/table/{}/branches/merge/", self.identifier))
|
||||
.json(&serde_json::json!({
|
||||
"from_branch": from_branch,
|
||||
"dry_run": dry_run,
|
||||
}));
|
||||
// No retry. 409 rejected merge is final and carries a body.
|
||||
let (request_id, response) = self.send(request, false).await?;
|
||||
let status = response.status();
|
||||
if status == StatusCode::NOT_FOUND {
|
||||
return Err(Error::TableNotFound {
|
||||
name: format!("{} (branch: {})", self.name, from_branch),
|
||||
source: format!("branch '{}' does not exist", from_branch).into(),
|
||||
});
|
||||
}
|
||||
// 200 and 409 both carry MergeBranchResult.
|
||||
if status != StatusCode::OK && status != StatusCode::CONFLICT {
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
return Err(Error::Http {
|
||||
source: format!("unexpected status {status} from merge_branch: {body}").into(),
|
||||
request_id,
|
||||
status_code: Some(status),
|
||||
});
|
||||
}
|
||||
let body = response.text().await.err_to_http(request_id.clone())?;
|
||||
serde_json::from_str(&body).map_err(|err| Error::Http {
|
||||
source: format!(
|
||||
"Failed to parse merge_branch response: {}, body: {}",
|
||||
err, body
|
||||
)
|
||||
.into(),
|
||||
request_id,
|
||||
status_code: Some(status),
|
||||
})
|
||||
}
|
||||
|
||||
fn current_branch(&self) -> Option<String> {
|
||||
self.branch.clone()
|
||||
}
|
||||
@@ -8211,6 +8286,147 @@ mod tests {
|
||||
assert!(matches!(err, Error::TableNotFound { .. }), "got {err:?}");
|
||||
}
|
||||
|
||||
fn sample_branch_diff_json() -> &'static str {
|
||||
r#"{
|
||||
"fromBranch":"exp",
|
||||
"parentVersion":1,
|
||||
"mainVersion":1,
|
||||
"branchVersion":2,
|
||||
"baseMoved":false,
|
||||
"rowCountMain":3,
|
||||
"rowCountBranch":3,
|
||||
"rowSummary":{
|
||||
"unchanged":3,
|
||||
"newOnBase":0,
|
||||
"newOnBranch":0,
|
||||
"staleRecompute":0,
|
||||
"inputsChanged":0,
|
||||
"deltaAvailable":false
|
||||
},
|
||||
"addedColumns":[{"name":"tag","dataType":"utf8","nullable":true}],
|
||||
"removedColumns":[],
|
||||
"changedColumns":[],
|
||||
"addedIndexes":[],
|
||||
"removedIndexes":[],
|
||||
"mergeable":true,
|
||||
"mergeBlockers":[]
|
||||
}"#
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_diff_branch() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
assert_eq!(request.method(), "POST");
|
||||
assert_eq!(request.url().path(), "/v1/table/my_table/branches/diff/");
|
||||
let body = request_body_json(&request);
|
||||
assert_eq!(body["from_branch"], "exp");
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(sample_branch_diff_json())
|
||||
.unwrap()
|
||||
});
|
||||
let diff = table.diff_branch("exp").await.unwrap();
|
||||
assert_eq!(diff.from_branch, "exp");
|
||||
assert!(diff.mergeable);
|
||||
assert_eq!(diff.added_columns.len(), 1);
|
||||
assert_eq!(diff.added_columns[0].name, "tag");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_merge_branch_dry_run() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
assert_eq!(request.url().path(), "/v1/table/my_table/branches/merge/");
|
||||
let body = request_body_json(&request);
|
||||
assert_eq!(body["from_branch"], "exp");
|
||||
assert_eq!(body["dry_run"], true);
|
||||
let resp = format!(
|
||||
r#"{{"status":"ready","diff":{},"preview":{{"promotedColumns":["tag"]}}}}"#,
|
||||
sample_branch_diff_json()
|
||||
);
|
||||
http::Response::builder().status(200).body(resp).unwrap()
|
||||
});
|
||||
let result = table.merge_branch("exp", true).await.unwrap();
|
||||
assert_eq!(result.status, crate::table::MergeBranchStatus::Ready);
|
||||
assert_eq!(result.preview.promoted_columns, vec!["tag".to_string()]);
|
||||
assert!(result.main_version_after.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_merge_branch_rejected_returns_ok_with_body() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
assert_eq!(request.url().path(), "/v1/table/my_table/branches/merge/");
|
||||
let body = request_body_json(&request);
|
||||
assert_eq!(body["dry_run"], false);
|
||||
let mut diff: serde_json::Value =
|
||||
serde_json::from_str(sample_branch_diff_json()).unwrap();
|
||||
diff["mergeable"] = serde_json::json!(false);
|
||||
diff["mergeBlockers"] = serde_json::json!([{
|
||||
"code": "baseMoved",
|
||||
"message": "main has advanced"
|
||||
}]);
|
||||
let resp = serde_json::json!({
|
||||
"status": "rejected",
|
||||
"diff": diff,
|
||||
"preview": { "promotedColumns": [] }
|
||||
});
|
||||
http::Response::builder()
|
||||
.status(409)
|
||||
.body(resp.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let result = table.merge_branch("exp", false).await.unwrap();
|
||||
assert_eq!(result.status, crate::table::MergeBranchStatus::Rejected);
|
||||
assert!(!result.diff.mergeable);
|
||||
assert_eq!(result.diff.merge_blockers.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_merge_branch_unknown_blocker_code_parses() {
|
||||
let table = Table::new_with_handler("my_table", |_| {
|
||||
let mut diff: serde_json::Value =
|
||||
serde_json::from_str(sample_branch_diff_json()).unwrap();
|
||||
diff["mergeable"] = serde_json::json!(false);
|
||||
diff["mergeBlockers"] = serde_json::json!([{
|
||||
"code": "multipleCommits",
|
||||
"message": "branch has more than one data commit"
|
||||
}]);
|
||||
let resp = serde_json::json!({
|
||||
"status": "rejected",
|
||||
"diff": diff,
|
||||
"preview": { "operation": "append", "rowsAdded": 2 }
|
||||
});
|
||||
http::Response::builder()
|
||||
.status(409)
|
||||
.body(resp.to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let result = table.merge_branch("exp", false).await.unwrap();
|
||||
assert_eq!(result.status, crate::table::MergeBranchStatus::Rejected);
|
||||
assert_eq!(
|
||||
result.diff.merge_blockers[0].code,
|
||||
crate::table::MergeBlockerCode::Unknown
|
||||
);
|
||||
assert!(result.preview.promoted_columns.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_merge_branch_unexpected_2xx_is_error() {
|
||||
let table = Table::new_with_handler("my_table", |_| {
|
||||
http::Response::builder()
|
||||
.status(204)
|
||||
.body(String::new())
|
||||
.unwrap()
|
||||
});
|
||||
let err = table.merge_branch("exp", false).await.unwrap_err();
|
||||
match err {
|
||||
Error::Http {
|
||||
status_code: Some(code),
|
||||
..
|
||||
} => assert_eq!(code, reqwest::StatusCode::NO_CONTENT),
|
||||
other => panic!("expected Http error, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_checkout_branch_validates_via_list() {
|
||||
let table = Table::new_with_handler("my_table", |request| {
|
||||
|
||||
@@ -62,6 +62,7 @@ use self::dataset::DatasetConsistencyWrapper;
|
||||
use self::merge::MergeInsertBuilder;
|
||||
|
||||
mod add_data;
|
||||
pub mod branch_merge;
|
||||
mod create_index;
|
||||
pub mod datafusion;
|
||||
pub(crate) mod dataset;
|
||||
@@ -77,6 +78,10 @@ use crate::index::waiter::wait_for_index;
|
||||
#[cfg(feature = "remote")]
|
||||
pub(crate) use add_data::PreprocessingOutput;
|
||||
pub use add_data::{AddDataBuilder, AddDataMode, AddResult, NaNVectorBehavior};
|
||||
pub use branch_merge::{
|
||||
BranchDiff, ColumnChange, ColumnSummary, IndexSummary, MergeBlocker, MergeBlockerCode,
|
||||
MergeBranchResult, MergeBranchStatus, MergePreview, RowCountSummary,
|
||||
};
|
||||
pub use chrono::Duration;
|
||||
pub use delete::DeleteResult;
|
||||
use futures::future::join_all;
|
||||
@@ -707,6 +712,19 @@ pub trait BaseTable: std::fmt::Display + std::fmt::Debug + Send + Sync {
|
||||
async fn list_branches(&self) -> Result<HashMap<String, BranchContents>>;
|
||||
/// Delete a branch.
|
||||
async fn delete_branch(&self, name: &str) -> Result<()>;
|
||||
/// Diff a branch against main. Remote only.
|
||||
async fn diff_branch(&self, _from_branch: &str) -> Result<BranchDiff> {
|
||||
Err(Error::NotSupported {
|
||||
message: "diff_branch is only supported on remote tables".into(),
|
||||
})
|
||||
}
|
||||
/// Merge a branch into main, or dry-run. Remote only.
|
||||
/// HTTP 409 still returns [`Ok`] with [`MergeBranchStatus::Rejected`].
|
||||
async fn merge_branch(&self, _from_branch: &str, _dry_run: bool) -> Result<MergeBranchResult> {
|
||||
Err(Error::NotSupported {
|
||||
message: "merge_branch is only supported on remote tables".into(),
|
||||
})
|
||||
}
|
||||
/// The branch this handle is scoped to, or `None` for `main`.
|
||||
fn current_branch(&self) -> Option<String>;
|
||||
/// Get the table definition.
|
||||
@@ -1953,6 +1971,21 @@ impl Table {
|
||||
self.inner.delete_branch(name).await
|
||||
}
|
||||
|
||||
/// Diff a branch against main. Remote only.
|
||||
pub async fn diff_branch(&self, from_branch: &str) -> Result<BranchDiff> {
|
||||
self.inner.diff_branch(from_branch).await
|
||||
}
|
||||
|
||||
/// Merge a branch into main, or dry-run. Remote only.
|
||||
/// HTTP 409 still returns [`Ok`] with [`MergeBranchStatus::Rejected`].
|
||||
pub async fn merge_branch(
|
||||
&self,
|
||||
from_branch: &str,
|
||||
dry_run: bool,
|
||||
) -> Result<MergeBranchResult> {
|
||||
self.inner.merge_branch(from_branch, dry_run).await
|
||||
}
|
||||
|
||||
/// The branch this handle is scoped to, or `None` for `main`.
|
||||
pub fn current_branch(&self) -> Option<String> {
|
||||
self.inner.current_branch()
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
//! Types for remote branch diff / merge against main.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ColumnSummary {
|
||||
pub name: String,
|
||||
pub data_type: String,
|
||||
pub nullable: bool,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ColumnChange {
|
||||
pub name: String,
|
||||
pub main: ColumnSummary,
|
||||
pub branch: ColumnSummary,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct IndexSummary {
|
||||
pub index_name: String,
|
||||
pub columns: Vec<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub index_type: Option<String>,
|
||||
pub status: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct RowCountSummary {
|
||||
pub unchanged: u64,
|
||||
pub new_on_base: u64,
|
||||
pub new_on_branch: u64,
|
||||
pub stale_recompute: u64,
|
||||
pub inputs_changed: u64,
|
||||
pub delta_available: bool,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub enum MergeBlockerCode {
|
||||
BaseMoved,
|
||||
RowCountMismatch,
|
||||
RowsChanged,
|
||||
ColumnRemoved,
|
||||
ColumnChanged,
|
||||
NoMergeableChanges,
|
||||
NoColumnChanges,
|
||||
InputColumnDependency,
|
||||
ParentNotMain,
|
||||
#[serde(other)]
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MergeBlocker {
|
||||
pub code: MergeBlockerCode,
|
||||
pub message: String,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct BranchDiff {
|
||||
pub from_branch: String,
|
||||
pub parent_version: u64,
|
||||
pub main_version: u64,
|
||||
pub branch_version: u64,
|
||||
pub base_moved: bool,
|
||||
pub row_count_main: u64,
|
||||
pub row_count_branch: u64,
|
||||
pub row_summary: RowCountSummary,
|
||||
pub added_columns: Vec<ColumnSummary>,
|
||||
pub removed_columns: Vec<ColumnSummary>,
|
||||
pub changed_columns: Vec<ColumnChange>,
|
||||
pub added_indexes: Vec<IndexSummary>,
|
||||
pub removed_indexes: Vec<IndexSummary>,
|
||||
pub mergeable: bool,
|
||||
pub merge_blockers: Vec<MergeBlocker>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MergePreview {
|
||||
#[serde(default)]
|
||||
pub promoted_columns: Vec<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub enum MergeBranchStatus {
|
||||
Ready,
|
||||
Rejected,
|
||||
NotImplemented,
|
||||
Merged,
|
||||
#[serde(other)]
|
||||
Unknown,
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct MergeBranchResult {
|
||||
pub status: MergeBranchStatus,
|
||||
pub diff: BranchDiff,
|
||||
pub preview: MergePreview,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub main_version_after: Option<u64>,
|
||||
}
|
||||
Reference in New Issue
Block a user