fix(mcp): execute_sql request limit caps, never raises, an explicit SQL LIMIT

UPDATING.md

@@ -80,6 +80,32 @@ format. Use a consistent timezone and precision for stored strings and bounds.
   able to provision a user regardless of `AUTH_USER_REGISTRATION`; it now
   requires the same gate as `/register/`.
 
+### SQL execution request limits are safety caps
+
+The MCP `execute_sql` request's `limit` no longer overrides a smaller outer SQL
+LIMIT. It caps the last statement at the smaller of the SQL LIMIT and the request
+limit, subject to `SQL_MAX_ROW`. For example, SQL `LIMIT 5` with request `limit: 10`
+keeps `LIMIT 5`. To return more rows, increase or remove the SQL LIMIT explicitly.
+Omitting the request limit still leaves SQL limits unchanged. Inner query and CTE
+limits are not changed.
+
+This contract change also applies to the public `Database.execute()` and
+`Database.execute_async()` APIs through `QueryOptions.limit`, including dry runs.
+
+SQL Lab also recognizes literal `FETCH FIRST` counts and parenthesized literal
+limits. SQL Server `TOP ... PERCENT` and `TOP ... WITH TIES` are no longer
+interpreted as fixed row counts: SQL Lab uses the row-limit dropdown (subject to
+server limits) and rewrites these clauses to a fixed `TOP` cap. For example,
+`TOP 5 PERCENT` with a 1000-row dropdown can return up to 1000 rows rather than
+the previous five.
+
+`BigQueryEngineSpec.fetch_data` no longer falls back to a plain `fetchall()` once
+it has read its initial sample, since a forward-only cursor cannot replay those
+rows. A BigQuery driver error after the sample now fails the query instead of
+returning the remaining rows as a success. This applies to every caller,
+including legacy SQL Lab execution and dataset column discovery. An error on the
+initial read still falls back as before.
+
 ### Doris SSL requests require TLS
 
 The Doris SSL toggle uses `ssl_mode=VERIFY_CA`. Saved `ssl_mode=REQUIRED` and

docs/admin_docs/configuration/mcp-server.mdx

@@ -1352,6 +1352,52 @@ session scope, without reloading the request user. Reporting an error therefore
 does not require metadata database access unless the hook itself uses it. Use the
 provided hook context for caller identity rather than `g.user`.
 
+## SQL execution row limits
+
+The `execute_sql` tool treats an explicit request `limit` as a safety cap on the
+last statement's outer SQL LIMIT. The smaller limit wins: SQL `LIMIT 5` with
+request `limit: 10` returns at most five rows, while SQL `LIMIT 10` with request
+`limit: 5` also returns at most five. Equal limits and SQL `LIMIT 0` are preserved.
+If SQL has no outer limit, the explicit request cap is applied. Inner query and
+CTE limits and SQL offsets are preserved. Literal `FETCH FIRST` counts and
+parenthesized literal limits follow the same rule. `PERCENT`, `WITH TIES`, and
+limit expressions that cannot be interpreted as a fixed row count are preserved
+inside a subquery with an outer row cap, rather than replaced. This conservative
+fallback cannot increase the original result's row count. For SQL Server, these
+restrictions are left unchanged instead: a derived table could reject duplicate
+or unnamed output columns. The executor caps the last statement's returned rows
+at the smaller of the explicit request limit and `SQL_MAX_ROW`, without changing
+percentage, tie, or expression semantics. Cursor reads are bounded by the same
+cap plus one probe row, including across multiple fetch batches, preserving
+engine-specific result processing. Only successfully returned rows count against the cursor
+budget, so a failed fetch leaves that budget available for an engine fallback.
+BigQuery only falls back if its initial fetch fails; errors after a successful
+sample, including during the EOF probe, fail the query rather than silently
+discarding buffered rows.
+This does not bound the work performed by the database.
+Because this cap applies to any row-returning last statement, a non-query
+statement such as `SHOW TABLES` or DML with `RETURNING` can also be capped. Each
+statement reports `truncated: true` when a fetch cap omits rows from the
+executed SQL result, including engine memory caps such as `BQ_FETCH_MAX_MB`.
+Exactly-at-cap and smaller results report `false` when no rows are omitted. This
+flag does not indicate rows excluded by a SQL LIMIT visible in `executed_sql`,
+and does not bound or count affected rows for a write with `RETURNING`. In
+`dry_run`, the SQL Server restriction remains unchanged because no results are
+fetched.
+
+Omitting `limit` leaves the SQL limit unchanged. Explicit request caps remain
+subject to the server's `SQL_MAX_ROW` configuration. This behavior also applies
+to `dry_run`, which returns the transformed SQL without executing it.
+
+The SQL-only `Database.apply_limit_to_sql()` helper used by dataset column
+discovery and alerts does not use the executor's bounded cursor. It retains the
+legacy SQL Server behavior of replacing `PERCENT`, `WITH TIES`, and non-fixed
+limit expressions with a fixed `TOP` cap, without introducing derived tables
+that could reject unnamed or duplicate columns. Smaller fixed limits, including
+zero, are preserved unless the caller explicitly requests `force=True`. This
+legacy helper's replacement semantics are separate from MCP execution's
+preservation of complex SQL restrictions.
+
 ## Result serialization
 
 Data-bearing tool responses sanitize row values, column samples, and statistics

superset-core/src/superset_core/queries/types.py

@@ -71,6 +71,8 @@ class QueryOptions:
     # Basic options
     catalog: str | None = None
     schema: str | None = None
+    # Cap the last statement's outer SQL limit without increasing a smaller one.
+    # None leaves SQL limits unchanged; explicit caps also respect SQL_MAX_ROW.
     limit: int | None = None
     timeout_seconds: int | None = None
 
@@ -100,6 +102,7 @@ class StatementResult:
     data: pd.DataFrame | None = None
     row_count: int = 0
     execution_time_ms: float | None = None
+    truncated: bool = False  # Rows omitted by the fetch cap, beyond executed_sql
 
 
 @dataclass

superset/db_engine_specs/bigquery.py

@@ -52,7 +52,7 @@
 from superset.errors import SupersetError, SupersetErrorType
 from superset.exceptions import SupersetException
 from superset.sql.parse import SQLScript, Table
-from superset.superset_typing import ResultSetColumnType
+from superset.superset_typing import FetchedRows, ResultSetColumnType
 from superset.utils import core as utils, json
 from superset.utils.hashing import hash_from_str
 
@@ -416,7 +416,8 @@ def fetch_data(cls, cursor: Any, limit: int | None = None) -> list[tuple[Any, ..
 
         Samples a first batch to estimate row size, then extrapolates the
         total number of rows that fit within ``BQ_FETCH_MAX_MB``.
-        Falls back to the parent implementation on any error.
+        Falls back to the parent implementation if the initial read fails.
+        Errors after consuming rows propagate instead of returning partial data.
         """
         # ``BQ_FETCH_MAX_MB`` has a default in ``config.py``, so use bracket
         # access in-context — a missing key should surface as a loud KeyError
@@ -428,13 +429,14 @@ def fetch_data(cls, cursor: Any, limit: int | None = None) -> list[tuple[Any, ..
         )
         max_bytes = max_mb * 1024 * 1024
 
+        first_batch: list[Any] = []
         try:
             initial_batch_size = (
                 min(_BQ_INITIAL_SAMPLE_ROWS, limit)
                 if limit
                 else _BQ_INITIAL_SAMPLE_ROWS
             )
-            first_batch: list[Any] = cursor.fetchmany(initial_batch_size)
+            first_batch = cursor.fetchmany(initial_batch_size)
 
             if not first_batch:
                 if has_request_context():
@@ -465,13 +467,17 @@ def fetch_data(cls, cursor: Any, limit: int | None = None) -> list[tuple[Any, ..
 
             # First batch already covers the budget or the result set
             if rows_fetched < initial_batch_size or remaining_rows <= 0:
+                # A full sample can also be the entire result. Confirm an omitted
+                # row; a bounded cursor leaves its own EOF probe to the executor.
                 memory_limited = (
-                    remaining_rows <= 0 and rows_fetched == initial_batch_size
+                    remaining_rows <= 0
+                    and rows_fetched == initial_batch_size
+                    and bool(cursor.fetchmany(1))
                 )
                 if has_request_context():
                     g.bq_memory_limited = memory_limited
                     g.bq_memory_limited_row_count = len(first_batch)
-                return first_batch
+                return FetchedRows(first_batch, truncated=memory_limited)
 
             # Fetch one extra row to confirm truncation without false positives
             second_batch: list[Any] = cursor.fetchmany(remaining_rows + 1) or []
@@ -488,16 +494,15 @@ def fetch_data(cls, cursor: Any, limit: int | None = None) -> list[tuple[Any, ..
             if has_request_context():
                 g.bq_memory_limited = memory_limited
                 g.bq_memory_limited_row_count = len(data)
-            return data
-
-        except Exception:  # pylint: disable=broad-except
-            # Broad catch on purpose: any failure in the size-estimation /
-            # progressive-fetch path (BigQuery DB-API errors, network or
-            # auth timeouts mid-fetch, ``sys.getsizeof`` raising on an
-            # unexpected cell type, or a future ``Row`` subclass we don't
-            # know how to unwrap) must degrade gracefully to the parent's
-            # straight fetch so the user still gets data.
-            # Fallback to parent implementation
+            return FetchedRows(data, truncated=memory_limited)
+
+        except Exception as ex:  # pylint: disable=broad-except
+            # A forward-only cursor cannot replay the consumed sample. Falling
+            # back after an estimation, second-batch, or EOF-probe failure would
+            # silently discard it and return only the remaining rows as success.
+            if first_batch:
+                raise cls.get_dbapi_mapped_exception(ex) from ex
+            # An initial read failure leaves no buffered rows to lose.
             data = super().fetch_data(cursor, limit)
             if data and type(data[0]).__name__ == "Row":
                 data = [r.values() for r in data]  # type: ignore

superset/mcp_service/sql_lab/schemas.py

@@ -72,9 +72,12 @@ class ExecuteSqlRequest(BaseModel):
     limit: int | None = Field(
         default=None,
         description=(
-            "Maximum number of rows to return. "
-            "If not specified, respects the LIMIT in your SQL query. "
-            "If specified, overrides any SQL LIMIT clause."
+            "Maximum rows returned by the last statement. "
+            "Omitted: respects SQL LIMIT. "
+            "If set, caps its outer LIMIT at min(SQL LIMIT, "
+            "this value), adding a LIMIT if absent. "
+            "Never raises stricter SQL limits or changes inner limits. "
+            "Explicit limits also obey server SQL_MAX_ROW."
         ),
         ge=1,
         le=10000,
@@ -151,6 +154,7 @@ class StatementInfo(BaseModel):
         ..., description="SQL after transformations (RLS, mutations, limits)"
     )
     row_count: RowCount = Field(..., description="Number of rows returned/affected")
+    truncated: bool = Field(False, description="Rows omitted by the fetch cap")
     execution_time_ms: float | None = Field(
         None, description="Statement execution time in milliseconds"
     )

superset/mcp_service/sql_lab/tool/execute_sql.py

@@ -373,6 +373,7 @@ def _convert_to_response(result: QueryResult) -> ExecuteSqlResponse:
                 original_sql=stmt.original_sql,
                 executed_sql=stmt.executed_sql,
                 row_count=stmt.row_count,
+                truncated=stmt.truncated,
                 execution_time_ms=stmt.execution_time_ms,
                 data=stmt_data,
             )

superset/models/core.py

@@ -1197,12 +1197,22 @@ def apply_limit_to_sql(
         limit: int = 1000,
         force: bool = False,
     ) -> str:
+        """Cap SQL for callers that do not enforce a separate cursor row budget.
+
+        Preserve smaller fixed limits unless forced. For SQL Server, retain the
+        legacy replacement of non-fixed limits: safely wrapping arbitrary output
+        columns is impossible, and these callers cannot rely on SQLExecutor's
+        bounded cursor to enforce the cap.
+        """
         script = SQLScript(sql, self.db_engine_spec.engine)
         statement = script.statements[-1]
-        current_limit = statement.get_limit_value() or float("inf")
-
-        if limit < current_limit or force:
+        if force or (
+            self.db_engine_spec.engine == "mssql"
+            and statement.get_limit_value() is None
+        ):
             statement.set_limit_value(limit, self.db_engine_spec.limit_method)
+        else:
+            statement.cap_limit_value(limit, self.db_engine_spec.limit_method)
 
         return script.format()
 

superset/result_set.py

@@ -185,7 +185,10 @@ def __init__(  # pylint: disable=too-many-locals  # noqa: C901
         data: DbapiResult,
         cursor_description: DbapiDescription,
         db_engine_spec: type[BaseEngineSpec],
+        *,
+        truncated: bool = False,
     ):
+        self.truncated = truncated
         self.db_engine_spec = db_engine_spec
         data = data or []
         column_names: list[str] = []

superset/sql/execution/celery_task.py

@@ -181,6 +181,7 @@ def _finalize_successful_query(
                     "data": data,
                     "columns": columns,
                     "row_count": result_set.size,
+                    "truncated": result_set.truncated,
                     "execution_time_ms": exec_time,
                 }
             )