From 8a890feb596c6d9307c2963579c20801a0b3daef Mon Sep 17 00:00:00 2001
From: Peng Ren <ia250@cummins.com>
Date: Fri, 24 Apr 2026 08:37:29 -0400
Subject: [PATCH 1/2] Add explain plan support

---
 README.md                         |  34 +++++
 pymongosql/executor.py            |  86 ++++++++++-
 pymongosql/sql/ast.py             |  39 +++++
 pymongosql/sql/explain_builder.py | 237 ++++++++++++++++++++++++++++++
 pymongosql/sql/parser.py          |  26 +++-
 tests/test_explain.py             | 224 ++++++++++++++++++++++++++++
 6 files changed, 642 insertions(+), 4 deletions(-)
 create mode 100644 pymongosql/sql/explain_builder.py
 create mode 100644 tests/test_explain.py

diff --git a/README.md b/README.md
index cc5e1ec..7a83b7c 100644
--- a/README.md
+++ b/README.md
@@ -92,6 +92,7 @@ pip install -e .
   - [UPDATE Statements](#update-statements)
   - [DELETE Statements](#delete-statements)
   - [View Management](#view-management)
+  - [Explain Statement](#explain-statement)
   - [Transaction Support](#transaction-support)
 - [SQL to MongoDB Mapping](#sql-to-mongodb-mapping)
 - [Apache Superset Integration](#apache-superset-integration)
@@ -542,6 +543,38 @@ cursor.execute("DROP VIEW active_users")
 
 **Note:** The pipeline must be a valid JSON array string enclosed in single quotes. `CREATE VIEW` maps to `db.command({"create": view_name, "viewOn": collection, "pipeline": [...]})` and `DROP VIEW` maps to `db.command({"drop": view_name})`.
 
+### Explain Statement
+
+Prefix any `SELECT` statement with `EXPLAIN` to inspect the MongoDB [query plan](https://www.mongodb.com/docs/compass/query-plan/). PyMongoSQL wraps the inner command with MongoDB's [`explain`](https://www.mongodb.com/docs/manual/reference/command/explain/) command and flattens the winning plan tree into a two-column result set (`stage`, `details`) that renders well in table views (e.g. Apache Superset, DB clients).
+
+```python
+# Default verbosity: queryPlanner
+cursor.execute("EXPLAIN SELECT * FROM users WHERE age > 21")
+for stage, details in cursor.fetchall():
+    print(stage, details)
+
+# Request executionStats (actual timing, docs/keys examined) or allPlansExecution
+cursor.execute("EXPLAIN (verbosity executionStats) SELECT * FROM users WHERE age > 21")
+```
+
+**Supported verbosities** (grammar-native options, unquoted identifiers):
+
+| Verbosity | What you get |
+|---|---|
+| `queryPlanner` _(default)_ | Winning plan, rejected plans, namespace, parsedQuery |
+| `executionStats` | `queryPlanner` output + execution time, documents returned/examined, index keys examined |
+| `allPlansExecution` | `executionStats` for the winning plan **and** each rejected candidate plan |
+
+Example output rows (`queryPlanner`):
+
+```
+stage                  details
+namespace              mydb.users
+parsedQuery            {"age": {"$gt": 21}}
+├─ COLLSCAN            {"direction": "forward", "filter": {...}}
+rejectedPlans          []
+```
+
 ### Transaction Support
 
 PyMongoSQL supports DB API 2.0 transactions for ACID-compliant database operations. Use the `begin()`, `commit()`, and `rollback()` methods to manage transactions:
@@ -588,6 +621,7 @@ The table below shows how PyMongoSQL translates SQL operations into MongoDB comm
 | `DELETE FROM col WHERE ...` | `{delete: col, deletes: [{q: filter, limit: 0}]}` | `db.command("delete", ...)` |
 | `CREATE VIEW v ON col AS '[...]'` | `{create: v, viewOn: col, pipeline: [...]}` | `db.command("create", ...)` |
 | `DROP VIEW v` | `{drop: v}` | `db.command("drop", ...)` |
+| `EXPLAIN <select>` | `{explain: <find\|aggregate cmd>, verbosity: "queryPlanner"}` | `db.command("explain", ...)` |
 
 ### SQL Clauses to MongoDB Query Components
 
diff --git a/pymongosql/executor.py b/pymongosql/executor.py
index 12ac7a6..80f0650 100644
--- a/pymongosql/executor.py
+++ b/pymongosql/executor.py
@@ -11,6 +11,7 @@
 from .helper import SQLHelper
 from .retry import execute_with_retry
 from .sql.delete_builder import DeleteExecutionPlan
+from .sql.explain_builder import ExplainExecutionPlan
 from .sql.insert_builder import InsertExecutionPlan
 from .sql.parser import SQLParser
 from .sql.query_builder import QueryExecutionPlan
@@ -758,10 +759,93 @@ def execute(
         return self._execute_execution_plan(self._execution_plan, connection, parameters)
 
 
+class ExplainExecution(ExecutionStrategy):
+    """Execution strategy for ``EXPLAIN [ (opt val, ...) ] <statement>`` wrappers.
+
+    Parses via :class:`SQLParser` (grammar-native EXPLAIN production) to obtain
+    an :class:`ExplainExecutionPlan`, delegates command construction and result
+    flattening to the plan, and runs the resulting ``explain`` command through
+    the shared connection/retry path.
+    """
+
+    _EXPLAIN_PATTERN = re.compile(r"^\s*EXPLAIN\b", re.IGNORECASE)
+
+    @property
+    def execution_plan(self) -> QueryExecutionPlan:
+        return self._execution_plan
+
+    def supports(self, context: ExecutionContext) -> bool:
+        return bool(self._EXPLAIN_PATTERN.match(context.query))
+
+    def _parse_sql(self, sql: str) -> ExplainExecutionPlan:
+        try:
+            parser = SQLParser(sql)
+            plan = parser.get_execution_plan()
+            if not isinstance(plan, ExplainExecutionPlan):
+                raise SqlSyntaxError("Expected EXPLAIN execution plan")
+            if not plan.validate():
+                raise SqlSyntaxError("Generated EXPLAIN plan is invalid")
+            return plan
+        except SqlSyntaxError:
+            raise
+        except Exception as e:
+            _logger.error(f"SQL parsing failed: {e}")
+            raise SqlSyntaxError(f"Failed to parse SQL: {e}")
+
+    def execute(
+        self,
+        context: ExecutionContext,
+        connection: Any,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
+    ) -> Optional[Dict[str, Any]]:
+        _logger.debug(f"Using explain execution for query: {context.query[:100]}")
+
+        # Normalize named parameters to positional, matching StandardQueryExecution.
+        processed_query = context.query
+        processed_params = parameters
+        if isinstance(parameters, dict):
+            param_names = re.findall(r":(\w+)", context.query)
+            processed_params = [parameters[name] for name in param_names]
+            processed_query = re.sub(r":(\w+)", "?", context.query)
+
+        explain_plan = self._parse_sql(processed_query)
+        # Store the synthesized result plan (QueryExecutionPlan) so the cursor
+        # can wire it directly into the ResultSet for column description.
+        self._execution_plan = explain_plan.result_plan
+
+        # Build the explain command (validates inner plan is a supported SELECT).
+        explain_cmd = explain_plan.build_command(processed_params)
+
+        if not connection:
+            raise OperationalError("No connection provided")
+
+        _logger.debug(f"Executing MongoDB explain command: {explain_cmd}")
+
+        try:
+            explain_result = _run_db_command(connection.database, explain_cmd, connection, "explain command")
+        except PyMongoError as e:
+            _logger.error(f"MongoDB explain execution failed: {e}")
+            raise DatabaseError(f"Explain execution failed: {e}")
+
+        # Return flattened rows as a command result. The cursor handles
+        # ExplainExecutionPlan -> result_plan translation when wiring the ResultSet.
+        return {
+            "cursor": {"id": 0, "firstBatch": explain_plan.flatten_result(explain_result)},
+            "ok": 1,
+        }
+
+
 class ExecutionPlanFactory:
     """Factory for creating appropriate execution strategy based on query context"""
 
-    _strategies = [ViewExecution(), StandardQueryExecution(), InsertExecution(), UpdateExecution(), DeleteExecution()]
+    _strategies = [
+        ExplainExecution(),
+        ViewExecution(),
+        StandardQueryExecution(),
+        InsertExecution(),
+        UpdateExecution(),
+        DeleteExecution(),
+    ]
 
     @classmethod
     def get_strategy(cls, context: ExecutionContext) -> ExecutionStrategy:
diff --git a/pymongosql/sql/ast.py b/pymongosql/sql/ast.py
index 0ab473b..9d2372f 100644
--- a/pymongosql/sql/ast.py
+++ b/pymongosql/sql/ast.py
@@ -38,6 +38,9 @@ def __init__(self) -> None:
         self._update_parse_result = UpdateParseResult.for_visitor()
         # Track current statement kind generically so UPDATE/DELETE can reuse this
         self._current_operation: str = "select"  # expected values: select | insert | update | delete
+        # EXPLAIN wrapper state (grammar: root : EXPLAIN? (options)? statement EOF)
+        self._is_explain: bool = False
+        self._explain_options: Dict[str, str] = {}
         self._handlers = self._initialize_handlers()
 
     def _initialize_handlers(self) -> Dict[str, BaseHandler]:
@@ -69,11 +72,47 @@ def current_operation(self) -> str:
         """Get the current operation type (select, insert, delete, or update)"""
         return self._current_operation
 
+    @property
+    def is_explain(self) -> bool:
+        """Whether the parsed statement was wrapped in EXPLAIN."""
+        return self._is_explain
+
+    @property
+    def explain_options(self) -> Dict[str, str]:
+        """Options collected from ``EXPLAIN (key value, ...)`` clause."""
+        return dict(self._explain_options)
+
     def visitRoot(self, ctx: PartiQLParser.RootContext) -> Any:
         """Visit root node and process child nodes"""
         _logger.debug("Starting to parse SQL query")
         # Reset to default SELECT operation at the start of each query
         self._current_operation = "select"
+        self._is_explain = False
+        self._explain_options = {}
+
+        # Detect EXPLAIN wrapper per grammar:
+        #   root : (EXPLAIN (PAREN_LEFT explainOption (COMMA explainOption)* PAREN_RIGHT)? )? statement EOF;
+        try:
+            explain_token = ctx.EXPLAIN() if hasattr(ctx, "EXPLAIN") else None
+        except Exception:
+            explain_token = None
+
+        if explain_token is not None:
+            self._is_explain = True
+            try:
+                option_ctxs = ctx.explainOption() or []
+            except Exception:
+                option_ctxs = []
+            for opt in option_ctxs:
+                try:
+                    param_tok = getattr(opt, "param", None)
+                    value_tok = getattr(opt, "value", None)
+                    if param_tok is not None and value_tok is not None:
+                        self._explain_options[param_tok.text] = value_tok.text
+                except Exception as e:
+                    _logger.warning(f"Error parsing explainOption: {e}")
+            _logger.debug(f"EXPLAIN detected with options: {self._explain_options}")
+
         try:
             result = self.visitChildren(ctx)
             return result
diff --git a/pymongosql/sql/explain_builder.py b/pymongosql/sql/explain_builder.py
new file mode 100644
index 0000000..204b3dd
--- /dev/null
+++ b/pymongosql/sql/explain_builder.py
@@ -0,0 +1,237 @@
+# -*- coding: utf-8 -*-
+import json
+import logging
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Sequence, Union
+
+from ..error import ProgrammingError, SqlSyntaxError
+from ..helper import SQLHelper
+from .builder import ExecutionPlan
+from .query_builder import QueryExecutionPlan
+
+_logger = logging.getLogger(__name__)
+
+
+_ALLOWED_VERBOSITIES = frozenset({"queryPlanner", "executionStats", "allPlansExecution"})
+
+
+@dataclass
+class ExplainExecutionPlan(ExecutionPlan):
+    """Execution plan that wraps another plan with server-side explain semantics."""
+
+    inner_plan: Optional[ExecutionPlan] = None
+    verbosity: str = "queryPlanner"
+    options: Dict[str, str] = field(default_factory=dict)
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "collection": self.collection,
+            "verbosity": self.verbosity,
+            "options": dict(self.options),
+            "inner_plan": self.inner_plan.to_dict() if self.inner_plan else None,
+        }
+
+    def validate(self) -> bool:
+        errors: list[str] = []
+
+        if self.inner_plan is None:
+            errors.append("EXPLAIN requires an inner statement")
+        elif not self.inner_plan.validate():
+            errors.append("Inner statement for EXPLAIN is invalid")
+
+        if self.verbosity not in _ALLOWED_VERBOSITIES:
+            errors.append(
+                f"Unsupported EXPLAIN verbosity '{self.verbosity}'. " f"Allowed: {sorted(_ALLOWED_VERBOSITIES)}"
+            )
+
+        if errors:
+            for err in errors:
+                _logger.warning(f"Explain plan validation error: {err}")
+            return False
+
+        return True
+
+    # ------------------------------------------------------------------ #
+    # Command construction
+    # ------------------------------------------------------------------ #
+    def build_inner_command(
+        self,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
+    ) -> Dict[str, Any]:
+        """Build the MongoDB command (``find`` or ``aggregate``) to be explained.
+
+        Only :class:`QueryExecutionPlan` inner plans are currently supported.
+        Placeholders in the filter stage are substituted using ``qmark`` style.
+        """
+        inner_plan = self.inner_plan
+        if not isinstance(inner_plan, QueryExecutionPlan):
+            raise SqlSyntaxError(
+                "EXPLAIN currently only supports SELECT statements "
+                f"(got {type(inner_plan).__name__ if inner_plan is not None else 'None'})"
+            )
+
+        if getattr(inner_plan, "is_aggregate_query", False):
+            try:
+                pipeline = json.loads(inner_plan.aggregate_pipeline or "[]")
+                options = json.loads(inner_plan.aggregate_options or "{}")
+            except json.JSONDecodeError as e:
+                raise ProgrammingError(f"Invalid JSON in aggregate pipeline or options: {e}")
+
+            command: Dict[str, Any] = {
+                "aggregate": inner_plan.collection,
+                "pipeline": pipeline,
+                "cursor": {},
+            }
+            for k, v in options.items():
+                if k not in command:
+                    command[k] = v
+            return command
+
+        if not inner_plan.collection:
+            raise ProgrammingError("No collection specified in query")
+
+        filter_stage = inner_plan.filter_stage or {}
+        if parameters:
+            filter_stage = SQLHelper.replace_placeholders_generic(filter_stage, parameters, "qmark")
+
+        command = {"find": inner_plan.collection, "filter": filter_stage}
+
+        if inner_plan.projection_stage:
+            command["projection"] = inner_plan.projection_stage
+        if inner_plan.sort_stage:
+            sort_spec: Dict[str, int] = {}
+            for sort_dict in inner_plan.sort_stage:
+                for field_name, direction in sort_dict.items():
+                    sort_spec[field_name] = direction
+            command["sort"] = sort_spec
+        if inner_plan.skip_stage:
+            command["skip"] = inner_plan.skip_stage
+        if inner_plan.limit_stage:
+            command["limit"] = inner_plan.limit_stage
+
+        return command
+
+    def build_command(
+        self,
+        parameters: Optional[Union[Sequence[Any], Dict[str, Any]]] = None,
+    ) -> Dict[str, Any]:
+        """Build the server-side ``{explain: <inner>, verbosity: <mode>}`` command."""
+        return {"explain": self.build_inner_command(parameters), "verbosity": self.verbosity}
+
+    # ------------------------------------------------------------------ #
+    # Result-set shaping
+    # ------------------------------------------------------------------ #
+    @property
+    def result_plan(self) -> QueryExecutionPlan:
+        """Synthetic :class:`QueryExecutionPlan` describing the ``(stage, details)`` result rows.
+
+        Used by :class:`~pymongosql.result_set.ResultSet` to build a stable column
+        description and row ordering for EXPLAIN output.
+        """
+        collection = (self.inner_plan.collection if self.inner_plan else None) or "explain"
+        return QueryExecutionPlan(
+            collection=collection,
+            projection_stage={"stage": 1, "details": 1},
+        )
+
+    @staticmethod
+    def flatten_result(explain_result: Dict[str, Any]) -> List[Dict[str, str]]:
+        """Flatten a MongoDB explain result into rows with ``stage`` and ``details`` columns.
+
+        The winning plan tree (from ``queryPlanner.winningPlan`` or, for aggregate
+        pipelines, ``stages[*].$cursor.queryPlanner.winningPlan``) is rendered as
+        an indented tree. Pipeline stages and high-level metadata (namespace,
+        rejected plan count, executionStats summary) are included as header /
+        footer rows.
+        """
+        rows: List[Dict[str, str]] = []
+
+        def fmt_details(d: Dict[str, Any]) -> str:
+            if not d:
+                return ""
+            try:
+                return json.dumps(d, default=str, ensure_ascii=False)
+            except (TypeError, ValueError):
+                return str(d)
+
+        def walk_plan(node: Dict[str, Any], depth: int, is_last: bool, prefix: str) -> None:
+            if not isinstance(node, dict):
+                return
+            stage_name = node.get("stage") or "UNKNOWN"
+            if depth == 0:
+                label = stage_name
+                child_prefix = ""
+            else:
+                connector = "└─ " if is_last else "├─ "
+                label = prefix + connector + stage_name
+                child_prefix = prefix + ("   " if is_last else "│  ")
+
+            details = {k: v for k, v in node.items() if k not in ("stage", "inputStage", "inputStages")}
+            rows.append({"stage": label, "details": fmt_details(details)})
+
+            children = []
+            if isinstance(node.get("inputStage"), dict):
+                children.append(node["inputStage"])
+            if isinstance(node.get("inputStages"), list):
+                children.extend(c for c in node["inputStages"] if isinstance(c, dict))
+            for i, child in enumerate(children):
+                walk_plan(child, depth + 1, i == len(children) - 1, child_prefix)
+
+        # Header metadata
+        qp = explain_result.get("queryPlanner") if isinstance(explain_result, dict) else None
+        if isinstance(qp, dict):
+            ns = qp.get("namespace")
+            if ns:
+                rows.append({"stage": "namespace", "details": str(ns)})
+            if "parsedQuery" in qp:
+                rows.append({"stage": "parsedQuery", "details": fmt_details(qp.get("parsedQuery") or {})})
+            rejected = qp.get("rejectedPlans") or []
+            if rejected:
+                rows.append({"stage": "rejectedPlans", "details": str(len(rejected))})
+
+            winning = qp.get("winningPlan")
+            if isinstance(winning, dict):
+                walk_plan(winning, 0, True, "")
+
+        # Aggregate pipeline explain: stages list with embedded $cursor plan
+        stages = explain_result.get("stages") if isinstance(explain_result, dict) else None
+        if isinstance(stages, list) and stages:
+            for idx, st in enumerate(stages):
+                if not isinstance(st, dict):
+                    continue
+                for stage_key, stage_val in st.items():
+                    rows.append({"stage": f"pipeline[{idx}]: {stage_key}", "details": ""})
+                    if stage_key == "$cursor" and isinstance(stage_val, dict):
+                        cur_qp = stage_val.get("queryPlanner", {})
+                        ns = cur_qp.get("namespace")
+                        if ns:
+                            rows.append({"stage": "  namespace", "details": str(ns)})
+                        winning = cur_qp.get("winningPlan")
+                        if isinstance(winning, dict):
+                            walk_plan(winning, 0, True, "  ")
+                    else:
+                        detail_val = stage_val if isinstance(stage_val, dict) else {"value": stage_val}
+                        rows.append({"stage": f"  {stage_key}", "details": fmt_details(detail_val)})
+
+        # Execution stats summary (when verbosity >= executionStats)
+        exec_stats = explain_result.get("executionStats") if isinstance(explain_result, dict) else None
+        if isinstance(exec_stats, dict):
+            summary_keys = (
+                "executionSuccess",
+                "nReturned",
+                "executionTimeMillis",
+                "totalKeysExamined",
+                "totalDocsExamined",
+            )
+            summary = {k: exec_stats.get(k) for k in summary_keys if k in exec_stats}
+            if summary:
+                rows.append({"stage": "executionStats", "details": fmt_details(summary)})
+
+        # Fallback: if we produced nothing, dump the raw explain result
+        if not rows:
+            rows.append({"stage": "explain", "details": fmt_details(explain_result or {})})
+
+        return rows
+
+
+__all__ = ["ExplainExecutionPlan"]
diff --git a/pymongosql/sql/parser.py b/pymongosql/sql/parser.py
index 33dc607..7dc39d6 100644
--- a/pymongosql/sql/parser.py
+++ b/pymongosql/sql/parser.py
@@ -10,6 +10,7 @@
 from .ast import MongoSQLLexer, MongoSQLParser, MongoSQLParserVisitor
 from .builder import ExecutionPlanBuilder
 from .delete_builder import DeleteExecutionPlan
+from .explain_builder import ExplainExecutionPlan
 from .insert_builder import InsertExecutionPlan
 from .query_builder import QueryExecutionPlan
 from .update_builder import UpdateExecutionPlan
@@ -132,8 +133,8 @@ def _validate_ast(self) -> None:
 
     def get_execution_plan(
         self,
-    ) -> Union[QueryExecutionPlan, InsertExecutionPlan, DeleteExecutionPlan, UpdateExecutionPlan]:
-        """Parse SQL and return an execution plan (SELECT, INSERT, DELETE, or UPDATE)."""
+    ) -> Union[QueryExecutionPlan, InsertExecutionPlan, DeleteExecutionPlan, UpdateExecutionPlan, ExplainExecutionPlan]:
+        """Parse SQL and return an execution plan (SELECT, INSERT, DELETE, UPDATE, or EXPLAIN)."""
         if self._ast is None:
             raise SqlSyntaxError("No AST available - parsing may have failed")
 
@@ -141,7 +142,7 @@ def get_execution_plan(
             self._visitor = MongoSQLParserVisitor()
             self._visitor.visit(self._ast)
 
-            # Use ExecutionPlanBuilder to create the plan from parse result
+            # Use ExecutionPlanBuilder to create the inner plan from parse result
             execution_plan = ExecutionPlanBuilder.build_from_parse_result(
                 self._visitor.parse_result, self._visitor.current_operation
             )
@@ -149,6 +150,25 @@ def get_execution_plan(
             if not execution_plan.validate():
                 raise SqlSyntaxError("Generated execution plan is invalid")
 
+            # If the statement was wrapped in EXPLAIN, wrap the inner plan.
+            if getattr(self._visitor, "is_explain", False):
+                options = self._visitor.explain_options
+                # Option key names are case-insensitive; normalize to lower-case lookups.
+                normalized = {k.lower(): v for k, v in options.items()}
+                verbosity = normalized.get("verbosity", "queryPlanner")
+                explain_plan = ExplainExecutionPlan(
+                    collection=execution_plan.collection,
+                    inner_plan=execution_plan,
+                    verbosity=verbosity,
+                    options=options,
+                )
+                if not explain_plan.validate():
+                    raise SqlSyntaxError("Generated EXPLAIN plan is invalid")
+                _logger.debug(
+                    f"Generated EXPLAIN plan (verbosity={verbosity}) for collection: {execution_plan.collection}"
+                )
+                return explain_plan
+
             _logger.debug(f"Generated execution plan for collection: {execution_plan.collection}")
             return execution_plan
 
diff --git a/tests/test_explain.py b/tests/test_explain.py
new file mode 100644
index 0000000..bf07a1c
--- /dev/null
+++ b/tests/test_explain.py
@@ -0,0 +1,224 @@
+# -*- coding: utf-8 -*-
+import json
+
+import pytest
+
+from pymongosql.error import SqlSyntaxError
+from pymongosql.executor import (
+    ExecutionContext,
+    ExecutionPlanFactory,
+    ExplainExecution,
+    StandardQueryExecution,
+)
+from pymongosql.sql.explain_builder import ExplainExecutionPlan
+
+
+class TestExplainParserUnit:
+    """Unit tests for EXPLAIN routing and rendering (no MongoDB required)."""
+
+    def setup_method(self):
+        self.strategy = ExplainExecution()
+
+    def test_supports_uppercase(self):
+        assert self.strategy.supports(ExecutionContext("EXPLAIN SELECT * FROM users"))
+
+    def test_supports_lowercase(self):
+        assert self.strategy.supports(ExecutionContext("explain select * from users"))
+
+    def test_supports_leading_whitespace(self):
+        assert self.strategy.supports(ExecutionContext("   EXPLAIN\n  SELECT 1"))
+
+    def test_does_not_support_plain_select(self):
+        assert not self.strategy.supports(ExecutionContext("SELECT * FROM users"))
+
+    def test_does_not_support_explain_substring(self):
+        # Must be a leading keyword, not just present in the query.
+        assert not self.strategy.supports(ExecutionContext("SELECT 'EXPLAIN' FROM users"))
+
+    def test_factory_routes_to_explain(self):
+        strategy = ExecutionPlanFactory.get_strategy(ExecutionContext("EXPLAIN SELECT * FROM users"))
+        assert isinstance(strategy, ExplainExecution)
+
+    def test_factory_still_routes_plain_select(self):
+        strategy = ExecutionPlanFactory.get_strategy(ExecutionContext("SELECT * FROM users"))
+        assert isinstance(strategy, StandardQueryExecution)
+
+
+class TestExplainFlatten:
+    """Unit tests for the explain-tree flattening / table formatting."""
+
+    def test_flatten_find_with_winning_plan(self):
+        explain_result = {
+            "queryPlanner": {
+                "namespace": "test_db.users",
+                "parsedQuery": {"age": {"$gt": 25}},
+                "winningPlan": {
+                    "stage": "LIMIT",
+                    "limitAmount": 10,
+                    "inputStage": {
+                        "stage": "FETCH",
+                        "inputStage": {
+                            "stage": "IXSCAN",
+                            "indexName": "age_1",
+                            "direction": "forward",
+                        },
+                    },
+                },
+                "rejectedPlans": [{"dummy": 1}, {"dummy": 2}],
+            },
+        }
+        rows = ExplainExecutionPlan.flatten_result(explain_result)
+        stages = [r["stage"] for r in rows]
+
+        assert "namespace" in stages
+        assert "parsedQuery" in stages
+        assert "rejectedPlans" in stages
+        # Tree structure (root has no tree marker, children use └─ / ├─)
+        assert "LIMIT" in stages
+        assert any(s.endswith("FETCH") and "└─" in s for s in stages)
+        assert any(s.endswith("IXSCAN") and "└─" in s for s in stages)
+
+        # rejectedPlans row reports count
+        rejected_row = next(r for r in rows if r["stage"] == "rejectedPlans")
+        assert rejected_row["details"] == "2"
+
+        # IXSCAN details should carry indexName / direction as JSON
+        ixscan_row = next(r for r in rows if r["stage"].endswith("IXSCAN"))
+        parsed = json.loads(ixscan_row["details"])
+        assert parsed["indexName"] == "age_1"
+        assert parsed["direction"] == "forward"
+
+    def test_flatten_multiple_input_stages(self):
+        explain_result = {
+            "queryPlanner": {
+                "namespace": "db.coll",
+                "winningPlan": {
+                    "stage": "OR",
+                    "inputStages": [
+                        {"stage": "IXSCAN", "indexName": "a_1"},
+                        {"stage": "IXSCAN", "indexName": "b_1"},
+                    ],
+                },
+            },
+        }
+        rows = ExplainExecutionPlan.flatten_result(explain_result)
+        stages = [r["stage"] for r in rows]
+        # First branch uses ├─, last uses └─
+        assert any("├─ IXSCAN" in s for s in stages)
+        assert any("└─ IXSCAN" in s for s in stages)
+
+    def test_flatten_execution_stats_summary(self):
+        explain_result = {
+            "queryPlanner": {
+                "namespace": "db.coll",
+                "winningPlan": {"stage": "COLLSCAN"},
+            },
+            "executionStats": {
+                "executionSuccess": True,
+                "nReturned": 5,
+                "executionTimeMillis": 3,
+                "totalKeysExamined": 0,
+                "totalDocsExamined": 42,
+            },
+        }
+        rows = ExplainExecutionPlan.flatten_result(explain_result)
+        summary_row = next(r for r in rows if r["stage"] == "executionStats")
+        parsed = json.loads(summary_row["details"])
+        assert parsed["nReturned"] == 5
+        assert parsed["totalDocsExamined"] == 42
+
+    def test_flatten_aggregate_stages(self):
+        explain_result = {
+            "stages": [
+                {
+                    "$cursor": {
+                        "queryPlanner": {
+                            "namespace": "db.coll",
+                            "winningPlan": {"stage": "COLLSCAN"},
+                        }
+                    }
+                },
+                {"$group": {"_id": "$category", "count": {"$sum": 1}}},
+            ],
+        }
+        rows = ExplainExecutionPlan.flatten_result(explain_result)
+        stages = [r["stage"] for r in rows]
+        assert any(s.startswith("pipeline[0]: $cursor") for s in stages)
+        assert any(s.startswith("pipeline[1]: $group") for s in stages)
+        assert any("COLLSCAN" in s for s in stages)
+
+    def test_flatten_fallback_on_empty(self):
+        rows = ExplainExecutionPlan.flatten_result({})
+        assert len(rows) == 1
+        assert rows[0]["stage"] == "explain"
+
+
+class TestExplainVerbosity:
+    """Tests for the optional ``EXPLAIN (key value, ...)`` option clause (grammar-native)."""
+
+    def test_invalid_verbosity_raises(self):
+        """An unsupported verbosity identifier fails validation at parse time."""
+        from pymongosql.sql.parser import SQLParser
+
+        with pytest.raises(SqlSyntaxError):
+            SQLParser("EXPLAIN (verbosity bogus) SELECT * FROM users").get_execution_plan()
+
+    def test_non_select_inner_raises(self):
+        """EXPLAIN of a non-SELECT inner statement is rejected at execute time."""
+        strategy = ExplainExecution()
+        ctx = ExecutionContext("EXPLAIN DELETE FROM users WHERE id = 1")
+        with pytest.raises(SqlSyntaxError):
+            strategy.execute(ctx, connection=None)
+
+    def test_verbosity_option_is_parsed(self):
+        """``EXPLAIN (verbosity executionStats) SELECT ...`` produces the expected plan."""
+        from pymongosql.sql.parser import SQLParser
+
+        plan = SQLParser("EXPLAIN (verbosity executionStats) SELECT * FROM users").get_execution_plan()
+        assert isinstance(plan, ExplainExecutionPlan)
+        assert plan.verbosity == "executionStats"
+        assert plan.inner_plan is not None
+        assert plan.inner_plan.collection == "users"
+
+    def test_bare_explain_defaults_to_queryPlanner(self):
+        from pymongosql.sql.parser import SQLParser
+
+        plan = SQLParser("EXPLAIN SELECT name FROM users WHERE age > 25").get_execution_plan()
+        assert isinstance(plan, ExplainExecutionPlan)
+        assert plan.verbosity == "queryPlanner"
+        assert plan.inner_plan.collection == "users"
+
+    def test_explain_options_are_preserved(self):
+        """All option pairs are preserved on the plan even if only 'verbosity' is recognized."""
+        from pymongosql.sql.parser import SQLParser
+
+        plan = SQLParser("EXPLAIN (verbosity queryPlanner, format tree) SELECT * FROM users").get_execution_plan()
+        assert plan.options == {"verbosity": "queryPlanner", "format": "tree"}
+
+
+@pytest.mark.integration
+class TestExplainIntegration:
+    """Integration tests that require a running MongoDB (use the ``conn`` fixture)."""
+
+    def test_explain_simple_select(self, conn):
+        cursor = conn.cursor()
+        cursor.execute("EXPLAIN SELECT name, email FROM users WHERE age > 25")
+        rows = cursor.fetchall()
+        assert len(rows) > 0
+        col_names = [d[0] for d in cursor.description]
+        assert col_names == ["stage", "details"]
+        # Winning plan must contribute at least one plan-stage row
+        stage_col_idx = col_names.index("stage")
+        stage_values = [r[stage_col_idx] for r in rows]
+        assert any(
+            any(tok in s for tok in ("COLLSCAN", "IXSCAN", "FETCH", "LIMIT", "SORT", "PROJECTION"))
+            for s in stage_values
+        )
+
+    def test_explain_with_execution_stats(self, conn):
+        cursor = conn.cursor()
+        cursor.execute("EXPLAIN (verbosity executionStats) SELECT * FROM users LIMIT 5")
+        rows = cursor.fetchall()
+        col_names = [d[0] for d in cursor.description]
+        stage_idx = col_names.index("stage")
+        assert any(r[stage_idx] == "executionStats" for r in rows)

From eb897e86bf8a5cc944ded336af1c02ddbb1bb3db Mon Sep 17 00:00:00 2001
From: Peng Ren <ia250@cummins.com>
Date: Fri, 24 Apr 2026 13:18:32 -0400
Subject: [PATCH 2/2] Bump version to 0.7.0

---
 README.md              | 10 +++++-----
 pymongosql/__init__.py |  2 +-
 pyproject.toml         |  3 ++-
 3 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/README.md b/README.md
index 7a83b7c..6884171 100644
--- a/README.md
+++ b/README.md
@@ -5,11 +5,11 @@
 [![Code Style](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)
 [![codecov](https://codecov.io/gh/passren/PyMongoSQL/branch/main/graph/badge.svg?token=2CTRL80NP2)](https://codecov.io/gh/passren/PyMongoSQL)
 [![License: MIT](https://img.shields.io/badge/License-MIT-purple.svg)](https://github.com/passren/PyMongoSQL/blob/0.1.2/LICENSE)
-[![Python Version](https://img.shields.io/badge/python-3.9+-blue.svg)](https://www.python.org/downloads/)
+[![Python Version](https://img.shields.io/badge/python-3.9|3.10|3.11|3.12|3.13|3.14-blue.svg)](https://www.python.org/downloads/)
 [![Downloads](https://static.pepy.tech/badge/pymongosql/month)](https://pepy.tech/projects/pymongosql)
-[![MongoDB](https://img.shields.io/badge/MongoDB-7.0+-green.svg)](https://www.mongodb.com/)
-[![SQLAlchemy](https://img.shields.io/badge/SQLAlchemy-1.4+_2.0+-darkgreen.svg)](https://www.sqlalchemy.org/)
-[![Superset](https://img.shields.io/badge/Apache_Superset-1.0+-blue.svg)](https://superset.apache.org/docs/6.0.0/configuration/databases)
+[![MongoDB](https://img.shields.io/badge/MongoDB-7.0+|8.0+-green.svg)](https://www.mongodb.com/)
+[![SQLAlchemy](https://img.shields.io/badge/SQLAlchemy-1.4+|2.0+-darkgreen.svg)](https://www.sqlalchemy.org/)
+[![Superset](https://img.shields.io/badge/Apache_Superset->1.0-blue.svg)](https://superset.apache.org/docs/6.0.0/configuration/databases)
 
 PyMongoSQL is a Python [DB API 2.0 (PEP 249)](https://www.python.org/dev/peps/pep-0249/) client for [MongoDB](https://www.mongodb.com/). It provides a familiar SQL interface to MongoDB, allowing developers to use SQL to interact with MongoDB collections.
 
@@ -35,7 +35,7 @@ PyMongoSQL implements the DB API 2.0 interfaces to provide SQL-like access to Mo
 
 ## Requirements
 
-- **Python**: 3.9, 3.10, 3.11, 3.12, 3.13+
+- **Python**: 3.9, 3.10, 3.11, 3.12, 3.13, 3.14
 - **MongoDB**: 7.0+
 
 ## Dependencies
diff --git a/pymongosql/__init__.py b/pymongosql/__init__.py
index 64603b0..700f368 100644
--- a/pymongosql/__init__.py
+++ b/pymongosql/__init__.py
@@ -6,7 +6,7 @@
 if TYPE_CHECKING:
     from .connection import Connection
 
-__version__: str = "0.6.0"
+__version__: str = "0.7.0"
 
 # Globals https://www.python.org/dev/peps/pep-0249/#globals
 apilevel: str = "2.0"
diff --git a/pyproject.toml b/pyproject.toml
index d083fac..d78614b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -16,7 +16,7 @@ maintainers = [
 ]
 keywords = ["mongodb", "sql", "database", "dbapi", "nosql"]
 classifiers = [
-    "Development Status :: 3 - Alpha",
+    "Development Status :: 4 - Beta",
     "Intended Audience :: Developers",
     "License :: OSI Approved :: MIT License",
     "Operating System :: OS Independent",
@@ -26,6 +26,7 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13",
+    "Programming Language :: Python :: 3.14",
     "Topic :: Database",
     "Topic :: Database :: Database Engines/Servers",
     "Topic :: Software Development :: Libraries :: Python Modules",