diff --git a/docs/source/custom_expectations.rst b/docs/source/custom_expectations.rst
index f7340871b..419814e8d 100644
--- a/docs/source/custom_expectations.rst
+++ b/docs/source/custom_expectations.rst
@@ -74,7 +74,7 @@ For SqlAlchemyDataset, the decorators work slightly differently. See the MetaSql
             mode_query = sa.select([
                 sa.column(column).label('value'),
                 sa.func.count(sa.column(column)).label('frequency')
-            ]).select_from(sa.table(self.table_name)).group_by(sa.column(column)).order_by(sa.desc(sa.column('frequency')))
+            ]).select_from(self._table).group_by(sa.column(column)).order_by(sa.desc(sa.column('frequency')))
 
             mode = self.engine.execute(mode_query).scalar()
             return {
diff --git a/docs/source/roadmap_changelog.rst b/docs/source/roadmap_changelog.rst
index 4563e1084..339696554 100644
--- a/docs/source/roadmap_changelog.rst
+++ b/docs/source/roadmap_changelog.rst
@@ -14,6 +14,7 @@ Planned Features
 
 v.0.4.4__develop
 ----------------
+* Add support for custom schema in SqlAlchemyDataset (#370)
 
 v.0.4.4
 ----------------
diff --git a/great_expectations/dataset/sqlalchemy_dataset.py b/great_expectations/dataset/sqlalchemy_dataset.py
index 8028e95fe..ec86b4416 100644
--- a/great_expectations/dataset/sqlalchemy_dataset.py
+++ b/great_expectations/dataset/sqlalchemy_dataset.py
@@ -67,7 +67,7 @@ class MetaSqlAlchemyDataset(Dataset):
                                       sa.column(column).is_(None) == False if None in ignore_values else True),
                               1)], else_=0)
                 ).label('unexpected_count')
-            ]).select_from(sa.table(self.table_name))
+            ]).select_from(self._table)
 
             count_results = dict(self.engine.execute(count_query).fetchone())
 
@@ -81,7 +81,7 @@ class MetaSqlAlchemyDataset(Dataset):
 
             # Retrieve unexpected  values
             unexpected_query_results = self.engine.execute(
-                sa.select([sa.column(column)]).select_from(sa.table(self.table_name)).where(
+                sa.select([sa.column(column)]).select_from(self._table).where(
                     sa.and_(sa.not_(expected_condition),
                             sa.or_(
                                 # SA normally evaluates `== None` as `IS NONE`. However `sa.in_()`
@@ -164,7 +164,7 @@ class MetaSqlAlchemyDataset(Dataset):
                     sa.func.sum(
                         sa.case([(sa.column(column) == None, 1)], else_=0)
                     ).label('null_count'),
-                ]).select_from(sa.table(self.table_name))
+                ]).select_from(self._table)
 
                 count_results = dict(self.engine.execute(count_query).fetchone())
 
@@ -204,11 +204,12 @@ class MetaSqlAlchemyDataset(Dataset):
 
 class SqlAlchemyDataset(MetaSqlAlchemyDataset):
 
-    def __init__(self, table_name=None, engine=None, connection_string=None, custom_sql=None, *args, **kwargs):
+    def __init__(self, table_name=None, engine=None, connection_string=None,
+                 custom_sql=None, schema=None, *args, **kwargs):
         if table_name is None:
             raise ValueError("No table_name provided.")
 
-        self.table_name = table_name
+        self._table = sa.Table(table_name, sa.MetaData(), schema=schema)
 
         if engine is None and connection_string is None:
             raise ValueError("Engine or connection_string must be provided.")
@@ -223,11 +224,17 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
                 # Currently we do no error handling if the engine doesn't work out of the box.
                 raise err
 
+        if schema is not None and custom_sql is not None:
+            # temporary table will be written to temp schema, so don't allow
+            # a user-defined schema
+            raise ValueError("Cannot specify both schema and custom_sql.")
+
+
         if custom_sql:
-            self.create_temporary_table(self.table_name, custom_sql)
+            self.create_temporary_table(table_name, custom_sql)
 
         insp = reflection.Inspector.from_engine(engine)
-        self.columns = insp.get_columns(self.table_name)
+        self.columns = insp.get_columns(table_name, schema=schema)
 
         # Only call super once connection is established and table_name and columns known to allow autoinspection
         super(SqlAlchemyDataset, self).__init__(*args, **kwargs)
@@ -282,7 +289,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
         if value is None:
             raise ValueError("value must be provided")
 
-        count_query = sa.select([sa.func.count()]).select_from(sa.table(self.table_name))
+        count_query = sa.select([sa.func.count()]).select_from(self._table)
         row_count = self.engine.execute(count_query).scalar()
 
         return {
@@ -310,7 +317,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
         except ValueError:
             raise ValueError("min_value and max_value must be integers")
 
-        count_query = sa.select([sa.func.count()]).select_from(sa.table(self.table_name))
+        count_query = sa.select([sa.func.count()]).select_from(self._table)
         row_count = self.engine.execute(count_query).scalar()
 
         if min_value != None and max_value != None:
@@ -521,7 +528,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
             raise ValueError("parse_strings_as_datetimes is not supported in SqlAlchemy")
 
         col_max = self.engine.execute(
-            sa.select([sa.func.max(sa.column(column))]).select_from(sa.table(self.table_name))
+            sa.select([sa.func.max(sa.column(column))]).select_from(self._table)
         ).scalar()
 
         # Handle possible missing values
@@ -568,7 +575,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
             raise ValueError("parse_strings_as_datetimes is not supported in SqlAlchemy")
 
         col_min = self.engine.execute(
-            sa.select([sa.func.min(sa.column(column))]).select_from(sa.table(self.table_name))
+            sa.select([sa.func.min(sa.column(column))]).select_from(self._table)
         ).scalar()
 
         # Handle possible missing values
@@ -609,7 +616,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
             raise ValueError("min_value and max_value cannot both be None")
 
         col_sum = self.engine.execute(
-            sa.select([sa.func.sum(sa.column(column))]).select_from(sa.table(self.table_name))
+            sa.select([sa.func.sum(sa.column(column))]).select_from(self._table)
         ).scalar()
 
         # Handle possible missing values
@@ -659,7 +666,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
             raise ValueError("column is not numeric")
 
         col_avg = self.engine.execute(
-            sa.select([sa.func.avg(sa.column(column))]).select_from(sa.table(self.table_name))
+            sa.select([sa.func.avg(sa.column(column))]).select_from(self._table)
         ).scalar()
 
         # Handle possible missing values
@@ -707,7 +714,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
                 sa.func.sum(
                     sa.case([(sa.column(column) == None, 1)], else_=0)
                 ).label('null_count')
-            ]).select_from(sa.table(self.table_name))
+            ]).select_from(self._table)
         )
 
         counts = dict(count_query.fetchone())
@@ -724,7 +731,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
         element_values = self.engine.execute(
             sa.select([sa.column(column)]).order_by(sa.column(column)).where(
                 sa.column(column) != None
-            ).offset(nonnull_count // 2 - 1).limit(2).select_from(sa.table(self.table_name))
+            ).offset(nonnull_count // 2 - 1).limit(2).select_from(self._table)
         )
 
         column_values = list(element_values.fetchall())
@@ -761,7 +768,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
                                           result_format=None, include_config=False, catch_exceptions=None, meta=None):
         # Duplicates are found by filtering a group by query
         dup_query = sa.select([sa.column(column)]).\
-            select_from(sa.table(self.table_name)).\
+            select_from(self._table).\
             group_by(sa.column(column)).\
             having(sa.func.count(sa.column(column)) > 1)
 
@@ -777,7 +784,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
             raise ValueError("min_value and max_value cannot both be None")
 
         unique_value_count = self.engine.execute(
-            sa.select([sa.func.count(sa.func.distinct(sa.column(column)))]).select_from(sa.table(self.table_name))
+            sa.select([sa.func.count(sa.func.distinct(sa.column(column)))]).select_from(self._table)
         ).scalar()
 
         # Handle possible missing values
@@ -814,7 +821,7 @@ class SqlAlchemyDataset(MetaSqlAlchemyDataset):
                     sa.case([(sa.column(column) == None, 1)], else_=0)
                 ).label('null_count'),
                 sa.func.count(sa.func.distinct(sa.column(column))).label('unique_value_count')
-            ]).select_from(sa.table(self.table_name))
+            ]).select_from(self._table)
         )
 
         counts = count_query.fetchone()

Allow passing a schema name to SqlAlchemyDataset
Currently, SqlAlchemyDataset accepts a table name but not a schema name. I'm currently working in Redshift. When I try to instantiate a `SqlAlchemyDataset`, I get a `KeyError` with `public.<tablename>`. After some inspection, I determined that the issue is popping up in this line:https://github.com/great-expectations/great_expectations/blob/d7a4ee03cfce8bea66c5f91573bb87e4385b1eb1/great_expectations/dataset/sqlalchemy_dataset.py#L222

Because I'm not passing a schema name, it uses the default schema, which is "public" for Redshift. When I run this line myself but specifying the schema name, the line runs successfully. Accessing non-default schemas seems like a common use case, so a schema argument would be helpful in many other cases.
diff --git a/tests/sqlalchemy_dataset/test_sqlalchemydataset.py b/tests/sqlalchemy_dataset/test_sqlalchemydataset.py
index 4c32c47b3..59c2c3eff 100644
--- a/tests/sqlalchemy_dataset/test_sqlalchemydataset.py
+++ b/tests/sqlalchemy_dataset/test_sqlalchemydataset.py
@@ -18,7 +18,7 @@ def custom_dataset():
             mode_query = sa.select([
                 sa.column(column).label('value'),
                 sa.func.count(sa.column(column)).label('frequency')
-            ]).select_from(sa.table(self.table_name)).group_by(sa.column(column)).order_by(
+            ]).select_from(self._table).group_by(sa.column(column)).order_by(
                 sa.desc(sa.column('frequency')))
 
             mode = self.engine.execute(mode_query).scalar()
@@ -85,6 +85,21 @@ def test_broken_decorator_errors(custom_dataset):
         assert "Column aggregate expectation failed to return required information: observed_value" in str(err)
 
 
+def test_missing_engine_error():
+    with pytest.raises(ValueError) as err:
+        SqlAlchemyDataset('test_engine', schema='example')
+        assert "Engine or connection_string must be provided." in str(err)
+
+
+def test_schema_custom_sql_error():
+    engine = sa.create_engine('sqlite://')
+
+    with pytest.raises(ValueError) as err:
+        SqlAlchemyDataset('test_schema_custom', schema='example', engine=engine,
+                          custom_sql='SELECT * FROM example.fake')
+        assert "Cannot specify both schema and custom_sql." in str(err)
+
+
 def test_sqlalchemydataset_with_custom_sql():
     engine = sa.create_engine('sqlite://')
 

pip install -e .
parse_log_pytest
requirements.txt
pytest
apt-get update apt-get install -y gcc
3.6
requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==22.2.0
certifi==2021.5.30
-e git+https://github.com/great-expectations/great_expectations.git@c0ab4f922b607f135caf6dac7bed16f61dbdeead#egg=great_expectations
greenlet==2.0.2
importlib-metadata==4.8.3
iniconfig==1.1.1
jsonschema==3.2.0
numpy==1.19.5
packaging==21.3
pandas==1.1.5
pluggy==1.0.0
py==1.11.0
pyparsing==3.1.4
pyrsistent==0.18.0
pytest==7.0.1
python-dateutil==2.9.0.post0
pytz==2025.2
scipy==1.5.4
six==1.17.0
SQLAlchemy==1.4.54
tomli==1.2.3
typing_extensions==4.1.1
zipp==3.6.0

name: great_expectations
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==22.2.0
      - greenlet==2.0.2
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - jsonschema==3.2.0
      - numpy==1.19.5
      - packaging==21.3
      - pandas==1.1.5
      - pluggy==1.0.0
      - py==1.11.0
      - pyparsing==3.1.4
      - pyrsistent==0.18.0
      - pytest==7.0.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - scipy==1.5.4
      - six==1.17.0
      - sqlalchemy==1.4.54
      - tomli==1.2.3
      - typing-extensions==4.1.1
      - zipp==3.6.0
prefix: /opt/conda/envs/great_expectations

tests/sqlalchemy_dataset/test_sqlalchemydataset.py::test_custom_sqlalchemydataset tests/sqlalchemy_dataset/test_sqlalchemydataset.py::test_schema_custom_sql_error
tests/sqlalchemy_dataset/test_sqlalchemydataset.py::test_sqlalchemydataset_with_custom_sql
tests/sqlalchemy_dataset/test_sqlalchemydataset.py::test_broken_decorator_errors tests/sqlalchemy_dataset/test_sqlalchemydataset.py::test_missing_engine_error
docs/source/roadmap_changelog.rst great_expectations/dataset/sqlalchemy_dataset.py docs/source/custom_expectations.rst
docs/source/roadmap_changelog.rst great_expectations/dataset/sqlalchemy_dataset.py docs/source/custom_expectations.rst