mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-13 15:22:30 +00:00
Jack YeandBrendan Clement
47b7babc08
Move the public website pages into `docs/web` beside the SDK, with
shared navigation and executable example sources in `docs/web-tests`.
The combined website is assembled by
[lancedb/docs#355](https://github.com/lancedb/docs/pull/355), with six
Enterprise replacements from
[lancedb/sophon#7681](https://github.com/lancedb/sophon/pull/7681).
This includes Function guidance, runnable Quickstarts, regenerated
snippets, production contact/demo fixes, the legacy SQL route, and the
LeRobot redirect. LeRobot remains attributed to Geneva; all 34 Geneva
pages remain in the docs repository until Function officially launches.
Materialized-view guidance follows the full rebuild refresh introduced
by #4373: local refresh writes the complete result, has no `full`
argument, supports `source_version`, and no longer requires stable row
IDs. Deployment SQL refresh options and stable-ID setup are qualified by
version/configuration. Function input guidance distinguishes ordinary
stored columns from computed inputs supported by newer deployments.
**Validation:** independently accepted local checks include 95 Python
tests (real Enterprise excluded), eight TypeScript tests, Rust
Quickstart, snippet parity, typos, deterministic assembly, Mint
validation and links/anchors/redirects. The combined baseline retains
all 181 production routes and 17 legacy redirects. Examples ran locally
against source-built Python SDK 0.40.0b13; existing GitHub workflows do
not run `docs/web-tests`. New main at `069b2f7d` is 0.41.0b0 and adds
computed Function inputs; that SDK was not exercised by the accepted
tests. Latest prose corrections receive focused site validation.
**Merge/rollout:** merge this and #7681 before #355's combined CI, which
reads producer `main`. This is migration infrastructure: section
overlays, source-owned skills and automatic versioned paired releases
remain follow-ups. The current manual publisher is transitional. Hosted
validation, a supported SDK/Enterprise release pair and a real
Enterprise walkthrough remain pending.
Production Mintlify watches `lancedb/docs:deploy-freeze`, directory
`docs`; preserve that source through review. The separate SDK GitHub
Pages reference workflow rebuilds on main merges.
**Updated-head CI (9a50f8db):**
[Typos](https://github.com/lancedb/lancedb/actions/runs/37654185443/job/112904990558),
[PR
title](https://github.com/lancedb/lancedb/actions/runs/37654179507/job/112904974595)
and
[Aikido](https://app.aikido.dev/featurebranch/scan/215386145?groupId=74229)
pass. Critical findings in the example lockfiles were patched: all Sharp
copies use 0.35.5, protobufjs 7.6.6, tar 7.5.22, quinn-proto 0.11.19,
zerovec 0.11.8 and zerovec-derive 0.11.6. The SDK's product dependencies
are unchanged. The revised lockfiles pass eight TypeScript examples,
HuggingFace decode/resize/crop/rescale/normalize plus AVIF roundtrip,
and Rust Quickstart. Aikido passing does not mean every advisory is
cleared; npm still reports 29 high and five moderate advisories in the
example dependency graph. No scanner bypass or suppression was added.
The Function prose correction passes combined/standalone Mint,
deterministic assembly, routes and assets. The revised 325-file site
checksum is
`175bec06864abb400dd292e667ba93448c0479b8359fd7bf6146afabf9040647`. This
PR remains open for independent review; further merges are on hold.
---------
Co-authored-by: Brendan Clement <brendan.clement@lancedb.com>
1517 lines
51 KiB
Rust
1517 lines
51 KiB
Rust
// SPDX-License-Identifier: Apache-2.0
|
|
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration as StdDuration;
|
|
|
|
use arrow_array::types::Float32Type;
|
|
use arrow_array::{
|
|
Array, FixedSizeListArray, Float32Array, Float64Array, Int32Array, Int64Array, RecordBatch,
|
|
RecordBatchIterator, RecordBatchReader, StringArray,
|
|
};
|
|
use arrow_schema::{DataType, Field, Schema};
|
|
use futures::TryStreamExt;
|
|
use lancedb::connect;
|
|
use lancedb::query::ExecutableQuery;
|
|
use lancedb::database::CreateTableMode;
|
|
use lancedb::table::{
|
|
ColumnAlteration, Duration, FieldMetadataUpdate, NewColumnTransform, OptimizeAction,
|
|
};
|
|
|
|
// --8<-- [start:update_make_users_reader]
|
|
fn make_users_reader(
|
|
ids: Vec<i64>,
|
|
names: Vec<&str>,
|
|
login_counts: Option<Vec<i64>>,
|
|
) -> Box<dyn RecordBatchReader + Send> {
|
|
let mut fields = vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("name", DataType::Utf8, false),
|
|
];
|
|
let mut columns: Vec<Arc<dyn Array>> =
|
|
vec![Arc::new(Int64Array::from(ids)), Arc::new(StringArray::from(names))];
|
|
|
|
if let Some(login_counts) = login_counts {
|
|
fields.push(Field::new("login_count", DataType::Int64, true));
|
|
columns.push(Arc::new(Int64Array::from(login_counts)));
|
|
}
|
|
|
|
let schema = Arc::new(Schema::new(fields));
|
|
let batch = RecordBatch::try_new(schema.clone(), columns).unwrap();
|
|
let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
|
|
Box::new(reader)
|
|
}
|
|
// --8<-- [end:update_make_users_reader]
|
|
|
|
// --8<-- [start:versioning_make_quotes_reader]
|
|
fn make_quotes_reader(rows: Vec<(i64, &str, &str)>) -> Box<dyn RecordBatchReader + Send> {
|
|
let ids: Vec<i64> = rows.iter().map(|(id, _, _)| *id).collect();
|
|
let authors: Vec<&str> = rows.iter().map(|(_, author, _)| *author).collect();
|
|
let quotes: Vec<&str> = rows.iter().map(|(_, _, quote)| *quote).collect();
|
|
|
|
let schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("author", DataType::Utf8, false),
|
|
Field::new("quote", DataType::Utf8, false),
|
|
]));
|
|
|
|
let batch = RecordBatch::try_new(
|
|
schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(ids)),
|
|
Arc::new(StringArray::from(authors)),
|
|
Arc::new(StringArray::from(quotes)),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
|
|
Box::new(reader)
|
|
}
|
|
// --8<-- [end:versioning_make_quotes_reader]
|
|
|
|
// Helper: a table with a `vector` and a `text` column, used to demonstrate
|
|
// building indexes on a branch.
|
|
fn make_products_reader(n: i32) -> Box<dyn RecordBatchReader + Send> {
|
|
const DIM: usize = 4;
|
|
let schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int32, false),
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(
|
|
Arc::new(Field::new("item", DataType::Float32, true)),
|
|
DIM as i32,
|
|
),
|
|
true,
|
|
),
|
|
Field::new("text", DataType::Utf8, false),
|
|
]));
|
|
|
|
let ids = Int32Array::from_iter_values(0..n);
|
|
let vectors = FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
(0..n).map(|i| {
|
|
Some(
|
|
(0..DIM)
|
|
.map(|d| Some((((i as usize * 31 + d * 7) % 97) as f32) / 97.0))
|
|
.collect::<Vec<Option<f32>>>(),
|
|
)
|
|
}),
|
|
DIM as i32,
|
|
);
|
|
let texts = StringArray::from_iter_values((0..n).map(|i| format!("product number {i}")));
|
|
|
|
let batch = RecordBatch::try_new(
|
|
schema.clone(),
|
|
vec![Arc::new(ids), Arc::new(vectors), Arc::new(texts)],
|
|
)
|
|
.unwrap();
|
|
let reader = RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema);
|
|
Box::new(reader)
|
|
}
|
|
|
|
#[allow(dead_code)]
|
|
async fn update_connect_enterprise_example() {
|
|
// --8<-- [start:update_connect_enterprise]
|
|
let uri = "db://your-project-slug";
|
|
let api_key = "your-api-key";
|
|
let region = "us-east-1";
|
|
// --8<-- [end:update_connect_enterprise]
|
|
let _ = (uri, api_key, region);
|
|
}
|
|
|
|
#[allow(dead_code)]
|
|
async fn update_connect_local_example() {
|
|
// --8<-- [start:update_connect_local]
|
|
let db = connect("./data").execute().await.unwrap();
|
|
// --8<-- [end:update_connect_local]
|
|
let _ = db;
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() {
|
|
let temp_dir = tempfile::tempdir().unwrap();
|
|
let db_uri = temp_dir.path().to_str().unwrap().to_string();
|
|
let db = connect(&db_uri).execute().await.unwrap();
|
|
|
|
// --8<-- [start:create_table_from_dicts]
|
|
struct Location {
|
|
vector: [f32; 2],
|
|
lat: f32,
|
|
long: f32,
|
|
}
|
|
|
|
let data = vec![
|
|
Location {
|
|
vector: [1.1, 1.2],
|
|
lat: 45.5,
|
|
long: -122.7,
|
|
},
|
|
Location {
|
|
vector: [0.2, 1.8],
|
|
lat: 40.1,
|
|
long: -74.1,
|
|
},
|
|
];
|
|
|
|
let schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 2),
|
|
false,
|
|
),
|
|
Field::new("lat", DataType::Float32, false),
|
|
Field::new("long", DataType::Float32, false),
|
|
]));
|
|
|
|
let batch = RecordBatch::try_new(
|
|
schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
data.iter()
|
|
.map(|row| Some(row.vector.iter().copied().map(Some).collect::<Vec<_>>())),
|
|
2,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.lat),
|
|
)),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.long),
|
|
)),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema.clone()));
|
|
let table = db
|
|
.create_table("test_table", reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_table_from_dicts]
|
|
assert_eq!(table.count_rows(None).await.unwrap(), 2);
|
|
|
|
// Seed an existing table so the conflict-handling examples have something to act on.
|
|
let conflict_seed = RecordBatch::try_new(
|
|
schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
data.iter()
|
|
.map(|row| Some(row.vector.iter().copied().map(Some).collect::<Vec<_>>())),
|
|
2,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.lat),
|
|
)),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.long),
|
|
)),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let conflict_seed_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(conflict_seed)].into_iter(), schema.clone()));
|
|
db.create_table("conflict_table", conflict_seed_reader)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Build readers for the rows we want to ingest. Outside the snippet markers
|
|
// because the docs only need to highlight the `.mode(...)` differences.
|
|
let make_reader = || {
|
|
let batch = RecordBatch::try_new(
|
|
schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
data.iter().map(|row| {
|
|
Some(row.vector.iter().copied().map(Some).collect::<Vec<_>>())
|
|
}),
|
|
2,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.lat),
|
|
)),
|
|
Arc::new(Float32Array::from_iter_values(
|
|
data.iter().map(|row| row.long),
|
|
)),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema.clone()));
|
|
reader
|
|
};
|
|
let exist_ok_reader = make_reader();
|
|
let overwrite_reader = make_reader();
|
|
|
|
// --8<-- [start:create_table_conflict_handling]
|
|
// Idempotent open: reuse the existing table if it exists.
|
|
// The provided data is ignored; the schema is validated against the
|
|
// existing table and a mismatch raises an error.
|
|
let _conflict_table = db
|
|
.create_table("conflict_table", exist_ok_reader)
|
|
.mode(CreateTableMode::exist_ok(|req| req))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Overwrite: drop the existing table and create a new one with the
|
|
// provided data. This permanently discards the old table's data.
|
|
let conflict_table = db
|
|
.create_table("conflict_table", overwrite_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_table_conflict_handling]
|
|
assert_eq!(conflict_table.count_rows(None).await.unwrap(), 2);
|
|
|
|
// --8<-- [start:create_table_custom_schema]
|
|
let custom_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 4),
|
|
false,
|
|
),
|
|
Field::new("lat", DataType::Float32, false),
|
|
Field::new("long", DataType::Float32, false),
|
|
]));
|
|
|
|
let custom_batch = RecordBatch::try_new(
|
|
custom_schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![
|
|
Some(vec![Some(1.1), Some(1.2), Some(1.3), Some(1.4)]),
|
|
Some(vec![Some(0.2), Some(1.8), Some(0.4), Some(3.6)]),
|
|
],
|
|
4,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from(vec![45.5, 40.1])),
|
|
Arc::new(Float32Array::from(vec![-122.7, -74.1])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let custom_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(custom_batch)].into_iter(), custom_schema.clone()));
|
|
let custom_table = db
|
|
.create_table("my_table_custom_schema", custom_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_table_custom_schema]
|
|
assert_eq!(custom_table.count_rows(None).await.unwrap(), 2);
|
|
|
|
// --8<-- [start:create_table_from_arrow]
|
|
let arrow_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 16),
|
|
false,
|
|
),
|
|
Field::new("text", DataType::Utf8, false),
|
|
]));
|
|
|
|
let arrow_batch = RecordBatch::try_new(
|
|
arrow_schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![Some(vec![Some(0.1); 16]), Some(vec![Some(0.2); 16])],
|
|
16,
|
|
),
|
|
),
|
|
Arc::new(StringArray::from(vec!["foo", "bar"])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let arrow_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(arrow_batch)].into_iter(), arrow_schema.clone()));
|
|
let arrow_table = db
|
|
.create_table("arrow_table_example", arrow_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_table_from_arrow]
|
|
assert_eq!(arrow_table.count_rows(None).await.unwrap(), 2);
|
|
|
|
// --8<-- [start:create_table_from_iterator]
|
|
let batch_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 4),
|
|
false,
|
|
),
|
|
Field::new("item", DataType::Utf8, false),
|
|
Field::new("price", DataType::Float32, false),
|
|
]));
|
|
|
|
let batches = (0..5)
|
|
.map(|i| {
|
|
RecordBatch::try_new(
|
|
batch_schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![
|
|
Some(vec![Some(3.1 + i as f32), Some(4.1), Some(5.1), Some(6.1)]),
|
|
Some(vec![
|
|
Some(5.9),
|
|
Some(26.5 + i as f32),
|
|
Some(4.7),
|
|
Some(32.8),
|
|
]),
|
|
],
|
|
4,
|
|
),
|
|
),
|
|
Arc::new(StringArray::from(vec![
|
|
format!("item{}", i * 2 + 1),
|
|
format!("item{}", i * 2 + 2),
|
|
])),
|
|
Arc::new(Float32Array::from(vec![
|
|
((i * 2 + 1) * 10) as f32,
|
|
((i * 2 + 2) * 10) as f32,
|
|
])),
|
|
],
|
|
)
|
|
.unwrap()
|
|
})
|
|
.collect::<Vec<_>>();
|
|
|
|
let batch_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(batches.into_iter().map(Ok), batch_schema.clone()));
|
|
let batch_table = db
|
|
.create_table("batched_table", batch_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_table_from_iterator]
|
|
assert_eq!(batch_table.count_rows(None).await.unwrap(), 10);
|
|
|
|
// --8<-- [start:open_existing_table]
|
|
let open_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 2),
|
|
false,
|
|
),
|
|
Field::new("lat", DataType::Float32, false),
|
|
Field::new("long", DataType::Float32, false),
|
|
]));
|
|
let open_batch = RecordBatch::try_new(
|
|
open_schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![Some(vec![Some(1.1), Some(1.2)])],
|
|
2,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from(vec![45.5])),
|
|
Arc::new(Float32Array::from(vec![-122.7])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let open_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(open_batch)].into_iter(), open_schema.clone()));
|
|
db.create_table("test_table", open_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
println!("{:?}", db.table_names().execute().await.unwrap());
|
|
|
|
let opened_table = db.open_table("test_table").execute().await.unwrap();
|
|
// --8<-- [end:open_existing_table]
|
|
assert_eq!(opened_table.count_rows(None).await.unwrap(), 1);
|
|
|
|
// --8<-- [start:create_empty_table]
|
|
let empty_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 2),
|
|
false,
|
|
),
|
|
Field::new("item", DataType::Utf8, false),
|
|
Field::new("price", DataType::Float32, false),
|
|
]));
|
|
let empty_table = db
|
|
.create_empty_table("test_empty_table", empty_schema)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:create_empty_table]
|
|
assert_eq!(empty_table.count_rows(None).await.unwrap(), 0);
|
|
|
|
// --8<-- [start:drop_table]
|
|
let drop_schema = Arc::new(Schema::new(vec![
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 2),
|
|
false,
|
|
),
|
|
Field::new("lat", DataType::Float32, false),
|
|
]));
|
|
let drop_batch = RecordBatch::try_new(
|
|
drop_schema.clone(),
|
|
vec![
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![Some(vec![Some(1.1), Some(1.2)])],
|
|
2,
|
|
),
|
|
),
|
|
Arc::new(Float32Array::from(vec![45.5])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let drop_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(drop_batch)].into_iter(), drop_schema.clone()));
|
|
db.create_table("my_table", drop_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
db.drop_table("my_table", &[]).await.unwrap();
|
|
// --8<-- [end:drop_table]
|
|
assert!(
|
|
!db.table_names()
|
|
.execute()
|
|
.await
|
|
.unwrap()
|
|
.contains(&"my_table".to_string())
|
|
);
|
|
|
|
// --8<-- [start:schema_add_setup]
|
|
let schema_add_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("name", DataType::Utf8, false),
|
|
Field::new("price", DataType::Float64, false),
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 128),
|
|
false,
|
|
),
|
|
]));
|
|
let schema_add_batch = RecordBatch::try_new(
|
|
schema_add_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![1, 2, 3])),
|
|
Arc::new(StringArray::from(vec!["Laptop", "Smartphone", "Headphones"])),
|
|
Arc::new(Float64Array::from(vec![1200.0, 800.0, 150.0])),
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![
|
|
Some(vec![Some(0.1_f32); 128]),
|
|
Some(vec![Some(0.2_f32); 128]),
|
|
Some(vec![Some(0.3_f32); 128]),
|
|
],
|
|
128,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let schema_add_reader: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
|
|
vec![Ok(schema_add_batch)].into_iter(),
|
|
schema_add_schema.clone(),
|
|
));
|
|
let schema_add_table = db
|
|
.create_table("schema_evolution_add_example", schema_add_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:schema_add_setup]
|
|
assert_eq!(schema_add_table.count_rows(None).await.unwrap(), 3);
|
|
|
|
// --8<-- [start:add_columns_calculated]
|
|
// Add a discounted price column (10% discount)
|
|
schema_add_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![(
|
|
"discounted_price".to_string(),
|
|
"cast((price * 0.9) as float)".to_string(),
|
|
)]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:add_columns_calculated]
|
|
|
|
// --8<-- [start:add_columns_default_values]
|
|
// Add a stock status column with default value
|
|
schema_add_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![(
|
|
"in_stock".to_string(),
|
|
"cast(true as boolean)".to_string(),
|
|
)]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:add_columns_default_values]
|
|
|
|
// --8<-- [start:add_columns_nullable]
|
|
// Add a nullable timestamp column
|
|
schema_add_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![(
|
|
"last_ordered".to_string(),
|
|
"cast(NULL as timestamp)".to_string(),
|
|
)]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:add_columns_nullable]
|
|
|
|
// --8<-- [start:add_feature_columns_sql]
|
|
schema_add_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![
|
|
(
|
|
"price_per_id".to_string(),
|
|
"cast(price / id as float)".to_string(),
|
|
),
|
|
("price_log".to_string(), "ln(price)".to_string()),
|
|
(
|
|
"price_score".to_string(),
|
|
"cast(price / (price + 100.0) as float)".to_string(),
|
|
),
|
|
]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:add_feature_columns_sql]
|
|
|
|
// --8<-- [start:schema_alter_setup]
|
|
let schema_alter_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("name", DataType::Utf8, false),
|
|
Field::new("price", DataType::Int32, false),
|
|
Field::new("discount_price", DataType::Float64, false),
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 128),
|
|
false,
|
|
),
|
|
]));
|
|
let schema_alter_batch = RecordBatch::try_new(
|
|
schema_alter_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![1, 2])),
|
|
Arc::new(StringArray::from(vec!["Laptop", "Smartphone"])),
|
|
Arc::new(Int32Array::from(vec![1200, 800])),
|
|
Arc::new(Float64Array::from(vec![1080.0, 720.0])),
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![Some(vec![Some(0.1_f32); 128]), Some(vec![Some(0.2_f32); 128])],
|
|
128,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let schema_alter_reader: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
|
|
vec![Ok(schema_alter_batch)].into_iter(),
|
|
schema_alter_schema.clone(),
|
|
));
|
|
let schema_alter_table = db
|
|
.create_table("schema_evolution_alter_example", schema_alter_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:schema_alter_setup]
|
|
assert_eq!(schema_alter_table.count_rows(None).await.unwrap(), 2);
|
|
|
|
// --8<-- [start:alter_columns_rename]
|
|
// Rename discount_price to sale_price
|
|
schema_alter_table
|
|
.alter_columns(&[ColumnAlteration::new("discount_price".to_string())
|
|
.rename("sale_price".to_string())])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_columns_rename]
|
|
|
|
// --8<-- [start:alter_columns_data_type]
|
|
// Change price from int32 to int64 for larger numbers
|
|
schema_alter_table
|
|
.alter_columns(&[ColumnAlteration::new("price".to_string()).cast_to(DataType::Int64)])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_columns_data_type]
|
|
|
|
// --8<-- [start:alter_columns_nullable]
|
|
// Make the name column nullable
|
|
schema_alter_table
|
|
.alter_columns(&[ColumnAlteration::new("name".to_string()).set_nullable(true)])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_columns_nullable]
|
|
|
|
// --8<-- [start:alter_columns_multiple]
|
|
// Rename, change type, and make nullable in one operation
|
|
schema_alter_table
|
|
.alter_columns(&[ColumnAlteration::new("sale_price".to_string())
|
|
.rename("final_price".to_string())
|
|
.cast_to(DataType::Float64)
|
|
.set_nullable(true)])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_columns_multiple]
|
|
|
|
// --8<-- [start:alter_columns_with_expression]
|
|
// For custom transforms, create a new column from a SQL expression.
|
|
let expression_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("price_text", DataType::Utf8, false),
|
|
]));
|
|
let expression_batch = RecordBatch::try_new(
|
|
expression_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![1])),
|
|
Arc::new(StringArray::from(vec!["$100"])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let expression_reader: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
|
|
vec![Ok(expression_batch)].into_iter(),
|
|
expression_schema.clone(),
|
|
));
|
|
let expression_table = db
|
|
.create_table("schema_evolution_expression_example", expression_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
expression_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![(
|
|
"price_numeric".to_string(),
|
|
"cast(replace(price_text, '$', '') as int)".to_string(),
|
|
)]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
expression_table.drop_columns(&["price_text"]).await.unwrap();
|
|
expression_table
|
|
.alter_columns(&[ColumnAlteration::new("price_numeric".to_string())
|
|
.rename("price".to_string())])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_columns_with_expression]
|
|
assert_eq!(expression_table.count_rows(None).await.unwrap(), 1);
|
|
|
|
// --8<-- [start:schema_drop_setup]
|
|
let schema_drop_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("name", DataType::Utf8, false),
|
|
Field::new("price", DataType::Float64, false),
|
|
Field::new("temp_col1", DataType::Utf8, false),
|
|
Field::new("temp_col2", DataType::Int32, false),
|
|
Field::new(
|
|
"vector",
|
|
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 128),
|
|
false,
|
|
),
|
|
]));
|
|
let schema_drop_batch = RecordBatch::try_new(
|
|
schema_drop_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![1, 2, 3])),
|
|
Arc::new(StringArray::from(vec!["Laptop", "Smartphone", "Headphones"])),
|
|
Arc::new(Float64Array::from(vec![1200.0, 800.0, 150.0])),
|
|
Arc::new(StringArray::from(vec!["X", "Y", "Z"])),
|
|
Arc::new(Int32Array::from(vec![100, 200, 300])),
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![
|
|
Some(vec![Some(0.1_f32); 128]),
|
|
Some(vec![Some(0.2_f32); 128]),
|
|
Some(vec![Some(0.3_f32); 128]),
|
|
],
|
|
128,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let schema_drop_reader: Box<dyn RecordBatchReader + Send> = Box::new(RecordBatchIterator::new(
|
|
vec![Ok(schema_drop_batch)].into_iter(),
|
|
schema_drop_schema.clone(),
|
|
));
|
|
let schema_drop_table = db
|
|
.create_table("schema_evolution_drop_example", schema_drop_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:schema_drop_setup]
|
|
assert_eq!(schema_drop_table.count_rows(None).await.unwrap(), 3);
|
|
|
|
// --8<-- [start:drop_columns_single]
|
|
// Remove the first temporary column
|
|
schema_drop_table.drop_columns(&["temp_col1"]).await.unwrap();
|
|
// --8<-- [end:drop_columns_single]
|
|
|
|
// --8<-- [start:drop_columns_multiple]
|
|
// Remove the second temporary column
|
|
schema_drop_table.drop_columns(&["temp_col2"]).await.unwrap();
|
|
// --8<-- [end:drop_columns_multiple]
|
|
|
|
// --8<-- [start:alter_vector_column]
|
|
let old_dim = 384;
|
|
let new_dim = 1024;
|
|
let vector_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new(
|
|
"embedding",
|
|
DataType::FixedSizeList(
|
|
Arc::new(Field::new("item", DataType::Float32, true)),
|
|
old_dim,
|
|
),
|
|
true,
|
|
),
|
|
]));
|
|
let vector_batch = RecordBatch::try_new(
|
|
vector_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![1])),
|
|
Arc::new(
|
|
FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
|
|
vec![Some(vec![Some(0.1_f32); old_dim as usize])],
|
|
old_dim,
|
|
),
|
|
),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let vector_reader: Box<dyn RecordBatchReader + Send> =
|
|
Box::new(RecordBatchIterator::new(vec![Ok(vector_batch)].into_iter(), vector_schema.clone()));
|
|
let vector_table = db
|
|
.create_table("vector_alter_example", vector_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Changing FixedSizeList dimensions (384 -> 1024) is not supported via alter_columns.
|
|
// Use add_columns + drop_columns + alter_columns(rename) to replace the column.
|
|
vector_table
|
|
.add_columns(
|
|
NewColumnTransform::SqlExpressions(vec![(
|
|
"embedding_v2".to_string(),
|
|
format!("arrow_cast(NULL, 'FixedSizeList({}, Float32)')", new_dim),
|
|
)]),
|
|
None,
|
|
)
|
|
.await
|
|
.unwrap();
|
|
vector_table.drop_columns(&["embedding"]).await.unwrap();
|
|
vector_table
|
|
.alter_columns(&[ColumnAlteration::new("embedding_v2".to_string())
|
|
.rename("embedding".to_string())])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:alter_vector_column]
|
|
assert_eq!(vector_table.count_rows(None).await.unwrap(), 1);
|
|
|
|
let field_metadata_schema = Arc::new(Schema::new(vec![
|
|
Field::new("id", DataType::Int64, false),
|
|
Field::new("category", DataType::Utf8, false),
|
|
]));
|
|
let field_metadata_batch = RecordBatch::try_new(
|
|
field_metadata_schema.clone(),
|
|
vec![
|
|
Arc::new(Int64Array::from(vec![0, 1])),
|
|
Arc::new(StringArray::from(vec!["a", "b"])),
|
|
],
|
|
)
|
|
.unwrap();
|
|
let field_metadata_reader: Box<dyn RecordBatchReader + Send> = Box::new(
|
|
RecordBatchIterator::new(vec![Ok(field_metadata_batch)].into_iter(), field_metadata_schema),
|
|
);
|
|
let field_metadata_table = db
|
|
.create_table("schema_field_metadata_example", field_metadata_reader)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// --8<-- [start:schema_field_metadata_merge]
|
|
// Set two metadata keys on the `category` field.
|
|
let res = field_metadata_table
|
|
.update_field_metadata(&[FieldMetadataUpdate::new("category")
|
|
.set("unit", "label")
|
|
.set("pii", "false")])
|
|
.await
|
|
.unwrap();
|
|
println!("version: {}", res.version);
|
|
|
|
// Merge: add a new key, delete one with `.remove`, keep the rest.
|
|
field_metadata_table
|
|
.update_field_metadata(&[FieldMetadataUpdate::new("category")
|
|
.set("source", "import")
|
|
.remove("pii")])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:schema_field_metadata_merge]
|
|
|
|
// --8<-- [start:schema_field_metadata_replace]
|
|
field_metadata_table
|
|
.update_field_metadata(&[FieldMetadataUpdate::new("category")
|
|
.set("owner", "search-team")
|
|
.replace()])
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:schema_field_metadata_replace]
|
|
|
|
// --8<-- [start:update_example_table_setup]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:update_example_table_setup]
|
|
let _ = table;
|
|
|
|
// --8<-- [start:update_operation]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
table
|
|
.update()
|
|
.only_if("id = 2")
|
|
.column("name", "'Bobby'")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:update_operation]
|
|
|
|
// --8<-- [start:update_using_sql]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
table
|
|
.update()
|
|
.only_if("id = 2")
|
|
.column("login_count", "login_count + 1")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:update_using_sql]
|
|
|
|
// --8<-- [start:merge_matched_update_only]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut merge_insert = table.merge_insert(&["id"]);
|
|
merge_insert.when_matched_update_all(None);
|
|
merge_insert
|
|
.execute(make_users_reader(
|
|
vec![2, 3],
|
|
vec!["Bobby", "Charlie"],
|
|
Some(vec![21, 5]),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:merge_matched_update_only]
|
|
|
|
// --8<-- [start:insert_if_not_exists]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut merge_insert = table.merge_insert(&["id"]);
|
|
merge_insert.when_not_matched_insert_all();
|
|
merge_insert
|
|
.execute(make_users_reader(
|
|
vec![2, 3],
|
|
vec!["Bobby", "Charlie"],
|
|
Some(vec![21, 5]),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:insert_if_not_exists]
|
|
|
|
// --8<-- [start:merge_update_insert]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut merge_insert = table.merge_insert(&["id"]);
|
|
merge_insert
|
|
.when_matched_update_all(None)
|
|
.when_not_matched_insert_all();
|
|
merge_insert
|
|
.execute(make_users_reader(
|
|
vec![2, 3],
|
|
vec!["Bobby", "Charlie"],
|
|
Some(vec![21, 5]),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:merge_update_insert]
|
|
|
|
// --8<-- [start:merge_delete_missing_by_source]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(
|
|
vec![1, 2, 3],
|
|
vec!["Alice", "Bob", "Charlie"],
|
|
Some(vec![10, 20, 5]),
|
|
),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut merge_insert = table.merge_insert(&["id"]);
|
|
merge_insert
|
|
.when_matched_update_all(None)
|
|
.when_not_matched_insert_all()
|
|
.when_not_matched_by_source_delete(None);
|
|
merge_insert
|
|
.execute(make_users_reader(
|
|
vec![2, 3],
|
|
vec!["Bobby", "Charlie"],
|
|
Some(vec![21, 5]),
|
|
))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:merge_delete_missing_by_source]
|
|
|
|
// --8<-- [start:merge_partial_columns]
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(vec![1, 2], vec!["Alice", "Bob"], Some(vec![10, 20])),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let mut merge_insert = table.merge_insert(&["id"]);
|
|
merge_insert
|
|
.when_matched_update_all(None)
|
|
.when_not_matched_insert_all();
|
|
merge_insert
|
|
.execute(make_users_reader(vec![2, 3], vec!["Bobby", "Charlie"], None))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:merge_partial_columns]
|
|
|
|
let table = db
|
|
.create_table(
|
|
"users_example",
|
|
make_users_reader(
|
|
vec![1, 2, 3],
|
|
vec!["Alice", "Bob", "Charlie"],
|
|
Some(vec![10, 20, 5]),
|
|
),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// --8<-- [start:delete_operation]
|
|
// delete data
|
|
let predicate = "id = 3";
|
|
table.delete(predicate).await.unwrap();
|
|
// --8<-- [end:delete_operation]
|
|
|
|
let table = db
|
|
.create_table(
|
|
"users_cleanup_example",
|
|
make_users_reader(
|
|
vec![1, 2, 3],
|
|
vec!["Alice", "Bob", "Charlie"],
|
|
Some(vec![10, 20, 5]),
|
|
),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// --8<-- [start:update_optimize_cleanup]
|
|
table
|
|
.optimize(OptimizeAction::Prune {
|
|
older_than: Some(Duration::days(1)),
|
|
delete_unverified: None,
|
|
error_if_tagged_old_versions: None,
|
|
})
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:update_optimize_cleanup]
|
|
|
|
// --8<-- [start:consistency_strong]
|
|
let strong_writer_db = connect(&db_uri).execute().await.unwrap();
|
|
let strong_reader_db = connect(&db_uri)
|
|
.read_consistency_interval(StdDuration::from_secs(0))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let strong_writer_table = strong_writer_db
|
|
.create_table(
|
|
"consistency_strong_table",
|
|
make_users_reader(vec![1], vec!["Alice"], None),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let strong_reader_table = strong_reader_db
|
|
.open_table("consistency_strong_table")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
strong_writer_table
|
|
.add(make_users_reader(vec![2], vec!["Bob"], None))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let strong_rows_after_write = strong_reader_table.count_rows(None).await.unwrap();
|
|
println!(
|
|
"Rows visible with strong consistency: {}",
|
|
strong_rows_after_write
|
|
);
|
|
// --8<-- [end:consistency_strong]
|
|
assert_eq!(strong_rows_after_write, 2);
|
|
|
|
// --8<-- [start:consistency_eventual]
|
|
let eventual_writer_db = connect(&db_uri).execute().await.unwrap();
|
|
let eventual_reader_db = connect(&db_uri)
|
|
.read_consistency_interval(StdDuration::from_secs(3600))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let eventual_writer_table = eventual_writer_db
|
|
.create_table(
|
|
"consistency_eventual_table",
|
|
make_users_reader(vec![1], vec!["Alice"], None),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let eventual_reader_table = eventual_reader_db
|
|
.open_table("consistency_eventual_table")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
eventual_writer_table
|
|
.add(make_users_reader(vec![2], vec!["Bob"], None))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let eventual_rows_after_write = eventual_reader_table.count_rows(None).await.unwrap();
|
|
println!(
|
|
"Rows visible before eventual refresh interval: {}",
|
|
eventual_rows_after_write
|
|
);
|
|
// --8<-- [end:consistency_eventual]
|
|
assert_eq!(eventual_rows_after_write, 1);
|
|
|
|
// --8<-- [start:consistency_checkout_latest]
|
|
let checkout_writer_db = connect(&db_uri).execute().await.unwrap();
|
|
let checkout_reader_db = connect(&db_uri).execute().await.unwrap();
|
|
let checkout_writer_table = checkout_writer_db
|
|
.create_table(
|
|
"consistency_checkout_latest_table",
|
|
make_users_reader(vec![1], vec!["Alice"], None),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let checkout_reader_table = checkout_reader_db
|
|
.open_table("consistency_checkout_latest_table")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
checkout_writer_table
|
|
.add(make_users_reader(vec![2], vec!["Bob"], None))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let rows_before_refresh = checkout_reader_table.count_rows(None).await.unwrap();
|
|
println!("Rows before checkout_latest: {}", rows_before_refresh);
|
|
checkout_reader_table.checkout_latest().await.unwrap();
|
|
let rows_after_refresh = checkout_reader_table.count_rows(None).await.unwrap();
|
|
println!("Rows after checkout_latest: {}", rows_after_refresh);
|
|
// --8<-- [end:consistency_checkout_latest]
|
|
assert_eq!(rows_before_refresh, 1);
|
|
assert_eq!(rows_after_refresh, 2);
|
|
|
|
// --8<-- [start:versioning_basic_setup]
|
|
let table_name = "quotes_versioning_example";
|
|
let data = vec![
|
|
(1, "Richard", "Wubba Lubba Dub Dub!"),
|
|
(2, "Morty", "Rick, what's going on?"),
|
|
(3, "Richard", "I turned myself into a pickle, Morty!"),
|
|
];
|
|
|
|
let table = db
|
|
.create_table(table_name, make_quotes_reader(data))
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:versioning_basic_setup]
|
|
assert_eq!(table.count_rows(None).await.unwrap(), 3);
|
|
|
|
// --8<-- [start:versioning_check_initial_version]
|
|
let versions = table.list_versions().await.unwrap();
|
|
let current_version = table.version().await.unwrap();
|
|
println!("Number of versions after creation: {}", versions.len());
|
|
println!("Current version: {}", current_version);
|
|
// --8<-- [end:versioning_check_initial_version]
|
|
assert_eq!(versions.len(), 1);
|
|
assert_eq!(current_version, versions.last().unwrap().version);
|
|
|
|
// --8<-- [start:versioning_update_data]
|
|
table
|
|
.update()
|
|
.only_if("author = 'Richard'")
|
|
.column("author", "'Richard Daniel Sanchez'")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
let rows_after_update = table
|
|
.count_rows(Some("author = 'Richard Daniel Sanchez'".to_string()))
|
|
.await
|
|
.unwrap();
|
|
println!(
|
|
"Rows updated to Richard Daniel Sanchez: {}",
|
|
rows_after_update
|
|
);
|
|
// --8<-- [end:versioning_update_data]
|
|
assert_eq!(rows_after_update, 2);
|
|
|
|
// --8<-- [start:versioning_add_data]
|
|
let more_data = vec![
|
|
(4, "Richard Daniel Sanchez", "That's the way the news goes!"),
|
|
(5, "Morty", "Aww geez, Rick!"),
|
|
];
|
|
table
|
|
.add(make_quotes_reader(more_data))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:versioning_add_data]
|
|
assert_eq!(table.count_rows(None).await.unwrap(), 5);
|
|
|
|
// --8<-- [start:versioning_check_versions_after_mod]
|
|
let versions_after_mod = table.list_versions().await.unwrap();
|
|
let version_count_after_mod = versions_after_mod.len();
|
|
let version_after_mod = table.version().await.unwrap();
|
|
println!(
|
|
"Number of versions after modifications: {}",
|
|
version_count_after_mod
|
|
);
|
|
println!("Current version: {}", version_after_mod);
|
|
// --8<-- [end:versioning_check_versions_after_mod]
|
|
assert!(version_count_after_mod >= 2);
|
|
assert_eq!(version_after_mod, versions_after_mod.last().unwrap().version);
|
|
|
|
// --8<-- [start:versioning_list_all_versions]
|
|
let all_versions = table.list_versions().await.unwrap();
|
|
for v in &all_versions {
|
|
println!("Version {}, created at {}", v.version, v.timestamp);
|
|
}
|
|
// --8<-- [end:versioning_list_all_versions]
|
|
assert!(!all_versions.is_empty());
|
|
|
|
// --8<-- [start:versioning_rollback]
|
|
table.checkout(version_after_mod).await.unwrap();
|
|
table.restore().await.unwrap();
|
|
let versions_after_rollback = table.list_versions().await.unwrap();
|
|
let version_count_after_rollback = versions_after_rollback.len();
|
|
println!(
|
|
"Total number of versions after rollback: {}",
|
|
version_count_after_rollback
|
|
);
|
|
// --8<-- [end:versioning_rollback]
|
|
assert_eq!(version_count_after_rollback, version_count_after_mod + 1);
|
|
assert_eq!(table.count_rows(None).await.unwrap(), 5);
|
|
|
|
// --8<-- [start:versioning_checkout_latest]
|
|
table.checkout_latest().await.unwrap();
|
|
// --8<-- [end:versioning_checkout_latest]
|
|
let latest_version = table.version().await.unwrap();
|
|
let versions_after_checkout = table.list_versions().await.unwrap();
|
|
assert_eq!(latest_version, versions_after_checkout.last().unwrap().version);
|
|
|
|
// --8<-- [start:versioning_delete_data]
|
|
table.delete("author = 'Morty'").await.unwrap();
|
|
let rows_after_deletion = table.count_rows(None).await.unwrap();
|
|
println!("Number of rows after deletion: {}", rows_after_deletion);
|
|
// --8<-- [end:versioning_delete_data]
|
|
assert_eq!(rows_after_deletion, 3);
|
|
|
|
// Setup: build a table with three versions to operate on with tags.
|
|
let tags_table = db
|
|
.create_table(
|
|
"quotes_tags_example",
|
|
make_quotes_reader(vec![(1, "Richard", "Wubba Lubba Dub Dub!")]),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap(); // v1
|
|
tags_table
|
|
.add(make_quotes_reader(vec![(2, "Morty", "Aww geez, Rick!")]))
|
|
.execute()
|
|
.await
|
|
.unwrap(); // v2
|
|
tags_table
|
|
.add(make_quotes_reader(vec![(3, "Summer", "Whatever, Grandpa")]))
|
|
.execute()
|
|
.await
|
|
.unwrap(); // v3
|
|
|
|
// --8<-- [start:versioning_tags]
|
|
let mut tags = tags_table.tags().await.unwrap();
|
|
|
|
// Create a tag pointing at a specific version
|
|
tags.create("baseline", 1).await.unwrap();
|
|
let current_version = tags_table.version().await.unwrap();
|
|
tags.create("with-edits", current_version).await.unwrap();
|
|
|
|
// List all tags on this table
|
|
let all_tags = tags.list().await.unwrap();
|
|
println!("Tags: {:?}", all_tags);
|
|
|
|
// Look up the version a tag points at
|
|
let baseline_version = tags.get_version("baseline").await.unwrap();
|
|
println!("baseline -> v{}", baseline_version);
|
|
|
|
// Move an existing tag to a different version
|
|
tags.update("baseline", 2).await.unwrap();
|
|
|
|
// Check out a version by tag name (separate method in Rust)
|
|
tags_table.checkout_tag("baseline").await.unwrap();
|
|
println!("Current version: {}", tags_table.version().await.unwrap());
|
|
|
|
// Delete a tag (does not delete the underlying version)
|
|
tags.delete("with-edits").await.unwrap();
|
|
|
|
// Return to the latest version
|
|
tags_table.checkout_latest().await.unwrap();
|
|
// --8<-- [end:versioning_tags]
|
|
assert_eq!(tags_table.version().await.unwrap(), 3);
|
|
let remaining = tags.list().await.unwrap();
|
|
assert!(remaining.contains_key("baseline"));
|
|
assert!(!remaining.contains_key("with-edits"));
|
|
|
|
// Setup: a fresh quotes table to demonstrate branches on.
|
|
let branches_table = db
|
|
.create_table(
|
|
"quotes_branches_example",
|
|
make_quotes_reader(vec![
|
|
(1, "Lancelot", "My lance never fails."),
|
|
(2, "Arthur", "Long live Camelot!"),
|
|
(3, "Merlin", "Magic always has a price."),
|
|
]),
|
|
)
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
use lancedb::table::Ref;
|
|
|
|
// --8<-- [start:branch_create]
|
|
// Fork an isolated, writable branch from main's latest version.
|
|
// `create_branch` returns a table handle scoped to the new branch.
|
|
let branch = branches_table
|
|
.create_branch("exp", Ref::Version(None, None))
|
|
.await
|
|
.unwrap();
|
|
// --8<-- [end:branch_create]
|
|
|
|
// --8<-- [start:branch_write]
|
|
// Writes land on the branch handle only; main is left untouched.
|
|
branch
|
|
.add(make_quotes_reader(vec![(4, "Lancelot", "For the realm!")]))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
println!("Branch rows: {}", branch.count_rows(None).await.unwrap()); // 4
|
|
println!("Main rows: {}", branches_table.count_rows(None).await.unwrap()); // 3
|
|
|
|
// List every branch, each mapped to its metadata (including its fork point).
|
|
println!("Branches: {:?}", branches_table.list_branches().await.unwrap());
|
|
// --8<-- [end:branch_write]
|
|
|
|
// --8<-- [start:branch_reopen]
|
|
// Reopen an existing branch by name from the table handle...
|
|
let checked_out = branches_table.checkout_branch("exp", None).await.unwrap();
|
|
// ...or open it directly via the connection's builder.
|
|
let opened = db
|
|
.open_table("quotes_branches_example")
|
|
.branch("exp")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
println!(
|
|
"Reopened rows: {}, {}",
|
|
checked_out.count_rows(None).await.unwrap(),
|
|
opened.count_rows(None).await.unwrap()
|
|
); // both 4
|
|
// --8<-- [end:branch_reopen]
|
|
|
|
// --8<-- [start:branch_delete]
|
|
// Delete the branch and its branch-local history. Data on main is safe.
|
|
branches_table.delete_branch("exp").await.unwrap();
|
|
// --8<-- [end:branch_delete]
|
|
|
|
assert_eq!(branches_table.count_rows(None).await.unwrap(), 3);
|
|
assert!(!branches_table.list_branches().await.unwrap().contains_key("exp"));
|
|
|
|
// Setup: a branch with row results that we want to apply to main.
|
|
let candidate = branches_table
|
|
.create_branch("candidate", Ref::Version(None, None))
|
|
.await
|
|
.unwrap();
|
|
candidate
|
|
.update()
|
|
.only_if("id = 1")
|
|
.column("quote", "'Revised on the branch'")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
candidate
|
|
.add(make_quotes_reader(vec![(
|
|
4,
|
|
"Galahad",
|
|
"The grail awaits.",
|
|
)]))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// --8<-- [start:branch_upsert_to_main]
|
|
// This is a row-level upsert, not a merge of branch histories.
|
|
// `merge_insert` updates matching rows and inserts new rows using a stable
|
|
// unique key. Filter the branch read if you only want to apply some results.
|
|
let schema = candidate.schema().await.unwrap();
|
|
let batches = candidate
|
|
.query()
|
|
.execute()
|
|
.await
|
|
.unwrap()
|
|
.try_collect::<Vec<_>>()
|
|
.await
|
|
.unwrap();
|
|
let rows_to_apply = RecordBatchIterator::new(batches.into_iter().map(Ok), schema);
|
|
|
|
let mut merge = branches_table.merge_insert(&["id"]);
|
|
merge
|
|
.when_matched_update_all(None) // update rows that already exist on main
|
|
.when_not_matched_insert_all(); // insert rows that are new on the branch
|
|
merge.execute(Box::new(rows_to_apply)).await.unwrap();
|
|
// --8<-- [end:branch_upsert_to_main]
|
|
|
|
assert_eq!(branches_table.count_rows(None).await.unwrap(), 4);
|
|
branches_table.delete_branch("candidate").await.unwrap();
|
|
|
|
// Setup: a larger table with a vector and a text column to index.
|
|
let products = db
|
|
.create_table("products_branch_index", make_products_reader(512))
|
|
.mode(CreateTableMode::Overwrite)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// --8<-- [start:branch_index]
|
|
use lancedb::index::scalar::FtsIndexBuilder;
|
|
use lancedb::index::Index;
|
|
|
|
// Build and validate indexes on a branch before using the configuration on
|
|
// main.
|
|
let dev = products
|
|
.create_branch("index-dev", Ref::Version(None, None))
|
|
.await
|
|
.unwrap();
|
|
|
|
// A vector (ANN) index and a full-text search index, both branch-scoped.
|
|
dev.create_index(&["vector"], Index::Auto)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
dev.create_index(&["text"], Index::FTS(FtsIndexBuilder::default()))
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
// Both indexes live only on the branch; main still has none.
|
|
println!("Branch indexes: {}", dev.list_indices().await.unwrap().len()); // 2
|
|
println!("Main indexes: {}", products.list_indices().await.unwrap().len()); // 0
|
|
// --8<-- [end:branch_index]
|
|
|
|
assert_eq!(dev.list_indices().await.unwrap().len(), 2);
|
|
assert_eq!(products.list_indices().await.unwrap().len(), 0);
|
|
products.delete_branch("index-dev").await.unwrap();
|
|
}
|