



"""simple docstring"""






import multiprocessing
from typing import TYPE_CHECKING, Optional, Union

from .. import Dataset, Features, config
from ..formatting import query_table
from ..packaged_modules.sql.sql import Sql
from ..utils import logging
from .abc import AbstractDatasetInputStream


if TYPE_CHECKING:
	import sqlitea

	import sqlalchemy



class 			A__   (  lowerCAmelCase_							):




	'''simple docstring'''





	def __init__(     self:		List[Any]       ,			_SCREAMING_SNAKE_CASE:		Union[str, "sqlalchemy.sql.Selectable"]       ,			_SCREAMING_SNAKE_CASE:		Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]       ,			_SCREAMING_SNAKE_CASE:		Optional[Features] = None       ,			_SCREAMING_SNAKE_CASE:		str = None       ,			_SCREAMING_SNAKE_CASE:		bool = False       ,			**_SCREAMING_SNAKE_CASE:		int       ,			)				->       Dict:

				"""simple docstring"""

				super().__init__(features=snake_case_       ,			cache_dir=snake_case_       ,			keep_in_memory=snake_case_       ,			**snake_case_)
				__lowerCAmelCase      :  int												=       Sql(
				    cache_dir=snake_case_       ,			features=snake_case_       ,			sql=snake_case_       ,			con=snake_case_       ,			**snake_case_       ,			)







	def 							_SCREAMING_SNAKE_CASE   (     self:		Tuple)				->       Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase      :  List[Any]												=       None
				__lowerCAmelCase      :  str												=       None
				__lowerCAmelCase      :  Optional[Any]												=       None
				__lowerCAmelCase      :  int												=       None

				self.builder.download_and_prepare(
				    download_config=snake_case_       ,			download_mode=snake_case_       ,			verification_mode=snake_case_       ,			base_path=snake_case_       ,			)

				# Build dataset for splits
				__lowerCAmelCase      :  Tuple												=       self.builder.as_dataset(
				    split="train"       ,			verification_mode=snake_case_       ,			in_memory=self.keep_in_memory)
				return dataset






class 			A__   :




	'''simple docstring'''





	def __init__(     self:		List[Any]       ,			_SCREAMING_SNAKE_CASE:		Dataset       ,			_SCREAMING_SNAKE_CASE:		str       ,			_SCREAMING_SNAKE_CASE:		Union[str, "sqlalchemy.engine.Connection", "sqlalchemy.engine.Engine", "sqlite3.Connection"]       ,			_SCREAMING_SNAKE_CASE:		Optional[int] = None       ,			_SCREAMING_SNAKE_CASE:		Optional[int] = None       ,			**_SCREAMING_SNAKE_CASE:		Union[str, Any]       ,			)				->       Tuple:

				"""simple docstring"""

				if num_proc is not None and num_proc <= 0:
							raise ValueError(F"""num_proc {num_proc} must be an integer > 0.""")

				__lowerCAmelCase      :  int												=       dataset
				__lowerCAmelCase      :  List[str]												=       name
				__lowerCAmelCase      :  Tuple												=       con
				__lowerCAmelCase      :  int												=       batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
				__lowerCAmelCase      :  str												=       num_proc
				__lowerCAmelCase      :  Optional[int]												=       to_sql_kwargs

	def 							_SCREAMING_SNAKE_CASE   (     self:		Dict)				->       Tuple:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[int]												=       self.to_sql_kwargs.pop("sql"       ,			snake_case_)
				__lowerCAmelCase      :  List[Any]												=       self.to_sql_kwargs.pop("con"       ,			snake_case_)
				__lowerCAmelCase      :  Union[str, Any]												=       self.to_sql_kwargs.pop("index"       ,			snake_case_)

				__lowerCAmelCase      :  Optional[Any]												=       self._write(index=snake_case_       ,			**self.to_sql_kwargs)
				return written

	def 							_SCREAMING_SNAKE_CASE   (     self:		Tuple       ,			_SCREAMING_SNAKE_CASE:		Any)				->       Any:

				"""simple docstring"""

				__lowerCAmelCase						,	__lowerCAmelCase						,	__lowerCAmelCase      :  List[Any]												=       args
				__lowerCAmelCase      :  List[Any]												=       {**to_sql_kwargs, "if_exists": "append"} if offset > 0 else to_sql_kwargs
				__lowerCAmelCase      :  str												=       query_table(
				    table=self.dataset.data       ,			key=slice(snake_case_       ,			offset + self.batch_size)       ,			indices=self.dataset._indices       ,			)
				__lowerCAmelCase      :  str												=       batch.to_pandas()
				__lowerCAmelCase      :  Optional[Any]												=       df.to_sql(self.name       ,			self.con       ,			index=snake_case_       ,			**snake_case_)
				return num_rows or len(snake_case_)







	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[int]       ,			_SCREAMING_SNAKE_CASE:		Optional[Any]       ,			**_SCREAMING_SNAKE_CASE:		str)				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  str												=       0

				if self.num_proc is None or self.num_proc == 1:
							for offset in logging.tqdm(
							    range(0       ,			len(self.dataset)       ,			self.batch_size)       ,			unit="ba"       ,			disable=not logging.is_progress_bar_enabled()       ,			desc="Creating SQL from Arrow format"       ,			):
										written += self._batch_sql((offset, index, to_sql_kwargs))
				else:
							__lowerCAmelCase						,	__lowerCAmelCase      :  List[str]												=       len(self.dataset), self.batch_size
							with multiprocessing.Pool(self.num_proc) as pool:
										for num_rows in logging.tqdm(
										    pool.imap(
										        self._batch_sql       ,			[(offset, index, to_sql_kwargs) for offset in range(0       ,			snake_case_       ,			snake_case_)]       ,			)       ,			total=(num_rows // batch_size) + 1 if num_rows % batch_size else num_rows // batch_size       ,			unit="ba"       ,			disable=not logging.is_progress_bar_enabled()       ,			desc="Creating SQL from Arrow format"       ,			):
													written += num_rows

				return written






'''simple docstring'''




import pytest

from datasets.utils.sharding import _distribute_shards, _number_of_shards_in_gen_kwargs, _split_gen_kwargs
@pytest.mark.parametrize(
    "kwargs, expected"			,     [
        ({"num_shards": 0, "max_num_jobs": 1}, []),
        ({"num_shards": 10, "max_num_jobs": 1}, [range(10       )]),
        ({"num_shards": 10, "max_num_jobs": 10}, [range(__lowercase			,     i + 1       ) for i in range(10       )]),
        ({"num_shards": 1, "max_num_jobs": 10}, [range(1       )]),
        ({"num_shards": 10, "max_num_jobs": 3}, [range(0			,     4       ), range(4			,     7       ), range(7			,     10       )]),
        ({"num_shards": 3, "max_num_jobs": 10}, [range(0			,     1       ), range(1			,     2       ), range(2			,     3       )]),
    ]			,     )
def 	UpperCAmelCase_							(					__lowercase							:      int			,     __lowercase							:      Dict       )		->       List[Any]:


     '''simple docstring'''



     _UpperCAmelCase     							=   _distribute_shards(**__lowercase       )
     assert out == expected
@pytest.mark.parametrize(
    "gen_kwargs, max_num_jobs, expected"			,     [
        ({"foo": 0}, 10, [{"foo": 0}]),
        ({"shards": [0, 1, 2, 3]}, 1, [{"shards": [0, 1, 2, 3]}]),
        ({"shards": [0, 1, 2, 3]}, 4, [{"shards": [0]}, {"shards": [1]}, {"shards": [2]}, {"shards": [3]}]),
        ({"shards": [0, 1]}, 4, [{"shards": [0]}, {"shards": [1]}]),
        ({"shards": [0, 1, 2, 3]}, 2, [{"shards": [0, 1]}, {"shards": [2, 3]}]),
    ]			,     )
def 	UpperCAmelCase_							(					__lowercase							:      Dict			,     __lowercase							:      Optional[Any]			,     __lowercase							:      int       )		->       str:


     '''simple docstring'''



     _UpperCAmelCase     							=   _split_gen_kwargs(__lowercase			,     __lowercase       )
     assert out == expected






@pytest.mark.parametrize(
    "gen_kwargs, expected"			,     [
        ({"foo": 0}, 1),
        ({"shards": [0]}, 1),
        ({"shards": [0, 1, 2, 3]}, 4),
        ({"shards": [0, 1, 2, 3], "foo": 0}, 4),
        ({"shards": [0, 1, 2, 3], "other": (0, 1)}, 4),
        ({"shards": [0, 1, 2, 3], "shards2": [0, 1]}, RuntimeError),
    ]			,     )
def 	UpperCAmelCase_							(					__lowercase							:      Optional[Any]			,     __lowercase							:      List[Any]       )		->       List[Any]:


     '''simple docstring'''



     if expected is RuntimeError:
          with pytest.raises(__lowercase       ):
               _number_of_shards_in_gen_kwargs(__lowercase       )
     else:
          _UpperCAmelCase     							=   _number_of_shards_in_gen_kwargs(__lowercase       )
          assert out == expected




