






'''simple docstring'''







from typing import List, Optional, Tuple, Union

import torch
from torch import nn
from torch.nn import CrossEntropyLoss

from ... import AutoBackbone
from ...modeling_outputs import SemanticSegmenterOutput
from ...modeling_utils import PreTrainedModel
from ...utils import add_start_docstrings, add_start_docstrings_to_model_forward, replace_return_docstrings
from ...utils.backbone_utils import BackboneMixin
from .configuration_upernet import UperNetConfig


__lowerCAmelCase       		=      [
    'openmmlab/upernet-convnext-tiny',
    # See all UperNet models at https://huggingface.co/models?filter=upernet
]

# General docstring
__lowerCAmelCase       		=      'UperNetConfig'





class 	SCREAMING_SNAKE_CASE					(					nn.Module       ):






			def __init__( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			Union[int, Tuple[int, int]]  ,						__SCREAMING_SNAKE_CASE      :			Union[int, Tuple[int, int], str] = 0  ,						__SCREAMING_SNAKE_CASE      :			bool = False  ,						__SCREAMING_SNAKE_CASE      :			Union[int, Tuple[int, int]] = 1  ,						)			->			None:
							super().__init__()
							a_		:  Optional[Any]											=      nn.Convad(
							    in_channels=__SCREAMING_SNAKE_CASE  ,						out_channels=__SCREAMING_SNAKE_CASE  ,						kernel_size=__SCREAMING_SNAKE_CASE  ,						padding=__SCREAMING_SNAKE_CASE  ,						bias=__SCREAMING_SNAKE_CASE  ,						dilation=__SCREAMING_SNAKE_CASE  ,						)
							a_		:  Any											=      nn.BatchNormad(__SCREAMING_SNAKE_CASE					)
							a_		:  Union[str, Any]											=      nn.ReLU()







			def 			SCREAMING_SNAKE_CASE      ( self      :			Any  ,						__SCREAMING_SNAKE_CASE      :			torch.Tensor					)			->			torch.Tensor:
							a_		:  Any											=      self.conv(__SCREAMING_SNAKE_CASE					)
							a_		:  Dict											=      self.batch_norm(__SCREAMING_SNAKE_CASE					)
							a_		:  Optional[Any]											=      self.activation(__SCREAMING_SNAKE_CASE					)

							return output





class 	SCREAMING_SNAKE_CASE					(					nn.Module       ):






			def __init__( self      :			Dict  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			int					)			->			None:
							super().__init__()
							a_		:  Union[str, Any]											=      [
							    nn.AdaptiveAvgPoolad(__SCREAMING_SNAKE_CASE					),
							    UperNetConvModule(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE  ,						kernel_size=1					),
							]
							for i, layer in enumerate(self.layers					):
											self.add_module(str(__SCREAMING_SNAKE_CASE					)  ,						__SCREAMING_SNAKE_CASE					)







			def 			SCREAMING_SNAKE_CASE      ( self      :			str  ,						__SCREAMING_SNAKE_CASE      :			torch.Tensor					)			->			torch.Tensor:
							a_		:  str											=      input
							for layer in self.layers:
											a_		:  List[Any]											=      layer(__SCREAMING_SNAKE_CASE					)
							return hidden_state





class 	SCREAMING_SNAKE_CASE					(					nn.Module       ):






			def __init__( self      :			List[Any]  ,						__SCREAMING_SNAKE_CASE      :			Tuple[int, ...]  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			bool					)			->			None:
							super().__init__()
							a_		:  Optional[int]											=      pool_scales
							a_		:  Any											=      align_corners
							a_		:  Optional[int]											=      in_channels
							a_		:  int											=      channels
							a_		:  Dict											=      []
							for i, pool_scale in enumerate(__SCREAMING_SNAKE_CASE					):
											a_		:  Any											=      UperNetPyramidPoolingBlock(pool_scale=__SCREAMING_SNAKE_CASE  ,						in_channels=__SCREAMING_SNAKE_CASE  ,						channels=__SCREAMING_SNAKE_CASE					)
											self.blocks.append(__SCREAMING_SNAKE_CASE					)
											self.add_module(str(__SCREAMING_SNAKE_CASE					)  ,						__SCREAMING_SNAKE_CASE					)







			def 			SCREAMING_SNAKE_CASE      ( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			torch.Tensor					)			->			List[torch.Tensor]:
							a_		:  List[Any]											=      []
							for ppm in self.blocks:
											a_		:  str											=      ppm(__SCREAMING_SNAKE_CASE					)
											a_		:  Dict											=      nn.functional.interpolate(
											    __SCREAMING_SNAKE_CASE  ,						size=x.size()[2:]  ,						mode='''bilinear'''  ,						align_corners=self.align_corners					)
											ppm_outs.append(__SCREAMING_SNAKE_CASE					)
							return ppm_outs





class 	SCREAMING_SNAKE_CASE					(					nn.Module       ):






			def __init__( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			Union[str, Any]					)			->			Union[str, Any]:
							super().__init__()

							a_		:  Optional[Any]											=      config
							a_		:  List[str]											=      config.pool_scales  # e.g. (1, 2, 3, 6)
							a_		:  str											=      in_channels
							a_		:  Optional[int]											=      config.hidden_size
							a_		:  str											=      False
							a_		:  Optional[Any]											=      nn.Convad(self.channels  ,						config.num_labels  ,						kernel_size=1					)

							# PSP Module
							a_		:  Optional[Any]											=      UperNetPyramidPoolingModule(
							    self.pool_scales  ,						self.in_channels[-1]  ,						self.channels  ,						align_corners=self.align_corners  ,						)
							a_		:  Optional[Any]											=      UperNetConvModule(
							    self.in_channels[-1] + len(self.pool_scales					) * self.channels  ,						self.channels  ,						kernel_size=3  ,						padding=1  ,						)
							# FPN Module
							a_		:  Tuple											=      nn.ModuleList()
							a_		:  Any											=      nn.ModuleList()
							for in_channels in self.in_channels[:-1]:  # skip the top layer
											a_		:  Dict											=      UperNetConvModule(__SCREAMING_SNAKE_CASE  ,						self.channels  ,						kernel_size=1					)
											a_		:  List[Any]											=      UperNetConvModule(self.channels  ,						self.channels  ,						kernel_size=3  ,						padding=1					)
											self.lateral_convs.append(__SCREAMING_SNAKE_CASE					)
											self.fpn_convs.append(__SCREAMING_SNAKE_CASE					)

							a_		:  Optional[int]											=      UperNetConvModule(
							    len(self.in_channels					) * self.channels  ,						self.channels  ,						kernel_size=3  ,						padding=1  ,						)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Any					)			->			str:
							self.apply(self._init_weights					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Any  ,						__SCREAMING_SNAKE_CASE      :			str					)			->			Any:
							if isinstance(__SCREAMING_SNAKE_CASE  ,						nn.Convad					):
											module.weight.data.normal_(mean=0.0  ,						std=self.config.initializer_range					)
											if module.bias is not None:
															module.bias.data.zero_()






			def 			SCREAMING_SNAKE_CASE      ( self      :			Optional[int]  ,						__SCREAMING_SNAKE_CASE      :			int					)			->			List[Any]:
							a_		:  Tuple											=      inputs[-1]
							a_		:  List[str]											=      [x]
							psp_outs.extend(self.psp_modules(__SCREAMING_SNAKE_CASE					)					)
							a_		:  Dict											=      torch.cat(__SCREAMING_SNAKE_CASE  ,						dim=1					)
							a_		:  Union[str, Any]											=      self.bottleneck(__SCREAMING_SNAKE_CASE					)

							return output







			def 			SCREAMING_SNAKE_CASE      ( self      :			List[str]  ,						__SCREAMING_SNAKE_CASE      :			torch.Tensor					)			->			torch.Tensor:
							# build laterals
							a_		:  Dict											=      [lateral_conv(encoder_hidden_states[i]					) for i, lateral_conv in enumerate(self.lateral_convs					)]

							laterals.append(self.psp_forward(__SCREAMING_SNAKE_CASE					)					)

							# build top-down path
							a_		:  Any											=      len(__SCREAMING_SNAKE_CASE					)
							for i in range(used_backbone_levels - 1  ,						0  ,						-1					):
											a_		:  Optional[Any]											=      laterals[i - 1].shape[2:]
											a_		:  List[Any]											=      laterals[i - 1] + nn.functional.interpolate(
											    laterals[i]  ,						size=__SCREAMING_SNAKE_CASE  ,						mode='''bilinear'''  ,						align_corners=self.align_corners					)

							# build outputs
							a_		:  Tuple											=      [self.fpn_convs[i](laterals[i]					) for i in range(used_backbone_levels - 1					)]
							# append psp feature
							fpn_outs.append(laterals[-1]					)

							for i in range(used_backbone_levels - 1  ,						0  ,						-1					):
											a_		:  str											=      nn.functional.interpolate(
											    fpn_outs[i]  ,						size=fpn_outs[0].shape[2:]  ,						mode='''bilinear'''  ,						align_corners=self.align_corners					)
							a_		:  List[Any]											=      torch.cat(__SCREAMING_SNAKE_CASE  ,						dim=1					)
							a_		:  List[str]											=      self.fpn_bottleneck(__SCREAMING_SNAKE_CASE					)
							a_		:  str											=      self.classifier(__SCREAMING_SNAKE_CASE					)

							return output





class 	SCREAMING_SNAKE_CASE					(					nn.Module       ):






			def __init__( self      :			Optional[int]  ,						__SCREAMING_SNAKE_CASE      :			int  ,						__SCREAMING_SNAKE_CASE      :			int = 2  ,						__SCREAMING_SNAKE_CASE      :			int = 3  ,						__SCREAMING_SNAKE_CASE      :			Union[int, Tuple[int, int]] = 1					)			->			None:
							super().__init__()

							a_		:  Any											=      config
							a_		:  Optional[Any]											=      config.auxiliary_in_channels
							a_		:  Optional[Any]											=      config.auxiliary_channels
							a_		:  Tuple											=      config.auxiliary_num_convs
							a_		:  Tuple											=      config.auxiliary_concat_input
							a_		:  Dict											=      in_index

							a_		:  str											=      (kernel_size // 2) * dilation
							a_		:  List[str]											=      []
							convs.append(
							    UperNetConvModule(
							        self.in_channels  ,						self.channels  ,						kernel_size=__SCREAMING_SNAKE_CASE  ,						padding=__SCREAMING_SNAKE_CASE  ,						dilation=__SCREAMING_SNAKE_CASE					)					)
							for i in range(self.num_convs - 1					):
											convs.append(
											    UperNetConvModule(
											        self.channels  ,						self.channels  ,						kernel_size=__SCREAMING_SNAKE_CASE  ,						padding=__SCREAMING_SNAKE_CASE  ,						dilation=__SCREAMING_SNAKE_CASE					)					)
							if self.num_convs == 0:
											a_		:  Optional[int]											=      nn.Identity()
							else:
											a_		:  Optional[int]											=      nn.Sequential(*__SCREAMING_SNAKE_CASE					)
							if self.concat_input:
											a_		:  str											=      UperNetConvModule(
											    self.in_channels + self.channels  ,						self.channels  ,						kernel_size=__SCREAMING_SNAKE_CASE  ,						padding=kernel_size // 2					)

							a_		:  int											=      nn.Convad(self.channels  ,						config.num_labels  ,						kernel_size=1					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			List[Any]					)			->			int:
							self.apply(self._init_weights					)






			def 			SCREAMING_SNAKE_CASE      ( self      :			Union[str, Any]  ,						__SCREAMING_SNAKE_CASE      :			Optional[Any]					)			->			Any:
							if isinstance(__SCREAMING_SNAKE_CASE  ,						nn.Convad					):
											module.weight.data.normal_(mean=0.0  ,						std=self.config.initializer_range					)
											if module.bias is not None:
															module.bias.data.zero_()







			def 			SCREAMING_SNAKE_CASE      ( self      :			int  ,						__SCREAMING_SNAKE_CASE      :			torch.Tensor					)			->			torch.Tensor:
							# just take the relevant feature maps
							a_		:  str											=      encoder_hidden_states[self.in_index]
							a_		:  Optional[int]											=      self.convs(__SCREAMING_SNAKE_CASE					)
							if self.concat_input:
											a_		:  int											=      self.conv_cat(torch.cat([hidden_states, output]  ,						dim=1					)					)
							a_		:  str											=      self.classifier(__SCREAMING_SNAKE_CASE					)
							return output





class 	SCREAMING_SNAKE_CASE					(					SCREAMING_SNAKE_CASE_       ):
			snake_case__  =				UperNetConfig
			snake_case__  =				"pixel_values"
			snake_case__  =				True






			def 			SCREAMING_SNAKE_CASE      ( self      :			Optional[Any]  ,						__SCREAMING_SNAKE_CASE      :			List[Any]					)			->			Union[str, Any]:
							if isinstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					):
											module.backbone.init_weights()
											module.decode_head.init_weights()
											module.auxiliary_head.init_weights()






			def 			SCREAMING_SNAKE_CASE      ( self      :			Dict					)			->			Optional[Any]:
							self.backbone.init_weights()
							self.decode_head.init_weights()
							self.auxiliary_head.init_weights()







			def 			SCREAMING_SNAKE_CASE      ( self      :			str  ,						__SCREAMING_SNAKE_CASE      :			List[Any]  ,						__SCREAMING_SNAKE_CASE      :			Optional[int]=False					)			->			str:
							if isinstance(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					):
											a_		:  Optional[Any]											=      value





__lowerCAmelCase       		=      r'\n    Parameters:\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use\n    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n        config ([`UperNetConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'

__lowerCAmelCase       		=      r'\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Padding will be ignored by default should you provide it. Pixel values can be obtained using\n            [`AutoImageProcessor`]. See [`SegformerImageProcessor.__call__`] for details.\n        output_attentions (`bool`, *optional*):\n            Whether or not to return the attentions tensors of all attention layers in case the backbone has them. See\n            `attentions` under returned tensors for more detail.\n        output_hidden_states (`bool`, *optional*):\n            Whether or not to return the hidden states of all layers of the backbone. See `hidden_states` under\n            returned tensors for more detail.\n        return_dict (`bool`, *optional*):\n            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.\n'





@add_start_docstrings(
    "UperNet framework leveraging any vision backbone e.g. for ADE20k, CityScapes."	,   SCREAMING_SNAKE_CASE_	,   )
class 	SCREAMING_SNAKE_CASE					(					SCREAMING_SNAKE_CASE_       ):






			def __init__( self      :			Any  ,						__SCREAMING_SNAKE_CASE      :			Optional[int]					)			->			Union[str, Any]:
							super().__init__(__SCREAMING_SNAKE_CASE					)

							a_		:  Optional[int]											=      AutoBackbone.from_config(config.backbone_config					)

							# Semantic segmentation head(s)
							a_		:  List[str]											=      UperNetHead(__SCREAMING_SNAKE_CASE  ,						in_channels=self.backbone.channels					)
							a_		:  int											=      UperNetFCNHead(__SCREAMING_SNAKE_CASE					) if config.use_auxiliary_head else None

							# Initialize weights and apply final processing
							self.post_init()







			@add_start_docstrings_to_model_forward(UPERNET_INPUTS_DOCSTRING.format('''batch_size, sequence_length'''					)					)
			@replace_return_docstrings(output_type=__SCREAMING_SNAKE_CASE  ,						config_class=_CONFIG_FOR_DOC					)
			def 			SCREAMING_SNAKE_CASE      ( self      :			Union[str, Any]  ,						__SCREAMING_SNAKE_CASE      :			Optional[torch.Tensor] = None  ,						__SCREAMING_SNAKE_CASE      :			Optional[bool] = None  ,						__SCREAMING_SNAKE_CASE      :			Optional[bool] = None  ,						__SCREAMING_SNAKE_CASE      :			Optional[torch.Tensor] = None  ,						__SCREAMING_SNAKE_CASE      :			Optional[bool] = None  ,						)			->			Union[tuple, SemanticSegmenterOutput]:
							a_		:  Optional[int]											=      return_dict if return_dict is not None else self.config.use_return_dict
							a_		:  Any											=      (
							    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
							)
							a_		:  List[str]											=      output_attentions if output_attentions is not None else self.config.output_attentions

							a_		:  Union[str, Any]											=      self.backbone.forward_with_filtered_kwargs(
							    __SCREAMING_SNAKE_CASE  ,						output_hidden_states=__SCREAMING_SNAKE_CASE  ,						output_attentions=__SCREAMING_SNAKE_CASE					)
							a_		:  Optional[Any]											=      outputs.feature_maps

							a_		:  Optional[Any]											=      self.decode_head(__SCREAMING_SNAKE_CASE					)
							a_		:  Optional[int]											=      nn.functional.interpolate(__SCREAMING_SNAKE_CASE  ,						size=pixel_values.shape[2:]  ,						mode='''bilinear'''  ,						align_corners=__SCREAMING_SNAKE_CASE					)

							a_		:  Any											=      None
							if self.auxiliary_head is not None:
											a_		:  Optional[int]											=      self.auxiliary_head(__SCREAMING_SNAKE_CASE					)
											a_		:  Optional[Any]											=      nn.functional.interpolate(
											    __SCREAMING_SNAKE_CASE  ,						size=pixel_values.shape[2:]  ,						mode='''bilinear'''  ,						align_corners=__SCREAMING_SNAKE_CASE					)

							a_		:  str											=      None
							if labels is not None:
											if self.config.num_labels == 1:
															raise ValueError('''The number of labels should be greater than one'''					)
											else:
															# compute weighted loss
															a_		:  List[str]											=      CrossEntropyLoss(ignore_index=self.config.loss_ignore_index					)
															a_		:  List[Any]											=      loss_fct(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
															a_		:  Tuple											=      loss_fct(__SCREAMING_SNAKE_CASE  ,						__SCREAMING_SNAKE_CASE					)
															a_		:  Dict											=      main_loss + self.config.auxiliary_loss_weight * auxiliary_loss

							if not return_dict:
											if output_hidden_states:
															a_		:  Union[str, Any]											=      (logits,) + outputs[1:]
											else:
															a_		:  List[Any]											=      (logits,) + outputs[2:]
											return ((loss,) + output) if loss is not None else output

							return SemanticSegmenterOutput(
							    loss=__SCREAMING_SNAKE_CASE  ,						logits=__SCREAMING_SNAKE_CASE  ,						hidden_states=outputs.hidden_states  ,						attentions=outputs.attentions  ,						)












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

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





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

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

				# Partitions 0 and 2
				a_		:  Dict											=      SparkExamplesIterable(__A      ).shard_data_sources(worker_id=0						,		num_workers=2      )
				assert shard_it_a.n_shards == 2
				a_		:  Optional[Any]											=      _get_expected_row_ids_and_row_dicts_for_partition_order(__A						,		[0, 2]      )
				for i, (row_id, row_dict) in enumerate(__A      ):
								a_				,			a_		:  Tuple											=      expected_row_ids_and_row_dicts_a[i]
								assert row_id == expected_row_id
								assert row_dict == expected_row_dict

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




