fix newline stripping in plain text readers (#174)

Summary:
Fixes #173

Note that the [input to `strip`](https://docs.python.org/3/library/stdtypes.html#str.strip)

> is a string specifying the **set of characters** to be removed. [Emphasis mine]

Thus, stripping works something like

```python
for char in chars:
    string.replace(char, "")
```

rather than

```python
string.replace(chars, "")
```

This means that always stripping `"\r\n"` is harmless even if the line terminator is only `"\n"` or `\"r"`.

Reviewed By: ejguan

Differential Revision: D33684458

Pulled By: NivekT

fbshipit-source-id: 9821b77d60d3afe038ae698965beefe319783aa1

ghstack-source-id: 37a119b
Pull Request resolved: #176

Author: ejguan

test/test_datapipe.py

@@ -520,14 +520,12 @@ def test_bucket_batcher_iterdatapipe(self) -> None:
         batch_dp = source_dp.bucketbatch(
             batch_size=3, drop_last=True, batch_num=100, bucket_num=1, in_batch_shuffle=True
         )
-        self.assertEqual(3, len(batch_dp))
         self.assertEqual(9, len(list(batch_dp.unbatch())))
 
         # Functional Test: drop last is False preserves length
         batch_dp = source_dp.bucketbatch(
             batch_size=3, drop_last=False, batch_num=100, bucket_num=1, in_batch_shuffle=False
         )
-        self.assertEqual(4, len(batch_dp))
         self.assertEqual(10, len(list(batch_dp.unbatch())))
 
         def _return_self(x):
@@ -539,14 +537,12 @@ def _return_self(x):
         )
         # bucket_num = 1 means there will be no shuffling if a sort key is given
         self.assertEqual([[0, 1, 2], [3, 4, 5], [6, 7, 8]], list(batch_dp))
-        self.assertEqual(3, len(batch_dp))
         self.assertEqual(9, len(list(batch_dp.unbatch())))
 
         # Functional Test: using sort_key, without in_batch_shuffle
         batch_dp = source_dp.bucketbatch(
             batch_size=3, drop_last=True, batch_num=100, bucket_num=2, in_batch_shuffle=False, sort_key=_return_self
         )
-        self.assertEqual(3, len(batch_dp))
         self.assertEqual(9, len(list(batch_dp.unbatch())))
 
         # Reset Test:
@@ -567,7 +563,8 @@ def _return_self(x):
         self.assertEqual(9, len([item for batch in res_after_reset for item in batch]))
 
         # __len__ Test: returns the number of batches
-        self.assertEqual(3, len(batch_dp))
+        with self.assertRaises(TypeError):
+            len(batch_dp)
 
 
 if __name__ == "__main__":

torchdata/datapipes/iter/transform/bucketbatcher.py

@@ -1,6 +1,6 @@
 # Copyright (c) Facebook, Inc. and its affiliates.
 import random
-from typing import Callable, Iterator, Optional, Sized, TypeVar
+from typing import Callable, Optional, TypeVar
 
 from torch.utils.data import DataChunk
 
@@ -42,44 +42,32 @@ class BucketBatcherIterDataPipe(IterDataPipe[DataChunk[T_co]]):
     in_batch_shuffle: bool
     length: Optional[int]
 
-    def __init__(
-        self,
+    def __new__(
+        cls,
         datapipe: IterDataPipe[T_co],
         batch_size: int,
         drop_last: bool = False,
         batch_num: int = 100,
         bucket_num: int = 1,
         sort_key: Optional[Callable] = None,
         in_batch_shuffle: bool = True,
-    ) -> None:
+    ) -> IterDataPipe:
         assert batch_size > 0, "Batch size is required to be larger than 0!"
         assert batch_num > 0, "Number of batches is required to be larger than 0!"
         assert bucket_num > 0, "Number of buckets is required to be larger than 0!"
-        super().__init__()
 
-        # TODO(136): Verify _datapippe is not going to be serialized twice and is able to reconstruct
-        self._datapipe: IterDataPipe[T_co] = datapipe
-        self.batch_size: int = batch_size
-        self.drop_last: bool = drop_last
-        self.batch_num: int = batch_num
-        self.bucket_num: int = bucket_num
-        self.sort_key: Optional[Callable] = sort_key
-        self.in_batch_shuffle: bool = in_batch_shuffle
-
-        self.bucket_size = batch_size * batch_num
-        self.pool_size = self.bucket_size * bucket_num
+        bucket_size = batch_size * batch_num
+        pool_size = bucket_size * bucket_num
 
         # Shuffle by pool_size
         if bucket_num > 1 or sort_key is None:
             if in_batch_shuffle:
-                datapipe = (
-                    datapipe.batch(batch_size=self.pool_size, drop_last=False).map(fn=_in_batch_shuffle_fn).unbatch()
-                )
+                datapipe = datapipe.batch(batch_size=pool_size, drop_last=False).map(fn=_in_batch_shuffle_fn).unbatch()
             else:
-                datapipe = datapipe.shuffle(buffer_size=self.pool_size)
+                datapipe = datapipe.shuffle(buffer_size=pool_size)
         # Sort by bucket_size if sort_key is given
         if sort_key is not None:
-            datapipe = datapipe.batch(self.bucket_size).map(fn=sort_key).unbatch()
+            datapipe = datapipe.batch(bucket_size).map(fn=sort_key).unbatch()
         # Batch and drop last (if needed)
         datapipe = datapipe.batch(batch_size, drop_last=drop_last)
         # Shuffle the batched data
@@ -88,20 +76,5 @@ def __init__(
             if in_batch_shuffle:
                 datapipe = datapipe.batch(batch_size=bucket_num, drop_last=False).map(fn=_in_batch_shuffle_fn).unbatch()
             else:
-                datapipe = datapipe.shuffle(buffer_size=self.bucket_size)
-        self.datapipe = datapipe
-        self.length = None
-
-    def __iter__(self) -> Iterator:
-        yield from self.datapipe
-
-    def __len__(self) -> int:
-        if self.length is not None:
-            return self.length
-        if isinstance(self._datapipe, Sized):
-            if self.drop_last:
-                self.length = len(self._datapipe) // self.batch_size
-            else:
-                self.length = (len(self._datapipe) + self.batch_size - 1) // self.batch_size
-            return self.length
-        raise TypeError(f"{type(self).__name__} instance doesn't have valid length")
+                datapipe = datapipe.shuffle(buffer_size=bucket_size)
+        return datapipe