






'''simple docstring'''







# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

__lowerCAmelCase       		=      TypeVar('T')





class 	SCREAMING_SNAKE_CASE					(					Generic[T]       ):






			def __init__( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			bool = True					)			->			None:
							a_		:  dict[T, list[T]]											=      {}  # dictionary of lists
							a_		:  Tuple											=      directed






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[Any]  ,						__SCREAMING_SNAKE_CASE      :			T  ,						__SCREAMING_SNAKE_CASE      :			T					)			->			GraphAdjacencyList[T]:

							if not self.directed:  # For undirected graphs
											# if both source vertex and destination vertex are both present in the
											# adjacency list, add destination vertex to source vertex list of adjacent
											# vertices and add source vertex to destination vertex list of adjacent
											# vertices.
											if source_vertex in self.adj_list and destination_vertex in self.adj_list:
															self.adj_list[source_vertex].append(__SCREAMING_SNAKE_CASE					)
															self.adj_list[destination_vertex].append(__SCREAMING_SNAKE_CASE					)
											# if only source vertex is present in adjacency list, add destination vertex
											# to source vertex list of adjacent vertices, then create a new vertex with
											# destination vertex as key and assign a list containing the source vertex
											# as it's first adjacent vertex.
											elif source_vertex in self.adj_list:
															self.adj_list[source_vertex].append(__SCREAMING_SNAKE_CASE					)
															a_		:  List[Any]											=      [source_vertex]
											# if only destination vertex is present in adjacency list, add source vertex
											# to destination vertex list of adjacent vertices, then create a new vertex
											# with source vertex as key and assign a list containing the source vertex
											# as it's first adjacent vertex.
											elif destination_vertex in self.adj_list:
															self.adj_list[destination_vertex].append(__SCREAMING_SNAKE_CASE					)
															a_		:  int											=      [destination_vertex]
											# if both source vertex and destination vertex are not present in adjacency
											# list, create a new vertex with source vertex as key and assign a list
											# containing the destination vertex as it's first adjacent vertex also
											# create a new vertex with destination vertex as key and assign a list
											# containing the source vertex as it's first adjacent vertex.
											else:
															a_		:  Optional[Any]											=      [destination_vertex]
															a_		:  int											=      [source_vertex]
							else:  # For directed graphs
											# if both source vertex and destination vertex are present in adjacency
											# list, add destination vertex to source vertex list of adjacent vertices.
											if source_vertex in self.adj_list and destination_vertex in self.adj_list:
															self.adj_list[source_vertex].append(__SCREAMING_SNAKE_CASE					)
											# if only source vertex is present in adjacency list, add destination
											# vertex to source vertex list of adjacent vertices and create a new vertex
											# with destination vertex as key, which has no adjacent vertex
											elif source_vertex in self.adj_list:
															self.adj_list[source_vertex].append(__SCREAMING_SNAKE_CASE					)
															a_		:  int											=      []
											# if only destination vertex is present in adjacency list, create a new
											# vertex with source vertex as key and assign a list containing destination
											# vertex as first adjacent vertex
											elif destination_vertex in self.adj_list:
															a_		:  Tuple											=      [destination_vertex]
											# if both source vertex and destination vertex are not present in adjacency
											# list, create a new vertex with source vertex as key and a list containing
											# destination vertex as it's first adjacent vertex. Then create a new vertex
											# with destination vertex as key, which has no adjacent vertex
											else:
															a_		:  int											=      [destination_vertex]
															a_		:  Optional[int]											=      []

							return self







			def __repr__( self      :			Tuple					)			->			str:
							return pformat(self.adj_list					)












'''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 		_UpperCAmelCase   (							__A							:   List[str]						,		__A							:   List[Any]      ):
				a_		:  Any											=      []
				for part_id in partition_order:
								a_		:  str											=      df.where(f'SPARK_PARTITION_ID() = {part_id}'      ).collect()
								for row_idx, row in enumerate(__A      ):
												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 		_UpperCAmelCase   (							):
				a_		:  List[str]											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  Union[str, Any]											=      spark.range(1_00      ).repartition(1      )
				a_		:  Any											=      Spark(__A      )
				# 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 		_UpperCAmelCase   (							):
				a_		:  List[Any]											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  int											=      spark.range(10      ).repartition(2      )
				a_		:  Tuple											=      [1, 0]
				a_		:  List[str]											=      _generate_iterable_examples(__A						,		__A      )  # Reverse the partitions.
				a_		:  int											=      _get_expected_row_ids_and_row_dicts_for_partition_order(__A						,		__A      )

				for i, (row_id, row_dict) in enumerate(generate_fn()      ):
								a_				,			a_		:  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 		_UpperCAmelCase   (							):
				a_		:  int											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  str											=      spark.range(10      ).repartition(1      )
				a_		:  Tuple											=      SparkExamplesIterable(__A      )
				assert it.n_shards == 1
				for i, (row_id, row_dict) in enumerate(__A      ):
								assert row_id == f'0_{i}'
								assert row_dict == {"id": i}





@require_not_windows
@require_dill_gt_0_3_2
def 		_UpperCAmelCase   (							):
				a_		:  Tuple											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  str											=      spark.range(30      ).repartition(3      )
				# Mock the generator so that shuffle reverses the partition indices.
				with patch('''numpy.random.Generator'''      ) as generator_mock:
								a_		:  Union[str, Any]											=      lambda __A      : x.reverse()
								a_		:  Any											=      _get_expected_row_ids_and_row_dicts_for_partition_order(__A						,		[2, 1, 0]      )

								a_		:  str											=      SparkExamplesIterable(__A      ).shuffle_data_sources(__A      )
								assert shuffled_it.n_shards == 3
								for i, (row_id, row_dict) in enumerate(__A      ):
												a_				,			a_		:  Optional[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 		_UpperCAmelCase   (							):
				a_		:  int											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  List[str]											=      spark.range(20      ).repartition(4      )

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

				# Partitions 1 and 3
				a_		:  List[Any]											=      SparkExamplesIterable(__A      ).shard_data_sources(worker_id=1						,		num_workers=2      )
				assert shard_it_a.n_shards == 2
				a_		:  Optional[int]											=      _get_expected_row_ids_and_row_dicts_for_partition_order(__A						,		[1, 3]      )
				for i, (row_id, row_dict) in enumerate(__A      ):
								a_				,			a_		:  Any											=      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 		_UpperCAmelCase   (							):
				a_		:  Any											=      pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				a_		:  List[Any]											=      spark.range(1_00      ).repartition(1      )
				a_		:  Optional[Any]											=      Spark(__A      )
				# 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() == 1_00




