


"""simple docstring"""

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_      (				UpperCamelCase__	: Tuple				,   UpperCamelCase__	: Dict ):






 """simple docstring"""







 __lowercase									=	[]
 for part_id in partition_order:
  __lowercase									=	df.where(f'''SPARK_PARTITION_ID() = {part_id}''' ).collect()
  for row_idx, row in enumerate(UpperCamelCase__ ):
   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_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(100 ).repartition(1 )
 __lowercase									=	Spark(UpperCamelCase__ )
 # 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_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(10 ).repartition(2 )
 __lowercase									=	[1, 0]
 __lowercase									=	_generate_iterable_examples(UpperCamelCase__				,   UpperCamelCase__ )  # Reverse the partitions.
 __lowercase									=	_get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase__				,   UpperCamelCase__ )

 for i, (row_id, row_dict) in enumerate(generate_fn() ):
  __lowercase							,		__lowercase									=	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_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(10 ).repartition(1 )
 __lowercase									=	SparkExamplesIterable(UpperCamelCase__ )
 assert it.n_shards == 1
 for i, (row_id, row_dict) in enumerate(UpperCamelCase__ ):
  assert row_id == f'''0_{i}'''
  assert row_dict == {"id": i}


@require_not_windows
@require_dill_gt_0_3_2
def lowerCAmelCase_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(30 ).repartition(3 )
 # Mock the generator so that shuffle reverses the partition indices.
 with patch("""numpy.random.Generator""" ) as generator_mock:
  __lowercase									=	lambda UpperCamelCase__ : x.reverse()
  __lowercase									=	_get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase__				,   [2, 1, 0] )

  __lowercase									=	SparkExamplesIterable(UpperCamelCase__ ).shuffle_data_sources(UpperCamelCase__ )
  assert shuffled_it.n_shards == 3
  for i, (row_id, row_dict) in enumerate(UpperCamelCase__ ):
   __lowercase							,		__lowercase									=	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_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(20 ).repartition(4 )

 # Partitions 0 and 2
 __lowercase									=	SparkExamplesIterable(UpperCamelCase__ ).shard_data_sources(worker_id=0				,   num_workers=2 )
 assert shard_it_a.n_shards == 2
 __lowercase									=	_get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase__				,   [0, 2] )
 for i, (row_id, row_dict) in enumerate(UpperCamelCase__ ):
  __lowercase							,		__lowercase									=	expected_row_ids_and_row_dicts_a[i]
  assert row_id == expected_row_id
  assert row_dict == expected_row_dict

 # Partitions 1 and 3
 __lowercase									=	SparkExamplesIterable(UpperCamelCase__ ).shard_data_sources(worker_id=1				,   num_workers=2 )
 assert shard_it_a.n_shards == 2
 __lowercase									=	_get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase__				,   [1, 3] )
 for i, (row_id, row_dict) in enumerate(UpperCamelCase__ ):
  __lowercase							,		__lowercase									=	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_      (				):






 """simple docstring"""







 __lowercase									=	pyspark.sql.SparkSession.builder.master("""local[*]""" ).appName("""pyspark""" ).getOrCreate()
 __lowercase									=	spark.range(100 ).repartition(1 )
 __lowercase									=	Spark(UpperCamelCase__ )
 # 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




"""simple docstring"""

import random


def lowerCAmelCase_      (				UpperCamelCase__	: int				,   UpperCamelCase__	: float				,   UpperCamelCase__	: bool = False ):






 """simple docstring"""







 __lowercase									=	{i: [] for i in range(UpperCamelCase__ )}

 # if probability is greater or equal than 1, then generate a complete graph
 if probability >= 1:
  return complete_graph(UpperCamelCase__ )
 # if probability is lower or equal than 0, then return a graph without edges
 if probability <= 0:
  return graph

 # for each couple of nodes, add an edge from u to v
 # if the number randomly generated is greater than probability probability
 for i in range(UpperCamelCase__ ):
  for j in range(i + 1				,   UpperCamelCase__ ):
   if random.random() < probability:
    graph[i].append(UpperCamelCase__ )
    if not directed:
     # if the graph is undirected, add an edge in from j to i, either
     graph[j].append(UpperCamelCase__ )
 return graph


def lowerCAmelCase_      (				UpperCamelCase__	: int ):






 """simple docstring"""







 return {
     i: [j for j in range(UpperCamelCase__ ) if i != j] for i in range(UpperCamelCase__ )
 }


if __name__ == "__main__":
       import doctest

       doctest.testmod()
