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 lowerCamelCase						(     SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			):
    '''simple docstring'''







    __UpperCamelCase  :Any    						=	[]
    for part_id in partition_order:
        __UpperCamelCase  :Optional[int]    						=	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 lowerCamelCase						(     ):
    '''simple docstring'''







    __UpperCamelCase  :Union[str, Any]    						=	pyspark.sql.SparkSession.builder.master('''local[*]'''			).appName('''pyspark'''			).getOrCreate()
    __UpperCamelCase  :List[Any]    						=	spark.range(100			).repartition(1			)
    __UpperCamelCase  :Any    						=	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 lowerCamelCase						(     ):
    '''simple docstring'''







    __UpperCamelCase  :Optional[Any]    						=	pyspark.sql.SparkSession.builder.master('''local[*]'''			).appName('''pyspark'''			).getOrCreate()
    __UpperCamelCase  :Union[str, Any]    						=	spark.range(10			).repartition(2			)
    __UpperCamelCase  :Optional[Any]    						=	[1, 0]
    __UpperCamelCase  :Any    						=	_generate_iterable_examples(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			)  # Reverse the partitions.
    __UpperCamelCase  :Optional[Any]    						=	_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()			):
        __UpperCamelCase						, __UpperCamelCase  :int    						=	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 lowerCamelCase						(     ):
    '''simple docstring'''







    __UpperCamelCase  :Optional[int]    						=	pyspark.sql.SparkSession.builder.master('''local[*]'''			).appName('''pyspark'''			).getOrCreate()
    __UpperCamelCase  :str    						=	spark.range(10			).repartition(1			)
    __UpperCamelCase  :Tuple    						=	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 lowerCamelCase						(     ):
    '''simple docstring'''







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

        __UpperCamelCase  :Union[str, Any]    						=	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			):
            __UpperCamelCase						, __UpperCamelCase  :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 lowerCamelCase						(     ):
    '''simple docstring'''







    __UpperCamelCase  :Dict    						=	pyspark.sql.SparkSession.builder.master('''local[*]'''			).appName('''pyspark'''			).getOrCreate()
    __UpperCamelCase  :List[str]    						=	spark.range(20			).repartition(4			)

    # Partitions 0 and 2
    __UpperCamelCase  :Union[str, Any]    						=	SparkExamplesIterable(SCREAMING_SNAKE_CASE			).shard_data_sources(worker_id=0						,    num_workers=2			)
    assert shard_it_a.n_shards == 2
    __UpperCamelCase  :Optional[Any]    						=	_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			):
        __UpperCamelCase						, __UpperCamelCase  :int    						=	expected_row_ids_and_row_dicts_a[i]
        assert row_id == expected_row_id
        assert row_dict == expected_row_dict

    # Partitions 1 and 3
    __UpperCamelCase  :Optional[Any]    						=	SparkExamplesIterable(SCREAMING_SNAKE_CASE			).shard_data_sources(worker_id=1						,    num_workers=2			)
    assert shard_it_a.n_shards == 2
    __UpperCamelCase  :Tuple    						=	_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			):
        __UpperCamelCase						, __UpperCamelCase  :str    						=	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 lowerCamelCase						(     ):
    '''simple docstring'''







    __UpperCamelCase  :Optional[Any]    						=	pyspark.sql.SparkSession.builder.master('''local[*]'''			).appName('''pyspark'''			).getOrCreate()
    __UpperCamelCase  :List[Any]    						=	spark.range(100			).repartition(1			)
    __UpperCamelCase  :List[Any]    						=	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() == 100






def lowerCamelCase						(     SCREAMING_SNAKE_CASE			):
    '''simple docstring'''







    __UpperCamelCase  :Optional[int]    						=	generate_pascal_triangle(SCREAMING_SNAKE_CASE			)
    for row_idx in range(SCREAMING_SNAKE_CASE			):
        # Print left spaces
        for _ in range(num_rows - row_idx - 1			):
            print(end=''' '''			)
        # Print row values
        for col_idx in range(row_idx + 1			):
            if col_idx != row_idx:
                print(triangle[row_idx][col_idx]						,    end=''' '''			)
            else:
                print(triangle[row_idx][col_idx]						,    end=''''''			)
        print()



def lowerCamelCase						(     SCREAMING_SNAKE_CASE			):
    '''simple docstring'''







    if not isinstance(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			):
        raise TypeError('''The input value of \'num_rows\' should be \'int\''''			)

    if num_rows == 0:
        return []
    elif num_rows < 0:
        raise ValueError(
            '''The input value of \'num_rows\' should be greater than or equal to 0'''			)

    __UpperCamelCase  :list[list[int]]    						=	[]
    for current_row_idx in range(SCREAMING_SNAKE_CASE			):
        __UpperCamelCase  :List[Any]    						=	populate_current_row(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			)
        triangle.append(SCREAMING_SNAKE_CASE			)
    return triangle



def lowerCamelCase						(     SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			):
    '''simple docstring'''







    __UpperCamelCase  :int    						=	[-1] * (current_row_idx + 1)
    # first and last elements of current row are equal to 1
    __UpperCamelCase						, __UpperCamelCase  :int    						=	1, 1
    for current_col_idx in range(1						,    SCREAMING_SNAKE_CASE			):
        calculate_current_element(
            SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			)
    return current_row



def lowerCamelCase						(     SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE						,    ):
    '''simple docstring'''







    __UpperCamelCase  :Dict    						=	triangle[current_row_idx - 1][current_col_idx - 1]
    __UpperCamelCase  :List[Any]    						=	triangle[current_row_idx - 1][current_col_idx]
    __UpperCamelCase  :List[Any]    						=	above_to_left_elt + above_to_right_elt



def lowerCamelCase						(     SCREAMING_SNAKE_CASE			):
    '''simple docstring'''







    if not isinstance(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			):
        raise TypeError('''The input value of \'num_rows\' should be \'int\''''			)

    if num_rows == 0:
        return []
    elif num_rows < 0:
        raise ValueError(
            '''The input value of \'num_rows\' should be greater than or equal to 0'''			)

    __UpperCamelCase  :list[list[int]]    						=	[[1]]

    for row_index in range(1						,    SCREAMING_SNAKE_CASE			):
        __UpperCamelCase  :Optional[Any]    						=	[0] + result[-1] + [0]
        __UpperCamelCase  :Any    						=	row_index + 1
        # Calculate the number of distinct elements in a row
        __UpperCamelCase  :Optional[Any]    						=	sum(divmod(SCREAMING_SNAKE_CASE						,    2			)			)
        __UpperCamelCase  :Union[str, Any]    						=	[
            temp_row[i - 1] + temp_row[i] for i in range(1						,    distinct_elements + 1			)
        ]
        __UpperCamelCase  :List[Any]    						=	row_first_half[: (row_index + 1) // 2]
        row_second_half.reverse()
        __UpperCamelCase  :List[str]    						=	row_first_half + row_second_half
        result.append(SCREAMING_SNAKE_CASE			)

    return result



def lowerCamelCase						(     ):
    '''simple docstring'''







    from collections.abc import Callable
    from timeit import timeit

    def benchmark_a_function(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			) -> None:
        __UpperCamelCase  :List[str]    						=	f"""{func.__name__}({value})"""
        __UpperCamelCase  :Optional[int]    						=	timeit(f"""__main__.{call}"""						,    setup='''import __main__'''			)
        # print(f"{call:38} = {func(value)} -- {timing:.4f} seconds")
        print(f"""{call:38} -- {timing:.4f} seconds"""			)

    for value in range(15			):  # (1, 7, 14):
        for func in (generate_pascal_triangle, generate_pascal_triangle_optimized):
            benchmark_a_function(SCREAMING_SNAKE_CASE						,    SCREAMING_SNAKE_CASE			)
        print()


if __name__ == "__main__":
    import doctest

    doctest.testmod()
    benchmark()





