"""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 					__a   (		A    ,				A      ):
     lowercase__				 =							[]
     for part_id in partition_order:
          lowercase__				 =							df.where(f'''SPARK_PARTITION_ID() = {part_id}'''      ).collect()
          for row_idx, row in enumerate(snake_case_      ):
               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 					__a   (		):
     lowercase__				 =							pyspark.sql.SparkSession.builder.master("local[*]"      ).appName("pyspark"      ).getOrCreate()
     lowercase__				 =							spark.range(1_00      ).repartition(1      )
     lowercase__				 =							Spark(snake_case_      )
     # 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 					__a   (		):
     lowercase__				 =							pyspark.sql.SparkSession.builder.master("local[*]"      ).appName("pyspark"      ).getOrCreate()
     lowercase__				 =							spark.range(10      ).repartition(2      )
     lowercase__				 =							[1, 0]
     lowercase__				 =							_generate_iterable_examples(snake_case_    ,				snake_case_      )  # Reverse the partitions.
     lowercase__				 =							_get_expected_row_ids_and_row_dicts_for_partition_order(snake_case_    ,				snake_case_      )

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







@require_not_windows
@require_dill_gt_0_3_2
def 					__a   (		):
     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 A      : x.reverse()
          lowercase__				 =							_get_expected_row_ids_and_row_dicts_for_partition_order(snake_case_    ,				[2, 1, 0]      )

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

     # Partitions 0 and 2
     lowercase__				 =							SparkExamplesIterable(snake_case_      ).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(snake_case_    ,				[0, 2]      )
     for i, (row_id, row_dict) in enumerate(snake_case_      ):
          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(snake_case_      ).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(snake_case_    ,				[1, 3]      )
     for i, (row_id, row_dict) in enumerate(snake_case_      ):
          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 					__a   (		):
     lowercase__				 =							pyspark.sql.SparkSession.builder.master("local[*]"      ).appName("pyspark"      ).getOrCreate()
     lowercase__				 =							spark.range(1_00      ).repartition(1      )
     lowercase__				 =							Spark(snake_case_      )
     # 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"""


from __future__ import annotations







def 					__a   (		A    ,				A      ):






     '''simple docstring'''



     if partitions <= 0:
          raise ValueError("partitions must be a positive number!"      )
     if partitions > number_of_bytes:
          raise ValueError("partitions can not > number_of_bytes!"      )
     lowercase__				 =							number_of_bytes // partitions
     lowercase__				 =							[]
     for i in range(A      ):
          lowercase__				 =							i * bytes_per_partition + 1
          lowercase__				 =							(
              number_of_bytes if i == partitions - 1 else (i + 1) * bytes_per_partition
          )
          allocation_list.append(f'''{start_bytes}-{end_bytes}'''      )
     return allocation_list


if __name__ == "__main__":
 import doctest

 doctest.testmod()




