


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

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


@require_not_windows
@require_dill_gt_0_3_2
def     lowercase			()   ->						int:
						SCREAMING_SNAKE_CASE													=					pyspark.sql.SparkSession.builder.master('local[*]'  ).appName('pyspark'  ).getOrCreate()
						SCREAMING_SNAKE_CASE													=					spark.range(30  ).repartition(3  )
						# Mock the generator so that shuffle reverses the partition indices.
						with patch('numpy.random.Generator'  ) as generator_mock:
												SCREAMING_SNAKE_CASE													=					lambda SCREAMING_SNAKE_CASE_  : x.reverse()
												SCREAMING_SNAKE_CASE													=					_get_expected_row_ids_and_row_dicts_for_partition_order(SCREAMING_SNAKE_CASE_ ,			[2, 1, 0]  )

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

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

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




__UpperCamelCase						=      frozenset(
    [
        '''prompt''',
        '''height''',
        '''width''',
        '''guidance_scale''',
        '''negative_prompt''',
        '''prompt_embeds''',
        '''negative_prompt_embeds''',
        '''cross_attention_kwargs''',
    ]
)

__UpperCamelCase						=      frozenset(['''prompt''', '''negative_prompt'''])

__UpperCamelCase						=      frozenset([])

__UpperCamelCase						=      frozenset(['''image'''])

__UpperCamelCase						=      frozenset(
    [
        '''image''',
        '''height''',
        '''width''',
        '''guidance_scale''',
    ]
)

__UpperCamelCase						=      frozenset(['''image'''])

__UpperCamelCase						=      frozenset(
    [
        '''prompt''',
        '''image''',
        '''height''',
        '''width''',
        '''guidance_scale''',
        '''negative_prompt''',
        '''prompt_embeds''',
        '''negative_prompt_embeds''',
    ]
)

__UpperCamelCase						=      frozenset(['''prompt''', '''image''', '''negative_prompt'''])

__UpperCamelCase						=      frozenset(
    [
        # Text guided image variation with an image mask
        '''prompt''',
        '''image''',
        '''mask_image''',
        '''height''',
        '''width''',
        '''guidance_scale''',
        '''negative_prompt''',
        '''prompt_embeds''',
        '''negative_prompt_embeds''',
    ]
)

__UpperCamelCase						=      frozenset(['''prompt''', '''image''', '''mask_image''', '''negative_prompt'''])

__UpperCamelCase						=      frozenset(
    [
        # image variation with an image mask
        '''image''',
        '''mask_image''',
        '''height''',
        '''width''',
        '''guidance_scale''',
    ]
)

__UpperCamelCase						=      frozenset(['''image''', '''mask_image'''])

__UpperCamelCase						=      frozenset(
    [
        '''example_image''',
        '''image''',
        '''mask_image''',
        '''height''',
        '''width''',
        '''guidance_scale''',
    ]
)

__UpperCamelCase						=      frozenset(['''example_image''', '''image''', '''mask_image'''])

__UpperCamelCase						=      frozenset(['''class_labels'''])

__UpperCamelCase						=      frozenset(['''class_labels'''])

__UpperCamelCase						=      frozenset(['''batch_size'''])

__UpperCamelCase						=      frozenset([])

__UpperCamelCase						=      frozenset(['''batch_size'''])

__UpperCamelCase						=      frozenset([])

__UpperCamelCase						=      frozenset(
    [
        '''prompt''',
        '''audio_length_in_s''',
        '''guidance_scale''',
        '''negative_prompt''',
        '''prompt_embeds''',
        '''negative_prompt_embeds''',
        '''cross_attention_kwargs''',
    ]
)

__UpperCamelCase						=      frozenset(['''prompt''', '''negative_prompt'''])
__UpperCamelCase						=      frozenset(['''input_tokens'''])

__UpperCamelCase						=      frozenset(['''input_tokens'''])


