



'''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 							SCREAMING_SNAKE_CASE(				SCREAMING_SNAKE_CASE__     ):







	"""simple docstring"""






	def __init__(		self     :	Dict						,     __snake_case     :	List[Any]						,     __snake_case     :	List[Any]						,     __snake_case     :	Tuple = None						,     __snake_case     :	Optional[int] = None						,     __snake_case     :	int = False						,     **__snake_case     :	Optional[int]						,     )     ->						int:
								super().__init__(features=__snake_case						,     cache_dir=__snake_case						,     keep_in_memory=__snake_case						,     **__snake_case						)
								UpperCAmelCase					:			Union[str, Any]									=						Sql(
								    cache_dir=__snake_case						,     features=__snake_case						,     sql=__snake_case						,     con=__snake_case						,     **__snake_case						,     )







	def 	A      (		self     :	List[Any]						)     ->						Tuple:
								UpperCAmelCase					:			Dict									=						None
								UpperCAmelCase					:			Any									=						None
								UpperCAmelCase					:			List[str]									=						None
								UpperCAmelCase					:			Optional[Any]									=						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
								UpperCAmelCase					:			str									=						self.builder.as_dataset(
								    split='''train'''						,     verification_mode=__snake_case						,     in_memory=self.keep_in_memory						)
								return dataset

class 							SCREAMING_SNAKE_CASE:







	"""simple docstring"""






	def __init__(		self     :	Any						,     __snake_case     :	Optional[int]						,     __snake_case     :	str						,     __snake_case     :	Tuple						,     __snake_case     :	Tuple = None						,     __snake_case     :	Optional[int] = None						,     **__snake_case     :	Any						,     )     ->						str:



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

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







	def 	A      (		self     :	Optional[Any]						)     ->						int:
								UpperCAmelCase					:			Union[str, Any]									=						self.to_sql_kwargs.pop('''sql'''						,     __snake_case						)
								UpperCAmelCase					:			Any									=						self.to_sql_kwargs.pop('''con'''						,     __snake_case						)
								UpperCAmelCase					:			Dict									=						self.to_sql_kwargs.pop('''index'''						,     __snake_case						)

								UpperCAmelCase					:			Tuple									=						self._write(index=__snake_case						,     **self.to_sql_kwargs						)
								return written







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







	def 	A      (		self     :	List[Any]						,     __snake_case     :	Optional[int]						,     **__snake_case     :	int						)     ->						int:
								UpperCAmelCase					:			Any									=						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:
															UpperCAmelCase				,						UpperCAmelCase					:			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











from packaging import version

from .import_utils import is_accelerate_available


if is_accelerate_available():
				import accelerate

def lowercase    (  SCREAMING_SNAKE_CASE )			->							Union[str, Any]:


			'''simple docstring'''







			if not is_accelerate_available():
						return method
			SCREAMING_SNAKE_CASE_								=						version.parse(accelerate.__version__ ).base_version
			if version.parse(SCREAMING_SNAKE_CASE ) < version.parse('0.17.0' ):
						return method

			def wrapper(self       ,    *SCREAMING_SNAKE_CASE       ,    **SCREAMING_SNAKE_CASE ):
						if hasattr(self       ,    '_hf_hook' ) and hasattr(self._hf_hook       ,    'pre_forward' ):
									self._hf_hook.pre_forward(self )
						return method(self       ,    *SCREAMING_SNAKE_CASE       ,    **SCREAMING_SNAKE_CASE )

			return wrapper
