



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 				snake_case_(_UpperCamelCase			,	_UpperCamelCase					)							->     Dict:



  """simple docstring"""






  _snake_case						=							[]
  for part_id in partition_order:
    _snake_case						=							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 				snake_case_()							->     Any:



  """simple docstring"""






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



  """simple docstring"""






  _snake_case						=							pyspark.sql.SparkSession.builder.master('''local[*]'''					).appName('''pyspark'''					).getOrCreate()
  _snake_case						=							spark.range(10					).repartition(2					)
  _snake_case						=							[1, 0]
  _snake_case						=							_generate_iterable_examples(_UpperCamelCase			,	_UpperCamelCase					)  # Reverse the partitions.
  _snake_case						=							_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCamelCase			,	_UpperCamelCase					)

  for i, (row_id, row_dict) in enumerate(generate_fn()					):
    _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 				snake_case_()							->     str:



  """simple docstring"""






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



  """simple docstring"""






  _snake_case						=							pyspark.sql.SparkSession.builder.master('''local[*]'''					).appName('''pyspark'''					).getOrCreate()
  _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:
    _snake_case						=							lambda _UpperCamelCase					: x.reverse()
    _snake_case						=							_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCamelCase			,	[2, 1, 0]					)

    _snake_case						=							SparkExamplesIterable(_UpperCamelCase					).shuffle_data_sources(_UpperCamelCase					)
    assert shuffled_it.n_shards == 3
    for i, (row_id, row_dict) in enumerate(_UpperCamelCase					):
      _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 				snake_case_()							->     Dict:



  """simple docstring"""






  _snake_case						=							pyspark.sql.SparkSession.builder.master('''local[*]'''					).appName('''pyspark'''					).getOrCreate()
  _snake_case						=							spark.range(20					).repartition(4					)

  # Partitions 0 and 2
  _snake_case						=							SparkExamplesIterable(_UpperCamelCase					).shard_data_sources(worker_id=0			,	num_workers=2					)
  assert shard_it_a.n_shards == 2
  _snake_case						=							_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCamelCase			,	[0, 2]					)
  for i, (row_id, row_dict) in enumerate(_UpperCamelCase					):
    _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
  _snake_case						=							SparkExamplesIterable(_UpperCamelCase					).shard_data_sources(worker_id=1			,	num_workers=2					)
  assert shard_it_a.n_shards == 2
  _snake_case						=							_get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCamelCase			,	[1, 3]					)
  for i, (row_id, row_dict) in enumerate(_UpperCamelCase					):
    _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 				snake_case_()							->     Optional[int]:



  """simple docstring"""






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




# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


__A      		=     {'''configuration_mra''': ['''MRA_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''MraConfig''']}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __A      		=     [
     '''MRA_PRETRAINED_MODEL_ARCHIVE_LIST''',
     '''MraForMaskedLM''',
     '''MraForMultipleChoice''',
     '''MraForQuestionAnswering''',
     '''MraForSequenceClassification''',
     '''MraForTokenClassification''',
     '''MraLayer''',
     '''MraModel''',
     '''MraPreTrainedModel''',
 ]


if TYPE_CHECKING:
 from .configuration_mra import MRA_PRETRAINED_CONFIG_ARCHIVE_MAP, MraConfig

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_mra import (
      MRA_PRETRAINED_MODEL_ARCHIVE_LIST,
      MraForMaskedLM,
      MraForMultipleChoice,
      MraForQuestionAnswering,
      MraForSequenceClassification,
      MraForTokenClassification,
      MraLayer,
      MraModel,
      MraPreTrainedModel,
  )
else:
 import sys

 __A      		=     _LazyModule(__name__, globals()['''__file__'''], _import_structure)

