Repository: Lightning-AI/litdata

Issue:
Add streaming support for huggingface parquet dataset similar to chunk stream
## 🚀 Feature Request  

**Add Streaming Support for Hugging Face Parquet Datasets Similar to Chunk-Based Streaming**  

### Motivation  
Currently, the entire dataset must be downloaded before indexing, resulting in significant waiting time before streaming can begin. This approach is inefficient, especially for datasets with a large number of chunks.  

### Pitch  
The proposed solution is to read only the metadata of all Parquet files using byte-range requests, allowing for quick indexing. Streaming can then proceed similarly to LitData's chunk-based streaming approach.  

Reference: [Introduction to Parquet Format](https://huggingface.co/blog/cfahlgren1/intro-to-parquet-format)

Discussion:
tchaton: @bhimrazy I wonder if we could benchmark pyarrow vs polars for streaming the data
bhimrazy: > @bhimrazy I wonder if we could benchmark pyarrow vs polars for streaming the data

Sure will check that.
duckdb also seems to be another option but none of them seem to provide efficient way of reading particular rows.
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Lightning-AI/litdata/pull/505?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Lightning-AI) Report
Attention: Patch coverage is `70.55556%` with `53 lines` in your changes missing coverage. Please review.
> Project coverage is 71%. Comparing base [(`a2b2570`)](https://app.codecov.io/gh/Lightning-AI/litdata/commit/a2b25706a993f171982dbbf19dd02473495968cb?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Lightning-AI) to head [(`30c6187`)](https://app.codecov.io/gh/Lightning-AI/litdata/commit/30c6187848375a64291e977c7d67272ae8af8e7b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Lightning-AI).

> :exclamation:  There is a different number of reports uploaded between BASE (a2b2570) and HEAD (30c6187). Click for more details.
> 
> <details><summary>HEAD has 15 uploads less than BASE</summary>
>
>| Flag | BASE (a2b2570) | HEAD (30c6187) |
>|------|------|------|
>|unittests|6|1|
>|windows-2022|2|1|
>|3.9|3|1|
>|3.10|3|0|
>|ubuntu-22.04|2|0|
>|macos-13|2|0|
></details>

<details><summary>Additional details and impacted files</summary>


```diff
@@         Coverage Diff          @@
##           main   #505    +/-   ##
====================================
- Coverage    79%    71%    -8%     
====================================
  Files        39     39            
  Lines      5859   5859            
====================================
- Hits       4621   4181   -440     
- Misses     1238   1678   +440     
```

</details>
<details><summary>🚀 New features to boost your workflow: </summary>

- ❄ [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
</details>
bhimrazy: ## Benchmarks for [open-thoughts/OpenThoughts-114k](https://huggingface.co/datasets/open-thoughts/OpenThoughts-114k)
- `bacth_size = 256`, `num_workers=32`, `machine=A10G`

<details>
  <summary>a) Results for this PR with `low_memory=True`:</summary>

```python
⚡ main ~/litdata-benchmark SHUFFLE=0 PRELOAD=0 LOW_MEMORY=1 python stream_fineweb_edu_10B.py 
Seed set to 42
Shuffle: False, Preload: False, Low Memory: True
Indexing HF dataset from hf://datasets/open-thoughts/OpenThoughts-114k/data into /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json.
Indexing progress:   0%|                                                                                                  | 0/6 [00:00<?, ?step/s]
Index file successfully written to: /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json
Indexing progress: 100%|██████████████████████████████████████████████████████████████████████████████████████████| 6/6 [00:00<00:00, 48.63step/s]
Total number of samples in the dataset: 113957
train-00001-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 175M/175M [00:00<00:00, 218MB/s]
train-00004-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 154M/154M [00:00<00:00, 193MB/s]
train-00003-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 174M/174M [00:00<00:00, 207MB/s]
train-00002-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 173M/173M [00:00<00:00, 226MB/s]
train-00005-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 152M/152M [00:00<00:00, 189MB/s]
train-00000-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 250M/250M [00:01<00:00, 135MB/s]
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:08<00:00, 51.68it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 0, streamed over 113957 samples in 8.631164073944092 or 13202.965949617279 samples/sec.
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:07<00:00, 61.00it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 1, streamed over 113957 samples in 7.311708211898804 or 15585.54473275472 samples/sec.
Finished benchmarking.
```

</details>

<details>
  <summary>b) Results for this PR with `low_memory=False`:</summary>

```python
⚡ main ~/litdata-benchmark SHUFFLE=0 PRELOAD=0 LOW_MEMORY=0 python stream_fineweb_edu_10B.py
Seed set to 42
Benchmarking using litdata version: 0.2.42
Shuffle: False, Preload: False, Low Memory: False
Indexing HF dataset from hf://datasets/open-thoughts/OpenThoughts-114k/data into /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json.
Indexing progress:   0%|                                                                                                  | 0/6 [00:00<?, ?step/s]
Index file successfully written to: /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json
Indexing progress: 100%|██████████████████████████████████████████████████████████████████████████████████████████| 6/6 [00:00<00:00, 23.32step/s]
Total number of samples in the dataset: 113957
train-00005-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 152M/152M [00:00<00:00, 234MB/s]
train-00004-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 154M/154M [00:00<00:00, 232MB/s]
train-00002-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 173M/173M [00:00<00:00, 226MB/s]
train-00003-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 174M/174M [00:00<00:00, 227MB/s]
train-00000-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 250M/250M [00:01<00:00, 205MB/s]
train-00001-of-00006.parquet: 100%|████████████████████████████████████████████████████████████████████████████| 175M/175M [00:02<00:00, 72.0MB/s]
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:09<00:00, 47.54it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 0, streamed over 113957 samples in 9.382962703704834 or 12145.093821165492 samples/sec.
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:08<00:00, 50.02it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 1, streamed over 113957 samples in 8.917301416397095 or 12779.310479351509 samples/sec.
Finished benchmarking.
```

</details>


<details>
  <summary>c) Results for this PR with `low_memory=False` and `pre_load_chunk=True`:</summary>

⚡ main ~/litdata-benchmark SHUFFLE=0 PRELOAD=1 LOW_MEMORY=0 python stream_fineweb_edu_10B.py
Seed set to 42
Shuffle: False, Preload: True, Low Memory: False
Indexing HF dataset from hf://datasets/open-thoughts/OpenThoughts-114k/data into /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json.
Indexing progress:   0%|                                                                                                  | 0/6 [00:00<?, ?step/s]
Index file successfully written to: /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json
Indexing progress: 100%|██████████████████████████████████████████████████████████████████████████████████████████| 6/6 [00:00<00:00, 42.90step/s]
Total number of samples in the dataset: 113957
train-00004-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 154M/154M [00:00<00:00, 221MB/s]
train-00005-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 152M/152M [00:00<00:00, 216MB/s]
train-00002-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 173M/173M [00:00<00:00, 224MB/s]
train-00003-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 174M/174M [00:00<00:00, 219MB/s]
train-00001-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 175M/175M [00:00<00:00, 203MB/s]
train-00000-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 250M/250M [00:01<00:00, 233MB/s]
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:09<00:00, 45.82it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 0, streamed over 113957 samples in 9.733589887619019 or 11707.597777877443 samples/sec.
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:10<00:00, 41.50it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 1, streamed over 113957 samples in 10.748220205307007 or 10602.402993807767 samples/sec.
Finished benchmarking.

</details>

<details>
  <summary>d) Results for this PR with `low_memory=False` and `shuffle=True`:</summary>

⚡ main ~/litdata-benchmark SHUFFLE=1 PRELOAD=1 LOW_MEMORY=0 python stream_fineweb_edu_10B.py
Seed set to 42
Shuffle: True, Preload: True, Low Memory: False
Indexing HF dataset from hf://datasets/open-thoughts/OpenThoughts-114k/data into /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json.
Indexing progress:   0%|                                                                                                  | 0/6 [00:00<?, ?step/s]
Index file successfully written to: /teamspace/studios/this_studio/.cache/litdata-cache-index-pq/0c2f2da21e97421c865837df5d13b4af5511701940e3c2d18f3659ae37907f91/index.json
Indexing progress: 100%|██████████████████████████████████████████████████████████████████████████████████████████| 6/6 [00:00<00:00, 12.96step/s]
Total number of samples in the dataset: 113957
train-00004-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 154M/154M [00:00<00:00, 221MB/s]
train-00002-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 173M/173M [00:00<00:00, 218MB/s]
train-00005-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 152M/152M [00:00<00:00, 191MB/s]
train-00001-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 175M/175M [00:00<00:00, 217MB/s]
train-00003-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 174M/174M [00:00<00:00, 192MB/s]
train-00000-of-00006.parquet: 100%|█████████████████████████████████████████████████████████████████████████████| 250M/250M [00:01<00:00, 214MB/s]
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:10<00:00, 44.21it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 0, streamed over 113957 samples in 10.087934494018555 or 11296.361705695894 samples/sec.
100%|███████████████████████████████████████████████████████████████████████████████████████████████████████████| 446/446 [00:10<00:00, 41.31it/s]
For /teamspace/studios/this_studio/litdata-benchmark/open-thoughts.py on 1, streamed over 113957 samples in 10.79666018486023 or 10554.834578638292 samples/sec.
Finished benchmarking.

</details>


Patch:
diff --git a/src/litdata/constants.py b/src/litdata/constants.py
index 49559d7..018271a 100644
--- a/src/litdata/constants.py
+++ b/src/litdata/constants.py
@@ -39,6 +39,7 @@ _AZURE_STORAGE_AVAILABLE = RequirementCache("azure.storage.blob")
 _TQDM_AVAILABLE = RequirementCache("tqdm")
 _LIGHTNING_SDK_AVAILABLE = RequirementCache("lightning_sdk")
 _HF_HUB_AVAILABLE = RequirementCache("huggingface_hub")
+_PYARROW_AVAILABLE = RequirementCache("pyarrow")
 _POLARS_AVAILABLE = RequirementCache("polars>1.0.0")
 _DEBUG = bool(int(os.getenv("DEBUG_LITDATA", "0")))
 
diff --git a/src/litdata/processing/readers.py b/src/litdata/processing/readers.py
index eff6fc5..a1ebdef 100644
--- a/src/litdata/processing/readers.py
+++ b/src/litdata/processing/readers.py
@@ -16,13 +16,10 @@ import os
 from abc import ABC, abstractmethod
 from typing import Any, List
 
-from lightning_utilities.core.imports import RequirementCache
-
+from litdata.constants import _PYARROW_AVAILABLE
 from litdata.streaming.dataloader import StreamingDataLoader
 from litdata.utilities.format import _get_tqdm_iterator_if_available
 
-_PYARROW_AVAILABLE = RequirementCache("pyarrow")
-
 
 class BaseReader(ABC):
     """The `BaseReader` interface defines how to read and preprocess data
diff --git a/src/litdata/streaming/dataset.py b/src/litdata/streaming/dataset.py
index b50e45e..f173794 100644
--- a/src/litdata/streaming/dataset.py
+++ b/src/litdata/streaming/dataset.py
@@ -100,11 +100,25 @@ class StreamingDataset(IterableDataset):
 
         if input_dir.url is not None and input_dir.url.startswith("hf://"):
             if index_path is None:
-                # no index path provide, load from cache, or try indexing on the go.
+                # No index_path was provided. Attempt to load it from cache or generate it dynamically on the fly.
                 index_path = index_hf_dataset(input_dir.url)
                 cache_dir.path = index_path
                 input_dir.path = index_path
-            item_loader = ParquetLoader()
+
+            if item_loader is not None and not isinstance(item_loader, ParquetLoader):
+                raise ValueError(
+                    "Invalid item_loader for hf://datasets. "
+                    "The item_loader must be an instance of ParquetLoader. "
+                    "Please provide a valid ParquetLoader instance."
+                )
+
+            if item_loader is not None and item_loader._low_memory and shuffle:
+                raise ValueError(
+                    "You have enabled shuffling when using low memory with ParquetLoader. "
+                    "This configuration may lead to performance issues during the training process. "
+                    "Consider disabling shuffling or using a ParquetLoader without low memory mode."
+                )
+            item_loader = item_loader or ParquetLoader()
 
         self.input_dir = input_dir
         self.cache_dir = cache_dir
@@ -548,9 +562,7 @@ class StreamingDataset(IterableDataset):
                     "The provided `item_loader` state doesn't match the current one. "
                     f"Found `{self.item_loader.state_dict()}` instead of `{state['item_loader']}`."
                 )
-            logger.warning(
-                f"Overriding state item_loader {state['item_loader']} " f"to {self.item_loader.state_dict()}."
-            )
+            logger.warning(f"Overriding state item_loader {state['item_loader']} to {self.item_loader.state_dict()}.")
             state["item_loader"] = self.item_loader.state_dict()
 
         if state["drop_last"] != self.drop_last:
diff --git a/src/litdata/streaming/downloader.py b/src/litdata/streaming/downloader.py
index 02a7904..5e595a3 100644
--- a/src/litdata/streaming/downloader.py
+++ b/src/litdata/streaming/downloader.py
@@ -15,6 +15,7 @@ import contextlib
 import os
 import shutil
 import subprocess
+import tempfile
 from abc import ABC
 from contextlib import suppress
 from typing import Any, Dict, List, Optional, Type
@@ -58,9 +59,9 @@ class Downloader(ABC):
         local_chunkpath = os.path.join(self._cache_dir, chunk_filename)
         remote_chunkpath = os.path.join(self._remote_dir, chunk_filename)
 
-        self.download_file(remote_chunkpath, local_chunkpath, chunk_filename)
+        self.download_file(remote_chunkpath, local_chunkpath)
 
-    def download_file(self, remote_chunkpath: str, local_chunkpath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_chunkpath: str, local_chunkpath: str) -> None:
         pass
 
 
@@ -74,7 +75,7 @@ class S3Downloader(Downloader):
         if not self._s5cmd_available or _DISABLE_S5CMD:
             self._client = S3Client(storage_options=self._storage_options)
 
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
         obj = parse.urlparse(remote_filepath)
 
         if obj.scheme != "s3":
@@ -158,7 +159,7 @@ class GCPDownloader(Downloader):
 
         super().__init__(remote_dir, cache_dir, chunks, storage_options)
 
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
         from google.cloud import storage
 
         obj = parse.urlparse(remote_filepath)
@@ -193,7 +194,7 @@ class AzureDownloader(Downloader):
 
         super().__init__(remote_dir, cache_dir, chunks, storage_options)
 
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
         from azure.storage.blob import BlobServiceClient
 
         obj = parse.urlparse(remote_filepath)
@@ -220,7 +221,7 @@ class AzureDownloader(Downloader):
 
 
 class LocalDownloader(Downloader):
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
         if not os.path.exists(remote_filepath):
             raise FileNotFoundError(f"The provided remote_path doesn't exist: {remote_filepath}")
 
@@ -248,32 +249,42 @@ class HFDownloader(Downloader):
             )
 
         super().__init__(remote_dir, cache_dir, chunks, storage_options)
-        from huggingface_hub import HfFileSystem
 
-        self.fs = HfFileSystem()
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
+        """Download a file from the Hugging Face Hub.
+        The remote_filepath should be in the format `hf://<repo_type>/<repo_org>/<repo_name>/path`. For more
+        information, see
+        https://huggingface.co/docs/huggingface_hub/en/guides/hf_file_system#integrations.
+        """
+        from huggingface_hub import hf_hub_download
 
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
-        # for HF dataset downloading, we don't need remote_filepath, but remote_chunk_filename
-        with suppress(Timeout), FileLock(local_filepath + ".lock", timeout=0):
-            temp_path = local_filepath + ".tmp"  # Avoid partial writes
-            try:
-                with self.fs.open(remote_chunk_filename, "rb") as cloud_file, open(temp_path, "wb") as local_file:
-                    for chunk in iter(lambda: cloud_file.read(4096), b""):  # Stream in 4KB chunks local_file.
-                        local_file.write(chunk)
+        obj = parse.urlparse(remote_filepath)
+
+        if obj.scheme != "hf":
+            raise ValueError(f"Expected obj.scheme to be `hf`, instead, got {obj.scheme} for remote={remote_filepath}")
 
-                os.rename(temp_path, local_filepath)  # Atomic move after successful write
+        if os.path.exists(local_filepath):
+            return
 
-            except Exception as e:
-                print(f"Error processing {remote_chunk_filename}: {e}")
+        with suppress(Timeout), FileLock(local_filepath + ".lock", timeout=0), tempfile.TemporaryDirectory() as tmpdir:
+            _, _, _, repo_org, repo_name, path = remote_filepath.split("/", 5)
+            repo_id = f"{repo_org}/{repo_name}"
 
-            finally:
-                # Ensure cleanup of temp file if an error occurs
-                if os.path.exists(temp_path):
-                    os.remove(temp_path)
+            downloaded_path = hf_hub_download(
+                repo_id,
+                path,
+                cache_dir=tmpdir,
+                repo_type="dataset",
+                **self._storage_options,
+            )
+            if downloaded_path != local_filepath and os.path.exists(downloaded_path):
+                temp_file_path = local_filepath + ".tmp"
+                shutil.copyfile(downloaded_path, temp_file_path)
+                os.rename(temp_file_path, local_filepath)
 
 
 class LocalDownloaderWithCache(LocalDownloader):
-    def download_file(self, remote_filepath: str, local_filepath: str, remote_chunk_filename: str = "") -> None:
+    def download_file(self, remote_filepath: str, local_filepath: str) -> None:
         remote_filepath = remote_filepath.replace("local:", "")
         super().download_file(remote_filepath, local_filepath)
 
diff --git a/src/litdata/streaming/item_loader.py b/src/litdata/streaming/item_loader.py
index 1c4b24e..ea295a9 100644
--- a/src/litdata/streaming/item_loader.py
+++ b/src/litdata/streaming/item_loader.py
@@ -10,8 +10,8 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 import functools
+import logging
 import os
 from abc import ABC, abstractmethod
 from collections import defaultdict, namedtuple
@@ -29,6 +29,7 @@ from litdata.constants import (
     _MAX_WAIT_TIME,
     _NUMPY_DTYPES_MAPPING,
     _POLARS_AVAILABLE,
+    _PYARROW_AVAILABLE,
     _TORCH_DTYPES_MAPPING,
 )
 from litdata.streaming.serializers import Serializer
@@ -37,6 +38,8 @@ from litdata.utilities.encryption import Encryption, EncryptionLevel
 
 Interval = namedtuple("Interval", ["chunk_start", "roi_start_idx", "roi_end_idx", "chunk_end"])
 
+logger = logging.getLogger(__name__)
+
 
 class BaseItemLoader(ABC):
     """The base item loader is responsible to decide how the items within a chunk are loaded."""
@@ -527,13 +530,25 @@ class TokensLoader(BaseItemLoader):
 
 
 class ParquetLoader(BaseItemLoader):
-    def __init__(self) -> None:
+    def __init__(self, pre_load_chunk: bool = False, low_memory: bool = True) -> None:
         if not _POLARS_AVAILABLE:
             raise ModuleNotFoundError(
                 "You are using the Parquet item loader, which depends on `Polars > 1.0.0`.",
                 "Please, run: `pip install polars>1.0.0`",
             )
+        if not _PYARROW_AVAILABLE:
+            raise ModuleNotFoundError("Please, run: `pip install pyarrow`")
+
         self._chunk_filepaths: Dict[str, bool] = {}
+        self._pre_load_chunk = pre_load_chunk
+        self._low_memory = low_memory
+
+        if not self._low_memory:
+            logger.warning(
+                "You have set low_memory=False in ParquetLoader. "
+                "This may result in high memory usage when processing large Parquet chunk files. "
+                "Consider setting low_memory=True to reduce memory consumption."
+            )
 
     def setup(
         self,
@@ -548,7 +563,9 @@ class ParquetLoader(BaseItemLoader):
         self._data_format = self._config["data_format"]
         self._shift_idx = len(self._data_format) * 4
         self.region_of_interest = region_of_interest
-        self._df: Dict[str, Any] = {}
+        self._df: Dict[int, Any] = {}
+        self._chunk_row_groups: Dict[int, Any] = {}
+        self._chunk_row_group_item_read_count: Dict[int, Any] = {}
 
     def generate_intervals(self) -> List[Interval]:
         intervals = []
@@ -566,11 +583,14 @@ class ParquetLoader(BaseItemLoader):
         return intervals
 
     def pre_load_chunk(self, chunk_index: int, chunk_filepath: str) -> None:
-        """Logic to load the chunk in background to gain some time."""
+        """Preload the chunk in the background to gain some time."""
+        if not self._pre_load_chunk or self._low_memory:
+            return
+
         import polars as pl
 
-        if chunk_filepath not in self._df:
-            self._df[chunk_filepath] = pl.scan_parquet(chunk_filepath).collect()
+        if chunk_index not in self._df and os.path.exists(chunk_filepath):
+            self._df[chunk_index] = pl.scan_parquet(chunk_filepath, low_memory=True).collect()
 
     def load_item_from_chunk(
         self,
@@ -580,7 +600,7 @@ class ParquetLoader(BaseItemLoader):
         begin: int,
         filesize_bytes: int,
     ) -> Any:
-        """Returns an item loaded from a chunk."""
+        """Returns an item loaded from a parquet chunk."""
         if chunk_filepath in self._chunk_filepaths and not os.path.isfile(chunk_filepath):
             del self._chunk_filepaths[chunk_filepath]
 
@@ -593,21 +613,112 @@ class ParquetLoader(BaseItemLoader):
 
             self._chunk_filepaths[chunk_filepath] = True
 
-        return self.get_df(chunk_filepath).row(index - begin)
+        # relative index of the desired row within the chunk.
+        relative_index = index - begin
+        if self._low_memory:
+            return self._get_item_with_low_memory(chunk_index, chunk_filepath, relative_index)
 
-    def get_df(self, chunk_filepath: str) -> Any:
+        return self._get_item(chunk_index, chunk_filepath, relative_index)
+
+    def _get_item_with_low_memory(self, chunk_index: int, chunk_filepath: str, row_index: int) -> Any:
+        """Retrieve a dataframe row from a parquet chunk in low memory mode.
+
+        This method reads only the necessary row group from the parquet file using PyArrow and Polars,
+        which helps in reducing memory usage.
+
+        Args:
+            chunk_index (int): The index of the chunk to be accessed.
+            chunk_filepath (str): The file path of the parquet chunk.
+            row_index (int): The relative row index within the loaded chunk.
+
+        Returns:
+            Any: The dataframe row corresponding to the specified index.
+        """
         import polars as pl
+        import pyarrow.parquet as pq
+
+        # Load the Parquet file metadata if not already loaded
+        if chunk_index not in self._df:
+            self._df[chunk_index] = pq.ParquetFile(chunk_filepath)
+
+        # Determine the row group and the row index within the row group
+        parquet_file = self._df[chunk_index]
+        num_rows_per_row_group = parquet_file.metadata.row_group(0).num_rows
+        row_group_index = row_index // num_rows_per_row_group
+        row_index_within_group = row_index % num_rows_per_row_group
+
+        # Check if the row group is already loaded
+        if chunk_index in self._chunk_row_groups and row_group_index in self._chunk_row_groups[chunk_index]:
+            # Use the cached row group
+            row_group_df = self._chunk_row_groups[chunk_index][row_group_index]
+            # update read count
+            self._chunk_row_group_item_read_count[chunk_index][row_group_index] += 1
+        else:
+            # Load the row group and convert it to a Polars DataFrame
+            row_group = self._df[chunk_index].read_row_group(row_group_index)
+            row_group_df = pl.from_arrow(row_group)
+
+            # Cache the loaded row group
+            if chunk_index not in self._chunk_row_groups:
+                self._chunk_row_groups[chunk_index] = {}
+                self._chunk_row_group_item_read_count[chunk_index] = {}
 
-        if chunk_filepath not in self._df:
-            self._df[chunk_filepath] = pl.scan_parquet(chunk_filepath).collect()
-        return self._df[chunk_filepath]
+            self._chunk_row_groups[chunk_index][row_group_index] = row_group_df
+            self._chunk_row_group_item_read_count[chunk_index][row_group_index] = 1
+
+        # Check if the row group has been fully read and release memory if necessary
+        read_count = self._chunk_row_group_item_read_count[chunk_index][row_group_index]
+        if read_count >= num_rows_per_row_group:
+            # Release memory for the fully read row group
+            del self._chunk_row_groups[chunk_index][row_group_index]
+            del self._chunk_row_group_item_read_count[chunk_index][row_group_index]
+
+        # Return the specific row from the dataframe
+        return row_group_df.row(row_index_within_group)  # type: ignore
+
+    def _get_item(self, chunk_index: int, chunk_filepath: str, index: int) -> Any:
+        """Retrieve a dataframe row from a parquet chunk by loading the entire chunk into memory.
+
+        Note:
+            This method reads the complete parquet file using Polars. Exercise caution with large files as it
+            may significantly increase memory usage.
+
+        Args:
+            chunk_index (int): The index of the chunk to be accessed.
+            chunk_filepath (str): The file path of the parquet chunk.
+            index (int): The relative row index within the loaded chunk.
+
+        Returns:
+            Any: The dataframe row corresponding to the specified index.
+        """
+        import polars as pl
+
+        if chunk_index not in self._df:
+            self._df[chunk_index] = pl.scan_parquet(chunk_filepath, low_memory=True).collect()
+        return self._df[chunk_index].row(index)
 
     def delete(self, chunk_index: int, chunk_filepath: str) -> None:
         """Delete a chunk from the local filesystem."""
+        if chunk_index in self._df:
+            del self._df[chunk_index]
+        if chunk_index in self._chunk_row_groups:
+            del self._chunk_row_groups[chunk_index]
+
+        if chunk_index in self._chunk_row_group_item_read_count:
+            del self._chunk_row_group_item_read_count[chunk_index]
         if os.path.exists(chunk_filepath):
             os.remove(chunk_filepath)
-        if chunk_filepath in self._df:
-            del self._df[chunk_filepath]
+
+    def close(self, chunk_index: int) -> None:
+        """Release the memory-mapped file for a specific chunk index."""
+        if chunk_index in self._df:
+            del self._df[chunk_index]
+
+        if chunk_index in self._chunk_row_groups:
+            del self._chunk_row_groups[chunk_index]
+
+        if chunk_index in self._chunk_row_group_item_read_count:
+            del self._chunk_row_group_item_read_count[chunk_index]
 
     def encode_data(self, data: List[bytes], sizes: List[int], flattened: List[Any]) -> Any:
         pass
diff --git a/src/litdata/streaming/reader.py b/src/litdata/streaming/reader.py
index 41c5afc..c71a350 100644
--- a/src/litdata/streaming/reader.py
+++ b/src/litdata/streaming/reader.py
@@ -24,7 +24,7 @@ from filelock import FileLock, Timeout
 
 from litdata.constants import _DEBUG
 from litdata.streaming.config import ChunksConfig, Interval
-from litdata.streaming.item_loader import BaseItemLoader, PyTreeLoader, TokensLoader
+from litdata.streaming.item_loader import BaseItemLoader, ParquetLoader, PyTreeLoader, TokensLoader
 from litdata.streaming.sampler import ChunkedIndex
 from litdata.streaming.serializers import Serializer, _get_serializers
 from litdata.utilities.encryption import Encryption
@@ -401,7 +401,7 @@ class BinaryReader:
 
         if index.chunk_index != self._last_chunk_index:
             # Close the memory-mapped file for the last chunk index
-            if isinstance(self._item_loader, TokensLoader) and self._last_chunk_index is not None:
+            if isinstance(self._item_loader, (TokensLoader, ParquetLoader)) and self._last_chunk_index is not None:
                 self._item_loader.close(self._last_chunk_index)
 
             # track the new chunk index as the latest one
diff --git a/src/litdata/streaming/writer.py b/src/litdata/streaming/writer.py
index 0adc34d..2771106 100644
--- a/src/litdata/streaming/writer.py
+++ b/src/litdata/streaming/writer.py
@@ -566,22 +566,21 @@ def index_parquet_dataset(
     pq_dir_url: str,
     cache_dir: Optional[str] = None,
     storage_options: Optional[Dict] = {},
-    remove_after_indexing: bool = True,
     num_workers: int = 4,
 ) -> None:
     """Index a Parquet dataset from a specified URL.
 
-    This function scans all `.parquet` files in the specified directory URL, extracts metadata
-    such as file size, chunk size, and data types, and indexes them. Optionally, the files
-    can be removed after indexing.
+
+    This function scans the metadata of all `.parquet` files in the specified directory URL, extracts metadata
+    such as file size, chunk size, and data types, and indexes them. The index is saved as a JSON file in the
+    specified cache directory.
 
     Args:
         pq_dir_url (str): URL of the directory containing the Parquet files.
         cache_dir (Optional[str]): Local cache directory for storing temporary files.
             For HF dataset, index.json file will be stored here.
         storage_options (Optional[Dict]): Additional storage options for accessing the Parquet files.
-        remove_after_indexing (bool): Whether to remove files after indexing (default is True).
-        num_workers (int): Number of workers to download files and index them.
+        num_workers (int): Number of workers to download metadata of Parquet files and index them.
 
     Raises:
         ModuleNotFoundError: If the required `polars` module is not installed.
@@ -589,8 +588,6 @@ def index_parquet_dataset(
     if not _POLARS_AVAILABLE:
         raise ModuleNotFoundError("Please, run: `pip install polars`")
 
-    import polars as pl
-
     pq_chunks_info = []
     config: Dict[str, Any] = {
         "compression": None,
@@ -602,7 +599,7 @@ def index_parquet_dataset(
         "item_loader": ParquetLoader.__name__,
     }
 
-    pq_dir_class = get_parquet_indexer_cls(pq_dir_url, cache_dir, storage_options, remove_after_indexing, num_workers)
+    pq_dir_class = get_parquet_indexer_cls(pq_dir_url, cache_dir, storage_options, num_workers)
 
     if _TQDM_AVAILABLE:
         from tqdm.auto import tqdm as _tqdm
@@ -616,13 +613,11 @@ def index_parquet_dataset(
             dynamic_ncols=True,
             unit="step",
         )
-    # iterate the directory and for all files ending in `.parquet` index them
-    for file_name, file_path in pq_dir_class:
-        file_size = os.path.getsize(file_path)
-        pq_polars = pl.scan_parquet(file_path)
-        chunk_dtypes = pq_polars.collect_schema().dtypes()
-        chunk_dtypes = [str(dt) for dt in chunk_dtypes]
-        chunk_size = pq_polars.select(pl.count()).collect().item()
+
+    results = {}
+    # iterate through the directory and index each file ending with ".parquet"
+    for file_metadata, order in pq_dir_class:
+        chunk_dtypes = file_metadata["data_types"]
 
         if len(config["data_format"]) != 0 and config["data_format"] != chunk_dtypes:
             raise Exception(
@@ -631,14 +626,18 @@ def index_parquet_dataset(
             )
         config["data_format"] = chunk_dtypes
         chunk_info = {
-            "chunk_bytes": file_size,
-            "chunk_size": chunk_size,
-            "filename": file_name,
+            "chunk_bytes": file_metadata["file_size"],
+            "chunk_size": file_metadata["num_rows"],
+            "filename": file_metadata["file_name"],
             "dim": None,
         }
-        pq_chunks_info.append(chunk_info)
+        results[order] = chunk_info
         if _TQDM_AVAILABLE:
             pbar.update(1)
 
+    for i in sorted(results.keys()):
+        pq_chunks_info.append(results[i])
+
+    del results
     print(flush=True)  # to prevent truncated printing when using concurrent threads/processes
     pq_dir_class.write_index(pq_chunks_info, config)
diff --git a/src/litdata/utilities/hf_dataset.py b/src/litdata/utilities/hf_dataset.py
index f9fce4b..7740c04 100644
--- a/src/litdata/utilities/hf_dataset.py
+++ b/src/litdata/utilities/hf_dataset.py
@@ -8,20 +8,27 @@ from litdata.utilities.parquet import default_cache_dir
 
 
 def index_hf_dataset(hf_url: str) -> str:
+    """Index a Hugging Face dataset and return the cache directory path.
+
+    Args:
+        hf_url (str): The URL of the Hugging Face dataset.
+
+    Returns:
+        str: The path to the cache directory containing the index.
+    """
     if not hf_url.startswith("hf://"):
-        raise ValueError(f"Invalid Hugging Face dataset URL: {hf_url}. Expected URL to start with 'hf://'.")
+        raise ValueError(
+            f"Invalid Hugging Face dataset URL: {hf_url}. "
+            "The URL should start with 'hf://'. Please check the URL and try again."
+        )
 
     cache_dir = default_cache_dir(hf_url)
-
     cache_index_path = os.path.join(cache_dir, _INDEX_FILENAME)
-    print("=" * 50)
 
     if os.path.exists(cache_index_path):
-        print(f"Found HF index.json file in {cache_index_path}.")
+        print(f"Found existing HF {_INDEX_FILENAME} file for {hf_url} at {cache_index_path}.")
     else:
-        print("Indexing HF dataset...")
-        index_parquet_dataset(hf_url, cache_dir, remove_after_indexing=False)
-
-    print("=" * 50)
+        print(f"Indexing HF dataset from {hf_url} into {cache_index_path}.")
+        index_parquet_dataset(hf_url, cache_dir, num_workers=os.cpu_count() or 4)
 
     return cache_dir
diff --git a/src/litdata/utilities/parquet.py b/src/litdata/utilities/parquet.py
index 03bb9d9..01bfbf2 100644
--- a/src/litdata/utilities/parquet.py
+++ b/src/litdata/utilities/parquet.py
@@ -1,97 +1,66 @@
 """contains utility functions to return parquet files from local, s3, or gs."""
 
 import hashlib
+import io
 import json
 import os
-import sys
-import threading
 from abc import ABC, abstractmethod
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import suppress
-from queue import Queue
-from time import sleep, time
+from time import time
 from typing import Any, Dict, Generator, List, Optional, Tuple, Union
 from urllib import parse
 
-from litdata.constants import _FSSPEC_AVAILABLE, _HF_HUB_AVAILABLE, _INDEX_FILENAME, _TQDM_AVAILABLE
+from litdata.constants import _FSSPEC_AVAILABLE, _HF_HUB_AVAILABLE, _INDEX_FILENAME, _PYARROW_AVAILABLE
 from litdata.streaming.resolver import Dir, _resolve_dir
 
 
-def delete_thread(rmq: Queue) -> None:
-    while True:
-        file_path = rmq.get()
-        if file_path is None:  # Sentinel value to exit
-            break
-        with suppress(FileNotFoundError):
-            os.remove(file_path)
-
-    # before exiting, just sleep for some time, to complete any pending deletions
-    sleep(0.5)
-
-
 class ParquetDir(ABC):
     def __init__(
         self,
         dir_path: Optional[Union[str, Dir]],
         cache_path: Optional[str] = None,
         storage_options: Optional[Dict] = {},
-        remove_after_indexing: bool = True,
         num_workers: int = 4,
     ):
         self.dir = _resolve_dir(dir_path)
         self.cache_path = cache_path
         self.storage_options = storage_options
-        self.remove_after_indexing = remove_after_indexing
         self.files: List[Any] = []
-        self.process_queue: Queue = Queue()
-        self.delete_queue: Queue = Queue()
         self.num_workers = num_workers
 
-    def __iter__(self) -> Generator[Tuple[str, str], None, None]:
-        # start worker in a separate thread, and then read values from `process_queue`, and yield
+    def __iter__(self) -> Generator[Tuple[Dict[str, Any], int], None, None]:
+        """Iterate over the Parquet files and yield their metadata.
 
-        self.is_delete_thread_running = self.dir.url is not None and self.remove_after_indexing
-
-        if self.is_delete_thread_running:
-            t = threading.Thread(
-                target=delete_thread,
-                name="delete_thread",
-                args=(self.delete_queue,),
-            )
-            t.start()
-
-        t_worker = threading.Thread(
-            target=self.worker,
-            name="worker_thread",
-            args=(),
-        )
-        t_worker.start()
-
-        while True:
-            file_name, file_path = self.process_queue.get()
-            if file_name is None and file_path is None:  # Sentinel value to exit
-                break
-            yield file_name, file_path
-            if self.is_delete_thread_running:
-                self.delete_queue.put_nowait(file_path)
-
-        if self.is_delete_thread_running:
-            self.delete_queue.put_nowait(None)  # so that it doesn't hang indefinitely
-            t.join()
+        Yields:
+            Generator[Tuple[str, int], None, None]: A generator yielding tuples of file name, file path, and order.
+        """
+        with ThreadPoolExecutor(max_workers=self.num_workers) as executor:
+            futures = {executor.submit(self.task, _file): (order, _file) for order, _file in enumerate(self.files)}
+            for future in futures:
+                file_metadata = future.result()
+                order, _ = futures[future]
+                yield file_metadata, order
 
-        t_worker.join()  # should've completed by now. But, just to be on safe side.
+    @abstractmethod
+    def task(self, _file: Any) -> Dict[str, Any]: ...
 
     @abstractmethod
-    def task(self, _file: Any) -> None: ...
+    def write_index(self, chunks_info: List[Dict[str, Any]], config: Dict[str, Any]) -> None:
+        """Write the index file to the cache directory.
 
-    def worker(self) -> None:
-        with ThreadPoolExecutor(max_workers=self.num_workers) as executor:
-            for _file in self.files:
-                executor.submit(self.task, _file)
-        self.process_queue.put_nowait((None, None))
+        Args:
+            chunks_info (List[Dict[str, Any]]): List of dictionaries containing chunk metadata.
+            config (Dict[str, Any]): Configuration dictionary containing metadata.
+        """
+        assert self.cache_path is not None, "Cache path is not set."
+        index_file_path = os.path.join(self.cache_path, _INDEX_FILENAME)
 
-    @abstractmethod
-    def write_index(self, chunks_info: List[Dict[str, Any]], config: Dict[str, Any]) -> None: ...
+        # write to index.json file
+        with open(index_file_path, "w") as f:
+            data = {"chunks": chunks_info, "config": config, "updated_at": str(time())}
+            json.dump(data, f, sort_keys=True)
+
+        print(f"Index file successfully written to: {index_file_path}")
 
 
 class LocalParquetDir(ParquetDir):
@@ -100,35 +69,47 @@ class LocalParquetDir(ParquetDir):
         dir_path: Optional[Union[str, Dir]],
         cache_path: Optional[str] = None,
         storage_options: Optional[Dict] = {},
-        remove_after_indexing: bool = True,
         num_workers: int = 4,
     ):
-        super().__init__(dir_path, cache_path, storage_options, remove_after_indexing, num_workers)
+        if not _PYARROW_AVAILABLE:
+            raise ModuleNotFoundError(
+                "The 'pyarrow' module is required for processing Parquet files. "
+                "Please install it by running: `pip install pyarrow`"
+            )
+        super().__init__(dir_path, cache_path, storage_options, num_workers)
 
         for _f in os.listdir(self.dir.path):
             if _f.endswith(".parquet"):
                 self.files.append(_f)
 
-    def task(self, _file: str) -> None:
-        assert isinstance(_file, str)
-
-        if _file.endswith(".parquet"):
-            assert self.dir.path is not None
-            assert self.dir.path != "", "Dir path can't be empty"
+    def task(self, _file: str) -> Dict[str, Any]:
+        """Extract metadata from a Parquet file on the local filesystem."""
+        import pyarrow.parquet as pq
 
-            file_path = os.path.join(self.dir.path, _file)
-            self.process_queue.put_nowait((_file, file_path))
+        assert isinstance(_file, str)
+        assert _file.endswith(".parquet")
+        assert self.dir.path is not None
+        assert self.dir.path != "", "Dir path can't be empty"
+
+        file_path = os.path.join(self.dir.path, _file)
+        parquet_file = pq.ParquetFile(file_path)
+        num_rows = parquet_file.metadata.num_rows
+        data_types = [str(col.type) for col in parquet_file.schema_arrow]
+        file_size = os.path.getsize(file_path)
+        file_name = os.path.basename(file_path)
+
+        return {
+            "file_name": file_name,
+            "num_rows": num_rows,
+            "file_size": file_size,
+            "data_types": data_types,
+        }
 
     def write_index(self, chunks_info: List[Dict[str, Any]], config: Dict[str, Any]) -> None:
-        # write to index.json file
+        """Write the index file to the cache directory."""
         if self.cache_path is None:
             self.cache_path = self.dir.path
-        assert self.cache_path is not None
-        with open(os.path.join(self.cache_path, _INDEX_FILENAME), "w") as f:
-            data = {"chunks": chunks_info, "config": config, "updated_at": str(time())}
-            json.dump(data, f, sort_keys=True)
-
-        print(f"Index file written to: {os.path.join(self.cache_path, _INDEX_FILENAME)}")
+        super().write_index(chunks_info, config)
 
 
 class CloudParquetDir(ParquetDir):
@@ -137,15 +118,18 @@ class CloudParquetDir(ParquetDir):
         dir_path: Optional[Union[str, Dir]],
         cache_path: Optional[str] = None,
         storage_options: Optional[Dict] = None,
-        remove_after_indexing: bool = True,
         num_workers: int = 4,
     ):
         if not _FSSPEC_AVAILABLE:
             raise ModuleNotFoundError(
                 "Support for Indexing cloud parquet files depends on `fsspec`.", "Please, run: `pip install fsspec`"
             )
-
-        super().__init__(dir_path, cache_path, storage_options, remove_after_indexing, num_workers)
+        if not _PYARROW_AVAILABLE:
+            raise ModuleNotFoundError(
+                "The 'pyarrow' module is required for processing Parquet files. "
+                "Please install it by running: `pip install pyarrow`"
+            )
+        super().__init__(dir_path, cache_path, storage_options, num_workers)
 
         assert self.dir.url is not None
 
@@ -169,48 +153,64 @@ class CloudParquetDir(ParquetDir):
             if _f["type"] == "file" and _f["name"].endswith(".parquet"):
                 self.files.append(_f)
 
-    def task(self, _file: Any) -> None:
-        if _file["type"] == "file" and _file["name"].endswith(".parquet"):
-            file_name = os.path.basename(_file["name"])
-            assert self.cache_path is not None
-            local_path = os.path.join(self.cache_path, file_name)
-            temp_path = local_path + ".tmp"  # Avoid partial writes
-            # if an existing temp file is present, means its corrupted
-            if os.path.exists(temp_path):
-                os.remove(temp_path)
+    def task(self, _file: Any) -> Dict[str, Any]:
+        """Extract metadata from a Parquet file on the cloud filesystem without downloading the entire file."""
+        import pyarrow.parquet as pq
+
+        # Validate inputs
+        if not isinstance(_file, dict) or "name" not in _file or "size" not in _file:
+            raise ValueError(f"Invalid file object: {_file}")
+
+        assert _file["name"].endswith(".parquet")
+
+        file_name = os.path.basename(_file["name"])
+        file_size = _file["size"]
 
-            # Download the file
-            with self.fs.open(_file["name"], "rb") as cloud_file, open(temp_path, "wb") as local_file:
-                for chunk in iter(lambda: cloud_file.read(4096), b""):  # Read in 4KB chunks
-                    local_file.write(chunk)
+        with self.fs.open(_file["name"], "rb") as f:
+            # Read footer size (last 8 bytes: 4 for footer size + 4 for magic number)
+            f.seek(file_size - 8)
+            footer_size = int.from_bytes(f.read(4), "little")
 
-            os.rename(temp_path, local_path)  # Atomic move after successful write
-            self.process_queue.put_nowait((file_name, local_path))
+            # seek to the start of the footer and read the footer data
+            footer_start = file_size - footer_size - 8
+            f.seek(footer_start)
+            footer_data = f.read(file_size - footer_start)
+
+            if len(footer_data) != file_size - footer_start:
+                raise ValueError(f"Failed to read complete footer data from {file_name}")
+
+        # Parse the footer data to extract schema information
+        with io.BytesIO(footer_data) as footer_buffer:
+            parquet_file = pq.ParquetFile(footer_buffer)
+            data_types = [str(col.type) for col in parquet_file.schema_arrow]
+            num_rows = parquet_file.metadata.num_rows
+
+        return {
+            "file_name": file_name,
+            "num_rows": num_rows,
+            "file_size": file_size,
+            "data_types": data_types,
+        }
 
     def write_index(self, chunks_info: List[Dict[str, Any]], config: Dict[str, Any]) -> None:
+        """Write the index file to the local cache directory and upload it to the cloud."""
         assert self.cache_path is not None
         assert self.dir.url is not None
 
-        # clear any `.lock` or `.tmp` chunk file if left
-        if self.is_delete_thread_running:
-            for file in os.listdir(self.cache_path):
-                if file != _INDEX_FILENAME:
-                    with suppress(FileNotFoundError):
-                        os.remove(file)
-
         index_file_path = os.path.join(self.cache_path, _INDEX_FILENAME)
         cloud_index_path = os.path.join(self.dir.url, _INDEX_FILENAME)
+
         # write to index.json file
         with open(index_file_path, "w") as f:
             data = {"chunks": chunks_info, "config": config, "updated_at": str(time())}
             json.dump(data, f, sort_keys=True)
 
-        # upload file to cloud
+        # upload index file to cloud
         with open(index_file_path, "rb") as local_file, self.fs.open(cloud_index_path, "wb") as cloud_file:
             for chunk in iter(lambda: local_file.read(4096), b""):  # Read in 4KB chunks
                 cloud_file.write(chunk)
 
-        print(f"Index file written to: {cloud_index_path}")
+        print(f"Index file successfully written to: {cloud_index_path}")
         if os.path.exists(index_file_path):
             os.remove(index_file_path)
 
@@ -221,15 +221,18 @@ class HFParquetDir(ParquetDir):
         dir_path: Optional[Union[str, Dir]],
         cache_path: Optional[str] = None,
         storage_options: Optional[Dict] = None,
-        remove_after_indexing: bool = True,
         num_workers: int = 4,
     ):
         if not _HF_HUB_AVAILABLE:
             raise ModuleNotFoundError(
                 "Support for Indexing HF depends on `huggingface_hub`.", "Please, run: `pip install huggingface_hub"
             )
-
-        super().__init__(dir_path, cache_path, storage_options, remove_after_indexing, num_workers)
+        if not _PYARROW_AVAILABLE:
+            raise ModuleNotFoundError(
+                "The 'pyarrow' module is required for processing Parquet files. "
+                "Please install it by running: `pip install pyarrow`"
+            )
+        super().__init__(dir_path, cache_path, storage_options, num_workers)
 
         assert self.dir.url is not None
         assert self.dir.url.startswith("hf")
@@ -243,66 +246,79 @@ class HFParquetDir(ParquetDir):
 
         self.fs = HfFileSystem()
 
-        for _f in self.fs.ls(self.dir.url, detail=False):
-            if _f.endswith(".parquet"):
+        for _f in self.fs.ls(self.dir.url, detail=True):
+            if isinstance(_f, dict) and _f["name"].endswith(".parquet"):
                 self.files.append(_f)
 
-    def task(self, _file: str) -> None:
-        assert isinstance(_file, str)
-        assert self.cache_path is not None
-        if _file.endswith(".parquet"):
-            file_name = os.path.basename(_file)
-            local_path = os.path.join(self.cache_path, file_name)
-            os.makedirs(os.path.dirname(local_path), exist_ok=True)
-            temp_path = local_path + ".tmp"  # Avoid partial writes
-            # if an existing temp file is present, means its corrupted
-            if os.path.exists(temp_path):
-                os.remove(temp_path)
+    def task(self, _file: dict) -> Dict[str, Any]:
+        """Extract metadata from a Parquet file on Hugging Face Hub without downloading the entire file."""
+        import pyarrow.parquet as pq
+
+        # Validate inputs
+        if not isinstance(_file, dict) or "name" not in _file or "size" not in _file:
+            raise ValueError(f"Invalid file object: {_file}")
 
-            with self.fs.open(_file, "rb") as cloud_file, open(temp_path, "wb") as local_file:
-                if _TQDM_AVAILABLE:
-                    from tqdm.auto import tqdm as _tqdm
+        assert _file["name"].endswith(".parquet")
 
-                    file_size = self.fs.info(_file)["size"]
-                    pbar = _tqdm(desc=f"Downloading {file_name}", total=file_size, unit="B", unit_scale=True)
+        file_name = os.path.basename(_file["name"])
+        file_size = _file["size"]
 
-                for chunk in iter(lambda: cloud_file.read(4096), b""):  # Read in 4KB chunks
-                    local_file.write(chunk)
+        with self.fs.open(_file["name"], "rb") as f:
+            # Read footer size (last 8 bytes: 4 for footer size + 4 for magic number)
+            f.seek(file_size - 8)
+            footer_size = int.from_bytes(f.read(4), "little")
 
-                    if _TQDM_AVAILABLE:
-                        pbar.update(len(chunk))
+            # seek to the start of the footer and read the footer data
+            footer_start = file_size - footer_size - 8
+            f.seek(footer_start)
+            footer_data = f.read(file_size - footer_start)
 
-            os.rename(temp_path, local_path)  # Atomic move after successful write
-            self.process_queue.put_nowait((file_name, local_path))
+            if len(footer_data) != file_size - footer_start:
+                raise ValueError(f"Failed to read complete footer data from {file_name}")
+
+        # Parse the footer data to extract schema information
+        with io.BytesIO(footer_data) as footer_buffer:
+            parquet_file = pq.ParquetFile(footer_buffer)
+            data_types = [str(col.type) for col in parquet_file.schema_arrow]
+            num_rows = parquet_file.metadata.num_rows
+
+        return {
+            "file_name": file_name,
+            "num_rows": num_rows,
+            "file_size": file_size,
+            "data_types": data_types,
+        }
 
     def write_index(self, chunks_info: List[Dict[str, Any]], config: Dict[str, Any]) -> None:
+        """Write the index file to the cache directory."""
         assert self.cache_path is not None
         assert self.dir.url is not None
 
-        # clear any `.lock` or `.tmp` chunk file if left
-        if self.is_delete_thread_running:
-            for file in os.listdir(self.cache_path):
-                if file != _INDEX_FILENAME:
-                    with suppress(FileNotFoundError):
-                        os.remove(file)
-
-        index_file_path = os.path.join(self.cache_path, _INDEX_FILENAME)
-        # write to index.json file
-        with open(index_file_path, "w") as f:
-            data = {"chunks": chunks_info, "config": config, "updated_at": str(time())}
-            json.dump(data, f, sort_keys=True)
-
-        print(f"Index file written to: {index_file_path}")
+        super().write_index(chunks_info, config)
 
 
 def get_parquet_indexer_cls(
     dir_path: str,
     cache_path: Optional[str] = None,
     storage_options: Optional[Dict] = {},
-    remove_after_indexing: bool = True,
     num_workers: int = 4,
 ) -> ParquetDir:
-    args = (dir_path, cache_path, storage_options, remove_after_indexing, num_workers)
+    """Get the appropriate ParquetDir class based on the directory path scheme.
+
+    Args:
+        dir_path (str): Path to the directory containing the Parquet files.
+        cache_path (Optional[str]): Local cache directory for storing temporary files.
+        storage_options (Optional[Dict]): Additional storage options for accessing the Parquet files.
+        remove_after_indexing (bool): Whether to remove files after indexing (default is True).
+        num_workers (int): Number of workers to download metadata of Parquet files and index them.
+
+    Returns:
+        ParquetDir: An instance of the appropriate ParquetDir class.
+
+    Raises:
+        ValueError: If the provided `dir_path` does not have an associated ParquetDir class.
+    """
+    args = (dir_path, cache_path, storage_options, num_workers)
 
     obj = parse.urlparse(dir_path)
 
@@ -313,18 +329,33 @@ def get_parquet_indexer_cls(
     if obj.scheme == "hf":
         return HFParquetDir(*args)
 
-    if sys.platform == "win32":
-        return LocalParquetDir(*args)
-
+    supported_schemes = ["local", "gs", "s3", "hf"]
     raise ValueError(
-        f"The provided `dir_path` {dir_path} doesn't have a ParquetDir class associated.",
-        f"Found scheme => {obj.scheme}",
+        f"The provided `dir_path` '{dir_path}' does not have an associated ParquetDir class. "
+        f"Found scheme: '{obj.scheme}'. Supported schemes are: {', '.join(supported_schemes)}. "
+        "Please provide a valid directory path with one of the supported schemes."
     )
 
 
 def default_cache_dir(url: str) -> str:
-    # Hash the URL using SHA256 and take the first 16 characters for brevity
+    """Generate a default cache directory path based on the given URL.
+
+    The directory is created under the user's home directory at
+    ~/.cache/litdata-cache-index-pq if it does not already exist.
+
+    Args:
+        url (str): The URL to be hashed for generating the cache directory path.
+
+    Returns:
+        str: The path to the generated cache directory.
+    """
+    # Hash the URL using SHA256
     url_hash = hashlib.sha256(url.encode()).hexdigest()
+
+    # Generate the cache directory path
     cache_path = os.path.join(os.path.expanduser("~"), ".cache", "litdata-cache-index-pq", url_hash)
-    os.makedirs(cache_path, exist_ok=True)  # Ensure the directory exists
+
+    # Ensure the directory exists
+    os.makedirs(cache_path, exist_ok=True)
+
     return cache_path


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 6143dee..ec3cd72 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -181,7 +181,17 @@ def clean_pq_index_cache():
 
 
 @pytest.fixture
-def huggingface_hub_mock(monkeypatch, write_pq_data, tmp_path):
+def huggingface_hub_mock(monkeypatch):
+    huggingface_hub = ModuleType("huggingface_hub")
+    monkeypatch.setitem(sys.modules, "huggingface_hub", huggingface_hub)
+    hf_hub_download = ModuleType("hf_hub_download")
+    monkeypatch.setitem(sys.modules, "huggingface_hub.hf_hub_download", hf_hub_download)
+    huggingface_hub.hf_hub_download = hf_hub_download
+    return huggingface_hub
+
+
+@pytest.fixture
+def huggingface_hub_fs_mock(monkeypatch, write_pq_data, tmp_path):
     huggingface_hub = ModuleType("huggingface_hub")
     hf_file_system = ModuleType("hf_file_system")
 
@@ -196,13 +206,25 @@ def huggingface_hub_mock(monkeypatch, write_pq_data, tmp_path):
         assert os.path.exists(file_path), "file hf is trying to access, doesn't exist."
         return open(file_path, mode)
 
+    def mock_ls(*args, **kwargs):
+        files = sorted(os.listdir(os.path.join(tmp_path, "pq-dataset")))
+        return [
+            {
+                "type": "file",
+                "name": file_name,
+                "size": os.path.getsize(os.path.join(tmp_path, "pq-dataset", file_name)),
+            }
+            for file_name in files
+        ]
+
+    def mock_hf_hub_download(repo_id, filename, cache_dir, repo_type, **kwargs):
+        return os.path.join(tmp_path, "pq-dataset", os.path.basename(filename))
+
     hf_fs_mock = Mock()
-    hf_fs_mock.ls = Mock(side_effect=lambda *args, **kwargs: os.listdir(os.path.join(tmp_path, "pq-dataset")))
+    hf_fs_mock.ls = Mock(side_effect=mock_ls)
     hf_fs_mock.open = Mock(side_effect=mock_open)
-    hf_fs_mock.info = Mock(
-        side_effect=lambda filename: {"size": os.path.getsize(os.path.join(tmp_path, "pq-dataset", filename))}
-    )
     huggingface_hub.HfFileSystem = Mock(return_value=hf_fs_mock)
+    huggingface_hub.hf_hub_download = Mock(side_effect=mock_hf_hub_download)
 
     return huggingface_hub
 
@@ -220,7 +242,13 @@ def fsspec_pq_mock(monkeypatch, write_pq_data, tmp_path, fsspec_mock):
     def mock_fsspec_ls(*args, **kwargs):
         file_list = []
         for file_name in os.listdir(os.path.join(tmp_path, "pq-dataset")):
-            file_list.append({"type": "file", "name": file_name})
+            file_list.append(
+                {
+                    "type": "file",
+                    "name": file_name,
+                    "size": os.path.getsize(os.path.join(tmp_path, "pq-dataset", file_name)),
+                }
+            )
         return file_list
 
     fs_mock = Mock()
diff --git a/tests/processing/test_readers.py b/tests/processing/test_readers.py
index ff4b33e..7b81809 100644
--- a/tests/processing/test_readers.py
+++ b/tests/processing/test_readers.py
@@ -4,7 +4,8 @@ import sys
 import pytest
 
 from litdata import map
-from litdata.processing.readers import _PYARROW_AVAILABLE, BaseReader, ParquetReader
+from litdata.constants import _PYARROW_AVAILABLE
+from litdata.processing.readers import BaseReader, ParquetReader
 
 
 class DummyReader(BaseReader):
diff --git a/tests/streaming/test_downloader.py b/tests/streaming/test_downloader.py
index 52bace5..2a9c144 100644
--- a/tests/streaming/test_downloader.py
+++ b/tests/streaming/test_downloader.py
@@ -1,4 +1,5 @@
 # ruff: noqa: S604
+import contextlib
 import os
 import sys
 from unittest import mock
@@ -11,6 +12,7 @@ from litdata.streaming.downloader import (
     AzureDownloader,
     Downloader,
     GCPDownloader,
+    HFDownloader,
     LocalDownloaderWithCache,
     S3Downloader,
     get_downloader,
@@ -312,3 +314,38 @@ def test_download_with_cache(tmpdir, monkeypatch):
         os_mock.assert_called()
     finally:
         os.remove("a.txt")
+
+
+@mock.patch("litdata.streaming.downloader._HF_HUB_AVAILABLE", True)
+def test_hf_downloader(tmpdir, huggingface_hub_mock):
+    # Create a mock for hf_hub_download
+    mock_hf_hub_download = MagicMock()
+    huggingface_hub_mock.hf_hub_download = mock_hf_hub_download
+
+    # Initialize the downloader
+    storage_options = {}
+    downloader = HFDownloader("hf://datasets/sample_org/sample_repo", tmpdir, [], storage_options)
+    local_filepath = os.path.join(tmpdir, "a.txt")
+
+    # Configure the mock to return the local_filepath
+    mock_hf_hub_download.return_value = local_filepath
+
+    # Test case 1: File doesn’t exist, should download
+    with contextlib.suppress(FileNotFoundError):
+        downloader.download_file("hf://datasets/sample_org/sample_repo/a.txt", local_filepath)
+
+    # Verify that hf_hub_download was called with the correct arguments
+    huggingface_hub_mock.hf_hub_download.assert_called_once()
+
+    # Reset the mock for the next test case
+    mock_hf_hub_download.reset_mock()
+
+    # Test case 2: File exists, should skip download
+    with open(local_filepath, "w") as f:
+        f.write("dummy content")
+
+    with contextlib.suppress(FileNotFoundError):
+        downloader.download_file("hf://datasets/sample_org/sample_repo/a.txt", local_filepath)
+
+    # Verify that hf_hub_download was not called
+    mock_hf_hub_download.assert_not_called()
diff --git a/tests/streaming/test_parquet.py b/tests/streaming/test_parquet.py
index 5a37870..1d41e1c 100644
--- a/tests/streaming/test_parquet.py
+++ b/tests/streaming/test_parquet.py
@@ -2,24 +2,22 @@ import hashlib
 import json
 import os
 import sys
-import threading
-import time
 from contextlib import nullcontext
-from queue import Queue
 from types import ModuleType
-from unittest.mock import Mock
+from unittest.mock import Mock, patch
 
 import pytest
 
 from litdata.constants import _INDEX_FILENAME
 from litdata.streaming.dataset import StreamingDataset
+from litdata.streaming.item_loader import ParquetLoader, PyTreeLoader
 from litdata.streaming.writer import index_parquet_dataset
+from litdata.utilities.hf_dataset import index_hf_dataset
 from litdata.utilities.parquet import (
     CloudParquetDir,
     HFParquetDir,
     LocalParquetDir,
     default_cache_dir,
-    delete_thread,
     get_parquet_indexer_cls,
 )
 
@@ -33,17 +31,16 @@ from litdata.utilities.parquet import (
         None,
         "s3://some_bucket/some_path",
         "gs://some_bucket/some_path",
-        "hf://some_bucket/some_path",
+        "hf://datasets/some_org/some_repo/some_path",
     ],
 )
-@pytest.mark.parametrize(("remove_after_indexing"), [True, False])
 @pytest.mark.parametrize(("num_worker"), [None, 1, 2, 4])
+@patch("litdata.utilities.parquet._HF_HUB_AVAILABLE", True)
+@patch("litdata.streaming.downloader._HF_HUB_AVAILABLE", True)
+@patch("litdata.utilities.parquet._FSSPEC_AVAILABLE", True)
 def test_parquet_index_write(
-    monkeypatch, tmp_path, pq_data, huggingface_hub_mock, fsspec_pq_mock, pq_dir_url, remove_after_indexing, num_worker
+    monkeypatch, tmp_path, pq_data, huggingface_hub_fs_mock, fsspec_pq_mock, pq_dir_url, num_worker
 ):
-    monkeypatch.setattr("litdata.utilities.parquet._HF_HUB_AVAILABLE", True)
-    monkeypatch.setattr("litdata.utilities.parquet._FSSPEC_AVAILABLE", True)
-
     if pq_dir_url is None:
         pq_dir_url = os.path.join(tmp_path, "pq-dataset")
 
@@ -57,30 +54,24 @@ def test_parquet_index_write(
 
     # call the write_parquet_index fn
     if num_worker is None:
-        index_parquet_dataset(pq_dir_url=pq_dir_url, remove_after_indexing=remove_after_indexing)
+        index_parquet_dataset(pq_dir_url=pq_dir_url)
     else:
-        index_parquet_dataset(
-            pq_dir_url=pq_dir_url, remove_after_indexing=remove_after_indexing, num_workers=num_worker
-        )
+        index_parquet_dataset(pq_dir_url=pq_dir_url, num_workers=num_worker)
 
     assert os.path.exists(index_file_path)
 
-    if remove_after_indexing and pq_dir_url.startswith(("gs://", "s3://", "hf://")):
-        time.sleep(1)  # Small delay to ensure deletion completes
-        if pq_dir_url.startswith("hf://"):
-            assert len(os.listdir(cache_dir)) == 1
-        else:
-            assert len(os.listdir(cache_dir)) == 0
+    if pq_dir_url.startswith("hf://"):
+        assert len(os.listdir(cache_dir)) == 1
+    elif pq_dir_url.startswith(("gs://", "s3://")):
+        assert len(os.listdir(cache_dir)) == 0
 
     # Read JSON file into a dictionary
     with open(index_file_path) as f:
         data = json.load(f)
-        print(f"index.json: {data}")
         assert len(data["chunks"]) == 5
         for cnk in data["chunks"]:
             assert cnk["chunk_size"] == 5
         assert data["config"]["item_loader"] == "ParquetLoader"
-        assert data["config"]["data_format"] == ["String", "Float64", "Float64"]
 
     # no test for streaming on s3 and gs
     if pq_dir_url is None or pq_dir_url.startswith("hf://"):
@@ -96,6 +87,20 @@ def test_parquet_index_write(
             assert _ds[2] == pq_data["height"][idx]
 
 
+@pytest.mark.skipif(condition=sys.platform == "win32", reason="Fails on windows and test gets cancelled")
+@pytest.mark.usefixtures("clean_pq_index_cache")
+@patch("litdata.utilities.parquet._HF_HUB_AVAILABLE", True)
+def test_index_hf_dataset(monkeypatch, tmp_path, huggingface_hub_fs_mock):
+    with pytest.raises(ValueError, match="Invalid Hugging Face dataset URL"):
+        index_hf_dataset("invalid_url")
+
+    hf_url = "hf://datasets/some_org/some_repo/some_path"
+    cache_dir = index_hf_dataset(hf_url)
+    assert os.path.exists(cache_dir)
+    assert len(os.listdir(cache_dir)) == 1
+    assert os.path.exists(os.path.join(cache_dir, _INDEX_FILENAME))
+
+
 def test_default_cache_dir(monkeypatch):
     os = ModuleType("os")
     os.path = Mock()
@@ -140,7 +145,7 @@ def test_default_cache_dir(monkeypatch):
         ("meow://some_bucket/somepath", None, pytest.raises(ValueError, match="The provided")),
     ],
 )
-def test_get_parquet_indexer_cls(pq_url, cls, expectation, monkeypatch, fsspec_mock, huggingface_hub_mock):
+def test_get_parquet_indexer_cls(pq_url, cls, expectation, monkeypatch, fsspec_mock, huggingface_hub_fs_mock):
     os = Mock()
     os.listdir = Mock(return_value=[])
 
@@ -150,7 +155,7 @@ def test_get_parquet_indexer_cls(pq_url, cls, expectation, monkeypatch, fsspec_m
 
     hf_fs_mock = Mock()
     hf_fs_mock.ls = Mock(return_value=[])
-    huggingface_hub_mock.HfFileSystem = Mock(return_value=hf_fs_mock)
+    huggingface_hub_fs_mock.HfFileSystem = Mock(return_value=hf_fs_mock)
 
     monkeypatch.setattr("litdata.utilities.parquet.os", os)
     monkeypatch.setattr("litdata.utilities.parquet._HF_HUB_AVAILABLE", True)
@@ -160,44 +165,46 @@ def test_get_parquet_indexer_cls(pq_url, cls, expectation, monkeypatch, fsspec_m
         assert isinstance(indexer_obj, cls)
 
 
-def test_delete_thread(tmp_path):
-    # create some random files in tmp_path dir and test if delete_thread deletes them
-    tmp_files = [f"tmp_file_{i}.txt" for i in range(10)]
-    for f in tmp_files:
-        with open(os.path.join(tmp_path, f), "w") as _:
-            pass
-
-    for f in tmp_files:
-        file_path = tmp_path / f
-        assert os.path.isfile(file_path)
-
-    q = Queue()
-    dt = threading.Thread(target=delete_thread, args=(q,))
-    dt.start()
-
-    for f in tmp_files:
-        file_path = tmp_path / f
-        q.put_nowait(file_path)
-
-    q.put_nowait(None)
-    dt.join()
-
-    # check if all files are deleted
-    for f in tmp_files:
-        file_path = tmp_path / f
-        assert not os.path.exists(file_path)
+@pytest.mark.usefixtures("clean_pq_index_cache")
+@patch("litdata.utilities.parquet._HF_HUB_AVAILABLE", True)
+@patch("litdata.streaming.downloader._HF_HUB_AVAILABLE", True)
+def test_stream_hf_parquet_dataset(monkeypatch, huggingface_hub_fs_mock, pq_data):
+    hf_url = "hf://datasets/some_org/some_repo/some_path"
 
+    # Test case 1: Invalid item_loader
+    with pytest.raises(ValueError, match="Invalid item_loader for hf://datasets."):
+        StreamingDataset(hf_url, item_loader=PyTreeLoader)
 
-@pytest.mark.usefixtures("clean_pq_index_cache")
-def test_stream_hf_parquet_dataset(huggingface_hub_mock, pq_data):
-    hf_url = "hf://some_bucket/some_path"
+    # Test case 2: Streaming without passing item_loader
     ds = StreamingDataset(hf_url)
-
     assert len(ds) == 25  # 5 datasets for 5 loops
+    for i, _ds in enumerate(ds):
+        idx = i % 5
+        assert len(_ds) == 3
+        assert _ds[0] == pq_data["name"][idx]
+        assert _ds[1] == pq_data["weight"][idx]
+        assert _ds[2] == pq_data["height"][idx]
 
+    # Test case 3: Streaming with ParquetLoader as item_loader and low_memory=False
+    ds = StreamingDataset(hf_url, item_loader=ParquetLoader(low_memory=False))
+    assert len(ds) == 25
     for i, _ds in enumerate(ds):
         idx = i % 5
         assert len(_ds) == 3
         assert _ds[0] == pq_data["name"][idx]
         assert _ds[1] == pq_data["weight"][idx]
         assert _ds[2] == pq_data["height"][idx]
+
+    # Test case 4: Streaming with ParquetLoader and low_memory=True
+    ds = StreamingDataset(hf_url, item_loader=ParquetLoader(low_memory=True))
+    assert len(ds) == 25
+    for i, _ds in enumerate(ds):
+        idx = i % 5
+        assert len(_ds) == 3
+        assert _ds[0] == pq_data["name"][idx]
+        assert _ds[1] == pq_data["weight"][idx]
+        assert _ds[2] == pq_data["height"][idx]
+
+    # Test case 5: Streaming with ParquetLoader and low_memory=True and shuffle=True
+    with pytest.raises(ValueError, match="You have enabled shuffling when using low memory with ParquetLoader."):
+        StreamingDataset(hf_url, item_loader=ParquetLoader(low_memory=True), shuffle=True)