From 3b4b4d1ea71749fdb82134b78bd0f09300316120 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Mon, 2 Nov 2020 13:13:19 -0500 Subject: [PATCH 01/25] Add a simple test suite for doing downstream tests --- .github/workflows/ci.yml | 38 ++++++ downstream_tests/test_dask_dataframe.py | 162 ++++++++++++++++++++++++ 2 files changed, 200 insertions(+) create mode 100644 downstream_tests/test_dask_dataframe.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a62dcb84..49593c13 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -51,3 +51,41 @@ jobs: run: | flake8 s3fs black s3fs --check + + test-downstream: + name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }}) + runs-on: ${{ matrix.os }} + strategy: + fail-fast: false + matrix: + os: ["ubuntu-latest", "macos-latest", "windows-latest"] + python-version: ["3.7", "3.8"] + pyarrow-version: [">=1,<2", ">=2,<3"] + + steps: + - uses: conda-incubator/setup-miniconda@v2 + with: + auto-update-conda: true + python-version: ${{ matrix.python-version }} + mamba-version: "*" + + - name: Checkout + uses: actions/checkout@v2 + + - name: Install conda dependencies + run: > + mamba create -n test-downstreams + -c conda-forge + 'python=${{ matrix.python-version }}' + 'pyarrow=${{ matrix.pyarrow-version }}' + pip botocore aiobotocore moto pytest fastparquet + + - name: Install portions from source + run: | + conda activate test-downstreams + pip install git+https://github.com/intake/filesystem_spec/ --no-deps + + - name: Run test suite + run: | + conda activate test-downstreams + pytest -vrsx downstream_tests diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py new file mode 100644 index 00000000..42ca5609 --- /dev/null +++ b/downstream_tests/test_dask_dataframe.py @@ -0,0 +1,162 @@ +import pytest +from pytest import fixture +import pandas as pd +import numpy as np +import time +import pyarrow as pa +import dask.dataframe as dd +import s3fs +import moto.server +import sys + + +@fixture(scope="session") +def partitioned_dataset() -> dict: + rows = 500000 + cds_df = pd.DataFrame( + { + "id": range(rows), + "part_key": np.random.choice(["A", "B", "C", "D"], rows), + "timestamp": np.random.randint(1051638817, 1551638817, rows), + "int_value": np.random.randint(0, 60000, rows), + } + ) + return {"dataframe": cds_df, "partitioning_column": "part_key"} + + +def free_port(): + import socketserver + + with socketserver.TCPServer(("localhost", 0), None) as s: + free_port = s.server_address[1] + return free_port + + +@fixture(scope="session") +def moto_server(): + import subprocess + + port = free_port() + process = subprocess.Popen([ + sys.executable, + moto.server.__file__, + '--port', str(port), + '--host', 'localhost', + 's3' + ]) + + s3fs_kwargs = dict( + client_kwargs={"endpoint_url": f'http://localhost:{port}'}, + ) + + start = time.time() + while True: + try: + fs = s3fs.S3FileSystem(skip_instance_cache=True, **s3fs_kwargs) + fs.ls("/") + except: + if time.time() - start > 30: + raise TimeoutError("Could not get a working moto server in time") + time.sleep(0.1) + + break + + yield s3fs_kwargs + + process.terminate() + + +@fixture(scope="session") +def moto_s3fs(moto_server): + return s3fs.S3FileSystem(**moto_server) + + +@fixture(scope="session") +def s3_bucket(moto_server): + test_bucket_name = 'test' + from botocore.session import Session + # NB: we use the sync botocore client for setup + session = Session() + client = session.create_client('s3', **moto_server['client_kwargs']) + client.create_bucket(Bucket=test_bucket_name, ACL='public-read') + return test_bucket_name + + +@fixture(scope="session") +def partitioned_parquet_path(partitioned_dataset, moto_s3fs, s3_bucket): + cds_df = partitioned_dataset["dataframe"] + table = pa.Table.from_pandas(cds_df, preserve_index=False) + path = s3_bucket + "/partitioned/dataset" + import pyarrow.parquet + + pyarrow.parquet.write_to_dataset( + table, + path, + filesystem=moto_s3fs, + partition_cols=[ + partitioned_dataset["partitioning_column"] + ], # new parameter included + ) + + # storage_options = dict(use_listings_cache=False) + # storage_options.update(docker_aws_s3.s3fs_kwargs) + # + # import dask.dataframe + # + # ddf = dask.dataframe.read_parquet( + # f"s3://{path}", storage_options=storage_options, gather_statistics=False + # ) + # all_rows = ddf.compute() + # assert "name" in all_rows.columns + return path + + +@pytest.fixture(scope='session', params=[ + pytest.param("pyarrow"), + pytest.param("fastparquet"), +]) +def parquet_engine(request): + return request.param + + +@pytest.fixture(scope='session', params=[ + pytest.param(False, id='gather_statistics=F'), + pytest.param(True, id='gather_statistics=T'), +]) +def gather_statistics(request): + return request.param + + +def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): + """The directory based reading is quite finicky""" + storage_options = moto_server.copy() + ddf = dd.read_parquet( + f"s3://{partitioned_parquet_path}", + storage_options=storage_options, + gather_statistics=gather_statistics, + engine=parquet_engine + ) + + assert 'part_key' in ddf.columns + actual = ddf.compute().sort_values('id') + + assert actual == partitioned_dataset["dataframe"] + + +def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): + """The directory based reading is quite finicky""" + storage_options = moto_server.copy() + ddf = dd.read_parquet( + f"s3://{partitioned_parquet_path}/part_key=A", + storage_options=storage_options, + gather_statistics=gather_statistics, + engine=parquet_engine + ) + + if parquet_engine == 'pyarrow': + assert 'part_key' in ddf.columns + actual = ddf.compute().sort_values('id') + expected = partitioned_dataset["dataframe"] + expected = expected.loc[expected.part_key == "A"] + + assert actual == expected From 34fc70e0cc494798ec2f76951bb977d826de49f3 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:24:19 -0500 Subject: [PATCH 02/25] Ensure that mamba works --- .github/workflows/ci.yml | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 49593c13..0fbdebe9 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,7 +52,7 @@ jobs: flake8 s3fs black s3fs --check - test-downstream: + test-downstream: &test-downstream name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }}) runs-on: ${{ matrix.os }} strategy: @@ -62,11 +62,17 @@ jobs: python-version: ["3.7", "3.8"] pyarrow-version: [">=1,<2", ">=2,<3"] + defaults: + run: + shell: bash -l {0} + steps: - uses: conda-incubator/setup-miniconda@v2 with: auto-update-conda: true python-version: ${{ matrix.python-version }} + channels: conda-forge,defaults + channel-priority: true mamba-version: "*" - name: Checkout @@ -74,7 +80,7 @@ jobs: - name: Install conda dependencies run: > - mamba create -n test-downstreams + mamba install -c conda-forge 'python=${{ matrix.python-version }}' 'pyarrow=${{ matrix.pyarrow-version }}' @@ -82,10 +88,8 @@ jobs: - name: Install portions from source run: | - conda activate test-downstreams pip install git+https://github.com/intake/filesystem_spec/ --no-deps - name: Run test suite run: | - conda activate test-downstreams pytest -vrsx downstream_tests From 149dd5010d7bb0f7412b13ba7329d43dd2db535c Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:25:52 -0500 Subject: [PATCH 03/25] Remove anchor --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0fbdebe9..d4247328 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,7 +52,7 @@ jobs: flake8 s3fs black s3fs --check - test-downstream: &test-downstream + test-downstream: name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }}) runs-on: ${{ matrix.os }} strategy: From b0ba969e922cb71006166d618a748c75fc4e8ba2 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:30:44 -0500 Subject: [PATCH 04/25] try reformatting string? --- .github/workflows/ci.yml | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d4247328..bb9dd0ea 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,7 +53,7 @@ jobs: black s3fs --check test-downstream: - name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }}) + name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ pyarrow-version }) runs-on: ${{ matrix.os }} strategy: fail-fast: false @@ -81,10 +81,10 @@ jobs: - name: Install conda dependencies run: > mamba install - -c conda-forge - 'python=${{ matrix.python-version }}' - 'pyarrow=${{ matrix.pyarrow-version }}' - pip botocore aiobotocore moto pytest fastparquet + -c conda-forge + 'python=${{ matrix.python-version }}' + 'pyarrow=${{ matrix.pyarrow-version }}' + pip botocore aiobotocore moto pytest fastparquet - name: Install portions from source run: | From bbd24983088c50690a1adddd60ae5a1495bd6602 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:31:44 -0500 Subject: [PATCH 05/25] try reformatting string? --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bb9dd0ea..827ebc02 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,7 +53,7 @@ jobs: black s3fs --check test-downstream: - name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ pyarrow-version }) + name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ matrix.pyarrow-version }) runs-on: ${{ matrix.os }} strategy: fail-fast: false From e5f6d855ac7db46d61b82a2f141a4cf7f52b5e48 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:31:59 -0500 Subject: [PATCH 06/25] try reformatting string? --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 827ebc02..07761b7d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,7 +53,7 @@ jobs: black s3fs --check test-downstream: - name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ matrix.pyarrow-version }) + name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ matrix.pyarrow-version }}) runs-on: ${{ matrix.os }} strategy: fail-fast: false From 95cc4b498d565bad47195d9338191a0438322647 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:34:26 -0500 Subject: [PATCH 07/25] try reformatting string? --- .github/workflows/ci.yml | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 07761b7d..84933645 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -83,8 +83,13 @@ jobs: mamba install -c conda-forge 'python=${{ matrix.python-version }}' - 'pyarrow=${{ matrix.pyarrow-version }}' - pip botocore aiobotocore moto pytest fastparquet + 'pyarrow${{ matrix.pyarrow-version }}' + pip + botocore + aiobotocore + moto + pytest + fastparquet - name: Install portions from source run: | From 7a15fd9443e6efad6a2d85e99b52fe4d26fba146 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:39:13 -0500 Subject: [PATCH 08/25] add dask as a dep --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 84933645..58a32e7c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -89,6 +89,7 @@ jobs: aiobotocore moto pytest + dask fastparquet - name: Install portions from source From c1b98da9b990a0305ff99a482c6c7aa5f747d2b9 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:43:41 -0500 Subject: [PATCH 09/25] install s3fs from the checkout --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 58a32e7c..76e8006b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,6 +95,7 @@ jobs: - name: Install portions from source run: | pip install git+https://github.com/intake/filesystem_spec/ --no-deps + pip install --no-deps . - name: Run test suite run: | From 04556d66c80644333270cf0f850f55045814fd41 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:48:04 -0500 Subject: [PATCH 10/25] Use fake aws credentials --- .github/workflows/ci.yml | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 76e8006b..ec572e5e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -66,6 +66,11 @@ jobs: run: shell: bash -l {0} + env: + BOTO_CONFIG: /dev/null + AWS_ACCESS_KEY_ID: foobar_key + AWS_SECRET_ACCESS_KEY: foobar_secret + steps: - uses: conda-incubator/setup-miniconda@v2 with: @@ -97,6 +102,12 @@ jobs: pip install git+https://github.com/intake/filesystem_spec/ --no-deps pip install --no-deps . + - name: conda packages + run: conda list + + - name: pip freeze + run: pip freeze + - name: Run test suite run: | pytest -vrsx downstream_tests From 7b1b3c70488f952e9cd4aec1f8e74880ea95c46d Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 10:55:26 -0500 Subject: [PATCH 11/25] Use fake aws credentials --- .github/workflows/ci.yml | 8 +++----- downstream_tests/test_dask_dataframe.py | 8 ++++++++ 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ec572e5e..c830704d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -66,11 +66,6 @@ jobs: run: shell: bash -l {0} - env: - BOTO_CONFIG: /dev/null - AWS_ACCESS_KEY_ID: foobar_key - AWS_SECRET_ACCESS_KEY: foobar_secret - steps: - uses: conda-incubator/setup-miniconda@v2 with: @@ -82,6 +77,9 @@ jobs: - name: Checkout uses: actions/checkout@v2 + with: + # we need the tags present + fetch-depth: 0 - name: Install conda dependencies run: > diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 42ca5609..1f7738be 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -8,6 +8,14 @@ import s3fs import moto.server import sys +import fsspec + + +@fixture(scope="session", autouse=True) +def aws_credentials(monkeypatch): + monkeypatch.setenv("BOTO_CONFIG", "/dev/null") + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "foobar_key") + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "foobar_secret") @fixture(scope="session") From 0e701f69e6eb2f2ba1ef17252073f4f564697113 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:09:09 -0500 Subject: [PATCH 12/25] fixed environment for pytest --- downstream_tests/test_dask_dataframe.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 1f7738be..491058b9 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -8,14 +8,14 @@ import s3fs import moto.server import sys -import fsspec +import os @fixture(scope="session", autouse=True) -def aws_credentials(monkeypatch): - monkeypatch.setenv("BOTO_CONFIG", "/dev/null") - monkeypatch.setenv("AWS_ACCESS_KEY_ID", "foobar_key") - monkeypatch.setenv("AWS_ACCESS_KEY_ID", "foobar_secret") +def aws_credentials(): + os.environ["BOTO_CONFIG"] = "/dev/null" + os.environ["AWS_ACCESS_KEY_ID"] = "foobar_key" + os.environ["AWS_ACCESS_KEY_ID"] = "foobar_secret" @fixture(scope="session") @@ -41,7 +41,7 @@ def free_port(): @fixture(scope="session") -def moto_server(): +def moto_server(aws_credentials): import subprocess port = free_port() From ee575a920d6629e68968c9c508e2fc38dea35739 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:12:17 -0500 Subject: [PATCH 13/25] fixed environment for pytest --- downstream_tests/test_dask_dataframe.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 491058b9..5bd83b1a 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -15,7 +15,7 @@ def aws_credentials(): os.environ["BOTO_CONFIG"] = "/dev/null" os.environ["AWS_ACCESS_KEY_ID"] = "foobar_key" - os.environ["AWS_ACCESS_KEY_ID"] = "foobar_secret" + os.environ["AWS_SECRET_ACCESS_KEY"] = "foobar_secret" @fixture(scope="session") From 45d4125753c9b323ccafe993081fe2554a6f30e1 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:20:19 -0500 Subject: [PATCH 14/25] Ensure that we have the correct tags --- .github/workflows/ci.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c830704d..dcd2328e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -81,6 +81,10 @@ jobs: # we need the tags present fetch-depth: 0 + - name: Ensure tags up to date + run: > + git fetch --tags https://github.com/dask/s3fs.git + - name: Install conda dependencies run: > mamba install From 84fc52a55acdad9465771404d5464f25ad1e3081 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:32:53 -0500 Subject: [PATCH 15/25] normalize column order for comparison --- .github/workflows/ci.yml | 1 + downstream_tests/test_dask_dataframe.py | 3 ++- 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dcd2328e..62c96877 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -98,6 +98,7 @@ jobs: pytest dask fastparquet + python-snappy - name: Install portions from source run: | diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 5bd83b1a..fbca6f21 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -147,8 +147,9 @@ def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_se assert 'part_key' in ddf.columns actual = ddf.compute().sort_values('id') + column_names = list(partitioned_dataset["dataframe"].columns) - assert actual == partitioned_dataset["dataframe"] + assert actual.loc[:, column_names] == partitioned_dataset["dataframe"].loc[:, column_names] def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): From 562dbb5b00ea927c920e096caa6931ce91e995ca Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:38:03 -0500 Subject: [PATCH 16/25] normalize columns for comparison --- downstream_tests/test_dask_dataframe.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index fbca6f21..05996a64 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -165,7 +165,8 @@ def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, mot if parquet_engine == 'pyarrow': assert 'part_key' in ddf.columns actual = ddf.compute().sort_values('id') - expected = partitioned_dataset["dataframe"] + expected: pd.DataFrame = partitioned_dataset["dataframe"] expected = expected.loc[expected.part_key == "A"] + expected = expected.drop(columns=['part_key']) assert actual == expected From cca6b0883775b67be224fe625040dd7906444116 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:42:37 -0500 Subject: [PATCH 17/25] show test summary --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 62c96877..cddc70c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,4 +113,4 @@ jobs: - name: Run test suite run: | - pytest -vrsx downstream_tests + pytest -vrsx -ra downstream_tests From ca170d9c4f626cc51400b3d92bcf040fb4e8942a Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:45:14 -0500 Subject: [PATCH 18/25] more colorful logs --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cddc70c1..f0f92798 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,4 +113,4 @@ jobs: - name: Run test suite run: | - pytest -vrsx -ra downstream_tests + pytest -vrsx -ra --color=yes downstream_tests From 4f61c54e519b8ba6f7d26871a78d203112cff7a9 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 11:59:55 -0500 Subject: [PATCH 19/25] better assertions --- .github/workflows/ci.yml | 2 +- downstream_tests/test_dask_dataframe.py | 19 +++++++++++++++---- 2 files changed, 16 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f0f92798..d1b55f60 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -113,4 +113,4 @@ jobs: - name: Run test suite run: | - pytest -vrsx -ra --color=yes downstream_tests + pytest -vrsx -rA --color=yes downstream_tests diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 05996a64..9230eabe 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -9,6 +9,7 @@ import moto.server import sys import os +from typing import List @fixture(scope="session", autouse=True) @@ -135,6 +136,15 @@ def gather_statistics(request): return request.param +def compare_dateframes(actual: pd.DataFrame, expected: pd.DataFrame, columns: List[str], id_column='id'): + from pandas.testing import assert_frame_equal + + actual = actual.set_index(id_column) + expected = expected.set_index(id_column) + + assert_frame_equal(actual.loc[:, columns], expected.loc[:, columns]) + + def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): """The directory based reading is quite finicky""" storage_options = moto_server.copy() @@ -149,7 +159,7 @@ def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_se actual = ddf.compute().sort_values('id') column_names = list(partitioned_dataset["dataframe"].columns) - assert actual.loc[:, column_names] == partitioned_dataset["dataframe"].loc[:, column_names] + compare_dateframes(actual, partitioned_dataset["dataframe"], column_names) def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): @@ -162,11 +172,12 @@ def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, mot engine=parquet_engine ) - if parquet_engine == 'pyarrow': - assert 'part_key' in ddf.columns + # if parquet_engine == 'pyarrow': + # assert 'part_key' in ddf.columns + actual = ddf.compute().sort_values('id') expected: pd.DataFrame = partitioned_dataset["dataframe"] expected = expected.loc[expected.part_key == "A"] expected = expected.drop(columns=['part_key']) - assert actual == expected + compare_dateframes(actual, partitioned_dataset["dataframe"], expected.column_names) From 72efe3010eb874b42e1d335ac5292672d0f87a95 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 12:01:21 -0500 Subject: [PATCH 20/25] better assertions --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d1b55f60..05e19813 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,8 +58,8 @@ jobs: strategy: fail-fast: false matrix: - os: ["ubuntu-latest", "macos-latest", "windows-latest"] - python-version: ["3.7", "3.8"] + os: ["ubuntu-latest", "windows-latest"] + python-version: ["3.8"] pyarrow-version: [">=1,<2", ">=2,<3"] defaults: From 41002ebdce67278f0609ba3b5ebf7439b2c2caec Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 12:05:44 -0500 Subject: [PATCH 21/25] better assertions --- .github/workflows/ci.yml | 2 +- downstream_tests/test_dask_dataframe.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 05e19813..f70f42a2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,7 +53,7 @@ jobs: black s3fs --check test-downstream: - name: Downstream tests (${{ matrix.python-version }}, ${{ matrix.os }} ${{ matrix.pyarrow-version }}) + name: Downstream (py${{ matrix.python-version }}, ${{ matrix.os }} arrow${{ matrix.pyarrow-version }}) runs-on: ${{ matrix.os }} strategy: fail-fast: false diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 9230eabe..69decba1 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -180,4 +180,4 @@ def test_non_partitioned_read(partitioned_dataset, partitioned_parquet_path, mot expected = expected.loc[expected.part_key == "A"] expected = expected.drop(columns=['part_key']) - compare_dateframes(actual, partitioned_dataset["dataframe"], expected.column_names) + compare_dateframes(actual, expected, list(expected.columns)) From c4845c5dbbf0ca067e38d46659f44358837eeba1 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 12:10:08 -0500 Subject: [PATCH 22/25] better assertions --- downstream_tests/test_dask_dataframe.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 69decba1..b4dc859b 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -139,6 +139,9 @@ def gather_statistics(request): def compare_dateframes(actual: pd.DataFrame, expected: pd.DataFrame, columns: List[str], id_column='id'): from pandas.testing import assert_frame_equal + if id_column in columns: + columns.remove(id_column) + actual = actual.set_index(id_column) expected = expected.set_index(id_column) From cf7c85f395a5215a8ac8ecd4515486bca0d77c42 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 13:49:30 -0500 Subject: [PATCH 23/25] better assertions --- downstream_tests/test_dask_dataframe.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index b4dc859b..f3c007c7 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -131,6 +131,7 @@ def parquet_engine(request): @pytest.fixture(scope='session', params=[ pytest.param(False, id='gather_statistics=F'), pytest.param(True, id='gather_statistics=T'), + pytest.param(None, id='gather_statistics=None'), ]) def gather_statistics(request): return request.param @@ -145,7 +146,7 @@ def compare_dateframes(actual: pd.DataFrame, expected: pd.DataFrame, columns: Li actual = actual.set_index(id_column) expected = expected.set_index(id_column) - assert_frame_equal(actual.loc[:, columns], expected.loc[:, columns]) + assert_frame_equal(actual.loc[:, columns], expected.loc[:, columns], check_dtype=False) def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_server, parquet_engine, gather_statistics): From 94a8369bccaa397b35d436c591206daa8fb393c9 Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 14:07:11 -0500 Subject: [PATCH 24/25] better assertions --- downstream_tests/test_dask_dataframe.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index f3c007c7..2b79fe34 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -161,6 +161,10 @@ def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_se assert 'part_key' in ddf.columns actual = ddf.compute().sort_values('id') + # dask converts our part_key into a categorical + actual[:, 'part_key'] = actual[:, 'part_key'].astype(str) + + column_names = list(partitioned_dataset["dataframe"].columns) compare_dateframes(actual, partitioned_dataset["dataframe"], column_names) From 483399feb2f1828a7da6a2a491c7743922f650bb Mon Sep 17 00:00:00 2001 From: Marius van Niekerk Date: Wed, 9 Dec 2020 14:19:38 -0500 Subject: [PATCH 25/25] asserts --- downstream_tests/test_dask_dataframe.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/downstream_tests/test_dask_dataframe.py b/downstream_tests/test_dask_dataframe.py index 2b79fe34..f202b31b 100644 --- a/downstream_tests/test_dask_dataframe.py +++ b/downstream_tests/test_dask_dataframe.py @@ -162,8 +162,7 @@ def test_partitioned_read(partitioned_dataset, partitioned_parquet_path, moto_se assert 'part_key' in ddf.columns actual = ddf.compute().sort_values('id') # dask converts our part_key into a categorical - actual[:, 'part_key'] = actual[:, 'part_key'].astype(str) - + actual['part_key'] = actual['part_key'].astype(str) column_names = list(partitioned_dataset["dataframe"].columns)