perspective-python 4.4.1

#  ┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
#  ┃ ██████ ██████ ██████       █      █      █      █      █ █▄  ▀███ █       ┃
#  ┃ ▄▄▄▄▄█ █▄▄▄▄▄ ▄▄▄▄▄█  ▀▀▀▀▀█▀▀▀▀▀ █ ▀▀▀▀▀█ ████████▌▐███ ███▄  ▀█ █ ▀▀▀▀▀ ┃
#  ┃ █▀▀▀▀▀ █▀▀▀▀▀ █▀██▀▀ ▄▄▄▄▄ █ ▄▄▄▄▄█ ▄▄▄▄▄█ ████████▌▐███ █████▄   █ ▄▄▄▄▄ ┃
#  ┃ █      ██████ █  ▀█▄       █ ██████      █      ███▌▐███ ███████▄ █       ┃
#  ┣━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┫
#  ┃ Copyright (c) 2017, the Perspective Authors.                              ┃
#  ┃ ╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌╌ ┃
#  ┃ This file is part of the Perspective library, distributed under the terms ┃
#  ┃ of the [Apache License 2.0](https://www.apache.org/licenses/LICENSE-2.0). ┃
#  ┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛

import os
import random
import uuid
import pyarrow as pa
import pandas as pd
import numpy as np
from datetime import date, datetime
from pytest import mark
import perspective as psp

client = psp.Server().new_local_client()
Table = client.table

SOURCE_STREAM_ARROW = os.path.join(
    os.path.dirname(__file__), "arrow", "int_float_str.arrow"
)
SOURCE_FILE_ARROW = os.path.join(
    os.path.dirname(__file__), "arrow", "int_float_str.arrow"
)
PARTIAL_ARROW = os.path.join(
    os.path.dirname(__file__), "arrow", "int_float_str_update.arrow"
)
DICT_ARROW = os.path.join(os.path.dirname(__file__), "arrow", "dict.arrow")
DICT_UPDATE_ARROW = os.path.join(
    os.path.dirname(__file__), "arrow", "dict_update.arrow"
)

names = ["a", "b", "c", "d"]


class TestUpdateArrow(object):
    # files

    def test_update_arrow_updates_stream_file(self):
        tbl = Table({"a": "integer", "b": "float", "c": "string"})

        with open(SOURCE_STREAM_ARROW, mode="rb") as file:  # b is important -> binary
            tbl.update(file.read())
            assert tbl.size() == 4
            assert tbl.schema() == {"a": "integer", "b": "float", "c": "string"}

        with open(SOURCE_FILE_ARROW, mode="rb") as file:
            tbl.update(file.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4] * 2,
                "b": [1.5, 2.5, 3.5, 4.5] * 2,
                "c": ["a", "b", "c", "d"] * 2,
            }

    def test_update_arrow_partial_updates_file(self):
        tbl = Table({"a": "integer", "b": "float", "c": "string"}, index="a")

        with open(SOURCE_STREAM_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 4

        with open(PARTIAL_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 4
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4],
                "b": [100.5, 2.5, 3.5, 400.5],
                "c": ["x", "b", "c", "y"],
            }

    def test_update_arrow_updates_dict_file(self):
        tbl = Table({"a": "string", "b": "string"})

        with open(DICT_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 5

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": ["abc", "def", "def", None, "abc", None, "update1", "update2"],
                "b": ["klm", "hij", None, "hij", "klm", "update3", None, "update4"],
            }

    @mark.skip
    def test_update_arrow_updates_dict_partial_file(self):
        tbl = None
        v = None

        with open(DICT_ARROW, mode="rb") as src:
            tbl = Table(src.read(), index="a")
            v = tbl.view()
            assert v.num_rows() == 2
            assert v.to_columns() == {"a": ["abc", "def"], "b": ["klm", "hij"]}

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            v.num_rows() == 4
            assert v.to_columns() == {
                "a": ["abc", "def", "update1", "update2"],
                "b": ["klm", "hij", None, "update4"],
            }

    # update with file arrow with more columns than in schema

    def test_update_arrow_updates_more_columns_stream_file(self):
        tbl = Table(
            {
                "a": "integer",
                "b": "float",
            }
        )

        with open(SOURCE_STREAM_ARROW, mode="rb") as file:  # b is important -> binary
            tbl.update(file.read())
            assert tbl.size() == 4
            assert tbl.schema() == {"a": "integer", "b": "float"}

        with open(SOURCE_FILE_ARROW, mode="rb") as file:
            tbl.update(file.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4] * 2,
                "b": [1.5, 2.5, 3.5, 4.5] * 2,
            }

    def test_update_arrow_partial_updates_more_columns_file(self):
        tbl = Table({"a": "integer", "c": "string"}, index="a")

        with open(SOURCE_STREAM_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 4

        with open(PARTIAL_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 4
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4],
                "c": ["x", "b", "c", "y"],
            }

    def test_update_arrow_updates_dict_more_columns_file(self):
        tbl = Table(
            {
                "a": "string",
            }
        )

        with open(DICT_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 5

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": ["abc", "def", "def", None, "abc", None, "update1", "update2"]
            }

    @mark.skip
    def test_update_arrow_updates_dict_more_columns_partial_file(self):
        tbl = Table({"a": "string"}, index="a")

        with open(DICT_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 4

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 4
            assert tbl.view().to_columns() == {
                "a": ["abc", "def", "update1", "update2"]
            }

    # update with file arrow with less columns than in schema

    def test_update_arrow_updates_less_columns_stream_file(self):
        tbl = Table(
            {
                "a": "integer",
                "x": "float",
            }
        )

        with open(SOURCE_STREAM_ARROW, mode="rb") as file:  # b is important -> binary
            tbl.update(file.read())
            assert tbl.size() == 4
            assert tbl.schema() == {"a": "integer", "x": "float"}

        with open(SOURCE_FILE_ARROW, mode="rb") as file:
            tbl.update(file.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4] * 2,
                "x": [None for i in range(8)],
            }

    def test_update_arrow_partial_updates_less_columns_file(self):
        tbl = Table({"a": "integer", "x": "string"}, index="a")

        with open(SOURCE_STREAM_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 4

        with open(PARTIAL_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 4
            assert tbl.view().to_columns() == {
                "a": [1, 2, 3, 4],
                "x": [None for i in range(4)],
            }

    def test_update_arrow_updates_dict_less_columns_file(self):
        tbl = Table({"a": "string", "x": "string"})

        with open(DICT_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 5

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 8
            assert tbl.view().to_columns() == {
                "a": ["abc", "def", "def", None, "abc", None, "update1", "update2"],
                "x": [None for i in range(8)],
            }

    @mark.skip
    def test_update_arrow_updates_dict_less_columns_partial_file(self):
        tbl = Table({"a": "string", "x": "string"}, index="a")

        with open(DICT_ARROW, mode="rb") as src:
            tbl.update(src.read())
            assert tbl.size() == 4

        with open(DICT_UPDATE_ARROW, mode="rb") as partial:
            tbl.update(partial.read())
            assert tbl.size() == 4
            assert tbl.view().to_columns() == {
                "a": ["abc", "def", "update1", "update2"],
                "x": [None for i in range(4)],
            }

    # update int schema with int

    def test_update_arrow_update_int_schema_with_uint8(self, util):
        array = [random.randint(0, 127) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint8)})

        schema = pa.schema({"a": pa.uint8()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_uint16(self, util):
        array = [random.randint(0, 32767) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint16)})

        schema = pa.schema({"a": pa.uint16()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_uint32(self, util):
        array = [random.randint(0, 2000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint32)})

        schema = pa.schema({"a": pa.uint32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_uint64(self, util):
        array = [random.randint(0, 20000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint64)})

        schema = pa.schema({"a": pa.uint64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_int8(self, util):
        array = [random.randint(-127, 127) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int8)})

        schema = pa.schema({"a": pa.int8()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_int16(self, util):
        array = [random.randint(-32767, 32767) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int16)})

        schema = pa.schema({"a": pa.int16()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_int32(self, util):
        array = [random.randint(-2000000, 2000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int32)})

        schema = pa.schema({"a": pa.int32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_int_schema_with_int64(self, util):
        array = [random.randint(-20000000, 20000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int64)})

        schema = pa.schema({"a": pa.int64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == [x * 1.0 for x in array]

    # updating float schema with int

    def test_update_arrow_update_float_schema_with_uint8(self, util):
        array = [random.randint(0, 127) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint8)})

        schema = pa.schema({"a": pa.uint8()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_uint16(self, util):
        array = [random.randint(0, 32767) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint16)})

        schema = pa.schema({"a": pa.uint16()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_uint32(self, util):
        array = [random.randint(0, 2000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint32)})

        schema = pa.schema({"a": pa.uint32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_uint64(self, util):
        array = [random.randint(0, 20000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.uint64)})

        schema = pa.schema({"a": pa.uint64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_int8(self, util):
        array = [random.randint(-127, 127) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int8)})

        schema = pa.schema({"a": pa.int8()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_int16(self, util):
        array = [random.randint(-32767, 32767) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int16)})

        schema = pa.schema({"a": pa.int16()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_int32(self, util):
        array = [random.randint(-2000000, 2000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int32)})

        schema = pa.schema({"a": pa.int32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_int64(self, util):
        array = [random.randint(-20000000, 20000000) for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.int64)})

        schema = pa.schema({"a": pa.int64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    # updating int schema with float
    def test_update_arrow_update_int_schema_with_float32(self, util):
        array = [random.randint(-2000000, 2000000) * 0.5 for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.float32)})

        schema = pa.schema({"a": pa.float32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == [int(x) for x in array]

    def test_update_arrow_update_int_schema_with_float64(self, util):
        array = [
            random.randint(-20000000, 20000000) * random.random() for i in range(100)
        ]
        data = pd.DataFrame({"a": np.array(array, dtype=np.float64)})

        schema = pa.schema({"a": pa.float64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "integer"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == [int(x) for x in array]

    # updating float schema with float

    def test_update_arrow_update_float_schema_with_float32(self, util):
        array = [random.randint(-2000000, 2000000) * 0.5 for i in range(100)]
        data = pd.DataFrame({"a": np.array(array, dtype=np.float32)})

        schema = pa.schema({"a": pa.float32()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    def test_update_arrow_update_float_schema_with_float64(self, util):
        array = [
            random.randint(-20000000, 20000000) * random.random() for i in range(100)
        ]
        data = pd.DataFrame({"a": np.array(array, dtype=np.float64)})

        schema = pa.schema({"a": pa.float64()})

        arrow = util.make_arrow_from_pandas(data, schema)
        tbl = Table({"a": "float"})
        tbl.update(arrow)
        assert tbl.view().to_columns()["a"] == array

    # updating date schema

    def test_update_arrow_update_date_schema_with_date32(self, util):
        array = [date(2019, 2, i) for i in range(1, 11)]
        data = pd.DataFrame({"a": array})

        schema = pa.schema({"a": pa.date32()})

        arrow = util.make_arrow_from_pandas(data, schema)

        tbl = Table({"a": "date"})

        tbl.update(arrow)

        assert tbl.view().to_columns()["a"] == [
            util.to_timestamp(datetime(2019, 2, i)) for i in range(1, 11)
        ]

    def test_update_arrow_update_date_schema_with_date64(self, util):
        array = [date(2019, 2, i) for i in range(1, 11)]
        data = pd.DataFrame({"a": array})

        schema = pa.schema({"a": pa.date64()})

        arrow = util.make_arrow_from_pandas(data, schema)

        tbl = Table({"a": "date"})

        tbl.update(arrow)

        assert tbl.view().to_columns()["a"] == [
            util.to_timestamp(datetime(2019, 2, i)) for i in range(1, 11)
        ]

    def test_update_arrow_update_datetime_schema_with_timestamp(self, util):
        data = [
            [datetime(2019, 2, i, 9) for i in range(1, 11)],
            [datetime(2019, 2, i, 10) for i in range(1, 11)],
            [datetime(2019, 2, i, 11) for i in range(1, 11)],
            [datetime(2019, 2, i, 12) for i in range(1, 11)],
        ]

        arrow_data = util.make_arrow(
            names,
            data,
            types=[
                pa.timestamp("s"),
                pa.timestamp("ms"),
                pa.timestamp("us"),
                pa.timestamp("ns"),
            ],
        )

        tbl = Table(
            {
                "a": "datetime",
                "b": "datetime",
                "c": "datetime",
                "d": "datetime",
            }
        )
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(d) for d in data[0]],
            "b": [util.to_timestamp(d) for d in data[1]],
            "c": [util.to_timestamp(d) for d in data[2]],
            "d": [util.to_timestamp(d) for d in data[3]],
        }

    # streams

    def test_update_arrow_updates_int_stream(self, util):
        data = [list(range(10)) for i in range(4)]
        arrow_data = util.make_arrow(names, data)
        tbl = Table({"a": "integer", "b": "integer", "c": "integer", "d": "integer"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {
            "a": data[0],
            "b": data[1],
            "c": data[2],
            "d": data[3],
        }

    def test_update_arrow_updates_float_stream(self, util):
        data = [[i for i in range(10)], [i * 1.5 for i in range(10)]]
        arrow_data = util.make_arrow(["a", "b"], data)
        tbl = Table(
            {
                "a": "integer",
                "b": "float",
            }
        )
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {"a": data[0], "b": data[1]}

    @mark.skip(reason="Decimal128 isn't part of our schema yet")
    def test_update_arrow_updates_decimal128_stream(self, util):
        data = [[i * 1000000000 for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.decimal128(10)])
        tbl = Table({"a": "integer"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {"a": data[0]}

    def test_update_arrow_updates_bool_stream(self, util):
        data = [[True if i % 2 == 0 else False for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data)
        tbl = Table({"a": "boolean"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {"a": data[0]}

    def test_update_arrow_updates_date32_stream(self, util):
        data = [[date(2019, 2, i) for i in range(1, 11)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.date32()])
        tbl = Table({"a": "date"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {
            "a": [int(1000 * datetime(2019, 2, i).timestamp()) for i in range(1, 11)]
        }

    def test_update_arrow_updates_date64_stream(self, util):
        data = [[date(2019, 2, i) for i in range(1, 11)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.date64()])
        tbl = Table({"a": "date"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(datetime(2019, 2, i)) for i in range(1, 11)]
        }

    def test_update_arrow_updates_timestamp_all_formats_stream(self, util):
        data = [
            [datetime(2019, 2, i, 9) for i in range(1, 11)],
            [datetime(2019, 2, i, 10) for i in range(1, 11)],
            [datetime(2019, 2, i, 11) for i in range(1, 11)],
            [datetime(2019, 2, i, 12) for i in range(1, 11)],
        ]
        arrow_data = util.make_arrow(
            names,
            data,
            types=[
                pa.timestamp("s"),
                pa.timestamp("ms"),
                pa.timestamp("us"),
                pa.timestamp("ns"),
            ],
        )
        tbl = Table(
            {"a": "datetime", "b": "datetime", "c": "datetime", "d": "datetime"}
        )
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(d) for d in data[0]],
            "b": [util.to_timestamp(d) for d in data[1]],
            "c": [util.to_timestamp(d) for d in data[2]],
            "d": [util.to_timestamp(d) for d in data[3]],
        }

    def test_update_arrow_updates_string_stream(self, util):
        data = [[str(i) for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.string()])
        tbl = Table({"a": "string"})
        tbl.update(arrow_data)
        assert tbl.size() == 10
        assert tbl.view().to_columns() == {"a": data[0]}

    def test_update_arrow_updates_dictionary_stream(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)
        tbl = Table({"a": "string", "b": "string"})
        tbl.update(arrow_data)

        assert tbl.size() == 4
        assert tbl.view().to_columns() == {
            "a": ["a", "b", "b", None],
            "b": ["x", "y", None, "z"],
        }

    @mark.skip(reason="Arrow no longer supports partial updates per row")
    def test_update_arrow_partial_updates_dictionary_stream(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)
        tbl = Table({"a": "string", "b": "string"}, index="a")
        tbl.update(arrow_data)
        assert tbl.size() == 3
        assert tbl.view().to_columns() == {"a": [None, "a", "b"], "b": ["z", "x", "y"]}

    @mark.skip
    def test_update_arrow_partial_updates_dictionary_stream_duplicates(self, util):
        """If there are duplicate values in the dictionary, primary keys
        may be duplicated if the column is used as an index. Skip this test
        for now - still looking for the best way to fix."""
        data = [
            ([0, 1, 1, None, 2], ["a", "b", "a"]),
            ([0, 1, None, 2, 1], ["x", "y", "z"]),
        ]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)

        tbl = Table({"a": "string", "b": "string"}, index="a")

        tbl.update(arrow_data)

        assert tbl.size() == 3
        assert tbl.view().to_columns() == {"a": [None, "a", "b"], "b": ["z", "x", "y"]}

    def test_update_arrow_partial_updates_more_columns_dictionary_stream(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)

        tbl = Table({"a": "string"}, index="a")

        tbl.update(arrow_data)

        assert tbl.size() == 3
        assert tbl.view().to_columns() == {"a": [None, "a", "b"]}

    @mark.skip(reason="Arrow no longer supports partial updates per row")
    def test_update_arrow_partial_updates_less_columns_dictionary_stream(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)
        tbl = Table({"a": "string", "b": "string", "x": "string"}, index="a")
        tbl.update(arrow_data)

        assert tbl.size() == 3
        assert tbl.view().to_columns() == {
            "a": [None, "a", "b"],
            "b": ["z", "x", "y"],
            "x": [None, None, None],
        }

    def test_update_arrow_arbitary_order(self, util):
        data = [[1, 2, 3, 4], ["a", "b", "c", "d"], [1, 2, 3, 4], ["a", "b", "c", "d"]]
        update_data = [[5, 6], ["e", "f"], [5, 6], ["e", "f"]]
        arrow = util.make_arrow(["a", "b", "c", "d"], data)
        update_arrow = util.make_arrow(["c", "b", "a", "d"], update_data)
        tbl = Table(arrow)
        assert tbl.schema() == {
            "a": "integer",
            "b": "string",
            "c": "integer",
            "d": "string",
        }
        tbl.update(update_arrow)
        assert tbl.size() == 6
        assert tbl.view().to_columns() == {
            "a": [1, 2, 3, 4, 5, 6],
            "b": ["a", "b", "c", "d", "e", "f"],
            "c": [1, 2, 3, 4, 5, 6],
            "d": ["a", "b", "c", "d", "e", "f"],
        }

    # append

    def test_update_arrow_updates_append_int_stream(self, util):
        data = [list(range(10)) for i in range(4)]
        arrow_data = util.make_arrow(names, data)
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {
            "a": data[0] + data[0],
            "b": data[1] + data[1],
            "c": data[2] + data[2],
            "d": data[3] + data[3],
        }

    def test_update_arrow_updates_append_float_stream(self, util):
        data = [[i for i in range(10)], [i * 1.5 for i in range(10)]]
        arrow_data = util.make_arrow(["a", "b"], data)
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {
            "a": data[0] + data[0],
            "b": data[1] + data[1],
        }

    def test_update_arrow_updates_append_decimal_stream(self, util):
        data = [[i * 1000 for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.decimal128(4)])
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {"a": data[0] + data[0]}

    def test_update_arrow_updates_append_bool_stream(self, util):
        data = [[True if i % 2 == 0 else False for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data)
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {"a": data[0] + data[0]}

    def test_update_arrow_updates_append_date32_stream(self, util):
        data = [[date(2019, 2, i) for i in range(1, 11)]]
        out_data = [datetime(2019, 2, i) for i in range(1, 11)]
        arrow_data = util.make_arrow(["a"], data, types=[pa.date32()])
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(d) for d in out_data + out_data]
        }

    def test_update_arrow_updates_append_date64_stream(self, util):
        data = [[date(2019, 2, i) for i in range(1, 11)]]
        out_data = [datetime(2019, 2, i) for i in range(1, 11)]
        arrow_data = util.make_arrow(["a"], data, types=[pa.date64()])
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(d) for d in out_data + out_data]
        }

    def test_update_arrow_updates_append_timestamp_all_formats_stream(self, util):
        data = [
            [datetime(2019, 2, i, 9) for i in range(1, 11)],
            [datetime(2019, 2, i, 10) for i in range(1, 11)],
            [datetime(2019, 2, i, 11) for i in range(1, 11)],
            [datetime(2019, 2, i, 12) for i in range(1, 11)],
        ]
        arrow_data = util.make_arrow(
            names,
            data,
            types=[
                pa.timestamp("s"),
                pa.timestamp("ms"),
                pa.timestamp("us"),
                pa.timestamp("ns"),
            ],
        )
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {
            "a": [util.to_timestamp(d) for d in data[0] + data[0]],
            "b": [util.to_timestamp(d) for d in data[1] + data[1]],
            "c": [util.to_timestamp(d) for d in data[2] + data[2]],
            "d": [util.to_timestamp(d) for d in data[3] + data[3]],
        }

    def test_update_arrow_updates_append_string_stream(self, util):
        data = [[str(i) for i in range(10)]]
        arrow_data = util.make_arrow(["a"], data, types=[pa.string()])
        tbl = Table(arrow_data)
        tbl.update(arrow_data)
        assert tbl.size() == 20
        assert tbl.view().to_columns() == {"a": data[0] + data[0]}

    def test_update_arrow_updates_append_dictionary_stream(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data)
        tbl = Table(arrow_data)
        tbl.update(arrow_data)

        assert tbl.size() == 8
        assert tbl.view().to_columns() == {
            "a": ["a", "b", "b", None, "a", "b", "b", None],
            "b": ["x", "y", None, "z", "x", "y", None, "z"],
        }

    def test_update_arrow_updates_append_dictionary_stream_legacy(self, util):
        data = [([0, 1, 1, None], ["a", "b"]), ([0, 1, None, 2], ["x", "y", "z"])]
        arrow_data = util.make_dictionary_arrow(["a", "b"], data, legacy=True)
        tbl = Table(arrow_data)
        tbl.update(arrow_data)

        assert tbl.size() == 8
        assert tbl.view().to_columns() == {
            "a": ["a", "b", "b", None, "a", "b", "b", None],
            "b": ["x", "y", None, "z", "x", "y", None, "z"],
        }

    # indexed

    def test_update_arrow_partial_indexed(self, util):
        data = [[1, 2, 3, 4], ["a", "b", "c", "d"]]
        update_data = [[2, 4], ["x", "y"]]
        arrow = util.make_arrow(["a", "b"], data)
        update_arrow = util.make_arrow(["a", "b"], update_data)
        tbl = Table(arrow, index="a")
        assert tbl.schema() == {"a": "integer", "b": "string"}
        tbl.update(update_arrow)
        assert tbl.size() == 4
        assert tbl.view().to_columns() == {"a": [1, 2, 3, 4], "b": ["a", "x", "c", "y"]}

    # update specific columns

    def test_update_arrow_specific_column(self, util):
        data = [[1, 2, 3, 4], ["a", "b", "c", "d"]]
        update_data = [[2, 3, 4]]
        arrow = util.make_arrow(["a", "b"], data)
        update_arrow = util.make_arrow(["a"], update_data)
        tbl = Table(arrow)
        assert tbl.schema() == {"a": "integer", "b": "string"}
        tbl.update(update_arrow)
        assert tbl.size() == 7
        assert tbl.view().to_columns() == {
            "a": [1, 2, 3, 4, 2, 3, 4],
            "b": ["a", "b", "c", "d", None, None, None],
        }

    # try to fuzz column order

    def test_update_arrow_column_order_str(self, util):
        # use str so it doesn't get promoted
        data = [["a", "b", "c"] for i in range(10)]
        names = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
        names_scrambled = names[::-1]
        arrow = util.make_arrow(names_scrambled, data)
        tbl = Table({name: "string" for name in names})
        tbl.update(arrow)
        assert tbl.size() == 3
        assert tbl.view().to_columns() == {name: data[0] for name in names}

    def test_update_arrow_column_order_int(self, util):
        data = [[1, 2, 3] for i in range(10)]
        names = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j"]
        names_scrambled = names[::-1]
        arrow = util.make_arrow(names_scrambled, data)
        tbl = Table({name: "integer" for name in names})
        tbl.update(arrow)
        assert tbl.size() == 3
        assert tbl.view().to_columns() == {name: data[0] for name in names}

    def test_update_arrow_thread_safe_int_index(self, util):
        data = [["a", "b", "c"] for i in range(10)]
        data += [[1, 2, 3]]
        names = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "uid"]
        arrow = util.make_arrow(names, data)
        tbl = Table(arrow, index="uid")

        for i in range(100):
            idx = (1, 2, 3)[random.randint(0, 2)]
            update_data = [
                [str(uuid.uuid4()) + str(random.randint(100, 1000000000))],
                [idx],
            ]
            update_names = [names[random.randint(0, 9)], "uid"]
            update_arrow = util.make_arrow(update_names, update_data)
            tbl.update(update_arrow)

        assert tbl.size() == 3

    def test_update_arrow_thread_safe_datetime_index(self, util):
        data = [["a", "b", "c"] for i in range(10)]
        data += [
            [
                datetime(2020, 1, 15, 12, 17),
                datetime(2020, 1, 15, 12, 18),
                datetime(2020, 1, 15, 12, 19),
            ]
        ]
        names = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "uid"]
        arrow = util.make_arrow(names, data)
        tbl = Table(arrow, index="uid")

        for i in range(100):
            idx = (
                datetime(2020, 1, 15, 12, 17),
                datetime(2020, 1, 15, 12, 18),
                datetime(2020, 1, 15, 12, 19),
            )[random.randint(0, 2)]
            update_data = [
                [str(uuid.uuid4()) + str(random.randint(100, 1000000000))],
                [idx],
            ]
            update_names = [names[random.randint(0, 9)], "uid"]
            update_arrow = util.make_arrow(update_names, update_data)
            tbl.update(update_arrow)

        assert tbl.size() == 3

    def test_update_arrow_thread_safe_str_index(self, util):
        data = [["a", "b", "c"] for i in range(11)]
        names = ["a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "uid"]
        arrow = util.make_arrow(names, data)
        tbl = Table(arrow, index="uid")

        for i in range(100):
            idx = ("a", "b", "c")[random.randint(0, 2)]
            update_data = [
                [str(uuid.uuid4()) + str(random.randint(100, 1000000000))],
                [idx],
            ]
            update_names = [names[random.randint(0, 9)], "uid"]
            update_arrow = util.make_arrow(update_names, update_data)
            tbl.update(update_arrow)

        assert tbl.size() == 3