# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import abc
from collections import OrderedDict
from decimal import Decimal
import io
import itertools
import json
import string
import unittest

try:
    import numpy as np
except ImportError:
    np = None
import pytest

import pyarrow as pa
from pyarrow.json import read_json, open_json, ReadOptions, ParseOptions


def generate_col_names():
    # 'a', 'b'... 'z', then 'aa', 'ab'...
    letters = string.ascii_lowercase
    yield from letters
    for first in letters:
        for second in letters:
            yield first + second


def make_random_json(num_cols=2, num_rows=10, linesep='\r\n'):
    arr = np.random.RandomState(42).randint(0, 1000, size=(num_cols, num_rows))
    col_names = list(itertools.islice(generate_col_names(), num_cols))
    lines = []
    for row in arr.T:
        json_obj = OrderedDict([(k, int(v)) for (k, v) in zip(col_names, row)])
        lines.append(json.dumps(json_obj))
    data = linesep.join(lines).encode()
    columns = [pa.array(col, type=pa.int64()) for col in arr]
    expected = pa.Table.from_arrays(columns, col_names)
    return data, expected


def check_options_class_pickling(cls, pickler, **attr_values):
    opts = cls(**attr_values)
    new_opts = pickler.loads(pickler.dumps(opts,
                                           protocol=pickler.HIGHEST_PROTOCOL))
    for name, value in attr_values.items():
        assert getattr(new_opts, name) == value


def test_read_options(pickle_module):
    cls = ReadOptions
    opts = cls()

    assert opts.block_size > 0
    opts.block_size = 12345
    assert opts.block_size == 12345

    assert opts.use_threads is True
    opts.use_threads = False
    assert opts.use_threads is False

    opts = cls(block_size=1234, use_threads=False)
    assert opts.block_size == 1234
    assert opts.use_threads is False

    check_options_class_pickling(cls, pickler=pickle_module,
                                 block_size=1234,
                                 use_threads=False)


def test_parse_options(pickle_module):
    cls = ParseOptions
    opts = cls()
    assert opts.newlines_in_values is False
    assert opts.explicit_schema is None

    opts.newlines_in_values = True
    assert opts.newlines_in_values is True

    schema = pa.schema([pa.field('foo', pa.int32())])
    opts.explicit_schema = schema
    assert opts.explicit_schema == schema

    assert opts.unexpected_field_behavior == "infer"
    for value in ["ignore", "error", "infer"]:
        opts.unexpected_field_behavior = value
        assert opts.unexpected_field_behavior == value

    with pytest.raises(ValueError):
        opts.unexpected_field_behavior = "invalid-value"

    check_options_class_pickling(cls, pickler=pickle_module,
                                 explicit_schema=schema,
                                 newlines_in_values=False,
                                 unexpected_field_behavior="ignore")


class BaseTestJSON(abc.ABC):
    @abc.abstractmethod
    def read_bytes(self, b, **kwargs):
        """
        :param b: bytes to be parsed
        :param kwargs: arguments passed on to open the json file
        :return: b parsed as a single Table
        """
        raise NotImplementedError

    def check_names(self, table, names):
        assert table.num_columns == len(names)
        assert [c.name for c in table.columns] == names

    def test_block_sizes(self):
        rows = b'{"a": 1}\n{"a": 2}\n{"a": 3}'
        read_options = ReadOptions()
        parse_options = ParseOptions()

        for data in [rows, rows + b'\n']:
            for newlines_in_values in [False, True]:
                parse_options.newlines_in_values = newlines_in_values
                read_options.block_size = 4
                with pytest.raises(ValueError,
                                   match="try to increase block size"):
                    self.read_bytes(data, read_options=read_options,
                                    parse_options=parse_options)

                # Validate reader behavior with various block sizes.
                # There used to be bugs in this area.
                for block_size in range(9, 20):
                    read_options.block_size = block_size
                    table = self.read_bytes(data, read_options=read_options,
                                            parse_options=parse_options)
                    assert table.to_pydict() == {'a': [1, 2, 3]}

    def test_no_newline_at_end(self):
        rows = b'{"a": 1,"b": 2, "c": 3}\n{"a": 4,"b": 5, "c": 6}'
        table = self.read_bytes(rows)
        assert table.to_pydict() == {
            'a': [1, 4],
            'b': [2, 5],
            'c': [3, 6],
        }

    def test_simple_ints(self):
        # Infer integer columns
        rows = b'{"a": 1,"b": 2, "c": 3}\n{"a": 4,"b": 5, "c": 6}\n'
        table = self.read_bytes(rows)
        schema = pa.schema([('a', pa.int64()),
                            ('b', pa.int64()),
                            ('c', pa.int64())])
        assert table.schema == schema
        assert table.to_pydict() == {
            'a': [1, 4],
            'b': [2, 5],
            'c': [3, 6],
        }

    def test_simple_varied(self):
        # Infer various kinds of data
        rows = (b'{"a": 1,"b": 2, "c": "3", "d": false}\n'
                b'{"a": 4.0, "b": -5, "c": "foo", "d": true}\n')
        table = self.read_bytes(rows)
        schema = pa.schema([('a', pa.float64()),
                            ('b', pa.int64()),
                            ('c', pa.string()),
                            ('d', pa.bool_())])
        assert table.schema == schema
        assert table.to_pydict() == {
            'a': [1.0, 4.0],
            'b': [2, -5],
            'c': ["3", "foo"],
            'd': [False, True],
        }

    def test_simple_nulls(self):
        # Infer various kinds of data, with nulls
        rows = (b'{"a": 1, "b": 2, "c": null, "d": null, "e": null}\n'
                b'{"a": null, "b": -5, "c": "foo", "d": null, "e": true}\n'
                b'{"a": 4.5, "b": null, "c": "nan", "d": null,"e": false}\n')
        table = self.read_bytes(rows)
        schema = pa.schema([('a', pa.float64()),
                            ('b', pa.int64()),
                            ('c', pa.string()),
                            ('d', pa.null()),
                            ('e', pa.bool_())])
        assert table.schema == schema
        assert table.to_pydict() == {
            'a': [1.0, None, 4.5],
            'b': [2, -5, None],
            'c': [None, "foo", "nan"],
            'd': [None, None, None],
            'e': [None, True, False],
        }

    def test_empty_lists(self):
        # ARROW-10955: Infer list(null)
        rows = b'{"a": []}'
        table = self.read_bytes(rows)
        schema = pa.schema([('a', pa.list_(pa.null()))])
        assert table.schema == schema
        assert table.to_pydict() == {'a': [[]]}

    def test_empty_rows(self):
        rows = b'{}\n{}\n'
        table = self.read_bytes(rows)
        schema = pa.schema([])
        assert table.schema == schema
        assert table.num_columns == 0
        assert table.num_rows == 2

    def test_explicit_schema_decimal(self):
        rows = (b'{"a": 1}\n'
                b'{"a": 1.45}\n'
                b'{"a": -23.456}\n'
                b'{}\n')
        expected = {
            'a': [Decimal("1"), Decimal("1.45"), Decimal("-23.456"), None],
        }

        decimal_types = (pa.decimal32, pa.decimal64, pa.decimal128, pa.decimal256)
        for type_factory in decimal_types:
            schema = pa.schema([('a', type_factory(9, 4))])
            opts = ParseOptions(explicit_schema=schema)
            table = self.read_bytes(rows, parse_options=opts)
            assert table.schema == schema
            assert table.to_pydict() == expected

    def test_explicit_schema_with_unexpected_behaviour(self):
        # infer by default
        rows = (b'{"foo": "bar", "num": 0}\n'
                b'{"foo": "baz", "num": 1}\n')
        schema = pa.schema([
            ('foo', pa.binary())
        ])

        opts = ParseOptions(explicit_schema=schema)
        table = self.read_bytes(rows, parse_options=opts)
        assert table.schema == pa.schema([
            ('foo', pa.binary()),
            ('num', pa.int64())
        ])
        assert table.to_pydict() == {
            'foo': [b'bar', b'baz'],
            'num': [0, 1],
        }

        # ignore the unexpected fields
        opts = ParseOptions(explicit_schema=schema,
                            unexpected_field_behavior="ignore")
        table = self.read_bytes(rows, parse_options=opts)
        assert table.schema == pa.schema([
            ('foo', pa.binary()),
        ])
        assert table.to_pydict() == {
            'foo': [b'bar', b'baz'],
        }

        # raise error
        opts = ParseOptions(explicit_schema=schema,
                            unexpected_field_behavior="error")
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error: unexpected field"):
            self.read_bytes(rows, parse_options=opts)

    @pytest.mark.numpy
    def test_small_random_json(self):
        data, expected = make_random_json(num_cols=2, num_rows=10)
        table = self.read_bytes(data)
        assert table.schema == expected.schema
        assert table.equals(expected)
        assert table.to_pydict() == expected.to_pydict()

    @pytest.mark.numpy
    def test_load_large_json(self):
        data, expected = make_random_json(num_cols=2, num_rows=100100)
        # set block size is 10MB
        read_options = ReadOptions(block_size=1024*1024*10)
        table = self.read_bytes(data, read_options=read_options)
        assert table.num_rows == 100100
        assert expected.num_rows == 100100

    @pytest.mark.numpy
    def test_stress_block_sizes(self):
        # Test a number of small block sizes to stress block stitching
        data_base, expected = make_random_json(num_cols=2, num_rows=100)
        read_options = ReadOptions()
        parse_options = ParseOptions()

        for data in [data_base, data_base.rstrip(b'\r\n')]:
            for newlines_in_values in [False, True]:
                parse_options.newlines_in_values = newlines_in_values
                for block_size in [22, 23, 37]:
                    read_options.block_size = block_size
                    table = self.read_bytes(data, read_options=read_options,
                                            parse_options=parse_options)
                    assert table.schema == expected.schema
                    if not table.equals(expected):
                        # Better error output
                        assert table.to_pydict() == expected.to_pydict()


class BaseTestJSONRead(BaseTestJSON):

    def read_bytes(self, b, **kwargs):
        return self.read_json(pa.py_buffer(b), **kwargs)

    def test_file_object(self):
        data = b'{"a": 1, "b": 2}\n'
        expected_data = {'a': [1], 'b': [2]}
        bio = io.BytesIO(data)
        table = self.read_json(bio)
        assert table.to_pydict() == expected_data
        # Text files not allowed
        sio = io.StringIO(data.decode())
        with pytest.raises(TypeError):
            self.read_json(sio)

    def test_reconcile_across_blocks(self):
        # ARROW-12065: reconciling inferred types across blocks
        first_row = b'{                               }\n'
        read_options = ReadOptions(block_size=len(first_row))
        for next_rows, expected_pylist in [
            (b'{"a": 0}', [None, 0]),
            (b'{"a": []}', [None, []]),
            (b'{"a": []}\n{"a": [[1]]}', [None, [], [[1]]]),
            (b'{"a": {}}', [None, {}]),
            (b'{"a": {}}\n{"a": {"b": {"c": 1}}}',
             [None, {"b": None}, {"b": {"c": 1}}]),
        ]:
            table = self.read_bytes(first_row + next_rows,
                                    read_options=read_options)
            expected = {"a": expected_pylist}
            assert table.to_pydict() == expected
            # Check that the issue was exercised
            assert table.column("a").num_chunks > 1


class BaseTestStreamingJSONRead(BaseTestJSON):
    def open_json(self, json, *args, **kwargs):
        """
        Reads the JSON file into memory using pyarrow's open_json
        json The JSON bytes
        args Positional arguments to be forwarded to pyarrow's open_json
        kwargs Keyword arguments to be forwarded to pyarrow's open_json
        """
        read_options = kwargs.setdefault('read_options', ReadOptions())
        read_options.use_threads = self.use_threads
        return open_json(json, *args, **kwargs)

    def open_bytes(self, b, **kwargs):
        return self.open_json(pa.py_buffer(b), **kwargs)

    def check_reader(self, reader, expected_schema, expected_data):
        assert reader.schema == expected_schema
        batches = list(reader)
        assert len(batches) == len(expected_data)
        for batch, expected_batch in zip(batches, expected_data):
            batch.validate(full=True)
            assert batch.schema == expected_schema
            assert batch.to_pydict() == expected_batch

    def read_bytes(self, b, **kwargs):
        return self.open_bytes(b, **kwargs).read_all()

    def test_file_object(self):
        data = b'{"a": 1, "b": 2}\n'
        expected_data = {'a': [1], 'b': [2]}
        bio = io.BytesIO(data)
        reader = self.open_json(bio)
        expected_schema = pa.schema([('a', pa.int64()),
                                     ('b', pa.int64())])
        self.check_reader(reader, expected_schema, [expected_data])

    def test_bad_first_chunk(self):
        bad_first_chunk = b'{"i": 0            }\n{"i": 1}'
        read_options = ReadOptions()
        read_options.block_size = 3
        with pytest.raises(
            pa.ArrowInvalid,
            match="straddling object straddles two block boundaries*"
        ):
            self.open_bytes(bad_first_chunk, read_options=read_options)

    def test_bad_middle_chunk(self):
        bad_middle_chunk = b'{"i": 0}\n{"i":     1}\n{"i": 2}'
        read_options = ReadOptions()
        read_options.block_size = 10
        expected_schema = pa.schema([('i', pa.int64())])

        reader = self.open_bytes(bad_middle_chunk, read_options=read_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == {
            'i': [0]
        }
        with pytest.raises(
            pa.ArrowInvalid,
            match="straddling object straddles two block boundaries*"
        ):
            reader.read_next_batch()

        with pytest.raises(StopIteration):
            reader.read_next_batch()

    def test_bad_first_parse(self):
        bad_first_block = b'{"n": }\n{"n": 10000}'
        read_options = ReadOptions()
        read_options.block_size = 16
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error: Invalid value.*"):
            self.open_bytes(bad_first_block, read_options=read_options)

    def test_bad_middle_parse_after_empty(self):
        bad_first_block = b'{            }{"n": }\n{"n": 10000}'
        read_options = ReadOptions()
        read_options.block_size = 16
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error: Invalid value.*"):
            self.open_bytes(bad_first_block, read_options=read_options)

    def test_bad_middle_parse(self):
        bad_middle_chunk = b'{"n": 1000}\n{"n": 200 00}\n{"n": 3000}'
        read_options = ReadOptions()
        read_options.block_size = 10
        expected_schema = pa.schema([('n', pa.int64())])

        reader = self.open_bytes(bad_middle_chunk, read_options=read_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == {
            'n': [1000]
        }
        with pytest.raises(
            pa.ArrowInvalid,
            match="JSON parse error:\
 Missing a comma or '}' after an object member*"
        ):
            reader.read_next_batch()

        with pytest.raises(StopIteration):
            reader.read_next_batch()

    def test_non_linewise_chunker_first_block(self):
        bad_middle_chunk = b'{"n": 0}{1}\n{"n": 2}'
        read_options = ReadOptions(block_size=10)
        parse_options = ParseOptions(newlines_in_values=True)
        expected_schema = pa.schema([('n', pa.int64())])

        reader = self.open_bytes(
            bad_middle_chunk,
            read_options=read_options,
            parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == {
            'n': [0]
        }
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error *"):
            reader.read_next_batch()

        with pytest.raises(StopIteration):
            reader.read_next_batch()

    def test_non_linewise_chunker_bad_first_block(self):
        bad_middle_chunk = b'{"n": 0}{1}\n{"n": 2}'
        read_options = ReadOptions(block_size=10)
        parse_options = ParseOptions(newlines_in_values=True)
        expected_schema = pa.schema([('n', pa.int64())])

        reader = self.open_bytes(
            bad_middle_chunk,
            read_options=read_options,
            parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == {
            'n': [0]
        }
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error *"):
            reader.read_next_batch()

        with pytest.raises(StopIteration):
            reader.read_next_batch()

    def test_non_linewise_chunker_bad_middle_block(self):
        bad_middle_chunk = b'{"n": 0}\n{"n":    1}\n{}"n":2}\n{"n": 3}'
        read_options = ReadOptions(block_size=10)
        parse_options = ParseOptions(newlines_in_values=True)
        expected_schema = pa.schema([('n', pa.int64())])

        reader = self.open_bytes(
            bad_middle_chunk,
            read_options=read_options,
            parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == {
            'n': [0]
        }
        assert reader.read_next_batch().to_pydict() == {
            'n': [1]
        }

        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error *"):
            reader.read_next_batch()

        with pytest.raises(StopIteration):
            reader.read_next_batch()

    def test_ignore_leading_empty_blocks(self):
        leading_empty_chunk = b'    \n{"b": true, "s": "foo"}'
        explicit_schema = pa.schema([
            ('b', pa.bool_()),
            ('s', pa.utf8())
        ])
        read_options = ReadOptions(block_size=24)
        parse_options = ParseOptions(explicit_schema=explicit_schema)
        expected_data = {
            'b': [True], 's': ["foo"]
        }

        reader = self.open_bytes(
            leading_empty_chunk,
            read_options=read_options,
            parse_options=parse_options)
        self.check_reader(reader, explicit_schema, [expected_data])

    def test_inference(self):
        rows = b'{"a": 0, "b": "foo"    }\n\
        {"a": 1, "c": true  }\n{"a": 2, "d": 4.0}'
        expected_schema = pa.schema([
            ('a', pa.int64()),
            ('b', pa.utf8())
        ])
        expected_data = {'a': [0], 'b': ["foo"]}

        read_options = ReadOptions(block_size=32)
        parse_options = ParseOptions(unexpected_field_behavior="infer")
        reader = self.open_bytes(
            rows,
            read_options=read_options,
            parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == expected_data
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error: unexpected field"):
            reader.read_next_batch()

        expected_schema = pa.schema([
            ('a', pa.int64()),
            ('b', pa.utf8()),
            ('c', pa.bool_()),
        ])
        expected_data = {'a': [0, 1], 'b': ["foo", None], 'c': [None, True]}
        read_options = ReadOptions(block_size=64)
        reader = self.open_bytes(rows, read_options=read_options,
                                 parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == expected_data
        with pytest.raises(pa.ArrowInvalid,
                           match="JSON parse error: unexpected field"):
            reader.read_next_batch()

        expected_schema = pa.schema([
            ('a', pa.int64()),
            ('b', pa.utf8()),
            ('c', pa.bool_()),
            ('d', pa.float64()),
        ])
        expected_data = {'a': [0, 1, 2], 'b': ["foo", None, None],
                         'c': [None, True, None], 'd': [None, None, 4.0]}
        read_options = ReadOptions(block_size=96)
        reader = self.open_bytes(rows, read_options=read_options,
                                 parse_options=parse_options)
        assert reader.schema == expected_schema
        assert reader.read_next_batch().to_pydict() == expected_data


class TestSerialJSONRead(BaseTestJSONRead, unittest.TestCase):

    def read_json(self, *args, **kwargs):
        read_options = kwargs.setdefault('read_options', ReadOptions())
        read_options.use_threads = False
        table = read_json(*args, **kwargs)
        table.validate(full=True)
        return table


class TestParallelJSONRead(BaseTestJSONRead, unittest.TestCase):

    def read_json(self, *args, **kwargs):
        read_options = kwargs.setdefault('read_options', ReadOptions())
        read_options.use_threads = True
        table = read_json(*args, **kwargs)
        table.validate(full=True)
        return table


class TestSerialStreamingJSONRead(BaseTestStreamingJSONRead, unittest.TestCase):

    use_threads = False


@pytest.mark.threading
class TestThreadedStreamingJSONRead(BaseTestStreamingJSONRead, unittest.TestCase):

    use_threads = True