diff --git a/rust/lance-index/src/scalar/inverted/index.rs b/rust/lance-index/src/scalar/inverted/index.rs
index a6db0e355c9..dbbbb3e8acb 100644
--- a/rust/lance-index/src/scalar/inverted/index.rs
+++ b/rust/lance-index/src/scalar/inverted/index.rs
@@ -18,7 +18,7 @@ use std::{
 use crate::metrics::NoOpMetricsCollector;
 use crate::prefilter::NoFilter;
 use crate::scalar::registry::{TrainingCriteria, TrainingOrdering};
-use arrow::array::{FixedSizeListBuilder, Float32Builder, Int32Builder};
+use arrow::array::{BooleanBuilder, FixedSizeListBuilder, Float32Builder, Int32Builder};
 use arrow::datatypes::{self, Float32Type, Int32Type, UInt64Type};
 use arrow::{
     array::{
@@ -28,8 +28,8 @@ use arrow::{
 };
 use arrow::{buffer::ScalarBuffer, datatypes::UInt32Type};
 use arrow_array::{
-    Array, ArrayRef, Float32Array, LargeBinaryArray, ListArray, OffsetSizeTrait, RecordBatch,
-    UInt32Array, UInt64Array,
+    Array, ArrayRef, BooleanArray, Float32Array, LargeBinaryArray, ListArray, OffsetSizeTrait,
+    RecordBatch, UInt32Array, UInt64Array,
 };
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
 use async_trait::async_trait;
@@ -7130,6 +7130,7 @@ fn do_flat_full_text_search_list<ListOffset: OffsetSizeTrait>(
 const FLAT_ROW_ID_COL_IDX: usize = 0;
 const FLAT_ALL_TOKENS_COL_IDX: usize = 1;
 const FLAT_QUERY_TOKEN_COUNTS_COL_IDX: usize = 2;
+const FLAT_PHRASE_MATCH_COL: &str = "phrase_match";
 
 /// If we accumulate this many bytes we warn the user they probably want to use an FTS index instead.
 const BYTES_ACCUMULATED_WARNING_THRESHOLD: u64 = 1024 * 1024 * 1024; // 1GB
@@ -7152,9 +7153,10 @@ async fn tokenize_and_count(
     tokenizer: Box<dyn LanceTokenizer>,
     query_tokens: Arc<Tokens>,
     doc_col_idx: usize,
+    phrase_slop: Option<u32>,
     elapsed_compute: Option<Time>,
 ) -> DataFusionResult<RecordBatch> {
-    let output_schema = Arc::new(Schema::new(vec![
+    let mut output_fields = vec![
         ROW_ID_FIELD.clone(),
         Field::new("all_tokens", DataType::UInt64, false),
         Field::new(
@@ -7165,9 +7167,15 @@ async fn tokenize_and_count(
             ),
             false,
         ),
-    ]));
+    ];
+    if phrase_slop.is_some() {
+        output_fields.push(Field::new(FLAT_PHRASE_MATCH_COL, DataType::Boolean, false));
+    }
+    let output_schema = Arc::new(Schema::new(output_fields));
     let output_schema_clone = output_schema.clone();
     let query_token_indices = Arc::new(query_token_indices(query_tokens.as_ref()));
+    let query_position_groups =
+        phrase_slop.map(|_| Arc::new(query_position_groups(query_tokens.as_ref())));
     let bytes_accumulated = Arc::new(AtomicU64::new(0));
     let bytes_warning_emitted = Arc::new(AtomicBool::new(false));
 
@@ -7177,6 +7185,7 @@ async fn tokenize_and_count(
             let output_schema = output_schema.clone();
             let query_tokens = query_tokens.clone();
             let query_token_indices = query_token_indices.clone();
+            let query_position_groups = query_position_groups.clone();
             let bytes_accumulated = bytes_accumulated.clone();
             let bytes_warning_emitted = bytes_warning_emitted.clone();
             let elapsed_compute = elapsed_compute.clone();
@@ -7194,28 +7203,61 @@ async fn tokenize_and_count(
                     query_tokens.len() as i32,
                     batch.num_rows(),
                 );
+                let mut phrase_matches = phrase_slop
+                    .map(|_| BooleanBuilder::with_capacity(batch.num_rows()));
                 let mut temp_query_token_counts = Vec::with_capacity(query_tokens.len());
-                let mut count_text = |doc: &str, temp_query_token_counts: &mut Vec<u64>| -> u64 {
+                let mut temp_query_token_positions = phrase_slop
+                    .map(|_| vec![Vec::<u32>::new(); query_tokens.len()]);
+                let mut count_text =
+                    |row_id: u64,
+                     doc: &str,
+                     temp_query_token_counts: &mut Vec<u64>,
+                     temp_query_token_positions: Option<&mut Vec<Vec<u32>>>|
+                     -> DataFusionResult<u64> {
                     let mut stream = tokenizer.token_stream_for_doc(doc);
                     let mut all_tokens = 0;
+                    let mut temp_query_token_positions = temp_query_token_positions;
                     while let Some(token) = stream.next() {
                         all_tokens += 1;
                         if let Some(token_indices) = query_token_indices.get(&token.text) {
+                            let position = if temp_query_token_positions.is_some() {
+                                Some(u32::try_from(token.position).map_err(|_| {
+                                    datafusion_common::DataFusionError::Execution(format!(
+                                        "token position overflow for row_id={row_id}: token_position={}",
+                                        token.position
+                                    ))
+                                })?)
+                            } else {
+                                None
+                            };
                             for token_index in token_indices {
                                 temp_query_token_counts[*token_index] += 1;
+                                if let (Some(positions), Some(position)) =
+                                    (temp_query_token_positions.as_deref_mut(), position)
+                                {
+                                    positions[*token_index].push(position);
+                                }
                             }
                         }
                     }
-                    all_tokens
+                    Ok(all_tokens)
                 };
                 let mut append_counts =
-                    |row_id: u64, all_tokens: u64, temp_query_token_counts: &[u64]| {
+                    |row_id: u64,
+                     all_tokens: u64,
+                     temp_query_token_counts: &[u64],
+                     is_phrase_match: Option<bool>| {
                         row_ids.append_value(row_id);
                         all_token_counts.append_value(all_tokens);
                         for count in temp_query_token_counts.iter().copied() {
                             query_token_counts.values().append_value(count);
                         }
                         query_token_counts.append(true);
+                        if let (Some(builder), Some(is_phrase_match)) =
+                            (phrase_matches.as_mut(), is_phrase_match)
+                        {
+                            builder.append_value(is_phrase_match);
+                        }
                     };
                 match batch.column(doc_col_idx).data_type() {
                     DataType::Utf8 | DataType::LargeUtf8 => {
@@ -7224,36 +7266,101 @@ async fn tokenize_and_count(
                             temp_query_token_counts.clear();
                             temp_query_token_counts
                                 .extend(std::iter::repeat_n(0, query_tokens.len()));
+                            if let Some(positions) = temp_query_token_positions.as_mut() {
+                                positions.iter_mut().for_each(Vec::clear);
+                            }
 
                             let Some(doc) = doc else {
                                 continue;
                             };
 
-                            let all_tokens = count_text(doc, &mut temp_query_token_counts);
+                            let all_tokens = count_text(
+                                *row_id,
+                                doc,
+                                &mut temp_query_token_counts,
+                                temp_query_token_positions.as_mut(),
+                            )?;
                             if all_tokens > 0 {
-                                append_counts(*row_id, all_tokens, &temp_query_token_counts);
+                                let is_phrase_match = phrase_slop.map(|slop| {
+                                    phrase_matches_positions(
+                                        query_tokens.as_ref(),
+                                        query_position_groups
+                                            .as_deref()
+                                            .expect("phrase groups exist when slop is set"),
+                                        temp_query_token_positions
+                                            .as_deref()
+                                            .expect("phrase positions exist when slop is set"),
+                                        slop,
+                                    )
+                                });
+                                append_counts(
+                                    *row_id,
+                                    all_tokens,
+                                    &temp_query_token_counts,
+                                    is_phrase_match,
+                                );
                             }
                         }
                     }
                     DataType::List(_) => {
-                        tokenize_and_count_list::<i32>(
-                            batch.column(doc_col_idx),
-                            row_id_array,
-                            &mut count_text,
-                            &mut append_counts,
-                            &mut temp_query_token_counts,
-                            query_tokens.len(),
-                        )?;
+                        if let Some(slop) = phrase_slop {
+                            tokenize_and_count_phrase_list::<i32>(
+                                batch.column(doc_col_idx),
+                                row_id_array,
+                                &mut count_text,
+                                &mut append_counts,
+                                &mut temp_query_token_counts,
+                                temp_query_token_positions
+                                    .as_mut()
+                                    .expect("phrase positions exist when slop is set"),
+                                query_tokens.as_ref(),
+                                query_position_groups
+                                    .as_deref()
+                                    .expect("phrase groups exist when slop is set"),
+                                slop,
+                            )?;
+                        } else {
+                            tokenize_and_count_list::<i32>(
+                                batch.column(doc_col_idx),
+                                row_id_array,
+                                &mut |doc, counts| count_text(0, doc, counts, None),
+                                &mut |row_id, all_tokens, counts| {
+                                    append_counts(row_id, all_tokens, counts, None)
+                                },
+                                &mut temp_query_token_counts,
+                                query_tokens.len(),
+                            )?;
+                        }
                     }
                     DataType::LargeList(_) => {
-                        tokenize_and_count_list::<i64>(
-                            batch.column(doc_col_idx),
-                            row_id_array,
-                            &mut count_text,
-                            &mut append_counts,
-                            &mut temp_query_token_counts,
-                            query_tokens.len(),
-                        )?;
+                        if let Some(slop) = phrase_slop {
+                            tokenize_and_count_phrase_list::<i64>(
+                                batch.column(doc_col_idx),
+                                row_id_array,
+                                &mut count_text,
+                                &mut append_counts,
+                                &mut temp_query_token_counts,
+                                temp_query_token_positions
+                                    .as_mut()
+                                    .expect("phrase positions exist when slop is set"),
+                                query_tokens.as_ref(),
+                                query_position_groups
+                                    .as_deref()
+                                    .expect("phrase groups exist when slop is set"),
+                                slop,
+                            )?;
+                        } else {
+                            tokenize_and_count_list::<i64>(
+                                batch.column(doc_col_idx),
+                                row_id_array,
+                                &mut |doc, counts| count_text(0, doc, counts, None),
+                                &mut |row_id, all_tokens, counts| {
+                                    append_counts(row_id, all_tokens, counts, None)
+                                },
+                                &mut temp_query_token_counts,
+                                query_tokens.len(),
+                            )?;
+                        }
                     }
                     data_type => {
                         return DataFusionResult::Err(datafusion_common::DataFusionError::Execution(
@@ -7264,13 +7371,17 @@ async fn tokenize_and_count(
                 let row_ids = row_ids.finish();
                 let all_token_counts = all_token_counts.finish();
                 let query_token_counts = query_token_counts.finish();
+                let mut columns = vec![
+                    Arc::new(row_ids) as ArrayRef,
+                    Arc::new(all_token_counts) as ArrayRef,
+                    Arc::new(query_token_counts) as ArrayRef,
+                ];
+                if let Some(mut phrase_matches) = phrase_matches {
+                    columns.push(Arc::new(phrase_matches.finish()) as ArrayRef);
+                }
                 let result_batch = RecordBatch::try_new(
                     output_schema,
-                    vec![
-                        Arc::new(row_ids) as ArrayRef,
-                        Arc::new(all_token_counts) as ArrayRef,
-                        Arc::new(query_token_counts) as ArrayRef,
-                    ],
+                    columns,
                 )?;
                 let bytes_accumulated = bytes_accumulated.fetch_add(result_batch.get_array_memory_size() as u64, Ordering::Relaxed);
                 if bytes_accumulated > BYTES_ACCUMULATED_WARNING_THRESHOLD && !bytes_warning_emitted.swap(true, Ordering::Relaxed) {
@@ -7296,7 +7407,7 @@ async fn tokenize_and_count(
 fn tokenize_and_count_list<ListOffset: OffsetSizeTrait>(
     doc_col: &ArrayRef,
     row_id_array: &arrow_array::PrimitiveArray<UInt64Type>,
-    count_text: &mut impl FnMut(&str, &mut Vec<u64>) -> u64,
+    count_text: &mut impl FnMut(&str, &mut Vec<u64>) -> DataFusionResult<u64>,
     append_counts: &mut impl FnMut(u64, u64, &[u64]),
     temp_query_token_counts: &mut Vec<u64>,
     query_tokens_len: usize,
@@ -7323,7 +7434,7 @@ fn tokenize_and_count_list<ListOffset: OffsetSizeTrait>(
         let elements = doc_array.value(i);
         let mut all_tokens = 0;
         for element in iter_str_array(elements.as_ref()).flatten() {
-            all_tokens += count_text(element, temp_query_token_counts);
+            all_tokens += count_text(element, temp_query_token_counts)?;
         }
 
         if all_tokens > 0 {
@@ -7334,6 +7445,77 @@ fn tokenize_and_count_list<ListOffset: OffsetSizeTrait>(
     Ok(())
 }
 
+#[allow(clippy::too_many_arguments)]
+fn tokenize_and_count_phrase_list<ListOffset: OffsetSizeTrait>(
+    doc_col: &ArrayRef,
+    row_id_array: &arrow_array::PrimitiveArray<UInt64Type>,
+    count_text: &mut impl FnMut(
+        u64,
+        &str,
+        &mut Vec<u64>,
+        Option<&mut Vec<Vec<u32>>>,
+    ) -> DataFusionResult<u64>,
+    append_counts: &mut impl FnMut(u64, u64, &[u64], Option<bool>),
+    temp_query_token_counts: &mut Vec<u64>,
+    temp_query_token_positions: &mut Vec<Vec<u32>>,
+    query_tokens: &Tokens,
+    query_position_groups: &[Vec<usize>],
+    slop: u32,
+) -> DataFusionResult<()> {
+    let doc_array = doc_col.as_list::<ListOffset>();
+    match doc_array.value_type() {
+        DataType::Utf8 | DataType::LargeUtf8 => {}
+        data_type => {
+            return Err(datafusion_common::DataFusionError::Execution(format!(
+                "unsupported list item data type {} for flat full text search",
+                data_type
+            )));
+        }
+    }
+
+    for i in 0..row_id_array.len() {
+        if doc_array.is_null(i) {
+            continue;
+        }
+
+        temp_query_token_counts.clear();
+        temp_query_token_counts.extend(std::iter::repeat_n(0, query_tokens.len()));
+        temp_query_token_positions.iter_mut().for_each(Vec::clear);
+
+        let elements = doc_array.value(i);
+        let mut document = String::new();
+        for element in iter_str_array(elements.as_ref()).flatten() {
+            if !document.is_empty() {
+                document.push(' ');
+            }
+            document.push_str(element);
+        }
+
+        let row_id = row_id_array.value(i);
+        let all_tokens = count_text(
+            row_id,
+            &document,
+            temp_query_token_counts,
+            Some(temp_query_token_positions),
+        )?;
+        if all_tokens > 0 {
+            append_counts(
+                row_id,
+                all_tokens,
+                temp_query_token_counts,
+                Some(phrase_matches_positions(
+                    query_tokens,
+                    query_position_groups,
+                    temp_query_token_positions,
+                    slop,
+                )),
+            );
+        }
+    }
+
+    Ok(())
+}
+
 fn query_token_indices(query_tokens: &Tokens) -> HashMap<String, Vec<usize>> {
     let mut indices = HashMap::new();
     for idx in 0..query_tokens.len() {
@@ -7409,6 +7591,7 @@ fn flat_bm25_score(
     counted_input: &RecordBatch,
     scorer: &MemBM25Scorer,
     operator: Operator,
+    require_phrase_match: bool,
 ) -> Result<RecordBatch> {
     let mut row_ids_builder = UInt64Builder::with_capacity(counted_input.num_rows());
     let mut scores_builder = Float32Builder::with_capacity(counted_input.num_rows());
@@ -7434,6 +7617,19 @@ fn flat_bm25_score(
         .values()
         .iter()
         .copied();
+    let mut phrase_matches_iter = if require_phrase_match {
+        Some(
+            counted_input
+                .column_by_name(FLAT_PHRASE_MATCH_COL)
+                .ok_or_else(|| Error::internal("flat phrase input is missing phrase_match"))?
+                .as_any()
+                .downcast_ref::<BooleanArray>()
+                .ok_or_else(|| Error::internal("flat phrase_match column is not boolean"))?
+                .iter(),
+        )
+    } else {
+        None
+    };
     for _ in 0..counted_input.num_rows() {
         let num_tokens_in_doc = all_token_counts_iter.next().expect_ok()?;
         let row_id = row_ids_iter.next().expect_ok()?;
@@ -7444,6 +7640,12 @@ fn flat_bm25_score(
         if num_tokens_in_doc == 0 {
             continue;
         }
+        if phrase_matches_iter
+            .as_mut()
+            .is_some_and(|matches| matches.next() != Some(Some(true)))
+        {
+            continue;
+        }
         if operator == Operator::And
             && !query_groups
                 .iter()
@@ -7490,6 +7692,57 @@ fn query_position_groups(query_tokens: &Tokens) -> Vec<Vec<usize>> {
     groups
 }
 
+fn phrase_matches_positions(
+    query_tokens: &Tokens,
+    query_position_groups: &[Vec<usize>],
+    token_positions: &[Vec<u32>],
+    slop: u32,
+) -> bool {
+    let mut reachable = Vec::<i64>::new();
+
+    for (group_idx, group) in query_position_groups.iter().enumerate() {
+        let Some(&first_token_idx) = group.first() else {
+            return false;
+        };
+        let query_position = i64::from(query_tokens.position(first_token_idx));
+        let mut relative_positions = group
+            .iter()
+            .flat_map(|token_idx| token_positions[*token_idx].iter())
+            .map(|position| i64::from(*position) - query_position)
+            .filter(|relative| slop > 0 || *relative >= 0)
+            .collect::<Vec<_>>();
+        relative_positions.sort_unstable();
+        relative_positions.dedup();
+        if relative_positions.is_empty() {
+            return false;
+        }
+
+        if group_idx == 0 {
+            reachable = relative_positions;
+            continue;
+        }
+
+        let slop = i64::from(slop);
+        let mut next_reachable = Vec::with_capacity(relative_positions.len());
+        let mut reachable_idx = 0;
+        for relative in relative_positions {
+            let least_previous = relative - slop;
+            while reachable_idx < reachable.len() && reachable[reachable_idx] < least_previous {
+                reachable_idx += 1;
+            }
+            if reachable_idx < reachable.len() && reachable[reachable_idx] <= relative {
+                next_reachable.push(relative);
+            }
+        }
+        if next_reachable.is_empty() {
+            return false;
+        }
+        reachable = next_reachable;
+    }
+
+    !reachable.is_empty()
+}
+
 #[deprecated(
     note = "use `flat_bm25_search_stream_with_metrics` to record CPU compute \
             time on a metric handle; pass `None` for the old behavior"
@@ -7579,6 +7832,61 @@ pub async fn flat_bm25_search_stream_with_metrics_and_operator(
     target_batch_size: usize,
     operator: Operator,
     elapsed_compute: Option<Time>,
+) -> DataFusionResult<SendableRecordBatchStream> {
+    flat_bm25_search_stream_impl(
+        input,
+        doc_col,
+        query,
+        tokenizer,
+        base_scorer,
+        target_batch_size,
+        operator,
+        None,
+        elapsed_compute,
+    )
+    .await
+}
+
+/// Flat-scan documents, retain exact phrase matches, and score them with BM25.
+///
+/// Phrase matching uses the configured analyzer's logical token positions and is
+/// applied before any caller-level limit.
+#[allow(clippy::too_many_arguments)]
+pub async fn flat_phrase_search_stream_with_metrics(
+    input: SendableRecordBatchStream,
+    doc_col: String,
+    query: String,
+    tokenizer: Box<dyn LanceTokenizer>,
+    base_scorer: Option<MemBM25Scorer>,
+    target_batch_size: usize,
+    slop: u32,
+    elapsed_compute: Option<Time>,
+) -> DataFusionResult<SendableRecordBatchStream> {
+    flat_bm25_search_stream_impl(
+        input,
+        doc_col,
+        query,
+        tokenizer,
+        base_scorer,
+        target_batch_size,
+        Operator::And,
+        Some(slop),
+        elapsed_compute,
+    )
+    .await
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn flat_bm25_search_stream_impl(
+    input: SendableRecordBatchStream,
+    doc_col: String,
+    query: String,
+    tokenizer: Box<dyn LanceTokenizer>,
+    base_scorer: Option<MemBM25Scorer>,
+    target_batch_size: usize,
+    operator: Operator,
+    phrase_slop: Option<u32>,
+    elapsed_compute: Option<Time>,
 ) -> DataFusionResult<SendableRecordBatchStream> {
     let mut tokenizer = tokenizer;
 
@@ -7626,6 +7934,7 @@ pub async fn flat_bm25_search_stream_with_metrics_and_operator(
         tokenizer,
         query_tokens.clone(),
         doc_col_idx,
+        phrase_slop,
         elapsed_compute.clone(),
     )
     .await?;
@@ -7634,7 +7943,13 @@ pub async fn flat_bm25_search_stream_with_metrics_and_operator(
     // All post-await work is synchronous; time the scorer + score + slicing loop together.
     let post_await_start = std::time::Instant::now();
     let scorer = initialize_scorer(base_scorer.as_ref(), query_tokens.as_ref(), &counted_input);
-    let scores = flat_bm25_score(query_tokens.as_ref(), &counted_input, &scorer, operator)?;
+    let scores = flat_bm25_score(
+        query_tokens.as_ref(),
+        &counted_input,
+        &scorer,
+        operator,
+        phrase_slop.is_some(),
+    )?;
 
     // Finally we emit batches according to the target batch size
     let num_out_batches = scores.num_rows().div_ceil(target_batch_size);
@@ -12735,6 +13050,7 @@ mod tests {
             query_tokens.clone(),
             1,
             None,
+            None,
         )
         .await
         .unwrap();
@@ -12861,6 +13177,129 @@ mod tests {
         assert_eq!(row_ids.values(), &[0]);
     }
 
+    async fn flat_phrase_row_ids(
+        docs: ArrayRef,
+        query: &str,
+        slop: u32,
+        tokenizer: Box<dyn LanceTokenizer>,
+    ) -> Vec<u64> {
+        let schema = Arc::new(Schema::new(vec![
+            ROW_ID_FIELD.clone(),
+            Field::new("text", docs.data_type().clone(), true),
+        ]));
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(UInt64Array::from(
+                    (0..docs.len() as u64).collect::<Vec<_>>(),
+                )) as ArrayRef,
+                docs,
+            ],
+        )
+        .unwrap();
+        let input: SendableRecordBatchStream = Box::pin(RecordBatchStreamAdapter::new(
+            schema,
+            stream::iter(vec![Ok(batch)]),
+        ));
+
+        let result_stream = flat_phrase_search_stream_with_metrics(
+            input,
+            "text".to_string(),
+            query.to_string(),
+            tokenizer,
+            None,
+            100,
+            slop,
+            None,
+        )
+        .await
+        .unwrap();
+        let batches: Vec<_> = result_stream.try_collect().await.unwrap();
+        let scored = arrow::compute::concat_batches(&FTS_SCHEMA, &batches).unwrap();
+        let mut row_ids = scored[ROW_ID]
+            .as_primitive::<UInt64Type>()
+            .values()
+            .to_vec();
+        row_ids.sort_unstable();
+        row_ids
+    }
+
+    fn simple_text_tokenizer() -> Box<dyn LanceTokenizer> {
+        Box::new(TextTokenizer::new(
+            TextAnalyzer::builder(SimpleTokenizer::default()).build(),
+        ))
+    }
+
+    #[tokio::test]
+    async fn flat_phrase_search_matches_exact_order_and_slop() {
+        let docs = Arc::new(StringArray::from(vec![
+            "alpha beta",
+            "alpha x beta",
+            "beta alpha",
+            "alpha beta beta",
+            "alpha alpha beta",
+        ])) as ArrayRef;
+
+        assert_eq!(
+            flat_phrase_row_ids(docs.clone(), "alpha beta", 0, simple_text_tokenizer()).await,
+            vec![0, 3, 4]
+        );
+        assert_eq!(
+            flat_phrase_row_ids(docs.clone(), "alpha beta", 1, simple_text_tokenizer()).await,
+            vec![0, 1, 3, 4]
+        );
+        assert_eq!(
+            flat_phrase_row_ids(docs, "alpha alpha", 0, simple_text_tokenizer()).await,
+            vec![4]
+        );
+    }
+
+    #[tokio::test]
+    async fn flat_phrase_search_respects_query_position_gaps() {
+        let docs = Arc::new(StringArray::from(vec![
+            "alpha beta",
+            "alpha quick beta",
+            "alpha the beta",
+            "beta alpha",
+        ])) as ArrayRef;
+        let tokenizer: Box<dyn LanceTokenizer> = Box::new(TextTokenizer::new(
+            TextAnalyzer::builder(SimpleTokenizer::default())
+                .filter(StopWordFilter::new(Language::English).unwrap())
+                .build(),
+        ));
+
+        assert_eq!(
+            flat_phrase_row_ids(docs, "alpha the beta", 0, tokenizer).await,
+            vec![1, 2]
+        );
+    }
+
+    #[tokio::test]
+    async fn flat_phrase_search_treats_string_lists_as_row_documents() {
+        let mut docs_builder =
+            GenericListBuilder::<i32, _>::new(GenericStringBuilder::<i32>::new());
+        docs_builder.values().append_value("alpha");
+        docs_builder.values().append_value("beta");
+        docs_builder.append(true);
+        docs_builder.values().append_value("alpha x");
+        docs_builder.values().append_value("beta");
+        docs_builder.append(true);
+        docs_builder.values().append_value("beta");
+        docs_builder.values().append_value("alpha");
+        docs_builder.append(true);
+
+        assert_eq!(
+            flat_phrase_row_ids(
+                Arc::new(docs_builder.finish()) as ArrayRef,
+                "alpha beta",
+                0,
+                simple_text_tokenizer(),
+            )
+            .await,
+            vec![0]
+        );
+    }
+
     #[tokio::test]
     async fn flat_bm25_search_code_and_uses_position_groups() {
         let schema = Arc::new(Schema::new(vec![
diff --git a/rust/lance/src/dataset/overlay.rs b/rust/lance/src/dataset/overlay.rs
index 83d4bc17938..3ed526a2b33 100644
--- a/rust/lance/src/dataset/overlay.rs
+++ b/rust/lance/src/dataset/overlay.rs
@@ -132,35 +132,8 @@ pub fn overlaid_fragments(fragments: &[Fragment]) -> HashMap<u32, &Fragment> {
         .collect()
 }
 
-/// Insert into `stale` the ids of fragments covered by `segment` whose index entries may be
-/// stale because an overlay committed after the segment was built touches a field the segment
-/// indexes. Field-aware and version-gated via [`overlay_exclusion_offsets`].
-///
-/// `overlaid_frags` holds only the fragments that actually carry overlays (rare), so the loop is
-/// `O(overlaid_frags)` rather than `O(fragments the segment covers)`.
-pub fn collect_overlay_stale_frags(
-    segment: &IndexMetadata,
-    overlaid_frags: &HashMap<u32, &Fragment>,
-    stale: &mut RoaringBitmap,
-    schema: &Schema,
-) -> Result<()> {
-    let coverage = segment.fragment_bitmap.as_ref();
-    for (&frag_id, fragment) in overlaid_frags {
-        if stale.contains(frag_id) || !covers_fragment(coverage, frag_id) {
-            continue;
-        }
-        if !stale_offsets_for_fragment(fragment, &segment.fields, segment.dataset_version, schema)?
-            .is_empty()
-        {
-            stale.insert(frag_id);
-        }
-    }
-    Ok(())
-}
-
-/// Like [`collect_overlay_stale_frags`] but with row-level granularity: instead of marking the
-/// whole fragment stale, it computes exactly which row offsets within each covered fragment are
-/// stale and accumulates them into `stale` (fragment_id → stale row offsets).
+/// Compute exactly which row offsets within each covered fragment are stale and accumulate them
+/// into `stale` (fragment_id → stale row offsets).
 ///
 /// Used by the scalar and vector paths to block only the affected rows from index results and
 /// re-evaluate only those rows on the flat path, keeping overhead proportional to the number of
@@ -1350,45 +1323,6 @@ mod tests {
         fragment
     }
 
-    #[test]
-    fn test_collect_frags_missing_bitmap_covers_all() {
-        let schema = flat_test_schema();
-        // A segment with no fragment_bitmap (legacy index predating bitmap tracking) must treat
-        // every overlaid fragment as covered so stale rows can't leak past the index unmasked.
-        let fragment = fragment_with_overlay(3, dense_overlay(vec![3], [1, 2], 9));
-        let overlaid: HashMap<u32, &Fragment> = HashMap::from([(3u32, &fragment)]);
-
-        let mut stale = RoaringBitmap::new();
-        collect_overlay_stale_frags(&segment(vec![3], 1, None), &overlaid, &mut stale, &schema)
-            .unwrap();
-        assert_eq!(stale, bitmap([3]), "missing bitmap must cover fragment 3");
-
-        // A present bitmap that excludes fragment 3 leaves it untouched.
-        let mut stale = RoaringBitmap::new();
-        collect_overlay_stale_frags(
-            &segment(vec![3], 1, Some(bitmap([0]))),
-            &overlaid,
-            &mut stale,
-            &schema,
-        )
-        .unwrap();
-        assert!(
-            stale.is_empty(),
-            "fragment absent from bitmap is not covered"
-        );
-
-        // A present bitmap that includes fragment 3 marks it stale.
-        let mut stale = RoaringBitmap::new();
-        collect_overlay_stale_frags(
-            &segment(vec![3], 1, Some(bitmap([3]))),
-            &overlaid,
-            &mut stale,
-            &schema,
-        )
-        .unwrap();
-        assert_eq!(stale, bitmap([3]));
-    }
-
     #[test]
     fn test_collect_rows_missing_bitmap_covers_all() {
         let schema = flat_test_schema();
@@ -1423,5 +1357,16 @@ mod tests {
             stale.is_empty(),
             "fragment absent from bitmap contributes no rows"
         );
+
+        // A present bitmap that includes fragment 3 contributes its stale rows.
+        let mut stale = HashMap::new();
+        collect_overlay_stale_rows_for_segment(
+            &segment(vec![3], 1, Some(bitmap([3]))),
+            &overlaid,
+            &mut stale,
+            &schema,
+        )
+        .unwrap();
+        assert_eq!(stale.get(&3), Some(&bitmap([1, 2])));
     }
 }
diff --git a/rust/lance/src/dataset/scanner.rs b/rust/lance/src/dataset/scanner.rs
index a4b0b6b864b..faa2c8fc26e 100644
--- a/rust/lance/src/dataset/scanner.rs
+++ b/rust/lance/src/dataset/scanner.rs
@@ -83,9 +83,7 @@ use tracing::{Span, info_span, instrument};
 use uuid::Uuid;
 
 use super::Dataset;
-use crate::dataset::overlay::{
-    collect_overlay_stale_frags, collect_overlay_stale_rows_for_segment, overlaid_fragments,
-};
+use crate::dataset::overlay::{collect_overlay_stale_rows_for_segment, overlaid_fragments};
 use crate::dataset::row_offsets_to_row_addresses;
 use crate::dataset::rowids::translate_addr_treemap_to_row_ids;
 use crate::dataset::utils::SchemaAdapter;
@@ -129,6 +127,15 @@ use lance_datafusion::substrait::parse_substrait;
 /// `LANCE_DEFAULT_BATCH_SIZE` specify one.
 pub const BATCH_SIZE_FALLBACK: usize = 8192;
 
+enum FtsOverlayPlan {
+    Unchanged,
+    RowLevel {
+        stale_rows: HashMap<u32, RoaringBitmap>,
+        segments: Vec<IndexMetadata>,
+    },
+    FullScan,
+}
+
 /// Parse an environment variable as a specific type, logging a warning on parse failure.
 fn parse_env_var<T: std::str::FromStr>(env_var_name: &str, default_val: &str) -> Option<T>
 where
@@ -3380,10 +3387,9 @@ impl Scanner {
                 self.fragments_covered_by_fts_query(&query).await?,
             )
             .await?;
-        // Data overlay masking: match queries drop stale segments and re-evaluate the affected
-        // fragments on the flat-text path (`plan_match_query` / `fts_stale_frags_and_fresh_segments`),
-        // and phrase queries exclude stale segments (`plan_phrase_query`). Both keep stale index
-        // hits out of the result.
+        // Data overlay masking: match and phrase queries drop stale segments and re-evaluate the
+        // affected fragments on their flat-text paths. This keeps stale index hits out while
+        // preserving current-value matches unless `fast_search` explicitly skips flat evaluation.
         let fts_exec = self
             .plan_fts(&query, &params, filter_plan, &prefilter_source)
             .await?;
@@ -3403,7 +3409,7 @@ impl Scanner {
                     .await?
             }
             FtsQuery::Phrase(query) => {
-                self.plan_phrase_query(query, params, prefilter_source)
+                self.plan_phrase_query(query, params, filter_plan, prefilter_source)
                     .await?
             }
 
@@ -3557,12 +3563,20 @@ impl Scanner {
         &self,
         query: &PhraseQuery,
         params: &FtsSearchParams,
+        filter_plan: &ExprFilterPlan,
         prefilter_source: &PreFilterSource,
     ) -> Result<Arc<dyn ExecutionPlan>> {
         let column = query.column.clone().ok_or(Error::invalid_input(
             "the column must be specified in the query".to_string(),
         ))?;
 
+        let index = self
+            .dataset
+            .load_scalar_index(IndexCriteria::default().for_column(&column).supports_fts())
+            .await?
+            .ok_or_else(|| {
+                Error::invalid_input(format!("No Inverted index found for column {}", column))
+            })?;
         let segments = load_segments(&self.dataset, &column)
             .await?
             .ok_or(Error::invalid_input(format!(
@@ -3576,40 +3590,93 @@ impl Scanner {
                 .to_string()));
         }
 
-        // Mask data overlay files: a fragment with an overlay committed after this FTS index can
-        // no longer be trusted to its inverted-index positions, so a stale phrase could still
-        // match. Exclude any segment covering such a fragment. Unlike match queries, phrase
-        // queries have no flat re-evaluation path, so — exactly as for unindexed fragments, which
-        // phrase queries already do not search — overlaid fragments are simply dropped from the
-        // phrase result; new phrase matches there surface once compaction folds the overlay into
-        // the base. This removes stale hits without introducing wrong ones.
-        let target_fragments = self
-            .fragments
-            .clone()
-            .unwrap_or_else(|| self.dataset.fragments().to_vec());
-        let (_flat_frag_ids, fresh_segments) = self
-            .fts_stale_frags_and_fresh_segments(&column, &target_fragments)
-            .await?;
+        let target_fragments = match self.fragments.as_ref() {
+            Some(fragments) if fragments.is_empty() => {
+                return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
+            }
+            Some(fragments) => fragments.clone(),
+            None => self.dataset.fragments().to_vec(),
+        };
+        let unindexed_fragments =
+            self.retain_target_fragments(self.dataset.unindexed_fragments(&index.name).await?);
+        if !target_fragments.is_empty() && unindexed_fragments.len() == target_fragments.len() {
+            if self.fast_search {
+                return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
+            }
+            let flat_plan = self
+                .plan_flat_phrase_query(
+                    target_fragments,
+                    HashMap::new(),
+                    query,
+                    params,
+                    filter_plan,
+                )
+                .await?;
+            return Self::combine_fts_plans(None, Some(flat_plan), params.limit);
+        }
 
-        let exec: Arc<dyn ExecutionPlan> = match fresh_segments {
-            // Every segment covers a stale fragment: with no flat phrase path there is nothing
-            // trustworthy left to search, so the phrase query returns no rows.
-            Some(segs) if segs.is_empty() => Arc::new(EmptyExec::new(FTS_SCHEMA.clone())),
-            Some(segs) => Arc::new(PhraseQueryExec::new_with_segments(
+        let overlay_plan = self.fts_overlay_plan(&column, &target_fragments).await?;
+        let (stale_rows, preset_segments) = match overlay_plan {
+            FtsOverlayPlan::Unchanged => (HashMap::new(), None),
+            FtsOverlayPlan::RowLevel {
+                stale_rows,
+                segments,
+            } => (stale_rows, Some(segments)),
+            FtsOverlayPlan::FullScan => {
+                if self.fast_search {
+                    return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
+                }
+                let flat_plan = self
+                    .plan_flat_phrase_query(
+                        target_fragments,
+                        HashMap::new(),
+                        query,
+                        params,
+                        filter_plan,
+                    )
+                    .await?;
+                return Self::combine_fts_plans(None, Some(flat_plan), params.limit);
+            }
+        };
+        let overlay_block = self.stale_rows_block_mask(&stale_rows).await?;
+
+        let mut indexed_exec = match preset_segments {
+            Some(segments) => PhraseQueryExec::new_with_segments(
                 self.dataset.clone(),
                 query.clone(),
                 params.clone(),
                 prefilter_source.clone(),
-                segs,
-            )),
-            None => Arc::new(PhraseQueryExec::new(
+                segments,
+            ),
+            None => PhraseQueryExec::new(
                 self.dataset.clone(),
                 query.clone(),
                 params.clone(),
                 prefilter_source.clone(),
-            )),
+            ),
         };
-        Ok(exec)
+        if let Some(overlay_block) = overlay_block {
+            indexed_exec = indexed_exec.with_overlay_block(overlay_block);
+        }
+        let indexed_plan = Some(Arc::new(indexed_exec) as Arc<dyn ExecutionPlan>);
+
+        let flat_plan =
+            if self.fast_search || (unindexed_fragments.is_empty() && stale_rows.is_empty()) {
+                None
+            } else {
+                Some(
+                    self.plan_flat_phrase_query(
+                        unindexed_fragments,
+                        stale_rows,
+                        query,
+                        params,
+                        filter_plan,
+                    )
+                    .await?,
+                )
+            };
+
+        Self::combine_fts_plans(indexed_plan, flat_plan, params.limit)
     }
 
     async fn plan_match_query(
@@ -3633,73 +3700,93 @@ impl Scanner {
             .await?;
 
         // Get target fragments
-        let target_fragments = self
-            .fragments
-            .clone()
-            .unwrap_or_else(|| self.dataset.fragments().to_vec());
+        let target_fragments = match self.fragments.as_ref() {
+            Some(fragments) if fragments.is_empty() => {
+                return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
+            }
+            Some(fragments) => fragments.clone(),
+            None => self.dataset.fragments().to_vec(),
+        };
 
         let (match_plan, flat_match_plan) = match &index {
             Some(index) => {
-                // Get unindexed fragments and filter to target fragments
                 let unindexed_fragments = self
                     .retain_target_fragments(self.dataset.unindexed_fragments(&index.name).await?);
 
-                // Fragments whose FTS index entries may be stale due to a newer data overlay.
-                // These are excluded from the indexed path and re-evaluated on the flat path.
-                let (stale_flat_frag_ids, fresh_segments) = self
-                    .fts_stale_frags_and_fresh_segments(&column, &target_fragments)
-                    .await?;
-
-                // Fragments that need flat evaluation: unindexed + stale (deduplicated).
-                let flat_fragments: Vec<Fragment> = {
-                    let mut seen = RoaringBitmap::new();
-                    let mut frags = Vec::new();
-                    for f in unindexed_fragments.iter().chain(
-                        target_fragments
-                            .iter()
-                            .filter(|f| stale_flat_frag_ids.contains(f.id as u32)),
-                    ) {
-                        if seen.insert(f.id as u32) {
-                            frags.push(f.clone());
-                        }
-                    }
-                    frags
-                };
-
-                // If all target fragments need flat evaluation, skip the indexed path.
-                if flat_fragments.len() == target_fragments.len() {
+                if !target_fragments.is_empty()
+                    && unindexed_fragments.len() == target_fragments.len()
+                {
                     if self.fast_search {
                         return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
                     }
                     let flat_match_plan = self
-                        .plan_flat_match_query(flat_fragments, query, params, filter_plan)
+                        .plan_flat_match_query(
+                            target_fragments,
+                            HashMap::new(),
+                            query,
+                            params,
+                            filter_plan,
+                        )
                         .await?;
-                    return Ok(flat_match_plan);
+                    return Self::combine_fts_plans(None, Some(flat_match_plan), params.limit);
                 }
 
-                // Build the indexed path. When overlays made some segments stale we use
-                // `new_with_segments` to restrict the search to fresh segments only.
-                let match_plan: Arc<dyn ExecutionPlan> = match fresh_segments {
-                    Some(segs) => Arc::new(MatchQueryExec::new_with_segments(
+                let overlay_plan = self.fts_overlay_plan(&column, &target_fragments).await?;
+                let (stale_rows, preset_segments) = match overlay_plan {
+                    FtsOverlayPlan::Unchanged => (HashMap::new(), None),
+                    FtsOverlayPlan::RowLevel {
+                        stale_rows,
+                        segments,
+                    } => (stale_rows, Some(segments)),
+                    FtsOverlayPlan::FullScan => {
+                        if self.fast_search {
+                            return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone())));
+                        }
+                        let flat_match_plan = self
+                            .plan_flat_match_query(
+                                target_fragments,
+                                HashMap::new(),
+                                query,
+                                params,
+                                filter_plan,
+                            )
+                            .await?;
+                        return Self::combine_fts_plans(None, Some(flat_match_plan), params.limit);
+                    }
+                };
+                let overlay_block = self.stale_rows_block_mask(&stale_rows).await?;
+
+                let mut indexed_exec = match preset_segments {
+                    Some(segments) => MatchQueryExec::new_with_segments(
                         self.dataset.clone(),
                         query.clone(),
                         params.clone(),
                         prefilter_source.clone(),
-                        segs,
-                    )),
-                    None => Arc::new(MatchQueryExec::new(
+                        segments,
+                    ),
+                    None => MatchQueryExec::new(
                         self.dataset.clone(),
                         query.clone(),
                         params.clone(),
                         prefilter_source.clone(),
-                    )),
+                    ),
                 };
+                if let Some(overlay_block) = overlay_block {
+                    indexed_exec = indexed_exec.with_overlay_block(overlay_block);
+                }
+                let match_plan = Arc::new(indexed_exec) as Arc<dyn ExecutionPlan>;
 
-                if self.fast_search || flat_fragments.is_empty() {
+                if self.fast_search || (unindexed_fragments.is_empty() && stale_rows.is_empty()) {
                     (Some(match_plan), None)
                 } else {
                     let flat_match_plan = self
-                        .plan_flat_match_query(flat_fragments, query, params, filter_plan)
+                        .plan_flat_match_query(
+                            unindexed_fragments,
+                            stale_rows,
+                            query,
+                            params,
+                            filter_plan,
+                        )
                         .await?;
                     (Some(match_plan), Some(flat_match_plan))
                 }
@@ -3710,41 +3797,56 @@ impl Scanner {
                 }
                 // No index: flat search all target fragments
                 let flat_match_plan = self
-                    .plan_flat_match_query(target_fragments.clone(), query, params, filter_plan)
+                    .plan_flat_match_query(
+                        target_fragments,
+                        HashMap::new(),
+                        query,
+                        params,
+                        filter_plan,
+                    )
                     .await?;
                 (None, Some(flat_match_plan))
             }
         };
 
-        // Combine plans
-        let plan = match (match_plan, flat_match_plan) {
-            (Some(match_plan), Some(flat_match_plan)) => {
-                let match_plan = UnionExec::try_new(vec![match_plan, flat_match_plan])?;
-                let match_plan = Arc::new(RepartitionExec::try_new(
-                    match_plan,
-                    Partitioning::RoundRobinBatch(1),
-                )?);
-                let sort_expr = PhysicalSortExpr {
-                    expr: expressions::col(SCORE_COL, match_plan.schema().as_ref())?,
-                    options: SortOptions {
-                        descending: true,
-                        nulls_first: false,
-                    },
-                };
-                Arc::new(SortExec::new([sort_expr].into(), match_plan).with_fetch(params.limit))
+        Self::combine_fts_plans(match_plan, flat_match_plan, params.limit)
+    }
+
+    fn combine_fts_plans(
+        indexed_plan: Option<Arc<dyn ExecutionPlan>>,
+        flat_plan: Option<Arc<dyn ExecutionPlan>>,
+        limit: Option<usize>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let plan = match (indexed_plan, flat_plan) {
+            (Some(indexed_plan), Some(flat_plan)) => {
+                UnionExec::try_new(vec![indexed_plan, flat_plan])?
             }
-            (Some(match_plan), None) => match_plan,
-            (None, Some(flat_match_plan)) => flat_match_plan,
-            (None, None) => unreachable!(),
+            (Some(indexed_plan), None) => return Ok(indexed_plan),
+            (None, Some(flat_plan)) if limit.is_none() => return Ok(flat_plan),
+            (None, Some(flat_plan)) => flat_plan,
+            (None, None) => return Ok(Arc::new(EmptyExec::new(FTS_SCHEMA.clone()))),
         };
-
-        Ok(plan)
+        let plan = Arc::new(RepartitionExec::try_new(
+            plan,
+            Partitioning::RoundRobinBatch(1),
+        )?);
+        let sort_expr = PhysicalSortExpr {
+            expr: expressions::col(SCORE_COL, plan.schema().as_ref())?,
+            options: SortOptions {
+                descending: true,
+                nulls_first: false,
+            },
+        };
+        Ok(Arc::new(
+            SortExec::new([sort_expr].into(), plan).with_fetch(limit),
+        ))
     }
 
     /// Plan match query on unindexed fragments
     async fn plan_flat_match_query(
         &self,
         fragments: Vec<Fragment>,
+        stale_rows: HashMap<u32, RoaringBitmap>,
         query: &MatchQuery,
         params: &FtsSearchParams,
         filter_plan: &ExprFilterPlan,
@@ -3757,9 +3859,59 @@ impl Scanner {
             ))?
             .clone();
 
-        let mut columns = vec![column.clone()];
-        if let Some(refine_expr) = filter_plan.refine_expr.as_ref() {
-            columns.extend(Planner::column_names_in_expr(refine_expr));
+        let plan = self
+            .plan_flat_fts_input(fragments, stale_rows, &column, filter_plan)
+            .await?;
+        Ok(Arc::new(FlatMatchQueryExec::new(
+            self.dataset.clone(),
+            query.clone(),
+            params.clone(),
+            plan,
+        )))
+    }
+
+    async fn plan_flat_phrase_query(
+        &self,
+        fragments: Vec<Fragment>,
+        stale_rows: HashMap<u32, RoaringBitmap>,
+        query: &PhraseQuery,
+        params: &FtsSearchParams,
+        filter_plan: &ExprFilterPlan,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let column = query
+            .column
+            .as_ref()
+            .ok_or(Error::invalid_input(
+                "the column must be specified in the query".to_string(),
+            ))?
+            .clone();
+
+        let plan = self
+            .plan_flat_fts_input(fragments, stale_rows, &column, filter_plan)
+            .await?;
+        Ok(Arc::new(FlatMatchQueryExec::new_phrase(
+            self.dataset.clone(),
+            query.clone(),
+            params.clone(),
+            plan,
+        )))
+    }
+
+    async fn plan_flat_fts_input(
+        &self,
+        fragments: Vec<Fragment>,
+        stale_rows: HashMap<u32, RoaringBitmap>,
+        column: &str,
+        filter_plan: &ExprFilterPlan,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        let mut columns = vec![column.to_string()];
+        let filter_expr = if stale_rows.is_empty() {
+            filter_plan.refine_expr.as_ref()
+        } else {
+            filter_plan.full_expr.as_ref()
+        };
+        if let Some(filter_expr) = filter_expr {
+            columns.extend(Planner::column_names_in_expr(filter_expr));
         }
         let scan_projection = self
             .dataset
@@ -3767,29 +3919,42 @@ impl Scanner {
             .with_row_id()
             .union_columns(&columns, OnMissing::Error)?;
 
-        let PlannedFilteredScan { mut plan, .. } = self
-            .filtered_read(
-                filter_plan,
-                scan_projection,
-                /*make_deletions_null=*/ false,
-                Some(Arc::new(fragments)),
-                None,
-                /*is_prefilter=*/ true,
-            )
-            .await?;
+        let mut inputs = Vec::with_capacity(2);
+        if !fragments.is_empty() {
+            let PlannedFilteredScan { mut plan, .. } = self
+                .filtered_read(
+                    filter_plan,
+                    scan_projection.clone(),
+                    /*make_deletions_null=*/ false,
+                    Some(Arc::new(fragments)),
+                    None,
+                    /*is_prefilter=*/ true,
+                )
+                .await?;
 
-        if let Some(refine_expr) = filter_plan.refine_expr.as_ref() {
-            plan = Arc::new(LanceFilterExec::try_new(refine_expr.clone(), plan)?);
+            if let Some(refine_expr) = filter_plan.refine_expr.as_ref() {
+                plan = Arc::new(LanceFilterExec::try_new(refine_expr.clone(), plan)?);
+            }
+            inputs.push(self.ensure_column_alias(plan, column)?);
         }
-        plan = self.ensure_column_alias(plan, &column)?;
 
-        let flat_match_plan = Arc::new(FlatMatchQueryExec::new(
-            self.dataset.clone(),
-            query.clone(),
-            params.clone(),
-            plan,
-        ));
-        Ok(flat_match_plan)
+        if !stale_rows.is_empty() {
+            let mut plan = self.stale_rows_take(&stale_rows, scan_projection).await?;
+            if let Some(filter) = filter_plan.full_expr.as_ref() {
+                let planner = Planner::new(plan.schema());
+                let filter = planner.optimize_expr(filter.clone())?;
+                plan = Arc::new(LanceFilterExec::try_new(filter, plan)?);
+            }
+            inputs.push(self.ensure_column_alias(plan, column)?);
+        }
+
+        match inputs.len() {
+            0 => Err(Error::internal(
+                "flat FTS input requires unindexed fragments or stale rows",
+            )),
+            1 => Ok(inputs.pop().unwrap()),
+            _ => Ok(UnionExec::try_new(inputs)?),
+        }
     }
 
     // ANN/KNN search execution node with optional prefilter
@@ -4388,69 +4553,56 @@ impl Scanner {
         Ok(stale)
     }
 
-    /// Compute which FTS segments are stale due to data overlay files committed after the
-    /// index was built, and which fragments must therefore fall back to the flat text path.
-    ///
-    /// Returns `(flat_frag_ids, Some(fresh_segments))` when overlays are present:
-    /// - `flat_frag_ids`: fragment IDs that must be scanned flat (stale fragments, plus any
-    ///   other fragments co-located in a segment that covers a stale one — the whole segment is
-    ///   excluded, so all fragments it covered must move to flat).
-    /// - `fresh_segments`: the subset of FTS segments that cover no stale fragment; safe to
-    ///   pass to `MatchQueryExec::new_with_segments`.
+    /// Plan FTS overlay handling at row granularity.
     ///
-    /// Returns `(empty, None)` on the fast path (no overlays, or no segments load).
-    async fn fts_stale_frags_and_fresh_segments(
+    /// Modern segments keep using their index while overlay-stale rows are blocked and
+    /// re-evaluated from current values. A legacy segment without fragment coverage falls back to
+    /// a full target scan when a relevant overlay exists because its indexed row set is unknown.
+    async fn fts_overlay_plan(
         &self,
         column: &str,
         target_fragments: &[Fragment],
-    ) -> Result<(RoaringBitmap, Option<Vec<IndexMetadata>>)> {
-        // Fast path: no overlays on any target fragment.
+    ) -> Result<FtsOverlayPlan> {
         if target_fragments.iter().all(|f| f.overlays.is_empty()) {
-            return Ok((RoaringBitmap::new(), None));
+            return Ok(FtsOverlayPlan::Unchanged);
         }
 
         let Some(segments) = load_segments(&self.dataset, column).await? else {
-            return Ok((RoaringBitmap::new(), None));
+            return Ok(FtsOverlayPlan::Unchanged);
         };
 
         let overlaid_frags = overlaid_fragments(target_fragments);
-        let mut stale_frag_ids = RoaringBitmap::new();
-        for seg in &segments {
-            collect_overlay_stale_frags(
-                seg,
-                &overlaid_frags,
-                &mut stale_frag_ids,
-                self.dataset.schema(),
-            )?;
-        }
-
-        if stale_frag_ids.is_empty() {
-            // Overlays exist but none are on this FTS column or predate the index.
-            return Ok((stale_frag_ids, None));
-        }
-
-        // Any segment covering a stale fragment is excluded from the indexed path.
-        // All fragments covered by that segment (stale + co-located fresh ones) must
-        // fall to the flat path, since the indexed path no longer covers them.
-        let mut flat_frag_ids = stale_frag_ids.clone();
-        let mut fresh_segments = Vec::with_capacity(segments.len());
-        for seg in segments {
-            match &seg.fragment_bitmap {
-                Some(bm) if !bm.is_disjoint(&stale_frag_ids) => {
-                    flat_frag_ids |= bm;
-                    // exclude this segment from the indexed path
-                }
-                Some(_) => fresh_segments.push(seg),
-                None => {
-                    // Coverage unknown (legacy segment without a fragment bitmap): we can neither
-                    // trust it to exclude overlay-stale rows nor tell which fragments it indexes.
-                    // Exclude it from the indexed path and route every target fragment to flat.
-                    flat_frag_ids.extend(target_fragments.iter().map(|f| f.id as u32));
+        let mut stale_rows = HashMap::new();
+        for segment in &segments {
+            if segment.fragment_bitmap.is_none() {
+                let mut legacy_stale_rows = HashMap::new();
+                collect_overlay_stale_rows_for_segment(
+                    segment,
+                    &overlaid_frags,
+                    &mut legacy_stale_rows,
+                    self.dataset.schema(),
+                )?;
+                if !legacy_stale_rows.is_empty() {
+                    return Ok(FtsOverlayPlan::FullScan);
                 }
+            } else {
+                collect_overlay_stale_rows_for_segment(
+                    segment,
+                    &overlaid_frags,
+                    &mut stale_rows,
+                    self.dataset.schema(),
+                )?;
             }
         }
 
-        Ok((flat_frag_ids, Some(fresh_segments)))
+        if stale_rows.is_empty() {
+            Ok(FtsOverlayPlan::Unchanged)
+        } else {
+            Ok(FtsOverlayPlan::RowLevel {
+                stale_rows,
+                segments,
+            })
+        }
     }
 
     /// Collect the stale rows into a [`RowAddrTreeMap`] in the domain the index results use.
diff --git a/rust/lance/src/dataset/tests/dataset_overlay_index_masking.rs b/rust/lance/src/dataset/tests/dataset_overlay_index_masking.rs
index cee96d37fc7..18a8a77ef9a 100644
--- a/rust/lance/src/dataset/tests/dataset_overlay_index_masking.rs
+++ b/rust/lance/src/dataset/tests/dataset_overlay_index_masking.rs
@@ -19,6 +19,7 @@ use lance_index::scalar::BuiltinIndexType;
 use lance_index::scalar::FullTextSearchQuery;
 use lance_index::scalar::ScalarIndexParams;
 use lance_index::scalar::inverted::InvertedIndexParams;
+use lance_index::scalar::inverted::query::{FtsQuery, PhraseQuery};
 use lance_io::utils::CachedFileSize;
 use lance_linalg::distance::MetricType;
 use lance_table::format::DataFile;
@@ -31,9 +32,11 @@ use lance_file::writer::{FileWriter, FileWriterOptions};
 use crate::Dataset;
 use crate::dataset::optimize::{CompactionOptions, compact_files, remapping};
 use crate::dataset::transaction::{DataOverlayGroup, Operation};
-use crate::dataset::{WriteDestination, WriteParams};
+use crate::dataset::{WriteDestination, WriteMode, WriteParams};
 use crate::index::vector::VectorIndexParams;
 use crate::index::{CreateIndexBuilder, DatasetIndexExt};
+use crate::io::exec::filtered_read::FilteredReadExec;
+use crate::io::exec::fts::FlatMatchQueryExec;
 
 /// Two-fragment Int32 dataset: `id` (field 0) = 0..12 and `age` (field 1) = id * 10,
 /// six rows per file (fragments 0 and 1). In-memory store so overlay files can be written
@@ -594,7 +597,7 @@ async fn test_overlay_stale_with_compound_index_expression() {
 
 /// Text dataset: two fragments, 6 rows each. Schema: id (Int32), text (Utf8).
 /// Texts are unique tokens so each row can be identified by its term.
-async fn create_text_dataset() -> Dataset {
+async fn create_text_dataset(stable_row_ids: bool) -> Dataset {
     let schema = Arc::new(ArrowSchema::new(vec![
         ArrowField::new("id", DataType::Int32, true),
         ArrowField::new("text", DataType::Utf8, true),
@@ -623,6 +626,7 @@ async fn create_text_dataset() -> Dataset {
     .unwrap();
     let write_params = WriteParams {
         max_rows_per_file: 6,
+        enable_stable_row_ids: stable_row_ids,
         ..Default::default()
     };
     let reader = RecordBatchIterator::new(vec![Ok(batch)], schema.clone());
@@ -683,7 +687,7 @@ async fn fts_ids_matching(dataset: &Dataset, term: &str) -> Vec<i32> {
 
 #[tokio::test]
 async fn test_ngram_optimize_preserves_overlay_staleness() {
-    let mut dataset = create_text_dataset().await;
+    let mut dataset = create_text_dataset(false).await;
     let params = ScalarIndexParams::for_builtin(BuiltinIndexType::NGram);
     let fragment_ids = dataset
         .get_fragments()
@@ -776,7 +780,7 @@ async fn test_btree_physical_merge_preserves_overlay_staleness() {
 
 #[tokio::test]
 async fn test_ngram_remap_excludes_newer_overlay_fragments() {
-    let mut dataset = create_text_dataset().await;
+    let mut dataset = create_text_dataset(false).await;
     let params = ScalarIndexParams::for_builtin(BuiltinIndexType::NGram);
     dataset
         .create_index(
@@ -838,8 +842,6 @@ async fn test_ngram_remap_excludes_newer_overlay_fragments() {
 }
 
 async fn fts_phrase_ids_matching(dataset: &Dataset, phrase: &str) -> Vec<i32> {
-    use lance_index::scalar::inverted::query::{FtsQuery, PhraseQuery};
-
     let query = FullTextSearchQuery::new_query(FtsQuery::Phrase(
         PhraseQuery::new(phrase.to_owned()).with_column(Some("text".to_owned())),
     ));
@@ -848,9 +850,10 @@ async fn fts_phrase_ids_matching(dataset: &Dataset, phrase: &str) -> Vec<i32> {
 
 /// An overlay committed after the FTS index is built replaces a row's text. Searching for
 /// the old term must not return the stale row; searching for the new term must find it.
+#[rstest]
 #[tokio::test]
-async fn test_fts_overlay_stale_drop_and_new_match() {
-    let mut dataset = create_text_dataset().await;
+async fn test_fts_overlay_stale_drop_and_new_match(#[values(false, true)] stable_row_ids: bool) {
+    let mut dataset = create_text_dataset(stable_row_ids).await;
     build_text_fts_index(&mut dataset).await;
 
     // fragment 0, row offset 1 (id=1): "apple banana" → "cherry mango"
@@ -894,13 +897,14 @@ async fn test_fts_overlay_stale_drop_and_new_match() {
     );
 }
 
-/// A phrase query must not return a stale hit for an overlaid FTS-indexed row. Phrase queries
-/// have no flat re-evaluation path, so the fragment is excluded from the indexed phrase search
-/// (like an unindexed fragment) rather than re-scored — the point of this test is that the
-/// pre-overlay phrase hit is dropped, not that the new value is found.
+/// Phrase queries re-evaluate overlay-stale segments from current values: stale hits disappear,
+/// new phrase matches surface, and unaffected rows co-located in the excluded segment remain.
+#[rstest]
 #[tokio::test]
-async fn test_fts_phrase_overlay_stale_drop() {
-    let mut dataset = create_text_dataset().await;
+async fn test_fts_phrase_overlay_stale_drop_and_new_match(
+    #[values(false, true)] stable_row_ids: bool,
+) {
+    let mut dataset = create_text_dataset(stable_row_ids).await;
     build_text_fts_index_with_positions(&mut dataset).await;
 
     // Before any overlay the phrase "apple banana" matches only id=1.
@@ -925,12 +929,296 @@ async fn test_fts_phrase_overlay_stale_drop() {
         fts_phrase_ids_matching(&dataset, "apple banana").await,
         Vec::<i32>::new()
     );
+    assert_eq!(
+        fts_phrase_ids_matching(&dataset, "cherry mango").await,
+        vec![1]
+    );
+    assert_eq!(
+        fts_phrase_ids_matching(&dataset, "apple pie").await,
+        vec![0]
+    );
+
+    let mut filtered_scan = dataset.scan();
+    filtered_scan
+        .full_text_search(FullTextSearchQuery::new_query(FtsQuery::Phrase(
+            PhraseQuery::new("cherry mango".to_owned()).with_column(Some("text".to_owned())),
+        )))
+        .unwrap();
+    filtered_scan.filter("id = 2").unwrap();
+    filtered_scan.project(&["id"]).unwrap();
+    assert_eq!(filtered_scan.try_into_batch().await.unwrap().num_rows(), 0);
+}
+
+#[tokio::test]
+async fn test_fts_empty_fragment_selection_is_empty() {
+    let mut dataset = create_text_dataset(false).await;
+    build_text_fts_index_with_positions(&mut dataset).await;
+
+    let mut match_scan = dataset.scan();
+    match_scan.with_fragments(Vec::new());
+    match_scan
+        .full_text_search(FullTextSearchQuery::new("apple".to_owned()))
+        .unwrap();
+    match_scan.project(&["id"]).unwrap();
+    let match_plan = match_scan.explain_plan(false).await.unwrap();
+    assert!(
+        match_plan.contains("EmptyExec"),
+        "explicit empty fragment selection should produce EmptyExec: {match_plan}"
+    );
+    assert_eq!(match_scan.try_into_batch().await.unwrap().num_rows(), 0);
+
+    let mut phrase_scan = dataset.scan();
+    phrase_scan.with_fragments(Vec::new());
+    phrase_scan
+        .full_text_search(FullTextSearchQuery::new_query(FtsQuery::Phrase(
+            PhraseQuery::new("apple pie".to_owned()).with_column(Some("text".to_owned())),
+        )))
+        .unwrap();
+    phrase_scan.project(&["id"]).unwrap();
+    let phrase_plan = phrase_scan.explain_plan(false).await.unwrap();
+    assert!(
+        phrase_plan.contains("EmptyExec"),
+        "explicit empty fragment selection should produce EmptyExec: {phrase_plan}"
+    );
+    assert_eq!(phrase_scan.try_into_batch().await.unwrap().num_rows(), 0);
+}
+
+#[rstest]
+#[tokio::test]
+async fn test_fts_combines_indexed_overlay_stale_and_unindexed_rows(
+    #[values(false, true)] stable_row_ids: bool,
+) {
+    let mut dataset = create_text_dataset(stable_row_ids).await;
+    build_text_fts_index_with_positions(&mut dataset).await;
+
+    let batch =
+        arrow_array::record_batch!(("id", Int32, [12]), ("text", Utf8, ["cherry mango"])).unwrap();
+    let schema = batch.schema();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+    let dataset = Dataset::write(
+        reader,
+        Arc::new(dataset),
+        Some(WriteParams {
+            mode: WriteMode::Append,
+            ..Default::default()
+        }),
+    )
+    .await
+    .unwrap();
+
+    let dataset = commit_overlay(
+        dataset,
+        "fts_combined_flat_paths",
+        0,
+        &[1],
+        OverlayCoverage::dense(RoaringBitmap::from_iter([1])),
+        vec![Arc::new(StringArray::from(vec![Some("cherry mango")]))],
+    )
+    .await;
+
+    assert_eq!(fts_ids_matching(&dataset, "mango").await, vec![1, 6, 12]);
+    assert_eq!(
+        fts_phrase_ids_matching(&dataset, "cherry mango").await,
+        vec![1, 12]
+    );
+}
+
+#[tokio::test]
+async fn test_fts_overlay_row_level_masking_under_fast_search() {
+    let mut dataset = create_text_dataset(false).await;
+    build_text_fts_index_with_positions(&mut dataset).await;
+
+    let dataset = commit_overlay(
+        dataset,
+        "fts_row_level_fast_search",
+        0,
+        &[1],
+        OverlayCoverage::dense(RoaringBitmap::from_iter([1])),
+        vec![Arc::new(StringArray::from(vec![Some("cherry mango")]))],
+    )
+    .await;
+
+    let mut match_scan = dataset.scan();
+    match_scan
+        .full_text_search(FullTextSearchQuery::new("apple".to_owned()))
+        .unwrap();
+    match_scan.project(&["id"]).unwrap();
+    match_scan.fast_search();
+    let match_result = match_scan.try_into_batch().await.unwrap();
+    assert_eq!(
+        ids_from_batches(std::slice::from_ref(&match_result)),
+        vec![0]
+    );
+
+    let mut indexed_phrase_scan = dataset.scan();
+    indexed_phrase_scan
+        .full_text_search(FullTextSearchQuery::new_query(FtsQuery::Phrase(
+            PhraseQuery::new("apple pie".to_owned()).with_column(Some("text".to_owned())),
+        )))
+        .unwrap();
+    indexed_phrase_scan.project(&["id"]).unwrap();
+    indexed_phrase_scan.fast_search();
+    let indexed_phrase_result = indexed_phrase_scan.try_into_batch().await.unwrap();
+    assert_eq!(
+        ids_from_batches(std::slice::from_ref(&indexed_phrase_result)),
+        vec![0]
+    );
+
+    let mut new_phrase_scan = dataset.scan();
+    new_phrase_scan
+        .full_text_search(FullTextSearchQuery::new_query(FtsQuery::Phrase(
+            PhraseQuery::new("cherry mango".to_owned()).with_column(Some("text".to_owned())),
+        )))
+        .unwrap();
+    new_phrase_scan.project(&["id"]).unwrap();
+    new_phrase_scan.fast_search();
+    assert_eq!(
+        new_phrase_scan.try_into_batch().await.unwrap().num_rows(),
+        0
+    );
+}
+
+#[tokio::test]
+async fn test_fts_overlay_flat_path_takes_only_stale_rows() {
+    let mut dataset = create_text_dataset(false).await;
+    build_text_fts_index(&mut dataset).await;
+
+    let dataset = commit_overlay(
+        dataset,
+        "fts_targeted_take",
+        0,
+        &[1],
+        OverlayCoverage::dense(RoaringBitmap::from_iter([1])),
+        vec![Arc::new(StringArray::from(vec![Some("cherry mango")]))],
+    )
+    .await;
+
+    let mut scan = dataset.scan();
+    scan.full_text_search(FullTextSearchQuery::new("cherry".to_owned()))
+        .unwrap();
+    scan.project(&["id"]).unwrap();
+    let plan = scan.create_plan().await.unwrap();
+
+    let mut nodes = vec![plan];
+    let mut flat_path_uses_targeted_take = false;
+    while let Some(node) = nodes.pop() {
+        if node.downcast_ref::<FlatMatchQueryExec>().is_some() {
+            let mut flat_nodes = node.children().into_iter().cloned().collect::<Vec<_>>();
+            while let Some(flat_node) = flat_nodes.pop() {
+                if flat_node
+                    .downcast_ref::<FilteredReadExec>()
+                    .is_some_and(|read| read.index_input().is_some())
+                {
+                    flat_path_uses_targeted_take = true;
+                    break;
+                }
+                flat_nodes.extend(flat_node.children().into_iter().cloned());
+            }
+        }
+        nodes.extend(node.children().into_iter().cloned());
+    }
+
+    assert!(
+        flat_path_uses_targeted_take,
+        "overlay-stale FTS rows must be re-evaluated through a targeted take"
+    );
+}
+
+#[tokio::test]
+async fn test_fts_phrase_searches_unindexed_fragments_unless_fast_search() {
+    let mut dataset = create_text_dataset(false).await;
+    build_text_fts_index_with_positions(&mut dataset).await;
+
+    let batch = arrow_array::record_batch!(
+        ("id", Int32, [12, 13, 14]),
+        (
+            "text",
+            Utf8,
+            [
+                "kiwi berry",
+                "kiwi berry filling",
+                "kiwi berry filling filling"
+            ]
+        )
+    )
+    .unwrap();
+    let schema = batch.schema();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+    let dataset = Dataset::write(
+        reader,
+        Arc::new(dataset),
+        Some(WriteParams {
+            mode: WriteMode::Append,
+            ..Default::default()
+        }),
+    )
+    .await
+    .unwrap();
+
+    let query = FullTextSearchQuery::new_query(FtsQuery::Phrase(
+        PhraseQuery::new("kiwi berry".to_owned()).with_column(Some("text".to_owned())),
+    ));
+    assert_eq!(fts_ids(&dataset, query.clone()).await, vec![12, 13, 14]);
+
+    let mut limited_scan = dataset.scan();
+    limited_scan.with_fragments(vec![dataset.fragments().last().unwrap().clone()]);
+    limited_scan
+        .full_text_search(query.clone().limit(Some(1)))
+        .unwrap();
+    limited_scan.project(&["id"]).unwrap();
+    let limited_result = limited_scan.try_into_batch().await.unwrap();
+    assert_eq!(
+        ids_from_batches(std::slice::from_ref(&limited_result)),
+        vec![12]
+    );
+
+    let mut filtered_scan = dataset.scan();
+    filtered_scan.full_text_search(query.clone()).unwrap();
+    filtered_scan.project(&["id"]).unwrap();
+    filtered_scan.filter("id < 12").unwrap();
+    assert_eq!(filtered_scan.try_into_batch().await.unwrap().num_rows(), 0);
+
+    let mut fast_scan = dataset.scan();
+    fast_scan.full_text_search(query).unwrap();
+    fast_scan.project(&["id"]).unwrap();
+    fast_scan.fast_search();
+    assert_eq!(fast_scan.try_into_batch().await.unwrap().num_rows(), 0);
+}
+
+#[tokio::test]
+async fn test_fts_phrase_stale_fragment_honors_query_limit() {
+    let mut dataset = create_text_dataset(false).await;
+    build_text_fts_index_with_positions(&mut dataset).await;
+
+    let dataset = commit_overlay(
+        dataset,
+        "phrase_limit_overlay",
+        0,
+        &[1],
+        OverlayCoverage::dense(RoaringBitmap::from_iter([1])),
+        vec![Arc::new(StringArray::from(vec![Some(
+            "apple banana filling filling",
+        )]))],
+    )
+    .await;
+
+    let query = FullTextSearchQuery::new_query(FtsQuery::Phrase(
+        PhraseQuery::new("apple".to_owned()).with_column(Some("text".to_owned())),
+    ))
+    .limit(Some(1));
+    let mut scan = dataset.scan();
+    scan.with_fragments(vec![dataset.fragments()[0].clone()]);
+    scan.full_text_search(query).unwrap();
+    scan.project(&["id"]).unwrap();
+
+    let result = scan.try_into_batch().await.unwrap();
+    assert_eq!(ids_from_batches(std::slice::from_ref(&result)), vec![0]);
 }
 
 /// An overlay on a non-FTS field must not exclude the fragment from phrase search.
 #[tokio::test]
 async fn test_fts_phrase_overlay_unrelated_field_not_excluded() {
-    let mut dataset = create_text_dataset().await;
+    let mut dataset = create_text_dataset(false).await;
     build_text_fts_index_with_positions(&mut dataset).await;
 
     // Overlay field 0 (`id`), not the FTS-indexed `text` column: phrase coverage is untouched.
@@ -953,7 +1241,7 @@ async fn test_fts_phrase_overlay_unrelated_field_not_excluded() {
 /// An overlay on a field the FTS index does NOT cover must not exclude anything.
 #[tokio::test]
 async fn test_fts_overlay_unrelated_field_not_excluded() {
-    let mut dataset = create_text_dataset().await;
+    let mut dataset = create_text_dataset(false).await;
     build_text_fts_index(&mut dataset).await;
 
     // Overlay field 0 (id) — not covered by the FTS index on `text`.
@@ -976,7 +1264,8 @@ async fn test_fts_overlay_unrelated_field_not_excluded() {
 
 /// Benchmark: measure query latency for BTree, FTS, and vector ANN with 0/4/16 overlay layers.
 ///
-/// Run with: cargo test -p lance --lib --release -- overlay_index_masking::bench --ignored --nocapture
+/// Run with:
+/// cargo test -p lance --lib --profile release-with-debug -- overlay_index_masking::bench --ignored --nocapture
 #[tokio::test]
 #[ignore = "benchmark"]
 #[allow(clippy::print_stdout)]
@@ -997,8 +1286,8 @@ async fn bench_index_query_overlay_overhead() {
     }
 
     // --- Build 1M-row dataset on local disk --------------------------------
-    // Schema: id(0), age(1), vec(2) — 3 top-level fields.
-    // Lance field IDs (depth-first): id=0, age=1, vec=2, vec.item=3.
+    // Schema: id, age, vec, text. Resolve field IDs from the Lance schema instead of
+    // assuming how nested Arrow child fields are numbered.
 
     println!("Building {ROWS}-row dataset at {uri} (this takes ~30 s)...");
 
@@ -1013,6 +1302,7 @@ async fn bench_index_query_overlay_overhead() {
             ),
             false,
         ),
+        ArrowField::new("text", DataType::Utf8, false),
     ]));
 
     let row_ids: Vec<i32> = (0..ROWS).collect();
@@ -1030,6 +1320,9 @@ async fn bench_index_query_overlay_overhead() {
         )
         .unwrap(),
     );
+    let text_col = Arc::new(StringArray::from_iter_values(
+        (0..ROWS).map(|row| if row == 42 { "needle" } else { "common" }),
+    ));
 
     let batch = RecordBatch::try_new(
         schema.clone(),
@@ -1037,6 +1330,7 @@ async fn bench_index_query_overlay_overhead() {
             Arc::new(Int32Array::from(row_ids)),
             Arc::new(Int32Array::from(ages)),
             vec_col,
+            text_col,
         ],
     )
     .unwrap();
@@ -1049,6 +1343,7 @@ async fn bench_index_query_overlay_overhead() {
     let mut dataset = Dataset::write(reader, uri, Some(write_params))
         .await
         .unwrap();
+    let text_field_id = dataset.schema().field_id("text").unwrap();
 
     println!("Building BTree index on age...");
     dataset
@@ -1074,6 +1369,18 @@ async fn bench_index_query_overlay_overhead() {
         .await
         .unwrap();
 
+    println!("Building FTS index on text...");
+    dataset
+        .create_index(
+            &["text"],
+            IndexType::Inverted,
+            None,
+            &InvertedIndexParams::default(),
+            true,
+        )
+        .await
+        .unwrap();
+
     println!("Indexes built.\n");
 
     // --- Timing helper ---------------------------------------------------
@@ -1093,20 +1400,18 @@ async fn bench_index_query_overlay_overhead() {
 
     // === Scenario A: BTree query overhead ================================
     //
-    // Overlay on `age` (field 1), covering only offset 0 of fragment 0.
-    // Fragment granularity: the entire fragment 0 (100k rows) falls to flat-scan.
+    // Overlay on `age` (field 1), covering only offset 0 of fragment 0. The stale row is
+    // blocked from the BTree and re-evaluated by targeted take.
     //
-    // btree_cold: `age = 420` → id=42 → in fragment 0 (rows 0..99999).
-    //   With overlays: 100k-row flat scan + per-overlay merge instead of index lookup.
-    //   Without overlays: O(log n) BTree lookup.
+    // btree_same_fragment: `age = 420` → id=42 → in fragment 0 (rows 0..99999).
+    //   The matching row stays indexed even though another row in the fragment is stale.
     //
-    // btree_warm: `age = 1000420` → id=100042 → in fragment 1 (rows 100000..199999).
-    //   Always served by the BTree index regardless of overlay count on fragment 0.
-    //   This isolates the index-lookup baseline.
-    println!("=== Scenario A: BTree (overlay on `age`, fragment 0 becomes stale) ===");
+    // btree_other_fragment: `age = 1000420` → id=100042 → in fragment 1.
+    //   This isolates the index-lookup baseline outside the overlaid fragment.
+    println!("=== Scenario A: BTree (one stale row in fragment 0) ===");
     println!(
         "{:>10}  {:>14}  {:>14}",
-        "overlays", "cold_frag0_ms", "warm_frag1_ms"
+        "overlays", "same_frag_ms", "other_frag_ms"
     );
 
     let mut committed_a = 0u32;
@@ -1127,9 +1432,9 @@ async fn bench_index_query_overlay_overhead() {
 
         let ds = Arc::new(dataset.clone());
 
-        // Cold path: stale fragment falls to flat scan when overlays > 0.
+        // Same-fragment indexed match plus targeted re-evaluation of the stale row.
         let ds2 = ds.clone();
-        let cold_ms = timeit(ITERS, || {
+        let same_fragment_ms = timeit(ITERS, || {
             let ds = ds2.clone();
             async move {
                 ds.scan()
@@ -1144,9 +1449,9 @@ async fn bench_index_query_overlay_overhead() {
         })
         .await;
 
-        // Warm path: fragment 1 never stale, always index-served.
+        // Fragment 1 never has a stale row and stays entirely index-served.
         let ds2 = ds.clone();
-        let warm_ms = timeit(ITERS, || {
+        let other_fragment_ms = timeit(ITERS, || {
             let ds = ds2.clone();
             async move {
                 ds.scan()
@@ -1161,7 +1466,7 @@ async fn bench_index_query_overlay_overhead() {
         })
         .await;
 
-        println!("{num_overlays:>10}  {cold_ms:>14.1}  {warm_ms:>14.1}");
+        println!("{num_overlays:>10}  {same_fragment_ms:>14.1}  {other_fragment_ms:>14.1}");
     }
 
     // === Scenario B: Vector ANN overhead =================================
@@ -1171,9 +1476,8 @@ async fn bench_index_query_overlay_overhead() {
     // the vector index (they touch field 1, not field 2). Only a vec overlay (field 2)
     // marks fragment 0 stale for the vector index.
     //
-    // With a vec overlay: 100k rows of fragment 0 are excluded from ANN prefilter
-    // bitmaps and re-scored brute-force (O(100k × DIM) distance computations).
-    println!("\n=== Scenario B: Vector ANN (overlay on `vec`, 100k rows brute-forced) ===");
+    // With a vec overlay, only the stale row is excluded from ANN and re-scored exactly.
+    println!("\n=== Scenario B: Vector ANN (one stale row re-scored) ===");
     println!("{:>12}  {:>10}", "vec_overlays", "ann_ms");
 
     let query_vec = Float32Array::from(vec![0.5f32; DIM as usize]);
@@ -1213,4 +1517,47 @@ async fn bench_index_query_overlay_overhead() {
 
         println!("{num_vec_overlays:>12}  {ann_ms:>10.1}");
     }
+
+    // === Scenario C: FTS overhead ========================================
+    //
+    // The FTS index has one segment spanning all 10 fragments. An overlay on one text row must
+    // keep that segment indexed, block just the stale row, and re-evaluate that row by targeted
+    // take. `needle` belongs to an unaffected row in the same fragment as the stale row.
+    println!("\n=== Scenario C: FTS (one stale row in a 1M-row segment) ===");
+    println!("{:>13}  {:>10}", "text_overlays", "fts_ms");
+
+    for num_text_overlays in [0u32, 1] {
+        if num_text_overlays == 1 {
+            dataset = commit_overlay(
+                dataset,
+                "text_ol0",
+                0, // fragment 0
+                &[text_field_id],
+                OverlayCoverage::dense(RoaringBitmap::from_iter([0u32])),
+                vec![Arc::new(StringArray::from(vec![Some("updated")]))],
+            )
+            .await;
+        }
+
+        let ds = Arc::new(dataset.clone());
+        let ds2 = ds.clone();
+        let fts_ms = timeit(ITERS, || {
+            let ds = ds2.clone();
+            async move {
+                let result = ds
+                    .scan()
+                    .full_text_search(FullTextSearchQuery::new("needle".to_owned()))
+                    .unwrap()
+                    .project(&["id"])
+                    .unwrap()
+                    .try_into_batch()
+                    .await
+                    .unwrap();
+                assert_eq!(result.num_rows(), 1);
+            }
+        })
+        .await;
+
+        println!("{num_text_overlays:>13}  {fts_ms:>10.1}");
+    }
 }
diff --git a/rust/lance/src/io/exec/fts.rs b/rust/lance/src/io/exec/fts.rs
index 7d66a3633a3..8d7f81e09fd 100644
--- a/rust/lance/src/io/exec/fts.rs
+++ b/rust/lance/src/io/exec/fts.rs
@@ -51,9 +51,10 @@ use lance_index::scalar::inverted::query::{
 use lance_index::scalar::inverted::tokenizer::document_tokenizer::TextTokenizer;
 use lance_index::scalar::inverted::{
     FTS_SCHEMA, InvertedIndex, MemBM25Scorer, SCORE_COL, build_global_bm25_scorer,
-    flat_bm25_search_stream_with_metrics_and_operator,
+    flat_bm25_search_stream_with_metrics_and_operator, flat_phrase_search_stream_with_metrics,
 };
 use lance_index::{prefilter::PreFilter, scalar::inverted::query::BooleanQuery};
+use lance_select::RowAddrMask;
 use lance_tokenizer::{SimpleTokenizer, TextAnalyzer};
 use tracing::instrument;
 use uuid::Uuid;
@@ -346,6 +347,8 @@ pub struct MatchQueryExec {
     /// scorer down to `InvertedIndex::bm25_search`.
     base_scorer: Option<Arc<MemBM25Scorer>>,
     segment_selection: FtsSegmentSelection,
+    /// Rows whose indexed values were superseded by newer data overlays.
+    overlay_block: Option<RowAddrMask>,
 
     properties: Arc<PlanProperties>,
     metrics: ExecutionPlanMetricsSet,
@@ -404,6 +407,7 @@ impl MatchQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::AllCommitted,
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -439,6 +443,7 @@ impl MatchQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::ExactResolved(Arc::from(segments)),
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -472,6 +477,7 @@ impl MatchQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::exact_uuids(segment_uuids),
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -494,6 +500,12 @@ impl MatchQueryExec {
         self
     }
 
+    /// Exclude rows whose indexed text was superseded by a newer data overlay.
+    pub fn with_overlay_block(mut self, overlay_block: RowAddrMask) -> Self {
+        self.overlay_block = Some(overlay_block);
+        self
+    }
+
     pub fn query(&self) -> &MatchQuery {
         &self.query
     }
@@ -568,6 +580,7 @@ impl ExecutionPlan for MatchQueryExec {
                     prefilter_source: PreFilterSource::None,
                     base_scorer: self.base_scorer.clone(),
                     segment_selection: self.segment_selection.clone(),
+                    overlay_block: self.overlay_block.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -595,6 +608,7 @@ impl ExecutionPlan for MatchQueryExec {
                     prefilter_source,
                     base_scorer: self.base_scorer.clone(),
                     segment_selection: self.segment_selection.clone(),
+                    overlay_block: self.overlay_block.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -620,6 +634,7 @@ impl ExecutionPlan for MatchQueryExec {
         let prefilter_source = self.prefilter_source.clone();
         let preset_base_scorer = self.base_scorer.clone();
         let segment_selection = self.segment_selection.clone();
+        let overlay_block = self.overlay_block.clone();
         let metrics = Arc::new(FtsIndexMetrics::new(&self.metrics, partition));
         let column = query.column.ok_or(DataFusionError::Execution(format!(
             "column not set for MatchQuery {}",
@@ -634,8 +649,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,
+                overlay_block,
+            )?;
             let deleted_fragments =
                 indices
                     .iter()
@@ -1026,6 +1047,7 @@ impl ExecutionPlan for FlatMatchFilterExec {
 pub struct FlatMatchQueryExec {
     dataset: Arc<Dataset>,
     query: MatchQuery,
+    phrase_slop: Option<u32>,
     params: FtsSearchParams,
     unindexed_input: Arc<dyn ExecutionPlan>,
     /// Optional override for the BM25 scorer normally built locally inside
@@ -1041,11 +1063,17 @@ pub struct FlatMatchQueryExec {
 
 impl DisplayAs for FlatMatchQueryExec {
     fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        let query_name = if self.phrase_slop.is_some() {
+            "FlatPhraseQuery"
+        } else {
+            "FlatMatchQuery"
+        };
         match t {
             DisplayFormatType::Default | DisplayFormatType::Verbose => {
                 write!(
                     f,
-                    "FlatMatchQuery: column={}, query={}",
+                    "{}: column={}, query={}",
+                    query_name,
                     self.query.column.as_deref().unwrap_or_default(),
                     self.query.terms
                 )
@@ -1053,7 +1081,8 @@ impl DisplayAs for FlatMatchQueryExec {
             DisplayFormatType::TreeRender => {
                 write!(
                     f,
-                    "FlatMatchQuery\ncolumn={}\nquery={}",
+                    "{}\ncolumn={}\nquery={}",
+                    query_name,
                     self.query.column.as_deref().unwrap_or_default(),
                     self.query.terms
                 )
@@ -1078,6 +1107,7 @@ impl FlatMatchQueryExec {
         Self {
             dataset,
             query,
+            phrase_slop: None,
             params,
             unindexed_input,
             base_scorer: None,
@@ -1104,6 +1134,7 @@ impl FlatMatchQueryExec {
         Self {
             dataset,
             query,
+            phrase_slop: None,
             params,
             unindexed_input,
             base_scorer: None,
@@ -1113,6 +1144,23 @@ impl FlatMatchQueryExec {
         }
     }
 
+    /// Create a flat executor that requires the query tokens to match in phrase order.
+    pub fn new_phrase(
+        dataset: Arc<Dataset>,
+        query: PhraseQuery,
+        mut params: FtsSearchParams,
+        unindexed_input: Arc<dyn ExecutionPlan>,
+    ) -> Self {
+        let slop = query.slop;
+        let match_query = MatchQuery::new(query.terms)
+            .with_column(query.column)
+            .with_operator(lance_index::scalar::inverted::query::Operator::And);
+        params = params.with_phrase_slop(Some(slop));
+        let mut exec = Self::new(dataset, match_query, params, unindexed_input);
+        exec.phrase_slop = Some(slop);
+        exec
+    }
+
     /// Override the local BM25 scorer; see [`MatchQueryExec::with_base_scorer`].
     pub fn with_base_scorer(mut self, scorer: Arc<MemBM25Scorer>) -> Self {
         self.base_scorer = Some(scorer);
@@ -1142,7 +1190,11 @@ impl FlatMatchQueryExec {
 
 impl ExecutionPlan for FlatMatchQueryExec {
     fn name(&self) -> &str {
-        "FlatMatchQueryExec"
+        if self.phrase_slop.is_some() {
+            "FlatPhraseQueryExec"
+        } else {
+            "FlatMatchQueryExec"
+        }
     }
 
     fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
@@ -1170,6 +1222,7 @@ impl ExecutionPlan for FlatMatchQueryExec {
         Ok(Arc::new(Self {
             dataset: self.dataset.clone(),
             query: self.query.clone(),
+            phrase_slop: self.phrase_slop,
             params: self.params.clone(),
             unindexed_input,
             base_scorer: self.base_scorer.clone(),
@@ -1179,13 +1232,14 @@ impl ExecutionPlan for FlatMatchQueryExec {
         }))
     }
 
-    #[instrument(name = "flat_match_query_exec", level = "debug", skip_all)]
+    #[instrument(name = "flat_fts_query_exec", level = "debug", skip_all)]
     fn execute(
         &self,
         partition: usize,
         context: Arc<datafusion::execution::TaskContext>,
     ) -> DataFusionResult<SendableRecordBatchStream> {
         let query = self.query.clone();
+        let phrase_slop = self.phrase_slop;
         let ds = self.dataset.clone();
         let preset_base_scorer = self.base_scorer.clone();
         let preset_segments = self.preset_segments.clone();
@@ -1249,17 +1303,34 @@ impl ExecutionPlan for FlatMatchQueryExec {
                 ),
             };
 
-            flat_bm25_search_stream_with_metrics_and_operator(
-                unindexed_input,
-                column,
-                query.terms,
-                tokenizer,
-                base_scorer,
-                target_batch_size,
-                query.operator,
-                Some(elapsed_compute),
-            )
-            .await
+            match phrase_slop {
+                Some(slop) => {
+                    flat_phrase_search_stream_with_metrics(
+                        unindexed_input,
+                        column,
+                        query.terms,
+                        tokenizer,
+                        base_scorer,
+                        target_batch_size,
+                        slop,
+                        Some(elapsed_compute),
+                    )
+                    .await
+                }
+                None => {
+                    flat_bm25_search_stream_with_metrics_and_operator(
+                        unindexed_input,
+                        column,
+                        query.terms,
+                        tokenizer,
+                        base_scorer,
+                        target_batch_size,
+                        query.operator,
+                        Some(elapsed_compute),
+                    )
+                    .await
+                }
+            }
         })
         .try_flatten()
         .map(move |batch| {
@@ -1303,6 +1374,8 @@ pub struct PhraseQueryExec {
     /// `execute()`. See [`MatchQueryExec::with_base_scorer`].
     base_scorer: Option<Arc<MemBM25Scorer>>,
     segment_selection: FtsSegmentSelection,
+    /// Rows whose indexed values were superseded by newer data overlays.
+    overlay_block: Option<RowAddrMask>,
     properties: Arc<PlanProperties>,
     metrics: ExecutionPlanMetricsSet,
 }
@@ -1352,6 +1425,7 @@ impl PhraseQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::AllCommitted,
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -1380,6 +1454,7 @@ impl PhraseQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::ExactResolved(Arc::from(segments)),
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -1414,6 +1489,7 @@ impl PhraseQueryExec {
             prefilter_source,
             base_scorer: None,
             segment_selection: FtsSegmentSelection::exact_uuids(segment_uuids),
+            overlay_block: None,
             properties,
             metrics: ExecutionPlanMetricsSet::new(),
         }
@@ -1425,6 +1501,12 @@ impl PhraseQueryExec {
         self
     }
 
+    /// Exclude rows whose indexed text was superseded by a newer data overlay.
+    pub fn with_overlay_block(mut self, overlay_block: RowAddrMask) -> Self {
+        self.overlay_block = Some(overlay_block);
+        self
+    }
+
     pub fn query(&self) -> &PhraseQuery {
         &self.query
     }
@@ -1492,6 +1574,7 @@ impl ExecutionPlan for PhraseQueryExec {
                 prefilter_source: PreFilterSource::None,
                 base_scorer: self.base_scorer.clone(),
                 segment_selection: self.segment_selection.clone(),
+                overlay_block: self.overlay_block.clone(),
                 properties: self.properties.clone(),
                 metrics: ExecutionPlanMetricsSet::new(),
             },
@@ -1517,6 +1600,7 @@ impl ExecutionPlan for PhraseQueryExec {
                     prefilter_source,
                     base_scorer: self.base_scorer.clone(),
                     segment_selection: self.segment_selection.clone(),
+                    overlay_block: self.overlay_block.clone(),
                     properties: self.properties.clone(),
                     metrics: ExecutionPlanMetricsSet::new(),
                 }
@@ -1542,6 +1626,7 @@ impl ExecutionPlan for PhraseQueryExec {
         let prefilter_source = self.prefilter_source.clone();
         let preset_base_scorer = self.base_scorer.clone();
         let segment_selection = self.segment_selection.clone();
+        let overlay_block = self.overlay_block.clone();
         let metrics = Arc::new(FtsIndexMetrics::new(&self.metrics, partition));
         let stream = stream::once(async move {
             let _timer = metrics.baseline_metrics.elapsed_compute().timer();
@@ -1556,8 +1641,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,
+                overlay_block,
+            )?;
             let deleted_fragments =
                 indices
                     .iter()
diff --git a/rust/lance/src/io/exec/utils.rs b/rust/lance/src/io/exec/utils.rs
index f4e4b99512f..44092f75459 100644
--- a/rust/lance/src/io/exec/utils.rs
+++ b/rust/lance/src/io/exec/utils.rs
@@ -52,6 +52,7 @@ pub(crate) fn build_prefilter(
     prefilter_source: &PreFilterSource,
     ds: Arc<Dataset>,
     index_meta: &[IndexMetadata],
+    overlay_block: Option<RowAddrMask>,
 ) -> Result<Arc<DatasetPreFilter>> {
     let prefilter_loader = match &prefilter_source {
         PreFilterSource::FilteredRowIds(src_node) => {
@@ -64,11 +65,11 @@ pub(crate) fn build_prefilter(
         }
         PreFilterSource::None => None,
     };
-    Ok(Arc::new(DatasetPreFilter::new(
-        ds,
-        index_meta,
-        prefilter_loader,
-    )))
+    let mut prefilter = DatasetPreFilter::new(ds, index_meta, prefilter_loader);
+    if let Some(overlay_block) = overlay_block {
+        prefilter = prefilter.with_overlay_block(overlay_block);
+    }
+    Ok(Arc::new(prefilter))
 }
 
 // Utility to convert an input (containing row ids) into a prefilter