Fix wrong results in SQL translation, the SQLAlchemy dialect and superset mode (GROUP BY, NOT/NULL, aliases, LIKE, Decimal, UUID, time grains)

README.md

@@ -743,6 +743,20 @@ PyMongoSQL can be used as a database driver in Apache Superset for querying and
 
 This allows seamless integration between MongoDB data and Superset's BI capabilities without requiring data migration to traditional SQL databases.
 
+**Time grains and decimals:**
+
+- `DATE_TRUNC('<unit>', field)` is translated to MongoDB's `$dateTrunc` (MongoDB 5.0+), in
+  projections and `GROUP BY`. Units: `second`, `minute`, `hour`, `day`, `week` (starting
+  Sunday), `week_monday`, `month`, `quarter`, `year`, `week_ending_saturday` and
+  `week_ending_sunday`. Truncation is in UTC. The same function is available in the
+  superset-mode SQLite stage, so virtual datasets group by time the same way.
+- In superset mode, a subquery's result is loaded into an in-memory SQLite database. Columns
+  holding `Decimal128` values are evaluated exactly there (install `pymongosql[superset]`,
+  which adds `sqlglot`): the column itself, `SUM`/`AVG`/`MIN`/`MAX` over it, `GROUP BY`,
+  `ORDER BY` and comparisons with numeric literals, with Decimal128's 34 significant
+  digits. Any other use of such a column (arithmetic, other functions, `DISTINCT`
+  aggregates) raises `NotSupportedError` rather than computing with doubles.
+
 **Important Note on Collection Names:**
 
 When using collection names containing special characters (`.`, `-`, `:`), you must wrap them in double quotes to prevent Superset's SQL parser from incorrectly interpreting them.

pymongosql/executor.py

@@ -41,6 +41,14 @@ def _run_db_command(db: Any, command: Dict[str, Any], connection: Any, operation
     )
 
 
+def _paging(limit: Any, skip: Any) -> Any:
+    """Validate bound LIMIT/OFFSET values; returns (limit, skip)."""
+    for name, value in (("LIMIT", limit), ("OFFSET", skip)):
+        if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value < 0):
+            raise ProgrammingError(f"{name} must be a non-negative integer, got {value!r}")
+    return limit, skip
+
+
 @dataclass
 class ExecutionContext:
     """Manages execution context for a single query"""
@@ -149,9 +157,16 @@ def _execute_find_plan(
             # Replace placeholders with parameters in filter_stage only (not in projection)
             filter_stage = execution_plan.filter_stage or {}
 
-            if parameters:
-                # Positional parameters with ? (named parameters are converted to positional in execute())
-                filter_stage = self._replace_placeholders(filter_stage, parameters)
+            # Positional parameters (named ones are converted to positional in execute()),
+            # in statement order: WHERE, then LIMIT, then OFFSET
+            bound, _ = SQLHelper.bind_filter(
+                {"filter": filter_stage, "limit": execution_plan.limit_stage, "skip": execution_plan.skip_stage},
+                parameters,
+            )
+            filter_stage = bound["filter"]
+            limit, skip = _paging(bound["limit"], bound["skip"])
+            if limit == 0:
+                return {"cursor": {"id": 0, "firstBatch": []}, "ok": 1}
 
             projection_stage = execution_plan.projection_stage or {}
 
@@ -170,13 +185,11 @@ def _execute_find_plan(
                         sort_spec[field_name] = direction
                 find_command["sort"] = sort_spec
 
-            # Apply skip if specified
-            if execution_plan.skip_stage:
-                find_command["skip"] = execution_plan.skip_stage
-
-            # Apply limit if specified
-            if execution_plan.limit_stage:
-                find_command["limit"] = execution_plan.limit_stage
+            # Apply skip and limit if specified (MongoDB reads limit 0 as "no limit")
+            if skip:
+                find_command["skip"] = skip
+            if limit is not None:
+                find_command["limit"] = limit
 
             _logger.debug(f"Executing MongoDB command: {find_command}")
 
@@ -223,7 +236,12 @@ def _execute_aggregate_plan(
 
             # Parse pipeline and options from JSON strings
             try:
-                pipeline = json.loads(execution_plan.aggregate_pipeline or "[]")
+                if execution_plan.aggregate_parameterized:
+                    from bson import json_util
+
+                    pipeline = json_util.loads(execution_plan.aggregate_pipeline or "[]")
+                else:
+                    pipeline = json.loads(execution_plan.aggregate_pipeline or "[]")
                 options = json.loads(execution_plan.aggregate_options or "{}")
             except json.JSONDecodeError as e:
                 raise ProgrammingError(f"Invalid JSON in aggregate pipeline or options: {e}")
@@ -232,6 +250,14 @@ def _execute_aggregate_plan(
             _logger.debug(f"Pipeline: {pipeline}")
             _logger.debug(f"Options: {options}")
 
+            # A pipeline generated from SQL carries parameter markers (WHERE, HAVING), then
+            # LIMIT and OFFSET
+            limit, skip = execution_plan.limit_stage, execution_plan.skip_stage
+            if execution_plan.aggregate_parameterized:
+                bound, _ = SQLHelper.bind_filter({"pipeline": pipeline, "limit": limit, "skip": skip}, parameters)
+                pipeline, limit, skip = bound["pipeline"], bound["limit"], bound["skip"]
+            limit, skip = _paging(limit, skip)
+
             # Get collection and call aggregate()
             collection = db[execution_plan.collection]
 
@@ -258,11 +284,11 @@ def _execute_aggregate_plan(
                         results = sorted(results, key=lambda x: x.get(field_name), reverse=reverse)
 
             # Apply skip and limit
-            if execution_plan.skip_stage:
-                results = results[execution_plan.skip_stage :]
+            if skip:
+                results = results[skip:]
 
-            if execution_plan.limit_stage:
-                results = results[: execution_plan.limit_stage]
+            if limit is not None:
+                results = results[:limit]
 
             # Apply projection if specified
             if execution_plan.projection_stage:
@@ -513,11 +539,8 @@ def _execute_execution_plan(
 
             filter_conditions = execution_plan.filter_conditions or {}
 
-            # Replace placeholders in filter if parameters provided
-            if parameters and filter_conditions:
-                filter_conditions = SQLHelper.replace_placeholders_generic(
-                    filter_conditions, parameters, execution_plan.parameter_style
-                )
+            # Bind the WHERE clause's parameter markers; every parameter must be used
+            filter_conditions, _ = SQLHelper.bind_filter(filter_conditions, parameters)
 
             command = {"delete": execution_plan.collection, "deletes": [{"q": filter_conditions, "limit": 0}]}
 
@@ -600,12 +623,10 @@ def _execute_execution_plan(
             # Replace placeholders if parameters provided
             # Note: We need to replace both update_fields and filter_conditions in one pass
             # to maintain correct parameter ordering (SET clause first, then WHERE clause)
-            if parameters:
-                # Combine structures for replacement in correct order
-                combined = {"update_fields": update_fields, "filter_conditions": filter_conditions}
-                replaced = SQLHelper.replace_placeholders_generic(combined, parameters, execution_plan.parameter_style)
-                update_fields = replaced["update_fields"]
-                filter_conditions = replaced["filter_conditions"]
+            # SET values and the WHERE clause carry parameter markers; bind them in
+            # statement order (SET first) and require every parameter to be used
+            bound, _ = SQLHelper.bind_filter({"u": update_fields, "q": filter_conditions}, parameters)
+            update_fields, filter_conditions = bound["u"], bound["q"]
 
             # MongoDB update command format
             # https://www.mongodb.com/docs/manual/reference/command/update/

pymongosql/helper.py

@@ -6,9 +6,12 @@
 """
 
 import logging
+from decimal import Decimal
 from typing import Any, Optional, Sequence, Tuple
 from urllib.parse import parse_qs, urlparse
 
+from bson import Decimal128
+
 from .error import ProgrammingError
 
 _logger = logging.getLogger(__name__)
@@ -102,6 +105,66 @@ def parse_connection_string(connection_string: Optional[str]) -> Tuple[Optional[
 class SQLHelper:
     """SQL-related helper utilities."""
 
+    @staticmethod
+    def to_bson_value(value: Any) -> Any:
+        """Convert a bound parameter to a type BSON can encode.
+
+        ``decimal.Decimal`` has no BSON encoding; ``Decimal128`` stores it exactly.
+        """
+        if isinstance(value, Decimal):
+            return Decimal128(value)
+        if isinstance(value, dict):
+            return {k: SQLHelper.to_bson_value(v) for k, v in value.items()}
+        if isinstance(value, (list, tuple)):
+            return [SQLHelper.to_bson_value(v) for v in value]
+        return value
+
+    @staticmethod
+    def bind_filter(value: Any, parameters: Any, exact: bool = True) -> Tuple[Any, int]:
+        """Bind positional parameters to the parameter markers of a translated filter.
+
+        Translated WHERE clauses mark each ``?`` with a dict marker, so a string literal
+        '?' is never taken for a parameter. With ``exact``, every parameter must be used:
+        a parameter the translation dropped (e.g. a LIMIT it could not read) must not
+        silently widen the query. Returns the bound value and the number used.
+        """
+        from .sql.where_tree import LIKE_KEY, is_like, is_param, like_operator
+
+        params = [] if parameters is None else parameters
+        if isinstance(params, dict) or not isinstance(params, Sequence) or isinstance(params, (str, bytes)):
+            raise ProgrammingError("Positional parameters must be provided as a sequence")
+        idx = [0]
+
+        def take() -> Any:
+            if idx[0] >= len(params):
+                raise ProgrammingError("Not enough parameters provided")
+            out = params[idx[0]]
+            idx[0] += 1
+            return out
+
+        def replace(val: Any) -> Any:
+            if is_param(val):
+                return SQLHelper.to_bson_value(take())
+            if is_like(val):
+                like = val[LIKE_KEY]
+                parts = []
+                for part in like["parts"]:
+                    part = take() if is_param(part) else part
+                    if not isinstance(part, str):
+                        raise ProgrammingError(f"A LIKE pattern parameter must be a string, got {part!r}")
+                    parts.append(part)
+                return like_operator("".join(parts), like["escape"], like["i"])
+            if isinstance(val, dict):
+                return {k: replace(v) for k, v in val.items()}
+            if isinstance(val, list):
+                return [replace(v) for v in val]
+            return val
+
+        bound = replace(value)
+        if exact and idx[0] != len(params):
+            raise ProgrammingError(f"{len(params)} parameters were given but the statement uses {idx[0]}")
+        return bound, idx[0]
+
     @staticmethod
     def replace_placeholders_generic(value: Any, parameters: Any, style: Optional[str]) -> Any:
         """Recursively replace placeholders in nested structures for qmark or named styles."""
@@ -120,7 +183,7 @@ def replace(val: Any) -> Any:
                         raise ProgrammingError("Not enough parameters provided")
                     out = parameters[idx[0]]
                     idx[0] += 1
-                    return out
+                    return SQLHelper.to_bson_value(out)
                 if isinstance(val, dict):
                     return {k: replace(v) for k, v in val.items()}
                 if isinstance(val, list):
@@ -138,7 +201,7 @@ def replace(val: Any) -> Any:
                     key = val[1:]
                     if key not in parameters:
                         raise ProgrammingError(f"Missing named parameter: {key}")
-                    return parameters[key]
+                    return SQLHelper.to_bson_value(parameters[key])
                 if isinstance(val, dict):
                     return {k: replace(v) for k, v in val.items()}
                 if isinstance(val, list):

pymongosql/result_set.py

@@ -4,6 +4,7 @@
 from typing import Any, Dict, List, Optional, Sequence, Tuple
 
 import jmespath
+from bson import Decimal128, Int64
 from pymongo.errors import PyMongoError
 
 from . import STRING
@@ -63,7 +64,7 @@ def _process_and_cache_batch(self, batch: List[Dict[str, Any]]) -> None:
         if not batch:
             return
         # Process results through projection mapping
-        processed_batch = [self._process_document(doc) for doc in batch]
+        processed_batch = [self._to_python(self._process_document(doc)) for doc in batch]
         # Convert dictionaries to output format (sequence or dict)
         formatted_batch = [self._format_result(doc) for doc in processed_batch]
         self._cached_results.extend(formatted_batch)
@@ -159,6 +160,24 @@ def _process_document(self, doc: Dict[str, Any]) -> Dict[str, Any]:
 
         return processed
 
+    @classmethod
+    def _to_python(cls, value: Any) -> Any:
+        """Return standard Python types for BSON-specific numbers.
+
+        DB API 2.0 consumers expect ``decimal.Decimal`` and ``int``; PyMongo returns
+        ``bson.Decimal128`` (which most libraries cannot sum or serialise) and, in
+        command responses, ``bson.Int64``.
+        """
+        if isinstance(value, Decimal128):
+            return value.to_decimal()
+        if isinstance(value, Int64):
+            return int(value)
+        if isinstance(value, dict):
+            return {k: cls._to_python(v) for k, v in value.items()}
+        if isinstance(value, list):
+            return [cls._to_python(v) for v in value]
+        return value
+
     def _mongo_to_bracket_key(self, field_path: str) -> str:
         """Convert Mongo dot-index notation to bracket notation.