fix error

Author: fredwang1012

great_expectations/expectations/conditions.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Iterable, List, Literal, Union
 
@@ -9,6 +10,8 @@
 if TYPE_CHECKING:
     from typing_extensions import TypeAlias
 
+logger = logging.getLogger(__name__)
+
 
 class ConditionParserError(ValueError):
     """Raised when unable to determine the Condition type from a dict."""
@@ -81,6 +84,31 @@ class Column(BaseModel):
     def __init__(self, name: str):
         super().__init__(name=name)
 
+    @validator("name", pre=True)
+    @classmethod
+    def _parse_column_name(cls, v: str) -> str:
+        """Parse column names that contain the col("...") syntax.
+
+        This handles cases where conditions are deserialized from GX Cloud with
+        unparsed column names like 'col("DeltaReturn10Year")' instead of the
+        clean name 'DeltaReturn10Year'.
+
+        Args:
+            v: The column name string, potentially containing col() syntax
+
+        Returns:
+            The clean column name without col() wrapper
+        """
+        if isinstance(v, str) and v.startswith('col("') and v.endswith('")'):
+            # Extract just the column name from col("name") syntax
+            clean_name = v[5:-2]  # Remove 'col("' from start and '")' from end
+            logger.debug(
+                f"Parsed unparsed column name from '{v}' to '{clean_name}'. "
+                "Column names should be provided without col() wrapper."
+            )
+            return clean_name
+        return v
+
     @override
     def __hash__(self) -> int:
         return hash(self.name)

great_expectations/expectations/metrics/util.py

@@ -1695,10 +1695,15 @@ def get_sqlalchemy_column_metadata(  # noqa: C901 # FIXME CoP
                     schema=schema_name,
                 )
             else:
-                logger.warning("unexpected table_selectable type")
-                columns = inspector.get_columns(  # type: ignore[assignment]
-                    table_name=str(table_selectable),
-                    schema=schema_name,
+                # For Select, Subquery, or other SQLAlchemy constructs (e.g., when row conditions are applied),
+                # we cannot use inspector.get_columns() as they are not simple table names.
+                # Raise an exception to trigger the fallback mechanism that uses column reflection.
+                logger.debug(
+                    f"table_selectable is of type {type(table_selectable).__name__}, "
+                    "using column reflection fallback"
+                )
+                raise AttributeError(
+                    "Cannot introspect columns from complex query; using reflection fallback"
                 )
         except (
             KeyError,