import json
import os
import unittest

from transformers.models.xlm.tokenization_xlm import VOCAB_FILES_NAMES, XLMTokenizer
from transformers.testing_utils import slow

from ...test_tokenization_common import TokenizerTesterMixin





class 		_lowercase						(					snake_case_						,      unittest.TestCase       ):
 lowercase				     =   XLMTokenizer
 lowercase				     =   False

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

  """simple docstring"""





  super().setUp()

  # Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
  UpperCamelCase_					:						List[Any] =					[
      'l',
      'o',
      'w',
      'e',
      'r',
      's',
      't',
      'i',
      'd',
      'n',
      'w</w>',
      'r</w>',
      't</w>',
      'lo',
      'low',
      'er</w>',
      'low</w>',
      'lowest</w>',
      'newer</w>',
      'wider</w>',
      '<unk>',
  ]
  UpperCamelCase_					:						str =					dict(zip(snake_case   ,      range(len(snake_case			)			)			)			)
  UpperCamelCase_					:						Tuple =					['l o 123', 'lo w 1456', 'e r</w> 1789', '']

  UpperCamelCase_					:						Tuple =					os.path.join(self.tmpdirname   ,      VOCAB_FILES_NAMES['vocab_file']			)
  UpperCamelCase_					:						Any =					os.path.join(self.tmpdirname   ,      VOCAB_FILES_NAMES['merges_file']			)
  with open(self.vocab_file   ,      'w'			) as fp:
   fp.write(json.dumps(snake_case			)			)
  with open(self.merges_file   ,      'w'			) as fp:
   fp.write('\n'.join(snake_case			)			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							List[Any]   ,      snake_case	:							Union[str, Any]			)   ->						int:

  """simple docstring"""





  UpperCamelCase_					:						Tuple =					'lower newer'
  UpperCamelCase_					:						Optional[int] =					'lower newer'
  return input_text, output_text

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

  """simple docstring"""





  UpperCamelCase_					:						int =					XLMTokenizer(self.vocab_file   ,      self.merges_file			)

  UpperCamelCase_					:						List[str] =					'lower'
  UpperCamelCase_					:						Optional[int] =					['low', 'er</w>']
  UpperCamelCase_					:						Optional[Any] =					tokenizer.tokenize(snake_case			)
  self.assertListEqual(snake_case   ,      snake_case			)

  UpperCamelCase_					:						List[Any] =					tokens + ['<unk>']
  UpperCamelCase_					:						int =					[1_4, 1_5, 2_0]
  self.assertListEqual(tokenizer.convert_tokens_to_ids(snake_case			)   ,      snake_case			)


 @slow
 def 						SCREAMING_SNAKE_CASE__   ( self	:							List[Any]			)   ->						Optional[int]:

  """simple docstring"""





  UpperCamelCase_					:						int =					XLMTokenizer.from_pretrained('xlm-mlm-en-2048'			)

  UpperCamelCase_					:						int =					tokenizer.encode('sequence builders'   ,      add_special_tokens=snake_case			)
  UpperCamelCase_					:						Optional[int] =					tokenizer.encode('multi-sequence build'   ,      add_special_tokens=snake_case			)

  UpperCamelCase_					:						Tuple =					tokenizer.build_inputs_with_special_tokens(snake_case			)
  UpperCamelCase_					:						Any =					tokenizer.build_inputs_with_special_tokens(snake_case   ,      snake_case			)

  assert encoded_sentence == [0] + text + [1]
  assert encoded_pair == [0] + text + [1] + text_a + [1]




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


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

if TYPE_CHECKING:
   import pyspark





@dataclass
class 		_lowercase						(					datasets.BuilderConfig       ):
 lowercase				     =   None





def    __lowercase     (						lowerCamelCase					:     "pyspark.sql.DataFrame"  ,  lowerCamelCase					:     List[int]  ,  ):
 import pyspark

 def generate_fn():
  UpperCamelCase_					:						Dict =					df.select('*'  ,  pyspark.sql.functions.spark_partition_id().alias('part_id'	)	)
  for partition_id in partition_order:
   UpperCamelCase_					:						Tuple =					df_with_partition_id.select('*'	).where(F"part_id = {partition_id}"	).drop('part_id'	)
   UpperCamelCase_					:						Union[str, Any] =					partition_df.collect()
   UpperCamelCase_					:						Any =					0
   for row in rows:
    yield F"{partition_id}_{row_id}", row.asDict()
    row_id += 1

 return generate_fn





class 		_lowercase						(					_BaseExamplesIterable       ):

 def __init__( self	:							Optional[int]   ,      snake_case	:							"pyspark.sql.DataFrame"   ,      snake_case	:							Tuple=None   ,      )   ->						Tuple:

  """simple docstring"""





  UpperCamelCase_					:						Dict =					df
  UpperCamelCase_					:						int =					partition_order or range(self.df.rdd.getNumPartitions()			)
  UpperCamelCase_					:						Optional[Any] =					_generate_iterable_examples(self.df   ,      self.partition_order			)

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

  """simple docstring"""





  yield from self.generate_examples_fn()

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any   ,      snake_case	:							np.random.Generator			)   ->						"SparkExamplesIterable":

  """simple docstring"""





  UpperCamelCase_					:						Optional[Any] =					list(range(self.df.rdd.getNumPartitions()			)			)
  generator.shuffle(snake_case			)
  return SparkExamplesIterable(self.df   ,      partition_order=snake_case			)

 def 						SCREAMING_SNAKE_CASE__   ( self	:							int   ,      snake_case	:							int   ,      snake_case	:							int			)   ->						"SparkExamplesIterable":

  """simple docstring"""





  UpperCamelCase_					:						Tuple =					self.split_shard_indices_by_worker(snake_case   ,      snake_case			)
  return SparkExamplesIterable(self.df   ,      partition_order=snake_case			)


 @property
 def 						SCREAMING_SNAKE_CASE__   ( self	:							int			)   ->						int:

  """simple docstring"""





  return len(self.partition_order			)





class 		_lowercase						(					datasets.DatasetBuilder       ):
 lowercase				     =   SparkConfig

 def __init__( self	:							List[Any]   ,      snake_case	:							"pyspark.sql.DataFrame"   ,      snake_case	:							str = None   ,      snake_case	:							str = None   ,      **snake_case	:							Optional[Any]   ,      )   ->						List[str]:

  """simple docstring"""





  import pyspark

  UpperCamelCase_					:						List[Any] =					pyspark.sql.SparkSession.builder.getOrCreate()
  UpperCamelCase_					:						str =					df
  UpperCamelCase_					:						Tuple =					working_dir

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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Any			)   ->						Dict:

  """simple docstring"""





  def create_cache_and_write_probe(snake_case	:							str			):
   # 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=snake_case			)
   UpperCamelCase_					:						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(snake_case   ,      '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:
   UpperCamelCase_					:						Tuple =					(
       self._spark.sparkContext.parallelize(range(1			)   ,      1			).mapPartitions(snake_case			).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	:							int			)   ->						Union[str, Any]:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[Any]   ,      snake_case	:							datasets.download.download_manager.DownloadManager			)   ->						Optional[int]:

  """simple docstring"""





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

 def 						SCREAMING_SNAKE_CASE__   ( self	:							Optional[Any]   ,      snake_case	:							Optional[int]			)   ->						List[Any]:

  """simple docstring"""





  import pyspark

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

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

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

  """simple docstring"""





  import pyspark

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

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

   if writer._num_bytes > 0:
    UpperCamelCase_, UpperCamelCase_					:						str =					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(snake_case			)			):
     UpperCamelCase_					:						Dict =					os.path.join(os.path.dirname(snake_case			)   ,      os.path.basename(snake_case			)			)
     shutil.move(snake_case   ,      snake_case			)

  UpperCamelCase_					:						int =					(
      self.df.mapInArrow(snake_case   ,      '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	:							int   ,      snake_case	:							"datasets.SplitGenerator"   ,      snake_case	:							str = "arrow"   ,      snake_case	:							Optional[Union[str, int]] = None   ,      snake_case	:							Optional[int] = None   ,      **snake_case	:							Any   ,      )   ->						int:

  """simple docstring"""





  self._validate_cache_dir()

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

  UpperCamelCase_					:						Optional[int] =					'-TTTTT-SSSSS-of-NNNNN'
  UpperCamelCase_					:						Dict =					f"{self.name}-{split_generator.name}{SUFFIX}.{file_format}"
  UpperCamelCase_					:						int =					path_join(self._output_dir   ,      snake_case			)

  UpperCamelCase_					:						int =					0
  UpperCamelCase_					:						Optional[int] =					0
  UpperCamelCase_					:						Union[str, Any] =					0
  UpperCamelCase_					:						Optional[Any] =					[]
  UpperCamelCase_					:						Any =					[]

  for task_id, content in self._prepare_split_single(snake_case   ,      snake_case   ,      snake_case			):
   (
       (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), (
       UpperCamelCase_
   ), 
   )					:						Optional[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(snake_case			)

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

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

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

   UpperCamelCase_					:						Any =					[]
   UpperCamelCase_					:						Optional[int] =					0
   for i in range(len(snake_case			)			):
    UpperCamelCase_, UpperCamelCase_					:						Union[str, Any] =					task_id_and_num_shards[i]
    for shard_id in range(snake_case			):
     args.append([task_id, shard_id, global_shard_id]			)
     global_shard_id += 1
   self._spark.sparkContext.parallelize(snake_case   ,      len(snake_case			)			).map(lambda snake_case			: _rename_shard(*snake_case			)			).collect()
  else:
   # don't use any pattern
   UpperCamelCase_					:						Tuple =					0
   UpperCamelCase_					:						Optional[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(snake_case   ,      ''			)   ,      )


 def 						SCREAMING_SNAKE_CASE__   ( self	:							Tuple   ,      snake_case	:							"datasets.SplitGenerator"   ,      )   ->						SparkExamplesIterable:

  """simple docstring"""





  return SparkExamplesIterable(self.df			)



