"""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]		) ->    str:
      '''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(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__       (				) ->    Any:
      '''simple docstring'''


      lowercase														=	pyspark.sql.SparkSession.builder.master("""local[*]"""		).appName("""pyspark"""		).getOrCreate()
      lowercase														=	spark.range(1_0_0		).repartition(1		)
      lowercase														=	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=1_6		)
      # Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
      assert spark_builder.df.rdd.getNumPartitions() == 5_0


@require_not_windows
@require_dill_gt_0_3_2
def 			UpperCAmelCase__       (				) ->    int:
      '''simple docstring'''


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

      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 			UpperCAmelCase__       (				) ->    Optional[int]:
      '''simple docstring'''


      lowercase														=	pyspark.sql.SparkSession.builder.master("""local[*]"""		).appName("""pyspark"""		).getOrCreate()
      lowercase														=	spark.range(1_0		).repartition(1		)
      lowercase														=	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__       (				) ->    Optional[Any]:
      '''simple docstring'''


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

            lowercase														=	SparkExamplesIterable(lowerCAmelCase__		).shuffle_data_sources(lowerCAmelCase__		)
            assert shuffled_it.n_shards == 3
            for i, (row_id, row_dict) in enumerate(lowerCAmelCase__		):
                  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 			UpperCAmelCase__       (				) ->    Union[str, Any]:
      '''simple docstring'''


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

      # Partitions 0 and 2
      lowercase														=	SparkExamplesIterable(lowerCAmelCase__		).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(lowerCAmelCase__					,		[0, 2]		)
      for i, (row_id, row_dict) in enumerate(lowerCAmelCase__		):
            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(lowerCAmelCase__		).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(lowerCAmelCase__					,		[1, 3]		)
      for i, (row_id, row_dict) in enumerate(lowerCAmelCase__		):
            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 			UpperCAmelCase__       (				) ->    Optional[Any]:
      '''simple docstring'''


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




"""simple docstring"""






from ..utils import DummyObject, requires_backends





class  _A						(       metaclass=lowerCAmelCase	):
      snake_case__			:   List[str]							  =	['onnx']

      def __init__(						self		,       *__lowerCAmelCase		,       **__lowerCAmelCase						):







            """simple docstring"""



            requires_backends(self		,       ["""onnx"""]						)

      @classmethod
      def 				A__       (						cls		,       *__lowerCAmelCase		,       **__lowerCAmelCase						):







            """simple docstring"""



            requires_backends(cls		,       ["""onnx"""]						)







      @classmethod
      def 				A__       (						cls		,       *__lowerCAmelCase		,       **__lowerCAmelCase						):







            """simple docstring"""



            requires_backends(cls		,       ["""onnx"""]						)



