diff --git a/Cargo.toml b/Cargo.toml
index 8bc90fe5..db8f3dff 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -23,7 +23,7 @@ arrow-select = "58"
 arrow-cast = { version = "58", features = ["prettyprint"] }
 arrow-ord = "58"
 
-datafusion = { version = "53", default-features = false, features = ["nested_expressions"] }
+datafusion = { version = "53", default-features = false, features = ["nested_expressions", "string_expressions"] }
 datafusion-physical-plan = "53"
 datafusion-physical-expr = "53"
 datafusion-execution = "53"
diff --git a/crates/omnigraph-compiler/src/ir/lower.rs b/crates/omnigraph-compiler/src/ir/lower.rs
index 1d5ca5de..ccec6d5e 100644
--- a/crates/omnigraph-compiler/src/ir/lower.rs
+++ b/crates/omnigraph-compiler/src/ir/lower.rs
@@ -4,7 +4,7 @@ use crate::catalog::Catalog;
 use crate::error::Result;
 use crate::query::ast::*;
 use crate::query::typecheck::TypeContext;
-use crate::types::Direction;
+use crate::types::{Direction, PropType, ScalarType};
 
 use super::*;
 
@@ -19,6 +19,15 @@ pub fn lower_query(
         ));
     }
     let param_names: HashSet<String> = query.params.iter().map(|p| p.name.clone()).collect();
+    // Param types were validated during typecheck; unknown names simply
+    // don't participate in `contains` overload resolution below.
+    let param_types: HashMap<String, PropType> = query
+        .params
+        .iter()
+        .filter_map(|p| {
+            PropType::from_param_type_name(&p.type_name, p.nullable).map(|t| (p.name.clone(), t))
+        })
+        .collect();
 
     let mut pipeline = Vec::new();
     let mut bound_vars = HashSet::new();
@@ -30,6 +39,7 @@ pub fn lower_query(
         &mut pipeline,
         &mut bound_vars,
         &param_names,
+        &param_types,
     )?;
 
     let return_exprs: Vec<IRProjection> = query
@@ -131,6 +141,7 @@ fn lower_clauses(
     pipeline: &mut Vec<IROp>,
     bound_vars: &mut HashSet<String>,
     param_names: &HashSet<String>,
+    param_types: &HashMap<String, PropType>,
 ) -> Result<()> {
     // Separate clause types for ordering: bindings first, then traversals, then filters
     let mut bindings = Vec::new();
@@ -371,11 +382,28 @@ fn lower_clauses(
         remaining = next_remaining;
     }
 
+    // Clause-local variable types for filter-op resolution: negation inners
+    // are typechecked into a discarded context clone (same asymmetry the
+    // `direction` fallback above documents), so `type_ctx` alone cannot
+    // resolve variables introduced inside `not { }`. Bindings declare their
+    // type; traversal endpoints take the edge's declared endpoint types
+    // (bindings win when both name a variable).
+    let mut local_var_types: HashMap<&str, &str> = HashMap::new();
+    for t in &traversals {
+        if let Some(edge) = catalog.lookup_edge_by_name(&t.edge_name) {
+            local_var_types.entry(t.src.as_str()).or_insert(&edge.from_type);
+            local_var_types.entry(t.dst.as_str()).or_insert(&edge.to_type);
+        }
+    }
+    for b in &bindings {
+        local_var_types.insert(b.variable.as_str(), b.type_name.as_str());
+    }
+
     // Lower explicit filters
     for filter in &filters {
         pipeline.push(IROp::Filter(IRFilter {
             left: lower_expr(&filter.left, param_names),
-            op: filter.op,
+            op: resolve_filter_op(catalog, type_ctx, param_types, &local_var_types, filter),
             right: lower_expr(&filter.right, param_names),
         }));
     }
@@ -394,6 +422,7 @@ fn lower_clauses(
             &mut inner_pipeline,
             &mut inner_bound,
             param_names,
+            param_types,
         )?;
 
         pipeline.push(IROp::AntiJoin {
@@ -405,6 +434,50 @@ fn lower_clauses(
     Ok(())
 }
 
+/// Resolve the overloaded `contains` keyword to its String-substring form
+/// (`StringContains`) when the left operand is a scalar String, so execution
+/// dispatches on the IR op alone and never re-derives operand types.
+///
+/// Variable types come from `local_var_types` (this clause list's bindings +
+/// traversal endpoints) first, then the outer `TypeContext` — negation
+/// inners never reach the outer context, while outer variables referenced
+/// inside a negation only exist there.
+fn resolve_filter_op(
+    catalog: &Catalog,
+    type_ctx: &TypeContext,
+    param_types: &HashMap<String, PropType>,
+    local_var_types: &HashMap<&str, &str>,
+    filter: &Filter,
+) -> CompOp {
+    if filter.op != CompOp::Contains {
+        return filter.op;
+    }
+    let left_is_scalar_string = match &filter.left {
+        Expr::PropAccess { variable, property } => local_var_types
+            .get(variable.as_str())
+            .copied()
+            .or_else(|| {
+                type_ctx
+                    .bindings
+                    .get(variable)
+                    .map(|bv| bv.type_name.as_str())
+            })
+            .and_then(|type_name| catalog.node_types.get(type_name))
+            .and_then(|nt| nt.properties.get(property))
+            .is_some_and(|p| !p.list && matches!(p.scalar, ScalarType::String)),
+        Expr::Literal(Literal::String(_)) => true,
+        Expr::Variable(v) => param_types
+            .get(v)
+            .is_some_and(|t| !t.list && matches!(t.scalar, ScalarType::String)),
+        _ => false,
+    };
+    if left_is_scalar_string {
+        CompOp::StringContains
+    } else {
+        CompOp::Contains
+    }
+}
+
 /// Build IR filters from a binding's inline property matches.
 fn build_binding_filters(
     binding: &Binding,
diff --git a/crates/omnigraph-compiler/src/ir/lower_tests.rs b/crates/omnigraph-compiler/src/ir/lower_tests.rs
index de5f69ef..05fa07f1 100644
--- a/crates/omnigraph-compiler/src/ir/lower_tests.rs
+++ b/crates/omnigraph-compiler/src/ir/lower_tests.rs
@@ -39,6 +39,47 @@ return { $f.name, $f.age }
     assert_eq!(ir.return_exprs.len(), 2);
 }
 
+#[test]
+fn test_lower_resolves_contains_overload_by_left_operand_type() {
+    // `contains` on a scalar String left operand lowers to StringContains
+    // (substring); on a list left operand it stays Contains (membership).
+    let schema = parse_schema("node Person { name: String  tags: [String]? }").unwrap();
+    let catalog = build_catalog(&schema).unwrap();
+    let qf = parse_query(
+        r#"
+query q($q: String) {
+match {
+    $p: Person
+    $p.name contains $q
+    $p.tags contains $q
+    $p.name starts_with $q
+}
+return { $p.name }
+}
+"#,
+    )
+    .unwrap();
+    let tc = typecheck_query(&catalog, &qf.queries[0]).unwrap();
+    let ir = lower_query(&catalog, &qf.queries[0], &tc).unwrap();
+
+    let filter_ops: Vec<CompOp> = ir
+        .pipeline
+        .iter()
+        .filter_map(|op| match op {
+            IROp::Filter(f) => Some(f.op),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        filter_ops,
+        vec![
+            CompOp::StringContains,
+            CompOp::Contains,
+            CompOp::StartsWith
+        ]
+    );
+}
+
 #[test]
 fn test_lower_undirected_traversal_to_direction_both() {
     let catalog = setup();
diff --git a/crates/omnigraph-compiler/src/query/ast.rs b/crates/omnigraph-compiler/src/query/ast.rs
index 802f7f4a..5ece2107 100644
--- a/crates/omnigraph-compiler/src/query/ast.rs
+++ b/crates/omnigraph-compiler/src/query/ast.rs
@@ -80,7 +80,16 @@ pub enum CompOp {
     Lt,
     Ge,
     Le,
+    /// The parser emits `Contains` for the `contains` keyword; typecheck
+    /// accepts a list left operand (membership) or a scalar String left
+    /// operand (substring), and lowering resolves the String form to
+    /// `StringContains` so downstream consumers never re-derive types.
     Contains,
+    /// Exact, case-sensitive prefix match on a scalar String property.
+    StartsWith,
+    /// Exact, case-sensitive substring match on a scalar String property.
+    /// Never produced by the parser — lowering resolves it from `Contains`.
+    StringContains,
 }
 
 impl std::fmt::Display for CompOp {
@@ -92,7 +101,8 @@ impl std::fmt::Display for CompOp {
             Self::Lt => write!(f, "<"),
             Self::Ge => write!(f, ">="),
             Self::Le => write!(f, "<="),
-            Self::Contains => write!(f, "contains"),
+            Self::Contains | Self::StringContains => write!(f, "contains"),
+            Self::StartsWith => write!(f, "starts_with"),
         }
     }
 }
diff --git a/crates/omnigraph-compiler/src/query/parser.rs b/crates/omnigraph-compiler/src/query/parser.rs
index 3330cea6..e21587ef 100644
--- a/crates/omnigraph-compiler/src/query/parser.rs
+++ b/crates/omnigraph-compiler/src/query/parser.rs
@@ -675,6 +675,7 @@ fn parse_comp_op(pair: pest::iterators::Pair<Rule>) -> Result<CompOp> {
 fn parse_filter_op(pair: pest::iterators::Pair<Rule>) -> Result<CompOp> {
     match pair.as_str() {
         "contains" => Ok(CompOp::Contains),
+        "starts_with" => Ok(CompOp::StartsWith),
         _ => parse_comp_op(pair),
     }
 }
diff --git a/crates/omnigraph-compiler/src/query/parser_tests.rs b/crates/omnigraph-compiler/src/query/parser_tests.rs
index 45a0832c..3379615a 100644
--- a/crates/omnigraph-compiler/src/query/parser_tests.rs
+++ b/crates/omnigraph-compiler/src/query/parser_tests.rs
@@ -458,6 +458,42 @@ delete Person where tags contains $tag
     assert!(parse_query(input).is_err());
 }
 
+#[test]
+fn test_parse_starts_with_filter() {
+    let input = r#"
+query autocomplete($q: String) {
+match {
+    $p: Person
+    $p.name starts_with $q
+}
+return { $p.name }
+}
+"#;
+    let qf = parse_query(input).unwrap();
+    let q = &qf.queries[0];
+    match &q.match_clause[1] {
+        Clause::Filter(f) => {
+            assert_eq!(f.op, CompOp::StartsWith);
+            assert!(matches!(
+                &f.left,
+                Expr::PropAccess { variable, property } if variable == "p" && property == "name"
+            ));
+            assert!(matches!(&f.right, Expr::Variable(v) if v == "q"));
+        }
+        _ => panic!("expected Filter"),
+    }
+}
+
+#[test]
+fn test_parse_starts_with_is_rejected_in_mutation_predicate() {
+    let input = r#"
+query drop_person($q: String) {
+delete Person where name starts_with $q
+}
+"#;
+    assert!(parse_query(input).is_err());
+}
+
 #[test]
 fn test_parse_triangle() {
     let input = r#"
diff --git a/crates/omnigraph-compiler/src/query/query.pest b/crates/omnigraph-compiler/src/query/query.pest
index 8818838d..ab18654d 100644
--- a/crates/omnigraph-compiler/src/query/query.pest
+++ b/crates/omnigraph-compiler/src/query/query.pest
@@ -93,7 +93,7 @@ agg_call = { agg_func ~ "(" ~ expr ~ ")" }
 agg_func = { "count" | "sum" | "avg" | "min" | "max" }
 
 comp_op = { ">=" | "<=" | "!=" | ">" | "<" | "=" }
-filter_op = { "contains" | comp_op }
+filter_op = { "starts_with" | "contains" | comp_op }
 
 // Terminals
 variable = @{ "$" ~ (ident_chars | "_") }
diff --git a/crates/omnigraph-compiler/src/query/typecheck.rs b/crates/omnigraph-compiler/src/query/typecheck.rs
index 91db57bb..58558366 100644
--- a/crates/omnigraph-compiler/src/query/typecheck.rs
+++ b/crates/omnigraph-compiler/src/query/typecheck.rs
@@ -879,9 +879,21 @@ fn typecheck_filter(
 
     if let (ResolvedType::Scalar(l), ResolvedType::Scalar(r)) = (&left_type, &right_type) {
         if filter.op == CompOp::Contains {
+            // Overloaded on the left operand: list → membership, scalar
+            // String → exact substring. Lowering resolves the String form to
+            // `StringContains` so execution never re-derives the dispatch.
+            if !l.list && matches!(l.scalar, ScalarType::String) {
+                if r.list || !matches!(r.scalar, ScalarType::String) {
+                    return Err(CompilerError::Type(format!(
+                        "T7: string contains requires a String right operand, got {}",
+                        r.display_name()
+                    )));
+                }
+                return Ok(());
+            }
             if !l.list {
                 return Err(CompilerError::Type(format!(
-                    "T7: contains requires a list property on the left, got {}",
+                    "T7: contains requires a list property (membership) or a String property (substring) on the left, got {}",
                     l.display_name()
                 )));
             }
@@ -909,6 +921,28 @@ fn typecheck_filter(
             return Ok(());
         }
 
+        if matches!(filter.op, CompOp::StartsWith | CompOp::StringContains) {
+            // Exact, case-sensitive string predicates: scalar String on both
+            // sides. (`StringContains` only exists post-lowering, but the
+            // check is written over both ops so re-typechecking IR-shaped
+            // input stays consistent.)
+            if l.list || !matches!(l.scalar, ScalarType::String) {
+                return Err(CompilerError::Type(format!(
+                    "T7: {} requires a String property on the left, got {}",
+                    filter.op,
+                    l.display_name()
+                )));
+            }
+            if r.list || !matches!(r.scalar, ScalarType::String) {
+                return Err(CompilerError::Type(format!(
+                    "T7: {} requires a String right operand, got {}",
+                    filter.op,
+                    r.display_name()
+                )));
+            }
+            return Ok(());
+        }
+
         // T7: check type compatibility
         if l.list || r.list {
             return Err(CompilerError::Type(
diff --git a/crates/omnigraph-compiler/src/query/typecheck_tests.rs b/crates/omnigraph-compiler/src/query/typecheck_tests.rs
index 63a0095a..ed7ba7da 100644
--- a/crates/omnigraph-compiler/src/query/typecheck_tests.rs
+++ b/crates/omnigraph-compiler/src/query/typecheck_tests.rs
@@ -217,7 +217,9 @@ return { $p.tags, $tags, $days }
 }
 
 #[test]
-fn test_contains_filter_requires_list_left_operand() {
+fn test_contains_filter_accepts_string_substring_overload() {
+    // A scalar String left operand resolves the overload to exact substring
+    // matching (previously a T7 error, so no existing query changes meaning).
     let catalog = setup();
     let qf = parse_query(
         r#"
@@ -228,13 +230,114 @@ match {
 }
 return { $p.name }
 }
+"#,
+    )
+    .unwrap();
+    assert!(typecheck_query(&catalog, &qf.queries[0]).is_ok());
+}
+
+#[test]
+fn test_string_contains_requires_string_right_operand() {
+    let catalog = setup();
+    let qf = parse_query(
+        r#"
+query q() {
+match {
+    $p: Person
+    $p.name contains 42
+}
+return { $p.name }
+}
+"#,
+    )
+    .unwrap();
+    let err = typecheck_query(&catalog, &qf.queries[0]).unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("string contains requires a String right operand")
+    );
+}
+
+#[test]
+fn test_contains_filter_requires_list_or_string_left_operand() {
+    let catalog = setup();
+    let qf = parse_query(
+        r#"
+query q() {
+match {
+    $p: Person
+    $p.age contains 3
+}
+return { $p.name }
+}
+"#,
+    )
+    .unwrap();
+    let err = typecheck_query(&catalog, &qf.queries[0]).unwrap_err();
+    assert!(err.to_string().contains(
+        "contains requires a list property (membership) or a String property (substring)"
+    ));
+}
+
+#[test]
+fn test_starts_with_accepts_string_operands() {
+    let catalog = setup();
+    let qf = parse_query(
+        r#"
+query q($q: String) {
+match {
+    $p: Person
+    $p.name starts_with $q
+}
+return { $p.name }
+}
+"#,
+    )
+    .unwrap();
+    assert!(typecheck_query(&catalog, &qf.queries[0]).is_ok());
+}
+
+#[test]
+fn test_starts_with_rejects_non_string_left_operand() {
+    let catalog = setup();
+    let qf = parse_query(
+        r#"
+query q() {
+match {
+    $p: Person
+    $p.age starts_with "4"
+}
+return { $p.name }
+}
+"#,
+    )
+    .unwrap();
+    let err = typecheck_query(&catalog, &qf.queries[0]).unwrap_err();
+    assert!(
+        err.to_string()
+            .contains("starts_with requires a String property on the left")
+    );
+}
+
+#[test]
+fn test_starts_with_rejects_non_string_right_operand() {
+    let catalog = setup();
+    let qf = parse_query(
+        r#"
+query q() {
+match {
+    $p: Person
+    $p.name starts_with 4
+}
+return { $p.name }
+}
 "#,
     )
     .unwrap();
     let err = typecheck_query(&catalog, &qf.queries[0]).unwrap_err();
     assert!(
         err.to_string()
-            .contains("contains requires a list property on the left")
+            .contains("starts_with requires a String right operand")
     );
 }
 
diff --git a/crates/omnigraph/src/db/omnigraph/table_ops.rs b/crates/omnigraph/src/db/omnigraph/table_ops.rs
index 123459b0..4a4bed82 100644
--- a/crates/omnigraph/src/db/omnigraph/table_ops.rs
+++ b/crates/omnigraph/src/db/omnigraph/table_ops.rs
@@ -645,9 +645,9 @@ pub(super) struct IndexWorkStatus {
 /// Per `build_indices_on_dataset_for_catalog`, nodes get BTree (id) plus, for
 /// each one-column `@index`/`@key` property, the index `node_prop_index_kind`
 /// assigns: a scalar BTREE for enums and orderable scalars
-/// (DateTime/Date/numeric/Bool), FTS for free-text Strings, or a Vector index.
-/// Edges get BTree only (id, src, dst). This helper and the builder share
-/// `node_prop_index_kind` so they cannot drift — see its doc comment.
+/// (DateTime/Date/numeric/Bool), FTS for free-text Strings, or a Vector
+/// index. Edges get BTree only (id, src, dst). This helper and the builder
+/// share `node_prop_index_kind` so they cannot drift — see its doc comment.
 #[derive(Default)]
 struct PlannedIndexWork {
     specs: Vec<crate::storage_layer::IndexBuildSpec>,
@@ -689,6 +689,7 @@ async fn plan_index_work_node(
     if !db.storage().has_btree_index(ds, "id").await? {
         work.push_spec(crate::storage_layer::IndexBuildSpec::BTree {
             column: "id".to_string(),
+            name: None,
         });
     }
     let Some(node_type) = catalog.node_types.get(type_name) else {
@@ -709,6 +710,18 @@ async fn plan_index_work_node(
                         column: prop_name.clone(),
                     });
                 }
+                // DEFERRED: a companion BTREE here would index-accelerate
+                // equality and `starts_with` on free-text Strings (Lance
+                // never consults an inverted index for either), and the
+                // staged machinery supports it (`IndexBuildSpec::BTree`
+                // explicit names; the same-column batch test). It is held
+                // back because Lance's second-generation shallow clones
+                // cannot read parent index files at all — every indexed read
+                // through a branch-of-a-branch fork hard-errors
+                // (lance-format/lance#7840), and the companion would widen
+                // that exposure to every `@key` equality lookup. Re-land when
+                // `lance_surface_guards::second_generation_branch_index_reads_fail_upstream`
+                // turns red (its panic message carries the checklist).
             }
             Some(NodePropIndexKind::Vector) => {
                 if !db.storage().has_vector_index(ds, prop_name).await? {
@@ -729,6 +742,7 @@ async fn plan_index_work_node(
                 if !db.storage().has_btree_index(ds, prop_name).await? {
                     work.push_spec(crate::storage_layer::IndexBuildSpec::BTree {
                         column: prop_name.clone(),
+                        name: None,
                     });
                 }
             }
@@ -773,6 +787,7 @@ async fn plan_index_work_edge_on_dataset(
         if !db.storage().has_btree_index(ds, column).await? {
             work.push_spec(crate::storage_layer::IndexBuildSpec::BTree {
                 column: column.to_string(),
+                name: None,
             });
         }
     }
diff --git a/crates/omnigraph/src/exec/mutation.rs b/crates/omnigraph/src/exec/mutation.rs
index e3706c5a..bac270a1 100644
--- a/crates/omnigraph/src/exec/mutation.rs
+++ b/crates/omnigraph/src/exec/mutation.rs
@@ -415,11 +415,18 @@ fn predicate_to_sql(
         CompOp::Lt => "<",
         CompOp::Ge => ">=",
         CompOp::Le => "<=",
-        CompOp::Contains => {
+        // The mutation grammar only admits comparison ops; these arms are
+        // defense in depth for IR built outside the parser.
+        CompOp::Contains | CompOp::StringContains => {
             return Err(OmniError::manifest(
                 "contains predicate not supported in mutations".to_string(),
             ));
         }
+        CompOp::StartsWith => {
+            return Err(OmniError::manifest(
+                "starts_with predicate not supported in mutations".to_string(),
+            ));
+        }
     };
 
     // #283: emit the column UNQUOTED. Lance's `Scanner::filter(&str)` (the
diff --git a/crates/omnigraph/src/exec/projection.rs b/crates/omnigraph/src/exec/projection.rs
index bb6e665f..28ebe5b3 100644
--- a/crates/omnigraph/src/exec/projection.rs
+++ b/crates/omnigraph/src/exec/projection.rs
@@ -5,6 +5,7 @@ pub(super) fn apply_filter(
     filter: &IRFilter,
     params: &ParamMap,
 ) -> Result<()> {
+    crate::instrumentation::record_in_memory_filter();
     let mask = evaluate_filter(batch, filter, params)?;
     let filtered = arrow_select::filter::filter_record_batch(batch, &mask)
         .map_err(|e| OmniError::Lance(e.to_string()))?;
@@ -24,6 +25,9 @@ fn evaluate_filter(
     if filter.op == CompOp::Contains {
         return evaluate_contains_filter(&left, &right);
     }
+    if matches!(filter.op, CompOp::StartsWith | CompOp::StringContains) {
+        return evaluate_string_match_filter(filter.op, &left, &right);
+    }
 
     // Cast right to match left's type if needed (e.g. Int64 literal vs Int32 column)
     let right = if left.data_type() != right.data_type() {
@@ -41,7 +45,9 @@ fn evaluate_filter(
         CompOp::Lt => cmp::lt(&left, &right),
         CompOp::Ge => cmp::gt_eq(&left, &right),
         CompOp::Le => cmp::lt_eq(&left, &right),
-        CompOp::Contains => unreachable!("handled above"),
+        CompOp::Contains | CompOp::StartsWith | CompOp::StringContains => {
+            unreachable!("handled above")
+        }
     }
     .map_err(|e| OmniError::Lance(e.to_string()))?;
 
@@ -134,6 +140,45 @@ fn evaluate_contains_filter(left: &ArrayRef, right: &ArrayRef) -> Result<Boolean
     Ok(BooleanArray::from(values))
 }
 
+/// Exact, case-sensitive string predicates (`starts_with` and the String
+/// overload of `contains`). NULL on either side is not a match, matching the
+/// pushdown arm's SQL semantics.
+fn evaluate_string_match_filter(
+    op: CompOp,
+    left: &ArrayRef,
+    right: &ArrayRef,
+) -> Result<BooleanArray> {
+    let left_str = left
+        .as_any()
+        .downcast_ref::<StringArray>()
+        .ok_or_else(|| OmniError::manifest(format!("{op} requires String operands")))?;
+    let right = if right.data_type() != left.data_type() {
+        arrow_cast::cast::cast(right, left.data_type())
+            .map_err(|e| OmniError::Lance(e.to_string()))?
+    } else {
+        Arc::clone(right)
+    };
+    let right_str = right
+        .as_any()
+        .downcast_ref::<StringArray>()
+        .ok_or_else(|| OmniError::manifest(format!("{op} requires String operands")))?;
+
+    let mut values = Vec::with_capacity(left_str.len());
+    for row in 0..left_str.len() {
+        if left_str.is_null(row) || right_str.is_null(row) {
+            values.push(Some(false));
+            continue;
+        }
+        let (l, r) = (left_str.value(row), right_str.value(row));
+        let matched = match op {
+            CompOp::StartsWith => l.starts_with(r),
+            _ => l.contains(r),
+        };
+        values.push(Some(matched));
+    }
+    Ok(BooleanArray::from(values))
+}
+
 fn array_value_eq(
     left: &dyn Array,
     left_index: usize,
diff --git a/crates/omnigraph/src/exec/query.rs b/crates/omnigraph/src/exec/query.rs
index 4262cbf7..8e5bc0b0 100644
--- a/crates/omnigraph/src/exec/query.rs
+++ b/crates/omnigraph/src/exec/query.rs
@@ -658,6 +658,14 @@ fn search_filter_variable(filter: &IRFilter) -> Option<&str> {
     }
 }
 
+/// Exact string predicates (`starts_with`, string `contains`) that should be
+/// pushed into the scan of the variable they constrain, where Lance can probe
+/// a covering BTREE (LikePrefix) or NGRAM (StringContains) index instead of
+/// filtering a full in-memory materialization of the type.
+fn is_string_match_filter(filter: &IRFilter) -> bool {
+    matches!(filter.op, CompOp::StartsWith | CompOp::StringContains)
+}
+
 fn execute_pipeline<'a>(
     pipeline: &'a [IROp],
     params: &'a ParamMap,
@@ -668,7 +676,20 @@ fn execute_pipeline<'a>(
     search_mode: &'a SearchMode,
 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<()>> + Send + 'a>> {
     Box::pin(async move {
-        // Pre-pass: collect search filters that need to be hoisted to NodeScan
+        // Pre-pass: collect filters that need to be hoisted to NodeScan —
+        // search filters (Lance full_text_search) and pushable string-match
+        // predicates (starts_with / string contains, which Lance can answer
+        // from a covering BTREE/NGRAM index at the scan). String-match hoists
+        // are gated on the variable actually having a NodeScan: a NodeScan
+        // silently ignores non-applicable filters, whereas leaving the Filter
+        // op in place is always correct (in-memory arm).
+        let scan_vars: HashSet<&str> = pipeline
+            .iter()
+            .filter_map(|op| match op {
+                IROp::NodeScan { variable, .. } => Some(variable.as_str()),
+                _ => None,
+            })
+            .collect();
         let mut hoisted_search_filters: HashMap<String, Vec<IRFilter>> = HashMap::new();
         let mut hoisted_indices: HashSet<usize> = HashSet::new();
         for (i, op) in pipeline.iter().enumerate() {
@@ -681,6 +702,25 @@ fn execute_pipeline<'a>(
                             .push(filter.clone());
                         hoisted_indices.insert(i);
                     }
+                } else if is_string_match_filter(filter)
+                    // The needle must be a literal or param: scan-level
+                    // lowering drops variable qualifiers (PropAccess becomes
+                    // a bare column ident), so a cross-variable predicate
+                    // hoisted into one scan would degenerate to comparing a
+                    // column with itself. Cross-variable forms stay in the
+                    // in-memory arm, which evaluates on the joined wide batch.
+                    && matches!(&filter.right, IRExpr::Literal(_) | IRExpr::Param(_))
+                    && ir_filter_to_expr(filter, params, None).is_some()
+                {
+                    if let IRExpr::PropAccess { variable, .. } = &filter.left {
+                        if scan_vars.contains(variable.as_str()) {
+                            hoisted_search_filters
+                                .entry(variable.clone())
+                                .or_default()
+                                .push(filter.clone());
+                            hoisted_indices.insert(i);
+                        }
+                    }
                 }
             }
         }
@@ -2227,10 +2267,12 @@ pub(super) fn build_lance_filter_expr(
     use datafusion::prelude::Expr;
 
     let mut acc: Option<Expr> = None;
+    let mut pushed = 0u64;
     for f in filters {
         let Some(e) = ir_filter_to_expr(f, params, schema) else {
             continue;
         };
+        pushed += 1;
         acc = Some(match acc {
             None => e,
             Some(prev) => Expr::BinaryExpr(datafusion::logical_expr::BinaryExpr::new(
@@ -2240,6 +2282,7 @@ pub(super) fn build_lance_filter_expr(
             )),
         });
     }
+    crate::instrumentation::record_pushed_filter_exprs(pushed);
     acc
 }
 
@@ -2268,6 +2311,22 @@ pub(super) fn ir_filter_to_expr(
         return Some(array_has(left, right));
     }
 
+    // Exact string predicates lower to the DataFusion `starts_with`/`contains`
+    // scalar functions. The function NAMES are load-bearing: Lance's scalar
+    // index expression parser matches them to probe a BTREE (`starts_with` →
+    // LikePrefix) or an NGRAM index (`contains` → StringContains + recheck)
+    // when one covers the column, and falls back to a plain filtered scan
+    // when none does — correct either way.
+    if matches!(filter.op, CompOp::StartsWith | CompOp::StringContains) {
+        use datafusion::functions::expr_fn::{contains, starts_with};
+        let left = ir_expr_to_expr(&filter.left, params, None)?;
+        let right = ir_expr_to_expr(&filter.right, params, None)?;
+        return Some(match filter.op {
+            CompOp::StartsWith => starts_with(left, right),
+            _ => contains(left, right),
+        });
+    }
+
     // A literal/param operand is coerced to the OTHER operand's column type so
     // the predicate stays a direct `col OP literal` and the scalar index is used.
     // Without this, DataFusion widens a narrow column (`CAST(col AS Int64)`),
@@ -2283,7 +2342,9 @@ pub(super) fn ir_filter_to_expr(
         CompOp::Lt => left.lt(right),
         CompOp::Ge => left.gt_eq(right),
         CompOp::Le => left.lt_eq(right),
-        CompOp::Contains => unreachable!("handled above"),
+        CompOp::Contains | CompOp::StartsWith | CompOp::StringContains => {
+            unreachable!("handled above")
+        }
     })
 }
 
diff --git a/crates/omnigraph/src/instrumentation.rs b/crates/omnigraph/src/instrumentation.rs
index 883af29e..8f55297d 100644
--- a/crates/omnigraph/src/instrumentation.rs
+++ b/crates/omnigraph/src/instrumentation.rs
@@ -68,6 +68,15 @@ pub struct QueryIoProbes {
     /// invocations). A cost test asserts a query referencing one edge builds only
     /// that edge, not every catalog edge (the cold-build shrink A2 ships).
     pub graph_edges_built: Arc<AtomicU64>,
+    /// IR filters lowered into a scan-level DataFusion `filter_expr` (summed
+    /// over `build_lance_filter_expr` calls). Lets a test assert a standalone
+    /// string-match predicate was HOISTED into the NodeScan (where Lance can
+    /// probe a covering index) rather than silently degrading to the
+    /// in-memory arm — a result-only assertion passes either way.
+    pub pushed_filter_exprs: Arc<AtomicU64>,
+    /// Filters evaluated by the in-memory arm (`projection.rs::apply_filter`),
+    /// the complement of `pushed_filter_exprs` for hoist assertions.
+    pub in_memory_filters: Arc<AtomicU64>,
 }
 
 tokio::task_local! {
@@ -164,6 +173,20 @@ pub(crate) fn record_graph_build(edges: usize) {
     });
 }
 
+/// Record `n` IR filters lowered into a scan-level `filter_expr`. No-op when
+/// no probes are installed (production) and when nothing was pushed.
+pub(crate) fn record_pushed_filter_exprs(n: u64) {
+    if n > 0 {
+        let _ = current(|p| p.pushed_filter_exprs.fetch_add(n, Ordering::Relaxed));
+    }
+}
+
+/// Record one in-memory filter application (`apply_filter`). No-op when no
+/// probes are installed (production).
+pub(crate) fn record_in_memory_filter() {
+    let _ = current(|p| p.in_memory_filters.fetch_add(1, Ordering::Relaxed));
+}
+
 /// Per-operation staged-write counts, installed for a task via
 /// [`with_merge_write_probes`]. Lets a cost-budget test assert WHICH staged-write
 /// primitive an operation invokes — e.g. that an append-only fast-forward merge
diff --git a/crates/omnigraph/src/storage_layer.rs b/crates/omnigraph/src/storage_layer.rs
index d8b433d2..03445dd7 100644
--- a/crates/omnigraph/src/storage_layer.rs
+++ b/crates/omnigraph/src/storage_layer.rs
@@ -113,9 +113,22 @@ pub(crate) mod sealed {
 /// outside this storage contract.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum IndexBuildSpec {
-    BTree { column: String },
-    FullText { column: String },
-    Vector { column: String },
+    /// `name: None` keeps Lance's default index name (`{column}_idx`).
+    /// An explicit name is REQUIRED whenever the column already carries an
+    /// index of another kind under the default name: Lance's `replace(true)`
+    /// removes existing indexes BY NAME, so an unnamed second build would
+    /// silently replace the first index instead of coexisting with it
+    /// (pinned by `lance_surface_guards::second_index_on_column_requires_explicit_distinct_name`).
+    BTree {
+        column: String,
+        name: Option<String>,
+    },
+    FullText {
+        column: String,
+    },
+    Vector {
+        column: String,
+    },
 }
 
 /// Logical semantics for an RFC-023 keyed write.
diff --git a/crates/omnigraph/src/table_store.rs b/crates/omnigraph/src/table_store.rs
index c3564471..240ec42d 100644
--- a/crates/omnigraph/src/table_store.rs
+++ b/crates/omnigraph/src/table_store.rs
@@ -2463,7 +2463,7 @@ impl TableStore {
 
         for spec in specs {
             let (column, index_type) = match spec {
-                IndexBuildSpec::BTree { column } => (column, "BTREE"),
+                IndexBuildSpec::BTree { column, .. } => (column, "BTREE"),
                 IndexBuildSpec::FullText { column } => (column, "FTS"),
                 IndexBuildSpec::Vector { column } => (column, "Vector"),
             };
@@ -2475,13 +2475,18 @@ impl TableStore {
 
             let mut ds_clone = ds.clone();
             let new_idx = match spec {
-                IndexBuildSpec::BTree { column } => {
+                IndexBuildSpec::BTree { column, name } => {
                     let params = ScalarIndexParams::default();
-                    ds_clone
-                        .create_index_builder(&[column.as_str()], IndexType::BTree, &params)
-                        .replace(true)
-                        .execute_uncommitted()
-                        .await
+                    let columns = [column.as_str()];
+                    let mut builder =
+                        ds_clone.create_index_builder(&columns, IndexType::BTree, &params);
+                    // An explicit name keeps this BTREE from replacing a
+                    // same-column index of another kind under Lance's shared
+                    // default name (see `IndexBuildSpec::BTree`).
+                    if let Some(name) = name {
+                        builder = builder.name(name.clone());
+                    }
+                    builder.replace(true).execute_uncommitted().await
                 }
                 IndexBuildSpec::FullText { column } => {
                     let params = InvertedIndexParams::default();
diff --git a/crates/omnigraph/src/table_store/staged_tests.rs b/crates/omnigraph/src/table_store/staged_tests.rs
index f354139d..734b42cf 100644
--- a/crates/omnigraph/src/table_store/staged_tests.rs
+++ b/crates/omnigraph/src/table_store/staged_tests.rs
@@ -2169,10 +2169,18 @@ async fn stage_create_indices_batches_mixed_types_into_one_exact_commit() {
             &[
                 IndexBuildSpec::BTree {
                     column: "id".to_string(),
+                    name: None,
                 },
                 IndexBuildSpec::FullText {
                     column: "body".to_string(),
                 },
+                // Second index on the SAME column: the explicit name keeps it
+                // from replacing the FTS index under Lance's shared default
+                // name (the dual-index shape free-text @index columns get).
+                IndexBuildSpec::BTree {
+                    column: "body".to_string(),
+                    name: Some("body_btree".to_string()),
+                },
                 IndexBuildSpec::Vector {
                     column: "embedding".to_string(),
                 },
@@ -2213,6 +2221,10 @@ async fn stage_create_indices_batches_mixed_types_into_one_exact_commit() {
     );
     assert!(store.has_btree_index(&new_ds, "id").await.unwrap());
     assert!(store.has_fts_index(&new_ds, "body").await.unwrap());
+    assert!(
+        store.has_btree_index(&new_ds, "body").await.unwrap(),
+        "the explicitly-named companion BTREE must land beside the FTS index"
+    );
     assert!(store.has_vector_index(&new_ds, "embedding").await.unwrap());
 }
 
diff --git a/crates/omnigraph/tests/lance_surface_guards.rs b/crates/omnigraph/tests/lance_surface_guards.rs
index 78b04419..40081eb9 100644
--- a/crates/omnigraph/tests/lance_surface_guards.rs
+++ b/crates/omnigraph/tests/lance_surface_guards.rs
@@ -25,7 +25,7 @@
 
 use std::sync::Arc;
 
-use arrow_array::{Int32Array, RecordBatch, RecordBatchIterator, StringArray};
+use arrow_array::{Array, Int32Array, RecordBatch, RecordBatchIterator, StringArray};
 use arrow_schema::{DataType, Field, Schema};
 use lance::Dataset;
 use lance::dataset::builder::DatasetBuilder;
@@ -2142,3 +2142,443 @@ async fn filtered_scan_tolerates_merge_update_row_id_overlap() {
         assert_eq!(rows, expected, "filtered read for {slug}");
     }
 }
+
+// --- Guard 21: starts_with routes to the BTREE (LikePrefix) and stays literal --
+//
+// The .gq `starts_with` predicate lowers to the DataFusion `starts_with`
+// scalar function pushed via `Scanner::filter_expr`. Lance's scalar-index
+// expression parser rewrites it to `SargableQuery::LikePrefix`, answered
+// exactly by a covering BTREE (range [prefix, next_prefix), unicode-safe
+// successor, no recheck). Two load-bearing behaviors are pinned:
+//
+//   1. The plan uses the scalar index (a result-only assertion would also
+//      pass on a silent full-scan fallback).
+//   2. The prefix is treated LITERALLY: `_`/`%` in the needle are plain
+//      bytes, never LIKE metacharacters ("a_b" must not match "axb").
+#[tokio::test]
+async fn starts_with_filter_routes_to_btree_and_is_literal() {
+    use datafusion::functions::expr_fn::starts_with;
+    use datafusion::physical_plan::displayable;
+    use datafusion::prelude::{ident, lit};
+    use futures::TryStreamExt;
+
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().join("guard21.lance");
+    let uri = uri.to_str().unwrap();
+
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Utf8, false),
+        Field::new("name", DataType::Utf8, true),
+    ]));
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(StringArray::from(vec!["1", "2", "3", "4", "5"])),
+            Arc::new(StringArray::from(vec![
+                Some("alice"),
+                Some("alps"),
+                Some("a_b"),
+                Some("axb"),
+                None,
+            ])),
+        ],
+    )
+    .unwrap();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+    let params = WriteParams {
+        mode: WriteMode::Create,
+        enable_stable_row_ids: true,
+        data_storage_version: Some(LanceFileVersion::V2_2),
+        ..Default::default()
+    };
+    let mut ds = Dataset::write(reader, uri, Some(params)).await.unwrap();
+    ds.create_index_builder(&["name"], IndexType::BTree, &ScalarIndexParams::default())
+        .replace(true)
+        .await
+        .unwrap();
+
+    async fn ids_for(ds: &Dataset, filter: datafusion::prelude::Expr) -> Vec<String> {
+        let mut scanner = ds.scan();
+        scanner.filter_expr(filter);
+        let batches: Vec<RecordBatch> = scanner
+            .try_into_stream()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        let mut ids: Vec<String> = Vec::new();
+        for b in &batches {
+            let col = b
+                .column_by_name("id")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .unwrap();
+            for i in 0..col.len() {
+                ids.push(col.value(i).to_string());
+            }
+        }
+        ids.sort();
+        ids
+    }
+
+    // Plan shape: the starts_with function expr must reach the scalar index.
+    let mut scanner = ds.scan();
+    scanner.filter_expr(starts_with(ident("name"), lit("al")));
+    let plan = scanner.create_plan().await.unwrap();
+    let plan_str = format!("{}", displayable(plan.as_ref()).indent(true));
+    assert!(
+        plan_str.contains("ScalarIndexQuery"),
+        "starts_with on a BTREE'd column must plan a scalar-index probe \
+         (LikePrefix); a red here means the Lance expression parser no longer \
+         maps the DataFusion `starts_with` function. plan:\n{plan_str}"
+    );
+
+    // Exact prefix semantics, NULL excluded.
+    assert_eq!(
+        ids_for(&ds, starts_with(ident("name"), lit("al"))).await,
+        vec!["1", "2"],
+        "prefix 'al' must match alice+alps only (never the NULL row)"
+    );
+    // Literal treatment of LIKE metacharacters: 'a_' matches only 'a_b'.
+    assert_eq!(
+        ids_for(&ds, starts_with(ident("name"), lit("a_"))).await,
+        vec!["3"],
+        "starts_with must treat '_' as a literal byte, not a LIKE wildcard \
+         ('a_' must not match 'axb')"
+    );
+}
+
+// --- Guard 22: contains routes to an NGRAM index and rechecks to exact results --
+//
+// The .gq String `contains` predicate lowers to the DataFusion `contains`
+// scalar function. With an NGRAM index on the column, Lance's expression
+// parser maps it to `TextQuery::StringContains` — an inexact (AtMost)
+// trigram-intersection probe followed by an automatic recheck, so final
+// results are exact. Pins: NGram index creation through the same
+// `create_index_builder` surface the engine uses, the plan probing the
+// index, exact substring semantics across token boundaries, and the
+// below-trigram-width needle (< 3 chars) degrading to a correct recheck-all
+// rather than an error or a wrong result.
+#[tokio::test]
+async fn contains_filter_routes_to_ngram_index_and_rechecks_exactly() {
+    use datafusion::functions::expr_fn::contains;
+    use datafusion::physical_plan::displayable;
+    use datafusion::prelude::{ident, lit};
+    use futures::TryStreamExt;
+
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().join("guard22.lance");
+    let uri = uri.to_str().unwrap();
+
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Utf8, false),
+        Field::new("text", DataType::Utf8, true),
+    ]));
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(StringArray::from(vec!["1", "2", "3", "4"])),
+            Arc::new(StringArray::from(vec![
+                Some("this ramen recipe simmers"),
+                Some("the beta ray shines"),
+                Some("nothing here"),
+                None,
+            ])),
+        ],
+    )
+    .unwrap();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+    let params = WriteParams {
+        mode: WriteMode::Create,
+        enable_stable_row_ids: true,
+        data_storage_version: Some(LanceFileVersion::V2_2),
+        ..Default::default()
+    };
+    let mut ds = Dataset::write(reader, uri, Some(params)).await.unwrap();
+    ds.create_index_builder(&["text"], IndexType::NGram, &ScalarIndexParams::default())
+        .replace(true)
+        .await
+        .unwrap();
+
+    async fn ids_for(ds: &Dataset, filter: datafusion::prelude::Expr) -> Vec<String> {
+        let mut scanner = ds.scan();
+        scanner.filter_expr(filter);
+        let batches: Vec<RecordBatch> = scanner
+            .try_into_stream()
+            .await
+            .unwrap()
+            .try_collect()
+            .await
+            .unwrap();
+        let mut ids: Vec<String> = Vec::new();
+        for b in &batches {
+            let col = b
+                .column_by_name("id")
+                .unwrap()
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .unwrap();
+            for i in 0..col.len() {
+                ids.push(col.value(i).to_string());
+            }
+        }
+        ids.sort();
+        ids
+    }
+
+    // Plan shape: the contains function expr must reach the NGRAM index.
+    let mut scanner = ds.scan();
+    scanner.filter_expr(contains(ident("text"), lit("ramen")));
+    let plan = scanner.create_plan().await.unwrap();
+    let plan_str = format!("{}", displayable(plan.as_ref()).indent(true));
+    assert!(
+        plan_str.contains("ScalarIndexQuery"),
+        "contains on an NGRAM'd column must plan a scalar-index probe \
+         (StringContains); a red here means the Lance expression parser no \
+         longer maps the DataFusion `contains` function. plan:\n{plan_str}"
+    );
+
+    // Exact substring semantics (recheck applied), NULL excluded.
+    assert_eq!(
+        ids_for(&ds, contains(ident("text"), lit("ramen"))).await,
+        vec!["1"]
+    );
+    // Substring crossing a token boundary — substring, not FTS token match.
+    assert_eq!(
+        ids_for(&ds, contains(ident("text"), lit("ta ray"))).await,
+        vec!["2"]
+    );
+    // KNOWN UPSTREAM BUG (pinned; lance-format/lance#7841): a needle below the trigram width (3)
+    // should degrade to a recheck-everything scan — the NGram index returns
+    // an `at_least(empty)` lower bound for it — but the scan planner treats
+    // the empty probe as authoritative and returns ZERO rows: silent row
+    // loss, not an error. Both rows here contain "ra" (the unindexed scan
+    // path returns them, proven by the in-memory-arm tests), so a red on
+    // this assertion means Lance FIXED sub-trigram containment: flip it to
+    // expect ["1", "2"] and lift the sub-trigram caveat before shipping the
+    // NGRAM `@index` kind — .gq String `contains` must never silently drop
+    // rows on short needles.
+    assert_eq!(
+        ids_for(&ds, contains(ident("text"), lit("ra"))).await,
+        Vec::<String>::new(),
+        "sub-trigram contains on an NGRAM'd column currently drops all rows \
+         (upstream bug); a non-empty result means Lance fixed it — update \
+         this guard and the NGRAM rollout caveat"
+    );
+}
+
+// --- Guard 23: a second index on a column requires an explicit distinct name ---
+//
+// Lance derives the default index name `{column}_idx` and `.replace(true)`
+// removes existing indexes BY NAME. So an unnamed second-index build on an
+// already-indexed column either replaces the first index or refuses — it
+// never yields two coexisting indexes. The engine therefore MUST pass an
+// explicit `.name(...)` whenever it adds a second index kind to one column
+// (dual BTREE beside FTS; opt-in NGRAM). Pins both halves: distinctly-named
+// indexes of different types coexist, and the unnamed path never silently
+// coexists. If Lance changes its naming/replace semantics, this turns red —
+// re-validate the engine's index-naming strategy in stage_create_indices.
+#[tokio::test]
+async fn second_index_on_column_requires_explicit_distinct_name() {
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().join("guard23.lance");
+    let uri = uri.to_str().unwrap();
+
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Utf8, false),
+        Field::new("text", DataType::Utf8, false),
+    ]));
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(StringArray::from(vec!["1", "2"])),
+            Arc::new(StringArray::from(vec!["hello world", "beta ray"])),
+        ],
+    )
+    .unwrap();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema);
+    let params = WriteParams {
+        mode: WriteMode::Create,
+        enable_stable_row_ids: true,
+        data_storage_version: Some(LanceFileVersion::V2_2),
+        ..Default::default()
+    };
+    let mut ds = Dataset::write(reader, uri, Some(params)).await.unwrap();
+
+    fn type_urls(indices: &[lance_table::format::IndexMetadata]) -> Vec<String> {
+        let mut v: Vec<String> = indices
+            .iter()
+            .filter_map(|m| m.index_details.as_ref().map(|d| d.type_url.clone()))
+            .collect();
+        v.sort();
+        v.dedup();
+        v
+    }
+
+    // Baseline: an unnamed FTS build lands under the default `text_idx` name.
+    ds.create_index_builder(
+        &["text"],
+        IndexType::Inverted,
+        &lance_index::scalar::InvertedIndexParams::default(),
+    )
+    .replace(true)
+    .await
+    .unwrap();
+    let after_fts = ds.load_indices().await.unwrap();
+    assert_eq!(after_fts.len(), 1);
+    assert_eq!(after_fts[0].name, "text_idx");
+
+    // The trap: an unnamed BTREE build on the same column must never leave
+    // BOTH indexes standing (today it replaces the FTS under the shared
+    // default name; an error would also satisfy the pin).
+    let unnamed = ds
+        .create_index_builder(&["text"], IndexType::BTree, &ScalarIndexParams::default())
+        .replace(true)
+        .await;
+    ds.checkout_latest().await.unwrap();
+    let after_unnamed = ds.load_indices().await.unwrap();
+    let distinct_types = type_urls(&after_unnamed).len();
+    assert!(
+        unnamed.is_err() || distinct_types == 1,
+        "an unnamed second-index build must replace or refuse, never coexist \
+         (got {} indexes with types {:?}) — if Lance now auto-uniquifies \
+         same-field names, re-validate the engine's explicit-naming strategy",
+        after_unnamed.len(),
+        type_urls(&after_unnamed),
+    );
+
+    // The contract the engine relies on: an explicitly-named second index of a
+    // different type coexists with the first.
+    ds.create_index_builder(
+        &["text"],
+        IndexType::Inverted,
+        &lance_index::scalar::InvertedIndexParams::default(),
+    )
+    .replace(true)
+    .await
+    .unwrap();
+    ds.create_index_builder(&["text"], IndexType::BTree, &ScalarIndexParams::default())
+        .name("text_btree_idx".to_string())
+        .replace(true)
+        .await
+        .unwrap();
+    ds.checkout_latest().await.unwrap();
+    let after_named = ds.load_indices().await.unwrap();
+    let names: Vec<&str> = {
+        let mut n: Vec<&str> = after_named.iter().map(|m| m.name.as_str()).collect();
+        n.sort();
+        n
+    };
+    assert_eq!(
+        names,
+        vec!["text_btree_idx", "text_idx"],
+        "distinctly-named indexes of different types must coexist on one column"
+    );
+    assert_eq!(
+        type_urls(&after_named).len(),
+        2,
+        "expected two distinct index types on the column, got {:?}",
+        type_urls(&after_named)
+    );
+}
+
+// --- Guard 24: second-generation shallow-clone index reads fail upstream ------
+//
+// Upstream tracking: lance-format/lance#7840.
+//
+// PURE-LANCE repro of the fork-lineage index bug (no omnigraph code): a
+// dataset's index files are recorded via `base_paths` redirects when a branch
+// is shallow-cloned, but cloning a CLONE records the redirect against the
+// immediate source tree instead of composing the source's own redirect to
+// where the files actually live. Every index-consuming read through the
+// second-generation branch then hard-errors with `Not found:
+// …/tree/<parent>/_indices/…` instead of degrading.
+//
+// Omnigraph hits this whenever a branch-of-a-branch materializes its own
+// table fork (e.g. a fast-forward `branch_merge` into a non-main target) and
+// a query then probes ANY index — BTREE equality, FTS `search`. It is the
+// reason the free-text companion BTREE (equality/prefix acceleration) is
+// deferred: it would widen the exposure to every `@key` equality lookup.
+//
+// This guard asserts the BUG (like the former blob-compaction guard): it
+// turns RED when a Lance bump fixes second-generation clone index reads —
+// then (1) delete this guard, (2) re-land the companion-BTREE dispatch
+// (`plan_index_work_node`) with its tests, and (3) re-check
+// `branch_merge_into_non_main_target_works` against the dual-index truth.
+#[tokio::test]
+async fn second_generation_branch_index_reads_fail_upstream() {
+    use futures::TryStreamExt;
+
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().join("guard24.lance");
+    let uri = uri.to_str().unwrap();
+
+    // Base dataset with a BTREE on `value` (the exact index-build call shape
+    // the engine uses).
+    let mut ds = fresh_dataset(uri).await;
+    ds.create_index_builder(&["value"], IndexType::BTree, &ScalarIndexParams::default())
+        .replace(true)
+        .await
+        .unwrap();
+
+    // First-generation branch (the engine's fork call shape), plus a write so
+    // the branch has its own commits.
+    let version = ds.version().version;
+    let mut feature = ds.create_branch("feature", version, None).await.unwrap();
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Utf8, false),
+        Field::new("value", DataType::Int32, false),
+    ]));
+    let batch = RecordBatch::try_new(
+        schema.clone(),
+        vec![
+            Arc::new(StringArray::from(vec!["carol"])),
+            Arc::new(Int32Array::from(vec![3])),
+        ],
+    )
+    .unwrap();
+    let reader = RecordBatchIterator::new(vec![Ok(batch)], schema.clone());
+    feature.append(reader, None).await.unwrap();
+
+    // An indexed read through the FIRST-generation branch works: the clone's
+    // base-path redirect resolves the index files in the root tree.
+    async fn indexed_rows(ds: &Dataset) -> lance::Result<usize> {
+        let mut scanner = ds.scan();
+        scanner.filter("value = 1").unwrap();
+        let batches: Vec<RecordBatch> = scanner.try_into_stream().await?.try_collect().await?;
+        Ok(batches.iter().map(|b| b.num_rows()).sum())
+    }
+    assert_eq!(
+        indexed_rows(&feature).await.unwrap(),
+        1,
+        "first-generation clone must resolve parent index files"
+    );
+
+    // Second-generation branch: clone the clone.
+    let feature_version = feature.version().version;
+    let experiment = feature
+        .create_branch("experiment", feature_version, None)
+        .await
+        .unwrap();
+
+    // The indexed read through the second-generation clone currently fails
+    // with a Not-found on `tree/feature/_indices/...` — the redirect points at
+    // the immediate source tree, where the index files never lived.
+    let result = indexed_rows(&experiment).await;
+    match result {
+        Err(e) => {
+            let msg = e.to_string();
+            assert!(
+                msg.contains("Not found") && msg.contains("_indices"),
+                "expected the known index-file Not-found failure, got: {msg}"
+            );
+        }
+        Ok(n) => panic!(
+            "second-generation clone indexed read SUCCEEDED ({n} rows) — Lance fixed \
+             clone-of-clone index base paths. Delete this guard, re-land the free-text \
+             companion BTREE dispatch, and re-validate the branch-merge topology tests."
+        ),
+    }
+}
diff --git a/crates/omnigraph/tests/literal_filters.rs b/crates/omnigraph/tests/literal_filters.rs
index 9fb480a0..acbfb361 100644
--- a/crates/omnigraph/tests/literal_filters.rs
+++ b/crates/omnigraph/tests/literal_filters.rs
@@ -17,6 +17,7 @@ use helpers::*;
 const SCHEMA: &str = r#"
 node Metric {
     name: String @key
+    label: String? @index
     score: F64?
     ratio: F32?
     count: I32?
@@ -24,13 +25,23 @@ node Metric {
     born: Date?
     seen: DateTime?
 }
+node Tag {
+    tname: String @key
+}
+edge Tagged: Metric -> Tag
 "#;
 
 // Seeds partition every predicate, so a dropped filter returns all 4 rows.
-const DATA: &str = r#"{"type":"Metric","data":{"name":"m1","score":2.5,"ratio":0.5,"count":1,"active":true,"born":"2024-06-01","seen":"2024-06-01T12:00:00Z"}}
-{"type":"Metric","data":{"name":"m2","score":1.0,"ratio":0.25,"count":2,"active":false,"born":"2023-01-01","seen":"2023-01-01T00:00:00Z"}}
-{"type":"Metric","data":{"name":"m3","score":3.0,"ratio":0.75,"count":3,"active":true,"born":"2025-01-01","seen":"2025-01-01T00:00:00Z"}}
-{"type":"Metric","data":{"name":"m4","score":0.5,"ratio":0.1,"count":4,"active":false,"born":"2022-12-31","seen":"2022-01-01T00:00:00Z"}}"#;
+// m4 carries no label, pinning NULL-is-not-a-match for the string predicates.
+const DATA: &str = r#"{"type":"Metric","data":{"name":"m1","label":"alpha one","score":2.5,"ratio":0.5,"count":1,"active":true,"born":"2024-06-01","seen":"2024-06-01T12:00:00Z"}}
+{"type":"Metric","data":{"name":"m2","label":"alps","score":1.0,"ratio":0.25,"count":2,"active":false,"born":"2023-01-01","seen":"2023-01-01T00:00:00Z"}}
+{"type":"Metric","data":{"name":"m3","label":"beta ray","score":3.0,"ratio":0.75,"count":3,"active":true,"born":"2025-01-01","seen":"2025-01-01T00:00:00Z"}}
+{"type":"Metric","data":{"name":"m4","score":0.5,"ratio":0.1,"count":4,"active":false,"born":"2022-12-31","seen":"2022-01-01T00:00:00Z"}}
+{"type":"Tag","data":{"tname":"alpine"}}
+{"type":"Tag","data":{"tname":"basalt"}}
+{"edge":"Tagged","from":"m1","to":"alpine"}
+{"edge":"Tagged","from":"m1","to":"basalt"}
+{"edge":"Tagged","from":"m3","to":"basalt"}"#;
 
 async fn metric_db(dir: &tempfile::TempDir) -> Omnigraph {
     let uri = dir.path().to_str().unwrap();
@@ -146,6 +157,235 @@ query seen_eq() { match { $m: Metric { seen: datetime("2024-06-01T12:00:00Z") }
     assert_eq!(sorted_metric_names(&mut db, q, "seen_eq").await, vec!["m1"]);
 }
 
+// Exact string predicates: `starts_with` and the String overload of
+// `contains`. Standalone filters on a scanned variable are hoisted into the
+// NodeScan (the pushdown arm — Lance probes a covering BTREE/NGRAM index when
+// one exists and scans otherwise); the same predicates on a variable
+// introduced by a traversal stay in the in-memory arm. Both arms must agree,
+// and NULL is never a match on either.
+#[tokio::test]
+async fn string_predicate_filters_execute_on_scanned_variable() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    let q = r#"
+query prefix() { match { $m: Metric  $m.label starts_with "alp" } return { $m.name } }
+query prefix_param($q: String) { match { $m: Metric  $m.label starts_with $q } return { $m.name } }
+query substr() { match { $m: Metric  $m.label contains "ta r" } return { $m.name } }
+query substr_all() { match { $m: Metric  $m.label contains "a" } return { $m.name } }
+query key_prefix() { match { $m: Metric  $m.name starts_with "m" } return { $m.name } }
+"#;
+    // Prefix: "alpha one", "alps" start with "alp"; NULL label (m4) is not a match.
+    assert_eq!(sorted_metric_names(&mut db, q, "prefix").await, vec!["m1", "m2"]);
+    // Substring across a token boundary — proves exact substring, not FTS
+    // token matching ("ta r" spans "beta ray").
+    assert_eq!(sorted_metric_names(&mut db, q, "substr").await, vec!["m3"]);
+    // Single-char needle (below the NGRAM trigram width) still returns exact results.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "substr_all").await,
+        vec!["m1", "m2", "m3"]
+    );
+    // Prefix over the BTREE'd @key column: every row matches.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "key_prefix").await,
+        vec!["m1", "m2", "m3", "m4"]
+    );
+    // Param-bound needle takes the same path as a literal.
+    let r = query_main(&mut db, q, "prefix_param", &params(&[("$q", "alph")]))
+        .await
+        .unwrap();
+    assert_eq!(r.num_rows(), 1, "only m1's label starts with 'alph'");
+}
+
+// A cross-variable string predicate ($b.label starts_with $a.label) must NOT
+// be hoisted into either scan: scan-level lowering drops variable qualifiers
+// (PropAccess becomes a bare column ident), so a hoisted cross-variable
+// predicate would compare $b.label with itself and return every cross-join
+// pair. It must evaluate in memory after both variables are bound.
+#[tokio::test]
+async fn cross_variable_string_predicate_is_not_hoisted() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    let q = r#"
+query cross() {
+    match {
+        $a: Metric
+        $b: Metric
+        $b.label starts_with $a.label
+    }
+    return { $a.name, $b.name }
+}
+"#;
+    let r = query_main(&mut db, q, "cross", &ParamMap::new()).await.unwrap();
+    // Only the three non-null self-pairs match; a wrongly-hoisted predicate
+    // degenerates to `label starts_with label` on $b and returns 3×4 pairs.
+    assert_eq!(
+        r.num_rows(),
+        3,
+        "cross-variable starts_with must evaluate after the cross join"
+    );
+}
+
+// The contains overload must resolve for variables introduced ONLY inside a
+// negation (`not {{ … }}`): negation inners are typechecked into a discarded
+// context clone, so resolution cannot depend on the outer TypeContext alone.
+// A String-column contains left unresolved would execute as list-membership
+// and error at runtime.
+#[tokio::test]
+async fn string_contains_resolves_inside_negation() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    let q = r#"
+query not_sal_tagged() {
+    match {
+        $m: Metric
+        not { $m tagged $t  $t.tname contains "sal" }
+    }
+    return { $m.name }
+}
+"#;
+    // m1 and m3 are tagged "basalt" (contains "sal") and drop out.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "not_sal_tagged").await,
+        vec!["m2", "m4"]
+    );
+}
+
+// Positional operand semantics (documented contract): `X contains Y` tests
+// that X contains Y whichever side each operand is on, and a same-variable
+// two-property predicate evaluates per row. Neither form is hoisted (the
+// needle isn't a literal/param), so both take the in-memory arm. The
+// reversed form uses a LITERAL left operand: a bare variable on the left is
+// shadowed by the traversal rule at parse time (`$q contains $m` reads as a
+// traversal over an edge named `contains`), a pre-existing grammar
+// precedence documented in the query-language page.
+#[tokio::test]
+async fn reversed_and_same_variable_string_predicates_execute() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    let q = r#"
+query reversed() {
+    match {
+        $m: Metric
+        "the alps are tall" contains $m.label
+    }
+    return { $m.name }
+}
+query same_var() {
+    match {
+        $m: Metric
+        $m.label starts_with $m.name
+    }
+    return { $m.name }
+}
+"#;
+    // "the alps are tall" contains the label "alps" (m2) only.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "reversed").await,
+        vec!["m2"],
+        "literal-contains-property matches m2 only"
+    );
+    // No label starts with its own row's name (m1..m4) — and the predicate
+    // must compare within each row, not degenerate or error.
+    assert!(
+        sorted_metric_names(&mut db, q, "same_var").await.is_empty(),
+        "no label starts with its row's name"
+    );
+}
+
+// Structural pin for the hoist: a standalone string predicate on a scanned
+// variable must be lowered into the NodeScan's `filter_expr` (pushdown arm,
+// where Lance can probe a covering index), NOT evaluated by the in-memory
+// arm — results alone cannot tell the two apart, because a silent full-scan
+// fallback returns the same rows. Same probe pattern as the merge
+// fast-forward structural gates (`omnigraph::instrumentation`).
+#[tokio::test]
+async fn standalone_string_predicate_is_hoisted_into_scan() {
+    use omnigraph::instrumentation::{QueryIoProbes, with_query_io_probes};
+    use std::sync::atomic::Ordering;
+
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    let q = r#"query prefix() { match { $m: Metric  $m.label starts_with "alp" } return { $m.name } }"#;
+
+    let probes = QueryIoProbes::default();
+    let r = with_query_io_probes(
+        probes.clone(),
+        query_main(&mut db, q, "prefix", &ParamMap::new()),
+    )
+    .await
+    .unwrap();
+    assert_eq!(r.num_rows(), 2, "alpha one + alps");
+    assert_eq!(
+        probes.pushed_filter_exprs.load(Ordering::Relaxed),
+        1,
+        "the standalone starts_with must be hoisted into the scan's filter_expr"
+    );
+    assert_eq!(
+        probes.in_memory_filters.load(Ordering::Relaxed),
+        0,
+        "no in-memory filter application — the hoist must fully consume the predicate"
+    );
+}
+
+// Composition shapes: a string predicate combined with an FTS search filter
+// on the same scanned variable (both reach the same NodeScan — FTS via
+// full_text_search, the predicate via filter_expr), and a string predicate
+// inside `not { }` (anti-join inner pipeline). `ensure_indices` runs first
+// so `search()` has its FTS index; the prefix predicate executes as a
+// correct scan-backed filter on the same scan.
+#[tokio::test]
+async fn string_predicates_compose_with_search_and_negation() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    db.ensure_indices().await.unwrap();
+    let q = r#"
+query prefix_and_search() {
+    match {
+        $m: Metric
+        search($m.label, "alps")
+        $m.label starts_with "al"
+    }
+    return { $m.name }
+}
+query not_alp_tagged() {
+    match {
+        $m: Metric
+        not { $m tagged $t  $t.tname starts_with "alp" }
+    }
+    return { $m.name }
+}
+"#;
+    // FTS matches the "alps" token (m2); the prefix filter keeps it.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "prefix_and_search").await,
+        vec!["m2"]
+    );
+    // Only m1 is tagged "alpine"; the anti-join drops it and keeps the rest.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "not_alp_tagged").await,
+        vec!["m2", "m3", "m4"]
+    );
+}
+
+#[tokio::test]
+async fn string_predicate_filters_execute_on_expanded_variable() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut db = metric_db(&dir).await;
+    // $t is introduced by the traversal, so these standalone filters take the
+    // in-memory arm — results must match the pushdown arm's semantics.
+    let q = r#"
+query tag_prefix() { match { $m: Metric  $m tagged $t  $t.tname starts_with "alp" } return { $m.name } }
+query tag_substr() { match { $m: Metric  $m tagged $t  $t.tname contains "sal" } return { $m.name } }
+"#;
+    // Only m1 is tagged "alpine".
+    assert_eq!(sorted_metric_names(&mut db, q, "tag_prefix").await, vec!["m1"]);
+    // "basalt" contains "sal": m1 and m3.
+    assert_eq!(
+        sorted_metric_names(&mut db, q, "tag_substr").await,
+        vec!["m1", "m3"]
+    );
+}
+
 // #283: a property-match on a camelCase `@index` field must execute, not fail
 // with "No field named reponame" at the Lance scan. Exercises the pushdown arm
 // (inline binding `Doc { repoName: $r }`) end-to-end.
diff --git a/crates/omnigraph/tests/scalar_indexes.rs b/crates/omnigraph/tests/scalar_indexes.rs
index 2e41cdfb..67309af8 100644
--- a/crates/omnigraph/tests/scalar_indexes.rs
+++ b/crates/omnigraph/tests/scalar_indexes.rs
@@ -5,7 +5,8 @@
 //! reports `Indexed` only when a BTREE covers the column (the same helper the
 //! traversal chooser uses). Enums and orderable scalars must get a BTREE so
 //! `=`/range/IN/IS NULL are index-accelerated; free-text Strings keep FTS
-//! (which `key_column_index_coverage` does not count as a BTREE, by design).
+//! only (a companion BTREE is deferred on the second-generation clone
+//! index-read bug — see `lance_surface_guards`).
 
 mod helpers;
 
@@ -32,9 +33,11 @@ const DATA: &str = r#"{"type":"Item","data":{"slug":"a","status":"active","publi
 
 // Enums and orderable scalars (DateTime, numeric) get a BTREE from the index
 // reconciler, so a `=`/range filter on them uses the index. Free-text
-// String `@index` keeps FTS (no BTREE), and an un-annotated column has no
-// scalar index — both report `Degraded`, which is the negative control that
-// keeps this test from being vacuously green.
+// String `@index` keeps FTS only (its companion BTREE is deferred on the
+// upstream clone-of-clone index-read bug), and an un-annotated column has no
+// scalar index — both report `Degraded`, the negative control that keeps
+// this test from being vacuously green. A second `ensure_indices` run must
+// be a no-op.
 #[tokio::test]
 async fn node_scalar_and_enum_index_columns_get_btree() {
     let dir = tempfile::tempdir().unwrap();
@@ -59,8 +62,9 @@ async fn node_scalar_and_enum_index_columns_get_btree() {
     let title_cov = ds.index_coverage("title").await.unwrap();
     assert!(
         matches!(title_cov, IndexCoverage::Degraded { .. }),
-        "free-text String @index should keep FTS (no BTREE), got {title_cov:?}"
+        "free-text String @index keeps FTS only while the companion BTREE is deferred, got {title_cov:?}"
     );
+    assert!(ds.has_fts_index("title").await.unwrap());
 
     // No @index annotation -> no scalar index at all -> Degraded.
     let note_cov = ds.index_coverage("note").await.unwrap();
@@ -68,4 +72,20 @@ async fn node_scalar_and_enum_index_columns_get_btree() {
         matches!(note_cov, IndexCoverage::Degraded { .. }),
         "un-annotated column should have no scalar index, got {note_cov:?}"
     );
+
+    // Idempotence: a second run finds every declared index present and
+    // publishes nothing (dual indexes must not re-plan as missing forever).
+    let version_before = ds.version();
+    db.ensure_indices().await.unwrap();
+    let snap2 = snapshot_main(&db).await.unwrap();
+    let ds2 = snap2.open("node:Item").await.unwrap();
+    assert_eq!(
+        version_before,
+        ds2.version(),
+        "second ensure_indices must be a no-op on a fully-indexed graph"
+    );
 }
+
+// (The companion-BTREE naming-collision regression — a column literally
+// named `{other}_btree` colliding with another column's companion index name
+// — lives on the `dual-btree-companion` branch with the deferred dispatch.)
diff --git a/docs/user/queries/index.md b/docs/user/queries/index.md
index dd468952..e166ae8f 100644
--- a/docs/user/queries/index.md
+++ b/docs/user/queries/index.md
@@ -24,9 +24,17 @@ Param types reuse all schema scalars; trailing `?` makes a param optional. The c
 - **Binding**: `$x: NodeType { prop: <literal | $param | now()>, … }`
 - **Traversal**: `$src EDGE_NAME { min, max? } $dst` — variable-length paths via hop bounds; default 1..1 if bounds omitted.
 - **Undirected traversal**: `$src <EDGE_NAME> $dst` — matches the edge in *either* direction with set semantics (a pair connected both ways, or a self-loop, appears once). Only valid on same-endpoint-type edges (e.g. `Related: Issue -> Issue`); an asymmetric edge is rejected at typecheck (`T22`) since it is well-typed in at most one orientation — use the directional form there. Composes with hop bounds (`$a <knows>{1,3} $b`) and `not { }` ("no edge in either direction").
-- **Filter**: `<expr> <op> <expr>` with operators `>=`, `<=`, `!=`, `>`, `<`, `=`, and string `contains`.
+- **Filter**: `<expr> <op> <expr>` with operators `>=`, `<=`, `!=`, `>`, `<`, `=`, plus the string predicates `contains` and `starts_with`.
 - **Negation**: `not { clause+ }` — desugars to anti-join over the inner pipeline.
 
+### String predicates
+
+- `$x.prop contains <needle>` is overloaded on the left operand's type: a **list** property tests membership; a scalar **String** property tests exact substring containment.
+- `$x.prop starts_with <needle>` tests an exact prefix on a scalar String property.
+- Both are **exact and case-sensitive** — no tokenization, stemming, or case folding (for token-based relevance matching use the [search functions](../search/index.md); for case-insensitive matching store a normalized column). A `NULL` value on either side is never a match. `_` and `%` in the needle are literal characters, not wildcards.
+- Operands are **positional**, like comparisons: `X contains Y` tests that X contains Y, and `X starts_with Y` tests that X begins with Y, whichever side each operand is on — `"a haystack" contains $p.name` asks whether the *literal* contains the row's name. Index acceleration applies to the canonical property-on-the-left form. One grammar caveat: a **bare variable** as the left operand (`$q contains $m`) parses as a traversal over an edge named `contains`, not as a filter — use a property access or literal on the left.
+- Both predicates are correct with or without an index. When the filtered variable is scanned directly (not introduced by a traversal) the predicate is pushed into the Lance scan, where a covering index accelerates it — BTREE for `starts_with` (exact prefix range), NGRAM for String `contains` (trigram probe + recheck). See [indexes](../search/indexes.md) for which columns get which index.
+
 ## RETURN clause
 
 `return { <expr> [as <alias>], … }` with expressions:
diff --git a/docs/user/search/index.md b/docs/user/search/index.md
index 1f0b12ad..a4c16e11 100644
--- a/docs/user/search/index.md
+++ b/docs/user/search/index.md
@@ -27,6 +27,23 @@ rank).
 - Scores and ranks propagate as ordinary columns, so you can `return` a score and
   `order` by it.
 
+## Exact string predicates vs. search functions
+
+The search functions above match **tokens** (after lowercasing, stemming, and
+stop-word removal). For exact matching on the stored string — prefix
+autocomplete, substring lookup — use the filter predicates
+[`starts_with` and String `contains`](../queries/index.md#string-predicates)
+instead. They are exact and case-sensitive, work with or without an index, and
+are accelerated by a covering BTREE (`starts_with`) or NGRAM (`contains`)
+index when the filtered variable is scanned directly. One tool per question:
+
+| Question | Use | Acceleration |
+|---|---|---|
+| "is this the prefix?" (autocomplete) | `starts_with` filter | BTREE, exact probe |
+| "does it contain this substring?" | String `contains` filter | NGRAM probe + recheck |
+| "did they misspell a word?" | `fuzzy()` | inverted index |
+| "what's most relevant?" | `bm25()` / `rrf()` | inverted index |
+
 ## Hybrid ranking with `rrf`
 
 Reciprocal Rank Fusion combines two independent rankings (typically one vector and
diff --git a/docs/user/search/indexes.md b/docs/user/search/indexes.md
index 473ae772..8c2ddf20 100644
--- a/docs/user/search/indexes.md
+++ b/docs/user/search/indexes.md
@@ -4,7 +4,7 @@
 
 | Index | Use | Notes |
 |---|---|---|
-| **BTREE scalar** | `=` / range / `IN` / `IS NULL` on a scalar | always on the node `id` and edge `src`/`dst`; and on each one-column `@index`/`@key` property that is an **enum** or an **orderable scalar** (`DateTime`/`Date`/`I32`/`I64`/`U32`/`U64`/`F32`/`F64`/`Bool`) |
+| **BTREE scalar** | `=` / range / `IN` / `IS NULL` / `starts_with` (prefix) on a scalar | always on the node `id` and edge `src`/`dst`; and on each one-column `@index`/`@key` property that is an **enum** or an **orderable scalar** (`DateTime`/`Date`/`I32`/`I64`/`U32`/`U64`/`F32`/`F64`/`Bool`) |
 | **Inverted (FTS)** | `search`, `fuzzy`, `match_text`, `bm25` | created on **free-text** (non-enum) `String` `@index`/`@key` columns |
 | **Vector** | `nearest()` k-NN | Lance picks IVF_PQ vs HNSW family by configuration; OmniGraph stores as FixedSizeList(Float32, dim) |
 
@@ -13,12 +13,18 @@ by the builder and the sidecar-pinning coverage check so they cannot drift):
 enums and orderable scalars → BTREE, free-text Strings → FTS, `Vector` → vector,
 list/`Blob` columns → none.
 
-> **Free-text Strings are not equality-indexed.** A non-enum `String` column
-> (including a `String @key` slug) gets an FTS inverted index, which Lance does
-> **not** consult for `=`/range — only for `search`/`match_text`/`bm25`. So an
-> equality filter on a free-text String falls back to a full scan. If you filter
-> a String identifier by equality on a large table, model it so the value is the
-> node id, or track it as a follow-up to also build a BTREE on such columns.
+> **Free-text Strings are not equality- or prefix-indexed (deferred).** A
+> non-enum `String` column (including a `String @key` slug) gets an FTS
+> inverted index, which Lance does **not** consult for `=`/range/`starts_with`
+> — only for `search`/`match_text`/`bm25`. Those filters stay correct but fall
+> back to a scan. A companion BTREE that closes this gap is implemented (see
+> the `dual-btree-companion` branch) but deferred on an upstream Lance bug
+> (lance-format/lance#7840): second-generation shallow clones (a branch of a
+> branch, e.g. after a fast-forward merge into a non-main target) cannot read
+> parent index files at all, and the companion would widen that failure to
+> every `@key` equality lookup. The bug is pinned by
+> `lance_surface_guards::second_generation_branch_index_reads_fail_upstream`,
+> which turns red when a Lance bump fixes it.
 
 > **Coverage and cost.** Each indexed column adds index files and build time, and
 > an index only covers the fragments it was built over. Rows appended after the