




import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map







def 				lowerCAmelCase_	(     snake_case_     ):
      _A							:     Optional[int]			=						[]
      if isinstance(snake_case_,snake_case_     ):
            for v in tree.values():
                  shapes.extend(_fetch_dims(snake_case_     )     )
      elif isinstance(snake_case_,(list, tuple)     ):
            for t in tree:
                  shapes.extend(_fetch_dims(snake_case_     )     )
      elif isinstance(snake_case_,torch.Tensor     ):
            shapes.append(tree.shape     )
      else:
            raise ValueError("""Not supported"""     )

      return shapes







@torch.jit.ignore
def 				lowerCAmelCase_	(     snake_case_,snake_case_     ):
      _A							:     int			=						[]
      for d in reversed(snake_case_     ):
            idx.append(flat_idx % d     )
            _A							:     List[str]			=						flat_idx // d

      return tuple(reversed(snake_case_     )     )







@torch.jit.ignore
def 				lowerCAmelCase_	(     snake_case_,snake_case_,snake_case_,snake_case_ = None,snake_case_ = None,):

      # start_edges and end_edges both indicate whether, starting from any given
      # dimension, the start/end index is at the top/bottom edge of the
      # corresponding tensor, modeled as a tree
      def reduce_edge_list(snake_case_     ) -> None:
            _A							:     List[str]			=						True
            for i in range(len(snake_case_     )     ):
                  _A							:     List[str]			=						-1 * (i + 1)
                  l[reversed_idx] &= tally
                  _A							:     List[Any]			=						l[reversed_idx]

      if start_edges is None:
            _A							:     str			=						[s == 0 for s in start]
            reduce_edge_list(snake_case_     )
      if end_edges is None:
            _A							:     Any			=						[e == (d - 1) for e, d in zip(snake_case_,snake_case_     )]
            reduce_edge_list(snake_case_     )

      # Base cases. Either start/end are empty and we're done, or the final,
      # one-dimensional tensor can be simply sliced
      if len(snake_case_     ) == 0:
            return [()]
      elif len(snake_case_     ) == 1:
            return [(slice(start[0],end[0] + 1     ),)]

      _A							:     List[Tuple[slice, ...]]			=						[]
      _A							:     List[slice]			=						[]

      # Dimensions common to start and end can be selected directly
      for s, e in zip(snake_case_,snake_case_     ):
            if s == e:
                  path_list.append(slice(snake_case_,s + 1     )     )
            else:
                  break

      _A							:     Tuple[slice, ...]			=						tuple(snake_case_     )
      _A							:     str			=						len(snake_case_     )

      # start == end, and we're done
      if divergence_idx == len(snake_case_     ):
            return [path]

      def upper() -> Tuple[Tuple[slice, ...], ...]:
            assert start_edges is not None
            assert end_edges is not None

            _A							:     Any			=						start[divergence_idx]
            return tuple(
                path + (slice(snake_case_,sdi + 1     ),) + s
                for s in _get_minimal_slice_set(
                    start[divergence_idx + 1 :],[d - 1 for d in dims[divergence_idx + 1 :]],dims[divergence_idx + 1 :],start_edges=start_edges[divergence_idx + 1 :],end_edges=[True for _ in end_edges[divergence_idx + 1 :]],)     )

      def lower() -> Tuple[Tuple[slice, ...], ...]:
            assert start_edges is not None
            assert end_edges is not None

            _A							:     Union[str, Any]			=						end[divergence_idx]
            return tuple(
                path + (slice(snake_case_,edi + 1     ),) + s
                for s in _get_minimal_slice_set(
                    [0 for _ in start[divergence_idx + 1 :]],end[divergence_idx + 1 :],dims[divergence_idx + 1 :],start_edges=[True for _ in start_edges[divergence_idx + 1 :]],end_edges=end_edges[divergence_idx + 1 :],)     )

      # If both start and end are at the edges of the subtree rooted at
      # divergence_idx, we can just select the whole subtree at once
      if start_edges[divergence_idx] and end_edges[divergence_idx]:
            slices.append(path + (slice(start[divergence_idx],end[divergence_idx] + 1     ),)     )
      # If just start is at the edge, we can grab almost all of the subtree,
      # treating only the ragged bottom edge as an edge case
      elif start_edges[divergence_idx]:
            slices.append(path + (slice(start[divergence_idx],end[divergence_idx]     ),)     )
            slices.extend(lower()     )
      # Analogous to the previous case, but the top is ragged this time
      elif end_edges[divergence_idx]:
            slices.extend(upper()     )
            slices.append(path + (slice(start[divergence_idx] + 1,end[divergence_idx] + 1     ),)     )
      # If both sides of the range are ragged, we need to handle both sides
      # separately. If there's contiguous meat in between them, we can index it
      # in one big chunk
      else:
            slices.extend(upper()     )
            _A							:     List[str]			=						end[divergence_idx] - start[divergence_idx]
            if middle_ground > 1:
                  slices.append(path + (slice(start[divergence_idx] + 1,end[divergence_idx]     ),)     )
            slices.extend(lower()     )

      return slices







@torch.jit.ignore
def 				lowerCAmelCase_	(     snake_case_,snake_case_,snake_case_,snake_case_     ):
      _A							:     Optional[int]			=						t.shape[:no_batch_dims]
      _A							:     Tuple			=						list(_flat_idx_to_idx(snake_case_,snake_case_     )     )
      # _get_minimal_slice_set is inclusive
      _A							:     str			=						list(_flat_idx_to_idx(flat_end - 1,snake_case_     )     )

      # Get an ordered list of slices to perform
      _A							:     int			=						_get_minimal_slice_set(
          snake_case_,snake_case_,snake_case_,)

      _A							:     Optional[int]			=						[t[s] for s in slices]

      return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]     ) for s in sliced_tensors]     )







def 				lowerCAmelCase_	(     snake_case_,snake_case_,snake_case_,snake_case_,snake_case_ = False,snake_case_ = None,snake_case_ = False,):
      if not (len(snake_case_     ) > 0):
            raise ValueError("""Must provide at least one input"""     )

      _A							:     int			=						[shape[:no_batch_dims] for shape in _fetch_dims(snake_case_     )]
      _A							:     Tuple			=						tuple([max(snake_case_     ) for s in zip(*snake_case_     )]     )

      def _prep_inputs(snake_case_     ) -> torch.Tensor:
            if not low_mem:
                  if not sum(t.shape[:no_batch_dims]     ) == no_batch_dims:
                        _A							:     Dict			=						t.expand(orig_batch_dims + t.shape[no_batch_dims:]     )
                  _A							:     Union[str, Any]			=						t.reshape(-1,*t.shape[no_batch_dims:]     )
            else:
                  _A							:     Dict			=						t.expand(orig_batch_dims + t.shape[no_batch_dims:]     )
            return t

      _A							:     Dict[str, Any]			=						tensor_tree_map(_prep_inputs,snake_case_     )
      _A							:     Tuple			=						None
      if _out is not None:
            _A							:     Optional[int]			=						tensor_tree_map(lambda snake_case_     : t.view([-1] + list(t.shape[no_batch_dims:]     )     ),_out     )

      _A							:     Optional[Any]			=						1
      for d in orig_batch_dims:
            flat_batch_dim *= d

      _A							:     str			=						flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

      def _select_chunk(snake_case_     ) -> torch.Tensor:
            return t[i : i + chunk_size] if t.shape[0] != 1 else t

      _A							:     Any			=						0
      _A							:     List[str]			=						prepped_outputs
      for _ in range(snake_case_     ):
            # Chunk the input
            if not low_mem:
                  _A							:     List[Any]			=						_select_chunk
            else:
                  _A							:     str			=						partial(
                      _chunk_slice,flat_start=snake_case_,flat_end=min(snake_case_,i + chunk_size     ),no_batch_dims=len(snake_case_     ),)

            _A							:     Dict[str, Any]			=						tensor_tree_map(snake_case_,snake_case_     )

            # Run the layer on the chunk
            _A							:     Tuple			=						layer(**snake_case_     )

            # Allocate space for the output
            if out is None:
                  _A							:     Optional[Any]			=						tensor_tree_map(lambda snake_case_     : t.new_zeros((flat_batch_dim,) + t.shape[1:]     ),snake_case_     )

            # Put the chunk in its pre-allocated space
            if isinstance(snake_case_,snake_case_     ):

                  def assign(snake_case_,snake_case_     ) -> None:
                        for k, v in da.items():
                              if isinstance(snake_case_,snake_case_     ):
                                    assign(snake_case_,da[k]     )
                              else:
                                    if _add_into_out:
                                          v[i : i + chunk_size] += da[k]
                                    else:
                                          _A							:     List[Any]			=						da[k]

                  assign(snake_case_,snake_case_     )
            elif isinstance(snake_case_,snake_case_     ):
                  for xa, xa in zip(snake_case_,snake_case_     ):
                        if _add_into_out:
                              xa[i : i + chunk_size] += xa
                        else:
                              _A							:     Tuple			=						xa
            elif isinstance(snake_case_,torch.Tensor     ):
                  if _add_into_out:
                        out[i : i + chunk_size] += output_chunk
                  else:
                        _A							:     Optional[int]			=						output_chunk
            else:
                  raise ValueError("""Not supported"""     )

            i += chunk_size

      _A							:     int			=						tensor_tree_map(lambda snake_case_     : t.view(orig_batch_dims + t.shape[1:]     ),snake_case_     )

      return out







class 						lowercase							:

     def __init__(  self		,     _a = 512		,     )						->							str:
           _A							:     Union[str, Any]			=						max_chunk_size
           _A							:     Optional[int]			=						None
           _A							:     Optional[tuple]			=						None

     def 					a__	(  self		,     _a		,     _a		,     _a				)						->							int:
           logging.info("""Tuning chunk size..."""				)

           if min_chunk_size >= self.max_chunk_size:
                 return min_chunk_size

           _A							:     List[int]			=						[2**l for l in range(int(math.log(self.max_chunk_size		,     2				)				) + 1				)]
           _A							:     List[Any]			=						[c for c in candidates if c > min_chunk_size]
           _A							:     Optional[Any]			=						[min_chunk_size] + candidates
           candidates[-1] += 4

           def test_chunk_size(_a				) -> bool:
                 try:
                       with torch.no_grad():
                             fn(*_a		,     chunk_size=_a				)
                       return True
                 except RuntimeError:
                       return False

           _A							:     Optional[int]			=						0
           _A							:     str			=						len(_a				) - 1
           while i > min_viable_chunk_size_index:
                 _A							:     Optional[Any]			=						test_chunk_size(candidates[i]				)
                 if not viable:
                       _A							:     Union[str, Any]			=						(min_viable_chunk_size_index + i) // 2
                 else:
                       _A							:     Union[str, Any]			=						i
                       _A							:     Optional[Any]			=						(i + len(_a				) - 1) // 2

           return candidates[min_viable_chunk_size_index]

     def 					a__	(  self		,     _a		,     _a				)						->							bool:
           _A							:     Optional[Any]			=						True
           for aa, aa in zip(_a		,     _a				):
                 assert type(_a				) == type(_a				)
                 if isinstance(_a		,     (list, tuple)				):
                       consistent &= self._compare_arg_caches(_a		,     _a				)
                 elif isinstance(_a		,     _a				):
                       _A							:     Union[str, Any]			=						[v for _, v in sorted(aa.items()		,     key=lambda _a				: x[0]				)]
                       _A							:     Union[str, Any]			=						[v for _, v in sorted(aa.items()		,     key=lambda _a				: x[0]				)]
                       consistent &= self._compare_arg_caches(_a		,     _a				)
                 else:
                       consistent &= aa == aa

           return consistent



     def 					a__	(  self		,     _a		,     _a		,     _a		,     )						->							int:
           _A							:     Tuple			=						True
           _A							:     tuple			=						tree_map(lambda _a				: a.shape if isinstance(_a		,     torch.Tensor				) else a		,     _a		,     _a				)
           if self.cached_arg_data is not None:
                 # If args have changed shape/value, we need to re-tune
                 assert len(self.cached_arg_data				) == len(_a				)
                 _A							:     List[Any]			=						self._compare_arg_caches(self.cached_arg_data		,     _a				)
           else:
                 # Otherwise, we can reuse the precomputed value
                 _A							:     Tuple			=						False

           if not consistent:
                 _A							:     int			=						self._determine_favorable_chunk_size(
                     _a		,     _a		,     _a		,     )
                 _A							:     int			=						arg_data

           assert self.cached_chunk_size is not None

           return self.cached_chunk_size







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 						lowercase							(   UpperCamelCase__		):

     def __init__(  self		,     _a		,     _a		,     _a = None		,     _a = None		,     _a = False		,     **_a		,     )						->							int:
           super().__init__(features=_a		,     cache_dir=_a		,     keep_in_memory=_a		,     **_a				)
           _A							:     Optional[int]			=						Sql(
               cache_dir=_a		,     features=_a		,     sql=_a		,     con=_a		,     **_a		,     )



     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Tuple			=						None
           _A							:     int			=						None
           _A							:     Tuple			=						None
           _A							:     Union[str, Any]			=						None

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

           # Build dataset for splits
           _A							:     int			=						self.builder.as_dataset(
               split="""train"""		,     verification_mode=_a		,     in_memory=self.keep_in_memory				)
           return dataset







class 						lowercase							:

     def __init__(  self		,     _a		,     _a		,     _a		,     _a = None		,     _a = None		,     **_a		,     )						->							Union[str, Any]:
           if num_proc is not None and num_proc <= 0:
                 raise ValueError(F'''num_proc {num_proc} must be an integer > 0.'''				)

           _A							:     Dict			=						dataset
           _A							:     int			=						name
           _A							:     Union[str, Any]			=						con
           _A							:     str			=						batch_size if batch_size else config.DEFAULT_MAX_BATCH_SIZE
           _A							:     str			=						num_proc
           _A							:     Optional[Any]			=						to_sql_kwargs

     def 					a__	(  self				)						->							int:
           _A							:     Any			=						self.to_sql_kwargs.pop("""sql"""		,     _a				)
           _A							:     List[str]			=						self.to_sql_kwargs.pop("""con"""		,     _a				)
           _A							:     int			=						self.to_sql_kwargs.pop("""index"""		,     _a				)

           _A							:     List[str]			=						self._write(index=_a		,     **self.to_sql_kwargs				)
           return written

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



     def 					a__	(  self		,     _a		,     **_a				)						->							int:
           _A							:     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:
                 _A			,   _A							:     Tuple			=						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		,     _a		,     _a				)]		,     )		,     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

