Add prefix and substring string predicates with index acceleration

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"

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,

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();

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"),
         }
     }
 }

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),
     }
 }

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#"

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 | "_") }

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(