feat: support segment-scoped scalar index scans

docs/scalar-segment-scans.md

@@ -0,0 +1,128 @@
+# Scalar index segment scans
+
+An ordinary scanner can use one physical BTree, Bitmap, or LabelList segment to
+generate candidates, then read them with the complete scanner filter. This
+does not run a global search of the other segments of the logical index. It does
+not subdivide a physical segment or make its own index search incremental.
+
+LabelList supports indexed array membership predicates. Every candidate search
+must return `SearchResult::Exact`.
+LabelList query values should match the array element type, for example
+`array_contains(int32_labels, CAST(42 AS INT))`; a cast on the indexed column
+can prevent the planner from finding an index driver and cause fallback.
+`AtMost` and `AtLeast` results still fall back; this mode does not enable FMIndex,
+NGram, BloomFilter, ZoneMap, or Inverted indices.
+
+## Configuring a task
+
+Open a fixed dataset version. Select the physical index UUID from that version's
+metadata and pass the task's complete fragment domain explicitly:
+
+```c
+LanceScanner *scanner = lance_scanner_new(dataset, columns, full_filter_sql);
+/* Check every return value in production. */
+lance_scanner_set_fragment_ids(scanner, fragment_ids, fragment_count);
+lance_scanner_set_scalar_index_segment(scanner, segment_uuid_16_bytes);
+lance_scanner_set_limit(scanner, 20000);
+/* The scanner-owning thread calls lance_scanner_next as usual. */
+```
+
+The segment setter copies the UUID; passing NULL clears it. Final option
+compatibility and snapshot metadata are checked when preparing the stream, not
+by the setter. Configure all options before the first `next`, Arrow stream
+export, or asynchronous scan. A failed preparation also freezes the options;
+create a new scanner to retry with different settings.
+
+SQL, Substrait and additional SQL filters keep their existing precedence and AND
+composition. The caller does not supply a separate driver predicate: Lance-C
+uses the typed filter planner and selects a necessary indexed leaf belonging to
+the requested logical index. It only descends through AND, never through OR or
+NOT, and chooses the first matching leaf in the planner's expression tree;
+this is not a selectivity-based choice or a guarantee of SQL text order.
+It then searches the selected UUID and applies the complete filter while
+reading candidates with automatic scalar-index planning disabled.
+
+Each task's fragment IDs define its result domain, including on fallback. A
+distributed planner must assign disjoint domains whose union covers the intended
+scan. Unindexed fragments need their own tasks, or an explicit domain including
+them (which causes that task to use fallback). Merely listing indexed segments
+does not include appended, unindexed data automatically.
+
+`use_scalar_index=false` disables segment search regardless of setter order.
+The scanner validates the selected snapshot UUID and fragment domain, then scans
+that domain with the full filter and LIMIT/OFFSET without opening the index or
+generating candidates. It reports `scalar_segment_fallback_disabled`.
+
+An unknown UUID, absent fragment, invalid option combination or segment metadata
+without one valid schema key field is an error. A known segment with
+incomplete/unknown coverage, no suitable driver, unsupported
+index type, nested key, overlays, unknown physical row counts, fragment reuse,
+non-exact results or unsupported
+row-ID domain falls back to a non-indexed scan of the entire explicit domain.
+Legacy (v1) storage also takes this fallback because ordinary scans cannot consume
+external row masks; it reports `scalar_segment_fallback_legacy_storage` without
+searching the index.
+I/O and corruption errors are propagated, not converted to empty results or
+successful fallback.
+
+The first implementation supports live-row ordinary scans and rejects vector/FTS
+queries and `include_deleted_rows=true` at stream creation, even when
+`use_scalar_index=false`. An index built after a delete does not contain the
+tombstoned rows, so even exact segment candidates
+cannot satisfy a scan that includes deleted rows. To read those rows, clear the
+segment setting and use an ordinary scan with `with_row_id=true`,
+`include_deleted_rows=true`, and `use_scalar_index=false`.
+Fragments removed from the current snapshot are not scanned by this option.
+Physical row-address
+results on stable-row-ID datasets currently fall back; results already expressed
+in the correct row-ID domain use the candidate path. Deletes and all remaining
+predicates are handled by the ordinary reader. No candidate-count limit is
+applied: LIMIT/OFFSET remain after the scanner's complete filter.
+
+If the host has additional predicates outside Lance, do not set a local limit
+before those predicates. Never divide the global limit by the number of tasks.
+Global OFFSET belongs to the coordinator, not independently to each task.
+
+## Stopping after the host limit
+
+This mode uses the existing scanner/stream lifecycle. Once the host has enough
+rows, it stops requesting further batches and closes the scanner after any active
+call has returned. Do not call `lance_scanner_close` concurrently with `next`.
+Exported Arrow streams remain owned by the caller and must also be released after
+their active consumers have finished.
+
+A host stop flag does not interrupt an in-progress `lance_scanner_next`: current
+index evaluation or I/O may finish before the host observes stop and closes the
+stream. No separate cancellation signal or thread is introduced. The host remains
+responsible for enforcing the global LIMIT across concurrent tasks.
+
+## Memory and statistics
+
+Candidate masks stay in Rust, and record batches are streamed. Each active task
+can still hold a complete segment's candidate set; scanner I/O buffer size does
+not cap that allocation. Control task concurrency and physical segment size.
+
+Successful exhaustion merges segment-search metrics into the existing statistics
+callback exactly once. New metrics include `scalar_segments_requested`,
+`scalar_segments_searched`, `scalar_segment_candidate_rows`,
+`scalar_segment_prepare_time`, `scalar_segment_search_time`, and
+`scalar_segment_fallbacks` plus `scalar_segment_fallback_*` reasons.
+`scalar_segment_prepare_time` includes search time. Metrics describe each
+successfully exhausted stream, including separately exported streams:
+
+- `scalar_segments_requested` is 1 even on fallback.
+- `scalar_segments_searched` is 1 after a completed index search, including one
+  whose inexact result causes fallback. A fallback can therefore include index work.
+- `scalar_segment_candidate_rows` counts TRUE rows in the exact segment result before fragment
+  restriction, residual filtering, deletion handling and LIMIT/OFFSET. It is not
+  the output or physical-read row count; a result without a known cardinality is
+  reported as 0. The reader's mask may also include NULL candidates that the full
+  filter subsequently discards.
+- A fallback records one reason, the first eligibility check that fails. Disabled
+  scalar indices and legacy storage bypass index opening and search after snapshot
+  validation. Other fallback reasons may be found after opening or searching an index.
+- Search and candidate metrics may be absent when their stage did not execute;
+  consumers should treat absent counts as 0.
+
+Early release, cancellation and errors retain the existing callback contract: final
+statistics are not guaranteed. Metrics do not establish global task concurrency.

include/lance/lance.h

@@ -1015,7 +1015,9 @@ int32_t lance_scanner_set_scan_in_order(LanceScanner* scanner, bool scan_in_orde
  * Configure whether scalar indices may be used to optimize filters.
  *
  * Scalar indices are enabled by default. Disable this to force filter
- * evaluation without scalar indices. This setting is independent of
+ * evaluation without scalar indices, including an explicitly selected scalar
+ * segment (which falls back to a scan of its explicit fragment_ids).
+ * This setting is independent of
  * `lance_scanner_set_use_index`, which controls vector ANN index usage.
  * Must be set before scanning starts.
  */
@@ -1051,7 +1053,11 @@ int32_t lance_scanner_with_row_address(LanceScanner* scanner, bool enable);
 
 /**
  * Configure whether deleted rows still present in storage are returned.
- * Deleted rows have a NULL `_rowid`; callers should also enable row IDs.
+ * Requires with_row_id=true; deleted rows have a NULL `_rowid`.
+ * For filtered scans, also set use_scalar_index=false: indices built after a
+ * deletion may omit tombstoned rows. Incompatible with scalar_index_segment,
+ * even when scalar indices are disabled.
+ * Fragments removed from the current snapshot are not scanned.
  * Must be set before scanning starts.
  */
 int32_t lance_scanner_set_include_deleted_rows(
@@ -1858,6 +1864,36 @@ int32_t lance_scanner_set_index_segments(
     size_t len
 );
 
+/**
+ * Accelerate an ordinary scalar-filtered scan with one physical index segment.
+ * segment_uuid points to 16 UUID bytes in RFC 4122 order; NULL clears the setting.
+ * Must be configured before scanning. Requires explicit nonempty fragment_ids,
+ * which define BOTH the read and fallback domain, independently of the segment.
+ * Missing snapshot UUIDs / fragment IDs are errors. Extra segment coverage is
+ * excluded by fragment_ids; incomplete coverage falls back to a full filtered
+ * scan of those fragment_ids. Callers distributing work must assign disjoint
+ * fragment domains and separately include any unindexed data they wish to read.
+ * The segment metadata must identify one key field present in the schema.
+ *
+ * BTree/Bitmap/LabelList searches use a necessary AND-conjunct of the
+ * full scanner filter on the selected logical index and require an Exact result.
+ * use_scalar_index=false skips segment search and uses the scoped fallback;
+ * snapshot UUID and fragment validation still applies.
+ * AtMost/AtLeast results fall back to a full filtered scan of fragment_ids.
+ * All predicates are reapplied during candidate reads; other scalar indices
+ * are disabled. Legacy storage, OR/NOT-only filters,
+ * overlays, fragment reuse, unsupported index types / result domains
+ * and missing coverage use the same domain without an index. No filter also
+ * falls back. LIMIT/OFFSET apply after the complete scanner filter, never to the
+ * unfiltered candidate set. Vector/FTS queries and include_deleted_rows=true
+ * are rejected even when use_scalar_index=false; segment mode is live-row-only.
+ *
+ * UUID bytes are copied. Metadata and final option compatibility are validated
+ * when creating the stream. Index corruption or I/O failures remain errors.
+ */
+int32_t lance_scanner_set_scalar_index_segment(
+    LanceScanner* scanner, const uint8_t* segment_uuid);
+
 /* ─── Full-text search (Phase 2) ─── */
 
 /**

include/lance/lance.hpp

@@ -1270,6 +1270,7 @@ class Scanner {
     }
 
     /// Configure whether scalar indices may be used to optimize filters.
+    /// False also disables explicit scalar segment search, retaining its fragment domain.
     Scanner& use_scalar_index(bool enable = true) {
         if (lance_scanner_set_use_scalar_index(handle_.get(), enable) != 0)
             check_error();
@@ -1305,12 +1306,30 @@ class Scanner {
     }
 
     /// Configure whether deleted rows still present in storage are returned.
+    /// Requires with_row_id(true); use_scalar_index(false) is needed for filtered scans.
+    /// Incompatible with scalar_index_segment. See lance.h.
     Scanner& include_deleted_rows(bool include_deleted_rows = true) {
         if (lance_scanner_set_include_deleted_rows(handle_.get(), include_deleted_rows) != 0)
             check_error();
         return *this;
     }
 
+    /// Generate exact candidates from one BTree/Bitmap/LabelList segment.
+    /// fragment_ids is required and defines the complete read/fallback domain. See lance.h.
+    /// Requires live rows only: include_deleted_rows(true) is rejected at stream creation.
+    /// use_scalar_index(false) selects the scoped fallback without searching the segment.
+    Scanner& scalar_index_segment(const std::array<uint8_t, 16>& segment_uuid) {
+        if (lance_scanner_set_scalar_index_segment(handle_.get(), segment_uuid.data()) != 0)
+            check_error();
+        return *this;
+    }
+
+    Scanner& clear_scalar_index_segment() {
+        if (lance_scanner_set_scalar_index_segment(handle_.get(), nullptr) != 0)
+            check_error();
+        return *this;
+    }
+
     /// Restrict scan to specific fragment IDs.
     Scanner& fragment_ids(const uint64_t* ids, size_t len) {
         if (lance_scanner_set_fragment_ids(handle_.get(), ids, len) != 0)

src/lib.rs

@@ -39,6 +39,7 @@ mod index_segment;
 mod merge_insert;
 mod restore;
 pub mod runtime;
+mod scalar_segment;
 mod scanner;
 mod session;
 pub mod stream_guard;