feat: advanced text search — fuzzy LD 2/3, suffix/infix, OR, proximity, BM25 scoring

- Parser: add LIKE handler, Boolean extraction for inorder param
- QueryBuilder: suffix/infix patterns, fuzzy LD 2-3, slop/inorder attributes, scoring
- Translator: score_alias on TranslatedQuery, WITHSCORES/SCORER args
- Executor: stride-3 response parsing for WITHSCORES (sync + async)
- Tests: 137 new tests (82 unit QB, 37 unit translator, 18 integration)
- All 402 tests pass, mypy clean

Author: nkanu17

sql_redis/executor.py

@@ -166,12 +166,25 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
         rows = []
 
         if translated.command == "FT.SEARCH":
-            # FT.SEARCH format: [count, key1, [fields1], key2, [fields2], ...]
-            # Skip document keys (odd indices), take field lists (even indices after count)
-            for i in range(2, len(raw_result), 2):
-                row_data = raw_result[i]
-                row = dict(zip(row_data[::2], row_data[1::2]))
-                rows.append(row)
+            # Check if WITHSCORES was requested — changes response format
+            with_scores = "WITHSCORES" in translated.args
+            score_alias = translated.score_alias
+
+            if with_scores:
+                # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...]
+                # Stride of 3: key, score, field_list
+                for i in range(1, len(raw_result) - 2, 3):
+                    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
+                    rows.append(row)
+            else:
+                # Standard format: [count, key1, [fields1], key2, [fields2], ...]
+                for i in range(2, len(raw_result), 2):
+                    row_data = raw_result[i]
+                    row = dict(zip(row_data[::2], row_data[1::2]))
+                    rows.append(row)
         else:
             # FT.AGGREGATE format: [count, [fields1], [fields2], ...]
             for row_data in raw_result[1:]:
@@ -252,11 +265,23 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
         rows = []
 
         if translated.command == "FT.SEARCH":
-            # FT.SEARCH format: [count, key1, [fields1], key2, [fields2], ...]
-            for i in range(2, len(raw_result), 2):
-                row_data = raw_result[i]
-                row = dict(zip(row_data[::2], row_data[1::2]))
-                rows.append(row)
+            with_scores = "WITHSCORES" in translated.args
+            score_alias = translated.score_alias
+
+            if with_scores:
+                # WITHSCORES format: [count, key1, score1, [fields1], ...]
+                for i in range(1, len(raw_result) - 2, 3):
+                    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
+                    rows.append(row)
+            else:
+                # Standard format: [count, key1, [fields1], key2, [fields2], ...]
+                for i in range(2, len(raw_result), 2):
+                    row_data = raw_result[i]
+                    row = dict(zip(row_data[::2], row_data[1::2]))
+                    rows.append(row)
         else:
             # FT.AGGREGATE format: [count, [fields1], [fields2], ...]
             for row_data in raw_result[1:]:

sql_redis/parser.py

@@ -164,6 +164,9 @@ class Condition:
     operator: str
     value: object
     negated: bool = False
+    fuzzy_level: int | None = None  # Levenshtein distance for FUZZY (1, 2, or 3)
+    slop: int | None = None  # Max distance between terms for proximity search
+    inorder: bool = False  # Require terms in order (used with slop)
 
 
 @dataclass
@@ -196,6 +199,17 @@ class GeoDistanceSelect:
     unit: str = "m"  # m, km, mi, ft (default: meters)
 
 
+@dataclass
+class ScoringSpec:
+    """Specification for relevance scoring.
+
+    Triggers WITHSCORES and optional SCORER on FT.SEARCH.
+    """
+
+    alias: str = "score"  # Column alias for the score
+    scorer: str = "BM25"  # Scorer algorithm (BM25, TFIDF, DISMAX, etc.)
+
+
 @dataclass
 class ParsedQuery:
     """Result of parsing a SQL query."""
@@ -219,6 +233,9 @@ class ParsedQuery:
     limit: int | None = None
     offset: int | None = None
     filters: list[str] = dataclasses.field(default_factory=list)
+    scoring: ScoringSpec | None = None  # Relevance scoring config
+    verbatim: bool = False  # If True, add VERBATIM to FT.SEARCH
+    nostopwords: bool = False  # If True, add NOSTOPWORDS to FT.SEARCH
 
 
 class SQLParser:
@@ -441,6 +458,17 @@ def _process_select_expression_inner(
             elif func_name_lower == "geo_distance":
                 # geo_distance(field, POINT(lon, lat), unit) in SELECT
                 self._process_geo_distance_select(expression, result, alias)
+            elif func_name_lower == "score":
+                # score() or score('BM25') — triggers WITHSCORES + SCORER
+                scorer = "BM25"
+                if expression.expressions:
+                    scorer_val = self._extract_literal_value(expression.expressions[0])
+                    if scorer_val is not None:
+                        scorer = str(scorer_val).upper()
+                result.scoring = ScoringSpec(
+                    alias=alias or "score",
+                    scorer=scorer,
+                )
             elif func_name_lower in redis_reducers:
                 # Redis-specific reducer functions
                 field_name = None
@@ -656,6 +684,9 @@ def _process_where_clause(
             self._add_between_condition(expression, result, negated)
         elif isinstance(expression, exp.In):
             self._add_in_condition(expression, result, negated)
+        elif isinstance(expression, exp.Like):
+            # LIKE 'pattern%' / '%pattern' / '%pattern%'
+            self._add_condition(expression, "LIKE", result, negated)
         elif isinstance(expression, exp.And):
             result.boolean_operator = "AND"
             self._process_where_clause(expression.this, result, negated)
@@ -938,25 +969,59 @@ def _add_in_condition(self, expression, result: ParsedQuery, negated: bool) -> N
     def _add_function_condition(
         self, expression, result: ParsedQuery, negated: bool
     ) -> None:
-        """Add a condition from a function call like fulltext(field, value)."""
+        """Add a condition from a function call like fulltext(field, value) or fuzzy(field, value, level)."""
         func_name = expression.name.upper()
-        if func_name == "FULLTEXT" and len(expression.expressions) >= 2:
-            first_arg = expression.expressions[0]
-            second_arg = expression.expressions[1]
+        args = expression.expressions
+
+        if func_name == "FULLTEXT" and len(args) >= 2:
+            field_name = args[0].name if isinstance(args[0], exp.Column) else None
+            value = self._extract_literal_value(args[1])
+
+            # Optional 3rd arg: slop (int)
+            slop = None
+            if len(args) >= 3:
+                slop_val = self._extract_literal_value(args[2])
+                if slop_val is not None:
+                    slop = int(slop_val)
+
+            # Optional 4th arg: inorder (boolean-like: 1/0 or true/false)
+            inorder = False
+            if len(args) >= 4:
+                inorder_val = self._extract_literal_value(args[3])
+                if inorder_val is not None:
+                    inorder = str(inorder_val).lower() in ("1", "true")
 
-            field_name = None
-            if isinstance(first_arg, exp.Column):
-                field_name = first_arg.name
+            if field_name is not None:
+                result.conditions.append(
+                    Condition(
+                        field=field_name,
+                        operator="FULLTEXT",
+                        value=value,
+                        negated=negated,
+                        slop=slop,
+                        inorder=inorder,
+                    )
+                )
 
-            value = self._extract_literal_value(second_arg)
+        elif func_name == "FUZZY" and len(args) >= 2:
+            field_name = args[0].name if isinstance(args[0], exp.Column) else None
+            value = self._extract_literal_value(args[1])
+
+            # Optional 3rd arg: fuzzy level (1, 2, or 3)
+            fuzzy_level = None
+            if len(args) >= 3:
+                level_val = self._extract_literal_value(args[2])
+                if level_val is not None:
+                    fuzzy_level = int(level_val)
 
             if field_name is not None:
                 result.conditions.append(
                     Condition(
                         field=field_name,
-                        operator="FULLTEXT",
+                        operator="FUZZY",
                         value=value,
                         negated=negated,
+                        fuzzy_level=fuzzy_level,
                     )
                 )
 
@@ -983,6 +1048,9 @@ def _extract_literal_value(self, expression, convert_dates: bool = False):
                 if timestamp is not None:
                     return timestamp
             return value
+        elif isinstance(expression, exp.Boolean):
+            # Handle TRUE/FALSE keywords parsed by sqlglot
+            return expression.this
         elif isinstance(expression, exp.Neg):
             # Handle negative numbers: Neg(Literal(122.4)) -> -122.4
             inner_value = self._extract_literal_value(expression.this)

sql_redis/query_builder.py

@@ -68,6 +68,10 @@ def build_text_condition(
         operator: str,
         value: str,
         negated: bool = False,
+        *,
+        fuzzy_level: int | None = None,
+        slop: int | None = None,
+        inorder: bool = False,
     ) -> str:
         """Build query syntax for TEXT field conditions.
 
@@ -76,6 +80,9 @@ def build_text_condition(
             operator: One of =, !=, FULLTEXT, LIKE, FUZZY.
             value: The search term or pattern.
             negated: If True, prefix with - for negation.
+            fuzzy_level: Levenshtein distance for FUZZY (1, 2, or 3). Default 1.
+            slop: Maximum distance between terms for proximity search.
+            inorder: If True with slop, require terms in order.
 
         Returns:
             RediSearch query syntax like @field:"exact phrase" or @field:(term1 term2).
@@ -91,19 +98,24 @@ def build_text_condition(
 
         # Handle different operators
         if operator == "LIKE":
-            # Convert SQL LIKE pattern (%) to RediSearch prefix (*)
+            # Convert SQL LIKE pattern (%) to RediSearch prefix/suffix/infix (*)
             search_value = value.replace("%", "*")
         elif operator == "FUZZY":
-            # Wrap with % for fuzzy matching
-            search_value = f"%{value}%"
+            # Wrap with % signs — count determined by fuzzy_level
+            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}"
         elif operator in ("=", "!="):
             # Exact phrase match — always wrap in quotes, preserve stopwords.
-            # This ensures "bank of america" stays as-is rather than
-            # being tokenized or having stopwords stripped.
             escaped = self._escape_text_value(value)
             search_value = f'"{escaped}"'
-        elif " " in value:
-            # MATCH with multi-word: tokenized search with stopword filtering
+        elif " " in value and " OR " not in value:
+            # FULLTEXT/MATCH with multi-word: tokenized search with stopword filtering
             words = value.split()
             removed_stopwords = [
                 w for w in words if w.lower() in REDIS_DEFAULT_STOPWORDS
@@ -122,13 +134,25 @@ def build_text_condition(
                     stacklevel=2,
                 )
 
-            # Use filtered words in parentheses (AND semantics), or original if all were stopwords
             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 = [t.strip() for t in value.split(" OR ")]
+            search_value = f"({'|'.join(or_terms)})"
         else:
             search_value = value
 
-        return f"{prefix}@{field}:{search_value}"
+        base = f"{prefix}@{field}:{search_value}"
+
+        # Append query attributes (slop, inorder) if specified
+        if slop is not None:
+            attrs = f"$slop: {slop};"
+            if inorder:
+                attrs += " $inorder: true;"
+            base = f"{base} => {{ {attrs} }}"
+
+        return base
 
     def _escape_tag_value(self, value: str) -> str:
         """Escape special characters in TAG values."""

sql_redis/translator.py

@@ -27,6 +27,7 @@ class TranslatedQuery:
     query_string: str
     args: list[str] = field(default_factory=list)
     params: dict[str, object] = field(default_factory=dict)  # Named parameters
+    score_alias: str | None = None  # Alias for score column when WITHSCORES is used
 
     def to_command_list(self) -> list[str]:
         """Return as a list suitable for redis.execute_command()."""
@@ -197,6 +198,9 @@ def _build_condition(self, condition: Condition, field_type: str | None) -> str:
                 operator,
                 str(condition.value),
                 is_negated,
+                fuzzy_level=condition.fuzzy_level,
+                slop=condition.slop,
+                inorder=condition.inorder,
             )
         elif field_type == "TAG":
             # Keep list value for IN clauses, convert scalar to string
@@ -301,6 +305,18 @@ def _build_search(
             offset = parsed.offset or 0
             args.extend(["LIMIT", str(offset), str(parsed.limit)])
 
+        # Scoring — WITHSCORES and SCORER
+        if parsed.scoring is not None:
+            args.append("WITHSCORES")
+            if parsed.scoring.scorer:
+                args.extend(["SCORER", parsed.scoring.scorer])
+
+        # Verbatim / nostopwords flags
+        if parsed.verbatim:
+            args.append("VERBATIM")
+        if parsed.nostopwords:
+            args.append("NOSTOPWORDS")
+
         # DIALECT 2 — unconditionally appended as the last arguments
         args.extend(["DIALECT", "2"])
 
@@ -310,6 +326,7 @@ def _build_search(
             query_string=query_string,
             args=args,
             params=params,
+            score_alias=(parsed.scoring.alias if parsed.scoring is not None else None),
         )
 
     def _build_geo_filter_args(self, geo_cond: GeoDistanceCondition) -> list[str]: