feat(sqlite-provider): implement streaming full-table scan for adaptive filtering#4
Conversation
…ve filtering SqliteLookupProvider.scan() previously returned NotImplemented, causing the adaptive filtered path in USearchExec to fail when a WHERE clause was combined with vector search. Add SqliteFullScanExec: a leaf ExecutionPlan that streams all rows from the SQLite table in 1024-row batches via a bounded tokio mpsc channel. The blocking SQLite cursor runs in spawn_blocking; the async consumer processes each batch through evaluate_filters() and drops it immediately, keeping peak memory at O(batch_size) rather than O(total_rows). The semaphore and connection pool are shared with fetch_by_keys so concurrent scans and key lookups stay within the configured pool size.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| if rows_in_batch > 0 { | ||
| if let Ok(batch) = build_scan_batch(&schema_task, col_bufs) { | ||
| let _ = tx_c.blocking_send(Ok(batch)); | ||
| } | ||
| } |
There was a problem hiding this comment.
P1 — final-batch error is silently swallowed
If build_scan_batch returns an Err here, it's dropped and the stream terminates cleanly — the consumer receives a successful (but truncated) result with no indication that the last partial batch failed.
| if rows_in_batch > 0 { | |
| if let Ok(batch) = build_scan_batch(&schema_task, col_bufs) { | |
| let _ = tx_c.blocking_send(Ok(batch)); | |
| } | |
| } | |
| if rows_in_batch > 0 { | |
| match build_scan_batch(&schema_task, col_bufs) { | |
| Ok(batch) => { let _ = tx_c.blocking_send(Ok(batch)); } | |
| Err(e) => { let _ = tx_c.blocking_send(Err(e)); } | |
| } | |
| } |
|
|
||
| let pool_c = pool.clone(); | ||
| let tx_c = tx.clone(); | ||
| let _ = tokio::task::spawn_blocking(move || { |
There was a problem hiding this comment.
P1 — spawn_blocking panic is silently discarded
let _ = spawn_blocking(...).await drops the Result<T, JoinError>. If the blocking task panics, the JoinError is thrown away, tx_c is dropped (the sender closes), and the consumer sees a clean end-of-stream with no error. A partial scan looks like a successful complete scan.
Propagate the join error:
| let _ = tokio::task::spawn_blocking(move || { | |
| if let Err(e) = tokio::task::spawn_blocking(move || { |
and at the closing .await:
}).await {
let _ = tx.send(Err(DataFusionError::Execution(format!("scan task panicked: {e}")))).await;
}
Address PR review comments: - Send build_scan_batch errors for the last partial batch instead of silently dropping them (truncated scan looked like success) - Propagate spawn_blocking JoinError so panics surface as stream errors instead of a clean end-of-stream
There was a problem hiding this comment.
Review
Issues
P1 — Existing test broken (tests/sqlite_provider_test.rs:147–160)
test_scan_returns_not_implemented asserts that scan() returns an Err containing "not support full table scans". After this PR, scan() returns Ok(...), so the test will fail. The test must be replaced with one that exercises the new streaming behavior (e.g., assert all rows are returned, assert batching works).
P1 — Connection pool error silently swallowed (src/sqlite_provider.rs:431–436)
When pool.lock() returns a poisoned mutex, the code converts the error to None and then reports "connection pool empty" instead of the real cause. fetch_by_keys propagates the poison error; this path should do the same.
Action Required
- Remove/replace
test_scan_returns_not_implemented— it asserts behavior that no longer exists and will fail CI. - Propagate the mutex-poison error in
execute()rather than silently converting it toNone.
| Ok(mut g) => g.pop(), | ||
| Err(_) => None, | ||
| } | ||
| }; |
There was a problem hiding this comment.
Swallowing the poison error and returning None means a poisoned mutex is reported as "connection pool empty", hiding the real cause. fetch_by_keys handles this correctly — propagate the error here too:
| }; | |
| let conn = match pool.lock() { | |
| Ok(mut g) => g.pop().ok_or_else(|| { | |
| DataFusionError::Execution("SqliteFullScanExec: connection pool empty".into()) | |
| }), | |
| Err(e) => Err(DataFusionError::Execution(format!( | |
| "connection pool mutex poisoned: {e}" | |
| ))), | |
| }; | |
| let conn = match conn { | |
| Ok(c) => c, |
| Ok(Some(row)) => { | ||
| let mut row_ok = true; | ||
| for (ci, buf) in col_bufs.iter_mut().enumerate() { | ||
| match row.get::<_, SqlValue>(ci) { | ||
| Ok(v) => buf.push(v), | ||
| Err(e) => { | ||
| let _ = tx_c.blocking_send(Err( | ||
| DataFusionError::Execution(e.to_string()), | ||
| )); | ||
| row_ok = false; | ||
| break; | ||
| } | ||
| } | ||
| } | ||
| if !row_ok { | ||
| break; | ||
| } |
There was a problem hiding this comment.
When a column read fails mid-row (say on column ci), columns 0..ci in col_bufs have one extra value that columns ci..n don't. The code correctly sends the error and breaks, but rows_in_batch still reflects the count of fully completed rows. If rows_in_batch > 0, the final flush at line 535 will call build_scan_batch with buffers of unequal length, causing RecordBatch::try_new to fail and a second error to be sent on the channel — masking the original one.
Fix: clear col_bufs (or just skip the flush) on the error path before breaking:
if !row_ok {
// Discard partial row data so the final flush doesn't see
// mismatched column buffer lengths.
for buf in col_bufs.iter_mut() {
buf.truncate(rows_in_batch);
}
break;
}
| Err(DataFusionError::NotImplemented( | ||
| "SqliteLookupProvider does not support full table scans; use fetch_by_keys".into(), | ||
| )) | ||
| Ok(Arc::new(SqliteFullScanExec::new( |
There was a problem hiding this comment.
The _projection parameter is ignored — the scan always fetches every column regardless of what DataFusion requests. For tables with many wide columns this wastes I/O and memory.\n\nConsider building a projected schema here (same pattern used in fetch_by_keys) and passing it to SqliteFullScanExec:\n\nrust\nlet schema = match projection {\n None => self.schema.clone(),\n Some(idxs) => Arc::new(arrow_schema::Schema::new(\n idxs.iter().map(|&i| self.schema.field(i).clone()).collect::<Vec<_>>(),\n )),\n};\nOk(Arc::new(SqliteFullScanExec::new(\n self.pool.clone(),\n self.sem.clone(),\n self.table_name.clone(),\n schema,\n)))\n
There was a problem hiding this comment.
Intentionally left unprojected. The only caller (plan_extension in planner.rs) always passes projection: None because the adaptive filter path needs all columns available to evaluate arbitrary WHERE predicates — projecting here risks dropping columns the filter references. Adding projection support would require the caller to compute the union of filter-referenced columns and key column, which adds complexity for no current benefit.
There was a problem hiding this comment.
Review
Issues
P1 — test_scan_returns_not_implemented will fail (tests/sqlite_provider_test.rs:147–160)
The existing regression test asserts that scan() returns a NotImplemented error containing "not support full table scans". This PR removes that error, so the test will fail. The test needs to be replaced with one that exercises the new happy path (full scan returns all rows). The PR test plan lists this scenario but no test was added.
P2 — Projection parameter ignored (inline comment on src/sqlite_provider.rs:320)
P2 — Partial column buffer corruption on column-read error (inline comment on src/sqlite_provider.rs:486–502)
Action Required
- Replace or update
test_scan_returns_not_implementedwith a test that registers the provider in aSessionContextand verifies a full-table SQL query returns the correct rows. - Address the partial-buffer truncation on the error path (see inline comment).
Prevents mismatched buffer lengths from causing a spurious second error during the final-batch flush, which would mask the original column read failure.
| Ok(Arc::new(SqliteFullScanExec::new( | ||
| self.pool.clone(), | ||
| self.sem.clone(), | ||
| self.table_name.clone(), | ||
| self.schema.clone(), | ||
| ))) |
There was a problem hiding this comment.
P2: projection hint is ignored — always scans all columns.
SqliteFullScanExec selects every column from SQLite regardless of the _projection argument. For wide tables where a query touches only a few columns, this wastes significant I/O per batch. The existing fetch_by_keys already handles projection correctly via out_schema.
SqliteFullScanExec should accept a projection: Option<Vec<usize>> field. The SQL col_list and the PlanProperties schema should both reflect only the projected columns, matching the pattern in execute_query_sync.
There was a problem hiding this comment.
Duplicate of the earlier thread — see reply at #discussion_r2945452082. The sole caller (plan_extension) always passes projection: None because the adaptive filter path requires all columns to evaluate arbitrary WHERE predicates. Adding projection support risks correctness (dropping filter-referenced columns) for no current benefit.
There was a problem hiding this comment.
Review: P1 issues found - see inline comments and summary below.
ISSUES:
P1 - test_scan_returns_not_implemented will fail (tests/sqlite_provider_test.rs:147-160)
An existing test explicitly asserts scan() returns an error containing 'not support full table scans'. This PR changes scan() to return Ok(...), making that test fail. The test file was not updated in this PR.
P1 - No test coverage for the new streaming scan path
The PR's own test plan items are all unchecked. SqliteFullScanExec has no tests: batching behavior, the flush boundary at exactly SCAN_BATCH_SIZE rows, concurrency under the semaphore, and error propagation paths in the blocking task are all untested.
ACTION REQUIRED:
- Delete or replace test_scan_returns_not_implemented in tests/sqlite_provider_test.rs - it asserts the old NotImplemented behavior and will break CI.
- Add at least one integration test that exercises the new scan path end-to-end and verifies row counts and data correctness.
Update test_scan_returns_not_implemented → test_scan_streams_all_rows: exercises the new SqliteFullScanExec end-to-end, verifying row count and data correctness across streamed batches.
| } | ||
|
|
||
| // Flush the last partial batch. | ||
| if rows_in_batch > 0 { |
There was a problem hiding this comment.
After a column-read error the loop breaks with rows_in_batch > 0, so this flush sends an Ok(batch) down the channel after an Err(...) was already sent. DataFusion consumers stop on the first error so this is harmless in practice, but it's confusing. Consider adding a had_error: bool flag and skipping the flush when it's set.
| Err(DataFusionError::NotImplemented( | ||
| "SqliteLookupProvider does not support full table scans; use fetch_by_keys".into(), | ||
| )) | ||
| Ok(Arc::new(SqliteFullScanExec::new( |
There was a problem hiding this comment.
Suggestion: thread _projection through to SqliteFullScanExec and use it to build a narrower SELECT column list. Right now every full scan fetches all columns from SQLite regardless of what the query needs. DataFusion will add a ProjectionExec on top so results are correct, but wide tables pay unnecessary deserialization cost.
Avoids sending Ok(batch) on the channel after an Err was already sent for a mid-row column read failure.
|
|
||
| fn execute( | ||
| &self, | ||
| _partition: usize, |
There was a problem hiding this comment.
Suggestion: validate partition before spawning. UnknownPartitioning(1) declares a single partition, so any other value is a caller bug — better to surface it than silently spawn a duplicate scan.
| _partition: usize, | |
| partition: usize, |
if partition != 0 {
return Err(DataFusionError::Internal(format!(
"SqliteFullScanExec only has 1 partition, got partition {partition}"
)));
}There was a problem hiding this comment.
DataFusion never calls execute() with partition > 0 when UnknownPartitioning(1) is declared — this is enforced by the framework. No other ExecutionPlan in this codebase (including USearchExec) adds this guard. Adding it here would be inconsistent defensive code for a scenario that can't happen.
| /// scan() returns a streaming ExecutionPlan that yields all rows in batches. | ||
| #[tokio::test] | ||
| async fn test_scan_returns_not_implemented() { | ||
| async fn test_scan_streams_all_rows() { |
There was a problem hiding this comment.
The existing 3-row test covers only the final partial-batch flush. The more complex path — where a full 1024-row batch is emitted mid-scan and buffers are drained — is never exercised. Consider adding a second test that inserts SCAN_BATCH_SIZE + 1 rows (or at minimum > SCAN_BATCH_SIZE) and asserts the total row count, to cover the batch-boundary drain logic.
1 participant
Summary
SqliteLookupProvider.scan()previously returnedNotImplemented, causing vector search queries with aWHEREclause to fail with "SqliteLookupProvider does not support full table scans"SqliteFullScanExec: a leafExecutionPlanthat streams all rows from the SQLite table in 1024-row batches via a boundedtokio::sync::mpscchannelspawn_blocking; each batch is evaluated throughevaluate_filters()and dropped immediately, keeping peak memory at O(batch_size) rather than O(total_rows)fetch_by_keys, so concurrent scans and key lookups stay within the configured pool sizeplanner.rsor any other file — the existingadaptive_filtered_executelogic works correctly oncescan()returns a valid streaming planTest plan
NotImplementederror