Add one_shot_download strategy.

Author: igorborgest

awswrangler/s3/_fs.py

@@ -192,8 +192,13 @@ def __init__(
         if mode not in {"rb", "wb", "r", "w"}:
             raise NotImplementedError("File mode must be {'rb', 'wb', 'r', 'w'}, not %s" % mode)
         self._mode: str = "rb" if mode is None else mode
-        if s3_block_size < 2:
-            raise exceptions.InvalidArgumentValue("s3_block_size MUST > 1")
+        self._one_shot_download: bool = False
+        if s3_block_size == 1:
+            raise exceptions.InvalidArgumentValue("s3_block_size MUST > 1 to define a valid size or "
+                                                  "< 1 to avoid blocks and always execute one shot downloads.")
+        elif s3_block_size < 1:
+            _logger.debug(f"s3_block_size of %d, enabling one_shot_download.", s3_block_size)
+            self._one_shot_download = True
         self._s3_block_size: int = s3_block_size
         self._s3_half_block_size: int = s3_block_size // 2
         self._s3_additional_kwargs: Dict[str, str] = {} if s3_additional_kwargs is None else s3_additional_kwargs
@@ -304,6 +309,12 @@ def _fetch(self, start: int, end: int) -> None:
             self._end = end
             return None
 
+        if self._one_shot_download:
+            self._start = 0
+            self._end = self._size
+            self._cache = self._fetch_range_proxy(self._start, self._end)
+            return None
+
         # Calculating block START and END positions
         _logger.debug("Downloading: %s (start) / %s (end)", start, end)
         mid: int = int(math.ceil((start + (end - 1)) / 2))
@@ -525,7 +536,7 @@ def open_s3_object(
     mode: str,
     use_threads: bool = False,
     s3_additional_kwargs: Optional[Dict[str, str]] = None,
-    s3_block_size: int = 4_194_304,  # 4 MB (4 * 2**20)
+    s3_block_size: int = -1,  # One shot download
     boto3_session: Optional[boto3.Session] = None,
     newline: Optional[str] = "\n",
     encoding: Optional[str] = "utf-8",

awswrangler/s3/_read_parquet.py

@@ -318,7 +318,7 @@ def _read_parquet_file(
         path=path,
         mode="rb",
         use_threads=use_threads,
-        s3_block_size=134_217_728,  # 128 MB (128 * 2**20)
+        s3_block_size=-1,  # One shot download
         s3_additional_kwargs=s3_additional_kwargs,
         boto3_session=boto3_session,
     ) as f:
@@ -348,30 +348,6 @@ def _count_row_groups(
         return n
 
 
-def _read_parquet_row_group(
-    row_group: int,
-    path: str,
-    columns: Optional[List[str]],
-    categories: Optional[List[str]],
-    boto3_primitives: _utils.Boto3PrimitivesType,
-    s3_additional_kwargs: Optional[Dict[str, str]],
-    use_threads: bool,
-) -> pa.Table:
-    boto3_session: boto3.Session = _utils.boto3_from_primitives(primitives=boto3_primitives)
-    with open_s3_object(
-        path=path,
-        mode="rb",
-        use_threads=use_threads,
-        s3_block_size=10_485_760,  # 10 MB (10 * 2**20)
-        s3_additional_kwargs=s3_additional_kwargs,
-        boto3_session=boto3_session,
-    ) as f:
-        pq_file: pyarrow.parquet.ParquetFile = pyarrow.parquet.ParquetFile(source=f, read_dictionary=categories)
-        num_row_groups: int = pq_file.num_row_groups
-        _logger.debug("Reading Row Group %s/%s [multi-threaded]", row_group + 1, num_row_groups)
-        return pq_file.read_row_group(i=row_group, columns=columns, use_threads=False, use_pandas_metadata=False)
-
-
 def _read_parquet(
     path: str,
     columns: Optional[List[str]],

awswrangler/s3/_read_text.py

@@ -77,7 +77,7 @@ def _read_text_file(
         path=path,
         mode=mode,
         use_threads=use_threads,
-        s3_block_size=134_217_728,  # 128 MB (128 * 2**20)
+        s3_block_size=-1,  # One shot download
         encoding=encoding,
         s3_additional_kwargs=s3_additional_kwargs,
         newline=newline,