From 3f9cc927e13fdb49a0b2fd8f85b39e8eb6ef2377 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Fri, 29 Jan 2021 16:36:36 -0600
Subject: [PATCH 01/12] feat: add `Database.list_tables` method

---
 google/cloud/spanner_v1/database.py | 21 ++++++++++++++++-
 google/cloud/spanner_v1/table.py    | 36 +++++++++++++++++++++++++++++
 2 files changed, 56 insertions(+), 1 deletion(-)
 create mode 100644 google/cloud/spanner_v1/table.py

diff --git a/google/cloud/spanner_v1/database.py b/google/cloud/spanner_v1/database.py
index c1c7953648..902e9daa25 100644
--- a/google/cloud/spanner_v1/database.py
+++ b/google/cloud/spanner_v1/database.py
@@ -48,7 +48,8 @@
 from google.cloud.spanner_admin_database_v1 import CreateDatabaseRequest
 from google.cloud.spanner_admin_database_v1 import UpdateDatabaseDdlRequest
 from google.cloud.spanner_v1 import ExecuteSqlRequest
-from google.cloud.spanner_v1 import (
+from google.cloud.spanner_v1.table import Table
+from google.cloud.spanner_v1.proto.transaction_pb2 import (
     TransactionSelector,
     TransactionOptions,
 )
@@ -67,6 +68,11 @@
 
 _DATABASE_METADATA_FILTER = "name:{0}/operations/"
 
+_LIST_TABLES_QUERY = """SELECT TABLE_NAME
+FROM INFORMATION_SCHEMA.TABLES
+WHERE SPANNER_STATE = 'COMMITTED'
+"""
+
 DEFAULT_RETRY_BACKOFF = Retry(initial=0.02, maximum=32, multiplier=1.3)
 
 
@@ -596,6 +602,19 @@ def list_database_operations(self, filter_="", page_size=None):
             filter_=database_filter, page_size=page_size
         )
 
+    def list_tables(self):
+        """List tables within the database.
+
+        :type: Iterable
+        :returns:
+            Iterable of :class:`~google.cloud.spanner_v1.table.Table`
+            resources within the current database.
+        """
+        with self.snapshot() as snapshot:
+            results = snapshot.execute_sql(_LIST_TABLES_QUERY)
+            for row in results:
+                yield Table(row[0], self)
+
 
 class BatchCheckout(object):
     """Context manager for using a batch from a database.
diff --git a/google/cloud/spanner_v1/table.py b/google/cloud/spanner_v1/table.py
new file mode 100644
index 0000000000..1b80c98d87
--- /dev/null
+++ b/google/cloud/spanner_v1/table.py
@@ -0,0 +1,36 @@
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""User friendly container for Cloud Spanner Table."""
+
+_GET_SCHEMA_TEMPLATE = "SELECT * FROM {} LIMIT 0"
+
+
+class Table(object):
+    """Representation of a Cloud Spanner Table.
+    """
+
+    def __init__(self, table_id, database):
+        self.table_id = table_id
+        self._database = database
+
+    def get_schema(self):
+        """
+        List of google.cloud.spanner_v1.types.Field
+        """
+        with self._database.snapshot() as snapshot:
+            query = _GET_SCHEMA_TEMPLATE.format(self.table_id)
+            results = snapshot.execute_sql(query)
+            # _ = list(results)
+            return list(results.fields)

From 40a60e3b0bcd2d208a9ca3fb4945926fbe19f286 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Mon, 1 Feb 2021 17:42:59 -0600
Subject: [PATCH 02/12] update docs, add tests

---
 google/cloud/spanner_v1/database.py |  6 ++--
 google/cloud/spanner_v1/table.py    | 29 +++++++++++++++---
 tests/system/test_system.py         | 47 +++++++++++++++++++++++++++++
 3 files changed, 75 insertions(+), 7 deletions(-)

diff --git a/google/cloud/spanner_v1/database.py b/google/cloud/spanner_v1/database.py
index 902e9daa25..25a2c8ed5b 100644
--- a/google/cloud/spanner_v1/database.py
+++ b/google/cloud/spanner_v1/database.py
@@ -47,12 +47,12 @@
 )
 from google.cloud.spanner_admin_database_v1 import CreateDatabaseRequest
 from google.cloud.spanner_admin_database_v1 import UpdateDatabaseDdlRequest
-from google.cloud.spanner_v1 import ExecuteSqlRequest
-from google.cloud.spanner_v1.table import Table
-from google.cloud.spanner_v1.proto.transaction_pb2 import (
+from google.cloud.spanner_v1 import (
+    ExecuteSqlRequest,
     TransactionSelector,
     TransactionOptions,
 )
+from google.cloud.spanner_v1.table import Table
 
 # pylint: enable=ungrouped-imports
 
diff --git a/google/cloud/spanner_v1/table.py b/google/cloud/spanner_v1/table.py
index 1b80c98d87..6f15799214 100644
--- a/google/cloud/spanner_v1/table.py
+++ b/google/cloud/spanner_v1/table.py
@@ -19,18 +19,39 @@
 
 class Table(object):
     """Representation of a Cloud Spanner Table.
+
+    :type table_id: str
+    :param table_id: The ID of the table.
+
+    :type instance: :class:`~google.cloud.spanner_v1.database.Database`
+    :param instance: The database that owns the table.
     """
 
     def __init__(self, table_id, database):
-        self.table_id = table_id
+        self._table_id = table_id
         self._database = database
 
-    def get_schema(self):
+    @property
+    def table_id(self):
+        """The ID of the table used in SQL.
+
+        :rtype: str
+        :returns: The table ID.
         """
-        List of google.cloud.spanner_v1.types.Field
+        return self._table_id
+
+    def get_schema(self):
+        """Get the schema of this table.
+
+        :rtype: list of :class:`~google.cloud.spanner_v1.types.Field`
+        :returns: The table schema.
         """
         with self._database.snapshot() as snapshot:
             query = _GET_SCHEMA_TEMPLATE.format(self.table_id)
             results = snapshot.execute_sql(query)
-            # _ = list(results)
+            # Start iterating to force the schema to download.
+            try:
+                next(iter(results))
+            except StopIteration:
+                pass
             return list(results.fields)
diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index 495824044b..b22af36a6f 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -42,6 +42,7 @@
 from google.cloud.spanner_v1 import KeySet
 from google.cloud.spanner_v1.instance import Backup
 from google.cloud.spanner_v1.instance import Instance
+from google.cloud.spanner_v1.table import Table
 
 from test_utils.retry import RetryErrors
 from test_utils.retry import RetryInstanceState
@@ -465,6 +466,52 @@ def _unit_of_work(transaction, name):
         self.assertEqual(len(rows), 2)
 
 
+class TestTableAPI(unittest.TestCase, _TestData):
+    DATABASE_NAME = "test_database" + unique_resource_id("_")
+
+    @classmethod
+    def setUpClass(cls):
+        pool = BurstyPool(labels={"testcase": "database_api"})
+        ddl_statements = EMULATOR_DDL_STATEMENTS if USE_EMULATOR else DDL_STATEMENTS
+        cls._db = Config.INSTANCE.database(
+            cls.DATABASE_NAME, ddl_statements=ddl_statements, pool=pool
+        )
+        operation = cls._db.create()
+        operation.result(30)  # raises on failure / timeout.
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._db.drop()
+
+    def test_list_tables(self):
+        tables = self._db.list_tables()
+        table_ids = set(table.table_id for table in tables)
+        self.assertIn("contacts", table_ids)
+        self.assertIn("contact_phones", table_ids)
+        self.assertIn("all_types", table_ids)
+
+    def test_list_tables_get_schema(self):
+        tables = self._db.list_tables()
+        for table in tables:
+            schema = table.get_schema()
+            self.assertIsInstance(schema, list)
+
+    def test_get_schema(self):
+        table = Table("all_types", self._db)
+        schema = table.get_schema()
+        names_and_types = set((field.name, field.type_.code) for field in schema)
+        self.assertIn(("pkey", TypeCode.INT64), names_and_types)
+        self.assertIn(("int_value", TypeCode.INT64), names_and_types)
+        self.assertIn(("int_array", TypeCode.ARRAY), names_and_types)
+        self.assertIn(("bool_value", TypeCode.BOOL), names_and_types)
+        self.assertIn(("bytes_value", TypeCode.BYTES), names_and_types)
+        self.assertIn(("date_value", TypeCode.DATE), names_and_types)
+        self.assertIn(("float_value", TypeCode.FLOAT64), names_and_types)
+        self.assertIn(("string_value", TypeCode.STRING), names_and_types)
+        self.assertIn(("timestamp_value", TypeCode.TIMESTAMP), names_and_types)
+        self.assertIn(("numeric_value", TypeCode.NUMERIC), names_and_types)
+
+
 @unittest.skipIf(USE_EMULATOR, "Skipping backup tests")
 @unittest.skipIf(SKIP_BACKUP_TESTS, "Skipping backup tests")
 class TestBackupAPI(unittest.TestCase, _TestData):

From e016b0fa3442c959c574413a7ddc01cb009e4077 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Mon, 1 Feb 2021 17:46:18 -0600
Subject: [PATCH 03/12] remove numeric from get_schema test

The NUMERIC column is not included in the emulator system tests.
---
 tests/system/test_system.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index b22af36a6f..c0409b301e 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -509,7 +509,6 @@ def test_get_schema(self):
         self.assertIn(("float_value", TypeCode.FLOAT64), names_and_types)
         self.assertIn(("string_value", TypeCode.STRING), names_and_types)
         self.assertIn(("timestamp_value", TypeCode.TIMESTAMP), names_and_types)
-        self.assertIn(("numeric_value", TypeCode.NUMERIC), names_and_types)
 
 
 @unittest.skipIf(USE_EMULATOR, "Skipping backup tests")

From f4072891691b9b85ad8003b2e447a0b0f6d436a8 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Mon, 1 Feb 2021 18:29:47 -0600
Subject: [PATCH 04/12] add unit tests

---
 tests/unit/test_database.py |  8 ++++++
 tests/unit/test_table.py    | 54 +++++++++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+)
 create mode 100644 tests/unit/test_table.py

diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py
index 175c269d50..948b260a5a 100644
--- a/tests/unit/test_database.py
+++ b/tests/unit/test_database.py
@@ -1243,6 +1243,14 @@ def test_list_database_operations_explicit_filter(self):
             filter_=expected_filter_, page_size=page_size
         )
 
+    def test_list_tables(self):
+        client = _Client()
+        instance = _Instance(self.INSTANCE_NAME, client=client)
+        pool = _Pool()
+        database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+        tables = database.list_tables()
+        self.assertIsNotNone(tables)
+
 
 class TestBatchCheckout(_BaseTest):
     def _get_target_class(self):
diff --git a/tests/unit/test_table.py b/tests/unit/test_table.py
new file mode 100644
index 0000000000..ba22289f95
--- /dev/null
+++ b/tests/unit/test_table.py
@@ -0,0 +1,54 @@
+# Copyright 2021 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import mock
+
+
+class _BaseTest(unittest.TestCase):
+    TABLE_ID = "test_table"
+
+    def _make_one(self, *args, **kwargs):
+        return self._get_target_class()(*args, **kwargs)
+
+
+class TestTable(_BaseTest):
+    def _get_target_class(self):
+        from google.cloud.spanner_v1.table import Table
+
+        return Table
+
+    def test_ctor(self):
+        from google.cloud.spanner_v1.database import Database
+
+        db = mock.create_autospec(Database, instance=True)
+        table = self._make_one(self.TABLE_ID, db)
+        self.assertEqual(table.table_id, self.TABLE_ID)
+
+    def test_get_schema(self):
+        from google.cloud.spanner_v1.database import Database, SnapshotCheckout
+        from google.cloud.spanner_v1.snapshot import Snapshot
+        from google.cloud.spanner_v1.table import _GET_SCHEMA_TEMPLATE
+
+        db = mock.create_autospec(Database, instance=True)
+        checkout = mock.create_autospec(SnapshotCheckout, instance=True)
+        snapshot = mock.create_autospec(Snapshot, instance=True)
+        db.snapshot.return_value = checkout
+        checkout.__enter__.return_value = snapshot
+        table = self._make_one(self.TABLE_ID, db)
+        schema = table.get_schema()
+        self.assertIsInstance(schema, list)
+        expected_query = _GET_SCHEMA_TEMPLATE.format(self.TABLE_ID)
+        snapshot.execute_sql.assert_called_with(expected_query)

From 21395ec497785a9142845e84fe5b61b226f8cc0e Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 2 Feb 2021 09:30:27 -0600
Subject: [PATCH 05/12] add docs for table api and usage

---
 docs/api-reference.rst |  1 +
 docs/index.rst         |  1 +
 docs/table-api.rst     |  6 ++++++
 docs/table-usage.rst   | 46 ++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 54 insertions(+)
 create mode 100644 docs/table-api.rst
 create mode 100644 docs/table-usage.rst

diff --git a/docs/api-reference.rst b/docs/api-reference.rst
index 30f67cd300..41046f78bf 100644
--- a/docs/api-reference.rst
+++ b/docs/api-reference.rst
@@ -10,6 +10,7 @@ Most likely, you will be interacting almost exclusively with these:
     client-api
     instance-api
     database-api
+    table-api
     session-api
     keyset-api
     snapshot-api
diff --git a/docs/index.rst b/docs/index.rst
index cabf56157c..a4ab1b27d7 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -11,6 +11,7 @@ Usage Documentation
   client-usage
   instance-usage
   database-usage
+  table-usage
   batch-usage
   snapshot-usage
   transaction-usage
diff --git a/docs/table-api.rst b/docs/table-api.rst
new file mode 100644
index 0000000000..86b81dc86e
--- /dev/null
+++ b/docs/table-api.rst
@@ -0,0 +1,6 @@
+Table API
+=========
+
+.. automodule:: google.cloud.spanner_v1.table
+  :members:
+  :show-inheritance:
diff --git a/docs/table-usage.rst b/docs/table-usage.rst
new file mode 100644
index 0000000000..aa939fd581
--- /dev/null
+++ b/docs/table-usage.rst
@@ -0,0 +1,46 @@
+Table Admin
+===========
+
+After creating an :class:`~google.cloud.spanner_v1.database.Database`, you can
+interact with individual tables for that instance.
+
+
+List Tables
+-----------
+
+To iterate over all existing tables for an database, use its
+:meth:`~google.cloud.spanner_v1.database.Database.list_tables` method:
+
+.. code:: python
+
+    for table in database.list_tables():
+        # `table` is a `Table` object.
+
+This method yields :class:`~google.cloud.spanner_v1.table.Table` objects.
+
+
+Constructing a Table
+--------------------
+
+A table object can be created by using the
+:class:`~google.cloud.spanner_v1.table.Table` constructor. Since table
+operations are executed via SQL, a
+:class:`~google.cloud.spanner_v1.database.Database` instance is required.
+
+.. code:: python
+
+    table = google.cloud.spanner_v1.table.Table(
+       "my_table_id", database
+    )
+
+Getting the Table Schema
+------------------------
+
+Use the :meth:`~google.cloud.spanner_v1.table.Table.get_schema` method to
+inspect the columns of a table as a list of
+:class:`~google.cloud.spanner_v1.types.Field` objects.
+
+.. code:: python
+
+    for field in table.get_schema():
+        # `field` is a `Field` object.

From 2b566fac693b98bdc58f6c4c226d9db5e50ef960 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 2 Feb 2021 09:35:44 -0600
Subject: [PATCH 06/12] fix link to Field class

---
 docs/table-usage.rst             | 2 +-
 google/cloud/spanner_v1/table.py | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/docs/table-usage.rst b/docs/table-usage.rst
index aa939fd581..c72b281753 100644
--- a/docs/table-usage.rst
+++ b/docs/table-usage.rst
@@ -38,7 +38,7 @@ Getting the Table Schema
 
 Use the :meth:`~google.cloud.spanner_v1.table.Table.get_schema` method to
 inspect the columns of a table as a list of
-:class:`~google.cloud.spanner_v1.types.Field` objects.
+:class:`~google.cloud.spanner_v1.types.StructType.Field` objects.
 
 .. code:: python
 
diff --git a/google/cloud/spanner_v1/table.py b/google/cloud/spanner_v1/table.py
index 6f15799214..11ebf45f59 100644
--- a/google/cloud/spanner_v1/table.py
+++ b/google/cloud/spanner_v1/table.py
@@ -43,7 +43,7 @@ def table_id(self):
     def get_schema(self):
         """Get the schema of this table.
 
-        :rtype: list of :class:`~google.cloud.spanner_v1.types.Field`
+        :rtype: list of :class:`~google.cloud.spanner_v1.types.StructType.Field`
         :returns: The table schema.
         """
         with self._database.snapshot() as snapshot:

From 66aa19277dabf7ebfca033e441bd6bc30da5b563 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 2 Feb 2021 09:38:12 -0600
Subject: [PATCH 07/12] typo in table constructor docs

---
 google/cloud/spanner_v1/table.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/google/cloud/spanner_v1/table.py b/google/cloud/spanner_v1/table.py
index 11ebf45f59..d559ebff00 100644
--- a/google/cloud/spanner_v1/table.py
+++ b/google/cloud/spanner_v1/table.py
@@ -23,8 +23,8 @@ class Table(object):
     :type table_id: str
     :param table_id: The ID of the table.
 
-    :type instance: :class:`~google.cloud.spanner_v1.database.Database`
-    :param instance: The database that owns the table.
+    :type database: :class:`~google.cloud.spanner_v1.database.Database`
+    :param database: The database that owns the table.
     """
 
     def __init__(self, table_id, database):

From bfbc105272dbeb0a4bcf4537c1453c836859e161 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 2 Feb 2021 11:47:57 -0600
Subject: [PATCH 08/12] add reload and exists methods

---
 docs/table-usage.rst             | 10 ++--
 google/cloud/spanner_v1/table.py | 87 ++++++++++++++++++++++++++++----
 tests/system/test_system.py      | 22 ++++++--
 tests/unit/test_table.py         | 74 ++++++++++++++++++++++++++-
 4 files changed, 173 insertions(+), 20 deletions(-)

diff --git a/docs/table-usage.rst b/docs/table-usage.rst
index c72b281753..48b455ce9f 100644
--- a/docs/table-usage.rst
+++ b/docs/table-usage.rst
@@ -19,8 +19,8 @@ To iterate over all existing tables for an database, use its
 This method yields :class:`~google.cloud.spanner_v1.table.Table` objects.
 
 
-Constructing a Table
---------------------
+Table Factory
+-------------
 
 A table object can be created by using the
 :class:`~google.cloud.spanner_v1.table.Table` constructor. Since table
@@ -36,11 +36,11 @@ operations are executed via SQL, a
 Getting the Table Schema
 ------------------------
 
-Use the :meth:`~google.cloud.spanner_v1.table.Table.get_schema` method to
-inspect the columns of a table as a list of
+Use the :attr:`~google.cloud.spanner_v1.table.Table.schema` property to inspect
+the columns of a table as a list of
 :class:`~google.cloud.spanner_v1.types.StructType.Field` objects.
 
 .. code:: python
 
-    for field in table.get_schema():
+    for field in table.schema
         # `field` is a `Field` object.
diff --git a/google/cloud/spanner_v1/table.py b/google/cloud/spanner_v1/table.py
index d559ebff00..4a31446509 100644
--- a/google/cloud/spanner_v1/table.py
+++ b/google/cloud/spanner_v1/table.py
@@ -14,6 +14,21 @@
 
 """User friendly container for Cloud Spanner Table."""
 
+from google.cloud.exceptions import NotFound
+
+from google.cloud.spanner_v1.types import (
+    Type,
+    TypeCode,
+)
+
+
+_EXISTS_TEMPLATE = """
+SELECT EXISTS(
+    SELECT TABLE_NAME
+    FROM INFORMATION_SCHEMA.TABLES
+    WHERE TABLE_NAME = @table_id
+)
+"""
 _GET_SCHEMA_TEMPLATE = "SELECT * FROM {} LIMIT 0"
 
 
@@ -31,6 +46,9 @@ def __init__(self, table_id, database):
         self._table_id = table_id
         self._database = database
 
+        # Calculated properties.
+        self._schema = None
+
     @property
     def table_id(self):
         """The ID of the table used in SQL.
@@ -40,18 +58,69 @@ def table_id(self):
         """
         return self._table_id
 
-    def get_schema(self):
+    def exists(self):
+        """Test whether this table exists.
+
+        :rtype: bool
+        :returns: True if the table exists, else false.
+        """
+        with self._database.snapshot() as snapshot:
+            return self._exists(snapshot)
+
+    def _exists(self, snapshot):
+        """Query to check that the table exists.
+
+        :type snapshot: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`
+        :param snapshot: snapshot to use for database queries
+
+        :rtype: bool
+        :returns: True if the table exists, else false.
+        """
+        results = snapshot.execute_sql(
+            _EXISTS_TEMPLATE,
+            params={"table_id": self.table_id},
+            param_types={"table_id": Type(code=TypeCode.STRING)},
+        )
+        return next(iter(results))[0]
+
+    @property
+    def schema(self):
+        """The schema of this table.
+
+        :rtype: list of :class:`~google.cloud.spanner_v1.types.StructType.Field`
+        :returns: The table schema.
+        """
+        if self._schema is None:
+            with self._database.snapshot() as snapshot:
+                self._schema = self._get_schema(snapshot)
+        return self._schema
+
+    def _get_schema(self, snapshot):
         """Get the schema of this table.
 
+        :type snapshot: :class:`~google.cloud.spanner_v1.snapshot.Snapshot`
+        :param snapshot: snapshot to use for database queries
+
         :rtype: list of :class:`~google.cloud.spanner_v1.types.StructType.Field`
         :returns: The table schema.
         """
+        query = _GET_SCHEMA_TEMPLATE.format(self.table_id)
+        results = snapshot.execute_sql(query)
+        # Start iterating to force the schema to download.
+        try:
+            next(iter(results))
+        except StopIteration:
+            pass
+        return list(results.fields)
+
+    def reload(self):
+        """Reload this table.
+
+        Refresh any configured schema into :attr:`schema`.
+
+        :raises NotFound: if the table does not exist
+        """
         with self._database.snapshot() as snapshot:
-            query = _GET_SCHEMA_TEMPLATE.format(self.table_id)
-            results = snapshot.execute_sql(query)
-            # Start iterating to force the schema to download.
-            try:
-                next(iter(results))
-            except StopIteration:
-                pass
-            return list(results.fields)
+            if not self._exists(snapshot):
+                raise NotFound("table '{}' does not exist".format(self.table_id))
+            self._schema = self._get_schema(snapshot)
diff --git a/tests/system/test_system.py b/tests/system/test_system.py
index c0409b301e..45585c043e 100644
--- a/tests/system/test_system.py
+++ b/tests/system/test_system.py
@@ -483,6 +483,14 @@ def setUpClass(cls):
     def tearDownClass(cls):
         cls._db.drop()
 
+    def test_exists(self):
+        table = Table("all_types", self._db)
+        self.assertTrue(table.exists())
+
+    def test_exists_not_found(self):
+        table = Table("table_does_not_exist", self._db)
+        self.assertFalse(table.exists())
+
     def test_list_tables(self):
         tables = self._db.list_tables()
         table_ids = set(table.table_id for table in tables)
@@ -490,15 +498,21 @@ def test_list_tables(self):
         self.assertIn("contact_phones", table_ids)
         self.assertIn("all_types", table_ids)
 
-    def test_list_tables_get_schema(self):
+    def test_list_tables_reload(self):
         tables = self._db.list_tables()
         for table in tables:
-            schema = table.get_schema()
+            self.assertTrue(table.exists())
+            schema = table.schema
             self.assertIsInstance(schema, list)
 
-    def test_get_schema(self):
+    def test_reload_not_found(self):
+        table = Table("table_does_not_exist", self._db)
+        with self.assertRaises(exceptions.NotFound):
+            table.reload()
+
+    def test_schema(self):
         table = Table("all_types", self._db)
-        schema = table.get_schema()
+        schema = table.schema
         names_and_types = set((field.name, field.type_.code) for field in schema)
         self.assertIn(("pkey", TypeCode.INT64), names_and_types)
         self.assertIn(("int_value", TypeCode.INT64), names_and_types)
diff --git a/tests/unit/test_table.py b/tests/unit/test_table.py
index ba22289f95..0a49a9b225 100644
--- a/tests/unit/test_table.py
+++ b/tests/unit/test_table.py
@@ -14,8 +14,15 @@
 
 import unittest
 
+from google.cloud.exceptions import NotFound
 import mock
 
+from google.cloud.spanner_v1.types import (
+    StructType,
+    Type,
+    TypeCode,
+)
+
 
 class _BaseTest(unittest.TestCase):
     TABLE_ID = "test_table"
@@ -37,7 +44,27 @@ def test_ctor(self):
         table = self._make_one(self.TABLE_ID, db)
         self.assertEqual(table.table_id, self.TABLE_ID)
 
-    def test_get_schema(self):
+    def test_exists_executes_query(self):
+        from google.cloud.spanner_v1.database import Database, SnapshotCheckout
+        from google.cloud.spanner_v1.snapshot import Snapshot
+        from google.cloud.spanner_v1.table import _EXISTS_TEMPLATE
+
+        db = mock.create_autospec(Database, instance=True)
+        checkout = mock.create_autospec(SnapshotCheckout, instance=True)
+        snapshot = mock.create_autospec(Snapshot, instance=True)
+        db.snapshot.return_value = checkout
+        checkout.__enter__.return_value = snapshot
+        snapshot.execute_sql.return_value = [[False]]
+        table = self._make_one(self.TABLE_ID, db)
+        exists = table.exists()
+        self.assertFalse(exists)
+        snapshot.execute_sql.assert_called_with(
+            _EXISTS_TEMPLATE,
+            params={"table_id": self.TABLE_ID},
+            param_types={"table_id": Type(code=TypeCode.STRING)},
+        )
+
+    def test_schema_executes_query(self):
         from google.cloud.spanner_v1.database import Database, SnapshotCheckout
         from google.cloud.spanner_v1.snapshot import Snapshot
         from google.cloud.spanner_v1.table import _GET_SCHEMA_TEMPLATE
@@ -48,7 +75,50 @@ def test_get_schema(self):
         db.snapshot.return_value = checkout
         checkout.__enter__.return_value = snapshot
         table = self._make_one(self.TABLE_ID, db)
-        schema = table.get_schema()
+        schema = table.schema
         self.assertIsInstance(schema, list)
         expected_query = _GET_SCHEMA_TEMPLATE.format(self.TABLE_ID)
         snapshot.execute_sql.assert_called_with(expected_query)
+
+    def test_schema_returns_cache(self):
+        from google.cloud.spanner_v1.database import Database
+
+        db = mock.create_autospec(Database, instance=True)
+        table = self._make_one(self.TABLE_ID, db)
+        table._schema = [StructType.Field(name="col1")]
+        schema = table.schema
+        self.assertEqual(schema, [StructType.Field(name="col1")])
+
+    def test_reload_raises_notfound(self):
+        from google.cloud.spanner_v1.database import Database, SnapshotCheckout
+        from google.cloud.spanner_v1.snapshot import Snapshot
+
+        db = mock.create_autospec(Database, instance=True)
+        checkout = mock.create_autospec(SnapshotCheckout, instance=True)
+        snapshot = mock.create_autospec(Snapshot, instance=True)
+        db.snapshot.return_value = checkout
+        checkout.__enter__.return_value = snapshot
+        snapshot.execute_sql.return_value = [[False]]
+        table = self._make_one(self.TABLE_ID, db)
+        with self.assertRaises(NotFound):
+            table.reload()
+
+    def test_reload_executes_queries(self):
+        from google.cloud.spanner_v1.database import Database, SnapshotCheckout
+        from google.cloud.spanner_v1.snapshot import Snapshot
+        from google.cloud.spanner_v1.streamed import StreamedResultSet
+
+        db = mock.create_autospec(Database, instance=True)
+        checkout = mock.create_autospec(SnapshotCheckout, instance=True)
+        snapshot = mock.create_autospec(Snapshot, instance=True)
+        results = mock.create_autospec(StreamedResultSet, instance=True)
+        db.snapshot.return_value = checkout
+        checkout.__enter__.return_value = snapshot
+        results.fields = [StructType.Field(name="col1")]
+        snapshot.execute_sql.side_effect = [
+            [[True]],
+            results,
+        ]
+        table = self._make_one(self.TABLE_ID, db)
+        table.reload()
+        self.assertEqual(table.schema, [StructType.Field(name="col1")])

From 1250d8ace73ca57995bfea80839a703d23d22656 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Wed, 17 Feb 2021 09:18:07 -0600
Subject: [PATCH 09/12] feat: add table method to database

---
 google/cloud/spanner_v1/database.py | 24 +++++++++++++++++++++++-
 google/cloud/spanner_v1/instance.py |  2 +-
 tests/unit/test_database.py         | 12 ++++++++++++
 3 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/google/cloud/spanner_v1/database.py b/google/cloud/spanner_v1/database.py
index 25a2c8ed5b..c37416af17 100644
--- a/google/cloud/spanner_v1/database.py
+++ b/google/cloud/spanner_v1/database.py
@@ -602,6 +602,28 @@ def list_database_operations(self, filter_="", page_size=None):
             filter_=database_filter, page_size=page_size
         )
 
+    def table(self, table_id):
+        """Factory to create a table object within this database.
+
+        Note: This method does not create a table in Cloud Spanner, but it can
+        be used to check if a table exists.
+
+        .. code-block:: python
+
+           my_table = database.table("my_table")
+           if my_table.exists():
+               print("Table with ID 'my_table' exists.")
+           else:
+               print("Table with ID 'my_table' does not exist.")
+
+        :type table_id: str
+        :param table_id: The ID of the table.
+
+        :rtype: :class:`~google.cloud.spanner_v1.table.Table`
+        :returns: a table owned by this database.
+        """
+        return Table(table_id, self)
+
     def list_tables(self):
         """List tables within the database.
 
@@ -613,7 +635,7 @@ def list_tables(self):
         with self.snapshot() as snapshot:
             results = snapshot.execute_sql(_LIST_TABLES_QUERY)
             for row in results:
-                yield Table(row[0], self)
+                yield self.table(row[0])
 
 
 class BatchCheckout(object):
diff --git a/google/cloud/spanner_v1/instance.py b/google/cloud/spanner_v1/instance.py
index b422c57afd..67208d430a 100644
--- a/google/cloud/spanner_v1/instance.py
+++ b/google/cloud/spanner_v1/instance.py
@@ -361,7 +361,7 @@ def database(self, database_id, ddl_statements=(), pool=None):
         """Factory to create a database within this instance.
 
         :type database_id: str
-        :param database_id: The ID of the instance.
+        :param database_id: The ID of the database.
 
         :type ddl_statements: list of string
         :param ddl_statements: (Optional) DDL statements, excluding the
diff --git a/tests/unit/test_database.py b/tests/unit/test_database.py
index 948b260a5a..d1b9bb4394 100644
--- a/tests/unit/test_database.py
+++ b/tests/unit/test_database.py
@@ -1243,6 +1243,18 @@ def test_list_database_operations_explicit_filter(self):
             filter_=expected_filter_, page_size=page_size
         )
 
+    def test_table_factory_defaults(self):
+        from google.cloud.spanner_v1.table import Table
+
+        client = _Client()
+        instance = _Instance(self.INSTANCE_NAME, client=client)
+        pool = _Pool()
+        database = self._make_one(self.DATABASE_ID, instance, pool=pool)
+        my_table = database.table("my_table")
+        self.assertIsInstance(my_table, Table)
+        self.assertIs(my_table._database, database)
+        self.assertEqual(my_table.table_id, "my_table")
+
     def test_list_tables(self):
         client = _Client()
         instance = _Instance(self.INSTANCE_NAME, client=client)

From 3f6eb3476c00683749bc58c04a04013be62f981b Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Wed, 17 Feb 2021 09:24:14 -0600
Subject: [PATCH 10/12] update usage docs to use factory method

---
 docs/table-usage.rst | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/docs/table-usage.rst b/docs/table-usage.rst
index 48b455ce9f..5754e4d858 100644
--- a/docs/table-usage.rst
+++ b/docs/table-usage.rst
@@ -22,16 +22,16 @@ This method yields :class:`~google.cloud.spanner_v1.table.Table` objects.
 Table Factory
 -------------
 
-A table object can be created by using the
-:class:`~google.cloud.spanner_v1.table.Table` constructor. Since table
-operations are executed via SQL, a
-:class:`~google.cloud.spanner_v1.database.Database` instance is required.
+A :class:`~google.cloud.spanner_v1.table.Table` object can be created with the
+:meth:`~google.cloud.spanner_v1.database.Database.table` factory method:
 
 .. code:: python
 
-    table = google.cloud.spanner_v1.table.Table(
-       "my_table_id", database
-    )
+    table = database.table("my_table_id")
+    if my_table.exists():
+        print("Table with ID 'my_table' exists.")
+    else:
+        print("Table with ID 'my_table' does not exist."
 
 Getting the Table Schema
 ------------------------

From 611b27fb881f33ea0332342de6ec14615cfeae50 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Mon, 22 Feb 2021 15:23:30 -0600
Subject: [PATCH 11/12] address warning in GitHub UI for sphinx header

---
 docs/table-usage.rst | 1 +
 1 file changed, 1 insertion(+)

diff --git a/docs/table-usage.rst b/docs/table-usage.rst
index 5754e4d858..315eaf63b4 100644
--- a/docs/table-usage.rst
+++ b/docs/table-usage.rst
@@ -33,6 +33,7 @@ A :class:`~google.cloud.spanner_v1.table.Table` object can be created with the
     else:
         print("Table with ID 'my_table' does not exist."
 
+
 Getting the Table Schema
 ------------------------
 

From d91e666741adcb6bb27fc98140d798d0ac64fa00 Mon Sep 17 00:00:00 2001
From: larkee <31196561+larkee@users.noreply.github.com>
Date: Fri, 26 Feb 2021 16:34:54 +1100
Subject: [PATCH 12/12] Update docs/table-usage.rst

---
 docs/table-usage.rst | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/table-usage.rst b/docs/table-usage.rst
index 315eaf63b4..9d28da1ebb 100644
--- a/docs/table-usage.rst
+++ b/docs/table-usage.rst
@@ -28,7 +28,7 @@ A :class:`~google.cloud.spanner_v1.table.Table` object can be created with the
 .. code:: python
 
     table = database.table("my_table_id")
-    if my_table.exists():
+    if table.exists():
         print("Table with ID 'my_table' exists.")
     else:
         print("Table with ID 'my_table' does not exist."