



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 						UpperCamelCase      (  snake_case__			:       Union[dict, list, tuple, torch.Tensor]      )      ->	List[Tuple[int, ...]]:
   UpperCamelCase       :    str		      =      []
   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 						UpperCamelCase      (  snake_case__			:       int			,	snake_case__			:       Tuple[int, ...]      )      ->	Tuple[int, ...]:
   UpperCamelCase       :    Optional[Any]		      =      []
   for d in reversed(snake_case__      ):
      idx.append(flat_idx % d      )
      UpperCamelCase       :    Optional[Any]		      =      flat_idx // d

   return tuple(reversed(snake_case__      )      )
@torch.jit.ignore
def 						UpperCamelCase      (  snake_case__			:       Sequence[int]			,	snake_case__			:       Sequence[int]			,	snake_case__			:       Sequence[int]			,	snake_case__			:       Optional[Sequence[bool]] = None			,	snake_case__			:       Optional[Sequence[bool]] = None			,	)      ->	List[Tuple[slice, ...]]:

   # 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__			:       List[bool]      ) -> None:
      UpperCamelCase       :    Optional[Any]		      =      True
      for i in range(len(snake_case__      )      ):
         UpperCamelCase       :    int		      =      -1 * (i + 1)
         l[reversed_idx] &= tally
         UpperCamelCase       :    Optional[Any]		      =      l[reversed_idx]

   if start_edges is None:
      UpperCamelCase       :    Dict		      =      [s == 0 for s in start]
      reduce_edge_list(snake_case__      )
   if end_edges is None:
      UpperCamelCase       :    int		      =      [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      ),)]

   UpperCamelCase       :    List[Tuple[slice, ...]]		      =      []
   UpperCamelCase       :    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

   UpperCamelCase       :    Tuple[slice, ...]		      =      tuple(snake_case__      )
   UpperCamelCase       :    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

      UpperCamelCase       :    Optional[int]		      =      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

      UpperCamelCase       :    List[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()      )
      UpperCamelCase       :    Dict		      =      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 						UpperCamelCase      (  snake_case__			:       torch.Tensor			,	snake_case__			:       int			,	snake_case__			:       int			,	snake_case__			:       int      )      ->	torch.Tensor:
   UpperCamelCase       :    Union[str, Any]		      =      t.shape[:no_batch_dims]
   UpperCamelCase       :    str		      =      list(_flat_idx_to_idx(snake_case__			,	snake_case__      )      )
   # _get_minimal_slice_set is inclusive
   UpperCamelCase       :    Optional[Any]		      =      list(_flat_idx_to_idx(flat_end - 1			,	snake_case__      )      )

   # Get an ordered list of slices to perform
   UpperCamelCase       :    Union[str, Any]		      =      _get_minimal_slice_set(
       snake_case__			,	snake_case__			,	snake_case__			,	)

   UpperCamelCase       :    Union[str, Any]		      =      [t[s] for s in slices]

   return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]      ) for s in sliced_tensors]      )
def 						UpperCamelCase      (  snake_case__			:       Callable			,	snake_case__			:       Dict[str, Any]			,	snake_case__			:       int			,	snake_case__			:       int			,	snake_case__			:       bool = False			,	snake_case__			:       Any = None			,	snake_case__			:       bool = False			,	)      ->	Any:
   if not (len(snake_case__      ) > 0):
      raise ValueError('Must provide at least one input'      )

   UpperCamelCase       :    Optional[int]		      =      [shape[:no_batch_dims] for shape in _fetch_dims(snake_case__      )]
   UpperCamelCase       :    Dict		      =      tuple([max(snake_case__      ) for s in zip(*snake_case__      )]      )

   def _prep_inputs(snake_case__			:       torch.Tensor      ) -> torch.Tensor:
      if not low_mem:
         if not sum(t.shape[:no_batch_dims]      ) == no_batch_dims:
            UpperCamelCase       :    Tuple		      =      t.expand(orig_batch_dims + t.shape[no_batch_dims:]      )
         UpperCamelCase       :    List[Any]		      =      t.reshape(-1			,	*t.shape[no_batch_dims:]      )
      else:
         UpperCamelCase       :    Optional[Any]		      =      t.expand(orig_batch_dims + t.shape[no_batch_dims:]      )
      return t

   UpperCamelCase       :    Dict[str, Any]		      =      tensor_tree_map(_prep_inputs			,	snake_case__      )
   UpperCamelCase       :    List[Any]		      =      None
   if _out is not None:
      UpperCamelCase       :    List[Any]		      =      tensor_tree_map(lambda snake_case__      : t.view([-1] + list(t.shape[no_batch_dims:]      )      )			,	_out      )

   UpperCamelCase       :    List[Any]		      =      1
   for d in orig_batch_dims:
      flat_batch_dim *= d

   UpperCamelCase       :    List[Any]		      =      flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

      UpperCamelCase       :    Dict[str, Any]		      =      tensor_tree_map(snake_case__			,	snake_case__      )

      # Run the layer on the chunk
      UpperCamelCase       :    List[Any]		      =      layer(**snake_case__      )

      # Allocate space for the output
      if out is None:
         UpperCamelCase       :    str		      =      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__			:       dict			,	snake_case__			:       dict      ) -> 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:
                     UpperCamelCase       :    Optional[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:
               UpperCamelCase       :    str		      =      xa
      elif isinstance(snake_case__			,	torch.Tensor      ):
         if _add_into_out:
            out[i : i + chunk_size] += output_chunk
         else:
            UpperCamelCase       :    Optional[Any]		      =      output_chunk
      else:
         raise ValueError('Not supported'      )

      i += chunk_size

   UpperCamelCase       :    Tuple		      =      tensor_tree_map(lambda snake_case__      : t.view(orig_batch_dims + t.shape[1:]      )			,	snake_case__      )

   return out



class 							lowerCAmelCase_						:
       def __init__(     self,     SCREAMING_SNAKE_CASE_ = 512,     )	->						List[Any]:
          UpperCamelCase       :    int		      =      max_chunk_size
          UpperCamelCase       :    Optional[int]		      =      None
          UpperCamelCase       :    Optional[tuple]		      =      None
       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)	->						int:
          logging.info('Tuning chunk size...'			)

          if min_chunk_size >= self.max_chunk_size:
             return min_chunk_size

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

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

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

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

          return consistent






       def    snake_case_			(     self,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     )	->						int:
          UpperCamelCase       :    int		      =      True
          UpperCamelCase       :    tuple		      =      tree_map(lambda SCREAMING_SNAKE_CASE_			: a.shape if isinstance(SCREAMING_SNAKE_CASE_,     torch.Tensor			) else a,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			)
          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(SCREAMING_SNAKE_CASE_			)
             UpperCamelCase       :    str		      =      self._compare_arg_caches(self.cached_arg_data,     SCREAMING_SNAKE_CASE_			)
          else:
             # Otherwise, we can reuse the precomputed value
             UpperCamelCase       :    Tuple		      =      False

          if not consistent:
             UpperCamelCase       :    Union[str, Any]		      =      self._determine_favorable_chunk_size(
                 SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_,     )
             UpperCamelCase       :    List[Any]		      =      arg_data

          assert self.cached_chunk_size is not None

          return self.cached_chunk_size




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 		snake_case__	(		lowercase							,				lowercase					):
   assert isinstance(lowercase							,				lowercase					)
   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 		snake_case__	(		lowercase							,				lowercase							,				lowercase							,				lowercase					):
   lowerCAmelCase_:							Any       =	tmp_path / "cache"
   lowerCAmelCase_:							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():
      lowerCAmelCase_:							Optional[int]       =	SqlDatasetReader(
          "dataset"							,				"sqlite:///" + sqlite_path							,				cache_dir=lowercase							,				keep_in_memory=lowercase					).read()
   _check_sql_dataset(lowercase							,				lowercase					)
@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 		snake_case__	(		lowercase							,				lowercase							,				lowercase							,				lowercase					):
   lowerCAmelCase_:							List[str]       =	tmp_path / "cache"
   lowerCAmelCase_:							int       =	{"col_1": "string", "col_2": "int64", "col_3": "float64"}
   lowerCAmelCase_:							List[str]       =	features.copy() if features else default_expected_features
   lowerCAmelCase_:							Any       =	(
       Features({feature: Value(lowercase					) for feature, dtype in features.items()}					) if features is not None else None
   )
   lowerCAmelCase_:							int       =	SqlDatasetReader("dataset"							,				"sqlite:///" + sqlite_path							,				features=lowercase							,				cache_dir=lowercase					).read()
   _check_sql_dataset(lowercase							,				lowercase					)
def 		snake_case__	(		lowercase					):
   with contextlib.closing(sqlitea.connect(lowercase					)					) as con:
      lowerCAmelCase_:							Any       =	con.cursor()
      cur.execute("SELECT * FROM dataset"					)
      for row in cur:
         yield row
@require_sqlalchemy
def 		snake_case__	(		lowercase							,				lowercase							,				lowercase					):
   lowerCAmelCase_:							Optional[int]       =	tmp_path / "cache"
   lowerCAmelCase_:							Optional[Any]       =	os.path.join(lowercase							,				"tmp.sql"					)
   lowerCAmelCase_:							str       =	SqlDatasetReader("dataset"							,				"sqlite:///" + sqlite_path							,				cache_dir=lowercase					).read()
   SqlDatasetWriter(lowercase							,				"dataset"							,				"sqlite:///" + output_sqlite_path							,				num_proc=1					).write()

   lowerCAmelCase_:							Union[str, Any]       =	iter_sql_file(lowercase					)
   lowerCAmelCase_:							str       =	iter_sql_file(lowercase					)

   for rowa, rowa in zip(lowercase							,				lowercase					):
      assert rowa == rowa
@require_sqlalchemy
def 		snake_case__	(		lowercase							,				lowercase							,				lowercase					):
   lowerCAmelCase_:							str       =	tmp_path / "cache"
   lowerCAmelCase_:							Optional[int]       =	os.path.join(lowercase							,				"tmp.sql"					)
   lowerCAmelCase_:							Optional[Any]       =	SqlDatasetReader("dataset"							,				"sqlite:///" + sqlite_path							,				cache_dir=lowercase					).read()
   SqlDatasetWriter(lowercase							,				"dataset"							,				"sqlite:///" + output_sqlite_path							,				num_proc=2					).write()

   lowerCAmelCase_:							Optional[Any]       =	iter_sql_file(lowercase					)
   lowerCAmelCase_:							Optional[int]       =	iter_sql_file(lowercase					)

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







@require_sqlalchemy
def 		snake_case__	(		lowercase							,				lowercase							,				lowercase					):
   lowerCAmelCase_:							Union[str, Any]       =	tmp_path / "cache"
   lowerCAmelCase_:							int       =	os.path.join(lowercase							,				"tmp.sql"					)
   lowerCAmelCase_:							Any       =	SqlDatasetReader("dataset"							,				"sqlite:///" + sqlite_path							,				cache_dir=lowercase					).read()
   with pytest.raises(lowercase					):
      SqlDatasetWriter(lowercase							,				"dataset"							,				"sqlite:///" + output_sqlite_path							,				num_proc=0					).write()