



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 			__lowerCAmelCase  (						datasets.BuilderConfig ):
 lowerCAmelCase__							       =	None





def 		a__      (		_UpperCamelCase      :			"pyspark.sql.DataFrame"			,_UpperCamelCase      :			List[int]			,):
   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 			__lowerCAmelCase  (						_BaseExamplesIterable ):




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



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



    '''simple docstring'''
    yield from self.generate_examples_fn()




 def 							lowerCamelCase   (    self						,					__UpperCAmelCase					):



    '''simple docstring'''
    __lowerCamelCase    				=     list(range(self.df.rdd.getNumPartitions()					)					)
    generator.shuffle(__UpperCAmelCase					)
    return SparkExamplesIterable(self.df						,					partition_order=__UpperCAmelCase					)




 def 							lowerCamelCase   (    self						,					__UpperCAmelCase						,					__UpperCAmelCase					):



    '''simple docstring'''
    __lowerCamelCase    				=     self.split_shard_indices_by_worker(__UpperCAmelCase						,					__UpperCAmelCase					)
    return SparkExamplesIterable(self.df						,					partition_order=__UpperCAmelCase					)




 @property
 def 							lowerCamelCase   (    self					):



    '''simple docstring'''
    return len(self.partition_order					)



class 			__lowerCAmelCase  (						datasets.DatasetBuilder ):
 lowerCAmelCase__							       =	SparkConfig




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



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



    '''simple docstring'''
    # Returns the path of the created file.
    def create_cache_and_write_probe(__UpperCAmelCase					):
       # 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					):



    '''simple docstring'''
    return datasets.DatasetInfo(features=self.config.features					)




 def 							lowerCamelCase   (    self						,					__UpperCAmelCase					):



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




 def 							lowerCamelCase   (    self						,					__UpperCAmelCase					):



    '''simple docstring'''
    import pyspark

    def get_arrow_batch_size(__UpperCAmelCase					):
       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						,					__UpperCAmelCase						,					__UpperCAmelCase						,					__UpperCAmelCase						,					):



    '''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					):
       # 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						,					__UpperCAmelCase						,					__UpperCAmelCase = "arrow"						,					__UpperCAmelCase = None						,					__UpperCAmelCase = None						,					**__UpperCAmelCase						,					):



    '''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						,					__UpperCAmelCase						,					__UpperCAmelCase						,					):
          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						,					__UpperCAmelCase						,					):



    '''simple docstring'''
    return SparkExamplesIterable(self.df					)











def 		a__      (		_UpperCamelCase      :			int			,_UpperCamelCase      :			int				):
   while second != 0:
      __lowerCamelCase    				=     first & second
      first ^= second
      __lowerCamelCase    				=     c << 1
   return first


if __name__ == "__main__":
  import doctest

  doctest.testmod()

  a_								=       int(input("""Enter the first number: """).strip())
  a_								=       int(input("""Enter the second number: """).strip())
  print(f"{add(first, second) = }")






