feat: expose row address allowlist/blocklist prefilters in C and C++

README.md

@@ -68,6 +68,7 @@ Based on the [liblance RFC](https://github.com/lance-format/lance/discussions/60
 | [x] | Async scan | Callback-based `lance_scanner_scan_async()` for non-blocking scans |
 | [x] | Dataset metadata | `lance_dataset_version()`, `lance_dataset_count_rows()`, `lance_dataset_latest_version()` |
 | [x] | Filter pushdown | `lance_scanner_set_substrait_filter()` accepts a serialized Substrait `ExtendedExpression`; `lance_scanner_additional_sql_filter()` adds SQL predicates with AND before scanning starts |
+| [x] | Row prefilter | `lance_scanner_set_row_addr_allowlist()` / `lance_scanner_set_row_addr_blocklist()` restrict scans, nearest-neighbor and full-text searches to (or exclude) a set of `_rowid`s before ranking |
 
 ## Segment-scoped array label filters
 

include/lance/lance.h

@@ -1287,6 +1287,42 @@ int32_t lance_scanner_set_fragment_ids(
     size_t len
 );
 
+/**
+ * Restrict the scan or search to rows whose `_rowid` is in `row_ids`.
+ *
+ * The ids are in the dataset's `_rowid` space: the stable row id when stable
+ * row ids are enabled, otherwise the row address. The allowlist is applied
+ * before ranking: a nearest or full-text search (including one using a
+ * prepared FTS query context) computes its top-k over the allowed rows only,
+ * and a plain scan reads only those rows. It is combined with any filter
+ * expression (AND) and with a blocklist (allowed AND NOT blocked). It cannot be
+ * combined with lance_scanner_set_scalar_index_segment().
+ * Calling it again replaces the previous allowlist. Must be called before
+ * iteration; the ids are copied.
+ * @param row_ids  Array of `_rowid` values, in any order (may be NULL when len is 0)
+ * @param len      Number of ids; 0 allows no rows
+ * @return 0 on success, -1 on error
+ */
+int32_t lance_scanner_set_row_addr_allowlist(
+    LanceScanner* scanner,
+    const uint64_t* row_ids,
+    size_t len
+);
+
+/**
+ * Exclude rows whose `_rowid` is in `row_ids` from the scan or search.
+ * Same `_rowid` space, ordering, and restrictions as
+ * lance_scanner_set_row_addr_allowlist().
+ * @param row_ids  Array of `_rowid` values, in any order (may be NULL when len is 0)
+ * @param len      Number of ids; 0 blocks no rows
+ * @return 0 on success, -1 on error
+ */
+int32_t lance_scanner_set_row_addr_blocklist(
+    LanceScanner* scanner,
+    const uint64_t* row_ids,
+    size_t len
+);
+
 /**
  * Set a Substrait filter on the scanner.
  *

include/lance/lance.hpp

@@ -1537,6 +1537,31 @@ class Scanner {
         return fragment_ids(ids.data(), ids.size());
     }
 
+    /// Restrict the scan or search to rows whose `_rowid` is in `row_ids`
+    /// (applied before ranking). `len == 0` allows no rows.
+    Scanner& row_addr_allowlist(const uint64_t* row_ids, size_t len) {
+        if (lance_scanner_set_row_addr_allowlist(handle_.get(), row_ids, len) != 0)
+            check_error();
+        return *this;
+    }
+
+    /// Restrict the scan or search to rows whose `_rowid` is in `row_ids` (vector overload).
+    Scanner& row_addr_allowlist(const std::vector<uint64_t>& row_ids) {
+        return row_addr_allowlist(row_ids.data(), row_ids.size());
+    }
+
+    /// Exclude rows whose `_rowid` is in `row_ids` (applied before ranking).
+    Scanner& row_addr_blocklist(const uint64_t* row_ids, size_t len) {
+        if (lance_scanner_set_row_addr_blocklist(handle_.get(), row_ids, len) != 0)
+            check_error();
+        return *this;
+    }
+
+    /// Exclude rows whose `_rowid` is in `row_ids` (vector overload).
+    Scanner& row_addr_blocklist(const std::vector<uint64_t>& row_ids) {
+        return row_addr_blocklist(row_ids.data(), row_ids.size());
+    }
+
     /// Set a Substrait filter (serialized ExtendedExpression bytes).
     /// Wins over any SQL filter passed to the Scanner constructor.
     Scanner& substrait_filter(const uint8_t* bytes, size_t len) {

src/scanner.rs

@@ -16,7 +16,8 @@ use datafusion::physical_plan::{ExecutionPlan, empty::EmptyExec};
 use futures::{FutureExt, Stream, StreamExt};
 use lance::Dataset;
 use lance::dataset::scanner::{
-    DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts,
+    DatasetRecordBatchStream, ExecutionStatsCallback, ExecutionSummaryCounts, RowAddrMask,
+    RowAddrTreeMap,
 };
 use lance::io::exec::fts::{FlatMatchQueryExec, MatchQueryExec, PhraseQueryExec};
 use lance_core::Result;
@@ -110,6 +111,10 @@ pub struct LanceScanner {
     with_row_address: bool,
     include_deleted_rows: bool,
     fragment_ids: Option<Vec<u64>>,
+    // External row prefilter keyed in the dataset's `_rowid` space; combined into
+    // a single `RowAddrMask` (allowed AND NOT blocked) when the scanner is built.
+    row_addr_allowlist: Option<RowAddrTreeMap>,
+    row_addr_blocklist: Option<RowAddrTreeMap>,
     index_segments: Option<Vec<Uuid>>,
     scalar_index_segment: Option<Uuid>,
     nearest: Option<NearestQuery>,
@@ -308,6 +313,8 @@ impl LanceScanner {
             with_row_address: false,
             include_deleted_rows: false,
             fragment_ids: None,
+            row_addr_allowlist: None,
+            row_addr_blocklist: None,
             index_segments: None,
             scalar_index_segment: None,
             nearest: None,
@@ -370,6 +377,20 @@ impl LanceScanner {
         Ok(())
     }
 
+    /// The external row prefilter, combined the same way as
+    /// `RowAddrMask::from_serialized_parts`: rows in the allowlist and not in the
+    /// blocklist. `None` when neither list is set (no mask, not "select nothing").
+    fn row_addr_mask(&self) -> Option<RowAddrMask> {
+        match (&self.row_addr_allowlist, &self.row_addr_blocklist) {
+            (Some(allow), Some(block)) => {
+                Some(RowAddrMask::from_allowed(allow.clone()).also_block(block.clone()))
+            }
+            (Some(allow), None) => Some(RowAddrMask::from_allowed(allow.clone())),
+            (None, Some(block)) => Some(RowAddrMask::from_block(block.clone())),
+            (None, None) => None,
+        }
+    }
+
     fn apply_filter(&self, scanner: &mut lance::dataset::scanner::Scanner) -> Result<()> {
         if let Some(substrait) = &self.substrait_filter {
             scanner.filter_substrait(substrait)?;
@@ -502,6 +523,18 @@ impl LanceScanner {
                 "fragment_ids cannot be combined with an FTS query context; split the query by FTS index segment UUID instead".into(),
             ));
         }
+        // A segment-scoped scalar scan installs its own row prefilter from the segment's
+        // candidates, which would replace this one.
+        let row_addr_mask = self.row_addr_mask().map(Arc::new);
+        if row_addr_mask.is_some() && self.scalar_index_segment.is_some() {
+            return Err(lance_core::Error::invalid_input_source(
+                "row address allowlist/blocklist cannot be combined with scalar_index_segment"
+                    .into(),
+            ));
+        }
+        if let Some(mask) = &row_addr_mask {
+            scanner.with_row_addr_prefilter(RowAddrMask::clone(mask));
+        }
         if self.prefilter {
             scanner.prefilter(true);
         }
@@ -577,6 +610,7 @@ impl LanceScanner {
                     && (self.filter.is_some()
                         || self.substrait_filter.is_some()
                         || !self.additional_sql_filters.is_empty()),
+                row_addr_mask,
                 batch_size: self.batch_size,
                 scan_statistics_callback: self.scan_statistics_callback.clone(),
             })
@@ -631,6 +665,9 @@ struct PreparedFtsExecution {
     context: Arc<FtsQueryContextInner>,
     segments: Vec<IndexMetadata>,
     scope_prefilter_to_fts_segments: bool,
+    // Re-applied to the rebuilt Match/Phrase execs: they are reconstructed from the
+    // planned ones, which do not expose the external mask Lance attached to them.
+    row_addr_mask: Option<Arc<RowAddrMask>>,
     batch_size: Option<usize>,
     scan_statistics_callback: Option<ExecutionStatsCallback>,
 }
@@ -689,6 +726,7 @@ impl PreparedScanner {
             &distributed_fts.segments,
             &distributed_fts.context.prepared,
             selected_segments_have_current_fragments,
+            distributed_fts.row_addr_mask.as_ref(),
         )?;
         if rewritten.indexed_query_execs > 1
             || rewritten.flat_match_query_execs > 1
@@ -785,6 +823,7 @@ fn rewrite_prepared_fts_plan(
     segments: &[IndexMetadata],
     prepared: &PreparedFtsQuery,
     selected_segments_have_current_fragments: bool,
+    row_addr_mask: Option<&Arc<RowAddrMask>>,
 ) -> Result<(Arc<dyn ExecutionPlan>, PreparedFtsPlanRewriteCounts)> {
     // Lance's ordinary FTS planner adds a flat-search branch for fragments not
     // covered by the logical index. A prepared INDEX_ONLY scan must omit that
@@ -812,6 +851,7 @@ fn rewrite_prepared_fts_plan(
                 segments,
                 prepared,
                 selected_segments_have_current_fragments,
+                row_addr_mask,
             )?;
             new_children.push(new_child);
             rewritten.indexed_query_execs += child_rewritten.indexed_query_execs;
@@ -841,7 +881,8 @@ fn rewrite_prepared_fts_plan(
             exec.prefilter_source().clone(),
             segments.to_vec(),
         )?
-        .with_base_scorer(Arc::clone(scorer));
+        .with_base_scorer(Arc::clone(scorer))
+        .with_external_mask(row_addr_mask.cloned());
         return Ok((Arc::new(replacement), rewritten));
     }
     if let Some(exec) = rebuilt.downcast_ref::<PhraseQueryExec>() {
@@ -861,7 +902,8 @@ fn rewrite_prepared_fts_plan(
             exec.prefilter_source().clone(),
             segments.to_vec(),
         )?
-        .with_base_scorer(Arc::clone(scorer));
+        .with_base_scorer(Arc::clone(scorer))
+        .with_external_mask(row_addr_mask.cloned());
         return Ok((Arc::new(replacement), rewritten));
     }
     Ok((rebuilt, rewritten))
@@ -1654,6 +1696,87 @@ unsafe fn scanner_set_fragment_ids_inner(
     Ok(0)
 }
 
+/// Restrict the scan or search to rows whose `_rowid` is in `row_ids`.
+///
+/// The ids are in the dataset's `_rowid` space: the stable row id when stable
+/// row ids are enabled, otherwise the row address. The allowlist is applied
+/// before ranking: a nearest or full-text search (including a prepared FTS
+/// query context) computes its top-k over the allowed rows only, and a plain
+/// scan reads only those rows. Combined with any filter expression (AND) and
+/// with a blocklist (allowed AND NOT blocked). `len == 0` allows no rows.
+/// Calling it again replaces the previous allowlist. Must be called before the
+/// scan starts; the ids are copied.
+///
+/// Returns 0 on success, -1 on error.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lance_scanner_set_row_addr_allowlist(
+    scanner: *mut LanceScanner,
+    row_ids: *const u64,
+    len: usize,
+) -> i32 {
+    scanner_poison_check!(scanner, -1);
+    scanner_ffi_try!(scanner, unsafe {
+        scanner_set_row_addr_list_inner(scanner, row_ids, len, RowAddrList::Allow)
+    })
+}
+
+/// Exclude rows whose `_rowid` is in `row_ids` from the scan or search.
+///
+/// Same `_rowid` space, timing, and replacement semantics as
+/// `lance_scanner_set_row_addr_allowlist`. `len == 0` blocks no rows.
+///
+/// Returns 0 on success, -1 on error.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn lance_scanner_set_row_addr_blocklist(
+    scanner: *mut LanceScanner,
+    row_ids: *const u64,
+    len: usize,
+) -> i32 {
+    scanner_poison_check!(scanner, -1);
+    scanner_ffi_try!(scanner, unsafe {
+        scanner_set_row_addr_list_inner(scanner, row_ids, len, RowAddrList::Block)
+    })
+}
+
+enum RowAddrList {
+    Allow,
+    Block,
+}
+
+unsafe fn scanner_set_row_addr_list_inner(
+    scanner: *mut LanceScanner,
+    row_ids: *const u64,
+    len: usize,
+    list: RowAddrList,
+) -> Result<i32> {
+    if scanner.is_null() {
+        return Err(lance_core::Error::invalid_input_source(
+            "scanner is NULL".into(),
+        ));
+    }
+    if row_ids.is_null() && len > 0 {
+        return Err(lance_core::Error::invalid_input_source(
+            "row_ids is NULL but len > 0".into(),
+        ));
+    }
+    let s = unsafe { &mut *scanner };
+    s.ensure_scan_not_started(match list {
+        RowAddrList::Allow => "row_addr_allowlist",
+        RowAddrList::Block => "row_addr_blocklist",
+    })?;
+    let ids = if len > 0 {
+        unsafe { std::slice::from_raw_parts(row_ids, len) }
+    } else {
+        &[]
+    };
+    let tree_map: RowAddrTreeMap = ids.iter().collect();
+    match list {
+        RowAddrList::Allow => s.row_addr_allowlist = Some(tree_map),
+        RowAddrList::Block => s.row_addr_blocklist = Some(tree_map),
+    }
+    Ok(0)
+}
+
 /// Set a Substrait filter on the scanner.
 ///
 /// `bytes` must point to a serialized Substrait
@@ -3839,6 +3962,7 @@ mod tests {
             &segments,
             &distributed.context.prepared,
             has_current_fragments,
+            None,
         )
         .unwrap();
         assert_eq!(counts.indexed_query_execs, 1);

tests/cpp/test_c_api.c

@@ -263,6 +263,58 @@ static void test_scan_with_limit(const char *uri) {
     printf("OK\n");
 }
 
+static uint64_t count_stream_rows(struct ArrowArrayStream *stream) {
+    uint64_t total_rows = 0;
+    while (1) {
+        struct ArrowArray array;
+        memset(&array, 0, sizeof(array));
+        ASSERT(stream->get_next(stream, &array) == 0, "get_next failed");
+        if (array.release == NULL) break;
+        total_rows += (uint64_t)array.length;
+        array.release(&array);
+    }
+    if (stream->release) stream->release(stream);
+    return total_rows;
+}
+
+static void test_scan_with_row_addr_lists(const char *uri) {
+    printf("  test_scan_with_row_addr_lists... ");
+
+    LanceDataset *ds = lance_dataset_open(uri, NULL, 0);
+    ASSERT(ds != NULL, "dataset open failed");
+    uint64_t expected_rows = lance_dataset_count_rows(ds);
+    CHECK_OK();
+
+    /* Rows 0, 2 and 4 of the first fragment, minus row 2. */
+    const uint64_t allowed[] = {4, 0, 2};
+    const uint64_t blocked[] = {2};
+    LanceScanner *scanner = lance_scanner_new(ds, NULL, NULL);
+    ASSERT(scanner != NULL, "scanner creation failed");
+    ASSERT(lance_scanner_set_row_addr_allowlist(scanner, allowed, 3) == 0, "set allowlist");
+    ASSERT(lance_scanner_set_row_addr_blocklist(scanner, blocked, 1) == 0, "set blocklist");
+    struct ArrowArrayStream stream;
+    memset(&stream, 0, sizeof(stream));
+    ASSERT(lance_scanner_to_arrow_stream(scanner, &stream) == 0, "to_arrow_stream failed");
+    ASSERT(count_stream_rows(&stream) == 2, "allowlist minus blocklist should keep 2 rows");
+    ASSERT(lance_scanner_set_row_addr_allowlist(scanner, allowed, 3) == -1,
+           "allowlist after the scan started must fail");
+    lance_scanner_close(scanner);
+
+    /* An empty blocklist keeps every row; NULL is valid when len is 0. */
+    scanner = lance_scanner_new(ds, NULL, NULL);
+    ASSERT(scanner != NULL, "scanner creation failed");
+    ASSERT(lance_scanner_set_row_addr_blocklist(scanner, NULL, 0) == 0, "set empty blocklist");
+    ASSERT(lance_scanner_set_row_addr_allowlist(scanner, NULL, 1) == -1,
+           "NULL row_ids with len > 0 must fail");
+    memset(&stream, 0, sizeof(stream));
+    ASSERT(lance_scanner_to_arrow_stream(scanner, &stream) == 0, "to_arrow_stream failed");
+    ASSERT(count_stream_rows(&stream) == expected_rows, "empty blocklist should keep every row");
+    lance_scanner_close(scanner);
+
+    lance_dataset_close(ds);
+    printf("OK\n");
+}
+
 /* Copy the Arrow C Data Interface format of the `blob` column of a stream's
  * schema into `out`; `out` is empty when the column is missing. */
 static void blob_column_format(struct ArrowArrayStream *stream, char *out, size_t out_len) {
@@ -1479,6 +1531,7 @@ int main(int argc, char **argv) {
     test_scan(uri);
     test_distance_range(uri);
     test_scan_with_limit(uri);
+    test_scan_with_row_addr_lists(uri);
     test_scanner_blob_handling(blob_uri);
     test_take_blobs(blob_uri);
     test_versions(uri);