[BUGFIX] query asset SQL ending in a comment fails validation

great_expectations/execution_engine/sqlalchemy_execution_engine.py

@@ -233,6 +233,39 @@ def _dialect_requires_persisted_connection(
     return return_val
 
 
+def _ensure_sql_text_ends_with_newline(text: str) -> str:
+    """Ensure raw SQL text ends on a fresh line before it is wrapped as a subquery.
+
+    SQLAlchemy compiles wrapped raw SQL text verbatim inside "(<text>) AS alias" on a
+    single line. If the text ends in a line comment (e.g. "-- note"), the appended ")"
+    and alias land inside that comment and the resulting statement is never terminated.
+    Ending the text on a fresh line keeps any appended syntax out of reach of a comment
+    that runs to the end of the text, without changing what the query selects.
+
+    Args:
+        text: The raw SQL text about to be wrapped as a subquery.
+
+    Returns:
+        The same text, with a trailing newline added if it didn't already have one.
+    """
+    return text if text.endswith("\n") else f"{text}\n"
+
+
+def _wrap_raw_sql_as_subquery(selectable: sqlalchemy.TextClause) -> Subquery:
+    """Wrap a raw-SQL TextClause (e.g. from a query asset) as a subquery.
+
+    See _ensure_sql_text_ends_with_newline for why the text is normalized first.
+
+    Args:
+        selectable: The raw-SQL TextClause to wrap.
+
+    Returns:
+        A Subquery built from the TextClause's columns.
+    """
+    text = _ensure_sql_text_ends_with_newline(selectable.text)
+    return sa.text(text).columns().subquery()
+
+
 class SqlAlchemyExecutionEngine(ExecutionEngine[SQLAColumnClause]):
     """SparkDFExecutionEngine instantiates the ExecutionEngine API to support computations using Spark platform.
 
@@ -652,7 +685,7 @@ def get_domain_records(  # noqa: C901, PLR0912, PLR0915 # FIXME CoP
         to TextualSelect using sa.columns() before it can be converted to type Subquery
         """
         if sqlalchemy.TextClause and isinstance(selectable, sqlalchemy.TextClause):  # type: ignore[truthy-function] # FIXME CoP
-            selectable = selectable.columns().subquery()
+            selectable = _wrap_raw_sql_as_subquery(selectable)
 
         # Filtering by row condition.
         if "row_condition" in domain_kwargs and domain_kwargs["row_condition"] is not None:
@@ -1010,7 +1043,7 @@ def resolve_metric_bundle(
                 """
                 if sqlalchemy.TextClause and isinstance(selectable, sqlalchemy.TextClause):  # type: ignore[truthy-function] # FIXME CoP
                     sa_query_object = sa.select(*query["select"]).select_from(
-                        selectable.columns().subquery()
+                        _wrap_raw_sql_as_subquery(selectable)
                     )
                 elif (sqlalchemy.Select and isinstance(selectable, sqlalchemy.Select)) or (  # type: ignore[truthy-function] # FIXME CoP
                     sqlalchemy.TextualSelect and isinstance(selectable, sqlalchemy.TextualSelect)  # type: ignore[truthy-function] # FIXME CoP
@@ -1220,7 +1253,7 @@ def _count_query_parameters(self, selectable: sqlalchemy.Selectable, select_list
         """
         DEFAULT_PARAMS_PER_SELECT = 2  # Conservative upper bound
         if isinstance(selectable, sqlalchemy.TextClause):
-            test_query = sa.select(*select_list).select_from(selectable.columns().subquery())
+            test_query = sa.select(*select_list).select_from(_wrap_raw_sql_as_subquery(selectable))
         elif isinstance(selectable, (sqlalchemy.Select, sqlalchemy.TextualSelect)):
             test_query = sa.select(*select_list).select_from(selectable.subquery())
         elif isinstance(selectable, sa.sql.FromClause):
@@ -1370,8 +1403,9 @@ def _subselectable(self, batch_spec: BatchSpec) -> sqlalchemy.Selectable:
             if not isinstance(query, str):
                 raise ValueError(f"SQL query should be a str but got {query}")  # noqa: TRY003 # FIXME CoP
             # Query is a valid SELECT query that begins with r"\w+select\w"
+            stripped_query = query.lstrip()[6:].strip().rstrip(";").rstrip()
             selectable = sa.select(
-                sa.text(query.lstrip()[6:].strip().rstrip(";").rstrip())
+                sa.text(_ensure_sql_text_ends_with_newline(stripped_query))
             ).subquery()
 
         return selectable

tests/integration/data_sources_and_expectations/data_sources/test_query_asset_sql_comments.py

@@ -0,0 +1,90 @@
+"""Regression tests for query assets whose raw SQL ends in a comment.
+
+See https://github.com/fivetran/great_expectations/issues/12122. GX wraps a query
+asset's raw SQL in a subquery before running metrics against it. SQLAlchemy compiles
+that as the user's text pasted verbatim inside "(<text>) AS anon_1" on a single line,
+so if the user's SQL ends in a line comment, the appended ")" and alias land inside
+that comment and the statement is never terminated.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pandas as pd
+
+import great_expectations.expectations as gxe
+from great_expectations.datasource.fluent.sql_datasource import SQLDatasource, TableAsset
+from tests.integration.conftest import parameterize_batch_for_data_sources
+from tests.integration.test_utils.data_source_config import (
+    PostgreSQLDatasourceTestConfig,
+    SqliteDatasourceTestConfig,
+)
+
+if TYPE_CHECKING:
+    from great_expectations.datasource.fluent.interfaces import Batch
+
+COL_A = "col_a"
+
+DATA = pd.DataFrame({COL_A: ["x", "y"]})
+
+
+def _add_query_asset(batch_for_datasource: Batch, name: str, query: str) -> Batch:
+    asset = batch_for_datasource.data_asset
+    assert isinstance(asset, TableAsset)
+    datasource = batch_for_datasource.datasource
+    assert isinstance(datasource, SQLDatasource)
+
+    query_asset = datasource.add_query_asset(name=name, query=query.format(table=asset.table_name))
+    return query_asset.add_batch_definition_whole_table(f"{name}_bd").get_batch()
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=[SqliteDatasourceTestConfig(), PostgreSQLDatasourceTestConfig()],
+    data=DATA,
+)
+def test_query_asset_sql_ending_in_line_comment(batch_for_datasource: Batch) -> None:
+    """A trailing `--` comment does not change what a query selects."""
+    batch = _add_query_asset(
+        batch_for_datasource,
+        "trailing_comment_asset",
+        f"SELECT {COL_A} FROM {{table}} -- only the rows we care about",
+    )
+
+    result = batch.validate(gxe.ExpectColumnValuesToNotBeNull(column=COL_A))
+
+    assert result.success, result.exception_info
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=[SqliteDatasourceTestConfig(), PostgreSQLDatasourceTestConfig()],
+    data=DATA,
+)
+def test_query_asset_sql_ending_in_terminated_block_comment(batch_for_datasource: Batch) -> None:
+    """A trailing, properly-closed `/* ... */` comment does not change what a query selects."""
+    batch = _add_query_asset(
+        batch_for_datasource,
+        "terminated_block_comment_asset",
+        f"SELECT {COL_A} FROM {{table}} /* only the rows we care about */",
+    )
+
+    result = batch.validate(gxe.ExpectColumnValuesToNotBeNull(column=COL_A))
+
+    assert result.success, result.exception_info
+
+
+@parameterize_batch_for_data_sources(
+    data_source_configs=[SqliteDatasourceTestConfig(), PostgreSQLDatasourceTestConfig()],
+    data=DATA,
+)
+def test_query_asset_sql_with_comment_not_at_end(batch_for_datasource: Batch) -> None:
+    """A comment in the middle of a query, followed by a later clause, is unaffected."""
+    batch = _add_query_asset(
+        batch_for_datasource,
+        "mid_query_comment_asset",
+        f"SELECT {COL_A} -- the column we care about\nFROM {{table}}",
+    )
+
+    result = batch.validate(gxe.ExpectColumnValuesToNotBeNull(column=COL_A))
+
+    assert result.success, result.exception_info