fix: score alias collision for SELECT *, reject float fuzzy levels

- When no RETURN clause exists (SELECT *), use first row's field names
  to detect score alias collisions instead of skipping the check
- Reject non-integer fuzzy levels (e.g. 2.9) with ValueError, matching
  the slop validation behavior
- Add test for fuzzy float rejection (351 total)

Author: nkanu17

sql_redis/executor.py

@@ -116,18 +116,28 @@ def _has_return_0(args: list[str]) -> bool:
             return False
 
     @staticmethod
-    def _resolve_score_alias(score_alias: str | None, args: list[str]) -> str:
+    def _resolve_score_alias(
+        score_alias: str | None,
+        args: list[str],
+        first_row_fields: set[str] | None = None,
+    ) -> str:
         """Determine a stable score column name that won't collide with
         document fields.  The alias is decided once before iterating rows
-        so every row uses the same column name."""
+        so every row uses the same column name.
+
+        When a RETURN clause is present, the returned field names are used
+        for collision detection.  When RETURN is absent (SELECT *), the
+        caller should pass ``first_row_fields`` — the field names from the
+        first result row — so we can detect collisions even when all
+        document attributes are returned."""
         alias = score_alias or "__score"
         # Extract RETURN field names from args to detect collision
         try:
             idx = args.index("RETURN")
             count = int(args[idx + 1])
             return_fields = set(args[idx + 2 : idx + 2 + count])
         except (ValueError, IndexError):
-            return_fields = set()
+            return_fields = first_row_fields or set()
         if alias in return_fields:
             alias = f"__score_{alias}"
         return alias
@@ -201,28 +211,35 @@ def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
             # RETURN 0 suppresses document fields (like NOCONTENT);
             # with WITHSCORES the reply is [count, id, score, id, score, ...]
             no_content = self._has_return_0(translated.args)
-            # Determine score column name once so every row is consistent
-            alias = (
-                self._resolve_score_alias(translated.score_alias, translated.args)
-                if with_scores
-                else ""
-            )
+
+            # Pre-resolve score alias; may be deferred for SELECT *
+            score_alias: str | None = None
 
             if with_scores and no_content:
                 # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...]
                 # Stride of 2: key, score (no field array)
+                score_alias = self._resolve_score_alias(
+                    translated.score_alias, translated.args
+                )
                 for i in range(1, len(raw_result) - 1, 2):
                     score = raw_result[i + 1]
-                    row = {alias: score}
+                    row = {score_alias: score}
                     rows.append(row)
             elif with_scores:
                 # WITHSCORES format: [count, key1, score1, [fields1], key2, score2, [fields2], ...]
                 # Stride of 3: key, score, field_list
+                # Resolve alias using first row's fields for SELECT * (no RETURN)
                 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[alias] = score
+                    if score_alias is None:
+                        score_alias = self._resolve_score_alias(
+                            translated.score_alias,
+                            translated.args,
+                            first_row_fields=set(row.keys()),
+                        )
+                    row[score_alias] = score
                     rows.append(row)
             else:
                 # Standard format: [count, key1, [fields1], key2, [fields2], ...]
@@ -312,25 +329,31 @@ async def execute(self, sql: str, *, params: dict | None = None) -> QueryResult:
         if translated.command == "FT.SEARCH":
             with_scores = "WITHSCORES" in translated.args
             no_content = self._has_return_0(translated.args)
-            alias = (
-                self._resolve_score_alias(translated.score_alias, translated.args)
-                if with_scores
-                else ""
-            )
+
+            score_alias: str | None = None
 
             if with_scores and no_content:
                 # WITHSCORES + RETURN 0: [count, id1, score1, id2, score2, ...]
+                score_alias = self._resolve_score_alias(
+                    translated.score_alias, translated.args
+                )
                 for i in range(1, len(raw_result) - 1, 2):
                     score = raw_result[i + 1]
-                    row = {alias: score}
+                    row = {score_alias: score}
                     rows.append(row)
             elif 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[alias] = score
+                    if score_alias is None:
+                        score_alias = self._resolve_score_alias(
+                            translated.score_alias,
+                            translated.args,
+                            first_row_fields=set(row.keys()),
+                        )
+                    row[score_alias] = score
                     rows.append(row)
             else:
                 # Standard format: [count, key1, [fields1], key2, [fields2], ...]

sql_redis/parser.py

@@ -1020,6 +1020,10 @@ def _add_function_condition(
             if len(args) >= 3:
                 level_val = self._extract_literal_value(args[2])
                 if level_val is not None:
+                    if isinstance(level_val, float) and level_val != int(level_val):
+                        raise ValueError(
+                            f"FUZZY level argument must be an integer (got {level_val})"
+                        )
                     fuzzy_level = int(level_val)
 
             if field_name is not None:

tests/test_sql_parser.py

@@ -871,3 +871,9 @@ def test_fulltext_float_slop_raises(self):
         parser = SQLParser()
         with pytest.raises(ValueError, match="must be an integer"):
             parser.parse("SELECT * FROM idx WHERE fulltext(title, 'hello world', 2.9)")
+
+    def test_fuzzy_float_level_raises(self):
+        """Float fuzzy level raises ValueError instead of silently truncating."""
+        parser = SQLParser()
+        with pytest.raises(ValueError, match="must be an integer"):
+            parser.parse("SELECT * FROM idx WHERE fuzzy(title, 'laptap', 2.9)")