


'''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


UpperCAmelCase_			:  Dict										=   datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
 import pyspark

@dataclass
class     _lowerCamelCase       (     datasets.BuilderConfig					):




 '''simple docstring'''
 __lowercase					:		Optional[datasets.Features]     				=    None


def       _lowercase							(					UpperCamelCase__							:       "pyspark.sql.DataFrame",					UpperCamelCase__							:       List[int],					):
 import pyspark

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

 return generate_fn

class     _lowerCamelCase       (     _BaseExamplesIterable					):




 '''simple docstring'''


 def __init__(   self   ,   __lowercase   ,   __lowercase=None   ,   ):

  """simple docstring"""







  __A	:		str	  =      df
  __A	:		Optional[int]	  =      partition_order or range(self.df.rdd.getNumPartitions()    )
  __A	:		str	  =      _generate_iterable_examples(self.df   ,   self.partition_order    )


 def __iter__(   self    ):

  """simple docstring"""







  yield from self.generate_examples_fn()


 def 			snake_case__ (   self   ,   __lowercase    ):

  """simple docstring"""







  __A	:		List[Any]	  =      list(range(self.df.rdd.getNumPartitions()    )    )
  generator.shuffle(__lowercase    )
  return SparkExamplesIterable(self.df   ,   partition_order=__lowercase    )


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase    ):

  """simple docstring"""







  __A	:		int	  =      self.split_shard_indices_by_worker(__lowercase   ,   __lowercase    )
  return SparkExamplesIterable(self.df   ,   partition_order=__lowercase    )







 @property
 def 			snake_case__ (   self    ):

  """simple docstring"""







  return len(self.partition_order    )


class     _lowerCamelCase       (     datasets.DatasetBuilder					):




 '''simple docstring'''
 __lowercase					:		Dict     				=    SparkConfig


 def __init__(   self   ,   __lowercase   ,   __lowercase = None   ,   __lowercase = None   ,   **__lowercase   ,   ):

  """simple docstring"""







  import pyspark

  __A	:		Dict	  =      pyspark.sql.SparkSession.builder.getOrCreate()
  __A	:		Dict	  =      df
  __A	:		Dict	  =      working_dir

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


 def 			snake_case__ (   self    ):

  """simple docstring"""







  def create_cache_and_write_probe(__lowercase    ):
   # 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=__lowercase    )
   __A	:		Optional[int]	  =      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(__lowercase   ,   '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:
   __A	:		Dict	  =      (
       self._spark.sparkContext.parallelize(range(1    )   ,   1    ).mapPartitions(__lowercase    ).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 			snake_case__ (   self    ):

  """simple docstring"""







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


 def 			snake_case__ (   self   ,   __lowercase    ):

  """simple docstring"""







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


 def 			snake_case__ (   self   ,   __lowercase    ):

  """simple docstring"""







  import pyspark

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

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


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase   ,   __lowercase   ,   ):

  """simple docstring"""







  import pyspark

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

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

   if writer._num_bytes > 0:
    __A  ,__A	:		Optional[int]	  =      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(__lowercase    )    ):
     __A	:		List[Any]	  =      os.path.join(os.path.dirname(__lowercase    )   ,   os.path.basename(__lowercase    )    )
     shutil.move(__lowercase   ,   __lowercase    )

  __A	:		Any	  =      (
      self.df.mapInArrow(__lowercase   ,   '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 			snake_case__ (   self   ,   __lowercase   ,   __lowercase = "arrow"   ,   __lowercase = None   ,   __lowercase = None   ,   **__lowercase   ,   ):

  """simple docstring"""







  self._validate_cache_dir()

  __A	:		List[Any]	  =      convert_file_size_to_int(max_shard_size or MAX_SHARD_SIZE    )
  self._repartition_df_if_needed(__lowercase    )
  __A	:		List[Any]	  =      not is_remote_filesystem(self._fs    )
  __A	:		Dict	  =      os.path.join if is_local else posixpath.join

  __A	:		str	  =      '-TTTTT-SSSSS-of-NNNNN'
  __A	:		Dict	  =      F"""{self.name}-{split_generator.name}{SUFFIX}.{file_format}"""
  __A	:		str	  =      path_join(self._output_dir   ,   __lowercase    )

  __A	:		Optional[Any]	  =      0
  __A	:		Optional[Any]	  =      0
  __A	:		int	  =      0
  __A	:		str	  =      []
  __A	:		Dict	  =      []

  for task_id, content in self._prepare_split_single(__lowercase   ,   __lowercase   ,   __lowercase    ):
   (
       (
       __A
   )  ,(
       __A
   )  ,(
       __A
   )  ,(
       __A
   )  ,
   )	:		str	  =      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(__lowercase    )

  __A	:		Optional[int]	  =      total_num_examples
  __A	:		Any	  =      total_num_bytes

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

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

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







 def 			snake_case__ (   self   ,   __lowercase   ,   ):

  """simple docstring"""







  return SparkExamplesIterable(self.df    )










'''simple docstring'''
import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_base import BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import PaddingStrategy, logging
from .tokenization_realm import RealmTokenizer


UpperCAmelCase_			:  int										=   logging.get_logger(__name__)

UpperCAmelCase_			:  int										=   {'vocab_file': 'vocab.txt', 'tokenizer_file': 'tokenizer.json'}

UpperCAmelCase_			:  List[str]										=   {
    'vocab_file': {
        'google/realm-cc-news-pretrained-embedder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-encoder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-scorer': (
            'https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/vocab.txt'
        ),
        'google/realm-cc-news-pretrained-openqa': (
            'https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/vocab.txt'
        ),
        'google/realm-orqa-nq-openqa': 'https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/vocab.txt',
        'google/realm-orqa-nq-reader': 'https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/vocab.txt',
        'google/realm-orqa-wq-openqa': 'https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/vocab.txt',
        'google/realm-orqa-wq-reader': 'https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/vocab.txt',
    },
    'tokenizer_file': {
        'google/realm-cc-news-pretrained-embedder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-embedder/resolve/main/tokenizer.jsont'
        ),
        'google/realm-cc-news-pretrained-encoder': (
            'https://huggingface.co/google/realm-cc-news-pretrained-encoder/resolve/main/tokenizer.json'
        ),
        'google/realm-cc-news-pretrained-scorer': (
            'https://huggingface.co/google/realm-cc-news-pretrained-scorer/resolve/main/tokenizer.json'
        ),
        'google/realm-cc-news-pretrained-openqa': (
            'https://huggingface.co/google/realm-cc-news-pretrained-openqa/aresolve/main/tokenizer.json'
        ),
        'google/realm-orqa-nq-openqa': (
            'https://huggingface.co/google/realm-orqa-nq-openqa/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-nq-reader': (
            'https://huggingface.co/google/realm-orqa-nq-reader/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-wq-openqa': (
            'https://huggingface.co/google/realm-orqa-wq-openqa/resolve/main/tokenizer.json'
        ),
        'google/realm-orqa-wq-reader': (
            'https://huggingface.co/google/realm-orqa-wq-reader/resolve/main/tokenizer.json'
        ),
    },
}

UpperCAmelCase_			:  str										=   {
    'google/realm-cc-news-pretrained-embedder': 5_1_2,
    'google/realm-cc-news-pretrained-encoder': 5_1_2,
    'google/realm-cc-news-pretrained-scorer': 5_1_2,
    'google/realm-cc-news-pretrained-openqa': 5_1_2,
    'google/realm-orqa-nq-openqa': 5_1_2,
    'google/realm-orqa-nq-reader': 5_1_2,
    'google/realm-orqa-wq-openqa': 5_1_2,
    'google/realm-orqa-wq-reader': 5_1_2,
}

UpperCAmelCase_			:  str										=   {
    'google/realm-cc-news-pretrained-embedder': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-encoder': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-scorer': {'do_lower_case': True},
    'google/realm-cc-news-pretrained-openqa': {'do_lower_case': True},
    'google/realm-orqa-nq-openqa': {'do_lower_case': True},
    'google/realm-orqa-nq-reader': {'do_lower_case': True},
    'google/realm-orqa-wq-openqa': {'do_lower_case': True},
    'google/realm-orqa-wq-reader': {'do_lower_case': True},
}

class     _lowerCamelCase       (     snake_case_					):




 '''simple docstring'''
 __lowercase					:		Union[str, Any]     				=    VOCAB_FILES_NAMES
 __lowercase					:		int     				=    PRETRAINED_VOCAB_FILES_MAP
 __lowercase					:		Union[str, Any]     				=    PRETRAINED_INIT_CONFIGURATION
 __lowercase					:		Union[str, Any]     				=    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 __lowercase					:		List[str]     				=    RealmTokenizer


 def __init__(   self   ,   __lowercase=None   ,   __lowercase=None   ,   __lowercase=True   ,   __lowercase="[UNK]"   ,   __lowercase="[SEP]"   ,   __lowercase="[PAD]"   ,   __lowercase="[CLS]"   ,   __lowercase="[MASK]"   ,   __lowercase=True   ,   __lowercase=None   ,   **__lowercase   ,   ):

  """simple docstring"""







  super().__init__(
      __lowercase   ,   tokenizer_file=__lowercase   ,   do_lower_case=__lowercase   ,   unk_token=__lowercase   ,   sep_token=__lowercase   ,   pad_token=__lowercase   ,   cls_token=__lowercase   ,   mask_token=__lowercase   ,   tokenize_chinese_chars=__lowercase   ,   strip_accents=__lowercase   ,   **__lowercase   ,   )

  __A	:		Optional[int]	  =      json.loads(self.backend_tokenizer.normalizer.__getstate__()    )
  if (
      normalizer_state.get('lowercase'   ,   __lowercase    ) != do_lower_case
      or normalizer_state.get('strip_accents'   ,   __lowercase    ) != strip_accents
      or normalizer_state.get('handle_chinese_chars'   ,   __lowercase    ) != tokenize_chinese_chars
  ):
   __A	:		Tuple	  =      getattr(__lowercase   ,   normalizer_state.pop('type'    )    )
   __A	:		Optional[int]	  =      do_lower_case
   __A	:		List[str]	  =      strip_accents
   __A	:		Dict	  =      tokenize_chinese_chars
   __A	:		List[Any]	  =      normalizer_class(**__lowercase    )

  __A	:		int	  =      do_lower_case


 def 			snake_case__ (   self   ,   __lowercase   ,   **__lowercase    ):

  """simple docstring"""







  __A	:		Dict	  =      PaddingStrategy.MAX_LENGTH

  __A	:		Optional[int]	  =      text
  __A	:		Union[str, Any]	  =      kwargs.pop('text_pair'   ,   __lowercase    )
  __A	:		List[str]	  =      kwargs.pop('return_tensors'   ,   __lowercase    )

  __A	:		int	  =      {
      'input_ids': [],
      'attention_mask': [],
      'token_type_ids': [],
  }

  for idx, candidate_text in enumerate(__lowercase    ):
   if batch_text_pair is not None:
    __A	:		Tuple	  =      batch_text_pair[idx]
   else:
    __A	:		Union[str, Any]	  =      None

   __A	:		Optional[int]	  =      super().__call__(__lowercase   ,   __lowercase   ,   return_tensors=__lowercase   ,   **__lowercase    )

   __A	:		str	  =      encoded_candidates.get('input_ids'    )
   __A	:		Union[str, Any]	  =      encoded_candidates.get('attention_mask'    )
   __A	:		Optional[int]	  =      encoded_candidates.get('token_type_ids'    )

   if encoded_input_ids is not None:
    output_data["input_ids"].append(__lowercase    )
   if encoded_attention_mask is not None:
    output_data["attention_mask"].append(__lowercase    )
   if encoded_token_type_ids is not None:
    output_data["token_type_ids"].append(__lowercase    )

  __A	:		List[str]	  =      {key: item for key, item in output_data.items() if len(__lowercase    ) != 0}

  return BatchEncoding(__lowercase   ,   tensor_type=__lowercase    )


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase=None    ):

  """simple docstring"""







  __A	:		Union[str, Any]	  =      [self.cls_token_id] + token_ids_a + [self.sep_token_id]

  if token_ids_a:
   output += token_ids_a + [self.sep_token_id]

  return output


 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase = None    ):

  """simple docstring"""







  __A	:		Union[str, Any]	  =      [self.sep_token_id]
  __A	:		Tuple	  =      [self.cls_token_id]
  if token_ids_a is None:
   return len(cls + token_ids_a + sep    ) * [0]
  return len(cls + token_ids_a + sep    ) * [0] + len(token_ids_a + sep    ) * [1]







 def 			snake_case__ (   self   ,   __lowercase   ,   __lowercase = None    ):

  """simple docstring"""







  __A	:		int	  =      self._tokenizer.model.save(__lowercase   ,   name=__lowercase    )
  return tuple(__lowercase    )






