






import os
from typing import BinaryIO, Optional, Union

import numpy as np
import pyarrow.parquet as pq

from .. import Audio, Dataset, Features, Image, NamedSplit, Value, config
from ..features.features import FeatureType, _visit
from ..formatting import query_table
from ..packaged_modules import _PACKAGED_DATASETS_MODULES
from ..packaged_modules.parquet.parquet import Parquet
from ..utils import logging
from ..utils.typing import NestedDataStructureLike, PathLike
from .abc import AbstractDatasetReader





def 	lowerCAmelCase_		(lowercase__						:       Features       )		->    Optional[int]:


				'''simple docstring'''




				lowerCAmelCase__      			=						np.inf

				def set_batch_size(lowercase__						:       FeatureType       ) -> None:
								nonlocal batch_size
								if isinstance(lowercase__     ,					lowercase__       ):
												lowerCAmelCase__      			=						min(lowercase__     ,					config.PARQUET_ROW_GROUP_SIZE_FOR_IMAGE_DATASETS       )
								elif isinstance(lowercase__     ,					lowercase__       ):
												lowerCAmelCase__      			=						min(lowercase__     ,					config.PARQUET_ROW_GROUP_SIZE_FOR_AUDIO_DATASETS       )
								elif isinstance(lowercase__     ,					lowercase__       ) and feature.dtype == "binary":
												lowerCAmelCase__      			=						min(lowercase__     ,					config.PARQUET_ROW_GROUP_SIZE_FOR_BINARY_DATASETS       )

				_visit(lowercase__     ,					lowercase__       )

				return None if batch_size is np.inf else batch_size
class  lowerCAmelCase_       (				snake_case__					):




							def __init__(   self     :   List[str] ,							SCREAMING_SNAKE_CASE_     :   NestedDataStructureLike[PathLike] ,							SCREAMING_SNAKE_CASE_     :   Optional[NamedSplit] = None ,							SCREAMING_SNAKE_CASE_     :   Optional[Features] = None ,							SCREAMING_SNAKE_CASE_     :   str = None ,							SCREAMING_SNAKE_CASE_     :   bool = False ,							SCREAMING_SNAKE_CASE_     :   bool = False ,							SCREAMING_SNAKE_CASE_     :   Optional[int] = None ,							**SCREAMING_SNAKE_CASE_     :   Any ,							):
											super().__init__(
											    SCREAMING_SNAKE_CASE_ ,							split=SCREAMING_SNAKE_CASE_ ,							features=SCREAMING_SNAKE_CASE_ ,							cache_dir=SCREAMING_SNAKE_CASE_ ,							keep_in_memory=SCREAMING_SNAKE_CASE_ ,							streaming=SCREAMING_SNAKE_CASE_ ,							num_proc=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_ ,							)
											lowerCAmelCase__      			=						path_or_paths if isinstance(SCREAMING_SNAKE_CASE_ ,							SCREAMING_SNAKE_CASE_	) else {self.split: path_or_paths}
											lowerCAmelCase__      			=						_PACKAGED_DATASETS_MODULES['''parquet'''][1]
											lowerCAmelCase__      			=						Parquet(
											    cache_dir=SCREAMING_SNAKE_CASE_ ,							data_files=SCREAMING_SNAKE_CASE_ ,							features=SCREAMING_SNAKE_CASE_ ,							hash=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_ ,							)




							def 						__snake_case					(   self     :   Optional[Any]	):
											# Build iterable dataset
											if self.streaming:
															lowerCAmelCase__      			=						self.builder.as_streaming_dataset(split=self.split	)
											# Build regular (map-style) dataset
											else:
															lowerCAmelCase__      			=						None
															lowerCAmelCase__      			=						None
															lowerCAmelCase__      			=						None
															lowerCAmelCase__      			=						None

															self.builder.download_and_prepare(
															    download_config=SCREAMING_SNAKE_CASE_ ,							download_mode=SCREAMING_SNAKE_CASE_ ,							verification_mode=SCREAMING_SNAKE_CASE_ ,							base_path=SCREAMING_SNAKE_CASE_ ,							num_proc=self.num_proc ,							)
															lowerCAmelCase__      			=						self.builder.as_dataset(
															    split=self.split ,							verification_mode=SCREAMING_SNAKE_CASE_ ,							in_memory=self.keep_in_memory	)
											return dataset



class  lowerCAmelCase_       :




							def __init__(   self     :   Any ,							SCREAMING_SNAKE_CASE_     :   Dataset ,							SCREAMING_SNAKE_CASE_     :   Union[PathLike, BinaryIO] ,							SCREAMING_SNAKE_CASE_     :   Optional[int] = None ,							**SCREAMING_SNAKE_CASE_     :   Optional[int] ,							):
											lowerCAmelCase__      			=						dataset
											lowerCAmelCase__      			=						path_or_buf
											lowerCAmelCase__      			=						batch_size or get_writer_batch_size(dataset.features	)
											lowerCAmelCase__      			=						parquet_writer_kwargs




							def 						__snake_case					(   self     :   List[str]	):
											lowerCAmelCase__      			=						self.batch_size if self.batch_size else config.DEFAULT_MAX_BATCH_SIZE

											if isinstance(self.path_or_buf ,							(str, bytes, os.PathLike)	):
															with open(self.path_or_buf ,							'''wb+'''	) as buffer:
																			lowerCAmelCase__      			=						self._write(file_obj=SCREAMING_SNAKE_CASE_ ,							batch_size=SCREAMING_SNAKE_CASE_ ,							**self.parquet_writer_kwargs	)
											else:
															lowerCAmelCase__      			=						self._write(file_obj=self.path_or_buf ,							batch_size=SCREAMING_SNAKE_CASE_ ,							**self.parquet_writer_kwargs	)
											return written




							def 						__snake_case					(   self     :   Union[str, Any] ,							SCREAMING_SNAKE_CASE_     :   BinaryIO ,							SCREAMING_SNAKE_CASE_     :   int ,							**SCREAMING_SNAKE_CASE_     :   Union[str, Any]	):
											lowerCAmelCase__      			=						0
											lowerCAmelCase__      			=						parquet_writer_kwargs.pop('''path_or_buf''' ,							SCREAMING_SNAKE_CASE_	)
											lowerCAmelCase__      			=						self.dataset.features.arrow_schema

											lowerCAmelCase__      			=						pq.ParquetWriter(SCREAMING_SNAKE_CASE_ ,							schema=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_	)

											for offset in logging.tqdm(
											    range(0 ,							len(self.dataset	) ,							SCREAMING_SNAKE_CASE_	) ,							unit='''ba''' ,							disable=not logging.is_progress_bar_enabled() ,							desc='''Creating parquet from Arrow format''' ,							):
															lowerCAmelCase__      			=						query_table(
															    table=self.dataset._data ,							key=slice(SCREAMING_SNAKE_CASE_ ,							offset + batch_size	) ,							indices=self.dataset._indices if self.dataset._indices is not None else None ,							)
															writer.write_table(SCREAMING_SNAKE_CASE_	)
															written += batch.nbytes
											writer.close()
											return written













from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader
class  lowerCAmelCase_       (				snake_case__					):




							def __init__(   self     :   str ,							SCREAMING_SNAKE_CASE_     :   pyspark.sql.DataFrame ,							SCREAMING_SNAKE_CASE_     :   Optional[NamedSplit] = None ,							SCREAMING_SNAKE_CASE_     :   Optional[Features] = None ,							SCREAMING_SNAKE_CASE_     :   bool = True ,							SCREAMING_SNAKE_CASE_     :   str = None ,							SCREAMING_SNAKE_CASE_     :   bool = False ,							SCREAMING_SNAKE_CASE_     :   str = None ,							SCREAMING_SNAKE_CASE_     :   bool = True ,							SCREAMING_SNAKE_CASE_     :   str = "arrow" ,							**SCREAMING_SNAKE_CASE_     :   Optional[Any] ,							):
											super().__init__(
											    split=SCREAMING_SNAKE_CASE_ ,							features=SCREAMING_SNAKE_CASE_ ,							cache_dir=SCREAMING_SNAKE_CASE_ ,							keep_in_memory=SCREAMING_SNAKE_CASE_ ,							streaming=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_ ,							)
											lowerCAmelCase__      			=						load_from_cache_file
											lowerCAmelCase__      			=						file_format
											lowerCAmelCase__      			=						Spark(
											    df=SCREAMING_SNAKE_CASE_ ,							features=SCREAMING_SNAKE_CASE_ ,							cache_dir=SCREAMING_SNAKE_CASE_ ,							working_dir=SCREAMING_SNAKE_CASE_ ,							**SCREAMING_SNAKE_CASE_ ,							)




							def 						__snake_case					(   self     :   List[str]	):
											if self.streaming:
															return self.builder.as_streaming_dataset(split=self.split	)
											lowerCAmelCase__      			=						None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
											self.builder.download_and_prepare(
											    download_mode=SCREAMING_SNAKE_CASE_ ,							file_format=self._file_format ,							)
											return self.builder.as_dataset(split=self.split	)





