mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-03 20:18:54 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 238028767a | |||
| 7357d63e87 | |||
| 624a75edf7 |
@@ -541,7 +541,17 @@ where(predicate): this
|
||||
|
||||
A filter statement to be applied to this query.
|
||||
|
||||
The filter should be supplied as an SQL query string. For example:
|
||||
Filters are applied before full-text and vector searches by default; no
|
||||
separate prefilter call is needed. For vector searches only, use
|
||||
[VectorQuery#postfilter](VectorQuery.md#postfilter) to apply the filter after the search.
|
||||
|
||||
The filter should be supplied as an SQL query string.
|
||||
|
||||
Filtering performance can often be improved by creating a scalar index
|
||||
on the filter column(s).
|
||||
|
||||
Calling this multiple times combines the filters with a logical AND rather
|
||||
than replacing the previous filter.
|
||||
|
||||
#### Parameters
|
||||
|
||||
@@ -551,18 +561,20 @@ The filter should be supplied as an SQL query string. For example:
|
||||
|
||||
`this`
|
||||
|
||||
#### Example
|
||||
#### Examples
|
||||
|
||||
```ts
|
||||
x > 10
|
||||
y > 0 AND y < 100
|
||||
x > 5 OR y = 'test'
|
||||
const results = await table
|
||||
.search("puppy", "fts")
|
||||
.where("meta = 'foo'")
|
||||
.limit(10)
|
||||
.toArray();
|
||||
```
|
||||
|
||||
Filtering performance can often be improved by creating a scalar index
|
||||
on the filter column(s).
|
||||
|
||||
Calling this multiple times combines the filters with a logical AND rather
|
||||
than replacing the previous filter.
|
||||
```ts
|
||||
query.where("x > 10");
|
||||
query.where("y > 0 AND y < 100");
|
||||
query.where("x > 5 OR y = 'test'");
|
||||
```
|
||||
|
||||
#### Inherited from
|
||||
|
||||
@@ -560,6 +560,9 @@ postfilter(): VectorQuery
|
||||
If this is called then filtering will happen after the vector search instead of
|
||||
before.
|
||||
|
||||
This method is only available for vector search queries. Full-text search
|
||||
filters are always applied before the search.
|
||||
|
||||
By default filtering will be performed before the vector search. This is how
|
||||
filtering is typically understood to work. This prefilter step does add some
|
||||
additional latency. Creating a scalar index on the filter column(s) can
|
||||
@@ -790,7 +793,17 @@ where(predicate): this
|
||||
|
||||
A filter statement to be applied to this query.
|
||||
|
||||
The filter should be supplied as an SQL query string. For example:
|
||||
Filters are applied before full-text and vector searches by default; no
|
||||
separate prefilter call is needed. For vector searches only, use
|
||||
[VectorQuery#postfilter](VectorQuery.md#postfilter) to apply the filter after the search.
|
||||
|
||||
The filter should be supplied as an SQL query string.
|
||||
|
||||
Filtering performance can often be improved by creating a scalar index
|
||||
on the filter column(s).
|
||||
|
||||
Calling this multiple times combines the filters with a logical AND rather
|
||||
than replacing the previous filter.
|
||||
|
||||
#### Parameters
|
||||
|
||||
@@ -800,18 +813,20 @@ The filter should be supplied as an SQL query string. For example:
|
||||
|
||||
`this`
|
||||
|
||||
#### Example
|
||||
#### Examples
|
||||
|
||||
```ts
|
||||
x > 10
|
||||
y > 0 AND y < 100
|
||||
x > 5 OR y = 'test'
|
||||
const results = await table
|
||||
.search("puppy", "fts")
|
||||
.where("meta = 'foo'")
|
||||
.limit(10)
|
||||
.toArray();
|
||||
```
|
||||
|
||||
Filtering performance can often be improved by creating a scalar index
|
||||
on the filter column(s).
|
||||
|
||||
Calling this multiple times combines the filters with a logical AND rather
|
||||
than replacing the previous filter.
|
||||
```ts
|
||||
query.where("x > 10");
|
||||
query.where("y > 0 AND y < 100");
|
||||
query.where("x > 5 OR y = 'test'");
|
||||
```
|
||||
|
||||
#### Inherited from
|
||||
|
||||
@@ -47,5 +47,22 @@ test("filtering examples", async () => {
|
||||
.limit(5)
|
||||
.toArray();
|
||||
// --8<-- [end:orderby_search]
|
||||
|
||||
const ftsTable = await db.createTable("myFts", [
|
||||
{ text: "Frodo was a happy puppy", category: "pet" },
|
||||
{ text: "A puppy training guide", category: "guide" },
|
||||
{ text: "There are several kittens playing", category: "pet" },
|
||||
]);
|
||||
await ftsTable.createIndex("text", { config: lancedb.Index.fts() });
|
||||
|
||||
// --8<-- [start:fts_prefilter]
|
||||
const ftsResults = await ftsTable
|
||||
.search("puppy", "fts")
|
||||
.where("category = 'pet'")
|
||||
.limit(10)
|
||||
.toArray();
|
||||
// --8<-- [end:fts_prefilter]
|
||||
expect(ftsResults).toHaveLength(1);
|
||||
expect(ftsResults[0].category).toBe("pet");
|
||||
});
|
||||
});
|
||||
|
||||
+24
-5
@@ -363,17 +363,33 @@ export class StandardQueryBase<
|
||||
/**
|
||||
* A filter statement to be applied to this query.
|
||||
*
|
||||
* The filter should be supplied as an SQL query string. For example:
|
||||
* @example
|
||||
* x > 10
|
||||
* y > 0 AND y < 100
|
||||
* x > 5 OR y = 'test'
|
||||
* Filters are applied before full-text and vector searches by default; no
|
||||
* separate prefilter call is needed. For vector searches only, use
|
||||
* {@link VectorQuery#postfilter} to apply the filter after the search.
|
||||
*
|
||||
* The filter should be supplied as an SQL query string.
|
||||
*
|
||||
* Filtering performance can often be improved by creating a scalar index
|
||||
* on the filter column(s).
|
||||
*
|
||||
* Calling this multiple times combines the filters with a logical AND rather
|
||||
* than replacing the previous filter.
|
||||
*
|
||||
* @example Filter a full-text search
|
||||
* ```ts
|
||||
* const results = await table
|
||||
* .search("puppy", "fts")
|
||||
* .where("meta = 'foo'")
|
||||
* .limit(10)
|
||||
* .toArray();
|
||||
* ```
|
||||
*
|
||||
* @example SQL filter expressions
|
||||
* ```ts
|
||||
* query.where("x > 10");
|
||||
* query.where("y > 0 AND y < 100");
|
||||
* query.where("x > 5 OR y = 'test'");
|
||||
* ```
|
||||
*/
|
||||
where(predicate: string): this {
|
||||
this.doCall((inner: NativeQueryType) => inner.onlyIf(predicate));
|
||||
@@ -669,6 +685,9 @@ export class VectorQuery extends StandardQueryBase<NativeVectorQuery> {
|
||||
* If this is called then filtering will happen after the vector search instead of
|
||||
* before.
|
||||
*
|
||||
* This method is only available for vector search queries. Full-text search
|
||||
* filters are always applied before the search.
|
||||
*
|
||||
* By default filtering will be performed before the vector search. This is how
|
||||
* filtering is typically understood to work. This prefilter step does add some
|
||||
* additional latency. Creating a scalar index on the filter column(s) can
|
||||
|
||||
@@ -707,6 +707,9 @@ class LanceDBConnection(DBConnection):
|
||||
self._namespace_client_properties = namespace_client_properties
|
||||
if _inner is not None:
|
||||
self._conn = _inner
|
||||
# Native-derived wrappers resolve this in their async reconstruction
|
||||
# path so construction never synchronously re-enters LOOP.
|
||||
self._read_consistency_interval = read_consistency_interval
|
||||
self._cached_namespace_client = None
|
||||
return
|
||||
|
||||
@@ -756,11 +759,14 @@ class LanceDBConnection(DBConnection):
|
||||
# storage_options. Also, this class really shouldn't be holding any state
|
||||
# beyond _conn.
|
||||
self._conn = AsyncConnection(LOOP.run(do_connect()))
|
||||
# Keep property access synchronous so debugger introspection cannot wait on
|
||||
# the background loop while that thread is suspended at a breakpoint.
|
||||
self._read_consistency_interval = read_consistency_interval
|
||||
self._cached_namespace_client: Optional[LanceNamespace] = None
|
||||
|
||||
@property
|
||||
def read_consistency_interval(self) -> Optional[timedelta]:
|
||||
return LOOP.run(self._conn.get_read_consistency_interval())
|
||||
return self._read_consistency_interval
|
||||
|
||||
@property
|
||||
def session(self) -> Optional[Session]:
|
||||
@@ -771,8 +777,16 @@ class LanceDBConnection(DBConnection):
|
||||
return self._conn.uri
|
||||
|
||||
@classmethod
|
||||
def from_inner(cls, inner: LanceDbConnection):
|
||||
return cls(None, _inner=inner)
|
||||
def from_inner(
|
||||
cls,
|
||||
inner: LanceDbConnection,
|
||||
read_consistency_interval: Optional[timedelta],
|
||||
):
|
||||
return cls(
|
||||
None,
|
||||
read_consistency_interval=read_consistency_interval,
|
||||
_inner=inner,
|
||||
)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.__class__.__name__}(uri={self._conn.uri!r})"
|
||||
|
||||
@@ -226,7 +226,7 @@ class PermutationBuilder:
|
||||
|
||||
async def do_execute():
|
||||
inner_tbl = await self._async.execute()
|
||||
return LanceTable.from_inner(inner_tbl)
|
||||
return await LanceTable.from_inner(inner_tbl)
|
||||
|
||||
return LOOP.run(do_execute())
|
||||
|
||||
|
||||
@@ -2182,11 +2182,15 @@ class LanceTable(Table):
|
||||
return self.name
|
||||
|
||||
@classmethod
|
||||
def from_inner(cls, tbl: LanceDBTable):
|
||||
from .db import LanceDBConnection
|
||||
async def from_inner(cls, tbl: LanceDBTable):
|
||||
from .db import AsyncConnection, LanceDBConnection
|
||||
|
||||
async_tbl = AsyncTable(tbl)
|
||||
conn = LanceDBConnection.from_inner(tbl.database())
|
||||
inner_conn = tbl.database()
|
||||
read_consistency_interval = await AsyncConnection(
|
||||
inner_conn
|
||||
).get_read_consistency_interval()
|
||||
conn = LanceDBConnection.from_inner(inner_conn, read_consistency_interval)
|
||||
return cls(
|
||||
conn,
|
||||
async_tbl.name,
|
||||
|
||||
@@ -77,6 +77,23 @@ def test_sync_repr_does_not_use_background_loop(tmp_path, monkeypatch):
|
||||
assert repr(table) == f"LanceTable(name='test', _conn={db!r})"
|
||||
|
||||
|
||||
def test_read_consistency_interval_does_not_use_background_loop(tmp_path, monkeypatch):
|
||||
from lancedb.background_loop import LOOP
|
||||
from lancedb.db import LanceDBConnection
|
||||
|
||||
consistency_interval = timedelta(seconds=5)
|
||||
db = lancedb.connect(tmp_path, read_consistency_interval=consistency_interval)
|
||||
db_from_inner = LanceDBConnection.from_inner(db._inner, consistency_interval)
|
||||
|
||||
def fail_run(*args, **kwargs):
|
||||
raise AssertionError("properties should not use the Python background loop")
|
||||
|
||||
monkeypatch.setattr(LOOP, "run", fail_run)
|
||||
|
||||
assert db.read_consistency_interval == consistency_interval
|
||||
assert db_from_inner.read_consistency_interval == consistency_interval
|
||||
|
||||
|
||||
def test_ingest_pd(tmp_path):
|
||||
db = lancedb.connect(tmp_path)
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ import math
|
||||
import pytest
|
||||
|
||||
from lancedb import DBConnection, Table, connect
|
||||
from lancedb.background_loop import LOOP
|
||||
from lancedb.permutation import Permutation, Permutations, permutation_builder
|
||||
|
||||
|
||||
@@ -31,6 +32,25 @@ def test_split_random_ratios(mem_db):
|
||||
assert 65 <= split_1_count <= 75 # ~70% ± tolerance
|
||||
|
||||
|
||||
def test_execute_does_not_reenter_background_loop(tmp_path, monkeypatch):
|
||||
import threading
|
||||
|
||||
db = connect(tmp_path)
|
||||
tbl = db.create_table("test_table", pa.table({"x": range(10)}))
|
||||
original_run = LOOP.run
|
||||
|
||||
def fail_on_reentry(future):
|
||||
assert threading.current_thread() is not LOOP.thread
|
||||
return original_run(future)
|
||||
|
||||
monkeypatch.setattr(LOOP, "run", fail_on_reentry)
|
||||
|
||||
permutation_tbl = permutation_builder(tbl).execute()
|
||||
|
||||
assert permutation_tbl.count_rows() == 10
|
||||
assert permutation_tbl._conn.read_consistency_interval is None
|
||||
|
||||
|
||||
def test_split_random_counts(mem_db):
|
||||
"""Test random splitting with absolute counts."""
|
||||
tbl = mem_db.create_table(
|
||||
|
||||
@@ -6,6 +6,7 @@ import os
|
||||
import sys
|
||||
import threading
|
||||
import warnings
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import date, datetime, timedelta
|
||||
from time import sleep
|
||||
from typing import List
|
||||
@@ -2124,6 +2125,27 @@ def test_delete(mem_db: DBConnection):
|
||||
assert table.to_arrow()["id"].to_pylist() == [1]
|
||||
|
||||
|
||||
def test_concurrent_deletes_are_thread_safe(mem_db: DBConnection):
|
||||
num_workers = 8
|
||||
table = mem_db.create_table(
|
||||
"my_table", data=[{"id": row_id} for row_id in range(num_workers)]
|
||||
)
|
||||
barrier = threading.Barrier(num_workers)
|
||||
|
||||
def delete(row_id: int):
|
||||
barrier.wait()
|
||||
return table.delete(f"id = {row_id}")
|
||||
|
||||
with ThreadPoolExecutor(max_workers=num_workers) as pool:
|
||||
results = list(pool.map(delete, range(num_workers)))
|
||||
|
||||
assert all(result.num_deleted_rows == 1 for result in results)
|
||||
assert sorted(result.version for result in results) == list(
|
||||
range(2, num_workers + 2)
|
||||
)
|
||||
assert table.count_rows() == 0
|
||||
|
||||
|
||||
def test_delete_expr(mem_db: DBConnection):
|
||||
table = mem_db.create_table(
|
||||
"my_table",
|
||||
|
||||
@@ -745,6 +745,9 @@ impl Table {
|
||||
|
||||
#[allow(private_interfaces)]
|
||||
pub fn delete(self_: PyRef<'_, Self>, condition: PredicateArg) -> PyResult<Bound<'_, PyAny>> {
|
||||
// Do not hold the Python borrow across the await. The cloned Rust table
|
||||
// handle is thread-safe and allows deletes on the same Python table to
|
||||
// run concurrently without PyO3 reporting "Already borrowed".
|
||||
let inner = self_.inner_ref()?.clone();
|
||||
future_into_py(self_.py(), async move {
|
||||
let result = match &condition {
|
||||
|
||||
@@ -420,11 +420,6 @@ impl Connection {
|
||||
///
|
||||
/// * `name` - The name of the table
|
||||
/// * `initial_data` - The initial data to write to the table
|
||||
///
|
||||
/// Floating-point `List` columns named `vec`, or with `vector` or `embedding`
|
||||
/// in their name, are inferred as vector columns when the first batch has a
|
||||
/// uniform, non-zero list length. The inferred dimension is validated for all
|
||||
/// subsequent batches and stored as a `FixedSizeList`.
|
||||
pub fn create_table<T: Scannable + 'static>(
|
||||
&self,
|
||||
name: impl Into<String>,
|
||||
|
||||
@@ -8,7 +8,7 @@ use lance_io::object_store::StorageOptionsProvider;
|
||||
use crate::{
|
||||
Error, Result, Table,
|
||||
connection::{merge_storage_options, set_storage_options_provider},
|
||||
data::scannable::{Scannable, WithEmbeddingsScannable, maybe_infer_vector_schema},
|
||||
data::scannable::{Scannable, WithEmbeddingsScannable},
|
||||
database::{CreateTableMode, CreateTableRequest, Database},
|
||||
embeddings::{EmbeddingDefinition, EmbeddingFunction, EmbeddingRegistry},
|
||||
table::WriteOptions,
|
||||
@@ -147,8 +147,6 @@ impl CreateTableBuilder {
|
||||
let embedding_registry = self.embedding_registry.clone();
|
||||
let parent = self.parent.clone();
|
||||
|
||||
self.request.data = maybe_infer_vector_schema(self.request.data).await?;
|
||||
|
||||
// If embeddings were configured via add_embedding(), wrap the data
|
||||
if !self.embeddings.is_empty() {
|
||||
let wrapped_data: Box<dyn Scannable> = Box::new(WithEmbeddingsScannable::try_new(
|
||||
|
||||
@@ -18,16 +18,13 @@ use crate::embeddings::{
|
||||
};
|
||||
use crate::table::{ColumnDefinition, ColumnKind, TableDefinition};
|
||||
use crate::{Error, Result};
|
||||
use arrow_array::{ArrayRef, RecordBatch, RecordBatchIterator, RecordBatchReader, cast::AsArray};
|
||||
use arrow_cast::{CastOptions, cast_with_options};
|
||||
use arrow_schema::{ArrowError, DataType, Schema, SchemaRef};
|
||||
use arrow_array::{ArrayRef, RecordBatch, RecordBatchIterator, RecordBatchReader};
|
||||
use arrow_schema::{ArrowError, SchemaRef};
|
||||
use async_trait::async_trait;
|
||||
use futures::StreamExt;
|
||||
use futures::stream::once;
|
||||
use lance_datafusion::utils::StreamingWriteSource;
|
||||
|
||||
use super::inspect::infer_dimension;
|
||||
|
||||
pub trait Scannable: Send {
|
||||
/// Returns the schema of the data.
|
||||
fn schema(&self) -> SchemaRef;
|
||||
@@ -500,146 +497,6 @@ impl Scannable for PeekedScannable {
|
||||
}
|
||||
}
|
||||
|
||||
fn name_suggests_vector_column(name: &str) -> bool {
|
||||
let name = name.to_ascii_lowercase();
|
||||
name == "vec" || name.contains("vector") || name.contains("embedding")
|
||||
}
|
||||
|
||||
/// Infer fixed dimensions for vector-like floating-point list columns.
|
||||
///
|
||||
/// Lance vector search requires `FixedSizeList` columns, but Arrow data assembled
|
||||
/// from runtime embedding models is often represented as `List`. For vector-like
|
||||
/// column names, inspect the first batch and convert uniform, non-empty lists to a
|
||||
/// fixed-size schema. Every subsequent batch is cast with strict length checking.
|
||||
pub(crate) async fn maybe_infer_vector_schema(
|
||||
data: Box<dyn Scannable>,
|
||||
) -> Result<Box<dyn Scannable>> {
|
||||
let input_schema = data.schema();
|
||||
let candidates = input_schema
|
||||
.fields()
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter(|(_, field)| {
|
||||
name_suggests_vector_column(field.name())
|
||||
&& matches!(
|
||||
field.data_type(),
|
||||
DataType::List(item) | DataType::LargeList(item)
|
||||
if item.data_type().is_floating()
|
||||
)
|
||||
})
|
||||
.map(|(index, _)| index)
|
||||
.collect::<Vec<_>>();
|
||||
if candidates.is_empty() {
|
||||
return Ok(data);
|
||||
}
|
||||
|
||||
let mut peeked = PeekedScannable::new(data);
|
||||
let Some(first_batch) = peeked.peek().await else {
|
||||
return Ok(Box::new(peeked));
|
||||
};
|
||||
|
||||
let mut fields = input_schema.fields().iter().cloned().collect::<Vec<_>>();
|
||||
let mut changed = false;
|
||||
for index in candidates {
|
||||
let array = first_batch.column(index);
|
||||
let dimension = match array.data_type() {
|
||||
DataType::List(_) => {
|
||||
infer_dimension::<arrow_array::types::Int32Type>(array.as_list::<i32>())?
|
||||
.map(i64::from)
|
||||
}
|
||||
DataType::LargeList(_) => {
|
||||
infer_dimension::<arrow_array::types::Int64Type>(array.as_list::<i64>())?
|
||||
}
|
||||
_ => unreachable!(),
|
||||
};
|
||||
let Some(dimension) = dimension.filter(|dimension| *dimension > 0) else {
|
||||
continue;
|
||||
};
|
||||
let dimension = i32::try_from(dimension).map_err(|_| Error::InvalidInput {
|
||||
message: format!(
|
||||
"Vector column '{}' has a dimension larger than i32::MAX",
|
||||
fields[index].name()
|
||||
),
|
||||
})?;
|
||||
let item = match fields[index].data_type() {
|
||||
DataType::List(item) | DataType::LargeList(item) => item.clone(),
|
||||
_ => unreachable!(),
|
||||
};
|
||||
fields[index] = Arc::new(
|
||||
fields[index]
|
||||
.as_ref()
|
||||
.clone()
|
||||
.with_data_type(DataType::FixedSizeList(item, dimension)),
|
||||
);
|
||||
changed = true;
|
||||
}
|
||||
|
||||
if !changed {
|
||||
return Ok(Box::new(peeked));
|
||||
}
|
||||
|
||||
let output_schema = Arc::new(Schema::new_with_metadata(
|
||||
fields,
|
||||
input_schema.metadata().clone(),
|
||||
));
|
||||
Ok(Box::new(InferredVectorScannable {
|
||||
inner: peeked,
|
||||
output_schema,
|
||||
}))
|
||||
}
|
||||
|
||||
struct InferredVectorScannable {
|
||||
inner: PeekedScannable,
|
||||
output_schema: SchemaRef,
|
||||
}
|
||||
|
||||
impl Scannable for InferredVectorScannable {
|
||||
fn schema(&self) -> SchemaRef {
|
||||
self.output_schema.clone()
|
||||
}
|
||||
|
||||
fn scan_as_stream(&mut self) -> SendableRecordBatchStream {
|
||||
let output_schema = self.output_schema.clone();
|
||||
let stream_schema = output_schema.clone();
|
||||
let stream = self.inner.scan_as_stream().map(move |batch| {
|
||||
let batch = batch?;
|
||||
let columns = batch
|
||||
.columns()
|
||||
.iter()
|
||||
.zip(output_schema.fields())
|
||||
.map(|(array, field)| {
|
||||
if array.data_type() == field.data_type() {
|
||||
Ok(array.clone())
|
||||
} else {
|
||||
cast_with_options(
|
||||
array,
|
||||
field.data_type(),
|
||||
&CastOptions {
|
||||
safe: false,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.map_err(Error::from)
|
||||
}
|
||||
})
|
||||
.collect::<Result<Vec<_>>>()?;
|
||||
Ok(RecordBatch::try_new(output_schema.clone(), columns)?)
|
||||
});
|
||||
Box::pin(SimpleRecordBatchStream {
|
||||
schema: stream_schema,
|
||||
stream,
|
||||
})
|
||||
}
|
||||
|
||||
fn num_rows(&self) -> Option<usize> {
|
||||
self.inner.num_rows()
|
||||
}
|
||||
|
||||
fn rescannable(&self) -> bool {
|
||||
self.inner.rescannable()
|
||||
}
|
||||
}
|
||||
|
||||
/// Compute the number of write partitions based on data size estimates.
|
||||
///
|
||||
/// `sample_bytes` and `sample_rows` come from a representative batch and are
|
||||
|
||||
@@ -72,9 +72,7 @@
|
||||
//!
|
||||
//! LanceDB uses [arrow-rs](https://github.com/apache/arrow-rs) to define schema, data types and array itself.
|
||||
//! It treats [`FixedSizeList<Float16/Float32>`](https://docs.rs/arrow/latest/arrow/array/struct.FixedSizeListArray.html)
|
||||
//! columns as vector columns. When creating a table with a floating-point `List`
|
||||
//! column named `vec`, or with `vector` or `embedding` in its name, LanceDB infers
|
||||
//! a uniform dimension from the first batch and stores it as a `FixedSizeList`.
|
||||
//! columns as vector columns.
|
||||
//!
|
||||
//! For more details, please refer to the [LanceDB documentation](https://docs.lancedb.com).
|
||||
//!
|
||||
@@ -84,8 +82,6 @@
|
||||
//! schema of the `RecordBatch` determines the schema of the table.
|
||||
//!
|
||||
//! Vector columns should be represented as `FixedSizeList<Float16/Float32>` data type.
|
||||
//! A vector-like `List<Float16/Float32>` input is also accepted when every vector
|
||||
//! has the same runtime dimension.
|
||||
//!
|
||||
//! ```rust
|
||||
//! # use std::sync::Arc;
|
||||
|
||||
@@ -1648,8 +1648,8 @@ mod tests {
|
||||
use super::*;
|
||||
use arrow::{array::downcast_array, compute::concat_batches, datatypes::Int32Type};
|
||||
use arrow_array::{
|
||||
FixedSizeListArray, Float32Array, Int32Array, ListArray, RecordBatch, StringArray,
|
||||
cast::AsArray, types::Float32Type,
|
||||
FixedSizeListArray, Float32Array, Int32Array, RecordBatch, StringArray, cast::AsArray,
|
||||
types::Float32Type,
|
||||
};
|
||||
use arrow_schema::{DataType, Field as ArrowField, Schema as ArrowSchema};
|
||||
use futures::{StreamExt, TryStreamExt};
|
||||
@@ -2282,91 +2282,6 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vector_search_infers_dimension_from_list_array() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let schema = Arc::new(ArrowSchema::new(vec![
|
||||
ArrowField::new("id", DataType::Int32, false),
|
||||
ArrowField::new(
|
||||
"vec",
|
||||
DataType::List(Arc::new(ArrowField::new("item", DataType::Float32, true))),
|
||||
true,
|
||||
),
|
||||
]));
|
||||
let vectors = ListArray::from_iter_primitive::<Float32Type, _, _>([
|
||||
Some([Some(0.0), Some(0.0)]),
|
||||
Some([Some(1.0), Some(1.0)]),
|
||||
]);
|
||||
let batch = RecordBatch::try_new(
|
||||
schema,
|
||||
vec![Arc::new(Int32Array::from(vec![0, 1])), Arc::new(vectors)],
|
||||
)
|
||||
.unwrap();
|
||||
let table = connect(tmp_dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.create_table("vectors", batch)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(matches!(
|
||||
table.schema().await.unwrap().field(1).data_type(),
|
||||
DataType::FixedSizeList(_, 2)
|
||||
));
|
||||
|
||||
let results = table
|
||||
.vector_search(&[0.0, 0.0])
|
||||
.unwrap()
|
||||
.limit(1)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.try_collect::<Vec<_>>()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(results[0]["id"].as_primitive::<Int32Type>().value(0), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn inferred_vector_dimension_is_validated_across_batches() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
let schema = Arc::new(ArrowSchema::new(vec![ArrowField::new(
|
||||
"vec",
|
||||
DataType::List(Arc::new(ArrowField::new("item", DataType::Float32, true))),
|
||||
true,
|
||||
)]));
|
||||
let first = RecordBatch::try_new(
|
||||
schema.clone(),
|
||||
vec![Arc::new(
|
||||
ListArray::from_iter_primitive::<Float32Type, _, _>([Some([Some(0.0), Some(0.0)])]),
|
||||
)],
|
||||
)
|
||||
.unwrap();
|
||||
let wrong_dimension = RecordBatch::try_new(
|
||||
schema,
|
||||
vec![Arc::new(
|
||||
ListArray::from_iter_primitive::<Float32Type, _, _>([Some([
|
||||
Some(1.0),
|
||||
Some(1.0),
|
||||
Some(1.0),
|
||||
])]),
|
||||
)],
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let result = connect(tmp_dir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.create_table("vectors", vec![first, wrong_dimension])
|
||||
.execute()
|
||||
.await;
|
||||
|
||||
assert!(result.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_fast_search_plan() {
|
||||
let tmp_dir = tempdir().unwrap();
|
||||
|
||||
Reference in New Issue
Block a user