2 posts tagged with "LanceDB"

LanceDB是一种使用持久存储构建的用于矢量搜索的开源数据库，极大简化了检索、过滤和嵌入管理。 LanceDB的主要功能包括：

生产规模的矢量搜索，无需管理服务器。
存储、查询和过滤矢量、元数据和多模态数据（文本、图像、视频、点云等）。
支持矢量相似性搜索、全文搜索和SQL。
本地支持Python和Javascript/Typescript。
零拷贝、自动版本管理，管理数据版本而无需额外基础设施。
GPU支持构建矢量索引（*）。
生态系统集成

LangChain🦜️🔗
LlamaIndex🦙
Apache-Arrow
Pandas
Polars
DuckDB

关于LanceDB中`Table`介绍

`Table`的方法

name(): 获取表的名称
schema(): 获取表的模式
count_rows(): 获取表中的行数
add(): 添加记录添加到表中，但传入的参数是需要实现IntoArrow的类型
query(): 查询表中的记录
update(): 更新表中的记录
delete(): 删除表中的记录
create_index(): 创建索引
merge_insert(): 合并插入
vector_search(): 矢量搜索
optimize(): 优化表
add_columns(): 添加(多)列
alter_columns(): 修改(多)列
drop_columns(): 删除(多)列
version(): 获取表的版本，由于LanceDB使用版本控制变化
checkpoint(): 根据指定版本获取检查点
checkpoint_latest(): 获取最新检查点
restore(): 恢复到指定版本
list_indices(): 列出表的索引

`Schema`模式定义`Table`定义方式

LanceDB Schema关系图

LanceDB Table依赖关系图

创建空表

1 完整代码

use arrow_schema::{DataType, Field, Schema};
use lancedb::{connect, Result};
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<()> {
  let created_empty_talbe = create_empty_table().await?;
  println!(
    "Created empty table: {}, Table name: {}",
    created_empty_talbe,
    created_empty_talbe.name()
  );
  Ok(())
}

#[allow(unused)]
async fn create_empty_table() -> Result<LanceDbTable> {
	// 创建模式定义
  let schema = Arc::new(Schema::new(vec![
    Field::new("id", DataType::Int32, false),
    Field::new("name", DataType::Utf8, false),
  ]));
	// 创建数据库URI目录
  let uri = "data/sample-lancedb";
	// 连接数据库
  let db = connect(uri).execute().await?;
	// 创建一个空表
  let table = db
    .create_empty_table("empty_talbe", schema)
    .execute()
    .await?;
  Ok(table)
}

包依赖文件 Cargo.toml文件内容如下：

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}
arrow-schema = "51.0"

运行结果如下:

创建带初始化数据的表

完整代码如下:

use arrow_schema::{DataType, Field, Schema};
use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator, StringArray};
use lancedb::{connect, Result, Table as LanceDbTable};
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<()> {
  let created_table_with_data = create_table_with_data().await?;
  println!(
    "Created table with data: {}, Table name: {}",
    created_table_with_data,
    created_table_with_data.name()
  );
  Ok(())
}

#[allow(unused)]
async fn create_table_with_data() -> Result<LanceDbTable> {
  // 创建本地数据库URI目录
  let uri = "data/sample-lancedb";
  // 连接数据库
  let db = connect(uri).execute().await?;

  // 创建模式定义
  let schema = Arc::new(Schema::new(vec![
    Field::new("id", DataType::Int32, false),
    Field::new("name", DataType::Utf8, false),
  ]));

  // 初始化`ids`列的数据
  let ids = Int32Array::from(vec![1, 2, 3]); 
  // 初始化`name`列的数据
  let names = StringArray::from(vec!["Alice", "Bob", "Lily"]); 
  // 使用`Schema`以及列数据创建`RecordBatch`
  let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(names)])?;
  // 使用`RecordBatch`创建`RecordBatchIterator`
  let batchs = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema);
  // 创建表，并插入初始化数据
  let table = db
    .create_table("table_with_person", batchs)
    .execute()
    .await?;
  Ok(table)
}

包依赖文件 Cargo.toml文件内容如下：

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}
arrow-schema = "51.0"

运行结果如下:

初始化记录并创建表

完整代码

use arrow_array::types::Float32Type;
use arrow_array::{FixedSizeListArray, Int32Array, RecordBatch, RecordBatchIterator, StringArray};
use arrow_schema::{DataType, Field, Schema};
use lancedb::arrow::IntoArrow;
use lancedb::{connect, Result, Table as LanceDbTable};
use std::sync::Arc;

#[tokio::main]
async fn main() -> Result<()> {
  let created_table_with_records = create_table_with_records().await?;
  println!(
    "Created table with records: {}, Table name: {}",
    created_table_with_records,
    created_table_with_records.name()
  );
  Ok(())
}

#[allow(unused)]
async fn create_table_with_records() -> Result<LanceDbTable> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;

  let initial_data = create_some_records()?;
  let tbl = db.create_table("my_table", initial_data).execute().await?;

  let new_data = create_some_records()?;
  // NOTICE: 只有实现了 IntoArrow 的类型才能使用`add`方法，即`create_some_records`返回的类型
  tbl.add(new_data).execute().await?;
  Ok(tbl)
}

#[allow(unused)]
fn create_some_records() -> Result<impl IntoArrow> {
  const TOTAL: usize = 1000;
  const DIM: usize = 128;

  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,
    ),
  ]));

  let batch = RecordBatch::try_new(
    schema.clone(),
    vec![
      Arc::new(Int32Array::from_iter_values(0..TOTAL as i32)),
      Arc::new(
        FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
          (0..TOTAL).map(|_| Some(vec![Some(1.0); DIM])),
          DIM as i32,
        ),
      ),
    ],
  )
  .unwrap();
  let batches = RecordBatchIterator::new(vec![batch].into_iter().map(Ok), schema.clone());
  Ok(Box::new(batches))
}

包依赖文件 Cargo.toml文件内容如下：

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}
arrow-schema = "51.0"
arrow-array = "51.0"

运行结果如下:

打开已存在的表

完整代码

use lancedb::{connect, Result, Table as LanceDbTable};

async fn main() -> Result<()> {
  let opened_table = open_with_existing_table().await?;
  println!(
    "Opened table: {}, Table name: {}",
    opened_table,
    opened_table.name()
  );
}

#[allow(unused)]
async fn open_with_existing_table() -> Result<LanceDbTable> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  let table = db.open_table("my_table").execute().await?;
  Ok(table)
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

运行结果如下:

删除表记录

完整代码

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {
  let queried_result = query_table().await?;
  println!("Query result: {:?}", queried_result);

  delete_table_records().await?; // 根据条件删除表中的记录
  Ok(())
}

#[allow(unused)]
async fn delete_table_records() -> Result<()> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  let table = db.open_table("my_table").execute().await?;
  table.delete("id > 24").await?;
  Ok(())
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

删除表

完整代码

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {
  let queried_result = query_table().await?;
  println!("Query result: {:?}", queried_result);

  drop_table().await?; // 删除 data/sample-lancedb/my_table
  Ok(())
}

#[allow(unused)]
async fn drop_table() -> Result<()> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  db.drop_table("my_table").await?;
  Ok(())
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

删除数据库

完整代码

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {
  let queried_result = query_table().await?;
  println!("Query result: {:?}", queried_result);

  drop_database().await?; // 删除 data/sample-lancedb
  Ok(())
}

#[allow(unused)]
async fn drop_database() -> Result<()> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  db.drop_db().await?;
  Ok(())
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

查询表记录

完整代码

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {

  let queried_result = query_table().await?;
  println!("Query result: {:?}", queried_result);
  Ok(())
}

#[allow(unused)]
async fn query_table() -> Result<VectorQuery> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  let table = db.open_table("my_table").execute().await?;
  let result = table.query().nearest_to(&[1.0; 128])?;
  Ok(result)
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

运行结果如下:

更新表记录

完整代码

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {

  update_table().await?;
  Ok(())
}

#[allow(unused)]
async fn update_table() -> Result<()> {
  let uri = "data/sample-lancedb";
  let db = connect(uri).execute().await?;
  let table = db.open_table("table_with_person").execute().await?;
  println!("Before update: {:?}", table.query());
  table
    .update()
    .only_if("id=0")
    .column("name", "Bob")
    .execute()
    .await?; // Bob -> Halzzz

  Ok(())
}

包依赖文件

lancedb = "0.7.0"
tokio = {version = "1.38.0", features = ["rt-multi-thread"]}

总结

本文详细介绍了如何使用Rust编程语言与LanceDB进行交互，包括表的创建、插入数据、查询、更新和删除操作。通过这些示例，展示了LanceDB在处理矢量数据和支持多模态数据方面的强大功能，以及如何通过Rust代码实现这些操作。 LanceDB提供了丰富的API接口，简化了数据库操作，使得开发者能够高效地管理和查询数据。

希望通过本文的讲解，您能够更好地理解并应用LanceDB来解决实际问题。

链接

Github Repo: https://github.com/yuxuetr/lancedb-examples/tree/main/about_a_table

LanceDB Get Started

Rust LanceDB

本文介绍从头开始使用LanceDB，每一个步骤会给出详细的说明和解释以及图示说明，帮助读者快速上手LanceDB。

LanceDB创建数据库基本调用关系图示

环境搭建

Rust
tokio
lancedb

Rust安装

curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh

创建Rust项目

cargo new lancedb-example
cd lancedb-example

安装Tokio

cargo add tokio --features rt-multi-thread

安装LanceDB

cargo add lancedb

代码解析

use lancedb::{connect, Result};

#[tokio::main]
async fn main() -> Result<()> {
  uri = "data/example-lancedb";
  db_builder = connect(uri);
  db_connect = connect(uri).execute().await?;
  println!("LanceDB builder: {:}", db_builder);
  println!("LanceDB connect: {}", db_connect);
  Ok(())
}

note

上述代码创建了一个最简单的LanceDB数据库创建和连接，并且打印ConnectBuilder和Connection:

uri: 表示数据库的URI
connect: 接收URI参数，返回ConnectBuilder, COnnectBuilder实现了Debug trait
execute: 返回Connection, 实现了Display trait，会在执行的目录创建 data/example-lancedb目录

`connect`函数的定义如下

/// Connect to a LanceDB database.
///
/// # Arguments
///
/// * `uri` - URI where the database is located, can be a local directory, supported remote cloud storage,
///           or a LanceDB Cloud database.  See [ConnectOptions::uri] for a list of accepted formats
pub fn connect(uri: &str) -> ConnectBuilder {
  ConnectBuilder::new(uri)
}

note

在connect函数中创建了ConnectBuilder，调用connect即创建了一个ConnectBuilder

`ConnectBuilder`

pub struct ConnectBuilder {
  /// Database URI
  ///
  /// ### Accpeted URI formats
  ///
  /// - `/path/to/database` - local database on file system.
  /// - `s3://bucket/path/to/database` or `gs://bucket/path/to/database` - database on cloud object store
  /// - `db://dbname` - LanceDB Cloud
  uri: String,

  /// LanceDB Cloud API key, required if using Lance Cloud
  api_key: Option<String>,
  /// LanceDB Cloud region, required if using Lance Cloud
  region: Option<String>,
  /// LanceDB Cloud host override, only required if using an on-premises Lance Cloud instance
  host_override: Option<String>,

  storage_options: HashMap<String, String>,

  /// The interval at which to check for updates from other processes.
  ///
  /// If None, then consistency is not checked. For performance
  /// reasons, this is the default. For strong consistency, set this to
  /// zero seconds. Then every read will check for updates from other
  /// processes. As a compromise, you can set this to a non-zero timedelta
  /// for eventual consistency. If more than that interval has passed since
  /// the last check, then the table will be checked for updates. Note: this
  /// consistency only applies to read operations. Write operations are
  /// always consistent.
  read_consistency_interval: Option<std::time::Duration>,
  embedding_registry: Option<Arc<dyn EmbeddingRegistry>>,
}

impl ConnectBuilder {
  /// Create a new [`ConnectOptions`] with the given database URI.
  pub fn new(uri: &str) -> Self {
    Self {
      uri: uri.to_string(),
      api_key: None,
      region: None,
      host_override: None,
      read_consistency_interval: None,
      storage_options: HashMap::new(),
      embedding_registry: None,
    }
  }

  // ......

  /// Establishes a connection to the database
  pub async fn execute(self) -> Result<Connection> {
    if self.uri.starts_with("db") {
      self.execute_remote()
    } else {
      let internal = Arc::new(Database::connect_with_options(&self).await?);
      Ok(Connection {
           internal,
           uri: self.uri,
      })
    }
  }
}

note

ConnectBuilder是一个结构体，用于配置和建立与LanceDB数据库的连接

作用:

ConnectBuilder通过存储连接参数(如URI、API秘钥、区域等)来构建数据库连接。

它提供了一些方法来设置这些参数，并通过execute方法建立实际的数据库连接。

主要字段和方法:

uri: String: 数据库的URI
api_key: Option<String>: LanceDB Cloud的API秘钥
region: Option<String>: LanceDB Cloud的区域
storage_options: HashMap<String, String>: 存储选项
read_consistency_interval: Option<std::time::Duration>: 读一致性检查间隔。
embedding_registry: Option<Arc<dyn EmbeddingRegistry>>: 嵌入注册表。

主要方法:

new(uri: &str) -> Self: 创建一个新的ConnectBuilder实例
execute(self) -> Result<Connection>: 执行连接建立，返回一个Connection实例

execute函数:

execute函数根据uri以及数据库连接选项，创建数据库连接
execute函数中首先判断uri是否是以db开头，如果是db开头，执行远程执行。
否则，使用Database设置连接选项，传入Connection对象，使用选项和uri创建Connection

`Database`

#[derive(Debug)]
struct Database {
  object_store: ObjectStore,
  query_string: Option<String>,

  pub(crate) uri: String,
  pub(crate) base_path: object_store::path::Path,

  // the object store wrapper to use on write path
  pub(crate) store_wrapper: Option<Arc<dyn WrappingObjectStore>>,

  read_consistency_interval: Option<std::time::Duration>,

  // Storage options to be inherited by tables created from this connection
  storage_options: HashMap<String, String>,
  embedding_registry: Arc<dyn EmbeddingRegistry>,
}

impl Database {
  async fn connect_with_options(options: &ConnectBuilder) -> Result<Self> {
    let uri = &options.uri;
    let parse_res = url::Url::parse(uri);

    // TODO: pass params regardless of OS
    match parse_res {
      Ok(url) if url.scheme().len() == 1 && cfg!(windows) => {
        Self::open_path(
          uri,
          options.read_consistency_interval,
          options.embedding_registry.clone(),
        )
       .await
     }
     Ok(mut url) => {
       // iter thru the query params and extract the commit store param
       let mut engine = None;
       let mut mirrored_store = None;
       let mut filtered_querys = vec![];

       // WARNING: specifying engine is NOT a publicly supported feature in lancedb yet
       // THE API WILL CHANGE
       for (key, value) in url.query_pairs() {
         if key == ENGINE {
           engine = Some(value.to_string());
         } else if key == MIRRORED_STORE {
           if cfg!(windows) {
             return Err(Error::NotSupported {
               message: "mirrored store is not supported on windows".into(),
             });
           }
           mirrored_store = Some(value.to_string());
         } else {
           // to owned so we can modify the url
           filtered_querys.push((key.to_string(), value.to_string()));
         }
       }

       // Filter out the commit store query param -- it's a lancedb param
       url.query_pairs_mut().clear();
       url.query_pairs_mut().extend_pairs(filtered_querys);
       // Take a copy of the query string so we can propagate it to lance
       let query_string = url.query().map(|s| s.to_string());
       // clear the query string so we can use the url as the base uri
       // use .set_query(None) instead of .set_query("") because the latter
       // will add a trailing '?' to the url
       url.set_query(None);

       let table_base_uri = if let Some(store) = engine {
         static WARN_ONCE: std::sync::Once = std::sync::Once::new();
         WARN_ONCE.call_once(|| {
           log::warn!("Specifing engine is not a publicly supported feature in lancedb yet. THE API WILL CHANGE");
         });
         let old_scheme = url.scheme().to_string();
         let new_scheme = format!("{}+{}", old_scheme, store);
         url.to_string().replacen(&old_scheme, &new_scheme, 1)
       } else {
         url.to_string()
       };

       let plain_uri = url.to_string();

       let storage_options = options.storage_options.clone();
       let os_params = ObjectStoreParams {
         storage_options: Some(storage_options.clone()),
         ..Default::default()
       };
       let (object_store, base_path) =
         ObjectStore::from_uri_and_params(&plain_uri, &os_params).await?;
       if object_store.is_local() {
         Self::try_create_dir(&plain_uri).context(CreateDirSnafu { path: plain_uri })?;
       }

       let write_store_wrapper = match mirrored_store {
         Some(path) => {
           let mirrored_store = Arc::new(LocalFileSystem::new_with_prefix(path)?);
           let wrapper = MirroringObjectStoreWrapper::new(mirrored_store);
           Some(Arc::new(wrapper) as Arc<dyn WrappingObjectStore>)
         }
         None => None,
       };

       let embedding_registry = options
           .embedding_registry
           .clone()
           .unwrap_or_else(|| Arc::new(MemoryRegistry::new()));
       Ok(Self {
         uri: table_base_uri,
         query_string,
         base_path,
         object_store,
         store_wrapper: write_store_wrapper,
         read_consistency_interval: options.read_consistency_interval,
         storage_options,
         embedding_registry,
       })
      }
      Err(_) => {
        Self::open_path(
          uri,
          options.read_consistency_interval,
          options.embedding_registry.clone(),
        )
        .await
      }
    }
  }
}

note

Database是一个结构体，表示LanceDB数据库实例

作用:

Database结构体封装了与实际数据库交互的逻辑。它负责管理数据库的基本路径、对象存储、嵌入注册表等信息，并提供方法来连接和操作数据库。

主要字段:

object_store: ObjectStore: 数据库的对象存储。
query_string: Option<String>: 查询字符串。
uri: String: 数据库的URI。
base_path: object_store::path::Path: 基础路径。
store_wrapper: Option<Arc<dyn WrappingObjectStore>>: 存储包装器。
read_consistency_interval: Option<std::time::Duration>: 读一致性检查间隔。
storage_options: HashMap<String, String>: 存储选项。
embedding_registry: Arc<dyn EmbeddingRegistry>: 嵌入注册表。

主要方法:

connect_with_options(options: &ConnectBuilder) -> Result<Self>: 根据ConnectBuilder配置建立数据库连接。

`Connection`

/// A connection to LanceDB
#[derive(Clone)]
pub struct Connection {
  uri: String,
  internal: Arc<dyn ConnectionInternal>,
}

note

Connection是一个结构体，表示与LanceDB数据库的连接实例

作用:

Connection结构体持有数据库连接的相关信息，并提供与数据库交互的接口。它是建立在ConnectBuilder配置基础上的实际连接对象。

主要字段:

uri: String: 数据库的URI
internal: Arc<dyn ConnectionInternal>: 内部连接实现

总结

connect: 一个函数，用于创建ConnectBuilder实例，并初始化数据库连接设置。
ConnectBuilder: 一个结构体，负责配置和建立数据库连接。它存储连接参数，并提供execute方法来建立实际的数据库连接。
Connection: 一个结构体，表示与LanceDB数据库的连接实例，持有连接信息并提供与数据库交互的接口。
Database: 一个结构体，表示LanceDB数据库实例，封装了与数据库交互的逻辑和相关信息。

链接

Github Repo: https://github.com/yuxuetr/lancedb-examples/tree/main/get_started

关于LanceDB中Table介绍​

Table的方法​

Schema模式定义Table定义方式​

创建空表​

创建带初始化数据的表​

初始化记录并创建表​

打开已存在的表​

删除表记录​

删除表​

删除数据库​

查询表记录​

更新表记录​

总结​

链接​

环境搭建​

Rust安装​

创建Rust项目​

安装Tokio​

安装LanceDB​

代码解析​

connect函数的定义如下​

ConnectBuilder​

Database​

Connection​

总结​

链接​

关于LanceDB中`Table`介绍

`Table`的方法

`Schema`模式定义`Table`定义方式

创建空表

创建带初始化数据的表

初始化记录并创建表

打开已存在的表

删除表记录

删除表

删除数据库

查询表记录

更新表记录

总结

链接

环境搭建

Rust安装

创建Rust项目

安装Tokio

安装LanceDB

代码解析

`connect`函数的定义如下

`ConnectBuilder`

`Database`

`Connection`

总结

链接