Added check for invalid hashing algorithm and removed list of hashing…

… algorithm support

docs/source/fundamentals/hashing.md

@@ -4,7 +4,7 @@ Streaming supports a variety of hash and checksum algorithms to verify data inte
 
 We optionally hash shards while serializing a streaming dataset, saving the resulting hashes in the index, which is written last. After the dataset is finished being written, we may hash the index file itself, the results of which must be stored elsewhere. Hashing during writing is controlled by the Writer argument `hashes: Optional[List[str]] = None`. We generally weakly recommend writing streaming datasets with one cryptographic hash algorithm and one fast hash algorithm for offline dataset validation in the future.
 
-Then, we optionally validate shard hashes upon download while reading a streaming dataset. Hashing during reading is controlled separately by the StreamingDataset argument `validate_hash: Optional[List[str]] = None`. We recommend reading streaming datasets for training purposes without validating hashes because of the extra cost in time and computation.
+Then, we optionally validate shard hashes upon download while reading a streaming dataset. Hashing during reading is controlled separately by the StreamingDataset argument `validate_hash: Optional[str] = None`. We recommend reading streaming datasets for training purposes without validating hashes because of the extra cost in time and computation.
 
 Available cryptographic hash functions:
 

streaming/base/dataset.py

@@ -14,7 +14,7 @@
 from math import ceil
 from threading import Event, Lock
 from time import sleep, time_ns
-from typing import Any, Dict, Iterator, List, Optional, Sequence, Tuple, Union
+from typing import Any, Dict, Iterator, Optional, Sequence, Tuple, Union
 
 import numpy as np
 from filelock import FileLock
@@ -251,7 +251,7 @@ class StreamingDataset(Array, IterableDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -312,7 +312,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[Union[int, str]] = None,
                  predownload: Optional[int] = None,

streaming/base/stream.py

@@ -81,7 +81,7 @@ class Stream:
             to ``None``.
         download_timeout (float, optional): Number of seconds to wait for a shard to download
             before raising an exception. Defaults to ``None``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool, optional): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep if and only if remote is local or no remote.
@@ -98,7 +98,7 @@ def __init__(self,
                  choose: Optional[int] = None,
                  download_retry: Optional[int] = None,
                  download_timeout: Optional[float] = None,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: Optional[bool] = None) -> None:
         self.remote = remote
         self._local = local
@@ -169,28 +169,6 @@ def _get_temporary_directory(self) -> str:
             hash = hashlib.blake2s(self.remote.encode('utf-8'), digest_size=16).hexdigest()
         return os.path.join(root, hash, self.split)
 
-    def _validate_hashes(self, hashes: List[str], file_info: FileInfo, filename: str,
-                         data: bytes) -> None:
-        """Validate the hashes of a file.
-
-        Args:
-            hashes (List[str]): Hashing algorithm name(s).
-            file_info (FileInfo): File information.
-            filename (str): Filename.
-            data (bytes): Data in the file.
-        """
-        # Validate what was downloaded.
-        if sorted(hashes) != sorted(file_info.hashes.keys()):
-            raise ValueError(f'Hash algorithms provided to validate ({hashes}) ' +
-                             f'do not match those provided during dataset creation ' +
-                             f'({sorted(file_info.hashes.keys())})')
-        for algo in hashes:
-            if algo not in file_info.hashes:
-                raise ValueError(f'Invalid Hash algorithm name: {algo}. Check the ' +
-                                 f'hash algorithm name during dataset creation.')
-            if get_hash(algo, data) != file_info.hashes[algo]:
-                raise ValueError(f'Checksum failure: {filename}')
-
     def apply_default(self, default: dict) -> None:
         """Apply defaults, setting any unset fields.
 
@@ -347,9 +325,15 @@ def _decompress_shard_part(self, zip_info: FileInfo, zip_filename: str, raw_file
         # Load compressed.
         data = open(zip_filename, 'rb').read()
 
-        # Validate the hash.
+        # Validate what was downloaded.
         if self.validate_hash:
-            self._validate_hashes(self.validate_hash, zip_info, zip_filename, data)
+            if self.validate_hash not in zip_info.hashes:
+                raise ValueError(
+                    f'Hash algorithm `{self.validate_hash}` provided for data ' +
+                    f'validation do not match with those provided during dataset ' +
+                    f'creation `{sorted(zip_info.hashes.keys())}`. Provide one of those.')
+            if get_hash(self.validate_hash, data) != zip_info.hashes[self.validate_hash]:
+                raise ValueError(f'Checksum failure: {zip_filename}')
 
         # Decompress and save that.
         data = decompress(compression, data)  # pyright: ignore
@@ -386,10 +370,6 @@ def _prepare_shard_part(self,
         raw_filename = os.path.join(self.local, self.split, raw_info.basename)
         if os.path.isfile(raw_filename):
             # Has raw.
-            # Validate the hash.
-            if self.validate_hash:
-                data = open(raw_filename, 'rb').read()
-                self._validate_hashes(self.validate_hash, raw_info, raw_filename, data)
             if zip_info and not self.safe_keep_zip:
                 zip_filename = os.path.join(self.local, self.split, zip_info.basename)
                 if os.path.isfile(zip_filename):
@@ -415,10 +395,16 @@ def _prepare_shard_part(self,
                 self._download_file(raw_info.basename)
                 delta += raw_info.bytes
 
-                # Validate the hash.
+                # Validate.
                 if self.validate_hash:
+                    if self.validate_hash not in raw_info.hashes:
+                        raise ValueError(
+                            f'Hash algorithm `{self.validate_hash}` provided for data ' +
+                            f'validation do not match with those provided during dataset ' +
+                            f'creation `{sorted(raw_info.hashes.keys())}`. Provide one of those.')
                     data = open(raw_filename, 'rb').read()
-                    self._validate_hashes(self.validate_hash, raw_info, raw_filename, data)
+                    if get_hash(self.validate_hash, data) != raw_info.hashes[self.validate_hash]:
+                        raise ValueError(f'Checksum failure: {raw_filename}')
         return delta
 
     def prepare_shard(self, shard: Reader) -> int:

streaming/multimodal/webvid.py

@@ -5,7 +5,7 @@
 
 import os
 from time import sleep
-from typing import Any, List, Optional
+from typing import Any, Optional
 
 from streaming.base import StreamingDataset
 from streaming.base.dataset import TICK, _Iterator
@@ -29,7 +29,7 @@ class StreamingInsideWebVid(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -102,7 +102,7 @@ class StreamingOutsideGIWebVid(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -152,7 +152,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,
@@ -233,7 +233,7 @@ class StreamingOutsideDTWebVid(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -283,7 +283,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/text/c4.py

@@ -7,7 +7,7 @@
 the `Common Crawl <https://commoncrawl.org>`_ dataset.
 """
 
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from transformers.models.auto.tokenization_auto import AutoTokenizer
 
@@ -31,7 +31,7 @@ class StreamingC4(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -83,7 +83,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/text/enwiki.py

@@ -3,7 +3,7 @@
 
 """English Wikipedia 2020-01-01 streaming dataset."""
 
-from typing import Any, List, Optional
+from typing import Any, Optional
 
 import numpy as np
 
@@ -27,7 +27,7 @@ class StreamingEnWiki(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -75,7 +75,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/text/pile.py

@@ -7,7 +7,7 @@
 high-quality datasets combined together.
 """
 
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from transformers.models.auto.tokenization_auto import AutoTokenizer
 
@@ -31,7 +31,7 @@ class StreamingPile(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -83,7 +83,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/vision/ade20k.py

@@ -7,7 +7,7 @@
 more details about this dataset.
 """
 
-from typing import Any, Callable, List, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 from streaming.base import StreamingDataset
 
@@ -29,7 +29,7 @@ class StreamingADE20K(StreamingDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -83,7 +83,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/vision/base.py

@@ -3,7 +3,7 @@
 
 """Base classes for computer vision :class:`StreamingDataset`s."""
 
-from typing import Any, Callable, List, Optional, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 from torchvision.datasets import VisionDataset
 from torchvision.transforms.functional import to_tensor
@@ -61,7 +61,7 @@ class StreamingVisionDataset(StreamingDataset, VisionDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to
@@ -115,7 +115,7 @@ def __init__(self,
                  split: Optional[str] = None,
                  download_retry: int = 2,
                  download_timeout: float = 60,
-                 validate_hash: Optional[List[str]] = None,
+                 validate_hash: Optional[str] = None,
                  keep_zip: bool = False,
                  epoch_size: Optional[int] = None,
                  predownload: Optional[int] = None,

streaming/vision/cifar10.py

@@ -27,7 +27,7 @@ class StreamingCIFAR10(StreamingVisionDataset):
         download_retry (int): Number of download re-attempts before giving up. Defaults to ``2``.
         download_timeout (float): Number of seconds to wait for a shard to download before raising
             an exception. Defaults to ``60``.
-        validate_hash (List[str], optional): Optional hash or checksum algorithm to use to validate
+        validate_hash (str, optional): Optional hash or checksum algorithm to use to validate
             shards. Defaults to ``None``.
         keep_zip (bool): Whether to keep or delete the compressed form when decompressing
             downloaded shards. If ``False``, keep iff remote is local or no remote. Defaults to