fix(ddl): replace byte-indexed NEW. rewriting with char-safe iteration

The check constraint, constraint validator, RLS, trigger, and
materialized view DDL parsers rewrote "NEW.col" references using byte
indexing after a to_uppercase() call. This produced incorrect results
for non-ASCII identifiers and is unsound for multi-byte UTF-8 sequences.

Switch all affected parsers to collect chars and iterate over the char
slice, matching the "NEW." prefix with eq_ignore_ascii_case on a
4-char window.

Also fix the SQL preprocessor's string literal scanner to advance by 2
on '' escapes instead of continuing without incrementing, which
previously caused an infinite loop on inputs with escaped single quotes.

Author: farhan-syah

nodedb-sql/src/parser/preprocess.rs

@@ -270,12 +270,14 @@ fn value_to_json(value: &nodedb_types::Value) -> String {
 fn find_matching_brace(chars: &[char], start: usize) -> Option<usize> {
     let mut depth = 0;
     let mut in_string = false;
-    for i in start..chars.len() {
+    let mut i = start;
+    while i < chars.len() {
         match chars[i] {
             '\'' if !in_string => in_string = true,
             '\'' if in_string => {
                 if i + 1 < chars.len() && chars[i + 1] == '\'' {
-                    // Skip escaped quote.
+                    // Skip '' escape — advance past both quotes.
+                    i += 2;
                     continue;
                 }
                 in_string = false;
@@ -289,6 +291,7 @@ fn find_matching_brace(chars: &[char], start: usize) -> Option<usize> {
             }
             _ => {}
         }
+        i += 1;
     }
     None
 }

nodedb/src/control/server/pgwire/ddl/collection/check_constraint.rs

@@ -271,24 +271,24 @@ fn restructure_subquery_check(expr: &str) -> RestructuredCheck {
 /// Converts `NEW.amount > 0` → `amount > 0` so the expression can be parsed
 /// as bare column references by `parse_generated_expr`.
 fn strip_new_prefix(sql: &str) -> String {
+    let chars: Vec<char> = sql.chars().collect();
     let mut result = String::with_capacity(sql.len());
-    let upper = sql.to_uppercase();
-    let bytes = sql.as_bytes();
     let mut i = 0;
 
-    while i < bytes.len() {
-        if i + 4 <= bytes.len() && upper[i..].starts_with("NEW.") {
-            // Check word boundary before.
-            if i > 0 && (bytes[i - 1].is_ascii_alphanumeric() || bytes[i - 1] == b'_') {
-                result.push(bytes[i] as char);
-                i += 1;
+    while i < chars.len() {
+        if i + 4 <= chars.len() {
+            let window: String = chars[i..i + 4].iter().collect();
+            if window.eq_ignore_ascii_case("NEW.") {
+                if i > 0 && (chars[i - 1].is_ascii_alphanumeric() || chars[i - 1] == '_') {
+                    result.push(chars[i]);
+                    i += 1;
+                    continue;
+                }
+                i += 4;
                 continue;
             }
-            // Skip "NEW." (4 chars).
-            i += 4;
-            continue;
         }
-        result.push(bytes[i] as char);
+        result.push(chars[i]);
         i += 1;
     }
     result
@@ -330,33 +330,35 @@ fn substitute_new_refs(sql: &str, fields: &HashMap<String, nodedb_types::Value>)
 
 /// Replace any remaining `NEW.xxx` references (not matched by known fields) with NULL.
 fn replace_remaining_new_refs(text: &str) -> String {
-    let upper = text.to_uppercase();
+    let chars: Vec<char> = text.chars().collect();
     let mut result = String::with_capacity(text.len());
     let mut i = 0;
-    let bytes = text.as_bytes();
 
-    while i < bytes.len() {
+    while i < chars.len() {
         // Check for "NEW." prefix (case insensitive).
-        if i + 4 <= bytes.len() && upper[i..].starts_with("NEW.") {
-            // Check word boundary before.
-            if i > 0 && (bytes[i - 1].is_ascii_alphanumeric() || bytes[i - 1] == b'_') {
-                result.push(bytes[i] as char);
-                i += 1;
-                continue;
-            }
-            // Find the end of the identifier after "NEW.".
-            let start = i + 4;
-            let mut end = start;
-            while end < bytes.len() && (bytes[end].is_ascii_alphanumeric() || bytes[end] == b'_') {
-                end += 1;
-            }
-            if end > start {
-                result.push_str("NULL");
-                i = end;
-                continue;
+        if i + 4 <= chars.len() {
+            let window: String = chars[i..i + 4].iter().collect();
+            if window.eq_ignore_ascii_case("NEW.") {
+                if i > 0 && (chars[i - 1].is_ascii_alphanumeric() || chars[i - 1] == '_') {
+                    result.push(chars[i]);
+                    i += 1;
+                    continue;
+                }
+                // Find the end of the identifier after "NEW.".
+                let start = i + 4;
+                let mut end = start;
+                while end < chars.len() && (chars[end].is_ascii_alphanumeric() || chars[end] == '_')
+                {
+                    end += 1;
+                }
+                if end > start {
+                    result.push_str("NULL");
+                    i = end;
+                    continue;
+                }
             }
         }
-        result.push(bytes[i] as char);
+        result.push(chars[i]);
         i += 1;
     }
     result

nodedb/src/control/server/pgwire/ddl/constraint/validate.rs

@@ -29,21 +29,23 @@ pub(super) fn validate_subquery_pattern(check_sql: &str) -> PgWireResult<()> {
 
 /// Strip `NEW.` prefix for validation parsing.
 pub(super) fn strip_new_prefix_for_validation(sql: &str) -> String {
-    let upper = sql.to_uppercase();
-    let bytes = sql.as_bytes();
+    let chars: Vec<char> = sql.chars().collect();
     let mut result = String::with_capacity(sql.len());
     let mut i = 0;
-    while i < bytes.len() {
-        if i + 4 <= bytes.len() && upper[i..].starts_with("NEW.") {
-            if i > 0 && (bytes[i - 1].is_ascii_alphanumeric() || bytes[i - 1] == b'_') {
-                result.push(bytes[i] as char);
-                i += 1;
+    while i < chars.len() {
+        if i + 4 <= chars.len() {
+            let window: String = chars[i..i + 4].iter().collect();
+            if window.eq_ignore_ascii_case("NEW.") {
+                if i > 0 && (chars[i - 1].is_ascii_alphanumeric() || chars[i - 1] == '_') {
+                    result.push(chars[i]);
+                    i += 1;
+                    continue;
+                }
+                i += 4;
                 continue;
             }
-            i += 4;
-            continue;
         }
-        result.push(bytes[i] as char);
+        result.push(chars[i]);
         i += 1;
     }
     result

nodedb/src/control/server/pgwire/ddl/materialized_view/parse.rs

@@ -50,16 +50,11 @@ pub fn parse_create_mv(sql: &str) -> PgWireResult<(String, String, String, Strin
         .ok_or_else(|| sqlstate_error("42601", "expected AS SELECT ... clause"))?;
     let query_start = after_on_start + as_pos + KW_AS.len();
 
-    // Find end of query: WITH clause or end of string.
-    let remaining = &upper[query_start..];
-    let with_pos = remaining.find(" WITH").or_else(|| {
-        if remaining.trim_start().starts_with("WITH") {
-            Some(0)
-        } else {
-            None
-        }
-    });
-    let query_end = with_pos.map(|p| query_start + p).unwrap_or(sql.len());
+    // Find end of query: a trailing `WITH (...)` options clause, or end of string.
+    // We must not match a CTE `WITH cte AS (...)` inside the SELECT body.
+    // The options clause is always `WITH (` — a CTE `WITH` is followed by
+    // an identifier, never `(`.
+    let query_end = find_trailing_with_options(&upper, query_start).unwrap_or(sql.len());
     let query_sql = sql[query_start..query_end].trim().to_string();
 
     if query_sql.is_empty() {
@@ -71,6 +66,35 @@ pub fn parse_create_mv(sql: &str) -> PgWireResult<(String, String, String, Strin
     Ok((name, source, query_sql, refresh_mode))
 }
 
+/// Find a trailing `WITH (...)` options clause that is NOT a CTE.
+///
+/// Scans backward from the end of the string for `WITH` followed (after
+/// optional whitespace) by `(`. CTE syntax is `WITH <name> AS (...)` — the
+/// word after `WITH` is an identifier, not `(`.
+fn find_trailing_with_options(upper: &str, query_start: usize) -> Option<usize> {
+    let region = &upper[query_start..];
+    // Search backward for the last " WITH" or "WITH" that is followed by "(".
+    let mut search_end = region.len();
+    loop {
+        let pos = region[..search_end].rfind("WITH")?;
+        // Verify word boundary before WITH.
+        if pos > 0 {
+            let before = region.as_bytes()[pos - 1];
+            if before.is_ascii_alphanumeric() || before == b'_' {
+                search_end = pos;
+                continue;
+            }
+        }
+        // Check that WITH is followed by `(` (possibly with whitespace).
+        let after_with = region[pos + 4..].trim_start();
+        if after_with.starts_with('(') {
+            return Some(query_start + pos);
+        }
+        // This WITH is a CTE or identifier — keep searching backward.
+        search_end = pos;
+    }
+}
+
 /// Extract refresh mode from WITH clause.
 fn extract_refresh_mode(upper: &str, sql: &str) -> String {
     let with_pos = match upper.rfind("WITH") {

nodedb/src/control/server/pgwire/ddl/rls/parse.rs

@@ -70,10 +70,7 @@ pub fn parse_create_rls_policy(
         .map(|i| using_idx + 1 + i)
         .unwrap_or(parts.len());
 
-    let predicate_str = parts[using_idx + 1..pred_end]
-        .join(" ")
-        .trim_matches(|c: char| c == '(' || c == ')')
-        .to_string();
+    let predicate_str = strip_outer_parens(&parts[using_idx + 1..pred_end].join(" "));
 
     let is_restrictive = parts[pred_end..]
         .iter()
@@ -114,7 +111,7 @@ pub fn parse_create_rls_policy(
 
         let field = pred_parts[0];
         let op = pred_parts[1];
-        let value_str = pred_parts[2..].join(" ").trim_matches('\'').to_string();
+        let value_str = strip_single_quotes(&pred_parts[2..].join(" "));
 
         let filter = crate::bridge::scan_filter::ScanFilter {
             field: field.to_string(),
@@ -161,3 +158,48 @@ pub fn parse_create_rls_policy(
         on_deny,
     })
 }
+
+/// Strip at most one matched pair of outer single quotes.
+///
+/// `"'hello'"` → `"hello"`, `"no_quotes"` → `"no_quotes"`.
+fn strip_single_quotes(s: &str) -> String {
+    let trimmed = s.trim();
+    if trimmed.len() >= 2 && trimmed.starts_with('\'') && trimmed.ends_with('\'') {
+        trimmed[1..trimmed.len() - 1].to_string()
+    } else {
+        trimmed.to_string()
+    }
+}
+
+/// Strip at most one matched pair of outer parentheses. Unlike
+/// `trim_matches('(' | ')')`, this preserves balanced inner parens.
+///
+/// `"((x > 0) AND (y = 1))"` → `"(x > 0) AND (y = 1)"`
+/// `"(x > 0)"` → `"x > 0"`
+/// `"x > 0"` → `"x > 0"` (no outer parens)
+fn strip_outer_parens(s: &str) -> String {
+    let trimmed = s.trim();
+    if trimmed.starts_with('(') && trimmed.ends_with(')') {
+        // Verify the outer parens are actually matched (not two separate groups).
+        let inner = &trimmed[1..trimmed.len() - 1];
+        let mut depth = 0i32;
+        let mut balanced = true;
+        for ch in inner.chars() {
+            match ch {
+                '(' => depth += 1,
+                ')' => {
+                    depth -= 1;
+                    if depth < 0 {
+                        balanced = false;
+                        break;
+                    }
+                }
+                _ => {}
+            }
+        }
+        if balanced && depth == 0 {
+            return inner.trim().to_string();
+        }
+    }
+    trimmed.to_string()
+}

nodedb/src/control/server/pgwire/ddl/trigger/parse.rs

@@ -320,21 +320,42 @@ fn extract_dollar_quoted_body(s: &str) -> Option<(&str, String)> {
 
 fn find_begin_pos(s: &str) -> Option<usize> {
     let upper = s.to_uppercase();
-    let mut search_from = 0;
-    loop {
-        let pos = upper[search_from..].find("BEGIN")?;
-        let abs_pos = search_from + pos;
-        let before_ok = abs_pos == 0
-            || !s.as_bytes()[abs_pos - 1].is_ascii_alphanumeric()
-                && s.as_bytes()[abs_pos - 1] != b'_';
-        let after_pos = abs_pos + 5;
-        let after_ok = after_pos >= s.len()
-            || !s.as_bytes()[after_pos].is_ascii_alphanumeric() && s.as_bytes()[after_pos] != b'_';
-        if before_ok && after_ok {
-            return Some(abs_pos);
+    let bytes = s.as_bytes();
+    let mut i = 0;
+
+    while i < bytes.len() {
+        // Skip single-quoted string literals so that 'BEGIN' inside a
+        // string (e.g. in a WHEN clause) is not mistaken for the body start.
+        if bytes[i] == b'\'' {
+            i += 1;
+            while i < bytes.len() {
+                if bytes[i] == b'\'' {
+                    i += 1;
+                    // Handle '' escape.
+                    if i < bytes.len() && bytes[i] == b'\'' {
+                        i += 1;
+                        continue;
+                    }
+                    break;
+                }
+                i += 1;
+            }
+            continue;
+        }
+
+        // Check for "BEGIN" keyword at this position.
+        if i + 5 <= bytes.len() && &upper[i..i + 5] == "BEGIN" {
+            let before_ok = i == 0 || !bytes[i - 1].is_ascii_alphanumeric() && bytes[i - 1] != b'_';
+            let after_pos = i + 5;
+            let after_ok = after_pos >= bytes.len()
+                || !bytes[after_pos].is_ascii_alphanumeric() && bytes[after_pos] != b'_';
+            if before_ok && after_ok {
+                return Some(i);
+            }
         }
-        search_from = abs_pos + 5;
+        i += 1;
     }
+    None
 }
 
 #[cfg(test)]

nodedb/src/control/server/pgwire/handler/plan.rs

@@ -124,6 +124,7 @@ pub(super) fn describe_plan(plan: &PhysicalPlan) -> PlanKind {
         | PhysicalPlan::Query(QueryOp::FacetCounts { .. })
         | PhysicalPlan::Query(QueryOp::HashJoin { .. })
         | PhysicalPlan::Query(QueryOp::InlineHashJoin { .. })
+        | PhysicalPlan::Query(QueryOp::RecursiveScan { .. })
         | PhysicalPlan::Graph(GraphOp::Algo { .. })
         | PhysicalPlan::Graph(GraphOp::Match { .. }) => PlanKind::MultiRow,