feat(scanner): add external row-address mask prefilter

rust/lance/src/io/exec.rs

@@ -18,6 +18,7 @@ pub(crate) mod knn;
 mod optimizer;
 mod projection;
 mod pushdown_scan;
+pub(crate) mod row_addr_mask;
 mod rowids;
 pub mod scalar_index;
 mod scan;
@@ -35,6 +36,7 @@ pub use lance_index::scalar::expression::FilterPlan;
 pub use optimizer::get_physical_optimizer;
 pub use projection::project;
 pub use pushdown_scan::{LancePushdownScanExec, ScanConfig};
+pub use row_addr_mask::RowAddrMaskFilterExec;
 pub use rowids::{AddRowAddrExec, AddRowOffsetExec};
 pub use scan::{LanceScanConfig, LanceScanExec};
 pub use take::TakeExec;

rust/lance/src/io/exec/filtered_read.rs

@@ -2314,7 +2314,12 @@ impl ExecutionPlan for FilteredReadExec {
         let mut updated_options = self.options.clone();
 
         if self.options.full_filter.is_none() && self.options.refine_filter.is_none() {
-            if self.options.scan_range_before_filter.is_some() {
+            // A before-filter range trims raw scan positions, which is only valid for
+            // an unindexed full scan. With an index_input (e.g. an external row mask or
+            // a scalar-index result) the rows are selected by that input, so a pre-range
+            // would apply before selection and keep the wrong rows; leave the limit to a
+            // node above the read instead.
+            if self.options.scan_range_before_filter.is_some() || self.index_input.is_some() {
                 return None;
             }
             updated_options.scan_range_before_filter = Some(0..(limit as u64));
@@ -3322,6 +3327,35 @@ mod tests {
             let result = plan.with_fetch(None);
             assert!(result.is_none());
         }
+
+        // Case 7: index_input present with no filter (the external-row-mask
+        // plain-scan shape) - with_fetch must reject before-filter pushdown, since
+        // the index_input selects the rows and a raw before-filter range would trim
+        // scan positions before that selection.
+        {
+            // Build a real scalar-index input, then attach it to options that carry
+            // no filter of their own.
+            let index_filter_plan = fixture.filter_plan("fully_indexed < 200", false).await;
+            let index_input = fixture
+                .index_input(&base_options.clone().with_filter_plan(index_filter_plan))
+                .await;
+            assert!(index_input.is_some(), "expected a scalar-index input");
+
+            let plan = FilteredReadExec::try_new(
+                fixture.dataset.clone(),
+                base_options.clone(),
+                index_input,
+            )
+            .unwrap();
+            assert!(plan.index_input().is_some());
+            assert!(plan.options().full_filter.is_none() && plan.options().refine_filter.is_none());
+
+            let result = plan.with_fetch(Some(100));
+            assert!(
+                result.is_none(),
+                "with_fetch must reject before-filter pushdown when index_input is present"
+            );
+        }
     }
 
     #[tokio::test]

rust/lance/src/io/exec/fts.rs

@@ -31,6 +31,7 @@ use lance_core::{
     utils::{tokio::get_num_compute_intensive_cpus, tracing::StreamTracingExt},
 };
 use lance_datafusion::utils::{ExecutionPlanMetricsSetExt, MetricsExt, PARTITIONS_SEARCHED_METRIC};
+use lance_select::RowAddrMask;
 use lance_table::format::IndexMetadata;
 
 use super::PreFilterSource;
@@ -238,6 +239,9 @@ pub struct MatchQueryExec {
     /// When set, `execute()` skips `load_segments` and searches exactly these
     /// segments.
     preset_segments: Option<Vec<IndexMetadata>>,
+    /// Optional external row-address mask combined (logical AND) with the BM25
+    /// prefilter so only masked rows are scored (see [`Self::with_external_mask`]).
+    external_mask: Option<Arc<RowAddrMask>>,
 
     properties: Arc<PlanProperties>,
     metrics: ExecutionPlanMetricsSet,
@@ -296,6 +300,7 @@ impl MatchQueryExec {
             prefilter_source,
             base_scorer: None,
             preset_segments: None,
+            external_mask: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -331,6 +336,7 @@ impl MatchQueryExec {
             prefilter_source,
             base_scorer: None,
             preset_segments: Some(segments),
+            external_mask: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -353,6 +359,15 @@ impl MatchQueryExec {
         self
     }
 
+    /// Restrict BM25 scoring to rows selected by an external row-address mask.
+    /// The mask is combined (logical AND) with the prefilter (see
+    /// [`build_prefilter`]) so top-k is computed over masked rows only. No-op
+    /// when `mask` is `None`.
+    pub fn with_external_mask(mut self, mask: Option<Arc<RowAddrMask>>) -> Self {
+        self.external_mask = mask;
+        self
+    }
+
     pub fn query(&self) -> &MatchQuery {
         &self.query
     }
@@ -418,6 +433,7 @@ impl ExecutionPlan for MatchQueryExec {
                     prefilter_source: PreFilterSource::None,
                     base_scorer: self.base_scorer.clone(),
                     preset_segments: self.preset_segments.clone(),
+                    external_mask: self.external_mask.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -445,6 +461,7 @@ impl ExecutionPlan for MatchQueryExec {
                     prefilter_source,
                     base_scorer: self.base_scorer.clone(),
                     preset_segments: self.preset_segments.clone(),
+                    external_mask: self.external_mask.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -468,6 +485,7 @@ impl ExecutionPlan for MatchQueryExec {
         let params = self.params.clone();
         let ds = self.dataset.clone();
         let prefilter_source = self.prefilter_source.clone();
+        let external_mask = self.external_mask.clone();
         let preset_base_scorer = self.base_scorer.clone();
         let preset_segments = self.preset_segments.clone();
         let metrics = Arc::new(FtsIndexMetrics::new(&self.metrics, partition));
@@ -490,8 +508,14 @@ impl ExecutionPlan for MatchQueryExec {
             let indices =
                 open_fts_segments(&ds, &column, &segments, &metrics.index_metrics).await?;
 
-            let mut pre_filter =
-                build_prefilter(context.clone(), partition, &prefilter_source, ds, &segments)?;
+            let mut pre_filter = build_prefilter(
+                context.clone(),
+                partition,
+                &prefilter_source,
+                ds,
+                &segments,
+                external_mask,
+            )?;
             let deleted_fragments =
                 indices
                     .iter()
@@ -1155,6 +1179,9 @@ pub struct PhraseQueryExec {
     /// Optional pre-resolved segment list. See
     /// [`MatchQueryExec::new_with_segments`].
     preset_segments: Option<Vec<IndexMetadata>>,
+    /// Optional external row-address mask combined (logical AND) with the BM25
+    /// prefilter so only masked rows are scored (see [`MatchQueryExec::with_external_mask`]).
+    external_mask: Option<Arc<RowAddrMask>>,
     properties: Arc<PlanProperties>,
     metrics: ExecutionPlanMetricsSet,
 }
@@ -1204,6 +1231,7 @@ impl PhraseQueryExec {
             prefilter_source,
             base_scorer: None,
             preset_segments: None,
+            external_mask: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -1232,6 +1260,7 @@ impl PhraseQueryExec {
             prefilter_source,
             base_scorer: None,
             preset_segments: Some(segments),
+            external_mask: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -1243,6 +1272,12 @@ impl PhraseQueryExec {
         self
     }
 
+    /// See [`MatchQueryExec::with_external_mask`].
+    pub fn with_external_mask(mut self, mask: Option<Arc<RowAddrMask>>) -> Self {
+        self.external_mask = mask;
+        self
+    }
+
     pub fn query(&self) -> &PhraseQuery {
         &self.query
     }
@@ -1301,6 +1336,7 @@ impl ExecutionPlan for PhraseQueryExec {
                 prefilter_source: PreFilterSource::None,
                 base_scorer: self.base_scorer.clone(),
                 preset_segments: self.preset_segments.clone(),
+                external_mask: self.external_mask.clone(),
                 properties: self.properties.clone(),
                 metrics: ExecutionPlanMetricsSet::new(),
             },
@@ -1326,6 +1362,7 @@ impl ExecutionPlan for PhraseQueryExec {
                     prefilter_source,
                     base_scorer: self.base_scorer.clone(),
                     preset_segments: self.preset_segments.clone(),
+                    external_mask: self.external_mask.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -1349,6 +1386,7 @@ impl ExecutionPlan for PhraseQueryExec {
         let params = self.params.clone();
         let ds = self.dataset.clone();
         let prefilter_source = self.prefilter_source.clone();
+        let external_mask = self.external_mask.clone();
         let preset_base_scorer = self.base_scorer.clone();
         let preset_segments = self.preset_segments.clone();
         let metrics = Arc::new(FtsIndexMetrics::new(&self.metrics, partition));
@@ -1371,8 +1409,14 @@ impl ExecutionPlan for PhraseQueryExec {
             let indices =
                 open_fts_segments(&ds, &column, &segments, &metrics.index_metrics).await?;
 
-            let mut pre_filter =
-                build_prefilter(context.clone(), partition, &prefilter_source, ds, &segments)?;
+            let mut pre_filter = build_prefilter(
+                context.clone(),
+                partition,
+                &prefilter_source,
+                ds,
+                &segments,
+                external_mask,
+            )?;
             let deleted_fragments =
                 indices
                     .iter()

rust/lance/src/io/exec/knn.rs

@@ -56,6 +56,7 @@ use lance_index::vector::{
 };
 use lance_linalg::distance::DistanceType;
 use lance_linalg::kernels::normalize_arrow;
+use lance_select::RowAddrMask;
 use lance_table::format::IndexMetadata;
 use tokio::sync::Notify;
 use uuid::Uuid;
@@ -67,6 +68,7 @@ use crate::index::vector::utils::{get_vector_type, validate_distance_type_for};
 use crate::{Error, Result};
 use lance_arrow::*;
 
+use super::row_addr_mask::MaskAndLoader;
 use super::utils::{
     FilteredRowIdsToPrefilter, IndexMetrics, InstrumentedRecordBatchStreamAdapter, PreFilterSource,
     SelectionVectorToPrefilter,
@@ -1104,6 +1106,7 @@ pub fn new_knn_exec(
     indices: &[IndexMetadata],
     query: &Query,
     prefilter_source: PreFilterSource,
+    external_mask: Option<Arc<RowAddrMask>>,
 ) -> Result<Arc<dyn ExecutionPlan>> {
     let ivf_node = ANNIvfPartitionExec::try_new(
         dataset.clone(),
@@ -1117,6 +1120,7 @@ pub fn new_knn_exec(
         indices.to_vec(),
         query.clone(),
         prefilter_source,
+        external_mask,
     )?;
 
     Ok(Arc::new(sub_index))
@@ -1377,6 +1381,10 @@ pub struct ANNIvfSubIndexExec {
     /// Prefiltering input
     prefilter_source: PreFilterSource,
 
+    /// Optional external row-address allow/block mask, combined with the
+    /// prefilter using logical AND.
+    external_mask: Option<Arc<RowAddrMask>>,
+
     /// Datafusion Plan Properties
     properties: Arc<PlanProperties>,
 
@@ -1390,6 +1398,7 @@ impl ANNIvfSubIndexExec {
         indices: Vec<IndexMetadata>,
         query: Query,
         prefilter_source: PreFilterSource,
+        external_mask: Option<Arc<RowAddrMask>>,
     ) -> Result<Self> {
         if input.schema().field_with_name(PART_ID_COLUMN).is_err() {
             return Err(Error::index(format!(
@@ -1409,6 +1418,7 @@ impl ANNIvfSubIndexExec {
             indices,
             query,
             prefilter_source,
+            external_mask,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         })
@@ -1880,6 +1890,7 @@ impl ExecutionPlan for ANNIvfSubIndexExec {
                 indices: self.indices.clone(),
                 query: self.query.clone(),
                 prefilter_source,
+                external_mask: self.external_mask.clone(),
                 properties: self.properties.clone(),
                 metrics: ExecutionPlanMetricsSet::new(),
             }
@@ -1960,6 +1971,14 @@ impl ExecutionPlan for ANNIvfSubIndexExec {
             PreFilterSource::None => None,
         };
 
+        // AND the external row-address mask into whatever the filter produced.
+        let prefilter_loader = match self.external_mask.clone() {
+            Some(mask) => {
+                Some(Box::new(MaskAndLoader::new(mask, prefilter_loader)) as Box<dyn FilterLoader>)
+            }
+            None => prefilter_loader,
+        };
+
         let pre_filter = Arc::new(DatasetPreFilter::new(
             ds.clone(),
             &indices,