



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 		lowercase						(			snake_case__):






	"""simple docstring"""


	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			str:
							UpperCAmelCase_=       pa.array(TypedSequence([1, 2, 3]					)					)
							self.assertEqual(arr.type						,  pa.intaa()					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			Union[str, Any]:
							with self.assertRaises(__UpperCAmelCase					):
													UpperCAmelCase_=       pa.array(TypedSequence([1, 2, 3]					)						,  type=pa.intaa()					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			Optional[Any]:
							with self.assertRaises(__UpperCAmelCase					):
													UpperCAmelCase_=       pa.array(TypedSequence([1, 2, 3]						,  try_type=Value("""bool"""					)						,  type=Value("""int64"""					)					)					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[Any]					)      ->			str:
							UpperCAmelCase_=       pa.array(TypedSequence([1, 2, 3]						,  type=Value("""int32"""					)					)					)
							self.assertEqual(arr.type						,  pa.intaa()					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Optional[int]					)      ->			str:
							with self.assertRaises((TypeError, pa.lib.ArrowInvalid)					):
													UpperCAmelCase_=       pa.array(TypedSequence(["""foo""", """bar"""]						,  type=Value("""int64"""					)					)					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Tuple					)      ->			Tuple:
							UpperCAmelCase_=       pa.array(TypedSequence([1, 2, 3]						,  try_type=Value("""int32"""					)					)					)
							self.assertEqual(arr.type						,  pa.intaa()					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						List[str]					)      ->			Union[str, Any]:
							UpperCAmelCase_=       pa.array(TypedSequence(["""foo""", """bar"""]						,  try_type=Value("""int64"""					)					)					)
							self.assertEqual(arr.type						,  pa.string()					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						Union[str, Any]					)      ->			int:
							UpperCAmelCase_=       pa.array(TypedSequence([[[1, 2, 3]]]						,  type=ArrayaD((1, 3)						,  """int64"""					)					)					)
							self.assertEqual(arr.type						,  ArrayaDExtensionType((1, 3)						,  """int64"""					)					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			List[Any]:
							with self.assertRaises((TypeError, pa.lib.ArrowInvalid)					):
													UpperCAmelCase_=       pa.array(TypedSequence(["""foo""", """bar"""]						,  type=ArrayaD((1, 3)						,  """int64"""					)					)					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			Dict:
							UpperCAmelCase_=       pa.array(TypedSequence([[[1, 2, 3]]]						,  try_type=ArrayaD((1, 3)						,  """int64"""					)					)					)
							self.assertEqual(arr.type						,  ArrayaDExtensionType((1, 3)						,  """int64"""					)					)




	def   _SCREAMING_SNAKE_CASE		(     self     :						int					)      ->			Optional[int]:
							UpperCAmelCase_=       pa.array(TypedSequence(["""foo""", """bar"""]						,  try_type=ArrayaD((1, 3)						,  """int64"""					)					)					)
							self.assertEqual(arr.type						,  pa.string()					)




	@require_pil
	def   _SCREAMING_SNAKE_CASE		(     self     :						List[Any]					)      ->			str:
							import PIL.Image

							UpperCAmelCase_=       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:
													UpperCAmelCase_=       pa.array(TypedSequence([{"""path""": None, """bytes""": B"""image_bytes"""}, pil_image]						,  type=Image()					)					)
													UpperCAmelCase_,       UpperCAmelCase_=       mock_cast_to_python_objects.call_args_list[-1]
													self.assertIn("""optimize_list_casting"""						,  __UpperCAmelCase					)
													self.assertFalse(kwargs["""optimize_list_casting"""]					)



def 	__a					(   lowerCAmelCase_       : List[Any]      ,lowerCAmelCase_       : int    )				->			str:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferReader(lowerCAmelCase_    ) if isinstance(lowerCAmelCase_      ,pa.Buffer    ) else pa.memory_map(lowerCAmelCase_    )
						UpperCAmelCase_=       pa.ipc.open_stream(lowerCAmelCase_    )
						UpperCAmelCase_=       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					(   lowerCAmelCase_       : Tuple      ,lowerCAmelCase_       : Tuple    )				->			Optional[int]:






						'''simple docstring'''



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



def 	__a					(   )				->			Dict:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						UpperCAmelCase_=       Features({"""labels""": ClassLabel(names=["""neg""", """pos"""]    )}    )
						with ArrowWriter(stream=lowerCAmelCase_      ,features=lowerCAmelCase_    ) as writer:
												writer.write({"""labels""": 0}    )
												writer.write({"""labels""": 1}    )
												UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()
						assert num_examples == 2
						assert num_bytes > 0
						assert writer._schema == features.arrow_schema
						assert writer._schema.metadata == features.arrow_schema.metadata
						UpperCAmelCase_=       pa.BufferReader(output.getvalue()    )
						UpperCAmelCase_=       pa.ipc.open_stream(lowerCAmelCase_    )
						UpperCAmelCase_=       f.read_all()
						UpperCAmelCase_=       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(lowerCAmelCase_    )



@pytest.mark.parametrize("""writer_batch_size"""      ,[None, 1, 10]    )
def 	__a					(   lowerCAmelCase_       : List[Any]    )				->			str:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						with ArrowWriter(
						    stream=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_      ,hash_salt="""split_name"""      ,check_duplicates=lowerCAmelCase_      ,) as writer:
												with pytest.raises(lowerCAmelCase_    ):
																		writer.write({"""col_1""": """foo""", """col_2""": 1}      ,key=[1, 2]    )
																		UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()



@pytest.mark.parametrize("""writer_batch_size"""      ,[None, 2, 10]    )
def 	__a					(   lowerCAmelCase_       : Tuple    )				->			List[Any]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						with ArrowWriter(
						    stream=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_      ,hash_salt="""split_name"""      ,check_duplicates=lowerCAmelCase_      ,) as writer:
												with pytest.raises(lowerCAmelCase_    ):
																		writer.write({"""col_1""": """foo""", """col_2""": 1}      ,key=10    )
																		writer.write({"""col_1""": """bar""", """col_2""": 2}      ,key=10    )
																		UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()



@pytest.mark.parametrize("""writer_batch_size"""      ,[None, 2, 10]    )
def 	__a					(   lowerCAmelCase_       : Dict    )				->			List[str]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						with ArrowWriter(
						    stream=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_      ,hash_salt="""split_name"""      ,check_duplicates=lowerCAmelCase_      ,) as writer:
												writer.write({"""col_1""": """foo""", """col_2""": 1}      ,key=1    )
												writer.write({"""col_1""": """bar""", """col_2""": 2}      ,key=2    )
												UpperCAmelCase_,       UpperCAmelCase_=       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					(   lowerCAmelCase_       : int      ,lowerCAmelCase_       : Dict    )				->			Any:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						UpperCAmelCase_=       pa.schema(lowerCAmelCase_    ) if fields else None
						with ArrowWriter(stream=lowerCAmelCase_      ,schema=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_    ) as writer:
												writer.write_batch({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}    )
												writer.write_batch({"""col_1""": [], """col_2""": []}    )
												UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()
						assert num_examples == 2
						assert num_bytes > 0
						if not fields:
												UpperCAmelCase_=       {"""col_1""": pa.string(), """col_2""": pa.intaa()}
						assert writer._schema == pa.schema(lowerCAmelCase_      ,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					(   lowerCAmelCase_       : Tuple      ,lowerCAmelCase_       : Optional[int]    )				->			Dict:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						UpperCAmelCase_=       pa.schema(lowerCAmelCase_    ) if fields else None
						with ArrowWriter(stream=lowerCAmelCase_      ,schema=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_    ) as writer:
												writer.write_table(pa.Table.from_pydict({"""col_1""": ["""foo""", """bar"""], """col_2""": [1, 2]}    )    )
												UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()
						assert num_examples == 2
						assert num_bytes > 0
						if not fields:
												UpperCAmelCase_=       {"""col_1""": pa.string(), """col_2""": pa.intaa()}
						assert writer._schema == pa.schema(lowerCAmelCase_      ,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					(   lowerCAmelCase_       : Optional[int]      ,lowerCAmelCase_       : Union[str, Any]    )				->			Union[str, Any]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						UpperCAmelCase_=       pa.schema(lowerCAmelCase_    ) if fields else None
						with ArrowWriter(stream=lowerCAmelCase_      ,schema=lowerCAmelCase_      ,writer_batch_size=lowerCAmelCase_    ) 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]}    )    )
												UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()
						assert num_examples == 2
						assert num_bytes > 0
						if not fields:
												UpperCAmelCase_=       {"""col_1""": pa.string(), """col_2""": pa.intaa()}
						assert writer._schema == pa.schema(lowerCAmelCase_      ,metadata=writer._schema.metadata    )
						_check_output(output.getvalue()      ,expected_num_chunks=num_examples if writer_batch_size == 1 else 1    )



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






						'''simple docstring'''



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



def 	__a					(   lowerCAmelCase_       : Optional[Any]    )				->			str:






						'''simple docstring'''



						if pa.types.is_list(lowerCAmelCase_    ):
												return get_base_dtype(arr_type.value_type    )
						else:
												return arr_type



def 	__a					(   lowerCAmelCase_       : Any      ,lowerCAmelCase_       : int    )				->			Optional[int]:






						'''simple docstring'''



						if isinstance(lst[0]      ,lowerCAmelCase_    ):
												change_first_primitive_element_in_list(lst[0]      ,lowerCAmelCase_    )
						else:
												UpperCAmelCase_=       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					(   lowerCAmelCase_       : Any      ,lowerCAmelCase_       : str      ,lowerCAmelCase_       : Dict    )				->			Optional[int]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.array(TypedSequence(lowerCAmelCase_      ,optimized_int_type=lowerCAmelCase_    )    )
						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					(   lowerCAmelCase_       : str      ,lowerCAmelCase_       : Optional[Any]      ,lowerCAmelCase_       : Any    )				->			List[str]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.array(OptimizedTypedSequence(lowerCAmelCase_      ,col=lowerCAmelCase_    )    )
						assert get_base_dtype(arr.type    ) == expected_dtype

						# not in range
						if col != "other":
												# avoids errors due to in-place modifications
												UpperCAmelCase_=       copy.deepcopy(lowerCAmelCase_    )
												UpperCAmelCase_=       np.iinfo(expected_dtype.to_pandas_dtype()    ).max + 1
												change_first_primitive_element_in_list(lowerCAmelCase_      ,lowerCAmelCase_    )
												UpperCAmelCase_=       pa.array(OptimizedTypedSequence(lowerCAmelCase_      ,col=lowerCAmelCase_    )    )
												assert get_base_dtype(arr.type    ) == pa.intaa()



@pytest.mark.parametrize("""raise_exception"""      ,[False, True]    )
def 	__a					(   lowerCAmelCase_       : int      ,lowerCAmelCase_       : Any    )				->			str:






						'''simple docstring'''



						UpperCAmelCase_=       str(tmp_path / """dataset-train.arrow"""    )
						try:
												with ArrowWriter(path=lowerCAmelCase_    ) 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					(   lowerCAmelCase_       : Union[str, Any]    )				->			Dict:






						'''simple docstring'''



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



def 	__a					(   )				->			List[str]:






						'''simple docstring'''



						UpperCAmelCase_=       pa.BufferOutputStream()
						with ParquetWriter(stream=lowerCAmelCase_    ) as writer:
												writer.write({"""col_1""": """foo""", """col_2""": 1}    )
												writer.write({"""col_1""": """bar""", """col_2""": 2}    )
												UpperCAmelCase_,       UpperCAmelCase_=       writer.finalize()
						assert num_examples == 2
						assert num_bytes > 0
						UpperCAmelCase_=       pa.BufferReader(output.getvalue()    )
						UpperCAmelCase_=       pq.read_table(lowerCAmelCase_    )
						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					(   lowerCAmelCase_       : Any      ,lowerCAmelCase_       : List[str]    )				->			Tuple:






						'''simple docstring'''



						import PIL.Image

						UpperCAmelCase_=       str(tmp_path / """test_image_rgb.jpg"""    )
						PIL.Image.fromarray(np.zeros((5, 5)      ,dtype=np.uinta    )    ).save(lowerCAmelCase_      ,format="""png"""    )
						UpperCAmelCase_=       pa.BufferOutputStream()
						with ParquetWriter(
						    stream=lowerCAmelCase_      ,features=Features({"""image""": Image()}    )      ,embed_local_files=lowerCAmelCase_    ) as writer:
												writer.write({"""image""": image_path}    )
												writer.finalize()
						UpperCAmelCase_=       pa.BufferReader(output.getvalue()    )
						UpperCAmelCase_=       pq.read_table(lowerCAmelCase_    )
						UpperCAmelCase_=       pa_table.to_pydict()
						if embed_local_files:
												assert isinstance(out["""image"""][0]["""path"""]      ,lowerCAmelCase_    )
												with open(lowerCAmelCase_      ,"""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					(   )				->			Union[str, Any]:






						'''simple docstring'''



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









from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader






class 		lowercase						(			snake_case__):






	"""simple docstring"""


	def __init__(     self     :						int						,  __UpperCAmelCase     :						pyspark.sql.DataFrame						,  __UpperCAmelCase     :						Optional[NamedSplit] = None						,  __UpperCAmelCase     :						Optional[Features] = None						,  __UpperCAmelCase     :						bool = True						,  __UpperCAmelCase     :						str = None						,  __UpperCAmelCase     :						bool = False						,  __UpperCAmelCase     :						str = None						,  __UpperCAmelCase     :						bool = True						,  __UpperCAmelCase     :						str = "arrow"						,  **__UpperCAmelCase     :						str						,  )      ->			Dict:
							super().__init__(
							    split=__UpperCAmelCase						,  features=__UpperCAmelCase						,  cache_dir=__UpperCAmelCase						,  keep_in_memory=__UpperCAmelCase						,  streaming=__UpperCAmelCase						,  **__UpperCAmelCase						,  )
							UpperCAmelCase_=       load_from_cache_file
							UpperCAmelCase_=       file_format
							UpperCAmelCase_=       Spark(
							    df=__UpperCAmelCase						,  features=__UpperCAmelCase						,  cache_dir=__UpperCAmelCase						,  working_dir=__UpperCAmelCase						,  **__UpperCAmelCase						,  )




	def   _SCREAMING_SNAKE_CASE		(     self     :						Any					)      ->			Dict:
							if self.streaming:
													return self.builder.as_streaming_dataset(split=self.split					)
							UpperCAmelCase_=       None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
							self.builder.download_and_prepare(
							    download_mode=__UpperCAmelCase						,  file_format=self._file_format						,  )
							return self.builder.as_dataset(split=self.split					)




