## Problem
`on_bad_vectors="drop"` is supposed to remove invalid vector rows before
write, but for some schema-defined vector columns it can still fail
later during Arrow cast instead of dropping the bad row.
Repro:
```python
class MySchema(LanceModel):
text: str
embedding: Vector(16)
table = db.create_table("test", schema=MySchema)
table.add(
[
{"text": "hello", "embedding": []},
{"text": "bar", "embedding": [0.1] * 16},
],
on_bad_vectors="drop",
)
```
Before:
```
RuntimeError
Arrow error: C Data interface error: Invalid: ListType can only be casted to FixedSizeListType if the lists are all the expected size.
```
After:
```
rows 1
texts ['bar']
```
## Solution
Make bad-vector sanitization use schema dimensions before cast, while
keeping the handling scoped to vector columns identified by schema
metadata or existing vector-name heuristics.
This also preserves existing integer vector inputs and avoids applying
on_bad_vectors to unrelated fixed-size float columns.
Fixes#1670
Signed-off-by: yaommen <myanstu@163.com>
Fixes#1540
I could not reproduce this on current `main` from Python, but I could
still reproduce it from the Rust SDK.
Python no longer reproduces because the current Python vector/hybrid
query paths re-chunk results into a `pyarrow.Table` before returning
batches. Rust still reproduced because `max_batch_length` was passed
into planning/scanning, but vector search could still emit larger
`RecordBatch`es later in execution (for example after KNN / TopK), so it
was not enforced on the final Rust output stream.
This PR enforces `max_batch_length` on the final Rust query output
stream and adds Rust regression coverage.
Before the fix, the Rust repro produced:
`num_batches=2, max_batch=8192, min_batch=1808, all_le_100=false`
After the fix, the same repro produces batches `<= 100`.
## Runnable Rust repro
Before this fix, current `main` could still return batches like `[8192,
1808]` here even with `max_batch_length = 100`:
```rust
use std::sync::Arc;
use arrow_array::{
types::Float32Type, FixedSizeListArray, RecordBatch, RecordBatchReader, StringArray,
};
use arrow_schema::{DataType, Field, Schema};
use futures::TryStreamExt;
use lancedb::query::{ExecutableQuery, QueryBase, QueryExecutionOptions};
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let tmp = tempfile::tempdir()?;
let uri = tmp.path().to_str().unwrap();
let rows = 10_000;
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
Field::new(
"vector",
DataType::FixedSizeList(Arc::new(Field::new("item", DataType::Float32, true)), 4),
false,
),
]));
let ids = StringArray::from_iter_values((0..rows).map(|i| format!("row-{i}")));
let vectors = FixedSizeListArray::from_iter_primitive::<Float32Type, _, _>(
(0..rows).map(|i| Some(vec![Some(i as f32), Some(1.0), Some(2.0), Some(3.0)])),
4,
);
let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(ids), Arc::new(vectors)])?;
let reader: Box<dyn RecordBatchReader + Send> = Box::new(
arrow_array::RecordBatchIterator::new(vec![Ok(batch)].into_iter(), schema),
);
let db = lancedb::connect(uri).execute().await?;
let table = db.create_table("test", reader).execute().await?;
let mut opts = QueryExecutionOptions::default();
opts.max_batch_length = 100;
let mut stream = table
.query()
.nearest_to(vec![0.0, 1.0, 2.0, 3.0])?
.limit(rows)
.execute_with_options(opts)
.await?;
let mut sizes = Vec::new();
while let Some(batch) = stream.try_next().await? {
sizes.push(batch.num_rows());
}
println!("{sizes:?}");
Ok(())
}
```
Signed-off-by: yaommen <myanstu@163.com>
yaommen