





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


lowercase__    :      List[Any]						=    datasets.utils.logging.get_logger(__name__)

if TYPE_CHECKING:
  import pyspark


@dataclass
class  SCREAMING_SNAKE_CASE    (datasets.BuilderConfig							):
     lowerCAmelCase					=       None






def        _lowerCAmelCase			(  __snake_case  : "pyspark.sql.DataFrame" ,							__snake_case  : List[int] ,							)     ->     Dict:
     import pyspark

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

     return generate_fn


class  SCREAMING_SNAKE_CASE    (_BaseExamplesIterable							):


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

          '''simple docstring'''






          __A						: Dict      =       df
          __A						: Optional[Any]      =       partition_order or range(self.df.rdd.getNumPartitions())
          __A						: 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						,	_UpperCAmelCase):

          '''simple docstring'''






          __A						: str      =       list(range(self.df.rdd.getNumPartitions()))
          generator.shuffle(__SCREAMING_SNAKE_CASE)
          return SparkExamplesIterable(self.df						,	partition_order=__SCREAMING_SNAKE_CASE)


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase						,	_UpperCAmelCase):

          '''simple docstring'''






          __A						: List[Any]      =       self.split_shard_indices_by_worker(__SCREAMING_SNAKE_CASE						,	__SCREAMING_SNAKE_CASE)
          return SparkExamplesIterable(self.df						,	partition_order=__SCREAMING_SNAKE_CASE)


     @property
     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          return len(self.partition_order)




class  SCREAMING_SNAKE_CASE    (datasets.DatasetBuilder							):
     lowerCAmelCase					=       SparkConfig


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

          '''simple docstring'''






          import pyspark

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

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


     def 							SCREAMING_SNAKE_CASE  (						self):

          '''simple docstring'''






          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=__SCREAMING_SNAKE_CASE)
               __A						: List[str]      =       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(__SCREAMING_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:
               __A						: List[Any]      =       (
                   self._spark.sparkContext.parallelize(range(1)						,	1).mapPartitions(__SCREAMING_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):

          '''simple docstring'''






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


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase):

          '''simple docstring'''






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


     def 							SCREAMING_SNAKE_CASE  (						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]})

          __A						: int      =       self.df.count()
          __A						: Union[str, Any]      =       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						: Dict      =       (
              self.df.limit(__SCREAMING_SNAKE_CASE)
              .repartition(1)
              .mapInArrow(__SCREAMING_SNAKE_CASE						,	'batch_bytes: long')
              .agg(pyspark.sql.functions.sum('batch_bytes').alias('sample_bytes'))
              .collect()[0]
              .sample_bytes
              / sample_num_rows
          )
          __A						: Union[str, 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.
               __A						: Dict      =       min(__SCREAMING_SNAKE_CASE						,	int(approx_total_size / max_shard_size))
               __A						: Optional[Any]      =       self.df.repartition(__SCREAMING_SNAKE_CASE)


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase						,	_UpperCAmelCase						,	_UpperCAmelCase						,	):

          '''simple docstring'''






          import pyspark

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

          def write_arrow(_UpperCAmelCase):
               # Within the same SparkContext, no two task attempts will share the same attempt ID.
               __A						: str      =       pyspark.TaskContext().taskAttemptId()
               __A						: Optional[int]      =       next(__SCREAMING_SNAKE_CASE						,	__SCREAMING_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']						,	)
               __A						: Union[str, Any]      =       0
               __A						: Union[str, Any]      =       writer_class(
                   features=__SCREAMING_SNAKE_CASE						,	path=working_fpath.replace('SSSSS'						,	F'{shard_id:05d}').replace('TTTTT'						,	F'{task_id:05d}')						,	writer_batch_size=__SCREAMING_SNAKE_CASE						,	storage_options=__SCREAMING_SNAKE_CASE						,	embed_local_files=__SCREAMING_SNAKE_CASE						,	)
               __A						: str      =       pa.Table.from_batches([first_batch])
               writer.write_table(__SCREAMING_SNAKE_CASE)
               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						: Tuple      =       writer_class(
                             features=writer._features						,	path=working_fpath.replace('SSSSS'						,	F'{shard_id:05d}').replace('TTTTT'						,	F'{task_id:05d}')						,	writer_batch_size=__SCREAMING_SNAKE_CASE						,	storage_options=__SCREAMING_SNAKE_CASE						,	embed_local_files=__SCREAMING_SNAKE_CASE						,	)
                    __A						: Optional[Any]      =       pa.Table.from_batches([batch])
                    writer.write_table(__SCREAMING_SNAKE_CASE)

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

          __A						: Dict      =       (
              self.df.mapInArrow(__SCREAMING_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						,	_UpperCAmelCase						,	_UpperCAmelCase = "arrow"						,	_UpperCAmelCase = None						,	_UpperCAmelCase = None						,	**_UpperCAmelCase						,	):

          '''simple docstring'''






          self._validate_cache_dir()

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

          __A						: Tuple      =       '-TTTTT-SSSSS-of-NNNNN'
          __A						: str      =       F'{self.name}-{split_generator.name}{SUFFIX}.{file_format}'
          __A						: Union[str, Any]      =       path_join(self._output_dir						,	__SCREAMING_SNAKE_CASE)

          __A						: List[str]      =       0
          __A						: Optional[Any]      =       0
          __A						: str      =       0
          __A						: Optional[int]      =       []
          __A						: Tuple      =       []

          for task_id, content in self._prepare_split_single(__SCREAMING_SNAKE_CASE						,	__SCREAMING_SNAKE_CASE						,	__SCREAMING_SNAKE_CASE):
               (
                   (
                   __A	
               ) ,(
                   __A	
               ) ,(
                   __A	
               ) ,(
                   __A	
               ) ,	
               )						: Optional[int]      =       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(__SCREAMING_SNAKE_CASE)

          __A						: List[str]      =       total_num_examples
          __A						: int      =       total_num_bytes

          # should rename everything at the end
          logger.debug(F'Renaming {total_shards} shards.')
          if total_shards > 1:
               __A						: Dict      =       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						: Any      =       self._fs

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

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


     def 							SCREAMING_SNAKE_CASE  (						self						,	_UpperCAmelCase						,	):

          '''simple docstring'''






          return SparkExamplesIterable(self.df)







"""simple docstring"""





def      lowerCamelCase__       (							_lowerCamelCase       :		int		,				_lowerCamelCase       :		int )      ->				int:
  lowerCamelCase_         =						1  # To kept the Calculated Value
  # Since C(n, k) = C(n, n-k)
  if k > (n - k):
    lowerCamelCase_         =						n - k
  # Calculate C(n,k)
  for i in range(_lowerCamelCase ):
    result *= n - i
    result //= i + 1
  return result







def      lowerCamelCase__       (							_lowerCamelCase       :		int )      ->				int:
  return binomial_coefficient(2 * node_count		,				_lowerCamelCase ) // (node_count + 1)







def      lowerCamelCase__       (							_lowerCamelCase       :		int )      ->				int:
  if n < 0:
    raise ValueError('factorial() not defined for negative values' )
  lowerCamelCase_         =						1
  for i in range(1		,				n + 1 ):
    result *= i
  return result







def      lowerCamelCase__       (							_lowerCamelCase       :		int )      ->				int:
  return catalan_number(_lowerCamelCase ) * factorial(_lowerCamelCase )


if __name__ == "__main__":
  _SCREAMING_SNAKE_CASE					:				int					    =						int(input('''Enter the number of nodes: ''').strip() or 0)
  if node_count <= 0:
    raise ValueError('''We need some nodes to work with.''')
  print(
      F'''Given {node_count} nodes, there are {binary_tree_count(node_count)} '''
      F'''binary trees and {catalan_number(node_count)} binary search trees.'''
  )




