feat: late materialization of vectors in filtered vector search

KNN search is performed when a vector index is not present. When a table
is partially covered by a vector index, we perform a union of an ANN
search over the indexed data, and a KNN search over the unindexed data.
If the table is completely unindexed it is just a KNN search on the
data.

Prior to this commit, when we would execute the KNN portion of a
filtered vector search, we would perform a scan of all columns and
remove results that did not match the filter. For large vectors, this
amounts to a lot of overfetch from storage.

When filters are selective, it is more efficient to read the filter
column (typically much smaller than the vector), apply the filter, and
then select matching vectors by row ID.

This patch implements that strategy as well as an adaptive mechanism for
deciding when to apply it. There is a new configuration concept in the
scanner for specifying the filter selectivity at which it will be
cheaper to do a scan. We will compute a target rowcount based on that
threshold and scan the filter column for matches. If we encounter more
matches than the target, we will give up and switch to a scan.

Author: wkalt

rust/lance-datafusion/src/exec.rs

@@ -411,6 +411,21 @@ pub struct ExecutionSummaryCounts {
     pub all_counts: HashMap<String, usize>,
 }
 
+impl ExecutionSummaryCounts {
+    /// Create a new ExecutionSummaryCounts with all values initialized to zero
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Create a new ExecutionSummaryCounts with only custom counts
+    pub fn with_counts(counts: impl IntoIterator<Item = (impl Into<String>, usize)>) -> Self {
+        Self {
+            all_counts: counts.into_iter().map(|(k, v)| (k.into(), v)).collect(),
+            ..Default::default()
+        }
+    }
+}
+
 fn visit_node(node: &dyn ExecutionPlan, counts: &mut ExecutionSummaryCounts) {
     if let Some(metrics) = node.metrics() {
         for (metric_name, count) in metrics.iter_counts() {

rust/lance-tools/src/meta.rs

@@ -47,7 +47,7 @@ impl LanceToolFileMetadata {
             .open_file(&path, &CachedFileSize::unknown())
             .await?;
         let file_metadata = FileReader::read_all_metadata(&file_scheduler).await?;
-        let lance_tool_file_metadata = LanceToolFileMetadata { file_metadata };
+        let lance_tool_file_metadata = Self { file_metadata };
         Ok(lance_tool_file_metadata)
     }
 }

rust/lance/src/dataset.rs

@@ -8873,12 +8873,7 @@ mod tests {
         }
 
         fn make_tx(read_version: u64) -> Transaction {
-            Transaction::new(
-                read_version,
-                Operation::Append { fragments: vec![] },
-                None,
-                None,
-            )
+            Transaction::new(read_version, Operation::Append { fragments: vec![] }, None)
         }
 
         async fn delete_external_tx_file(ds: &Dataset) {
@@ -8939,7 +8934,6 @@ mod tests {
                 ds.load_indices().await.unwrap().as_ref().clone(),
                 &tx_file,
                 &ManifestWriteConfig::default(),
-                None,
             )
             .unwrap();
         let location = write_manifest_file(