def       lowerCAmelCase_     (		lowercase:							int  ,					lowercase:							bool = False		)    ->				bool:







 '''simple docstring'''
 if n == 2:
  return True
 if not n % 2 or n < 2:
  return False
 if n > 5 and n % 10 not in (1, 3, 7, 9):  # can quickly check last digit
  return False
 if n > 3_317_044_064_679_887_385_961_981 and not allow_probable:
  raise ValueError(
      '''Warning: upper bound of deterministic test is exceeded. '''
      '''Pass allow_probable=True to allow probabilistic test. '''
      '''A return value of True indicates a probable prime.'''		)
 # array bounds provided by analysis
 _UpperCamelCase:     Optional[Any]							= [
     2_047,
     1_373_653,
     25_326_001,
     3_215_031_751,
     2_152_302_898_747,
     3_474_749_660_383,
     341_550_071_728_321,
     1,
     3_825_123_056_546_413_051,
     1,
     1,
     318_665_857_834_031_151_167_461,
     3_317_044_064_679_887_385_961_981,
 ]

 _UpperCamelCase:     Tuple							= [2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41]
 for idx, _p in enumerate(lowercase  ,					1		):
  if n < _p:
   # then we have our last prime to check
   _UpperCamelCase:     Optional[int]							= primes[:idx]
   break
 _UpperCamelCase	,							_UpperCamelCase:     str							= n - 1, 0
 # break up n -1 into a power of 2 (s) and
 # remaining odd component
 # essentially, solve for d * 2 ** s == n - 1
 while d % 2 == 0:
  d //= 2
  s += 1
 for prime in plist:
  _UpperCamelCase:     Optional[int]							= False
  for r in range(lowercase		):
   _UpperCamelCase:     List[Any]							= pow(lowercase  ,					d * 2**r  ,					lowercase		)
   # see article for analysis explanation for m
   if (r == 0 and m == 1) or ((m + 1) % n == 0):
    _UpperCamelCase:     Dict							= True
    # this loop will not determine compositeness
    break
  if pr:
   continue
  # if pr is False, then the above loop never evaluated to true,
  # and the n MUST be composite
  return False
 return True




def       lowerCAmelCase_     (		)    ->				None:







 '''simple docstring'''
 assert not miller_rabin(561		)
 assert miller_rabin(563		)
 # 2047

 assert not miller_rabin(838_201		)
 assert miller_rabin(838_207		)
 # 1_373_653

 assert not miller_rabin(17_316_001		)
 assert miller_rabin(17_316_017		)
 # 25_326_001

 assert not miller_rabin(3_078_386_641		)
 assert miller_rabin(3_078_386_653		)
 # 3_215_031_751

 assert not miller_rabin(1_713_045_574_801		)
 assert miller_rabin(1_713_045_574_819		)
 # 2_152_302_898_747

 assert not miller_rabin(2_779_799_728_307		)
 assert miller_rabin(2_779_799_728_327		)
 # 3_474_749_660_383

 assert not miller_rabin(113_850_023_909_441		)
 assert miller_rabin(113_850_023_909_527		)
 # 341_550_071_728_321

 assert not miller_rabin(1_275_041_018_848_804_351		)
 assert miller_rabin(1_275_041_018_848_804_391		)
 # 3_825_123_056_546_413_051

 assert not miller_rabin(79_666_464_458_507_787_791_867		)
 assert miller_rabin(79_666_464_458_507_787_791_951		)
 # 318_665_857_834_031_151_167_461

 assert not miller_rabin(552_840_677_446_647_897_660_333		)
 assert miller_rabin(552_840_677_446_647_897_660_359		)
 # 3_317_044_064_679_887_385_961_981
 # upper limit for probabilistic test


if __name__ == "__main__":
  test_miller_rabin()
from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)




def       lowerCAmelCase_     (		lowercase:							Optional[Any]  ,					lowercase:							Tuple		)    ->				Any:







 '''simple docstring'''
 _UpperCamelCase:     Union[str, Any]							= []
 for part_id in partition_order:
  _UpperCamelCase:     int							= df.where(F"""SPARK_PARTITION_ID() = {part_id}"""		).collect()
  for row_idx, row in enumerate(lowercase		):
   expected_row_ids_and_row_dicts.append((F"""{part_id}_{row_idx}""", row.asDict())		)
 return expected_row_ids_and_row_dicts




@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				List[str]:







 '''simple docstring'''
 _UpperCamelCase:     Optional[int]							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     int							= spark.range(100		).repartition(1		)
 _UpperCamelCase:     int							= Spark(lowercase		)
 # The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
 # that each partition can hold 2 rows.
 spark_builder._repartition_df_if_needed(max_shard_size=16		)
 # Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
 assert spark_builder.df.rdd.getNumPartitions() == 50




@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				List[str]:







 '''simple docstring'''
 _UpperCamelCase:     List[Any]							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     Optional[Any]							= spark.range(10		).repartition(2		)
 _UpperCamelCase:     int							= [1, 0]
 _UpperCamelCase:     Any							= _generate_iterable_examples(lowercase  ,					lowercase		)  # Reverse the partitions.
 _UpperCamelCase:     List[str]							= _get_expected_row_ids_and_row_dicts_for_partition_order(lowercase  ,					lowercase		)

 for i, (row_id, row_dict) in enumerate(generate_fn()		):
  _UpperCamelCase	,							_UpperCamelCase:     List[Any]							= expected_row_ids_and_row_dicts[i]
  assert row_id == expected_row_id
  assert row_dict == expected_row_dict




@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				Any:







 '''simple docstring'''
 _UpperCamelCase:     Optional[Any]							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     Any							= spark.range(10		).repartition(1		)
 _UpperCamelCase:     int							= SparkExamplesIterable(lowercase		)
 assert it.n_shards == 1
 for i, (row_id, row_dict) in enumerate(lowercase		):
  assert row_id == F"""0_{i}"""
  assert row_dict == {"id": i}




@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				Union[str, Any]:







 '''simple docstring'''
 _UpperCamelCase:     Any							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     str							= spark.range(30		).repartition(3		)
 # Mock the generator so that shuffle reverses the partition indices.
 with patch('''numpy.random.Generator'''		) as generator_mock:
  _UpperCamelCase:     Union[str, Any]							= lambda lowercase		: x.reverse()
  _UpperCamelCase:     List[str]							= _get_expected_row_ids_and_row_dicts_for_partition_order(lowercase  ,					[2, 1, 0]		)

  _UpperCamelCase:     Union[str, Any]							= SparkExamplesIterable(lowercase		).shuffle_data_sources(lowercase		)
  assert shuffled_it.n_shards == 3
  for i, (row_id, row_dict) in enumerate(lowercase		):
   _UpperCamelCase	,							_UpperCamelCase:     List[Any]							= expected_row_ids_and_row_dicts[i]
   assert row_id == expected_row_id
   assert row_dict == expected_row_dict




@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				Union[str, Any]:







 '''simple docstring'''
 _UpperCamelCase:     str							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     Tuple							= spark.range(20		).repartition(4		)

 # Partitions 0 and 2
 _UpperCamelCase:     List[Any]							= SparkExamplesIterable(lowercase		).shard_data_sources(worker_id=0  ,					num_workers=2		)
 assert shard_it_a.n_shards == 2
 _UpperCamelCase:     Dict							= _get_expected_row_ids_and_row_dicts_for_partition_order(lowercase  ,					[0, 2]		)
 for i, (row_id, row_dict) in enumerate(lowercase		):
  _UpperCamelCase	,							_UpperCamelCase:     Optional[int]							= expected_row_ids_and_row_dicts_a[i]
  assert row_id == expected_row_id
  assert row_dict == expected_row_dict

 # Partitions 1 and 3
 _UpperCamelCase:     str							= SparkExamplesIterable(lowercase		).shard_data_sources(worker_id=1  ,					num_workers=2		)
 assert shard_it_a.n_shards == 2
 _UpperCamelCase:     str							= _get_expected_row_ids_and_row_dicts_for_partition_order(lowercase  ,					[1, 3]		)
 for i, (row_id, row_dict) in enumerate(lowercase		):
  _UpperCamelCase	,							_UpperCamelCase:     str							= expected_row_ids_and_row_dicts_a[i]
  assert row_id == expected_row_id
  assert row_dict == expected_row_dict





@require_not_windows
@require_dill_gt_0_3_2
def       lowerCAmelCase_     (		)    ->				Optional[int]:







 '''simple docstring'''
 _UpperCamelCase:     Any							= pyspark.sql.SparkSession.builder.master('''local[*]'''		).appName('''pyspark'''		).getOrCreate()
 _UpperCamelCase:     Union[str, Any]							= spark.range(100		).repartition(1		)
 _UpperCamelCase:     Optional[Any]							= Spark(lowercase		)
 # Choose a small max_shard_size for maximum partitioning.
 spark_builder._repartition_df_if_needed(max_shard_size=1		)
 # The new number of partitions should not be greater than the number of rows.
 assert spark_builder.df.rdd.getNumPartitions() == 100