fix: address Codex/Copilot review feedback (cycle 2)

- score() only SELECT emits RETURN 0 to prevent full document payload leak
- score() with FT.AGGREGATE raises ValueError (WITHSCORES is FT.SEARCH-only)
- Score alias collision guard in executor (both sync/async)
- Add _escape_fulltext_term with double-quote escaping
- Apply escaping to LIKE and FUZZY operators for special chars
- Escape multi-field non-exact text search values
- Mark verbatim/nostopwords as API-only fields (no SQL parser path yet)
- 9 new tests (128 total QB+translator), 411 total pass

Author: nkanu17

sql_redis/executor.py

@@ -177,7 +177,11 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
                     score = raw_result[i + 1]
                     row_data = raw_result[i + 2]
                     row = dict(zip(row_data[::2], row_data[1::2]))
-                    row[score_alias or "__score"] = score
+                    alias = score_alias or "__score"
+                    # Guard: if alias collides with a document field, prefix it
+                    if alias in row:
+                        alias = f"__score_{alias}"
+                    row[alias] = score
                     rows.append(row)
             else:
                 # Standard format: [count, key1, [fields1], key2, [fields2], ...]
@@ -274,7 +278,10 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
                     score = raw_result[i + 1]
                     row_data = raw_result[i + 2]
                     row = dict(zip(row_data[::2], row_data[1::2]))
-                    row[score_alias or "__score"] = score
+                    alias = score_alias or "__score"
+                    if alias in row:
+                        alias = f"__score_{alias}"
+                    row[alias] = score
                     rows.append(row)
             else:
                 # Standard format: [count, key1, [fields1], key2, [fields2], ...]

sql_redis/parser.py

@@ -234,6 +234,7 @@ class ParsedQuery:
     offset: int | None = None
     filters: list[str] = dataclasses.field(default_factory=list)
     scoring: ScoringSpec | None = None  # Relevance scoring config
+    # API-only flags — not set by SQL parsing, available for programmatic use
     verbatim: bool = False  # If True, add VERBATIM to FT.SEARCH
     nostopwords: bool = False  # If True, add NOSTOPWORDS to FT.SEARCH
 

sql_redis/query_builder.py

@@ -52,30 +52,18 @@ class QueryBuilder:
     TAG_SPECIAL_CHARS = r".,<>{}[]\"':;!@#$%^&*()-+=~"
 
     # Characters that have special meaning in RediSearch free-text queries
-    # (outside of double-quoted phrases) and must be escaped with a backslash.
-    # Only characters likely to appear accidentally in user data are included;
-    # intentional RediSearch features (~, *, %, ^) are intentionally excluded.
-    TEXT_QUERY_SPECIAL_CHARS = frozenset({"\\", "-", "@", "|", "(", ")"})
-
-    @staticmethod
-    def _escape_text_value(value: str) -> str:
-        """Escape characters that are special inside RediSearch double-quoted phrases.
-
-        Backslashes and double quotes must be escaped so they don't break
-        the query syntax or alter its meaning.
-        """
-        # Escape backslashes first (so we don't double-escape the quote escapes),
-        # then escape double quotes.
-        return value.replace("\\", "\\\\").replace('"', '\\"')
+    # (outside double-quoted phrases). Must be escaped with backslash.
+    # Includes double-quote to prevent starting/ending quoted phrases.
+    TEXT_QUERY_SPECIAL_CHARS = set('\\|-()"@~!{}[]^$><=;:')
 
     @classmethod
     def _escape_fulltext_term(cls, term: str) -> str:
         """Escape characters that have special meaning in RediSearch free-text queries.
 
         Applied to individual terms used outside of double-quoted phrases (e.g.,
-        in parenthesized FULLTEXT expressions) so that user input containing
-        RediSearch operator characters like |, -, (, ), @ does not alter the
-        query semantics or produce syntax errors.
+        in parenthesized FULLTEXT expressions, LIKE, FUZZY) so that user input
+        containing RediSearch operator characters does not alter query semantics
+        or produce syntax errors.
         """
         result = []
         for char in term:
@@ -85,6 +73,17 @@ def _escape_fulltext_term(cls, term: str) -> str:
                 result.append(char)
         return "".join(result)
 
+    @staticmethod
+    def _escape_text_value(value: str) -> str:
+        """Escape characters that are special inside RediSearch double-quoted phrases.
+
+        Backslashes and double quotes must be escaped so they don't break
+        the query syntax or alter its meaning.
+        """
+        # Escape backslashes first (so we don't double-escape the quote escapes),
+        # then escape double quotes.
+        return value.replace("\\", "\\\\").replace('"', '\\"')
+
     def build_text_condition(
         self,
         field: str | list[str],
@@ -124,30 +123,35 @@ def build_text_condition(
             if operator in ("=", "!="):
                 escaped = self._escape_text_value(value)
                 return f'{prefix}(@{field_str}:"{escaped}")'
-            return f"{prefix}(@{field_str}:{value})"
+            escaped = self._escape_fulltext_term(value)
+            return f"{prefix}(@{field_str}:{escaped})"
 
         # Handle different operators
         if operator == "LIKE":
-            # Convert SQL LIKE pattern (%) to RediSearch prefix/suffix/infix (*)
-            search_value = value.replace("%", "*")
+            # Escape special chars in the non-wildcard portion, then convert % → *
+            # Split on %, escape each segment, rejoin with *
+            parts = value.split("%")
+            escaped_parts = [self._escape_fulltext_term(p) for p in parts]
+            search_value = "*".join(escaped_parts)
         elif operator == "FUZZY":
-            # Wrap with % signs — count determined by fuzzy_level
+            # Escape special chars before wrapping with % markers
+            escaped = self._escape_fulltext_term(value)
             level = fuzzy_level if fuzzy_level is not None else 1
             if level not in (1, 2, 3):
                 raise ValueError(
                     f"Fuzzy level must be 1, 2, or 3 (got {level}). "
                     "RediSearch supports a maximum Levenshtein distance of 3."
                 )
             pct = "%" * level
-            search_value = f"{pct}{value}{pct}"
+            search_value = f"{pct}{escaped}{pct}"
         elif operator in ("=", "!="):
             # Exact phrase match — always wrap in quotes, preserve stopwords.
             escaped = self._escape_text_value(value)
             search_value = f'"{escaped}"'
         elif " " in value and " OR " not in value:
-            # FULLTEXT with multi-word: tokenized search with stopword filtering.
-            # Each term is escaped to prevent RediSearch operator characters in
-            # user input from changing query semantics.
+            # FULLTEXT/MATCH with multi-word: tokenized search with stopword filtering.
+            # FULLTEXT is intentionally pass-through — users craft RediSearch queries
+            # via fulltext(), so operator chars like ~, |, - are preserved.
             words = value.split()
             removed_stopwords = [
                 w for w in words if w.lower() in REDIS_DEFAULT_STOPWORDS
@@ -166,23 +170,14 @@ def build_text_condition(
                     stacklevel=2,
                 )
 
-            if filtered_words:
-                escaped_terms = " ".join(
-                    self._escape_fulltext_term(w) for w in filtered_words
-                )
-            else:
-                # All words were stopwords; pass them through (escaped) so the
-                # query doesn't become empty. RediSearch will still skip them at
-                # query time, but this avoids a syntax error from an empty clause.
-                escaped_terms = " ".join(self._escape_fulltext_term(w) for w in words)
-            search_value = f"({escaped_terms})"
+            terms = " ".join(filtered_words) if filtered_words else value
+            search_value = f"({terms})"
         elif " OR " in value:
             # OR union within text field: split on ' OR ' and join with |
-            or_terms = [
-                self._escape_fulltext_term(t.strip()) for t in value.split(" OR ")
-            ]
+            or_terms = [t.strip() for t in value.split(" OR ")]
             search_value = f"({'|'.join(or_terms)})"
         else:
+            # Single-word FULLTEXT — escape to prevent accidental operator injection
             search_value = self._escape_fulltext_term(value)
 
         base = f"{prefix}@{field}:{search_value}"

sql_redis/translator.py

@@ -129,6 +129,13 @@ def _build_command(self, analyzed: AnalyzedQuery) -> TranslatedQuery:
         query_string = self._build_query_string(analyzed)
 
         if use_aggregate:
+            if parsed.scoring is not None:
+                raise ValueError(
+                    "score() is not supported with FT.AGGREGATE queries. "
+                    "WITHSCORES / SCORER are FT.SEARCH-only features. "
+                    "Remove score() or avoid GROUP BY / aggregation functions "
+                    "in the same query."
+                )
             return self._build_aggregate(analyzed, query_string)
         else:
             return self._build_search(analyzed, query_string)
@@ -290,7 +297,16 @@ def _build_search(
             if analyzed.vector_search.alias not in return_fields:
                 return_fields.append(analyzed.vector_search.alias)
 
-        if return_fields and return_fields != ["*"]:
+        # When score() is the only SELECT expression, parsed.fields is empty.
+        # We still need a RETURN clause to avoid leaking full document payloads.
+        # Score itself is delivered via WITHSCORES (not RETURN), but we must
+        # emit RETURN 0 so Redis returns no document attributes beyond the score.
+        score_only_select = parsed.scoring is not None and not return_fields
+
+        if score_only_select:
+            # RETURN 0 — suppress all document fields, score comes via WITHSCORES
+            args.extend(["RETURN", "0"])
+        elif return_fields and return_fields != ["*"]:
             args.append("RETURN")
             args.append(str(len(return_fields)))
             args.extend(return_fields)

tests/test_query_builder.py

@@ -531,3 +531,51 @@ def test_build_missing_condition_is_not_null(self):
         builder = QueryBuilder()
         result = builder.build_missing_condition("email", is_missing=False)
         assert result == "-ismissing(@email)"
+
+
+class TestQueryBuilderEscaping:
+    """Tests for escaping special characters in text search values."""
+
+    def test_escape_fulltext_term_special_chars(self):
+        """_escape_fulltext_term escapes RediSearch operator characters."""
+        builder = QueryBuilder()
+        result = builder._escape_fulltext_term("hello|world")
+        assert result == "hello\\|world"
+
+    def test_escape_fulltext_term_double_quote(self):
+        """_escape_fulltext_term escapes double quotes."""
+        builder = QueryBuilder()
+        result = builder._escape_fulltext_term('say "hello"')
+        assert result == 'say \\"hello\\"'
+
+    def test_escape_fulltext_term_at_sign(self):
+        """_escape_fulltext_term escapes @ to prevent field injection."""
+        builder = QueryBuilder()
+        result = builder._escape_fulltext_term("user@email")
+        assert result == "user\\@email"
+
+    def test_like_escapes_special_chars(self):
+        """LIKE pattern escapes special chars in the non-wildcard portion."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "LIKE", "%hello|world%")
+        assert result == "@title:*hello\\|world*"
+
+    def test_fuzzy_escapes_special_chars(self):
+        """FUZZY escapes special chars in the term before wrapping with %."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FUZZY", "hello@world")
+        assert result == "@title:%hello\\@world%"
+
+    def test_fuzzy_escapes_double_quote(self):
+        """FUZZY escapes double quotes in term."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition("title", "FUZZY", 'say"hi')
+        assert result == '@title:%say\\"hi%'
+
+    def test_multi_field_non_exact_escapes(self):
+        """Multi-field search with non-exact operator escapes special chars."""
+        builder = QueryBuilder()
+        result = builder.build_text_condition(
+            ["title", "description"], "FULLTEXT", "hello|world"
+        )
+        assert result == "(@title|description:hello\\|world)"