
"""simple docstring"""



import os
import posixpath
import uuid
from dataclasses import dataclass
from typing import TYPE_CHECKING, Iterable, List, Optional, Tuple, Union

import numpy as np
import pyarrow as pa

import datasets
from datasets.arrow_writer import ArrowWriter, ParquetWriter
from datasets.config import MAX_SHARD_SIZE
from datasets.filesystems import (
    is_remote_filesystem,
    rename,
)
from datasets.iterable_dataset import _BaseExamplesIterable
from datasets.utils.py_utils import convert_file_size_to_int


_snake_case          =						datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
						import pyspark
@dataclass
class        _SCREAMING_SNAKE_CASE  (  datasets.BuilderConfig							):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:							Optional[datasets.Features]							=					None






def 	__snake_case  (    SCREAMING_SNAKE_CASE:					"pyspark.sql.DataFrame"     ,  SCREAMING_SNAKE_CASE:					List[int]     ,  ):






							"""simple docstring"""



							import pyspark

							def generate_fn():
														_lowerCAmelCase  	=			df.select('*'     ,  pyspark.sql.functions.spark_partition_id().alias('part_id'						)						)
														for partition_id in partition_order:
																					_lowerCAmelCase  	=			df_with_partition_id.select('*'						).where(f"""part_id = {partition_id}"""						).drop('part_id'						)
																					_lowerCAmelCase  	=			partition_df.collect()
																					_lowerCAmelCase  	=			0
																					for row in rows:
																												yield f"""{partition_id}_{row_id}""", row.asDict()
																												row_id += 1

							return generate_fn
class        _SCREAMING_SNAKE_CASE  (  _BaseExamplesIterable							):






				'''simple docstring'''



				def __init__(					self	:			Union[str, Any]				,     UpperCAmelCase_	:			"pyspark.sql.DataFrame"				,     UpperCAmelCase_	:			int=None				,     )		->  List[str]:







											"""simple docstring"""




											_lowerCAmelCase  	=			df
											_lowerCAmelCase  	=			partition_order or range(self.df.rdd.getNumPartitions()  )
											_lowerCAmelCase  	=			_generate_iterable_examples(self.df				,     self.partition_order  )


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







											"""simple docstring"""




											yield from self.generate_examples_fn()


				def 				__lowerCamelCase	(					self	:			Union[str, Any]				,     UpperCAmelCase_	:			np.random.Generator  )		->  "SparkExamplesIterable":







											"""simple docstring"""




											_lowerCAmelCase  	=			list(range(self.df.rdd.getNumPartitions()  )  )
											generator.shuffle(UpperCAmelCase_  )
											return SparkExamplesIterable(self.df				,     partition_order=UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			Dict				,     UpperCAmelCase_	:			int				,     UpperCAmelCase_	:			int  )		->  "SparkExamplesIterable":







											"""simple docstring"""




											_lowerCAmelCase  	=			self.split_shard_indices_by_worker(UpperCAmelCase_				,     UpperCAmelCase_  )
											return SparkExamplesIterable(self.df				,     partition_order=UpperCAmelCase_  )







				@property
				def 				__lowerCamelCase	(					self	:			Dict  )		->  int:







											"""simple docstring"""




											return len(self.partition_order  )






class        _SCREAMING_SNAKE_CASE  (  datasets.DatasetBuilder							):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:							str							=					SparkConfig


				def __init__(					self	:			Dict				,     UpperCAmelCase_	:			"pyspark.sql.DataFrame"				,     UpperCAmelCase_	:			str = None				,     UpperCAmelCase_	:			str = None				,     **UpperCAmelCase_	:			str				,     )		->  str:







											"""simple docstring"""




											import pyspark

											_lowerCAmelCase  	=			pyspark.sql.SparkSession.builder.getOrCreate()
											_lowerCAmelCase  	=			df
											_lowerCAmelCase  	=			working_dir

											super().__init__(
											    cache_dir=UpperCAmelCase_				,     config_name=str(self.df.semanticHash()  )				,     **UpperCAmelCase_				,     )


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







											"""simple docstring"""




											def create_cache_and_write_probe(UpperCAmelCase_	:			Tuple  ):
																		# makedirs with exist_ok will recursively create the directory. It will not throw an error if directories
																		# already exist.
																		os.makedirs(self._cache_dir				,     exist_ok=UpperCAmelCase_  )
																		_lowerCAmelCase  	=			os.path.join(self._cache_dir				,     'fs_test' + uuid.uuida().hex  )
																		# Opening the file in append mode will create a new file unless it already exists, in which case it will not
																		# change the file contents.
																		open(UpperCAmelCase_				,     'a'  )
																		return [probe_file]

											if self._spark.conf.get('spark.master'				,     ''  ).startswith('local'  ):
																		return

											# If the cluster is multi-node, make sure that the user provided a cache_dir and that it is on an NFS
											# accessible to the driver.
											# TODO: Stream batches to the driver using ArrowCollectSerializer instead of throwing an error.
											if self._cache_dir:
																		_lowerCAmelCase  	=			(
																		    self._spark.sparkContext.parallelize(range(1  )				,     1  ).mapPartitions(UpperCAmelCase_  ).collect()
																		)
																		if os.path.isfile(probe[0]  ):
																									return

											raise ValueError(
											    'When using Dataset.from_spark on a multi-node cluster, the driver and all workers should be able to access cache_dir'  )


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







											"""simple docstring"""




											return datasets.DatasetInfo(features=self.config.features  )


				def 				__lowerCamelCase	(					self	:			Dict				,     UpperCAmelCase_	:			datasets.download.download_manager.DownloadManager  )		->  Dict:







											"""simple docstring"""




											return [datasets.SplitGenerator(name=datasets.Split.TRAIN  )]


				def 				__lowerCamelCase	(					self	:			Any				,     UpperCAmelCase_	:			Optional[int]  )		->  Optional[int]:







											"""simple docstring"""




											import pyspark

											def get_arrow_batch_size(UpperCAmelCase_	:			Union[str, Any]  ):
																		for batch in it:
																									yield pa.RecordBatch.from_pydict({'batch_bytes': [batch.nbytes]}  )

											_lowerCAmelCase  	=			self.df.count()
											_lowerCAmelCase  	=			df_num_rows if df_num_rows <= 100 else 100
											# Approximate the size of each row (in Arrow format) by averaging over a max-100-row sample.
											_lowerCAmelCase  	=			(
											    self.df.limit(UpperCAmelCase_  )
											    .repartition(1  )
											    .mapInArrow(UpperCAmelCase_				,     'batch_bytes: long'  )
											    .agg(pyspark.sql.functions.sum('batch_bytes'  ).alias('sample_bytes'  )  )
											    .collect()[0]
											    .sample_bytes
											    / sample_num_rows
											)
											_lowerCAmelCase  	=			approx_bytes_per_row * df_num_rows
											if approx_total_size > max_shard_size:
																		# Make sure there is at least one row per partition.
																		_lowerCAmelCase  	=			min(UpperCAmelCase_				,     int(approx_total_size / max_shard_size  )  )
																		_lowerCAmelCase  	=			self.df.repartition(UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			Any				,     UpperCAmelCase_	:			str				,     UpperCAmelCase_	:			str				,     UpperCAmelCase_	:			int				,     )		->  Iterable[Tuple[int, bool, Union[int, tuple]]]:







											"""simple docstring"""




											import pyspark

											_lowerCAmelCase  	=			ParquetWriter if file_format == 'parquet' else ArrowWriter
											_lowerCAmelCase  	=			os.path.join(self._working_dir				,     os.path.basename(UpperCAmelCase_  )  ) if self._working_dir else fpath
											_lowerCAmelCase  	=			file_format == 'parquet'

											# Define these so that we don't reference self in write_arrow, which will result in a pickling error due to
											# pickling the SparkContext.
											_lowerCAmelCase  	=			self.config.features
											_lowerCAmelCase  	=			self._writer_batch_size
											_lowerCAmelCase  	=			self._fs.storage_options

											def write_arrow(UpperCAmelCase_	:			Any  ):
																		# Within the same SparkContext, no two task attempts will share the same attempt ID.
																		_lowerCAmelCase  	=			pyspark.TaskContext().taskAttemptId()
																		_lowerCAmelCase  	=			next(UpperCAmelCase_				,     UpperCAmelCase_  )
																		if first_batch is None:
																									# Some partitions might not receive any data.
																									return pa.RecordBatch.from_arrays(
																									    [[task_id], [0], [0]]				,     names=['task_id', 'num_examples', 'num_bytes']				,     )
																		_lowerCAmelCase  	=			0
																		_lowerCAmelCase  	=			writer_class(
																		    features=UpperCAmelCase_				,     path=working_fpath.replace('SSSSS'				,     F"""{shard_id:05d}"""  ).replace('TTTTT'				,     F"""{task_id:05d}"""  )				,     writer_batch_size=UpperCAmelCase_				,     storage_options=UpperCAmelCase_				,     embed_local_files=UpperCAmelCase_				,     )
																		_lowerCAmelCase  	=			pa.Table.from_batches([first_batch]  )
																		writer.write_table(UpperCAmelCase_  )
																		for batch in it:
																									if max_shard_size is not None and writer._num_bytes >= max_shard_size:
																																_lowerCAmelCase	,		_lowerCAmelCase  	=			writer.finalize()
																																writer.close()
																																yield pa.RecordBatch.from_arrays(
																																    [[task_id], [num_examples], [num_bytes]]				,     names=['task_id', 'num_examples', 'num_bytes']				,     )
																																shard_id += 1
																																_lowerCAmelCase  	=			writer_class(
																																    features=writer._features				,     path=working_fpath.replace('SSSSS'				,     F"""{shard_id:05d}"""  ).replace('TTTTT'				,     F"""{task_id:05d}"""  )				,     writer_batch_size=UpperCAmelCase_				,     storage_options=UpperCAmelCase_				,     embed_local_files=UpperCAmelCase_				,     )
																									_lowerCAmelCase  	=			pa.Table.from_batches([batch]  )
																									writer.write_table(UpperCAmelCase_  )

																		if writer._num_bytes > 0:
																									_lowerCAmelCase	,		_lowerCAmelCase  	=			writer.finalize()
																									writer.close()
																									yield pa.RecordBatch.from_arrays(
																									    [[task_id], [num_examples], [num_bytes]]				,     names=['task_id', 'num_examples', 'num_bytes']				,     )

																		if working_fpath != fpath:
																									for file in os.listdir(os.path.dirname(UpperCAmelCase_  )  ):
																																_lowerCAmelCase  	=			os.path.join(os.path.dirname(UpperCAmelCase_  )				,     os.path.basename(UpperCAmelCase_  )  )
																																shutil.move(UpperCAmelCase_				,     UpperCAmelCase_  )

											_lowerCAmelCase  	=			(
											    self.df.mapInArrow(UpperCAmelCase_				,     'task_id: long, num_examples: long, num_bytes: long'  )
											    .groupBy('task_id'  )
											    .agg(
											        pyspark.sql.functions.sum('num_examples'  ).alias('total_num_examples'  )				,     pyspark.sql.functions.sum('num_bytes'  ).alias('total_num_bytes'  )				,     pyspark.sql.functions.count('num_bytes'  ).alias('num_shards'  )				,     pyspark.sql.functions.collect_list('num_examples'  ).alias('shard_lengths'  )				,     )
											    .collect()
											)
											for row in stats:
																		yield row.task_id, (row.total_num_examples, row.total_num_bytes, row.num_shards, row.shard_lengths)


				def 				__lowerCamelCase	(					self	:			Any				,     UpperCAmelCase_	:			"datasets.SplitGenerator"				,     UpperCAmelCase_	:			str = "arrow"				,     UpperCAmelCase_	:			Optional[Union[str, int]] = None				,     UpperCAmelCase_	:			Optional[int] = None				,     **UpperCAmelCase_	:			Dict				,     )		->  Optional[int]:







											"""simple docstring"""




											self._validate_cache_dir()

											_lowerCAmelCase  	=			convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE  )
											self._repartition_df_if_needed(UpperCAmelCase_  )
											_lowerCAmelCase  	=			not is_remote_filesystem(self._fs  )
											_lowerCAmelCase  	=			os.path.join if is_local else posixpath.join

											_lowerCAmelCase  	=			'-TTTTT-SSSSS-of-NNNNN'
											_lowerCAmelCase  	=			F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
											_lowerCAmelCase  	=			path_join(self._output_dir				,     UpperCAmelCase_  )

											_lowerCAmelCase  	=			0
											_lowerCAmelCase  	=			0
											_lowerCAmelCase  	=			0
											_lowerCAmelCase  	=			[]
											_lowerCAmelCase  	=			[]

											for task_id, content in self._prepare_split_single(UpperCAmelCase_				,     UpperCAmelCase_				,     UpperCAmelCase_  ):
																		(
																		    (
																		    _lowerCAmelCase		
																		)	,		(
																		    _lowerCAmelCase		
																		)	,		(
																		    _lowerCAmelCase		
																		)	,		(
																		    _lowerCAmelCase		
																		)	,				
																		)  	=			content
																		if num_bytes > 0:
																									total_num_examples += num_examples
																									total_num_bytes += num_bytes
																									total_shards += num_shards
																									task_id_and_num_shards.append((task_id, num_shards)  )
																									all_shard_lengths.extend(UpperCAmelCase_  )

											_lowerCAmelCase  	=			total_num_examples
											_lowerCAmelCase  	=			total_num_bytes

											# should rename everything at the end
											logger.debug(F"""Renaming {total_shards} shards."""  )
											if total_shards > 1:
																		_lowerCAmelCase  	=			all_shard_lengths

																		# Define fs outside of _rename_shard so that we don't reference self in the function, which will result in a
																		# pickling error due to pickling the SparkContext.
																		_lowerCAmelCase  	=			self._fs

																		# use the -SSSSS-of-NNNNN pattern
																		def _rename_shard(
																		    UpperCAmelCase_	:			int				,     UpperCAmelCase_	:			int				,     UpperCAmelCase_	:			int				,     ):
																									rename(
																									    UpperCAmelCase_				,     fpath.replace('SSSSS'				,     F"""{shard_id:05d}"""  ).replace('TTTTT'				,     F"""{task_id:05d}"""  )				,     fpath.replace('TTTTT-SSSSS'				,     F"""{global_shard_id:05d}"""  ).replace('NNNNN'				,     F"""{total_shards:05d}"""  )				,     )

																		_lowerCAmelCase  	=			[]
																		_lowerCAmelCase  	=			0
																		for i in range(len(UpperCAmelCase_  )  ):
																									_lowerCAmelCase	,		_lowerCAmelCase  	=			task_id_and_num_shards[i]
																									for shard_id in range(UpperCAmelCase_  ):
																																args.append([task_id, shard_id, global_shard_id]  )
																																global_shard_id += 1
																		self._spark.sparkContext.parallelize(UpperCAmelCase_				,     len(UpperCAmelCase_  )  ).map(lambda UpperCAmelCase_  : _rename_shard(*UpperCAmelCase_  )  ).collect()
											else:
																		# don't use any pattern
																		_lowerCAmelCase  	=			0
																		_lowerCAmelCase  	=			task_id_and_num_shards[0][0]
																		self._rename(
																		    fpath.replace('SSSSS'				,     F"""{shard_id:05d}"""  ).replace('TTTTT'				,     F"""{task_id:05d}"""  )				,     fpath.replace(UpperCAmelCase_				,     ''  )				,     )







				def 				__lowerCamelCase	(					self	:			int				,     UpperCAmelCase_	:			"datasets.SplitGenerator"				,     )		->  SparkExamplesIterable:







											"""simple docstring"""




											return SparkExamplesIterable(self.df  )








"""simple docstring"""



import unittest

from datasets import load_dataset

from transformers import BloomTokenizerFast
from transformers.testing_utils import require_tokenizers

from ...test_tokenization_common import TokenizerTesterMixin
@require_tokenizers
class        _SCREAMING_SNAKE_CASE  (  UpperCAmelCase      ,				unittest.TestCase							):






				'''simple docstring'''



				SCREAMING_SNAKE_CASE_:							str							=					None
				SCREAMING_SNAKE_CASE_:							List[str]							=					BloomTokenizerFast
				SCREAMING_SNAKE_CASE_:							List[str]							=					BloomTokenizerFast
				SCREAMING_SNAKE_CASE_:							Any							=					True
				SCREAMING_SNAKE_CASE_:							List[Any]							=					False
				SCREAMING_SNAKE_CASE_:							Optional[int]							=					"tokenizer_file"
				SCREAMING_SNAKE_CASE_:							str							=					{"bos_token": "<s>", "eos_token": "</s>", "unk_token": "<unk>", "pad_token": "<pad>"}


				def 				__lowerCamelCase	(					self	:			Tuple  )		->  str:







											"""simple docstring"""




											super().setUp()
											_lowerCAmelCase  	=			BloomTokenizerFast.from_pretrained('bigscience/tokenizer'  )
											tokenizer.save_pretrained(self.tmpdirname  )


				def 				__lowerCamelCase	(					self	:			List[str]				,     **UpperCAmelCase_	:			Any  )		->  Optional[Any]:







											"""simple docstring"""




											kwargs.update(self.special_tokens_map  )
											return BloomTokenizerFast.from_pretrained(self.tmpdirname				,     **UpperCAmelCase_  )


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







											"""simple docstring"""




											_lowerCAmelCase  	=			self.get_rust_tokenizer()

											_lowerCAmelCase  	=			['The quick brown fox</s>', 'jumps over the lazy dog</s>']
											_lowerCAmelCase  	=			[[2_175, 23_714, 73_173, 144_252, 2], [77, 132_619, 3_478, 368, 109_586, 35_433, 2]]

											_lowerCAmelCase  	=			tokenizer.batch_encode_plus(UpperCAmelCase_  )['input_ids']
											self.assertListEqual(UpperCAmelCase_				,     UpperCAmelCase_  )

											_lowerCAmelCase  	=			tokenizer.batch_decode(UpperCAmelCase_  )
											self.assertListEqual(UpperCAmelCase_				,     UpperCAmelCase_  )


				def 				__lowerCamelCase	(					self	:			Any				,     UpperCAmelCase_	:			List[str]=6  )		->  Any:







											"""simple docstring"""




											for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
																		with self.subTest(F"""{tokenizer.__class__.__name__} ({pretrained_name})"""  ):
																									_lowerCAmelCase  	=			self.rust_tokenizer_class.from_pretrained(UpperCAmelCase_				,     **UpperCAmelCase_  )
																									# tokenizer_r.pad_token = None # Hotfixing padding = None
																									# Simple input
																									_lowerCAmelCase  	=			'This is a simple input'
																									_lowerCAmelCase  	=			['This is a simple input 1', 'This is a simple input 2']
																									_lowerCAmelCase  	=			('This is a simple input', 'This is a pair')
																									_lowerCAmelCase  	=			[
																									    ('This is a simple input 1', 'This is a simple input 2'),
																									    ('This is a simple pair 1', 'This is a simple pair 2'),
																									]

																									# Simple input tests
																									try:
																																tokenizer_r.encode(UpperCAmelCase_				,     max_length=UpperCAmelCase_  )
																																tokenizer_r.encode_plus(UpperCAmelCase_				,     max_length=UpperCAmelCase_  )

																																tokenizer_r.batch_encode_plus(UpperCAmelCase_				,     max_length=UpperCAmelCase_  )
																																tokenizer_r.encode(UpperCAmelCase_				,     max_length=UpperCAmelCase_  )
																																tokenizer_r.batch_encode_plus(UpperCAmelCase_				,     max_length=UpperCAmelCase_  )
																									except ValueError:
																																self.fail('Bloom Tokenizer should be able to deal with padding'  )

																									_lowerCAmelCase  	=			None  # Hotfixing padding = None
																									self.assertRaises(UpperCAmelCase_				,     tokenizer_r.encode				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'  )

																									# Simple input
																									self.assertRaises(UpperCAmelCase_				,     tokenizer_r.encode_plus				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'  )

																									# Simple input
																									self.assertRaises(
																									    UpperCAmelCase_				,     tokenizer_r.batch_encode_plus				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'				,     )

																									# Pair input
																									self.assertRaises(UpperCAmelCase_				,     tokenizer_r.encode				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'  )

																									# Pair input
																									self.assertRaises(UpperCAmelCase_				,     tokenizer_r.encode_plus				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'  )

																									# Pair input
																									self.assertRaises(
																									    UpperCAmelCase_				,     tokenizer_r.batch_encode_plus				,     UpperCAmelCase_				,     max_length=UpperCAmelCase_				,     padding='max_length'				,     )


				def 				__lowerCamelCase	(					self	:			str  )		->  Any:







											"""simple docstring"""




											_lowerCAmelCase  	=			self.get_rust_tokenizer()
											_lowerCAmelCase  	=			load_dataset('xnli'				,     'all_languages'				,     split='test'				,     streaming=UpperCAmelCase_  )

											_lowerCAmelCase  	=			next(iter(UpperCAmelCase_  )  )['premise']  # pick up one data
											_lowerCAmelCase  	=			list(sample_data.values()  )

											_lowerCAmelCase  	=			list(map(tokenizer.encode				,     UpperCAmelCase_  )  )
											_lowerCAmelCase  	=			[tokenizer.decode(UpperCAmelCase_				,     clean_up_tokenization_spaces=UpperCAmelCase_  ) for x in output_tokens]
											self.assertListEqual(UpperCAmelCase_				,     UpperCAmelCase_  )







				def 				__lowerCamelCase	(					self	:			Tuple  )		->  List[Any]:







											"""simple docstring"""




											self.assertGreaterEqual(len(self.tokenizer_class.pretrained_vocab_files_map  )				,     1  )
											self.assertGreaterEqual(len(list(self.tokenizer_class.pretrained_vocab_files_map.values()  )[0]  )				,     1  )






