


import copy
import os
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
import pyarrow as pa
import pyarrow.parquet as pq
import pytest

from datasets.arrow_writer import ArrowWriter, OptimizedTypedSequence, ParquetWriter, TypedSequence
from datasets.features import ArrayaD, ClassLabel, Features, Image, Value
from datasets.features.features import ArrayaDExtensionType, cast_to_python_objects
from datasets.keyhash import DuplicatedKeysError, InvalidKeyError

from .utils import require_pil


class 	_a							(      UpperCamelCase__      ):

							def 				lowerCamelCase_			(							self:     Optional[int]							)      ->       List[Any]:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence([1, 2, 3]							)							)
									self.assertEqual(arr.type   ,						pa.intaa()							)

							def 				lowerCamelCase_			(							self:     Any							)      ->       str:





									"""simple docstring"""







									with self.assertRaises(UpperCamelCase_							):
											lowercase__						=				pa.array(TypedSequence([1, 2, 3]							)   ,						type=pa.intaa()							)

							def 				lowerCamelCase_			(							self:     Union[str, Any]							)      ->       int:





									"""simple docstring"""







									with self.assertRaises(UpperCamelCase_							):
											lowercase__						=				pa.array(TypedSequence([1, 2, 3]   ,						try_type=Value('''bool'''							)   ,						type=Value('''int64'''							)							)							)

							def 				lowerCamelCase_			(							self:     str							)      ->       Optional[int]:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence([1, 2, 3]   ,						type=Value('''int32'''							)							)							)
									self.assertEqual(arr.type   ,						pa.intaa()							)

							def 				lowerCamelCase_			(							self:     Any							)      ->       List[Any]:





									"""simple docstring"""







									with self.assertRaises((TypeError, pa.lib.ArrowInvalid)							):
											lowercase__						=				pa.array(TypedSequence(['''foo''', '''bar''']   ,						type=Value('''int64'''							)							)							)

							def 				lowerCamelCase_			(							self:     Union[str, Any]							)      ->       Union[str, Any]:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence([1, 2, 3]   ,						try_type=Value('''int32'''							)							)							)
									self.assertEqual(arr.type   ,						pa.intaa()							)

							def 				lowerCamelCase_			(							self:     Union[str, Any]							)      ->       Any:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence(['''foo''', '''bar''']   ,						try_type=Value('''int64'''							)							)							)
									self.assertEqual(arr.type   ,						pa.string()							)

							def 				lowerCamelCase_			(							self:     int							)      ->       List[str]:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence([[[1, 2, 3]]]   ,						type=ArrayaD((1, 3)   ,						'''int64'''							)							)							)
									self.assertEqual(arr.type   ,						ArrayaDExtensionType((1, 3)   ,						'''int64'''							)							)

							def 				lowerCamelCase_			(							self:     str							)      ->       List[Any]:





									"""simple docstring"""







									with self.assertRaises((TypeError, pa.lib.ArrowInvalid)							):
											lowercase__						=				pa.array(TypedSequence(['''foo''', '''bar''']   ,						type=ArrayaD((1, 3)   ,						'''int64'''							)							)							)

							def 				lowerCamelCase_			(							self:     Optional[Any]							)      ->       int:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence([[[1, 2, 3]]]   ,						try_type=ArrayaD((1, 3)   ,						'''int64'''							)							)							)
									self.assertEqual(arr.type   ,						ArrayaDExtensionType((1, 3)   ,						'''int64'''							)							)

							def 				lowerCamelCase_			(							self:     Optional[int]							)      ->       Dict:





									"""simple docstring"""







									lowercase__						=				pa.array(TypedSequence(['''foo''', '''bar''']   ,						try_type=ArrayaD((1, 3)   ,						'''int64'''							)							)							)
									self.assertEqual(arr.type   ,						pa.string()							)


							@require_pil
							def 				lowerCamelCase_			(							self:     Dict							)      ->       Any:





									"""simple docstring"""







									import PIL.Image

									lowercase__						=				PIL.Image.fromarray(np.arange(10   ,						dtype=np.uinta							).reshape(2   ,						5							)							)
									with patch(
									    '''datasets.arrow_writer.cast_to_python_objects'''   ,						side_effect=UpperCamelCase_							) as mock_cast_to_python_objects:
											lowercase__						=				pa.array(TypedSequence([{'''path''': None, '''bytes''': b'''image_bytes'''}, pil_image]   ,						type=Image()							)							)
											lowercase__					,		lowercase__						=				mock_cast_to_python_objects.call_args_list[-1]
											self.assertIn('''optimize_list_casting'''   ,						UpperCamelCase_							)
											self.assertFalse(kwargs['''optimize_list_casting''']							)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferReader(SCREAMING_SNAKE_CASE					) if isinstance(SCREAMING_SNAKE_CASE						,    pa.Buffer					) else pa.memory_map(SCREAMING_SNAKE_CASE					)
		lowercase__						=				pa.ipc.open_stream(SCREAMING_SNAKE_CASE					)
		lowercase__						=				f.read_all()
		assert len(pa_table.to_batches()					) == expected_num_chunks
		assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
		del pa_table
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 1, 10]					)
@pytest.mark.parametrize(
    '''fields'''						,    [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		lowercase__						=				pa.schema(SCREAMING_SNAKE_CASE					) if fields else None
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE						,    schema=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE					) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__						=				{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(SCREAMING_SNAKE_CASE						,    metadata=writer._schema.metadata					)
		_check_output(output.getvalue()						,    expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)
def       _a			(				):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		lowercase__						=				Features({'''labels''': ClassLabel(names=['''neg''', '''pos''']					)}					)
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE						,    features=SCREAMING_SNAKE_CASE					) as writer:
				writer.write({'''labels''': 0}					)
				writer.write({'''labels''': 1}					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		assert writer._schema == features.arrow_schema
		assert writer._schema.metadata == features.arrow_schema.metadata
		lowercase__						=				pa.BufferReader(output.getvalue()					)
		lowercase__						=				pa.ipc.open_stream(SCREAMING_SNAKE_CASE					)
		lowercase__						=				f.read_all()
		lowercase__						=				pa_table.schema
		assert pa_table.num_rows == 2
		assert schema == features.arrow_schema
		assert schema.metadata == features.arrow_schema.metadata
		assert features == Features.from_arrow_schema(SCREAMING_SNAKE_CASE					)
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 1, 10]					)
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		with ArrowWriter(
		    stream=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE						,    hash_salt='''split_name'''						,    check_duplicates=SCREAMING_SNAKE_CASE						,    ) as writer:
				with pytest.raises(SCREAMING_SNAKE_CASE					):
						writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,    key=[1, 2]					)
						lowercase__					,		lowercase__						=				writer.finalize()
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 2, 10]					)
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		with ArrowWriter(
		    stream=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE						,    hash_salt='''split_name'''						,    check_duplicates=SCREAMING_SNAKE_CASE						,    ) as writer:
				with pytest.raises(SCREAMING_SNAKE_CASE					):
						writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,    key=10					)
						writer.write({'''col_1''': '''bar''', '''col_2''': 2}						,    key=10					)
						lowercase__					,		lowercase__						=				writer.finalize()
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 2, 10]					)
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		with ArrowWriter(
		    stream=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE						,    hash_salt='''split_name'''						,    check_duplicates=SCREAMING_SNAKE_CASE						,    ) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}						,    key=1					)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}						,    key=2					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		_check_output(output.getvalue()						,    expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 1, 10]					)
@pytest.mark.parametrize(
    '''fields'''						,    [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		lowercase__						=				pa.schema(SCREAMING_SNAKE_CASE					) if fields else None
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE						,    schema=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE					) as writer:
				writer.write_batch({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}					)
				writer.write_batch({'''col_1''': [], '''col_2''': []}					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__						=				{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(SCREAMING_SNAKE_CASE						,    metadata=writer._schema.metadata					)
		_check_output(output.getvalue()						,    expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 1, 10]					)
@pytest.mark.parametrize(
    '''fields'''						,    [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		lowercase__						=				pa.schema(SCREAMING_SNAKE_CASE					) if fields else None
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE						,    schema=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE					) as writer:
				writer.write_table(pa.Table.from_pydict({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}					)					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__						=				{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(SCREAMING_SNAKE_CASE						,    metadata=writer._schema.metadata					)
		_check_output(output.getvalue()						,    expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)
@pytest.mark.parametrize('''writer_batch_size'''						,    [None, 1, 10]					)
@pytest.mark.parametrize(
    '''fields'''						,    [None, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}, {'''col_1''': pa.string(), '''col_2''': pa.intaa()}]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		lowercase__						=				pa.schema(SCREAMING_SNAKE_CASE					) if fields else None
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE						,    schema=SCREAMING_SNAKE_CASE						,    writer_batch_size=SCREAMING_SNAKE_CASE					) as writer:
				writer.write_row(pa.Table.from_pydict({'''col_1''': ['''foo'''], '''col_2''': [1]}					)					)
				writer.write_row(pa.Table.from_pydict({'''col_1''': ['''bar'''], '''col_2''': [2]}					)					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		if not fields:
				lowercase__						=				{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
		assert writer._schema == pa.schema(SCREAMING_SNAKE_CASE						,    metadata=writer._schema.metadata					)
		_check_output(output.getvalue()						,    expected_num_chunks=num_examples if writer_batch_size == 1 else 1					)
def       _a			(				):


		"""simple docstring"""




		with tempfile.TemporaryDirectory() as tmp_dir:
				lowercase__						=				{'''col_1''': pa.string(), '''col_2''': pa.intaa()}
				lowercase__						=				os.path.join(SCREAMING_SNAKE_CASE						,    '''test.arrow'''					)
				with ArrowWriter(path=SCREAMING_SNAKE_CASE						,    schema=pa.schema(SCREAMING_SNAKE_CASE					)					) as writer:
						writer.write_batch({'''col_1''': ['''foo''', '''bar'''], '''col_2''': [1, 2]}					)
						lowercase__					,		lowercase__						=				writer.finalize()
				assert num_examples == 2
				assert num_bytes > 0
				assert writer._schema == pa.schema(SCREAMING_SNAKE_CASE						,    metadata=writer._schema.metadata					)
				_check_output(SCREAMING_SNAKE_CASE						,    1					)
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		if pa.types.is_list(SCREAMING_SNAKE_CASE					):
				return get_base_dtype(arr_type.value_type					)
		else:
				return arr_type
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		if isinstance(lst[0]						,    SCREAMING_SNAKE_CASE					):
				change_first_primitive_element_in_list(lst[0]						,    SCREAMING_SNAKE_CASE					)
		else:
				lowercase__						=				value
@pytest.mark.parametrize('''optimized_int_type, expected_dtype'''						,    [(None, pa.intaa()), (Value('''int32'''					), pa.intaa())]					)
@pytest.mark.parametrize('''sequence'''						,    [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.array(TypedSequence(SCREAMING_SNAKE_CASE						,    optimized_int_type=SCREAMING_SNAKE_CASE					)					)
		assert get_base_dtype(arr.type					) == expected_dtype
@pytest.mark.parametrize(
    '''col, expected_dtype'''						,    [
        ('''attention_mask''', pa.inta()),
        ('''special_tokens_mask''', pa.inta()),
        ('''token_type_ids''', pa.inta()),
        ('''input_ids''', pa.intaa()),
        ('''other''', pa.intaa()),
    ]						,    )
@pytest.mark.parametrize('''sequence'''						,    [[1, 2, 3], [[1, 2, 3]], [[[1, 2, 3]]]]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				pa.array(OptimizedTypedSequence(SCREAMING_SNAKE_CASE						,    col=SCREAMING_SNAKE_CASE					)					)
		assert get_base_dtype(arr.type					) == expected_dtype

		# not in range
		if col != "other":
				# avoids errors due to in-place modifications
				lowercase__						=				copy.deepcopy(SCREAMING_SNAKE_CASE					)
				lowercase__						=				np.iinfo(expected_dtype.to_pandas_dtype()					).max + 1
				change_first_primitive_element_in_list(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					)
				lowercase__						=				pa.array(OptimizedTypedSequence(SCREAMING_SNAKE_CASE						,    col=SCREAMING_SNAKE_CASE					)					)
				assert get_base_dtype(arr.type					) == pa.intaa()
@pytest.mark.parametrize('''raise_exception'''						,    [False, True]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				str(tmp_path / '''dataset-train.arrow'''					)
		try:
				with ArrowWriter(path=SCREAMING_SNAKE_CASE					) as writer:
						if raise_exception:
								raise pa.lib.ArrowInvalid()
						else:
								writer.stream.close()
		except pa.lib.ArrowInvalid:
				pass
		finally:
				assert writer.stream.closed
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				'''mock://dataset-train.arrow'''
		with ArrowWriter(path=SCREAMING_SNAKE_CASE						,    storage_options=mockfs.storage_options					) as writer:
				assert isinstance(writer._fs						,    type(SCREAMING_SNAKE_CASE					)					)
				assert writer._fs.storage_options == mockfs.storage_options
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		assert mockfs.exists(SCREAMING_SNAKE_CASE					)
def       _a			(				):


		"""simple docstring"""




		lowercase__						=				pa.BufferOutputStream()
		with ParquetWriter(stream=SCREAMING_SNAKE_CASE					) as writer:
				writer.write({'''col_1''': '''foo''', '''col_2''': 1}					)
				writer.write({'''col_1''': '''bar''', '''col_2''': 2}					)
				lowercase__					,		lowercase__						=				writer.finalize()
		assert num_examples == 2
		assert num_bytes > 0
		lowercase__						=				pa.BufferReader(output.getvalue()					)
		lowercase__						=				pq.read_table(SCREAMING_SNAKE_CASE					)
		assert pa_table.to_pydict() == {"col_1": ["foo", "bar"], "col_2": [1, 2]}
@require_pil
@pytest.mark.parametrize('''embed_local_files'''						,    [False, True]					)
def       _a			(				SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		import PIL.Image

		lowercase__						=				str(tmp_path / '''test_image_rgb.jpg'''					)
		PIL.Image.fromarray(np.zeros((5, 5)						,    dtype=np.uinta					)					).save(SCREAMING_SNAKE_CASE						,    format='''png'''					)
		lowercase__						=				pa.BufferOutputStream()
		with ParquetWriter(
		    stream=SCREAMING_SNAKE_CASE						,    features=Features({'''image''': Image()}					)						,    embed_local_files=SCREAMING_SNAKE_CASE					) as writer:
				writer.write({'''image''': image_path}					)
				writer.finalize()
		lowercase__						=				pa.BufferReader(output.getvalue()					)
		lowercase__						=				pq.read_table(SCREAMING_SNAKE_CASE					)
		lowercase__						=				pa_table.to_pydict()
		if embed_local_files:
				assert isinstance(out['''image'''][0]['''path''']						,    SCREAMING_SNAKE_CASE					)
				with open(SCREAMING_SNAKE_CASE						,    '''rb'''					) as f:
						assert out["image"][0]["bytes"] == f.read()
		else:
				assert out["image"][0]["path"] == image_path
				assert out["image"][0]["bytes"] is None
def       _a			(				):


		"""simple docstring"""




		lowercase__						=				pa.schema([pa.field('''col_1'''						,    pa.string()						,    nullable=SCREAMING_SNAKE_CASE					)]					)
		lowercase__						=				pa.BufferOutputStream()
		with ArrowWriter(stream=SCREAMING_SNAKE_CASE					) as writer:
				writer._build_writer(inferred_schema=SCREAMING_SNAKE_CASE					)
				assert writer._schema == pa.schema([pa.field('''col_1'''						,    pa.string()					)]					)






lowerCAmelCase	=							[sum(int(c, 10) ** 2 for c in i.__str__()) for i in range(10_0000)]
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		lowercase__						=				0
		while number:
				# Increased Speed Slightly by checking every 5 digits together.
				sum_of_digits_squared += DIGITS_SQUARED[number % 10_00_00]
				number //= 10_00_00

		return sum_of_digits_squared


# There are 2 Chains made,
# One ends with 89 with the chain member 58 being the one which when declared first,
# there will be the least number of iterations for all the members to be checked.

# The other one ends with 1 and has only one element 1.

# So 58 and 1 are chosen to be declared at the starting.

# Changed dictionary to an array to quicken the solution
lowerCAmelCase	=							[None] * 1000_0000
lowerCAmelCase	=							True
lowerCAmelCase	=							False
def       _a			(				SCREAMING_SNAKE_CASE					):


		"""simple docstring"""




		if CHAINS[number - 1] is not None:
				return CHAINS[number - 1]  # type: ignore

		lowercase__						=				chain(next_number(SCREAMING_SNAKE_CASE					)					)
		lowercase__						=				number_chain

		while number < 10_00_00_00:
				lowercase__						=				number_chain
				number *= 10

		return number_chain
def       _a			(				SCREAMING_SNAKE_CASE = 10_00_00_00					):


		"""simple docstring"""




		for i in range(1						,    SCREAMING_SNAKE_CASE					):
				if CHAINS[i] is None:
						chain(i + 1					)

		return CHAINS[:number].count(SCREAMING_SNAKE_CASE					)


if __name__ == "__main__":
			import doctest

			doctest.testmod()
			print(f"""{solution() = }""")


