




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

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



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

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

  # Partitions 0 and 2
  lowerCAmelCase__				:  Union[str, Any]		       =     SparkExamplesIterable(UpperCamelCase      ).shard_data_sources(worker_id=0	,      num_workers=2      )
  assert shard_it_a.n_shards == 2
  lowerCAmelCase__				:  Any		       =     _get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase	,      [0, 2]      )
  for i, (row_id, row_dict) in enumerate(UpperCamelCase      ):
    lowerCAmelCase__       ,			lowerCAmelCase__				:  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
  lowerCAmelCase__				:  int		       =     SparkExamplesIterable(UpperCamelCase      ).shard_data_sources(worker_id=1	,      num_workers=2      )
  assert shard_it_a.n_shards == 2
  lowerCAmelCase__				:  Optional[int]		       =     _get_expected_row_ids_and_row_dicts_for_partition_order(UpperCamelCase	,      [1, 3]      )
  for i, (row_id, row_dict) in enumerate(UpperCamelCase      ):
    lowerCAmelCase__       ,			lowerCAmelCase__				:  Dict		       =     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			(					)     ->	Any:
  lowerCAmelCase__				:  Tuple		       =     pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
  lowerCAmelCase__				:  int		       =     spark.range(100      ).repartition(1      )
  lowerCAmelCase__				:  int		       =     Spark(UpperCamelCase      )
  # 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












lowerCAmelCase_					    = 6_55_21



def      __lowerCAmelCase			(					UpperCamelCase      )     ->	int:
  lowerCAmelCase__				:  List[str]		       =     1
  lowerCAmelCase__				:  List[Any]		       =     0
  for plain_chr in plain_text:
    lowerCAmelCase__				:  Union[str, Any]		       =     (a + ord(UpperCamelCase      )) % MOD_ADLER
    lowerCAmelCase__				:  Optional[Any]		       =     (b + a) % MOD_ADLER
  return (b << 16) | a

