Repository: great-expectations/great_expectations
Pull request #6999 (2023-01-31 03:57:09)

[FEATURE] Add SQL query data asset for new experimental datasources
- [x] My code follows the Great Expectations [style guide](https://docs.greatexpectations.io/docs/contributing/style_guides/code_style)
- [x] I have performed a [self-review](https://docs.greatexpectations.io/docs/contributing/contributing_checklist) of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] I have added [unit tests](https://docs.greatexpectations.io/docs/contributing/contributing_test#writing-unit-and-integration-tests) where applicable and made sure that new and existing tests are passing.
- [x] I have run any local integration tests and made sure that nothing is broken.


netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *niobium-lead-7998* processing.


|  Name | Link |
|---------------------------------|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | d7f2c100af26091a335676c25aee9e0444af3f25 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/niobium-lead-7998/deploys/63d89197a946bf00089fd691 |
codesee-maps[bot]: <details>
<summary><b>👇 <a href="https://app.codesee.io/r/reviews?pr=6999&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations">Click on the image</a> for a new way to code review</b>
</summary>

  - Make big changes easier — review code in small groups of related files
  - Know where to start — see the whole change at a glance
  - Take a code tour — explore the change with an interactive tour
  - Make comments and review — all fully sync’ed with github

    [Try it now!](https://app.codesee.io/r/reviews?pr=6999&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)
</details>

[![Review these changes using an interactive CodeSee Map](https://s3.us-east-2.amazonaws.com/maps.codesee.io/images/github/great-expectations/great_expectations/6999/b2919366/d7f2c100af26091a335676c25aee9e0444af3f25.svg)](https://app.codesee.io/r/reviews?pr=6999&src=https%3A%2F%2Fgithub.com%2Fgreat-expectations%2Fgreat_expectations)

#### Legend
![CodeSee Map Legend](https://app.codesee.io/Legend_v2.png)

Patch:
diff --git a/great_expectations/execution_engine/sqlalchemy_execution_engine.py b/great_expectations/execution_engine/sqlalchemy_execution_engine.py
index da703acb8..039acdc8d 100644
--- a/great_expectations/execution_engine/sqlalchemy_execution_engine.py
+++ b/great_expectations/execution_engine/sqlalchemy_execution_engine.py
@@ -1147,6 +1147,15 @@ class SqlAlchemyExecutionEngine(ExecutionEngine):
     def _build_selectable_from_batch_spec(
         self, batch_spec: BatchSpec
     ) -> Union[Selectable, str]:
+        if (
+            batch_spec.get("query") is not None
+            and batch_spec.get("sampling_method") is not None
+        ):
+            raise ValueError(
+                "Sampling is not supported on query data. "
+                "It is currently only supported on table data."
+            )
+
         if "splitter_method" in batch_spec:
             splitter_fn: Callable = self._get_splitter_method(
                 splitter_method_name=batch_spec["splitter_method"]
@@ -1162,7 +1171,7 @@ class SqlAlchemyExecutionEngine(ExecutionEngine):
             else:
                 split_clause = sa.true()
 
-        table_name: str = batch_spec["table_name"]
+        selectable: Selectable = self._subselectable(batch_spec)
         sampling_method: Optional[str] = batch_spec.get("sampling_method")
         if sampling_method is not None:
             if sampling_method in [
@@ -1181,9 +1190,7 @@ class SqlAlchemyExecutionEngine(ExecutionEngine):
                 sampler_fn = self._data_sampler.get_sampler_method(sampling_method)
                 return (
                     sa.select("*")
-                    .select_from(
-                        sa.table(table_name, schema=batch_spec.get("schema_name", None))
-                    )
+                    .select_from(selectable)
                     .where(
                         sa.and_(
                             split_clause,
@@ -1192,13 +1199,23 @@ class SqlAlchemyExecutionEngine(ExecutionEngine):
                     )
                 )
 
-        return (
-            sa.select("*")
-            .select_from(
-                sa.table(table_name, schema=batch_spec.get("schema_name", None))
+        return sa.select("*").select_from(selectable).where(split_clause)
+
+    def _subselectable(self, batch_spec: BatchSpec) -> Selectable:
+        table_name = batch_spec.get("table_name")
+        query = batch_spec.get("query")
+        selectable: Selectable
+        if table_name:
+            selectable = sa.table(
+                table_name, schema=batch_spec.get("schema_name", None)
             )
-            .where(split_clause)
-        )
+        else:
+            if not isinstance(query, str):
+                raise ValueError(f"SQL query should be a str but got {query}")
+            # Query is a valid SELECT query that begins with r"\w+select\w"
+            selectable = sa.select(sa.text(query.lstrip()[6:].lstrip())).subquery()
+
+        return selectable
 
     def get_batch_data_and_markers(
         self, batch_spec: BatchSpec
@@ -1211,6 +1228,18 @@ class SqlAlchemyExecutionEngine(ExecutionEngine):
         RuntimeQueryBatchSpec (illegal type "{str(type(batch_spec))}" was received).
                         """
             )
+        if (
+            sum(
+                1 if x else 0
+                for x in [batch_spec.get("query"), batch_spec.get("table_name")]
+            )
+            != 1
+        ):
+            raise InvalidBatchSpecError(
+                "SqlAlchemyExecutionEngine only accepts a batch_spec where exactly 1 of "
+                "'query' or 'table_name' is specified. "
+                f"table_name={batch_spec.get('table_name')}, query={batch_spec.get('query')}"
+            )
 
         batch_data: Optional[SqlAlchemyBatchData] = None
         batch_markers = BatchMarkers(
diff --git a/great_expectations/experimental/datasources/schemas/PostgresDatasource.json b/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
index 9c0bc12be..f93b6e9d9 100644
--- a/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
+++ b/great_expectations/experimental/datasources/schemas/PostgresDatasource.json
@@ -20,7 +20,14 @@
             "default": {},
             "type": "object",
             "additionalProperties": {
-                "$ref": "#/definitions/TableAsset"
+                "anyOf": [
+                    {
+                        "$ref": "#/definitions/TableAsset"
+                    },
+                    {
+                        "$ref": "#/definitions/QueryAsset"
+                    }
+                ]
             }
         },
         "connection_string": {
@@ -111,17 +118,55 @@
                         "$ref": "#/definitions/BatchSorter"
                     }
                 },
+                "column_splitter": {
+                    "$ref": "#/definitions/SqlYearMonthSplitter"
+                },
                 "table_name": {
                     "title": "Table Name",
                     "type": "string"
+                }
+            },
+            "required": [
+                "name",
+                "table_name"
+            ],
+            "additionalProperties": false
+        },
+        "QueryAsset": {
+            "title": "QueryAsset",
+            "description": "Base model for most ZEP pydantic models.\n\nAdds yaml dumping and parsing methods.\n\nExtra fields are not allowed.\n\nSerialization methods default to `exclude_unset = True` to prevent serializing\nconfigs full of mostly unset default values.\nAlso prevents passing along unset kwargs to BatchSpec.\nhttps://docs.pydantic.dev/usage/exporting_models/",
+            "type": "object",
+            "properties": {
+                "name": {
+                    "title": "Name",
+                    "type": "string"
+                },
+                "type": {
+                    "title": "Type",
+                    "default": "query",
+                    "enum": [
+                        "query"
+                    ],
+                    "type": "string"
+                },
+                "order_by": {
+                    "title": "Order By",
+                    "type": "array",
+                    "items": {
+                        "$ref": "#/definitions/BatchSorter"
+                    }
                 },
                 "column_splitter": {
                     "$ref": "#/definitions/SqlYearMonthSplitter"
+                },
+                "query": {
+                    "title": "Query",
+                    "type": "string"
                 }
             },
             "required": [
                 "name",
-                "table_name"
+                "query"
             ],
             "additionalProperties": false
         }
diff --git a/great_expectations/experimental/datasources/schemas/QueryAsset.json b/great_expectations/experimental/datasources/schemas/QueryAsset.json
new file mode 100644
index 000000000..82f606fa6
--- /dev/null
+++ b/great_expectations/experimental/datasources/schemas/QueryAsset.json
@@ -0,0 +1,90 @@
+{
+    "title": "QueryAsset",
+    "description": "Base model for most ZEP pydantic models.\n\nAdds yaml dumping and parsing methods.\n\nExtra fields are not allowed.\n\nSerialization methods default to `exclude_unset = True` to prevent serializing\nconfigs full of mostly unset default values.\nAlso prevents passing along unset kwargs to BatchSpec.\nhttps://docs.pydantic.dev/usage/exporting_models/",
+    "type": "object",
+    "properties": {
+        "name": {
+            "title": "Name",
+            "type": "string"
+        },
+        "type": {
+            "title": "Type",
+            "default": "query",
+            "enum": [
+                "query"
+            ],
+            "type": "string"
+        },
+        "order_by": {
+            "title": "Order By",
+            "type": "array",
+            "items": {
+                "$ref": "#/definitions/BatchSorter"
+            }
+        },
+        "column_splitter": {
+            "$ref": "#/definitions/SqlYearMonthSplitter"
+        },
+        "query": {
+            "title": "Query",
+            "type": "string"
+        }
+    },
+    "required": [
+        "name",
+        "query"
+    ],
+    "additionalProperties": false,
+    "definitions": {
+        "BatchSorter": {
+            "title": "BatchSorter",
+            "type": "object",
+            "properties": {
+                "key": {
+                    "title": "Key",
+                    "type": "string"
+                },
+                "reverse": {
+                    "title": "Reverse",
+                    "default": false,
+                    "type": "boolean"
+                }
+            },
+            "required": [
+                "key"
+            ]
+        },
+        "SqlYearMonthSplitter": {
+            "title": "SqlYearMonthSplitter",
+            "type": "object",
+            "properties": {
+                "column_name": {
+                    "title": "Column Name",
+                    "type": "string"
+                },
+                "method_name": {
+                    "title": "Method Name",
+                    "default": "split_on_year_and_month",
+                    "enum": [
+                        "split_on_year_and_month"
+                    ],
+                    "type": "string"
+                },
+                "param_names": {
+                    "title": "Param Names",
+                    "type": "array",
+                    "items": {
+                        "enum": [
+                            "year",
+                            "month"
+                        ],
+                        "type": "string"
+                    }
+                }
+            },
+            "required": [
+                "column_name"
+            ]
+        }
+    }
+}
diff --git a/great_expectations/experimental/datasources/schemas/SQLDatasource.json b/great_expectations/experimental/datasources/schemas/SQLDatasource.json
index 1a4110fe6..5fd576cab 100644
--- a/great_expectations/experimental/datasources/schemas/SQLDatasource.json
+++ b/great_expectations/experimental/datasources/schemas/SQLDatasource.json
@@ -1,6 +1,6 @@
 {
     "title": "SQLDatasource",
-    "description": "Adds a generic SQL datasource to the data context.\n\nArgs:\n    name: The name of this datasource.\n    connection_string: The SQLAlchemy connection string used to connect to the database.\n        For example: \"postgresql+psycopg2://postgres:@localhost/test_database\"\n    assets: An optional dictionary whose keys are TableAsset names and whose values\n        are TableAsset objects.",
+    "description": "Adds a generic SQL datasource to the data context.\n\nArgs:\n    name: The name of this datasource.\n    connection_string: The SQLAlchemy connection string used to connect to the database.\n        For example: \"postgresql+psycopg2://postgres:@localhost/test_database\"\n    assets: An optional dictionary whose keys are SQL DataAsset names and whose values\n        are SQL DataAsset objects.",
     "type": "object",
     "properties": {
         "type": {
@@ -20,7 +20,14 @@
             "default": {},
             "type": "object",
             "additionalProperties": {
-                "$ref": "#/definitions/TableAsset"
+                "anyOf": [
+                    {
+                        "$ref": "#/definitions/TableAsset"
+                    },
+                    {
+                        "$ref": "#/definitions/QueryAsset"
+                    }
+                ]
             }
         },
         "connection_string": {
@@ -108,17 +115,55 @@
                         "$ref": "#/definitions/BatchSorter"
                     }
                 },
+                "column_splitter": {
+                    "$ref": "#/definitions/SqlYearMonthSplitter"
+                },
                 "table_name": {
                     "title": "Table Name",
                     "type": "string"
+                }
+            },
+            "required": [
+                "name",
+                "table_name"
+            ],
+            "additionalProperties": false
+        },
+        "QueryAsset": {
+            "title": "QueryAsset",
+            "description": "Base model for most ZEP pydantic models.\n\nAdds yaml dumping and parsing methods.\n\nExtra fields are not allowed.\n\nSerialization methods default to `exclude_unset = True` to prevent serializing\nconfigs full of mostly unset default values.\nAlso prevents passing along unset kwargs to BatchSpec.\nhttps://docs.pydantic.dev/usage/exporting_models/",
+            "type": "object",
+            "properties": {
+                "name": {
+                    "title": "Name",
+                    "type": "string"
+                },
+                "type": {
+                    "title": "Type",
+                    "default": "query",
+                    "enum": [
+                        "query"
+                    ],
+                    "type": "string"
+                },
+                "order_by": {
+                    "title": "Order By",
+                    "type": "array",
+                    "items": {
+                        "$ref": "#/definitions/BatchSorter"
+                    }
                 },
                 "column_splitter": {
                     "$ref": "#/definitions/SqlYearMonthSplitter"
+                },
+                "query": {
+                    "title": "Query",
+                    "type": "string"
                 }
             },
             "required": [
                 "name",
-                "table_name"
+                "query"
             ],
             "additionalProperties": false
         }
diff --git a/great_expectations/experimental/datasources/schemas/TableAsset.json b/great_expectations/experimental/datasources/schemas/TableAsset.json
index d94c025b6..2b57c6a21 100644
--- a/great_expectations/experimental/datasources/schemas/TableAsset.json
+++ b/great_expectations/experimental/datasources/schemas/TableAsset.json
@@ -22,12 +22,12 @@
                 "$ref": "#/definitions/BatchSorter"
             }
         },
+        "column_splitter": {
+            "$ref": "#/definitions/SqlYearMonthSplitter"
+        },
         "table_name": {
             "title": "Table Name",
             "type": "string"
-        },
-        "column_splitter": {
-            "$ref": "#/definitions/SqlYearMonthSplitter"
         }
     },
     "required": [
diff --git a/great_expectations/experimental/datasources/sql_datasource.py b/great_expectations/experimental/datasources/sql_datasource.py
index 510c034d6..cd8ee9ef3 100644
--- a/great_expectations/experimental/datasources/sql_datasource.py
+++ b/great_expectations/experimental/datasources/sql_datasource.py
@@ -5,6 +5,7 @@ import itertools
 from datetime import datetime
 from typing import (
     TYPE_CHECKING,
+    Any,
     Callable,
     Dict,
     List,
@@ -12,6 +13,7 @@ from typing import (
     Optional,
     Sequence,
     Type,
+    Union,
     cast,
 )
 
@@ -56,7 +58,7 @@ class ColumnSplitter:
     method_name: str
     param_names: Sequence[str]
 
-    def param_defaults(self, table_asset: TableAsset) -> Dict[str, List]:
+    def param_defaults(self, sql_asset: _SQLAsset) -> Dict[str, List]:
         raise NotImplementedError
 
     @pydantic.validator("method_name")
@@ -91,33 +93,34 @@ class SqlYearMonthSplitter(ColumnSplitter):
         default_factory=lambda: ["year", "month"]
     )
 
-    def param_defaults(self, table_asset: TableAsset) -> Dict[str, List]:
+    def param_defaults(self, sql_asset: _SQLAsset) -> Dict[str, List]:
         """Query sql database to get the years and months to split over.
 
         Args:
-            table_asset: A TableAsset over which we want to split the data.
+            sql_asset: A SQL DataAsset over which we want to split the data.
         """
         return _query_for_year_and_month(
-            table_asset, self.column_name, _get_sql_datetime_range
+            sql_asset, self.column_name, _get_sql_datetime_range
         )
 
 
 def _query_for_year_and_month(
-    table_asset: TableAsset,
+    sql_asset: _SQLAsset,
     column_name: str,
     query_datetime_range: Callable[
-        [sqlalchemy.engine.base.Connection, str, str], DatetimeRange
+        [sqlalchemy.engine.base.Connection, sqlalchemy.sql.Selectable, str],
+        DatetimeRange,
     ],
 ) -> Dict[str, List]:
     from great_expectations.execution_engine import SqlAlchemyExecutionEngine
 
-    execution_engine = table_asset.datasource.get_execution_engine()
+    execution_engine = sql_asset.datasource.get_execution_engine()
     assert isinstance(execution_engine, SqlAlchemyExecutionEngine)
 
     with execution_engine.engine.connect() as conn:
         datetimes: DatetimeRange = query_datetime_range(
             conn,
-            table_asset.table_name,
+            sql_asset.as_selectable(),
             column_name,
         )
     year: List[int] = list(range(datetimes.min.year, datetimes.max.year + 1))
@@ -130,43 +133,39 @@ def _query_for_year_and_month(
 
 
 def _get_sql_datetime_range(
-    conn: sqlalchemy.engine.base.Connection, table_name: str, col_name: str
+    conn: sqlalchemy.engine.base.Connection,
+    selectable: sqlalchemy.sql.Selectable,
+    col_name: str,
 ) -> DatetimeRange:
-    q = f"select min({col_name}), max({col_name}) from {table_name}"
-    min_max_dt = list(conn.execute(q))[0]
+    import sqlalchemy as sa
+
+    column = sa.column(col_name)
+    query = sa.select([sa.func.min(column), sa.func.max(column)]).select_from(
+        selectable
+    )
+    min_max_dt = list(
+        conn.execute(query.compile(compile_kwargs={"literal_binds": True}))
+    )[0]
+    if min_max_dt[0] is None or min_max_dt[1] is None:
+        raise SQLDatasourceError(
+            f"Data date range can not be determined for the query: {query}. The returned range was {min_max_dt}."
+        )
     return DatetimeRange(min=min_max_dt[0], max=min_max_dt[1])
 
 
-class TableAsset(DataAsset):
+class _SQLAsset(DataAsset):
     # Instance fields
-    type: Literal["table"] = "table"
-    table_name: str
+    type: Literal["_sqlasset"] = "_sqlasset"
     column_splitter: Optional[SqlYearMonthSplitter] = None
     name: str
 
     def test_connection(self) -> None:
-        """Test the connection for the TableAsset.
+        """Test the connection for the _SQLAsset.
 
         Raises:
             TestConnectionError
         """
         assert isinstance(self.datasource, SQLDatasource)
-        schema: Optional[str] = None
-        table_name: str
-        if "." in self.table_name:
-            schema, table_name = self.table_name.split(".")
-        else:
-            table_name = self.table_name
-        engine: sqlalchemy.engine.Engine = self.datasource.get_engine()
-        exists = sqlalchemy.inspect(engine).has_table(
-            table_name=table_name,
-            schema=schema,
-        )
-        if not exists:
-            raise TestConnectionError(
-                f"Attempt to connect to table: {self.table_name} failed because the table "
-                "does not exist."
-            )
 
     def batch_request_options_template(
         self,
@@ -186,14 +185,14 @@ class TableAsset(DataAsset):
     def add_year_and_month_splitter(
         self,
         column_name: str,
-    ) -> TableAsset:
+    ) -> _SQLAsset:
         """Associates a year month splitter with this DataAsset
 
         Args:
             column_name: A column name of the date column where year and month will be parsed out.
 
         Returns:
-            This TableAsset so we can use this method fluently.
+            This DataAsset so we can use this method fluently.
         """
         self.column_splitter = SqlYearMonthSplitter(
             column_name=column_name,
@@ -272,12 +271,7 @@ class TableAsset(DataAsset):
         column_splitter = self.column_splitter
         for request in self._fully_specified_batch_requests(batch_request):
             batch_metadata = copy.deepcopy(request.options)
-            batch_spec_kwargs = {
-                "type": "table",
-                "data_asset_name": self.name,
-                "table_name": self.table_name,
-                "batch_identifiers": {},
-            }
+            batch_spec_kwargs = self._create_batch_spec_kwargs()
             if column_splitter:
                 batch_spec_kwargs["splitter_method"] = column_splitter.method_name
                 batch_spec_kwargs["splitter_kwargs"] = {
@@ -329,6 +323,101 @@ class TableAsset(DataAsset):
         self.sort_batches(batch_list)
         return batch_list
 
+    def _create_batch_spec_kwargs(self) -> dict[str, Any]:
+        """Creates batch_spec_kwargs used to instantiate a SqlAlchemyDatasourceBatchSpec
+
+        This is called by get_batch_list_from_batch_request to generate the batches.
+
+        Returns:
+            A dictionary that will be passed to SqlAlchemyDatasourceBatchSpec(**returned_dict)
+        """
+        raise NotImplementedError
+
+    def as_selectable(self) -> sqlalchemy.sql.Selectable:
+        """Returns a Selectable that can be used to query this data
+
+        Returns:
+            A Selectable that can be used in a from clause to query this data
+        """
+        raise NotImplementedError
+
+
+class QueryAsset(_SQLAsset):
+    # Instance fields
+    type: Literal["query"] = "query"  # type: ignore[assignment]
+    query: str
+
+    @pydantic.validator("query")
+    def query_must_start_with_select(cls, v: str):
+        query = v.lstrip()
+        if not (query.upper().startswith("SELECT") and query[6].isspace()):
+            raise ValueError("query must start with 'SELECT' followed by a whitespace.")
+        return v
+
+    def as_selectable(self) -> sqlalchemy.sql.Selectable:
+        """Returns the Selectable that is used to retrieve the data.
+
+        This can be used in a subselect FROM clause for queries against this data.
+        """
+        import sqlalchemy as sa
+
+        return sa.select(sa.text(self.query.lstrip()[6:])).subquery()
+
+    def _create_batch_spec_kwargs(self) -> dict[str, Any]:
+        return {
+            "data_asset_name": self.name,
+            "query": self.query,
+            "temp_table_schema_name": None,
+            "batch_identifiers": {},
+        }
+
+
+class TableAsset(_SQLAsset):
+    # Instance fields
+    type: Literal["table"] = "table"  # type: ignore[assignment]
+    table_name: str
+
+    def test_connection(self) -> None:
+        """Test the connection for the TableAsset.
+
+        Raises:
+            TestConnectionError
+        """
+        assert isinstance(self.datasource, SQLDatasource)
+        schema: Optional[str] = None
+        table_name: str
+        if "." in self.table_name:
+            schema, table_name = self.table_name.split(".")
+        else:
+            table_name = self.table_name
+        engine: sqlalchemy.engine.Engine = self.datasource.get_engine()
+        exists = sqlalchemy.inspect(engine).has_table(
+            table_name=table_name,
+            schema=schema,
+        )
+        if not exists:
+            raise TestConnectionError(
+                f"Attempt to connect to table: {self.table_name} failed because the table "
+                "does not exist."
+            )
+
+    def as_selectable(self) -> sqlalchemy.sql.Selectable:
+        """Returns the table as a sqlalchemy Selectable.
+
+        This can be used in a from clause for a query against this data.
+        """
+        import sqlalchemy as sa
+
+        return sa.text(self.table_name)
+
+    def _create_batch_spec_kwargs(self) -> dict[str, Any]:
+        return {
+            "type": "table",
+            "data_asset_name": self.name,
+            "table_name": self.table_name,
+            "batch_identifiers": {},
+        }
+
 
 class SQLDatasource(Datasource):
     """Adds a generic SQL datasource to the data context.
@@ -337,18 +426,20 @@ class SQLDatasource(Datasource):
         name: The name of this datasource.
         connection_string: The SQLAlchemy connection string used to connect to the database.
             For example: "postgresql+psycopg2://postgres:@localhost/test_database"
-        assets: An optional dictionary whose keys are TableAsset names and whose values
-            are TableAsset objects.
+        assets: An optional dictionary whose keys are SQL DataAsset names and whose values
+            are SQL DataAsset objects.
     """
 
     # class var definitions
-    asset_types: ClassVar[List[Type[DataAsset]]] = [TableAsset]
+    asset_types: ClassVar[List[Type[DataAsset]]] = [TableAsset, QueryAsset]
 
     # right side of the operator determines the type name
     # left side enforces the names on instance creation
     type: Literal["sql"] = "sql"
     connection_string: str
-    assets: Dict[str, TableAsset] = {}
+    # We need to explicitly add each asset type to the Union due to how
+    # deserialization is implemented in our pydantic base model.
+    assets: Dict[str, Union[TableAsset, QueryAsset]] = {}
 
     # private attrs
     _engine: sqlalchemy.engine.Engine = pydantic.PrivateAttr()
@@ -430,3 +521,26 @@ class SQLDatasource(Datasource):
             # see DataAsset._parse_order_by_sorter()
         )
         return self.add_asset(asset)
+
+    def add_query_asset(
+        self,
+        name: str,
+        query: str,
+        order_by: Optional[BatchSortersDefinition] = None,
+    ) -> QueryAsset:
+        """Adds a query asset to this datasource.
+
+        Args:
+            name: The name of this table asset.
+            query: The SELECT query to selects the data to validate. It must begin with the "SELECT".
+            order_by: A list of BatchSorters or BatchSorter strings.
+
+        Returns:
+            The QueryAsset that is added to the datasource.
+        """
+        asset = QueryAsset(
+            name=name,
+            query=query,
+            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+        )
+        return self.add_asset(asset)
diff --git a/great_expectations/experimental/datasources/sqlite_datasource.py b/great_expectations/experimental/datasources/sqlite_datasource.py
index c7176fbfb..99781094b 100644
--- a/great_expectations/experimental/datasources/sqlite_datasource.py
+++ b/great_expectations/experimental/datasources/sqlite_datasource.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import TYPE_CHECKING, Dict, List, Optional, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Type, TypeVar, Union
 
 import pydantic
 from pydantic import dataclasses as pydantic_dc
@@ -18,24 +18,23 @@ from great_expectations.experimental.datasources.sql_datasource import (
     BatchSortersDefinition,
     ColumnSplitter,
     DatetimeRange,
+    QueryAsset,
     SQLDatasource,
     SQLDatasourceError,
     TableAsset,
     _query_for_year_and_month,
+    _SQLAsset,
 )
 
 
-class SqliteTableAsset(TableAsset):
-    # Subclass overrides
-    type: Literal["sqlite_table"] = "sqlite_table"  # type: ignore[assignment]
-    column_splitter: Optional[SqliteYearMonthSplitter] = None  # type: ignore[assignment]
-
-    # This asset type will support a variety of splitters
+class _SplitterMixin:
     def add_year_and_month_splitter(
         self,
         column_name: str,
-    ) -> SqliteTableAsset:
-        """Associates a year month splitter with this sqlite table data asset
+    ) -> _SQLAsset:
+        """Associates a year month splitter with this sqlite data asset.
+
+        This must be used as a mixin for a sqlite data asset.
 
         Args:
             column_name: A column name of the date column where year and month will be parsed out.
@@ -43,10 +42,25 @@ class SqliteTableAsset(TableAsset):
         Returns:
             This SqliteTableAsset so we can use this method fluently.
         """
+        assert isinstance(
+            self, tuple(_SqliteAssets)
+        ), "_SplitterMixin can only be mixed into a sqlite data asset."
         self.column_splitter = SqliteYearMonthSplitter(
             column_name=column_name,
         )
-        return self
+        return self  # type: ignore[return-value]  # See isinstance check above.
+
+
+class SqliteTableAsset(_SplitterMixin, TableAsset):
+    # Subclass overrides
+    type: Literal["sqlite_table"] = "sqlite_table"  # type: ignore[assignment]
+    column_splitter: Optional[SqliteYearMonthSplitter] = None  # type: ignore[assignment]
+
+
+class SqliteQueryAsset(_SplitterMixin, QueryAsset):
+    # Subclass overrides
+    type: Literal["sqlite_query"] = "sqlite_query"  # type: ignore[assignment]
+    column_splitter: Optional[SqliteYearMonthSplitter] = None  # type: ignore[assignment]
 
 
 @pydantic_dc.dataclass(frozen=True)
@@ -57,31 +71,57 @@ class SqliteYearMonthSplitter(ColumnSplitter):
         default_factory=lambda: ["year", "month"]
     )
 
-    def param_defaults(self, table_asset: TableAsset) -> Dict[str, List]:
+    def param_defaults(self, sql_asset: _SQLAsset) -> Dict[str, List]:
         """Query sqlite database to get the years and months to split over.
 
         Args:
-            table_asset: A SqliteTableAsset over which we want to split the data.
+            sql_asset: A Sqlite*Asset over which we want to split the data.
         """
-        if not isinstance(table_asset, SqliteTableAsset):
+        if not isinstance(sql_asset, tuple(_SqliteAssets)):
             raise SQLDatasourceError(
-                "Table asset passed to SqliteYearMonthSplitter is not a SqliteTableAsset. It is "
-                f"{table_asset}"
+                "SQL asset passed to SqliteYearMonthSplitter is not a Sqlite*Asset. It is "
+                f"{sql_asset}"
             )
 
         return _query_for_year_and_month(
-            table_asset, self.column_name, _get_sqlite_datetime_range
+            sql_asset, self.column_name, _get_sqlite_datetime_range
         )
 
 
 def _get_sqlite_datetime_range(
-    conn: sqlalchemy.engine.base.Connection, table_name: str, col_name: str
+    conn: sqlalchemy.engine.base.Connection,
+    selectable: sqlalchemy.sql.Selectable,
+    col_name: str,
 ) -> DatetimeRange:
-    q = f"select STRFTIME('%Y%m%d', min({col_name})), STRFTIME('%Y%m%d', max({col_name})) from {table_name}"
-    min_max_dt = [datetime.strptime(dt, "%Y%m%d") for dt in list(conn.execute(q))[0]]
+    import sqlalchemy as sa
+
+    column = sa.column(col_name)
+    query = sa.select(
+        [
+            sa.func.strftime("%Y%m%d", sa.func.min(column)),
+            sa.func.strftime("%Y%m%d", sa.func.max(column)),
+        ]
+    ).select_from(selectable)
+    min_max_dt = [
+        datetime.strptime(dt, "%Y%m%d")
+        for dt in list(
+            conn.execute(query.compile(compile_kwargs={"literal_binds": True}))
+        )[0]
+    ]
+    if min_max_dt[0] is None or min_max_dt[1] is None:
+        raise SQLDatasourceError(
+            f"Data date range can not be determined for the query: {query}. The returned range was {min_max_dt}."
+        )
     return DatetimeRange(min=min_max_dt[0], max=min_max_dt[1])
 
 
+_SqliteAssets: List[Type[DataAsset]] = [SqliteTableAsset, SqliteQueryAsset]
+# Unfortunately the following types can't be derived from _SqliteAssets above because mypy doesn't
+# support programmatically unrolling this list, eg Union[*_SqliteAssets] is not supported.
+SqliteAssetTypes = Union[SqliteTableAsset, SqliteQueryAsset]
+SqliteAssetType = TypeVar("SqliteAssetType", SqliteTableAsset, SqliteQueryAsset)
+
+
 class SqliteDsn(pydantic.AnyUrl):
     allowed_schemes = {
         "sqlite",
@@ -104,14 +144,14 @@ class SqliteDatasource(SQLDatasource):
     """
 
     # class var definitions
-    asset_types: ClassVar[List[Type[DataAsset]]] = [SqliteTableAsset]
+    asset_types: ClassVar[List[Type[DataAsset]]] = _SqliteAssets
 
     # Subclass instance var overrides
     # right side of the operator determines the type name
     # left side enforces the names on instance creation
     type: Literal["sqlite"] = "sqlite"  # type: ignore[assignment]
     connection_string: SqliteDsn
-    assets: Dict[str, SqliteTableAsset] = {}  # type: ignore[assignment]
+    assets: Dict[str, SqliteAssetTypes] = {}  # type: ignore[assignment]
 
     def add_table_asset(
         self,
@@ -138,3 +178,29 @@ class SqliteDatasource(SQLDatasource):
         asset._datasource = self
         self.assets[name] = asset
         return asset
+
+    def add_query_asset(
+        self,
+        name: str,
+        query: str,
+        order_by: Optional[BatchSortersDefinition] = None,
+    ) -> SqliteQueryAsset:
+        """Adds a sqlite query asset to this sqlite datasource.
+
+        Args:
+            name: The name of this table asset.
+            query: The SELECT query to selects the data to validate. It must begin with the "SELECT".
+            order_by: A list of BatchSorters or BatchSorter strings.
+
+        Returns:
+            The SqliteTableAsset that is added to the datasource.
+        """
+        asset = SqliteQueryAsset(
+            name=name,
+            query=query,
+            order_by=order_by or [],  # type: ignore[arg-type]  # coerce list[str]
+            # see TableAsset._parse_order_by_sorter()
+        )
+        asset._datasource = self
+        self.assets[name] = asset
+        return asset


Test patch:
diff --git a/tests/experimental/datasources/integration/conftest.py b/tests/experimental/datasources/integration/conftest.py
index 8c9c191b7..8e045c9bb 100644
--- a/tests/experimental/datasources/integration/conftest.py
+++ b/tests/experimental/datasources/integration/conftest.py
@@ -6,6 +6,7 @@ import pathlib
 import pytest
 
 from great_expectations.data_context import AbstractDataContext
+from great_expectations.experimental.datasources import SqliteDatasource
 from great_expectations.experimental.datasources.interfaces import (
     BatchRequest,
     DataAsset,
@@ -35,9 +36,9 @@ def pandas_data(
     return context, pandas_ds, asset, batch_request
 
 
-def sql_data(
-    context: AbstractDataContext,
-) -> tuple[AbstractDataContext, Datasource, DataAsset, BatchRequest]:
+def sqlite_datasource(
+    context: AbstractDataContext, db_filename: str
+) -> SqliteDatasource:
     relative_path = pathlib.Path(
         "..",
         "..",
@@ -45,13 +46,20 @@ def sql_data(
         "test_sets",
         "taxi_yellow_tripdata_samples",
         "sqlite",
-        "yellow_tripdata.db",
+        db_filename,
     )
     db_file = pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
     datasource = context.sources.add_sqlite(
         name="test_datasource",
         connection_string=f"sqlite:///{db_file}",
     )
+    return datasource
+
+
+def sql_data(
+    context: AbstractDataContext,
+) -> tuple[AbstractDataContext, Datasource, DataAsset, BatchRequest]:
+    datasource = sqlite_datasource(context, "yellow_tripdata.db")
     asset = (
         datasource.add_table_asset(
             name="my_asset",
@@ -109,19 +117,8 @@ def multibatch_pandas_data(
 def multibatch_sql_data(
     context: AbstractDataContext,
 ) -> tuple[AbstractDataContext, Datasource, DataAsset, BatchRequest]:
-    relative_path = pathlib.Path(
-        "..",
-        "..",
-        "..",
-        "test_sets",
-        "taxi_yellow_tripdata_samples",
-        "sqlite",
-        "yellow_tripdata_sample_2020_all_months_combined.db",
-    )
-    db_file = pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
-    datasource = context.sources.add_sqlite(
-        name="test_datasource",
-        connection_string=f"sqlite:///{db_file}",
+    datasource = sqlite_datasource(
+        context, "yellow_tripdata_sample_2020_all_months_combined.db"
     )
     asset = (
         datasource.add_table_asset(
diff --git a/tests/experimental/datasources/integration/test_integration_datasource.py b/tests/experimental/datasources/integration/test_integration_datasource.py
index cfc1d4855..aeced9b6c 100644
--- a/tests/experimental/datasources/integration/test_integration_datasource.py
+++ b/tests/experimental/datasources/integration/test_integration_datasource.py
@@ -8,6 +8,7 @@ from great_expectations.experimental.datasources.interfaces import (
     DataAsset,
     Datasource,
 )
+from tests.experimental.datasources.integration.conftest import sqlite_datasource
 from tests.experimental.datasources.integration.integration_test_utils import (
     run_batch_head,
     run_checkpoint_and_data_doc,
@@ -87,3 +88,26 @@ def test_batch_head(
         n_rows=n_rows,
         success=success,
     )
+
+
+def test_sql_query_data_asset(empty_data_context):
+    context = empty_data_context
+    datasource = sqlite_datasource(context, "yellow_tripdata.db")
+    passenger_count_value = 5
+    asset = (
+        datasource.add_query_asset(
+            name="query_asset",
+            query=f"   SELECT * from yellow_tripdata_sample_2019_02 WHERE passenger_count = {passenger_count_value}",
+        )
+        .add_year_and_month_splitter(column_name="pickup_datetime")
+        .add_sorters(["year"])
+    )
+    validator = context.get_validator(
+        batch_request=asset.get_batch_request({"year": 2019})
+    )
+    result = validator.expect_column_distinct_values_to_equal_set(
+        column="passenger_count",
+        value_set=[passenger_count_value],
+        result_format={"result_format": "BOOLEAN_ONLY"},
+    )
+    assert result.success
diff --git a/tests/experimental/datasources/test_config.py b/tests/experimental/datasources/test_config.py
index 7fb9dfce2..1e237bc3e 100644
--- a/tests/experimental/datasources/test_config.py
+++ b/tests/experimental/datasources/test_config.py
@@ -341,9 +341,13 @@ def test_catch_bad_asset_configs(
     print(f"\n{exc_info.typename}:{exc_info.value}")
 
     all_errors = exc_info.value.errors()
-    assert len(all_errors) == 1, "Expected 1 error"
-    assert expected_error_loc == all_errors[0]["loc"]
-    assert expected_msg == all_errors[0]["msg"]
+    assert len(all_errors) >= 1, "Expected at least 1 error"
+    test_msg = ""
+    for error in all_errors:
+        if expected_error_loc == all_errors[0]["loc"]:
+            test_msg = error["msg"]
+            break
+    assert expected_msg == test_msg
 
 
 @pytest.mark.unit
diff --git a/tests/experimental/datasources/test_postgres_datasource.py b/tests/experimental/datasources/test_postgres_datasource.py
index 509446261..b0501b7fa 100644
--- a/tests/experimental/datasources/test_postgres_datasource.py
+++ b/tests/experimental/datasources/test_postgres_datasource.py
@@ -700,3 +700,35 @@ def test_validate_invalid_postgres_connection_string(
             connection_string=connection_string,
         ):
             pass
+
+
+@pytest.mark.unit
+def test_query_data_asset(create_source):
+    query = "SELECT * FROM my_table"
+
+    def validate_batch_spec(spec: SqlAlchemyDatasourceBatchSpec) -> None:
+        assert spec == {
+            "data_asset_name": "query_asset",
+            "query": "SELECT * FROM my_table",
+            "temp_table_schema_name": None,
+            "batch_identifiers": {},
+        }
+
+    with create_source(
+        validate_batch_spec=validate_batch_spec, dialect="postgresql"
+    ) as source:
+        asset = source.add_query_asset(
+            name="query_asset", query="SELECT * FROM my_table"
+        )
+        assert asset.name == "query_asset"
+        assert asset.query.lower() == query.lower()
+        source.get_batch_list_from_batch_request(asset.get_batch_request())
+
+
+@pytest.mark.unit
+def test_non_select_query_data_asset(create_source):
+    with create_source(
+        validate_batch_spec=lambda _: None, dialect="postgresql"
+    ) as source:
+        with pytest.raises(ValueError):
+            source.add_query_asset(name="query_asset", query="* FROM my_table")
diff --git a/tests/experimental/datasources/test_schemas.py b/tests/experimental/datasources/test_schemas.py
index 122486b9a..34cace99a 100644
--- a/tests/experimental/datasources/test_schemas.py
+++ b/tests/experimental/datasources/test_schemas.py
@@ -51,6 +51,7 @@ def test_vcs_schemas_match(zep_ds_or_asset_model: Type[pydantic.BaseModel]):
     if schema_path.name in (
         "SqliteDatasource.json",
         "SqliteTableAsset.json",
+        "SqliteQueryAsset.json",
     ):
         pytest.xfail(f"{schema_path.name} does not exist")
 
diff --git a/tests/experimental/datasources/test_sqlite_datasource.py b/tests/experimental/datasources/test_sqlite_datasource.py
index 14893f98c..951b69ed1 100644
--- a/tests/experimental/datasources/test_sqlite_datasource.py
+++ b/tests/experimental/datasources/test_sqlite_datasource.py
@@ -7,10 +7,13 @@ from great_expectations.experimental.datasources import SqliteDatasource
 from great_expectations.experimental.datasources.sqlite_datasource import SqliteDsn
 
 
-@pytest.mark.unit
-def test_connection_string_starts_with_sqlite():
-    # The actual file doesn't matter only it's existence since SqlAlchemy does a check
-    # when it creates the database engine.
+@pytest.fixture
+def sqlite_datasource_name() -> str:
+    return "sqlite_datasource"
+
+
+@pytest.fixture
+def sqlite_database_path() -> pathlib.Path:
     relative_path = pathlib.Path(
         "..",
         "..",
@@ -19,15 +22,27 @@ def test_connection_string_starts_with_sqlite():
         "sqlite",
         "yellow_tripdata.db",
     )
-    db_file = pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
-    name = "sqlite_datasource"
-    connection_string = f"sqlite:///{db_file}"
+    return pathlib.Path(__file__).parent.joinpath(relative_path).resolve(strict=True)
+
+
+@pytest.fixture
+def sqlite_datasource(sqlite_database_path, sqlite_datasource_name) -> SqliteDatasource:
+    connection_string = f"sqlite:///{sqlite_database_path}"
     datasource = SqliteDatasource(
-        name=name,
-        connection_string=connection_string,
+        name=sqlite_datasource_name,
+        connection_string=connection_string,  # type: ignore[arg-type]  # pydantic will coerce
     )
-    assert datasource.name == name
-    assert datasource.connection_string == connection_string
+    return datasource
+
+
+@pytest.mark.unit
+def test_connection_string_starts_with_sqlite(
+    sqlite_datasource, sqlite_database_path, sqlite_datasource_name
+):
+    # The actual file doesn't matter only it's existence since SqlAlchemy does a check
+    # when it creates the database engine.
+    assert sqlite_datasource.name == sqlite_datasource_name
+    assert sqlite_datasource.connection_string == f"sqlite:///{sqlite_database_path}"
 
 
 @pytest.mark.unit
@@ -45,3 +60,9 @@ def test_connection_string_that_does_not_start_with_sqlite():
         "  URL scheme not permitted (type=value_error.url.scheme; "
         f"allowed_schemes={SqliteDsn.allowed_schemes})"
     )
+
+
+@pytest.mark.unit
+def test_non_select_query_asset(sqlite_datasource):
+    with pytest.raises(ValueError):
+        sqlite_datasource.add_query_asset(name="query_asset", query="* from table")