
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


lowercase  :					Tuple						    =							datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
  import pyspark

@dataclass
class __snake_case		(      datasets.BuilderConfig     ):
    _a		:						Tuple=				None







def 						_snake_case(  SCREAMING_SNAKE_CASE__					,	SCREAMING_SNAKE_CASE__					,	) ->    Optional[int]:
       import pyspark

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

       return generate_fn

class __snake_case		(      _BaseExamplesIterable     ):
    def __init__(  self		,snake_case		,snake_case=None		,):



           '''simple docstring'''







           lowercase							:     Optional[Any]       	=      df
           lowercase							:     Any       	=      partition_order or range(self.df.rdd.getNumPartitions()      )
           lowercase							:     List[str]       	=      _generate_iterable_examples(self.df		,self.partition_order      )
    def __iter__(  self      ):



           '''simple docstring'''







           yield from self.generate_examples_fn()
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           lowercase							:     Union[str, Any]       	=      list(range(self.df.rdd.getNumPartitions()      )      )
           generator.shuffle(a_      )
           return SparkExamplesIterable(self.df		,partition_order=a_      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case      ):



           '''simple docstring'''







           lowercase							:     Optional[int]       	=      self.split_shard_indices_by_worker(a_		,a_      )
           return SparkExamplesIterable(self.df		,partition_order=a_      )
    @property
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           return len(self.partition_order      )






class __snake_case		(      datasets.DatasetBuilder     ):
    _a		:						str=				SparkConfig
    def __init__(  self		,snake_case		,snake_case = None		,snake_case = None		,**snake_case		,):



           '''simple docstring'''







           import pyspark

           lowercase							:     Union[str, Any]       	=      pyspark.sql.SparkSession.builder.getOrCreate()
           lowercase							:     Union[str, Any]       	=      df
           lowercase							:     List[Any]       	=      working_dir

           super().__init__(
               cache_dir=a_		,config_name=str(self.df.semanticHash()      )		,**a_		,)
    def       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           def create_cache_and_write_probe(snake_case      ):
                  # 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=a_      )
                  lowercase							:     Tuple       	=      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(a_		,"""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:
                  lowercase							:     str       	=      (
                      self._spark.sparkContext.parallelize(range(1      )		,1      ).mapPartitions(a_      ).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       _SCREAMING_SNAKE_CASE       (  self      ):



           '''simple docstring'''







           return datasets.DatasetInfo(features=self.config.features      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           return [datasets.SplitGenerator(name=datasets.Split.TRAIN      )]
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case      ):



           '''simple docstring'''







           import pyspark

           def get_arrow_batch_size(snake_case      ):
                  for batch in it:
                         yield pa.RecordBatch.from_pydict({"""batch_bytes""": [batch.nbytes]}      )

           lowercase							:     Dict       	=      self.df.count()
           lowercase							:     Optional[int]       	=      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.
           lowercase							:     Optional[Any]       	=      (
               self.df.limit(a_      )
               .repartition(1      )
               .mapInArrow(a_		,"""batch_bytes: long"""      )
               .agg(pyspark.sql.functions.sum("""batch_bytes"""      ).alias("""sample_bytes"""      )      )
               .collect()[0]
               .sample_bytes
               / sample_num_rows
           )
           lowercase							:     Optional[Any]       	=      approx_bytes_per_row * df_num_rows
           if approx_total_size > max_shard_size:
                  # Make sure there is at least one row per partition.
                  lowercase							:     Dict       	=      min(a_		,int(approx_total_size / max_shard_size      )      )
                  lowercase							:     List[Any]       	=      self.df.repartition(a_      )
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case		,snake_case		,):



           '''simple docstring'''







           import pyspark

           lowercase							:     Dict       	=      ParquetWriter if file_format == '''parquet''' else ArrowWriter
           lowercase							:     List[str]       	=      os.path.join(self._working_dir		,os.path.basename(a_      )      ) if self._working_dir else fpath
           lowercase							:     str       	=      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.
           lowercase							:     Tuple       	=      self.config.features
           lowercase							:     Dict       	=      self._writer_batch_size
           lowercase							:     Union[str, Any]       	=      self._fs.storage_options

           def write_arrow(snake_case      ):
                  # Within the same SparkContext, no two task attempts will share the same attempt ID.
                  lowercase							:     int       	=      pyspark.TaskContext().taskAttemptId()
                  lowercase							:     Any       	=      next(a_		,a_      )
                  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"""]		,)
                  lowercase							:     Dict       	=      0
                  lowercase							:     Tuple       	=      writer_class(
                      features=a_		,path=working_fpath.replace("""SSSSS"""		,f"{shard_id:05d}"      ).replace("""TTTTT"""		,f"{task_id:05d}"      )		,writer_batch_size=a_		,storage_options=a_		,embed_local_files=a_		,)
                  lowercase							:     Optional[Any]       	=      pa.Table.from_batches([first_batch]      )
                  writer.write_table(a_      )
                  for batch in it:
                         if max_shard_size is not None and writer._num_bytes >= max_shard_size:
                                lowercase							:     Tuple       	=      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
                                lowercase							:     Dict       	=      writer_class(
                                    features=writer._features		,path=working_fpath.replace("""SSSSS"""		,f"{shard_id:05d}"      ).replace("""TTTTT"""		,f"{task_id:05d}"      )		,writer_batch_size=a_		,storage_options=a_		,embed_local_files=a_		,)
                         lowercase							:     Union[str, Any]       	=      pa.Table.from_batches([batch]      )
                         writer.write_table(a_      )

                  if writer._num_bytes > 0:
                         lowercase							:     Dict       	=      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(a_      )      ):
                                lowercase							:     List[str]       	=      os.path.join(os.path.dirname(a_      )		,os.path.basename(a_      )      )
                                shutil.move(a_		,a_      )

           lowercase							:     str       	=      (
               self.df.mapInArrow(a_		,"""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       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,snake_case = "arrow"		,snake_case = None		,snake_case = None		,**snake_case		,):



           '''simple docstring'''







           self._validate_cache_dir()

           lowercase							:     Optional[int]       	=      convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE      )
           self._repartition_df_if_needed(a_      )
           lowercase							:     List[str]       	=      not is_remote_filesystem(self._fs      )
           lowercase							:     Union[str, Any]       	=      os.path.join if is_local else posixpath.join

           lowercase							:     str       	=      '''-TTTTT-SSSSS-of-NNNNN'''
           lowercase							:     List[Any]       	=      f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
           lowercase							:     Dict       	=      path_join(self._output_dir		,a_      )

           lowercase							:     Any       	=      0
           lowercase							:     Tuple       	=      0
           lowercase							:     Any       	=      0
           lowercase							:     Optional[Any]       	=      []
           lowercase							:     Dict       	=      []

           for task_id, content in self._prepare_split_single(a_		,a_		,a_      ):
                  (
                      lowercase					
                  )							:     List[Any]       	=      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(a_      )

           lowercase							:     Optional[Any]       	=      total_num_examples
           lowercase							:     Optional[Any]       	=      total_num_bytes

           # should rename everything at the end
           logger.debug(f"Renaming {total_shards} shards."      )
           if total_shards > 1:
                  lowercase							:     int       	=      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.
                  lowercase							:     List[str]       	=      self._fs

                  # use the -SSSSS-of-NNNNN pattern
                  def _rename_shard(
                      snake_case		,snake_case		,snake_case		,):
                         rename(
                             a_		,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}"      )		,)

                  lowercase							:     Tuple       	=      []
                  lowercase							:     Any       	=      0
                  for i in range(len(a_      )      ):
                         lowercase							:     Optional[int]       	=      task_id_and_num_shards[i]
                         for shard_id in range(a_      ):
                                args.append([task_id, shard_id, global_shard_id]      )
                                global_shard_id += 1
                  self._spark.sparkContext.parallelize(a_		,len(a_      )      ).map(lambda snake_case      : _rename_shard(*a_      )      ).collect()
           else:
                  # don't use any pattern
                  lowercase							:     Dict       	=      0
                  lowercase							:     Any       	=      task_id_and_num_shards[0][0]
                  self._rename(
                      fpath.replace("""SSSSS"""		,f"{shard_id:05d}"      ).replace("""TTTTT"""		,f"{task_id:05d}"      )		,fpath.replace(a_		,""""""      )		,)
    def       _SCREAMING_SNAKE_CASE       (  self		,snake_case		,):



           '''simple docstring'''







           return SparkExamplesIterable(self.df      )





import warnings
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__A   			=logging.get_logger(__name__)

class 		UpperCAmelCase__			(			__UpperCamelCase   ):




			'''simple docstring'''






			UpperCamelCase									=       ["""input_ids""", """attention_mask"""]




			def __init__(   self							: List[str]       ,       a_							: List[Any]="</s>"       ,       a_							: str="<unk>"       ,       a_							: Tuple="<pad>"       ,       a_							: List[Any]=1_25       ,       a_							: Any=None       ,       **a_							: Tuple       ,       ):
					'''simple docstring'''





					if extra_ids > 0 and additional_special_tokens is None:
							__UpperCAmelCase    :     Optional[Any]													=	[F'<extra_id_{i}>' for i in range(a_ )]
					elif extra_ids > 0 and additional_special_tokens is not None:
							# Check that we have the right number of extra_id special tokens
							__UpperCAmelCase    :     Optional[Any]													=	len(set(filter(lambda a_ : bool('''extra_id''' in str(a_ ) )       ,       a_ ) ) )
							if extra_tokens != extra_ids:
									raise ValueError(
									    F'Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are'
									    ''' provided to ByT5Tokenizer. In this case the additional_special_tokens must include the'''
									    ''' extra_ids tokens''' )

					__UpperCAmelCase    :     str													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else pad_token
					__UpperCAmelCase    :     str													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else eos_token
					__UpperCAmelCase    :     int													=	AddedToken(a_       ,       lstrip=a_       ,       rstrip=a_ ) if isinstance(a_       ,       a_ ) else unk_token

					super().__init__(
					    eos_token=a_       ,       unk_token=a_       ,       pad_token=a_       ,       extra_ids=a_       ,       additional_special_tokens=a_       ,       **a_       ,       )

					__UpperCAmelCase    :     Dict													=	extra_ids

					__UpperCAmelCase    :     Any													=	2**8  # utf is 8 bits

					# define special tokens dict
					__UpperCAmelCase    :     Dict[int, str]													=	{
					    self.pad_token: 0,
					    self.eos_token: 1,
					    self.unk_token: 2,
					}
					__UpperCAmelCase    :     Optional[int]													=	len(self.special_tokens_encoder )
					__UpperCAmelCase    :     Union[str, Any]													=	len(a_ )
					for i, token in enumerate(a_ ):
							__UpperCAmelCase    :     Union[str, Any]													=	self.vocab_size + i - n
					__UpperCAmelCase    :     Dict[str, int]													=	{v: k for k, v in self.special_tokens_encoder.items()}




			@property
			def  snake_case__     (   self							: Tuple ):
					'''simple docstring'''





					return self._utf_vocab_size + self._num_special_tokens + self._extra_ids




			def  snake_case__     (   self							: Any       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None       ,       a_							: bool = False ):
					'''simple docstring'''





					if already_has_special_tokens:
							return super().get_special_tokens_mask(
							    token_ids_a=a_       ,       token_ids_a=a_       ,       already_has_special_tokens=a_ )

					# normal case: some special tokens
					if token_ids_a is None:
							return ([0] * len(a_ )) + [1]
					return ([0] * len(a_ )) + [1] + ([0] * len(a_ )) + [1]




			def  snake_case__     (   self							: Any       ,       a_							: List[int] ):
					'''simple docstring'''





					if len(a_ ) > 0 and token_ids[-1] == self.eos_token_id:
							warnings.warn(
							    F'This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated'
							    ''' eos tokens being added.''' )
							return token_ids
					else:
							return token_ids + [self.eos_token_id]




			def  snake_case__     (   self							: Union[str, Any]       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None ):
					'''simple docstring'''





					__UpperCAmelCase    :     str													=	[self.eos_token_id]

					if token_ids_a is None:
							return len(token_ids_a + eos ) * [0]
					return len(token_ids_a + eos + token_ids_a + eos ) * [0]




			def  snake_case__     (   self							: Optional[Any]       ,       a_							: List[int]       ,       a_							: Optional[List[int]] = None ):
					'''simple docstring'''





					__UpperCAmelCase    :     Union[str, Any]													=	self._add_eos_if_not_present(a_ )
					if token_ids_a is None:
							return token_ids_a
					else:
							__UpperCAmelCase    :     List[str]													=	self._add_eos_if_not_present(a_ )
							return token_ids_a + token_ids_a




			def  snake_case__     (   self							: List[Any]       ,       a_							: str ):
					'''simple docstring'''





					__UpperCAmelCase    :     str													=	[chr(a_ ) for i in text.encode('''utf-8''' )]
					return tokens




			def  snake_case__     (   self							: Optional[Any]       ,       a_							: int ):
					'''simple docstring'''





					if token in self.special_tokens_encoder:
							__UpperCAmelCase    :     Union[str, Any]													=	self.special_tokens_encoder[token]
					elif token in self.added_tokens_encoder:
							__UpperCAmelCase    :     Optional[Any]													=	self.added_tokens_encoder[token]
					elif len(a_ ) != 1:
							__UpperCAmelCase    :     int													=	self.unk_token_id
					else:
							__UpperCAmelCase    :     List[str]													=	ord(a_ ) + self._num_special_tokens
					return token_id




			def  snake_case__     (   self							: Union[str, Any]       ,       a_							: Optional[int] ):
					'''simple docstring'''





					if index in self.special_tokens_decoder:
							__UpperCAmelCase    :     Dict													=	self.special_tokens_decoder[index]
					else:
							__UpperCAmelCase    :     List[Any]													=	chr(index - self._num_special_tokens )
					return token




			def  snake_case__     (   self							: int       ,       a_							: Dict ):
					'''simple docstring'''





					__UpperCAmelCase    :     List[Any]													=	b''''''
					for token in tokens:
							if token in self.special_tokens_decoder:
									__UpperCAmelCase    :     Optional[int]													=	self.special_tokens_decoder[token].encode('''utf-8''' )
							elif token in self.added_tokens_decoder:
									__UpperCAmelCase    :     List[str]													=	self.special_tokens_decoder[token].encode('''utf-8''' )
							elif token in self.special_tokens_encoder:
									__UpperCAmelCase    :     Tuple													=	token.encode('''utf-8''' )
							elif token in self.added_tokens_encoder:
									__UpperCAmelCase    :     int													=	token.encode('''utf-8''' )
							else:
									__UpperCAmelCase    :     Any													=	bytes([ord(a_ )] )
							bstring += tok_string
					__UpperCAmelCase    :     Dict													=	bstring.decode('''utf-8'''       ,       errors='''ignore''' )
					return string




			def  snake_case__     (   self							: Tuple       ,       a_							: str       ,       a_							: Optional[str] = None ):
					'''simple docstring'''





					return ()





