mirror of
https://github.com/lancedb/lancedb.git
synced 2026-06-04 21:00:44 +00:00
test: assert branch handle read-consistency behavior
This commit is contained in:
@@ -904,7 +904,7 @@ async def test_async_tags(mem_db_async: AsyncConnection):
|
||||
|
||||
|
||||
def test_branches(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(0))
|
||||
table = db.create_table(
|
||||
"test",
|
||||
data=[
|
||||
@@ -942,6 +942,22 @@ def test_branches(tmp_path):
|
||||
assert "exp" not in table.branches.list()
|
||||
|
||||
|
||||
def test_branch_handle_tracks_concurrent_writes(tmp_path):
|
||||
db = lancedb.connect(tmp_path, read_consistency_interval=timedelta(0))
|
||||
table = db.create_table("t", [{"id": 1}])
|
||||
|
||||
# two independent handles on the same branch
|
||||
writer = table.branches.create("exp")
|
||||
reader = db.open_table("t", branch="exp")
|
||||
assert reader.count_rows() == 1
|
||||
|
||||
# a concurrent write on the branch is visible to the other handle
|
||||
writer.add([{"id": 2}])
|
||||
assert reader.count_rows() == 2
|
||||
# main is unaffected
|
||||
assert table.count_rows() == 1
|
||||
|
||||
|
||||
def test_branch_name_validation(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
table = db.create_table("t", [{"id": 1}])
|
||||
|
||||
@@ -3594,6 +3594,75 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_branch_handle_tracks_concurrent_writes() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = tmp_dir.path().to_str().unwrap();
|
||||
|
||||
// interval = 0 so every read checks storage for new commits
|
||||
let conn = ConnectBuilder::new(uri)
|
||||
.read_consistency_interval(Duration::from_secs(0))
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let table = conn
|
||||
.create_table("my_table", some_sample_data())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let v1 = table.version().await.unwrap();
|
||||
|
||||
// two independent handles on the same branch
|
||||
let writer = table.create_branch("exp", v1).await.unwrap();
|
||||
let reader = conn
|
||||
.open_table("my_table")
|
||||
.branch("exp")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(reader.count_rows(None).await.unwrap(), 1);
|
||||
|
||||
// a concurrent write on the branch is visible to the other handle, which
|
||||
// tracks the branch's HEAD (not main's)
|
||||
writer.add(some_sample_data()).execute().await.unwrap();
|
||||
assert_eq!(reader.count_rows(None).await.unwrap(), 2);
|
||||
// main is untouched
|
||||
assert_eq!(table.count_rows(None).await.unwrap(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_branch_handle_without_consistency_interval_is_pinned() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let uri = tmp_dir.path().to_str().unwrap();
|
||||
|
||||
// default interval (None): handles do not auto-refresh
|
||||
let conn = ConnectBuilder::new(uri).execute().await.unwrap();
|
||||
let table = conn
|
||||
.create_table("my_table", some_sample_data())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
let v1 = table.version().await.unwrap();
|
||||
|
||||
let writer = table.create_branch("exp", v1).await.unwrap();
|
||||
let reader = conn
|
||||
.open_table("my_table")
|
||||
.branch("exp")
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(reader.count_rows(None).await.unwrap(), 1);
|
||||
|
||||
// without a consistency interval the reader stays on the version it
|
||||
// opened, exactly like a main-branch handle...
|
||||
writer.add(some_sample_data()).execute().await.unwrap();
|
||||
assert_eq!(reader.count_rows(None).await.unwrap(), 1);
|
||||
|
||||
// ...until it explicitly refreshes
|
||||
reader.checkout_latest().await.unwrap();
|
||||
assert_eq!(reader.count_rows(None).await.unwrap(), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_index() {
|
||||
use arrow_array::RecordBatch;
|
||||
|
||||
Reference in New Issue
Block a user