





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

   for i, (row_id, row_dict) in enumerate(generate_fn()			):
      lowercase_      ,							lowercase_      :Union[str, 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_  (						):
   lowercase_      :Any         =			pyspark.sql.SparkSession.builder.master("local[*]"			).appName("pyspark"			).getOrCreate()
   lowercase_      :str         =			spark.range(10			).repartition(1			)
   lowercase_      :Optional[Any]         =			SparkExamplesIterable(__lowerCamelCase			)
   assert it.n_shards == 1
   for i, (row_id, row_dict) in enumerate(__lowerCamelCase			):
      assert row_id == F'0_{i}'
      assert row_dict == {"id": i}
@require_not_windows
@require_dill_gt_0_3_2
def 							UpperCAmelCase_  (						):
   lowercase_      :Tuple         =			pyspark.sql.SparkSession.builder.master("local[*]"			).appName("pyspark"			).getOrCreate()
   lowercase_      :Optional[Any]         =			spark.range(30			).repartition(3			)
   # Mock the generator so that shuffle reverses the partition indices.
   with patch("numpy.random.Generator"			) as generator_mock:
      lowercase_      :List[Any]         =			lambda __lowerCamelCase			: x.reverse()
      lowercase_      :Union[str, Any]         =			_get_expected_row_ids_and_row_dicts_for_partition_order(__lowerCamelCase     ,[2, 1, 0]			)

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

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









'''simple docstring'''


def 							UpperCAmelCase_  (						__lowerCamelCase : dict			):
   lowercase_      :set[int]         =			set()
   # To detect a back edge, keep track of vertices currently in the recursion stack
   lowercase_      :set[int]         =			set()
   return any(
       node not in visited and depth_first_search(__lowerCamelCase     ,__lowerCamelCase     ,__lowerCamelCase     ,__lowerCamelCase			)
       for node in graph			)
def 							UpperCAmelCase_  (						__lowerCamelCase : dict     ,__lowerCamelCase : int     ,__lowerCamelCase : set     ,__lowerCamelCase : set			):
   visited.add(__lowerCamelCase			)
   rec_stk.add(__lowerCamelCase			)

   for node in graph[vertex]:
      if node not in visited:
         if depth_first_search(__lowerCamelCase     ,__lowerCamelCase     ,__lowerCamelCase     ,__lowerCamelCase			):
            return True
      elif node in rec_stk:
         return True

    # The node needs to be removed from recursion stack before function ends
   rec_stk.remove(__lowerCamelCase			)
   return False


if __name__ == "__main__":
 from doctest import testmod

 testmod()


