Repository: great-expectations/great_expectations (python)

Issue:
UnexpectedRowsExpectation unexpected_rows limit

**Is your feature request related to a problem? Please describe.**
Currently, when using UnexpectedRowsExpectation or retrieving unexpected rows in Great Expectations (e.g., result_format: COMPLETE with Spark/SQL), the system returns a maximum of 200 unexpected rows. This limit is hardcoded and prevents users from accessing all failing rows in large datasets. For example, if there are 432 failing rows, only 200 are returned. This makes it difficult to fully analyze validation results without writing extra custom queries.

**Describe the solution you'd like**
I would like to have a configurable option to increase or remove this limit, such as:
result_format:
  result_format: COMPLETE
  include_unexpected_rows: True
  max_unexpected_rows: -1  # -1 or None for unlimited

This would allow retrieving all unexpected rows directly from Great Expectations validation results, without memory issues or extra workarounds.

**Describe alternatives you've considered**
Using Spark filters manually after validation to retrieve all failing rows.
Running multiple partial validations with offsets.

Both alternatives require extra code and duplicate work outside of Great Expectations, which is not ideal.

**Additional context**
[Add any other context, screenshots, or code samples about the feature request here.](https://discourse.greatexpectations.io/t/unexpectedrowsexpectation-unexpected-rows-limit/2026)



Patch:
diff --git a/great_expectations/__init__.py b/great_expectations/__init__.py
index c912f998364e..42f1c7e6c629 100644
--- a/great_expectations/__init__.py
+++ b/great_expectations/__init__.py
@@ -3,7 +3,7 @@
 # Set up version information immediately
 from ._version import get_versions as _get_versions
 
-__version__ = _get_versions()["version"]
+__version__: str = _get_versions()["version"]
 del _get_versions
 
 # Submodules must be imported first to avoid circular dependencies
diff --git a/great_expectations/core/expectation_validation_result.py b/great_expectations/core/expectation_validation_result.py
index 6f5a1784b6c3..3da2a433542c 100644
--- a/great_expectations/core/expectation_validation_result.py
+++ b/great_expectations/core/expectation_validation_result.py
@@ -32,6 +32,7 @@
 )
 
 if TYPE_CHECKING:
+    from great_expectations.expectations.expectation import Expectation
     from great_expectations.expectations.expectation_configuration import (
         ExpectationConfiguration,
     )
@@ -114,6 +115,14 @@ def __init__(  # noqa: PLR0913 # FIXME CoP
         }
         self.rendered_content = rendered_content
 
+    @property
+    @public_api
+    def expectation(self) -> Expectation:
+        """The Expectation object that produced this result."""
+        if self.expectation_config is None:
+            raise TypeError("Cannot construct Expectation: expectation_config is None.")  # noqa: TRY003
+        return self.expectation_config.to_domain_obj()
+
     @override
     def __eq__(self, other):
         """ExpectationValidationResult equality ignores instance identity, relying only on properties."""  # noqa: E501 # FIXME CoP
@@ -529,6 +538,12 @@ def asset_name(self) -> str | None:
             return self.meta["active_batch_definition"].get("data_asset_name")
         return None
 
+    @property
+    @public_api
+    def batch_parameters(self) -> dict | None:
+        """The batch parameters used for this validation run, if any."""
+        return self.meta.get("batch_parameters")
+
     def __eq__(self, other):  # type: ignore[explicit-override] # FIXME
         """ExpectationSuiteValidationResult equality ignores instance identity, relying only on properties."""  # noqa: E501 # FIXME CoP
         if not isinstance(other, self.__class__):
diff --git a/great_expectations/core/validation_definition.py b/great_expectations/core/validation_definition.py
index 656423f07e1e..7589541e1d0b 100644
--- a/great_expectations/core/validation_definition.py
+++ b/great_expectations/core/validation_definition.py
@@ -22,6 +22,7 @@
 from great_expectations.core.result_format import DEFAULT_RESULT_FORMAT
 from great_expectations.core.run_identifier import RunIdentifier
 from great_expectations.core.serdes import _EncodedValidationData, _IdentifierBundle
+from great_expectations.core.suite_parameters import parse_suite_parameter
 from great_expectations.data_context.cloud_constants import GXCloudRESTResource
 from great_expectations.data_context.data_context.context_factory import project_manager
 from great_expectations.data_context.types.refs import GXCloudResourceRef
@@ -39,6 +40,11 @@
     StoreBackendError,
     ValidationDefinitionNotFoundError,
 )
+from great_expectations.expectations.core.unexpected_rows_expectation import (
+    UnexpectedRowsExpectation,
+)
+from great_expectations.metrics.metric_results import MetricErrorResult
+from great_expectations.metrics.query.batch_table import QueryBatchTable
 from great_expectations.validator.v1_validator import Validator
 
 if TYPE_CHECKING:
@@ -52,6 +58,7 @@
     )
     from great_expectations.datasource.fluent.batch_request import BatchParameters
     from great_expectations.datasource.fluent.interfaces import DataAsset, Datasource
+    from great_expectations.expectations.expectation import Expectation
 
 
 @public_api
@@ -378,6 +385,69 @@ def save(self) -> None:
 
         store.update(key=key, value=self)
 
+    @public_api
+    def get_unexpected_rows(
+        self,
+        expectation: Expectation,
+        batch_parameters: Optional[BatchParameters] = None,
+        expectation_parameters: Optional[SuiteParameterDict] = None,
+    ) -> list[dict]:
+        """Fetch all failing rows for an UnexpectedRowsExpectation without the 200-row limit.
+
+        Args:
+            expectation: The UnexpectedRowsExpectation to fetch rows for. Only
+                UnexpectedRowsExpectation is currently supported; other types raise ValueError.
+            batch_parameters: Optional batch parameters for selecting the correct batch.
+                Pass result.batch_parameters when using partitioned data.
+            expectation_parameters: Optional suite parameter values for resolving
+                parameterized queries (the $PARAMETER syntax). Only needed if the
+                expectation uses a suite parameter reference for unexpected_rows_query.
+
+        Returns:
+            A list of dicts, one per failing row.
+
+        Raises:
+            ValueError: If the expectation is not an UnexpectedRowsExpectation, if
+                unexpected_rows_query is a suite parameter reference but no
+                expectation_parameters are provided, or if the suite parameter does
+                not resolve to a string.
+            TypeError: If unexpected_rows_query is neither a string nor a supported
+                suite parameter reference.
+            RuntimeError: If the underlying metric computation fails.
+            SuiteParameterError: If a referenced suite parameter is missing or invalid.
+        """
+        if not isinstance(expectation, UnexpectedRowsExpectation):
+            raise ValueError(  # noqa: TRY003, TRY004
+                "Only UnexpectedRowsExpectation is currently supported. "
+                f"Got {type(expectation).__name__}."
+            )
+        query = expectation.unexpected_rows_query
+        if isinstance(query, dict) and "$PARAMETER" in query:
+            if not expectation_parameters:
+                raise ValueError(  # noqa: TRY003
+                    "unexpected_rows_query is a suite parameter reference "
+                    f"({query['$PARAMETER']!r}) but no expectation_parameters were provided."
+                )
+            resolved = parse_suite_parameter(
+                query["$PARAMETER"],
+                suite_parameters=expectation_parameters,
+            )
+            if not isinstance(resolved, str):
+                raise ValueError(  # noqa: TRY003
+                    f"Suite parameter {query['$PARAMETER']!r} did not resolve to a string."
+                )
+            query = resolved
+        elif not isinstance(query, str):
+            raise TypeError(  # noqa: TRY003
+                "unexpected_rows_query must be a string or suite parameter reference, "
+                f"got {type(query).__name__}."
+            )
+        batch = self.batch_definition.get_batch(batch_parameters)
+        metric_result = batch.compute_metrics(QueryBatchTable(query=query, fetch_all=True))
+        if isinstance(metric_result, MetricErrorResult):
+            raise RuntimeError(metric_result.value.exception_message)  # noqa: TRY004
+        return metric_result.value
+
     def _add_to_store(self) -> None:
         """This is used to persist a validation_definition before we run it.
 
diff --git a/great_expectations/expectations/metrics/query_metric_provider.py b/great_expectations/expectations/metrics/query_metric_provider.py
index 18b15560e009..abcd968c8465 100644
--- a/great_expectations/expectations/metrics/query_metric_provider.py
+++ b/great_expectations/expectations/metrics/query_metric_provider.py
@@ -205,10 +205,12 @@ def _get_sqlalchemy_records_from_substituted_batch_subquery(
         cls,
         substituted_batch_subquery: str,
         execution_engine: SqlAlchemyExecutionEngine,
+        fetch_all: bool = False,
     ) -> list[dict]:
-        result: Union[Sequence[sa.Row[Any]], Any] = execution_engine.execute_query(
-            sa.text(substituted_batch_subquery)
-        ).fetchmany(MAX_RESULT_RECORDS)
+        query_result = execution_engine.execute_query(sa.text(substituted_batch_subquery))
+        result: Union[Sequence[sa.Row[Any]], Any] = (
+            query_result.fetchall() if fetch_all else query_result.fetchmany(MAX_RESULT_RECORDS)
+        )
 
         if isinstance(result, Sequence):
             return [element._asdict() for element in result]
diff --git a/great_expectations/expectations/metrics/query_metrics/query_table/query_table.py b/great_expectations/expectations/metrics/query_metrics/query_table/query_table.py
index 3d39247c336f..6ef7e1623f59 100644
--- a/great_expectations/expectations/metrics/query_metrics/query_table/query_table.py
+++ b/great_expectations/expectations/metrics/query_metrics/query_table/query_table.py
@@ -19,7 +19,7 @@
 
 class QueryTable(QueryMetricProvider):
     metric_name = "query.table"
-    value_keys = ("query",)
+    value_keys: tuple[str, ...] = ("query", "fetch_all")
 
     @metric_value(engine=SqlAlchemyExecutionEngine)
     def _sqlalchemy(
@@ -41,9 +41,11 @@ def _sqlalchemy(
                 execution_engine=execution_engine,
             )
         )
+        fetch_all = metric_value_kwargs.get("fetch_all", False)
         return cls._get_sqlalchemy_records_from_substituted_batch_subquery(
             substituted_batch_subquery=substituted_batch_subquery,
             execution_engine=execution_engine,
+            fetch_all=fetch_all,
         )
 
     @metric_value(engine=SparkDFExecutionEngine)
@@ -66,6 +68,10 @@ def _spark(
         query = query.format(batch="tmp_view")
 
         engine: pyspark.SparkSession = execution_engine.spark
-        result: List[pyspark.Row] = engine.sql(query).limit(MAX_RESULT_RECORDS).collect()
+        fetch_all = metric_value_kwargs.get("fetch_all", False)
+        if fetch_all:
+            result: List[pyspark.Row] = engine.sql(query).collect()
+        else:
+            result = engine.sql(query).limit(MAX_RESULT_RECORDS).collect()
 
         return [element.asDict() for element in result]
diff --git a/great_expectations/metrics/__init__.py b/great_expectations/metrics/__init__.py
index 86fad7f7b3c8..8266ed47c8dd 100644
--- a/great_expectations/metrics/__init__.py
+++ b/great_expectations/metrics/__init__.py
@@ -22,5 +22,6 @@
 from .column_pair.values_in_set import ColumnPairValuesInSetUnexpectedCount
 from .metric import Metric
 from .multi_column.sum_equal import MultiColumnSumEqualUnexpectedCount
+from .query.batch_table import QueryBatchTable
 from .query.data_source_table import QueryDataSourceTable
 from .query.row_count import QueryRowCount
diff --git a/great_expectations/metrics/query/batch_table.py b/great_expectations/metrics/query/batch_table.py
new file mode 100644
index 000000000000..c427f4cebe4c
--- /dev/null
+++ b/great_expectations/metrics/query/batch_table.py
@@ -0,0 +1,13 @@
+from great_expectations.metrics.metric import NonEmptyString
+from great_expectations.metrics.metric_results import MetricResult
+from great_expectations.metrics.query import QueryMetric
+
+
+class QueryBatchTableResult(MetricResult[list[dict]]): ...
+
+
+class QueryBatchTable(QueryMetric[QueryBatchTableResult]):
+    name = "query.table"
+
+    query: NonEmptyString
+    fetch_all: bool = False


Test patch:
diff --git a/tests/core/test_expectation_validation_result.py b/tests/core/test_expectation_validation_result.py
index a221e7cbfc08..61c7457f9cd1 100644
--- a/tests/core/test_expectation_validation_result.py
+++ b/tests/core/test_expectation_validation_result.py
@@ -708,6 +708,59 @@ def test_hash_stable_across_runs(self):
         assert hash1 == hash2 == hash3
 
 
+class TestExpectationProperty:
+    @pytest.mark.unit
+    def test_returns_typed_expectation_from_config(self):
+        config = ExpectationConfiguration(
+            type="expect_column_values_to_not_be_null",
+            kwargs={"column": "foo"},
+        )
+        evr = ExpectationValidationResult(
+            success=True,
+            expectation_config=config,
+        )
+        expectation = evr.expectation
+        assert isinstance(expectation, gxe.ExpectColumnValuesToNotBeNull)
+        assert expectation.column == "foo"
+
+    @pytest.mark.unit
+    def test_works_for_unexpected_rows_expectation(self):
+        config = ExpectationConfiguration(
+            type="unexpected_rows_expectation",
+            kwargs={"unexpected_rows_query": "SELECT * FROM {batch} WHERE col > 5"},
+        )
+        evr = ExpectationValidationResult(
+            success=False,
+            expectation_config=config,
+        )
+        expectation = evr.expectation
+        assert isinstance(expectation, gxe.UnexpectedRowsExpectation)
+        assert expectation.unexpected_rows_query == "SELECT * FROM {batch} WHERE col > 5"
+
+
+class TestBatchParametersProperty:
+    @pytest.mark.unit
+    def test_returns_batch_parameters_from_meta(self):
+        params = {"year": 2026, "month": 3}
+        esvr = ExpectationSuiteValidationResult(
+            success=True,
+            results=[],
+            suite_name="test_suite",
+            meta={"batch_parameters": params},
+        )
+        assert esvr.batch_parameters == params
+
+    @pytest.mark.unit
+    def test_returns_none_when_missing(self):
+        esvr = ExpectationSuiteValidationResult(
+            success=True,
+            results=[],
+            suite_name="test_suite",
+            meta={},
+        )
+        assert esvr.batch_parameters is None
+
+
 class TestGetMaxSeverityFailure:
     @pytest.mark.unit
     def test_get_max_severity_failure_no_results(self):
diff --git a/tests/core/test_validation_definition.py b/tests/core/test_validation_definition.py
index 2d8db48e7f22..633e79a826e2 100644
--- a/tests/core/test_validation_definition.py
+++ b/tests/core/test_validation_definition.py
@@ -14,6 +14,7 @@
 import great_expectations.expectations as gxe
 from great_expectations import RunIdentifier
 from great_expectations import __version__ as GX_VERSION
+from great_expectations.core.batch_definition import BatchDefinition
 from great_expectations.core.expectation_suite import ExpectationSuite
 from great_expectations.core.expectation_validation_result import (
     ExpectationSuiteValidationResult,
@@ -60,7 +61,6 @@
 
     from pytest_mock import MockerFixture
 
-    from great_expectations.core.batch_definition import BatchDefinition
     from great_expectations.data_context.data_context.cloud_data_context import (
         CloudDataContext,
     )
@@ -980,6 +980,99 @@ def test_is_fresh_raises_error_when_validation_definition_not_found(in_memory_ru
     assert isinstance(diagnostics.errors[0], ValidationDefinitionNotFoundError)
 
 
+class TestGetUnexpectedRows:
+    @pytest.mark.unit
+    def test_raises_value_error_for_non_unexpected_rows_expectation(
+        self, validation_definition: ValidationDefinition
+    ):
+        expectation = gxe.ExpectColumnValuesToNotBeNull(column="foo")
+        with pytest.raises(ValueError, match="Only UnexpectedRowsExpectation"):
+            validation_definition.get_unexpected_rows(expectation)
+
+    @pytest.mark.unit
+    def test_calls_compute_metrics_with_correct_args(
+        self, validation_definition: ValidationDefinition, mocker: MockerFixture
+    ):
+        expectation = gxe.UnexpectedRowsExpectation(
+            unexpected_rows_query="SELECT * FROM {batch} WHERE col > 5"
+        )
+        mock_batch = mocker.MagicMock()
+        mock_metric_result = mocker.MagicMock()
+        mock_metric_result.value = [{"col": 10}]
+        mock_batch.compute_metrics.return_value = mock_metric_result
+
+        mocker.patch.object(
+            BatchDefinition,
+            "get_batch",
+            return_value=mock_batch,
+        )
+
+        result = validation_definition.get_unexpected_rows(expectation)
+        assert result == [{"col": 10}]
+        mock_batch.compute_metrics.assert_called_once()
+        call_arg = mock_batch.compute_metrics.call_args[0][0]
+        assert call_arg.name == "query.table"
+        assert call_arg.query == "SELECT * FROM {batch} WHERE col > 5"
+        assert call_arg.fetch_all is True
+
+    @pytest.mark.unit
+    def test_passes_batch_parameters(
+        self, validation_definition: ValidationDefinition, mocker: MockerFixture
+    ):
+        expectation = gxe.UnexpectedRowsExpectation(
+            unexpected_rows_query="SELECT * FROM {batch} WHERE col > 5"
+        )
+        mock_batch = mocker.MagicMock()
+        mock_metric_result = mocker.MagicMock()
+        mock_metric_result.value = []
+        mock_batch.compute_metrics.return_value = mock_metric_result
+
+        mock_get_batch = mocker.patch.object(
+            BatchDefinition,
+            "get_batch",
+            return_value=mock_batch,
+        )
+
+        batch_params = {"year": 2026, "month": 3}
+        validation_definition.get_unexpected_rows(expectation, batch_parameters=batch_params)
+        mock_get_batch.assert_called_once_with(batch_params)
+
+    @pytest.mark.unit
+    def test_resolves_suite_parameter_query(
+        self, validation_definition: ValidationDefinition, mocker: MockerFixture
+    ):
+        expectation = gxe.UnexpectedRowsExpectation(
+            unexpected_rows_query={"$PARAMETER": "my_query"}
+        )
+        mock_batch = mocker.MagicMock()
+        mock_metric_result = mocker.MagicMock()
+        mock_metric_result.value = [{"col": 99}]
+        mock_batch.compute_metrics.return_value = mock_metric_result
+
+        mocker.patch.object(
+            BatchDefinition,
+            "get_batch",
+            return_value=mock_batch,
+        )
+
+        result = validation_definition.get_unexpected_rows(
+            expectation,
+            expectation_parameters={"my_query": "SELECT * FROM {batch} WHERE col > 5"},
+        )
+        assert result == [{"col": 99}]
+        call_arg = mock_batch.compute_metrics.call_args[0][0]
+        assert call_arg.query == "SELECT * FROM {batch} WHERE col > 5"
+        assert call_arg.fetch_all is True
+
+    @pytest.mark.unit
+    def test_raises_when_suite_parameter_missing(self, validation_definition: ValidationDefinition):
+        expectation = gxe.UnexpectedRowsExpectation(
+            unexpected_rows_query={"$PARAMETER": "my_query"}
+        )
+        with pytest.raises(ValueError, match="no expectation_parameters were provided"):
+            validation_definition.get_unexpected_rows(expectation)
+
+
 @pytest.mark.unit
 def test_is_fresh_raises_error_when_child_deps_not_found(in_memory_runtime_context):
     context = in_memory_runtime_context
diff --git a/tests/expectations/metrics/query_metrics/test_query_metrics.py b/tests/expectations/metrics/query_metrics/test_query_metrics.py
index 825c5f623b4c..5b91e8e3d5de 100644
--- a/tests/expectations/metrics/query_metrics/test_query_metrics.py
+++ b/tests/expectations/metrics/query_metrics/test_query_metrics.py
@@ -176,9 +176,14 @@ def test_sqlalchemy_query_metrics_that_return_records(
             batch_selectable=batch_selectable,
             execution_engine=mock_sqlalchemy_execution_engine,
         )
+    expected_kwargs = {
+        "substituted_batch_subquery": mock_substituted_batch_subquery,
+        "execution_engine": mock_sqlalchemy_execution_engine,
+    }
+    if issubclass(metric_class, QueryTable):
+        expected_kwargs["fetch_all"] = False
     mock_get_sqlalchemy_records_from_substituted_batch_subquery.assert_called_once_with(
-        substituted_batch_subquery=mock_substituted_batch_subquery,
-        execution_engine=mock_sqlalchemy_execution_engine,
+        **expected_kwargs,
     )
 
 
diff --git a/tests/integration/data_sources_and_expectations/expectations/test_unexpected_rows_expectation.py b/tests/integration/data_sources_and_expectations/expectations/test_unexpected_rows_expectation.py
index 1df0af29d5a8..ed71de40b221 100644
--- a/tests/integration/data_sources_and_expectations/expectations/test_unexpected_rows_expectation.py
+++ b/tests/integration/data_sources_and_expectations/expectations/test_unexpected_rows_expectation.py
@@ -6,7 +6,10 @@
 import pytest
 
 import great_expectations.expectations as gxe
+from great_expectations.constants import MAX_RESULT_RECORDS
+from great_expectations.core.expectation_suite import ExpectationSuite
 from great_expectations.core.result_format import ResultFormat
+from great_expectations.core.validation_definition import ValidationDefinition
 from great_expectations.datasource.fluent.interfaces import Batch
 from tests.integration.conftest import parameterize_batch_for_data_sources
 from tests.integration.test_utils.data_source_config import (
@@ -625,3 +628,44 @@ def test_sql_server_order_by_without_top_works_transparently_failure(
     result = batch_for_datasource.validate(expectation)
     assert result.success is False
     assert result.exception_info.get("raised_exception") is False
+
+
+GET_UNEXPECTED_ROWS_DATA = pd.DataFrame(
+    {"id": list(range(300)), "value": [i % 2 for i in range(300)]}
+)
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=[PostgreSQLDatasourceTestConfig()],
+    data=GET_UNEXPECTED_ROWS_DATA,
+)
+def test_get_unexpected_rows_returns_all_rows(
+    asset_for_datasource,
+    _batch_setup_for_datasource,
+) -> None:
+    """get_unexpected_rows() returns >200 rows, bypassing MAX_RESULT_RECORDS."""
+    context = _batch_setup_for_datasource.context
+    batch_definition = asset_for_datasource.add_batch_definition_whole_table(
+        name="get_unexpected_rows_bd"
+    )
+    suite = context.suites.add(ExpectationSuite(name="get_unexpected_rows_suite"))
+    expectation = gxe.UnexpectedRowsExpectation(
+        unexpected_rows_query="SELECT * FROM {batch} WHERE id > 0"
+    )
+    suite.add_expectation(expectation)
+
+    vd = context.validation_definitions.add(
+        ValidationDefinition(
+            name="get_unexpected_rows_vd",
+            data=batch_definition,
+            suite=suite,
+        )
+    )
+
+    result = vd.run()
+    assert not result.success
+
+    evr = result.results[0]
+    rows = vd.get_unexpected_rows(evr.expectation)
+    assert len(rows) == 299
+    assert len(rows) > MAX_RESULT_RECORDS
diff --git a/tests/integration/metrics/query/test_batch_table.py b/tests/integration/metrics/query/test_batch_table.py
new file mode 100644
index 000000000000..700a5ff984ee
--- /dev/null
+++ b/tests/integration/metrics/query/test_batch_table.py
@@ -0,0 +1,98 @@
+from typing import Sequence
+
+import pandas as pd
+
+from great_expectations.constants import MAX_RESULT_RECORDS
+from great_expectations.metrics.query.batch_table import (
+    QueryBatchTable,
+    QueryBatchTableResult,
+)
+from tests.integration.conftest import (
+    parameterize_batch_for_data_sources,
+)
+from tests.integration.test_utils.data_source_config import (
+    DatabricksDatasourceTestConfig,
+    DataSourceTestConfig,
+    PostgreSQLDatasourceTestConfig,
+    SnowflakeDatasourceTestConfig,
+    SparkFilesystemCsvDatasourceTestConfig,
+    SqliteDatasourceTestConfig,
+)
+
+SQL_DATA_SOURCES: Sequence[DataSourceTestConfig] = [
+    PostgreSQLDatasourceTestConfig(),
+    SqliteDatasourceTestConfig(),
+    SnowflakeDatasourceTestConfig(),
+    DatabricksDatasourceTestConfig(),
+]
+
+SPARK_DATA_SOURCES: Sequence[DataSourceTestConfig] = [
+    SparkFilesystemCsvDatasourceTestConfig(),
+]
+
+SMALL_DF = pd.DataFrame({"id": [1, 2, 3], "name": ["A", "B", "C"]})
+BIG_DF = pd.DataFrame({"id": list(range(300)), "name": ["X"] * 300})
+
+
+class TestQueryBatchTable:
+    @parameterize_batch_for_data_sources(
+        data_source_configs=SQL_DATA_SOURCES,
+        data=SMALL_DF,
+    )
+    def test_success_sql(self, batch_for_datasource):
+        metric = QueryBatchTable(
+            query="SELECT * FROM {batch} WHERE name = 'A'",
+            fetch_all=False,
+        )
+        result = batch_for_datasource.compute_metrics(metric)
+        assert isinstance(result, QueryBatchTableResult)
+        assert len(result.value) == 1
+
+    @parameterize_batch_for_data_sources(
+        data_source_configs=SPARK_DATA_SOURCES,
+        data=SMALL_DF,
+    )
+    def test_success_spark(self, batch_for_datasource):
+        metric = QueryBatchTable(
+            query="SELECT * FROM {batch} WHERE name = 'A'",
+            fetch_all=False,
+        )
+        result = batch_for_datasource.compute_metrics(metric)
+        assert isinstance(result, QueryBatchTableResult)
+        assert len(result.value) == 1
+
+    @parameterize_batch_for_data_sources(
+        data_source_configs=SQL_DATA_SOURCES,
+        data=BIG_DF,
+    )
+    def test_fetch_all_false_caps_at_200(self, batch_for_datasource):
+        metric = QueryBatchTable(
+            query="SELECT * FROM {batch} WHERE id > 0",
+            fetch_all=False,
+        )
+        result = batch_for_datasource.compute_metrics(metric)
+        assert len(result.value) == MAX_RESULT_RECORDS
+
+    @parameterize_batch_for_data_sources(
+        data_source_configs=SQL_DATA_SOURCES,
+        data=BIG_DF,
+    )
+    def test_fetch_all_true_returns_all_rows_sql(self, batch_for_datasource):
+        metric = QueryBatchTable(
+            query="SELECT * FROM {batch} WHERE id > 0",
+            fetch_all=True,
+        )
+        result = batch_for_datasource.compute_metrics(metric)
+        assert len(result.value) == 299
+
+    @parameterize_batch_for_data_sources(
+        data_source_configs=SPARK_DATA_SOURCES,
+        data=BIG_DF,
+    )
+    def test_fetch_all_true_returns_all_rows_spark(self, batch_for_datasource):
+        metric = QueryBatchTable(
+            query="SELECT * FROM {batch} WHERE id > 0",
+            fetch_all=True,
+        )
+        result = batch_for_datasource.compute_metrics(metric)
+        assert len(result.value) == 299