


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 						_A						(						_lowercase ,   _lowercase      ) ->      Any:


     """simple docstring"""


     __UpperCamelCase      =       []
     for part_id in partition_order:
          __UpperCamelCase      =       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 						_A						(						) ->      Union[str, Any]:


     """simple docstring"""


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


     """simple docstring"""


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

     for i, (row_id, row_dict) in enumerate(generate_fn()      ):
          __UpperCamelCase      =       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 						_A						(						) ->      int:


     """simple docstring"""


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


     """simple docstring"""


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

          __UpperCamelCase      =       SparkExamplesIterable(_UpperCAmelCase      ).shuffle_data_sources(_UpperCAmelCase      )
          assert shuffled_it.n_shards == 3
          for i, (row_id, row_dict) in enumerate(_UpperCAmelCase      ):
               __UpperCamelCase      =       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 						_A						(						) ->      List[str]:


     """simple docstring"""


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

     # Partitions 0 and 2
     __UpperCamelCase      =       SparkExamplesIterable(_UpperCAmelCase      ).shard_data_sources(worker_id=0 ,   num_workers=2      )
     assert shard_it_a.n_shards == 2
     __UpperCamelCase      =       _get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase ,   [0, 2]      )
     for i, (row_id, row_dict) in enumerate(_UpperCAmelCase      ):
          __UpperCamelCase      =       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      =       SparkExamplesIterable(_UpperCAmelCase      ).shard_data_sources(worker_id=1 ,   num_workers=2      )
     assert shard_it_a.n_shards == 2
     __UpperCamelCase      =       _get_expected_row_ids_and_row_dicts_for_partition_order(_UpperCAmelCase ,   [1, 3]      )
     for i, (row_id, row_dict) in enumerate(_UpperCAmelCase      ):
          __UpperCamelCase      =       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 						_A						(						) ->      Tuple:


     """simple docstring"""


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






'''simple docstring'''







import contextlib
import os
import sqlitea

import pytest

from datasets import Dataset, Features, Value
from datasets.io.sql import SqlDatasetReader, SqlDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases, require_sqlalchemy


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Any    ,_UpperCAmelCase	:	str  )  ->			Dict:
       assert isinstance(_UpperCAmelCase    ,_UpperCAmelCase  )
       assert dataset.num_rows == 4
       assert dataset.num_columns == 3
       assert dataset.column_names == ["col_1", "col_2", "col_3"]
       for feature, expected_dtype in expected_features.items():
              assert dataset.features[feature].dtype == expected_dtype


@require_sqlalchemy
@pytest.mark.parametrize("""keep_in_memory"""    ,[False, True]  )
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]    ,_UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	str  )  ->			Optional[Any]:
       _a    : Any     					=tmp_path / """cache"""
       _a    : int     					={"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
       with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
              _a    : Tuple     					=SqlDatasetReader(
                  """dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase    ,keep_in_memory=_UpperCAmelCase  ).read()
       _check_sql_dataset(_UpperCAmelCase    ,_UpperCAmelCase  )


@require_sqlalchemy
@pytest.mark.parametrize(
    """features"""    ,[
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]    ,)
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	List[Any]    ,_UpperCAmelCase	:	Dict    ,_UpperCAmelCase	:	Optional[Any]    ,_UpperCAmelCase	:	int  )  ->			List[Any]:
       _a    : Union[str, Any]     					=tmp_path / """cache"""
       _a    : str     					={"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""}
       _a    : Optional[int]     					=features.copy() if features else default_expected_features
       _a    : Union[str, Any]     					=(
           Features({feature: Value(_UpperCAmelCase  ) for feature, dtype in features.items()}  ) if features is not None else None
       )
       _a    : Optional[Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,features=_UpperCAmelCase    ,cache_dir=_UpperCAmelCase  ).read()
       _check_sql_dataset(_UpperCAmelCase    ,_UpperCAmelCase  )


def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[Any]  )  ->			List[str]:
       with contextlib.closing(sqlitea.connect(_UpperCAmelCase  )  ) as con:
              _a    : Any     					=con.cursor()
              cur.execute("""SELECT * FROM dataset"""  )
              for row in cur:
                     yield row


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Dict    ,_UpperCAmelCase	:	Tuple    ,_UpperCAmelCase	:	List[str]  )  ->			Union[str, Any]:
       _a    : Union[str, Any]     					=tmp_path / """cache"""
       _a    : Union[str, Any]     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Tuple     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=1  ).write()

       _a    : Tuple     					=iter_sql_file(_UpperCAmelCase  )
       _a    : List[Any]     					=iter_sql_file(_UpperCAmelCase  )

       for rowa, rowa in zip(_UpperCAmelCase    ,_UpperCAmelCase  ):
              assert rowa == rowa


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Optional[int]    ,_UpperCAmelCase	:	Any    ,_UpperCAmelCase	:	List[Any]  )  ->			Optional[int]:
       _a    : int     					=tmp_path / """cache"""
       _a    : Any     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Union[str, Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=2  ).write()

       _a    : List[Any]     					=iter_sql_file(_UpperCAmelCase  )
       _a    : str     					=iter_sql_file(_UpperCAmelCase  )

       for rowa, rowa in zip(_UpperCAmelCase    ,_UpperCAmelCase  ):
              assert rowa == rowa


@require_sqlalchemy
def 	SCREAMING_SNAKE_CASE_					(  _UpperCAmelCase	:	Union[str, Any]    ,_UpperCAmelCase	:	str    ,_UpperCAmelCase	:	List[Any]  )  ->			List[str]:
       _a    : List[str]     					=tmp_path / """cache"""
       _a    : Dict     					=os.path.join(_UpperCAmelCase    ,"""tmp.sql"""  )
       _a    : Optional[Any]     					=SqlDatasetReader("""dataset"""    ,"""sqlite:///""" + sqlite_path    ,cache_dir=_UpperCAmelCase  ).read()
       with pytest.raises(_UpperCAmelCase  ):
              SqlDatasetWriter(_UpperCAmelCase    ,"""dataset"""    ,"""sqlite:///""" + output_sqlite_path    ,num_proc=0  ).write()


