From 868133580d06ea227c574aa6f1b91e12778e9689 Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 26 Mar 2019 11:07:08 -0700
Subject: [PATCH 1/4] Allow string for copy sources, query destination, and
 default dataset

This removes a few more cases where it was required to create a
TableReference or DatasetReference.

* Allow a string for destination in QueryJobConfig .
* Allow a string for default_dataset in QueryJobConfig .
* Allow a list of strings for sources in Client.copy_table.
---
 bigquery/google/cloud/bigquery/client.py | 65 +++++++++---------------
 bigquery/google/cloud/bigquery/job.py    | 36 ++++++++++---
 bigquery/google/cloud/bigquery/table.py  |  9 ++++
 bigquery/tests/unit/test_client.py       | 27 ++++++++++
 bigquery/tests/unit/test_job.py          | 34 +++++++++++++
 5 files changed, 125 insertions(+), 46 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/client.py b/bigquery/google/cloud/bigquery/client.py
index 7fe412478bfa..b93eca1c75ff 100644
--- a/bigquery/google/cloud/bigquery/client.py
+++ b/bigquery/google/cloud/bigquery/client.py
@@ -48,6 +48,7 @@
 from google.cloud.bigquery import job
 from google.cloud.bigquery.query import _QueryResults
 from google.cloud.bigquery.retry import DEFAULT_RETRY
+from google.cloud.bigquery.table import _str_or_ref_to_table_ref
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import TableListItem
 from google.cloud.bigquery.table import TableReference
@@ -379,8 +380,8 @@ def create_table(self, table, exists_ok=False, retry=DEFAULT_RETRY):
             google.cloud.bigquery.table.Table:
                 A new ``Table`` returned from the service.
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
+
         if isinstance(table, TableReference):
             table = Table(table)
 
@@ -446,11 +447,7 @@ def get_table(self, table_ref, retry=DEFAULT_RETRY):
             google.cloud.bigquery.table.Table:
                 A ``Table`` instance.
         """
-        if isinstance(table_ref, str):
-            table_ref = TableReference.from_string(
-                table_ref, default_project=self.project
-            )
-
+        table_ref = _str_or_ref_to_table_ref(table_ref, default_project=self.project)
         api_response = self._call_api(retry, method="GET", path=table_ref.path)
         return Table.from_api_repr(api_response)
 
@@ -653,8 +650,7 @@ def delete_table(self, table, retry=DEFAULT_RETRY, not_found_ok=False):
                 Defaults to ``False``. If ``True``, ignore "not found" errors
                 when deleting the table.
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
 
         if not isinstance(table, (Table, TableReference)):
             raise TypeError("table must be a Table or a TableReference")
@@ -966,11 +962,9 @@ def load_table_from_uri(
         if isinstance(source_uris, six.string_types):
             source_uris = [source_uris]
 
-        if isinstance(destination, str):
-            destination = TableReference.from_string(
-                destination, default_project=self.project
-            )
-
+        destination = _str_or_ref_to_table_ref(
+            destination, default_project=self.project
+        )
         load_job = job.LoadJob(job_ref, source_uris, destination, self, job_config)
         load_job._begin(retry=retry)
 
@@ -1045,11 +1039,9 @@ def load_table_from_file(
         if location is None:
             location = self.location
 
-        if isinstance(destination, str):
-            destination = TableReference.from_string(
-                destination, default_project=self.project
-            )
-
+        destination = _str_or_ref_to_table_ref(
+            destination, default_project=self.project
+        )
         job_ref = job._JobReference(job_id, project=project, location=location)
         load_job = job.LoadJob(job_ref, None, destination, self, job_config)
         job_resource = load_job.to_api_repr()
@@ -1288,6 +1280,7 @@ def copy_table(
             sources (Union[ \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
+                Sequence[str], \
                 Sequence[ \
                     :class:`~google.cloud.bigquery.table.TableReference`], \
             ]):
@@ -1327,18 +1320,18 @@ def copy_table(
             location = self.location
 
         job_ref = job._JobReference(job_id, project=project, location=location)
-
-        if isinstance(sources, str):
-            sources = TableReference.from_string(sources, default_project=self.project)
-
-        if isinstance(destination, str):
-            destination = TableReference.from_string(
-                destination, default_project=self.project
-            )
+        sources = _str_or_ref_to_table_ref(sources, default_project=self.project)
+        destination = _str_or_ref_to_table_ref(
+            destination, default_project=self.project
+        )
 
         if not isinstance(sources, collections_abc.Sequence):
             sources = [sources]
 
+        sources = [
+            _str_or_ref_to_table_ref(source, default_project=self.project)
+            for source in sources
+        ]
         copy_job = job.CopyJob(
             job_ref, sources, destination, client=self, job_config=job_config
         )
@@ -1405,9 +1398,7 @@ def extract_table(
             location = self.location
 
         job_ref = job._JobReference(job_id, project=project, location=location)
-
-        if isinstance(source, str):
-            source = TableReference.from_string(source, default_project=self.project)
+        source = _str_or_ref_to_table_ref(source, default_project=self.project)
 
         if isinstance(destination_uris, six.string_types):
             destination_uris = [destination_uris]
@@ -1529,8 +1520,7 @@ def insert_rows(self, table, rows, selected_fields=None, **kwargs):
         Raises:
             ValueError: if table's schema is not set
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
 
         if selected_fields is not None:
             schema = selected_fields
@@ -1596,9 +1586,7 @@ def insert_rows_json(
                 identifies the row, and the "errors" key contains a list of
                 the mappings describing one or more problems with the row.
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
-
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
         rows_info = []
         data = {"rows": rows_info}
 
@@ -1647,9 +1635,7 @@ def list_partitions(self, table, retry=DEFAULT_RETRY):
             List[str]:
                 A list of the partition ids present in the partitioned table
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
-
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
         meta_table = self.get_table(
             TableReference(
                 self.dataset(table.dataset_id, project=table.project),
@@ -1724,8 +1710,7 @@ def list_rows(
                 (this is distinct from the total number of rows in the
                 current page: ``iterator.page.num_items``).
         """
-        if isinstance(table, str):
-            table = TableReference.from_string(table, default_project=self.project)
+        table = _str_or_ref_to_table_ref(table, default_project=self.project)
 
         if selected_fields is not None:
             schema = selected_fields
diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 4df7a92ba63c..2cb8e2ee2290 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -17,6 +17,7 @@
 import copy
 import threading
 
+import six
 from six.moves import http_client
 
 import google.api_core.future.polling
@@ -33,6 +34,7 @@
 from google.cloud.bigquery.schema import SchemaField
 from google.cloud.bigquery.table import _EmptyRowIterator
 from google.cloud.bigquery.table import EncryptionConfiguration
+from google.cloud.bigquery.table import _str_or_ref_to_table_ref
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import TimePartitioning
@@ -1999,6 +2001,13 @@ def default_dataset(self):
         to use for unqualified table names in the query or :data:`None` if not
         set.
 
+        The ``dataset`` setter accepts:
+
+        - a :class:`~google.cloud.bigquery.dataset.DatasetReference`, or
+        - a :class:`str` of the fully-qualified dataset ID in standard SQL
+          format. The value must included a project ID and dataset ID
+          separated by ``.``.
+
         See
         https://g.co/cloud/bigquery/docs/reference/v2/jobs#configuration.query.defaultDataset
         """
@@ -2009,9 +2018,14 @@ def default_dataset(self):
 
     @default_dataset.setter
     def default_dataset(self, value):
-        resource = None
-        if value is not None:
-            resource = value.to_api_repr()
+        if value is None:
+            self._set_sub_prop("defaultDataset", None)
+            return
+
+        if isinstance(value, six.string_types):
+            value = DatasetReference.from_string(value)
+
+        resource = value.to_api_repr()
         self._set_sub_prop("defaultDataset", resource)
 
     @property
@@ -2019,6 +2033,13 @@ def destination(self):
         """google.cloud.bigquery.table.TableReference: table where results are
         written or :data:`None` if not set.
 
+        The ``destination`` setter accepts:
+
+        - a :class:`~google.cloud.bigquery.table.TableReference`, or
+        - a :class:`str` of the fully-qualified table ID in standard SQL
+          format. The value must included a project ID, dataset ID, and table
+          ID, each separated by ``.``.
+
         See
         https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.destinationTable
         """
@@ -2029,9 +2050,12 @@ def destination(self):
 
     @destination.setter
     def destination(self, value):
-        resource = None
-        if value is not None:
-            resource = value.to_api_repr()
+        if value is None:
+            self._set_sub_prop("destinationTable", None)
+            return
+
+        value = _str_or_ref_to_table_ref(value)
+        resource = value.to_api_repr()
         self._set_sub_prop("destinationTable", resource)
 
     @property
diff --git a/bigquery/google/cloud/bigquery/table.py b/bigquery/google/cloud/bigquery/table.py
index 35f950e7d94a..12d5caafffb0 100644
--- a/bigquery/google/cloud/bigquery/table.py
+++ b/bigquery/google/cloud/bigquery/table.py
@@ -1661,3 +1661,12 @@ def _rows_page_start(iterator, page, response):
 
 
 # pylint: enable=unused-argument
+
+
+def _str_or_ref_to_table_ref(value, default_project=None):
+    """Helper to convert a string to TableReference or keep a
+    TableReference unchanged.
+    """
+    if isinstance(value, six.string_types):
+        value = TableReference.from_string(value, default_project=default_project)
+    return value
diff --git a/bigquery/tests/unit/test_client.py b/bigquery/tests/unit/test_client.py
index 794b76a0a9f4..4ffd589d8a49 100644
--- a/bigquery/tests/unit/test_client.py
+++ b/bigquery/tests/unit/test_client.py
@@ -2820,6 +2820,33 @@ def test_copy_table_w_client_location(self):
             method="POST", path="/projects/other-project/jobs", data=resource
         )
 
+    def test_copy_table_w_source_strings(self):
+        creds = _make_credentials()
+        http = object()
+        client = self._make_one(project=self.PROJECT, credentials=creds, _http=http)
+        client._connection = _make_connection({})
+        sources = [
+            "dataset_wo_proj.some_table",
+            "other_project.other_dataset.other_table",
+            client.dataset("dataset_from_ref").table("table_from_ref"),
+        ]
+        destination = "some_project.some_dataset.destination_table"
+
+        job = client.copy_table(sources, destination)
+
+        expected_sources = [
+            client.dataset("dataset_wo_proj").table("some_table"),
+            client.dataset("other_dataset", project="other_project").table(
+                "other_table"
+            ),
+            client.dataset("dataset_from_ref").table("table_from_ref"),
+        ]
+        self.assertEqual(list(job.sources), expected_sources)
+        expected_destination = client.dataset(
+            "some_dataset", project="some_project"
+        ).table("destination_table")
+        self.assertEqual(job.destination, expected_destination)
+
     def test_extract_table(self):
         from google.cloud.bigquery.job import ExtractJob
 
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index 5023c41e6374..6bc73e82fb6d 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -3163,6 +3163,40 @@ def test_ctor_w_properties(self):
         self.assertFalse(config.use_query_cache)
         self.assertTrue(config.use_legacy_sql)
 
+    def test_ctor_w_string_default_dataset(self):
+        from google.cloud.bigquery import dataset
+
+        default_dataset = "default-proj.default_dset"
+        config = self._get_target_class()(default_dataset=default_dataset)
+        expected = dataset.DatasetReference.from_string(default_dataset)
+        self.assertEqual(config.default_dataset, expected)
+
+    def test_ctor_w_string_destinaton(self):
+        from google.cloud.bigquery import table
+
+        destination = "dest-proj.dest_dset.dest_tbl"
+        config = self._get_target_class()(destination=destination)
+        expected = table.TableReference.from_string(destination)
+        self.assertEqual(config.destination, expected)
+
+    def test_default_dataset_w_string(self):
+        from google.cloud.bigquery import dataset
+
+        default_dataset = "default-proj.default_dset"
+        config = self._make_one()
+        config.default_dataset = default_dataset
+        expected = dataset.DatasetReference.from_string(default_dataset)
+        self.assertEqual(config.default_dataset, expected)
+
+    def test_destinaton_w_string(self):
+        from google.cloud.bigquery import table
+
+        destination = "dest-proj.dest_dset.dest_tbl"
+        config = self._make_one()
+        config.destination = destination
+        expected = table.TableReference.from_string(destination)
+        self.assertEqual(config.destination, expected)
+
     def test_time_partitioning(self):
         from google.cloud.bigquery import table
 

From 3ab95d4ad7da8b2be27171716e294f024882536d Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Tue, 26 Mar 2019 14:46:47 -0700
Subject: [PATCH 2/4] Refactor str/table/ref to ref helper

Rename the helper to `_table_arg_to_table_ref` and always return a
TableReference when something sensible is passed in.

Add a second helper `_table_arg_to_table` that always returns a `Table`,
which is needed in `list_rows` and `insert_rows` because only a `Table`
contains a schema.
---
 bigquery/google/cloud/bigquery/client.py | 110 ++++++++++++++---------
 bigquery/google/cloud/bigquery/job.py    |  16 +++-
 bigquery/google/cloud/bigquery/table.py  |  23 ++++-
 bigquery/tests/unit/test_client.py       |   2 +-
 bigquery/tests/unit/test_job.py          |   9 ++
 5 files changed, 112 insertions(+), 48 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/client.py b/bigquery/google/cloud/bigquery/client.py
index b93eca1c75ff..3f79e948a5a8 100644
--- a/bigquery/google/cloud/bigquery/client.py
+++ b/bigquery/google/cloud/bigquery/client.py
@@ -48,7 +48,8 @@
 from google.cloud.bigquery import job
 from google.cloud.bigquery.query import _QueryResults
 from google.cloud.bigquery.retry import DEFAULT_RETRY
-from google.cloud.bigquery.table import _str_or_ref_to_table_ref
+from google.cloud.bigquery.table import _table_arg_to_table
+from google.cloud.bigquery.table import _table_arg_to_table_ref
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import TableListItem
 from google.cloud.bigquery.table import TableReference
@@ -69,6 +70,9 @@
 _READ_LESS_THAN_SIZE = (
     "Size {:d} was specified but the file-like object only had " "{:d} bytes remaining."
 )
+_NEED_TABLE_ARGUMENT = (
+    "The table argument should be a table ID string, Table, or TableReference"
+)
 
 
 class Project(object):
@@ -380,10 +384,7 @@ def create_table(self, table, exists_ok=False, retry=DEFAULT_RETRY):
             google.cloud.bigquery.table.Table:
                 A new ``Table`` returned from the service.
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
-
-        if isinstance(table, TableReference):
-            table = Table(table)
+        table = _table_arg_to_table(table, default_project=self.project)
 
         path = "/projects/%s/datasets/%s/tables" % (table.project, table.dataset_id)
         data = table.to_api_repr()
@@ -433,6 +434,7 @@ def get_table(self, table_ref, retry=DEFAULT_RETRY):
 
         Args:
             table_ref (Union[ \
+                :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
             ]):
@@ -447,7 +449,7 @@ def get_table(self, table_ref, retry=DEFAULT_RETRY):
             google.cloud.bigquery.table.Table:
                 A ``Table`` instance.
         """
-        table_ref = _str_or_ref_to_table_ref(table_ref, default_project=self.project)
+        table_ref = _table_arg_to_table_ref(table_ref, default_project=self.project)
         api_response = self._call_api(retry, method="GET", path=table_ref.path)
         return Table.from_api_repr(api_response)
 
@@ -650,10 +652,9 @@ def delete_table(self, table, retry=DEFAULT_RETRY, not_found_ok=False):
                 Defaults to ``False``. If ``True``, ignore "not found" errors
                 when deleting the table.
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
-
-        if not isinstance(table, (Table, TableReference)):
-            raise TypeError("table must be a Table or a TableReference")
+        table = _table_arg_to_table_ref(table, default_project=self.project)
+        if not isinstance(table, TableReference):
+            raise TypeError("Unable to get TableReference for table '{}'".format(table))
 
         try:
             self._call_api(retry, method="DELETE", path=table.path)
@@ -921,6 +922,7 @@ def load_table_from_uri(
                 URIs of data files to be loaded; in format
                 ``gs://<bucket_name>/<object_name_or_glob>``.
             destination (Union[ \
+                :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
             ]):
@@ -962,7 +964,7 @@ def load_table_from_uri(
         if isinstance(source_uris, six.string_types):
             source_uris = [source_uris]
 
-        destination = _str_or_ref_to_table_ref(
+        destination = _table_arg_to_table_ref(
             destination, default_project=self.project
         )
         load_job = job.LoadJob(job_ref, source_uris, destination, self, job_config)
@@ -991,6 +993,7 @@ def load_table_from_file(
         Arguments:
             file_obj (file): A file handle opened in binary mode for reading.
             destination (Union[ \
+                :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
             ]):
@@ -1039,7 +1042,7 @@ def load_table_from_file(
         if location is None:
             location = self.location
 
-        destination = _str_or_ref_to_table_ref(
+        destination = _table_arg_to_table_ref(
             destination, default_project=self.project
         )
         job_ref = job._JobReference(job_id, project=project, location=location)
@@ -1278,14 +1281,20 @@ def copy_table(
 
         Arguments:
             sources (Union[ \
+                :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
-                Sequence[str], \
                 Sequence[ \
-                    :class:`~google.cloud.bigquery.table.TableReference`], \
+                    Union[ \
+                        :class:`~google.cloud.bigquery.table.Table`, \
+                        :class:`~google.cloud.bigquery.table.TableReference`, \
+                        str, \
+                    ] \
+                ], \
             ]):
                 Table or tables to be copied.
             destination (Union[
+                :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
             ]):
@@ -1320,18 +1329,26 @@ def copy_table(
             location = self.location
 
         job_ref = job._JobReference(job_id, project=project, location=location)
-        sources = _str_or_ref_to_table_ref(sources, default_project=self.project)
-        destination = _str_or_ref_to_table_ref(
-            destination, default_project=self.project
-        )
+
+        # sources can be one of many different input types. (string, Table,
+        # TableReference, or a sequence of any of those.) Convert them all to a
+        # list of TableReferences.
+        #
+        # _table_arg_to_table_ref leaves lists unmodified.
+        sources = _table_arg_to_table_ref(sources, default_project=self.project)
 
         if not isinstance(sources, collections_abc.Sequence):
             sources = [sources]
 
         sources = [
-            _str_or_ref_to_table_ref(source, default_project=self.project)
+            _table_arg_to_table_ref(source, default_project=self.project)
             for source in sources
         ]
+
+        destination = _table_arg_to_table_ref(
+            destination, default_project=self.project
+        )
+
         copy_job = job.CopyJob(
             job_ref, sources, destination, client=self, job_config=job_config
         )
@@ -1357,6 +1374,7 @@ def extract_table(
 
         Arguments:
             source (Union[ \
+                :class:`google.cloud.bigquery.table.Table`, \
                 :class:`google.cloud.bigquery.table.TableReference`, \
                 src, \
             ]):
@@ -1398,7 +1416,7 @@ def extract_table(
             location = self.location
 
         job_ref = job._JobReference(job_id, project=project, location=location)
-        source = _str_or_ref_to_table_ref(source, default_project=self.project)
+        source = _table_arg_to_table_ref(source, default_project=self.project)
 
         if isinstance(destination_uris, six.string_types):
             destination_uris = [destination_uris]
@@ -1520,18 +1538,22 @@ def insert_rows(self, table, rows, selected_fields=None, **kwargs):
         Raises:
             ValueError: if table's schema is not set
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
+        table = _table_arg_to_table(table, default_project=self.project)
 
+        if not isinstance(table, Table):
+            raise TypeError(_NEED_TABLE_ARGUMENT)
+
+        schema = table.schema
+
+        # selected_fields can override the table schema.
         if selected_fields is not None:
             schema = selected_fields
-        elif isinstance(table, TableReference):
-            raise ValueError("need selected_fields with TableReference")
-        elif isinstance(table, Table):
-            if len(table.schema) == 0:
-                raise ValueError(_TABLE_HAS_NO_SCHEMA)
-            schema = table.schema
-        else:
-            raise TypeError("table should be Table or TableReference")
+
+        if len(schema) == 0:
+            raise ValueError((
+                "Could not determine schema for table '{}'. Call client.get_table() "
+                "or pass in a list of schema fields to the selected_fields argument."
+            ).format(table))
 
         json_rows = [_record_field_to_json(schema, row) for row in rows]
 
@@ -1586,7 +1608,10 @@ def insert_rows_json(
                 identifies the row, and the "errors" key contains a list of
                 the mappings describing one or more problems with the row.
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
+        # Convert table to just a reference because unlike insert_rows,
+        # insert_rows_json doesn't need the table schema. It's not doing any
+        # type conversions.
+        table = _table_arg_to_table_ref(table, default_project=self.project)
         rows_info = []
         data = {"rows": rows_info}
 
@@ -1635,7 +1660,7 @@ def list_partitions(self, table, retry=DEFAULT_RETRY):
             List[str]:
                 A list of the partition ids present in the partitioned table
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
+        table = _table_arg_to_table_ref(table, default_project=self.project)
         meta_table = self.get_table(
             TableReference(
                 self.dataset(table.dataset_id, project=table.project),
@@ -1710,18 +1735,23 @@ def list_rows(
                 (this is distinct from the total number of rows in the
                 current page: ``iterator.page.num_items``).
         """
-        table = _str_or_ref_to_table_ref(table, default_project=self.project)
+        table = _table_arg_to_table(table, default_project=self.project)
+
+        if not isinstance(table, Table):
+            raise TypeError(_NEED_TABLE_ARGUMENT)
+
+        schema = table.schema
 
+        # selected_fields can override the table schema.
         if selected_fields is not None:
             schema = selected_fields
-        elif isinstance(table, TableReference):
-            raise ValueError("need selected_fields with TableReference")
-        elif isinstance(table, Table):
-            if len(table.schema) == 0 and table.created is None:
-                raise ValueError(_TABLE_HAS_NO_SCHEMA)
-            schema = table.schema
-        else:
-            raise TypeError("table should be Table or TableReference")
+
+        # Allow listing rows of an empty table by not raising if the table exists.
+        elif len(schema) == 0 and table.created is None:
+            raise ValueError((
+                "Could not determine schema for table '{}'. Call client.get_table() "
+                "or pass in a list of schema fields to the selected_fields argument."
+            ).format(table))
 
         params = {}
         if selected_fields is not None:
diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 2cb8e2ee2290..4fc470acd1e6 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -23,6 +23,8 @@
 import google.api_core.future.polling
 from google.cloud import exceptions
 from google.cloud.exceptions import NotFound
+from google.cloud.bigquery.dataset import Dataset
+from google.cloud.bigquery.dataset import DatasetListItem
 from google.cloud.bigquery.dataset import DatasetReference
 from google.cloud.bigquery.external_config import ExternalConfig
 from google.cloud.bigquery.query import _query_param_from_api_repr
@@ -34,7 +36,7 @@
 from google.cloud.bigquery.schema import SchemaField
 from google.cloud.bigquery.table import _EmptyRowIterator
 from google.cloud.bigquery.table import EncryptionConfiguration
-from google.cloud.bigquery.table import _str_or_ref_to_table_ref
+from google.cloud.bigquery.table import _table_arg_to_table_ref
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.table import Table
 from google.cloud.bigquery.table import TimePartitioning
@@ -2003,10 +2005,11 @@ def default_dataset(self):
 
         The ``dataset`` setter accepts:
 
+        - a :class:`~google.cloud.bigquery.dataset.Dataset`, or
         - a :class:`~google.cloud.bigquery.dataset.DatasetReference`, or
         - a :class:`str` of the fully-qualified dataset ID in standard SQL
           format. The value must included a project ID and dataset ID
-          separated by ``.``.
+          separated by ``.``. For example: ``your-project.your_dataset``.
 
         See
         https://g.co/cloud/bigquery/docs/reference/v2/jobs#configuration.query.defaultDataset
@@ -2025,6 +2028,9 @@ def default_dataset(self, value):
         if isinstance(value, six.string_types):
             value = DatasetReference.from_string(value)
 
+        if isinstance(value, (Dataset, DatasetListItem)):
+            value = value.reference
+
         resource = value.to_api_repr()
         self._set_sub_prop("defaultDataset", resource)
 
@@ -2035,10 +2041,12 @@ def destination(self):
 
         The ``destination`` setter accepts:
 
+        - a :class:`~google.cloud.bigquery.table.Table`, or
         - a :class:`~google.cloud.bigquery.table.TableReference`, or
         - a :class:`str` of the fully-qualified table ID in standard SQL
           format. The value must included a project ID, dataset ID, and table
-          ID, each separated by ``.``.
+          ID, each separated by ``.``. For example:
+          ``your-project.your_dataset.your_table``.
 
         See
         https://g.co/cloud/bigquery/docs/reference/rest/v2/jobs#configuration.query.destinationTable
@@ -2054,7 +2062,7 @@ def destination(self, value):
             self._set_sub_prop("destinationTable", None)
             return
 
-        value = _str_or_ref_to_table_ref(value)
+        value = _table_arg_to_table_ref(value)
         resource = value.to_api_repr()
         self._set_sub_prop("destinationTable", resource)
 
diff --git a/bigquery/google/cloud/bigquery/table.py b/bigquery/google/cloud/bigquery/table.py
index 12d5caafffb0..efab67d8bc69 100644
--- a/bigquery/google/cloud/bigquery/table.py
+++ b/bigquery/google/cloud/bigquery/table.py
@@ -1663,10 +1663,27 @@ def _rows_page_start(iterator, page, response):
 # pylint: enable=unused-argument
 
 
-def _str_or_ref_to_table_ref(value, default_project=None):
-    """Helper to convert a string to TableReference or keep a
-    TableReference unchanged.
+def _table_arg_to_table_ref(value, default_project=None):
+    """Helper to convert a string or Table to TableReference.
+
+    This function keeps TableReference and other kinds of objects unchanged.
+    """
+    if isinstance(value, six.string_types):
+        value = TableReference.from_string(value, default_project=default_project)
+    if isinstance(value, (Table, TableListItem)):
+        value = value.reference
+    return value
+
+
+def _table_arg_to_table(value, default_project=None):
+    """Helper to convert a string or TableReference to a Table.
+
+    This function keeps Table and other kinds of objects unchanged.
     """
     if isinstance(value, six.string_types):
         value = TableReference.from_string(value, default_project=default_project)
+    if isinstance(value, TableReference):
+        value = Table(value)
+
     return value
+
diff --git a/bigquery/tests/unit/test_client.py b/bigquery/tests/unit/test_client.py
index 4ffd589d8a49..4310a50deebf 100644
--- a/bigquery/tests/unit/test_client.py
+++ b/bigquery/tests/unit/test_client.py
@@ -3443,7 +3443,7 @@ def test_insert_rows_wo_schema(self):
         with self.assertRaises(ValueError) as exc:
             client.insert_rows(table, ROWS)
 
-        self.assertEqual(exc.exception.args, (_TABLE_HAS_NO_SCHEMA,))
+        self.assertIn("Could not determine schema for table", exc.exception.args[0])
 
     def test_insert_rows_w_schema(self):
         import datetime
diff --git a/bigquery/tests/unit/test_job.py b/bigquery/tests/unit/test_job.py
index 6bc73e82fb6d..833081ce066d 100644
--- a/bigquery/tests/unit/test_job.py
+++ b/bigquery/tests/unit/test_job.py
@@ -3188,6 +3188,15 @@ def test_default_dataset_w_string(self):
         expected = dataset.DatasetReference.from_string(default_dataset)
         self.assertEqual(config.default_dataset, expected)
 
+    def test_default_dataset_w_dataset(self):
+        from google.cloud.bigquery import dataset
+
+        default_dataset = "default-proj.default_dset"
+        expected = dataset.DatasetReference.from_string(default_dataset)
+        config = self._make_one()
+        config.default_dataset = dataset.Dataset(expected)
+        self.assertEqual(config.default_dataset, expected)
+
     def test_destinaton_w_string(self):
         from google.cloud.bigquery import table
 

From 84dbed9d3b277628b142bd4d7da777871809bbde Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Wed, 27 Mar 2019 10:13:23 -0700
Subject: [PATCH 3/4] Rename table_ref arg to table. Fix lint errors.

---
 bigquery/google/cloud/bigquery/client.py | 41 +++++++++++-------------
 bigquery/google/cloud/bigquery/table.py  |  1 -
 bigquery/tests/unit/test_client.py       |  2 +-
 3 files changed, 20 insertions(+), 24 deletions(-)

diff --git a/bigquery/google/cloud/bigquery/client.py b/bigquery/google/cloud/bigquery/client.py
index 3f79e948a5a8..fd69502f6ea1 100644
--- a/bigquery/google/cloud/bigquery/client.py
+++ b/bigquery/google/cloud/bigquery/client.py
@@ -54,7 +54,6 @@
 from google.cloud.bigquery.table import TableListItem
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.table import RowIterator
-from google.cloud.bigquery.table import _TABLE_HAS_NO_SCHEMA
 
 
 _DEFAULT_CHUNKSIZE = 1048576  # 1024 * 1024 B = 1 MB
@@ -429,11 +428,11 @@ def get_dataset(self, dataset_ref, retry=DEFAULT_RETRY):
         api_response = self._call_api(retry, method="GET", path=dataset_ref.path)
         return Dataset.from_api_repr(api_response)
 
-    def get_table(self, table_ref, retry=DEFAULT_RETRY):
-        """Fetch the table referenced by ``table_ref``.
+    def get_table(self, table, retry=DEFAULT_RETRY):
+        """Fetch the table referenced by ``table``.
 
         Args:
-            table_ref (Union[ \
+            table (Union[ \
                 :class:`~google.cloud.bigquery.table.Table`, \
                 :class:`~google.cloud.bigquery.table.TableReference`, \
                 str, \
@@ -449,7 +448,7 @@ def get_table(self, table_ref, retry=DEFAULT_RETRY):
             google.cloud.bigquery.table.Table:
                 A ``Table`` instance.
         """
-        table_ref = _table_arg_to_table_ref(table_ref, default_project=self.project)
+        table_ref = _table_arg_to_table_ref(table, default_project=self.project)
         api_response = self._call_api(retry, method="GET", path=table_ref.path)
         return Table.from_api_repr(api_response)
 
@@ -964,9 +963,7 @@ def load_table_from_uri(
         if isinstance(source_uris, six.string_types):
             source_uris = [source_uris]
 
-        destination = _table_arg_to_table_ref(
-            destination, default_project=self.project
-        )
+        destination = _table_arg_to_table_ref(destination, default_project=self.project)
         load_job = job.LoadJob(job_ref, source_uris, destination, self, job_config)
         load_job._begin(retry=retry)
 
@@ -1042,9 +1039,7 @@ def load_table_from_file(
         if location is None:
             location = self.location
 
-        destination = _table_arg_to_table_ref(
-            destination, default_project=self.project
-        )
+        destination = _table_arg_to_table_ref(destination, default_project=self.project)
         job_ref = job._JobReference(job_id, project=project, location=location)
         load_job = job.LoadJob(job_ref, None, destination, self, job_config)
         job_resource = load_job.to_api_repr()
@@ -1345,9 +1340,7 @@ def copy_table(
             for source in sources
         ]
 
-        destination = _table_arg_to_table_ref(
-            destination, default_project=self.project
-        )
+        destination = _table_arg_to_table_ref(destination, default_project=self.project)
 
         copy_job = job.CopyJob(
             job_ref, sources, destination, client=self, job_config=job_config
@@ -1550,10 +1543,12 @@ def insert_rows(self, table, rows, selected_fields=None, **kwargs):
             schema = selected_fields
 
         if len(schema) == 0:
-            raise ValueError((
-                "Could not determine schema for table '{}'. Call client.get_table() "
-                "or pass in a list of schema fields to the selected_fields argument."
-            ).format(table))
+            raise ValueError(
+                (
+                    "Could not determine schema for table '{}'. Call client.get_table() "
+                    "or pass in a list of schema fields to the selected_fields argument."
+                ).format(table)
+            )
 
         json_rows = [_record_field_to_json(schema, row) for row in rows]
 
@@ -1748,10 +1743,12 @@ def list_rows(
 
         # Allow listing rows of an empty table by not raising if the table exists.
         elif len(schema) == 0 and table.created is None:
-            raise ValueError((
-                "Could not determine schema for table '{}'. Call client.get_table() "
-                "or pass in a list of schema fields to the selected_fields argument."
-            ).format(table))
+            raise ValueError(
+                (
+                    "Could not determine schema for table '{}'. Call client.get_table() "
+                    "or pass in a list of schema fields to the selected_fields argument."
+                ).format(table)
+            )
 
         params = {}
         if selected_fields is not None:
diff --git a/bigquery/google/cloud/bigquery/table.py b/bigquery/google/cloud/bigquery/table.py
index efab67d8bc69..46419e70a83e 100644
--- a/bigquery/google/cloud/bigquery/table.py
+++ b/bigquery/google/cloud/bigquery/table.py
@@ -1686,4 +1686,3 @@ def _table_arg_to_table(value, default_project=None):
         value = Table(value)
 
     return value
-
diff --git a/bigquery/tests/unit/test_client.py b/bigquery/tests/unit/test_client.py
index 4310a50deebf..780f13bcbe3e 100644
--- a/bigquery/tests/unit/test_client.py
+++ b/bigquery/tests/unit/test_client.py
@@ -3427,7 +3427,7 @@ def test_query_w_query_parameters(self):
         )
 
     def test_insert_rows_wo_schema(self):
-        from google.cloud.bigquery.table import Table, _TABLE_HAS_NO_SCHEMA
+        from google.cloud.bigquery.table import Table
 
         creds = _make_credentials()
         http = object()

From fd8b06d76d25f98cef94d372600e73a1da4ef50b Mon Sep 17 00:00:00 2001
From: Tim Swast <swast@google.com>
Date: Wed, 27 Mar 2019 10:31:07 -0700
Subject: [PATCH 4/4] Fix typo in docstring.

---
 bigquery/google/cloud/bigquery/job.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/bigquery/google/cloud/bigquery/job.py b/bigquery/google/cloud/bigquery/job.py
index 4fc470acd1e6..f4c919a2d8fc 100644
--- a/bigquery/google/cloud/bigquery/job.py
+++ b/bigquery/google/cloud/bigquery/job.py
@@ -2003,7 +2003,7 @@ def default_dataset(self):
         to use for unqualified table names in the query or :data:`None` if not
         set.
 
-        The ``dataset`` setter accepts:
+        The ``default_dataset`` setter accepts:
 
         - a :class:`~google.cloud.bigquery.dataset.Dataset`, or
         - a :class:`~google.cloud.bigquery.dataset.DatasetReference`, or