Add basic translation->execution stack

Author: abrookins

pyproject.toml

@@ -18,4 +18,15 @@ build-backend = "hatchling.build"
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+addopts = "--cov=sql_redis --cov-report=term-missing"
+
+[tool.coverage.run]
+source = ["sql_redis"]
+branch = true
+
+[tool.coverage.report]
+exclude_lines = [
+    "pragma: no cover",
+    "raise NotImplementedError",
+]
 

sql_redis/__init__.py

@@ -1,6 +1,6 @@
 """SQL to Redis command translation utility."""
 
-from sql_redis.translator import translate_sql
+from sql_redis.translator import Translator, TranslatedQuery
 
-__all__ = ["translate_sql"]
+__all__ = ["Translator", "TranslatedQuery"]
 

sql_redis/analyzer.py

@@ -40,26 +40,91 @@ def get_conditions_by_type(self, field_type: str) -> list[Condition]:
 
 class Analyzer:
     """Analyzes parsed SQL queries with schema context."""
-    
+
     def __init__(self, schemas: dict[str, dict[str, str]]):
         """Initialize analyzer with schema registry data.
-        
+
         Args:
             schemas: Dict mapping index names to field->type dicts.
         """
         self._schemas = schemas
-    
+
     def analyze(self, parsed: ParsedQuery) -> AnalyzedQuery:
         """Analyze a parsed query, resolving field types.
-        
+
         Args:
             parsed: The parsed SQL query.
-            
+
         Returns:
             An AnalyzedQuery with field types resolved.
-            
+
         Raises:
             ValueError: If the index or a field is unknown.
         """
-        raise NotImplementedError("Analyzer.analyze is not yet implemented")
+        # Validate index exists
+        if parsed.index not in self._schemas:
+            raise ValueError(f"Unknown index: {parsed.index}")
+
+        schema = self._schemas[parsed.index]
+        result = AnalyzedQuery(parsed=parsed)
+
+        # Collect all fields referenced in the query
+        referenced_fields: set[str] = set()
+
+        # Fields from SELECT
+        for field_name in parsed.fields:
+            if field_name != "*":
+                referenced_fields.add(field_name)
+
+        # Fields from conditions
+        for condition in parsed.conditions:
+            referenced_fields.add(condition.field)
+
+        # Fields from aggregations
+        for agg in parsed.aggregations:
+            if agg.field:
+                referenced_fields.add(agg.field)
+
+        # Fields from computed fields (extract field references from expressions)
+        for computed in parsed.computed_fields:
+            # Simple extraction - look for field names in the expression
+            for field_name in schema.keys():
+                if field_name in computed.expression:
+                    referenced_fields.add(field_name)
+
+        # Fields from vector search
+        if parsed.vector_search:
+            referenced_fields.add(parsed.vector_search.field)
+
+        # Fields from GROUP BY
+        for field_name in parsed.groupby_fields:
+            referenced_fields.add(field_name)
+
+        # Resolve field types
+        for field_name in referenced_fields:
+            if field_name not in schema:
+                raise ValueError(f"Unknown field: {field_name}")
+            result.field_types[field_name] = schema[field_name]
+
+        # Copy aggregations and computed fields
+        result.aggregations = parsed.aggregations
+        result.computed_fields = parsed.computed_fields
+        result.groupby_fields = parsed.groupby_fields
+
+        # Determine if this is a global aggregation
+        result.is_global_aggregation = (
+            len(parsed.aggregations) > 0 and len(parsed.groupby_fields) == 0
+        )
+
+        # Analyze vector search
+        if parsed.vector_search:
+            result.vector_search = VectorSearchAnalysis(
+                field=parsed.vector_search.field,
+                k=parsed.limit or parsed.vector_search.k or 10,
+                alias=parsed.vector_search.alias,
+            )
+            # Has prefilter if there are conditions
+            result.has_prefilter = len(parsed.conditions) > 0
+
+        return result
 

sql_redis/executor.py

@@ -0,0 +1,83 @@
+"""SQL Executor - executes translated queries against Redis."""
+
+from dataclasses import dataclass
+
+import redis
+
+from sql_redis.schema import SchemaRegistry
+from sql_redis.translator import Translator
+
+
+@dataclass
+class QueryResult:
+    """Result of executing a SQL query."""
+
+    rows: list[dict]
+    count: int
+
+
+class Executor:
+    """Executes SQL queries against Redis."""
+
+    def __init__(self, client: redis.Redis, schema_registry: SchemaRegistry):
+        """Initialize executor with Redis client and schema registry."""
+        self._client = client
+        self._schema_registry = schema_registry
+        self._translator = Translator(schema_registry)
+
+    def execute(
+        self, sql: str, *, params: dict | None = None
+    ) -> QueryResult:
+        """Execute a SQL query and return results."""
+        params = params or {}
+
+        # Substitute non-bytes params in SQL
+        for key, value in params.items():
+            placeholder = f":{key}"
+            if isinstance(value, (int, float)):
+                sql = sql.replace(placeholder, str(value))
+            elif isinstance(value, str):
+                sql = sql.replace(placeholder, f"'{value}'")
+            # bytes (vectors) are handled via Redis PARAMS
+
+        # Translate SQL to Redis command
+        translated = self._translator.translate(sql)
+
+        # Build command list and substitute vector params
+        cmd = list(translated.to_command_list())
+
+        # Find any bytes params (vectors) to substitute
+        vector_param = None
+        for value in params.values():
+            if isinstance(value, bytes):
+                vector_param = value
+                break
+
+        # Replace $vector placeholder with actual bytes
+        if vector_param:
+            for i, arg in enumerate(cmd):
+                if arg == "$vector":
+                    cmd[i] = vector_param
+
+        # Execute command
+        raw_result = self._client.execute_command(*cmd)
+
+        # Parse result based on command type
+        count = raw_result[0] if raw_result else 0
+        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)
+        else:
+            # FT.AGGREGATE format: [count, [fields1], [fields2], ...]
+            for row_data in raw_result[1:]:
+                row = dict(zip(row_data[::2], row_data[1::2]))
+                rows.append(row)
+
+        return QueryResult(rows=rows, count=count)
+