Repository: ibis-project/ibis-bigquery (language: python)

Issue:
add datamgr code to create and populate tables for system tests
See: https://github.com/ibis-project/ibis/blob/98e2cd84478f9c8325879bb5b77527a23145f419/ci/datamgr.py#L484-L617 from ibis-framework 1.4.0
add datamgr code to create and populate tables for system tests
See: https://github.com/ibis-project/ibis/blob/98e2cd84478f9c8325879bb5b77527a23145f419/ci/datamgr.py#L484-L617 from ibis-framework 1.4.0
add datamgr code to create and populate tables for system tests
See: https://github.com/ibis-project/ibis/blob/98e2cd84478f9c8325879bb5b77527a23145f419/ci/datamgr.py#L484-L617 from ibis-framework 1.4.0


Discussion / hints:
We might want to do this as part of the test fixtures, instead. Either way, we need some way of populating these data.
@tswast I started to look at this while getting my dev environment set up.  Ok to proceed implementing this to get the test data populated?

```python
@pytest.fixture(scope='session')
def bqclient(client):
    return client.client


@pytest.fixture(scope='session')
def testing_dataset(bqclient):
    dataset_ref = bigquery.DatasetReference(bqclient.project, DATASET_ID)
    try:
        bqclient.get_dataset(dataset_ref)
    except NotFound:
        bqclient.create_dataset(dataset_ref)
    yield dataset_ref
    bqclient.delete_dataset(dataset_ref, delete_contents=True)


@pytest.fixture(scope='session')
def date_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'date_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_date_table(bqclient, date_table):
    table = bigquery.Table(date_table)
    table.schema = [
        bigquery.SchemaField('my_date_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_date_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)


@pytest.fixture(scope='session')
def timestamp_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'timestamp_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_timestamp_table(bqclient, timestamp_table):
    table = bigquery.Table(timestamp_table)
    table.schema = [
        bigquery.SchemaField('my_timestamp_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_timestamp_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)
```
@loudinb Fixtures like that in tests/system/conftest.py sound perfect.
We might want to do this as part of the test fixtures, instead. Either way, we need some way of populating these data.
@tswast I started to look at this while getting my dev environment set up.  Ok to proceed implementing this to get the test data populated?

```python
@pytest.fixture(scope='session')
def bqclient(client):
    return client.client


@pytest.fixture(scope='session')
def testing_dataset(bqclient):
    dataset_ref = bigquery.DatasetReference(bqclient.project, DATASET_ID)
    try:
        bqclient.get_dataset(dataset_ref)
    except NotFound:
        bqclient.create_dataset(dataset_ref)
    yield dataset_ref
    bqclient.delete_dataset(dataset_ref, delete_contents=True)


@pytest.fixture(scope='session')
def date_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'date_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_date_table(bqclient, date_table):
    table = bigquery.Table(date_table)
    table.schema = [
        bigquery.SchemaField('my_date_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_date_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)


@pytest.fixture(scope='session')
def timestamp_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'timestamp_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_timestamp_table(bqclient, timestamp_table):
    table = bigquery.Table(timestamp_table)
    table.schema = [
        bigquery.SchemaField('my_timestamp_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_timestamp_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)
```
@loudinb Fixtures like that in tests/system/conftest.py sound perfect.
We might want to do this as part of the test fixtures, instead. Either way, we need some way of populating these data.
@tswast I started to look at this while getting my dev environment set up.  Ok to proceed implementing this to get the test data populated?

```python
@pytest.fixture(scope='session')
def bqclient(client):
    return client.client


@pytest.fixture(scope='session')
def testing_dataset(bqclient):
    dataset_ref = bigquery.DatasetReference(bqclient.project, DATASET_ID)
    try:
        bqclient.get_dataset(dataset_ref)
    except NotFound:
        bqclient.create_dataset(dataset_ref)
    yield dataset_ref
    bqclient.delete_dataset(dataset_ref, delete_contents=True)


@pytest.fixture(scope='session')
def date_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'date_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_date_table(bqclient, date_table):
    table = bigquery.Table(date_table)
    table.schema = [
        bigquery.SchemaField('my_date_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_date_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)


@pytest.fixture(scope='session')
def timestamp_table(testing_dataset):
    return bigquery.TableReference(testing_dataset, 'timestamp_column_parted')


@pytest.fixture(autouse=True, scope='session')
def create_timestamp_table(bqclient, timestamp_table):
    table = bigquery.Table(timestamp_table)
    table.schema = [
        bigquery.SchemaField('my_timestamp_parted_col', 'DATE'),
        bigquery.SchemaField('string_col', 'STRING'),
        bigquery.SchemaField('int_col', 'INTEGER'),
    ]
    table.time_partitioning = bigquery.TimePartitioning(
        field='my_timestamp_parted_col'
    )
    bqclient.create_table(table, exists_ok=True)
```
@loudinb Fixtures like that in tests/system/conftest.py sound perfect.

Fix patch:
diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml
index 17cb9a3..d9a17e0 100644
--- a/.github/workflows/main.yml
+++ b/.github/workflows/main.yml
@@ -53,14 +53,14 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python_version: ["3.7", "3.8", "3.9"]
-        ibis_version: ["github"]
         include:
           - python_version: "3.7"
             "ibis_version": "1.2.0"
-          - python_version: "3.7"
-            "ibis_version":  "1.3.0"
           - python_version: "3.8"
+            "ibis_version":  "1.3.0"
+          - python_version: "3.9"
+            "ibis_version": "1.4.0"
+          - python_version: "3.10"
             "ibis_version": "1.4.0"
 
     steps:
diff --git a/environment.yml b/environment.yml
index 7339062..97f177e 100644
--- a/environment.yml
+++ b/environment.yml
@@ -1,13 +1,13 @@
-name: ibis-bigquery-dev
+name: ibis-v1-bigquery-dev
 channels:
 - conda-forge
 dependencies:
 
 # core
-- ibis-framework  # TODO: require Ibis 2.0 when it's released
-- cached_property  # New in Ibis 2.0 https://github.com/ibis-project/ibis/pull/2805
+- ibis-framework <=2.0.0dev
 - google-cloud-bigquery-core >=1.12.0,<3.0.0dev
 - pydata-google-auth
+- sqlalchemy=1.3.23
 
 # dev
 - black=19.10b0 # Same as ibis
diff --git a/setup.py b/setup.py
index 845d07f..786ec14 100644
--- a/setup.py
+++ b/setup.py
@@ -41,12 +41,14 @@
     packages=setuptools.find_packages(),
     python_requires=">=3.7",
     install_requires=[
-        # TODO require ibis 2.0 when it's released
-        "ibis-framework >= 1.2.0,<3.0.0dev",
+        "ibis-framework >=1.2.0,<2.0.0dev",
         "google-cloud-bigquery >=1.12.0,<3.0.0dev",
         "google-cloud-bigquery-storage >=1.0.0,<3.0.0dev",
         "pyarrow >=1.0.0,<5.0.0dev",
         "pydata-google-auth",
+        # Workaround for ibis 1.x incompatibility with SQLAlchemy 1.4
+        # AttributeError: module 'sqlalchemy' has no attribute 'Binary'   
+        "sqlalchemy <1.4.0dev",
     ],
     classifiers=[
         release_status,


Test patch:
diff --git a/.github/workflows/system-tests-pr.yml b/.github/workflows/system-tests-pr.yml
index 7300256..0d2d50c 100644
--- a/.github/workflows/system-tests-pr.yml
+++ b/.github/workflows/system-tests-pr.yml
@@ -17,14 +17,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python_version: ["3.7", "3.8", "3.9"]
-        ibis_version: ["github"]
         include:
           - python_version: "3.7"
             "ibis_version": "1.2.0"
-          - python_version: "3.7"
-            "ibis_version":  "1.3.0"
           - python_version: "3.8"
+            "ibis_version":  "1.3.0"
+          - python_version: "3.10"
             "ibis_version": "1.4.0"
 
     steps:
@@ -55,7 +53,7 @@ jobs:
         GCLOUD_KEY_PASSPHRASE: ${{ secrets.GCLOUD_KEY_PASSPHRASE }}
 
     - name: run tests
-      run: GOOGLE_BIGQUERY_PROJECT_ID="ibis-gbq" GOOGLE_APPLICATION_CREDENTIALS="$HOME/secrets/gcloud-service-key.json" python -m pytest tests/system
+      run: GOOGLE_BIGQUERY_PROJECT_ID="ibis-gbq" GOOGLE_APPLICATION_CREDENTIALS="$HOME/secrets/gcloud-service-key.json" python -m pytest tests/system --save-dataset
 
     # Remove the label so that it can be added again to re-trigger a build
     - name: reset label
diff --git a/.github/workflows/system-tests.yml b/.github/workflows/system-tests.yml
index 001b10c..a66dfa4 100644
--- a/.github/workflows/system-tests.yml
+++ b/.github/workflows/system-tests.yml
@@ -15,14 +15,12 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python_version: ["3.7", "3.8", "3.9"]
         ibis_version: ["github"]
-        include:
           - python_version: "3.7"
             "ibis_version": "1.2.0"
-          - python_version: "3.7"
-            "ibis_version":  "1.3.0"
           - python_version: "3.8"
+            "ibis_version":  "1.3.0"
+          - python_version: "3.10"
             "ibis_version": "1.4.0"
 
     steps:
@@ -51,4 +49,4 @@ jobs:
         GCLOUD_KEY_PASSPHRASE: ${{ secrets.GCLOUD_KEY_PASSPHRASE }}
 
     - name: run tests
-      run: GOOGLE_BIGQUERY_PROJECT_ID="ibis-gbq" GOOGLE_APPLICATION_CREDENTIALS="$HOME/secrets/gcloud-service-key.json" python -m pytest tests/system
+      run: GOOGLE_BIGQUERY_PROJECT_ID="ibis-gbq" GOOGLE_APPLICATION_CREDENTIALS="$HOME/secrets/gcloud-service-key.json" python -m pytest tests/system --save-dataset
diff --git a/tests/system/conftest.py b/tests/system/conftest.py
index 7b230d4..c706215 100644
--- a/tests/system/conftest.py
+++ b/tests/system/conftest.py
@@ -1,46 +1,62 @@
+import io
 import os
+import shutil
+import tempfile
+import urllib.request
 
 import ibis  # noqa: F401
 import pytest
-from google.oauth2 import service_account
+from google.api_core.exceptions import NotFound
+from google.cloud import bigquery
+import google.auth
+import google.auth.exceptions
 
 import ibis_bigquery
 
-PROJECT_ID = os.environ.get("GOOGLE_BIGQUERY_PROJECT_ID", "ibis-gbq")
-DATASET_ID = "testing"
+DEFAULT_PROJECT_ID = "ibis-gbq"
+PROJECT_ID_ENV_VAR = "GOOGLE_BIGQUERY_PROJECT_ID"
+DATASET_ID = "ibis_gbq_testing"
+TESTING_DATA_URI = (
+    'https://raw.githubusercontent.com/ibis-project/testing-data/master')
 
 bq = ibis_bigquery.Backend()
 
 
-def _credentials():
-    google_application_credentials = os.environ.get(
-        "GOOGLE_APPLICATION_CREDENTIALS", None
-    )
-    if google_application_credentials is None:
-        pytest.skip(
-            "Environment variable GOOGLE_APPLICATION_CREDENTIALS is " "not defined"
-        )
-    elif not google_application_credentials:
-        pytest.skip("Environment variable GOOGLE_APPLICATION_CREDENTIALS is empty")
-    elif not os.path.exists(google_application_credentials):
-        pytest.skip(
-            "Environment variable GOOGLE_APPLICATION_CREDENTIALS points "
-            "to {}, which does not exist".format(google_application_credentials)
-        )
+def pytest_addoption(parser):
+    parser.addoption('--save-dataset', action='store_true', default=False,
+                     help='saves all test data in the testing dataset')
+    parser.addoption('--no-refresh-dataset', action='store_true', default=False,
+                     help='do not refresh the test data in the testing dataset')
 
-    return service_account.Credentials.from_service_account_file(
-        google_application_credentials
-    )
+
+@pytest.fixture(scope="session")
+def dataset_id() -> str:
+    return DATASET_ID   
+
+@pytest.fixture(scope="session")
+def default_credentials():
+    try:
+        credentials, project_id = google.auth.default(scopes=ibis_bigquery.EXTERNAL_DATA_SCOPES)
+    except google.auth.excecptions.DefaultCredentialsError as exc:
+        pytest.skip(f"Could not get GCP credentials: {exc}")
+
+    return credentials, project_id
 
 
 @pytest.fixture(scope="session")
-def project_id():
-    return PROJECT_ID
+def project_id(default_credentials):
+    project_id = os.getenv(PROJECT_ID_ENV_VAR)
+    if project_id is None:
+        _, project_id = default_credentials
+    if project_id is None:
+        project_id = DEFAULT_PROJECT_ID
+    return project_id
 
 
 @pytest.fixture(scope="session")
-def credentials():
-    return _credentials()
+def credentials(default_credentials):
+    credentials, _ = default_credentials
+    return credentials
 
 
 @pytest.fixture(scope="session")
@@ -94,3 +110,243 @@ def public(project_id, credentials):
         dataset_id="bigquery-public-data.stackoverflow",
         credentials=credentials,
     )
+
+
+# Native BigQuery client fixtures
+# required to dynamically create the testing dataset,
+# the tables, and to populate data into the tables.
+@pytest.fixture(scope='session')
+def bqclient(client):
+    return client.client
+
+
+# Create testing dataset.
+@pytest.fixture(autouse=True, scope='session')
+def testing_dataset(bqclient, request, dataset_id):
+    dataset_ref = bigquery.DatasetReference(bqclient.project, dataset_id)
+    try:
+        bqclient.create_dataset(dataset_ref, exists_ok=True)
+    except NotFound:
+        pass
+    yield dataset_ref
+    if not request.config.getoption("--save-dataset"):
+        bqclient.delete_dataset(
+            dataset_ref, delete_contents=True, not_found_ok=True
+        )
+
+
+@pytest.fixture(scope='session')
+def functional_alltypes_table(testing_dataset):
+    return bigquery.TableReference(testing_dataset, 'functional_alltypes')
+
+
+@pytest.fixture(autouse=True, scope='session')
+def create_functional_alltypes_table(bqclient, functional_alltypes_table):
+    table = bigquery.Table(functional_alltypes_table)
+    table.schema = [
+        bigquery.SchemaField('index', 'INTEGER'),
+        bigquery.SchemaField('Unnamed_0', 'INTEGER'),
+        bigquery.SchemaField('id', 'INTEGER'),
+        bigquery.SchemaField('bool_col', 'BOOLEAN'),
+        bigquery.SchemaField('tinyint_col', 'INTEGER'),
+        bigquery.SchemaField('smallint_col', 'INTEGER'),
+        bigquery.SchemaField('int_col', 'INTEGER'),
+        bigquery.SchemaField('bigint_col', 'INTEGER'),
+        bigquery.SchemaField('float_col', 'FLOAT'),
+        bigquery.SchemaField('double_col', 'FLOAT'),
+        bigquery.SchemaField('date_string_col', 'STRING'),
+        bigquery.SchemaField('string_col', 'STRING'),
+        bigquery.SchemaField('timestamp_col', 'TIMESTAMP'),
+        bigquery.SchemaField('year', 'INTEGER'),
+        bigquery.SchemaField('month', 'INTEGER'),
+    ]
+    bqclient.create_table(table, exists_ok=True)
+    return table
+
+
+@pytest.fixture(autouse=True, scope='session')
+def load_functional_alltypes_data(request, bqclient, create_functional_alltypes_table):
+    if request.config.getoption("--no-refresh-dataset"):
+        return
+
+    table = create_functional_alltypes_table
+    load_config = bigquery.LoadJobConfig()
+    load_config.skip_leading_rows = 1  # skip the header row.
+    load_config.write_disposition = 'WRITE_TRUNCATE'
+    filepath = download_file(
+        '{}/functional_alltypes.csv'.format(TESTING_DATA_URI))
+    with open(filepath.name, 'rb') as csvfile:
+        job = bqclient.load_table_from_file(
+            csvfile,
+            table,
+            job_config=load_config,
+        ).result()
+    if job.error_result:
+        print('error')
+
+
+# Ingestion time partitioned table.
+@pytest.fixture(scope='session')
+def functional_alltypes_parted_table(testing_dataset):
+    return bigquery.TableReference(
+        testing_dataset, 'functional_alltypes_parted')
+
+
+@pytest.fixture(scope='session')
+def create_functional_alltypes_parted_table(
+        bqclient, functional_alltypes_parted_table):
+    table = bigquery.Table(functional_alltypes_parted_table)
+    table.schema = [
+        bigquery.SchemaField('index', 'INTEGER'),
+        bigquery.SchemaField('Unnamed_0', 'INTEGER'),
+        bigquery.SchemaField('id', 'INTEGER'),
+        bigquery.SchemaField('bool_col', 'BOOLEAN'),
+        bigquery.SchemaField('tinyint_col', 'INTEGER'),
+        bigquery.SchemaField('smallint_col', 'INTEGER'),
+        bigquery.SchemaField('int_col', 'INTEGER'),
+        bigquery.SchemaField('bigint_col', 'INTEGER'),
+        bigquery.SchemaField('float_col', 'FLOAT'),
+        bigquery.SchemaField('double_col', 'FLOAT'),
+        bigquery.SchemaField('date_string_col', 'STRING'),
+        bigquery.SchemaField('string_col', 'STRING'),
+        bigquery.SchemaField('timestamp_col', 'TIMESTAMP'),
+        bigquery.SchemaField('year', 'INTEGER'),
+        bigquery.SchemaField('month', 'INTEGER'),
+    ]
+    table.time_partitioning = bigquery.TimePartitioning(
+        type_=bigquery.TimePartitioningType.DAY
+    )
+    table.require_partition_filter = False
+    bqclient.create_table(table, exists_ok=True)
+    return table
+
+
+@pytest.fixture(autouse=True, scope='session')
+def load_functional_alltypes_parted_data(
+        request, bqclient, create_functional_alltypes_parted_table):
+    if request.config.getoption("--no-refresh-dataset"):
+        return
+
+    table = create_functional_alltypes_parted_table
+    load_config = bigquery.LoadJobConfig()
+    load_config.write_disposition = 'WRITE_TRUNCATE'
+    load_config.skip_leading_rows = 1  # skip the header row.
+    filepath = download_file(
+        '{}/functional_alltypes.csv'.format(TESTING_DATA_URI))
+    with open(filepath.name, 'rb') as csvfile:
+        job = bqclient.load_table_from_file(
+            csvfile,
+            table,
+            job_config=load_config,
+        ).result()
+    if job.error_result:
+        print('error')
+
+
+# Create a table with complex data types (nested and repeated).
+@pytest.fixture(scope='session')
+def struct_bq_table(testing_dataset):
+    return bigquery.TableReference(testing_dataset, 'struct_table')
+
+
+@pytest.fixture(autouse=True, scope='session')
+def load_struct_table_data(request, bqclient, struct_bq_table):
+    if request.config.getoption("--no-refresh-dataset"):
+        return
+
+    load_config = bigquery.LoadJobConfig()
+    load_config.write_disposition = 'WRITE_TRUNCATE'
+    load_config.source_format = 'AVRO'
+    filepath = download_file(
+        '{}/struct_table.avro'.format(TESTING_DATA_URI))
+    with open(filepath.name, 'rb') as avrofile:
+        job = bqclient.load_table_from_file(
+            avrofile,
+            struct_bq_table,
+            job_config=load_config,
+        ).result()
+    if job.error_result:
+        print('error')
+
+
+# Create empty date-partitioned table.
+@pytest.fixture(scope='session')
+def date_table(testing_dataset):
+    return bigquery.TableReference(testing_dataset, 'date_column_parted')
+
+
+@pytest.fixture(autouse=True, scope='session')
+def create_date_table(bqclient, date_table):
+    table = bigquery.Table(date_table)
+    table.schema = [
+        bigquery.SchemaField('my_date_parted_col', 'DATE'),
+        bigquery.SchemaField('string_col', 'STRING'),
+        bigquery.SchemaField('int_col', 'INTEGER'),
+    ]
+    table.time_partitioning = bigquery.TimePartitioning(
+        field='my_date_parted_col'
+    )
+    bqclient.create_table(table, exists_ok=True)
+    return table
+
+
+# Create empty timestamp-partitioned tables.
+@pytest.fixture(scope='session')
+def timestamp_table(testing_dataset):
+    return bigquery.TableReference(testing_dataset, 'timestamp_column_parted')
+
+
+@pytest.fixture(autouse=True, scope='session')
+def create_timestamp_table(bqclient, timestamp_table):
+    table = bigquery.Table(timestamp_table)
+    table.schema = [
+        bigquery.SchemaField('my_timestamp_parted_col', 'DATE'),
+        bigquery.SchemaField('string_col', 'STRING'),
+        bigquery.SchemaField('int_col', 'INTEGER'),
+    ]
+    table.time_partitioning = bigquery.TimePartitioning(
+        field='my_timestamp_parted_col'
+    )
+    bqclient.create_table(table, exists_ok=True)
+
+
+# Create a table with a numeric column
+@pytest.fixture(scope='session')
+def numeric_bq_table(testing_dataset):
+    return bigquery.TableReference(testing_dataset, 'numeric_table')
+
+
+@pytest.fixture(scope='session')
+def create_numeric_table(bqclient, numeric_bq_table):
+    table = bigquery.Table(numeric_bq_table)
+    table.schema = [
+        bigquery.SchemaField('string_col', 'STRING'),
+        bigquery.SchemaField('numeric_col', 'NUMERIC'),
+    ]
+    bqclient.create_table(table, exists_ok=True)
+    return table
+
+
+@pytest.fixture(autouse=True, scope='session')
+def load_numeric_data(request, bqclient, create_numeric_table):
+    if request.config.getoption("--no-refresh-dataset"):
+        return
+
+    load_config = bigquery.LoadJobConfig()
+    load_config.write_disposition = 'WRITE_TRUNCATE'
+    load_config.source_format = 'NEWLINE_DELIMITED_JSON'
+    data = u'''{"string_col": "1st value", "numeric_col": 0.999999999}\n\
+               {"string_col": "2nd value", "numeric_col": 0.000000002}'''
+    jsonfile = io.StringIO(data)
+    table = create_numeric_table
+    job = bqclient.load_table_from_file(
+        jsonfile, table, job_config=load_config).result()
+    if job.error_result:
+        print('error')
+
+
+def download_file(url):
+    with urllib.request.urlopen(url) as response:
+        with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
+            shutil.copyfileobj(response, tmp_file)
+    return tmp_file
diff --git a/tests/system/test_client.py b/tests/system/test_client.py
index 1d4b476..8bf31a8 100644
--- a/tests/system/test_client.py
+++ b/tests/system/test_client.py
@@ -1,6 +1,7 @@
 import collections
 import datetime
 import decimal
+from google.cloud.bigquery import dataset
 
 import ibis
 import ibis.expr.datatypes as dt
@@ -57,8 +58,8 @@ def test_list_tables(client):
     assert set(tables) == {"functional_alltypes", "functional_alltypes_parted"}
 
 
-def test_current_database(client):
-    assert client.current_database.name == "testing"
+def test_current_database(client, dataset_id):
+    assert client.current_database.name == dataset_id
     assert client.current_database.name == client.dataset_id
     assert client.current_database.tables == client.list_tables()
 
@@ -203,7 +204,7 @@ def test_different_partition_col_name(monkeypatch, client):
     assert col in parted_alltypes.columns
 
 
-def test_subquery_scalar_params(alltypes, project_id):
+def test_subquery_scalar_params(alltypes, project_id, dataset_id):
     t = alltypes
     param = ibis.param("timestamp").name("my_param")
     expr = (
@@ -221,12 +222,12 @@ def test_subquery_scalar_params(alltypes, project_id):
   SELECT `string_col`, sum(`float_col`) AS `foo`
   FROM (
     SELECT `float_col`, `timestamp_col`, `int_col`, `string_col`
-    FROM `{}.testing.functional_alltypes`
+    FROM `{}.{}.functional_alltypes`
     WHERE `timestamp_col` < @my_param
   ) t1
   GROUP BY 1
 ) t0""".format(
-        project_id
+        project_id, dataset_id
     )
     assert result == expected
 
@@ -456,7 +457,7 @@ def test_raw_sql(client):
     assert client.raw_sql("SELECT 1").fetchall() == [(1,)]
 
 
-def test_scalar_param_scope(alltypes, project_id):
+def test_scalar_param_scope(alltypes, project_id, dataset_id):
     t = alltypes
     param = ibis.param("timestamp")
     mut = t.mutate(param=param).compile(params={param: "2017-01-01"})
@@ -464,8 +465,8 @@ def test_scalar_param_scope(alltypes, project_id):
         mut
         == """\
 SELECT *, @param AS `param`
-FROM `{}.testing.functional_alltypes`""".format(
-            project_id
+FROM `{}.{}.functional_alltypes`""".format(
+            project_id, dataset_id
         )
     )
 
@@ -491,8 +492,8 @@ def test_exists_table(client):
     assert not client.exists_table("footable")
 
 
-def test_exists_database(client):
-    assert client.exists_database("testing")
+def test_exists_database(client, dataset_id):
+    assert client.exists_database(dataset_id)
     assert not client.exists_database("foodataset")
 
 
@@ -620,8 +621,8 @@ def test_string_to_timestamp(client):
     assert result_tz == timestamp_tz
 
 
-def test_client_sql_query(client):
-    expr = client.sql("select * from testing.functional_alltypes limit 20")
+def test_client_sql_query(client, dataset_id):
+    expr = client.sql(f"select * from {dataset_id}.functional_alltypes limit 20")
     result = expr.execute()
     expected = client.table("functional_alltypes").head(20).execute()
     tm.assert_frame_equal(result, expected)
@@ -633,7 +634,7 @@ def test_timestamp_column_parted_is_not_renamed(client):
     assert "PARTITIONTIME" not in t.columns
 
 
-def test_prevent_rewrite(alltypes, project_id):
+def test_prevent_rewrite(alltypes, project_id, dataset_id):
     t = alltypes
     expr = (
         t.groupby(t.string_col)
@@ -646,11 +647,11 @@ def test_prevent_rewrite(alltypes, project_id):
 SELECT *
 FROM (
   SELECT `string_col`, ARRAY_AGG(`double_col`) AS `collected_double`
-  FROM `{}.testing.functional_alltypes`
+  FROM `{}.{}.functional_alltypes`
   GROUP BY 1
 ) t0
 WHERE `string_col` != 'wat'""".format(
-        project_id
+        project_id, dataset_id
     )
     assert result == expected
 
@@ -704,6 +705,8 @@ def test_numeric_sum(numeric_table):
     expr = t.numeric_col.sum()
     result = expr.execute()
     assert isinstance(result, decimal.Decimal)
+    compare = result.compare(decimal.Decimal('1.000000001'))
+    assert compare == decimal.Decimal('0')
 
 
 def test_boolean_casting(alltypes):
@@ -711,9 +714,9 @@ def test_boolean_casting(alltypes):
     expr = t.groupby(k=t.string_col.nullif("1") == "9").count()
     result = expr.execute().set_index("k")
     count = result["count"]
-    assert count.loc[False] == 5840
-    assert count.loc[True] == 730
-    assert count.loc[None] == 730
+    assert count.at[False] == 5840
+    assert count.at[True] == 730
+    assert count.at[None] == 730
 
 
 def test_approx_median(alltypes):
diff --git a/tests/system/test_compiler.py b/tests/system/test_compiler.py
index e23a691..e5377ec 100644
--- a/tests/system/test_compiler.py
+++ b/tests/system/test_compiler.py
@@ -9,7 +9,7 @@
 IBIS_1_VERSION = packaging.version.Version("1.4.0")
 
 
-def test_timestamp_accepts_date_literals(alltypes, project_id):
+def test_timestamp_accepts_date_literals(alltypes, project_id, dataset_id):
     date_string = "2009-03-01"
     param = ibis.param(dt.timestamp).name("param_0")
     expr = alltypes.mutate(param=param)
@@ -17,63 +17,63 @@ def test_timestamp_accepts_date_literals(alltypes, project_id):
     result = expr.compile(params=params)
     expected = f"""\
 SELECT *, @param AS `param`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
 
 @pytest.mark.parametrize(
     ("distinct", "expected_keyword"), [(True, "DISTINCT"), (False, "ALL")]
 )
-def test_union(alltypes, distinct, expected_keyword, project_id):
+def test_union(alltypes, distinct, expected_keyword, project_id, dataset_id):
     expr = alltypes.union(alltypes, distinct=distinct)
     result = expr.compile()
     expected = f"""\
 SELECT *
-FROM `{project_id}.testing.functional_alltypes`
+FROM `{project_id}.{dataset_id}.functional_alltypes`
 UNION {expected_keyword}
 SELECT *
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
 
-def test_ieee_divide(alltypes, project_id):
+def test_ieee_divide(alltypes, project_id, dataset_id):
     expr = alltypes.double_col / 0
     result = expr.compile()
     expected = f"""\
 SELECT IEEE_DIVIDE(`double_col`, 0) AS `tmp`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
 
-def test_identical_to(alltypes, project_id):
+def test_identical_to(alltypes, project_id, dataset_id):
     t = alltypes
     pred = t.string_col.identical_to("a") & t.date_string_col.identical_to("b")
     expr = t[pred]
     result = expr.compile()
     expected = f"""\
 SELECT *
-FROM `{project_id}.testing.functional_alltypes`
+FROM `{project_id}.{dataset_id}.functional_alltypes`
 WHERE (((`string_col` IS NULL) AND ('a' IS NULL)) OR (`string_col` = 'a')) AND
       (((`date_string_col` IS NULL) AND ('b' IS NULL)) OR (`date_string_col` = 'b'))"""  # noqa: E501
     assert result == expected
 
 
 @pytest.mark.parametrize("timezone", [None, "America/New_York"])
-def test_to_timestamp(alltypes, timezone, project_id):
+def test_to_timestamp(alltypes, timezone, project_id, dataset_id):
     expr = alltypes.date_string_col.to_timestamp("%F", timezone)
     result = expr.compile()
     if timezone:
         expected = f"""\
 SELECT PARSE_TIMESTAMP('%F', `date_string_col`, 'America/New_York') AS `tmp`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     else:
         expected = f"""\
 SELECT PARSE_TIMESTAMP('%F', `date_string_col`) AS `tmp`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
 
-def test_window_function(alltypes, project_id):
+def test_window_function(alltypes, project_id, dataset_id):
     t = alltypes
     w1 = ibis.window(
         preceding=1, following=0, group_by="year", order_by="timestamp_col"
@@ -83,7 +83,7 @@ def test_window_function(alltypes, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (PARTITION BY `year` ORDER BY `timestamp_col` ROWS BETWEEN 1 PRECEDING AND CURRENT ROW) AS `win_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
     w2 = ibis.window(
@@ -94,7 +94,7 @@ def test_window_function(alltypes, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (PARTITION BY `year` ORDER BY `timestamp_col` ROWS BETWEEN CURRENT ROW AND 2 FOLLOWING) AS `win_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
     w3 = ibis.window(preceding=(4, 2), group_by="year", order_by="timestamp_col")
@@ -103,11 +103,11 @@ def test_window_function(alltypes, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (PARTITION BY `year` ORDER BY `timestamp_col` ROWS BETWEEN 4 PRECEDING AND 2 PRECEDING) AS `win_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_range_window_function(alltypes, project_id):
+def test_range_window_function(alltypes, project_id, dataset_id):
     if IBIS_VERSION <= IBIS_1_VERSION:
         pytest.skip("requires ibis 2.x")
     t = alltypes
@@ -117,7 +117,7 @@ def test_range_window_function(alltypes, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (PARTITION BY `year` ORDER BY `month` RANGE BETWEEN 1 PRECEDING AND CURRENT ROW) AS `two_month_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
     w3 = ibis.range_window(preceding=(4, 2), group_by="year", order_by="timestamp_col")
@@ -126,7 +126,7 @@ def test_range_window_function(alltypes, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (PARTITION BY `year` ORDER BY UNIX_MICROS(`timestamp_col`) RANGE BETWEEN 4 PRECEDING AND 2 PRECEDING) AS `win_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
@@ -144,7 +144,7 @@ def test_range_window_function(alltypes, project_id):
         (ibis.interval(weeks=1), 1000000 * 60 * 60 * 24 * 7),
     ],
 )
-def test_trailing_range_window(alltypes, preceding, value, project_id):
+def test_trailing_range_window(alltypes, preceding, value, project_id, dataset_id):
     if IBIS_VERSION <= IBIS_1_VERSION:
         pytest.skip("requires ibis 2.x")
     t = alltypes
@@ -154,7 +154,7 @@ def test_trailing_range_window(alltypes, preceding, value, project_id):
     expected = f"""\
 SELECT *,
        avg(`float_col`) OVER (ORDER BY UNIX_MICROS(`timestamp_col`) RANGE BETWEEN {value} PRECEDING AND CURRENT ROW) AS `win_avg`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
@@ -178,7 +178,7 @@ def test_trailing_range_window_unsupported(alltypes, preceding, value):
         (False, False, "UNION ALL", "UNION ALL"),
     ],
 )
-def test_union_cte(alltypes, distinct1, distinct2, expected1, expected2, project_id):
+def test_union_cte(alltypes, distinct1, distinct2, expected1, expected2, project_id, dataset_id):
     t = alltypes
     expr1 = t.group_by(t.string_col).aggregate(metric=t.double_col.sum())
     expr2 = expr1.view()
@@ -188,64 +188,64 @@ def test_union_cte(alltypes, distinct1, distinct2, expected1, expected2, project
     expected = f"""\
 WITH t0 AS (
   SELECT `string_col`, sum(`double_col`) AS `metric`
-  FROM `{project_id}.testing.functional_alltypes`
+  FROM `{project_id}.{dataset_id}.functional_alltypes`
   GROUP BY 1
 )
 SELECT *
 FROM t0
 {expected1}
 SELECT `string_col`, sum(`double_col`) AS `metric`
-FROM `{project_id}.testing.functional_alltypes`
+FROM `{project_id}.{dataset_id}.functional_alltypes`
 GROUP BY 1
 {expected2}
 SELECT `string_col`, sum(`double_col`) AS `metric`
-FROM `{project_id}.testing.functional_alltypes`
+FROM `{project_id}.{dataset_id}.functional_alltypes`
 GROUP BY 1"""
     assert result == expected
 
 
-def test_bool_reducers(alltypes, project_id):
+def test_bool_reducers(alltypes, project_id, dataset_id):
     b = alltypes.bool_col
     expr = b.mean()
     result = expr.compile()
     expected = f"""\
 SELECT avg(CAST(`bool_col` AS INT64)) AS `mean`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     expr2 = b.sum()
     result = expr2.compile()
     expected = f"""\
 SELECT sum(CAST(`bool_col` AS INT64)) AS `sum`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
 
-def test_bool_reducers_where(alltypes, project_id):
+def test_bool_reducers_where(alltypes, project_id, dataset_id):
     b = alltypes.bool_col
     m = alltypes.month
     expr = b.mean(where=m > 6)
     result = expr.compile()
     expected = f"""\
 SELECT avg(CASE WHEN `month` > 6 THEN CAST(`bool_col` AS INT64) ELSE NULL END) AS `mean`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
     expr2 = b.sum(where=((m > 6) & (m < 10)))
     result = expr2.compile()
     expected = f"""\
 SELECT sum(CASE WHEN (`month` > 6) AND (`month` < 10) THEN CAST(`bool_col` AS INT64) ELSE NULL END) AS `sum`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_approx_nunique(alltypes, project_id):
+def test_approx_nunique(alltypes, project_id, dataset_id):
     d = alltypes.double_col
     expr = d.approx_nunique()
     result = expr.compile()
     expected = f"""\
 SELECT APPROX_COUNT_DISTINCT(`double_col`) AS `approx_nunique`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     b = alltypes.bool_col
@@ -254,17 +254,17 @@ def test_approx_nunique(alltypes, project_id):
     result = expr2.compile()
     expected = f"""\
 SELECT APPROX_COUNT_DISTINCT(CASE WHEN `month` > 6 THEN `bool_col` ELSE NULL END) AS `approx_nunique`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_approx_median(alltypes, project_id):
+def test_approx_median(alltypes, project_id, dataset_id):
     d = alltypes.double_col
     expr = d.approx_median()
     result = expr.compile()
     expected = f"""\
 SELECT APPROX_QUANTILES(`double_col`, 2)[OFFSET(1)] AS `approx_median`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     m = alltypes.month
@@ -272,11 +272,11 @@ def test_approx_median(alltypes, project_id):
     result = expr2.compile()
     expected = f"""\
 SELECT APPROX_QUANTILES(CASE WHEN `month` > 6 THEN `double_col` ELSE NULL END, 2)[OFFSET(1)] AS `approx_median`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_bit_and(alltypes, project_id):
+def test_bit_and(alltypes, project_id, dataset_id):
     if IBIS_VERSION <= IBIS_1_VERSION:
         pytest.skip("requires ibis 2.x")
     i = alltypes.int_col
@@ -284,7 +284,7 @@ def test_bit_and(alltypes, project_id):
     result = expr.compile()
     expected = f"""\
 SELECT BIT_AND(`int_col`) AS `bit_and`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     b = alltypes.bigint_col
@@ -292,11 +292,11 @@ def test_bit_and(alltypes, project_id):
     result = expr2.compile()
     expected = f"""\
 SELECT BIT_AND(CASE WHEN `bigint_col` > 6 THEN `int_col` ELSE NULL END) AS `bit_and`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_bit_or(alltypes, project_id):
+def test_bit_or(alltypes, project_id, dataset_id):
     if IBIS_VERSION <= IBIS_1_VERSION:
         pytest.skip("requires ibis 2.x")
     i = alltypes.int_col
@@ -304,7 +304,7 @@ def test_bit_or(alltypes, project_id):
     result = expr.compile()
     expected = f"""\
 SELECT BIT_OR(`int_col`) AS `bit_or`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     b = alltypes.bigint_col
@@ -312,11 +312,11 @@ def test_bit_or(alltypes, project_id):
     result = expr2.compile()
     expected = f"""\
 SELECT BIT_OR(CASE WHEN `bigint_col` > 6 THEN `int_col` ELSE NULL END) AS `bit_or`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_bit_xor(alltypes, project_id):
+def test_bit_xor(alltypes, project_id, dataset_id):
     if IBIS_VERSION <= IBIS_1_VERSION:
         pytest.skip("requires ibis 2.x")
     i = alltypes.int_col
@@ -324,7 +324,7 @@ def test_bit_xor(alltypes, project_id):
     result = expr.compile()
     expected = f"""\
 SELECT BIT_XOR(`int_col`) AS `bit_xor`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     b = alltypes.bigint_col
@@ -332,24 +332,24 @@ def test_bit_xor(alltypes, project_id):
     result = expr2.compile()
     expected = f"""\
 SELECT BIT_XOR(CASE WHEN `bigint_col` > 6 THEN `int_col` ELSE NULL END) AS `bit_xor`
-FROM `{project_id}.testing.functional_alltypes`"""  # noqa: E501
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""  # noqa: E501
     assert result == expected
 
 
-def test_cov(alltypes, project_id):
+def test_cov(alltypes, project_id, dataset_id):
     d = alltypes.double_col
     expr = d.cov(d)
     result = expr.compile()
     expected = f"""\
 SELECT COVAR_SAMP(`double_col`, `double_col`) AS `tmp`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     expr = d.cov(d, how="pop")
     result = expr.compile()
     expected = f"""\
 SELECT COVAR_POP(`double_col`, `double_col`) AS `tmp`
-FROM `{project_id}.testing.functional_alltypes`"""
+FROM `{project_id}.{dataset_id}.functional_alltypes`"""
     assert result == expected
 
     with pytest.raises(ValueError):