Repository: Lightning-AI/pytorch-lightning
Pull request: Enforce eval shuffle warning only for default samplers
## What does this PR do?

<!--
Please include a summary of the change and which issue is fixed.
Please also include relevant motivation and context.
List any dependencies that are required for this change.

If we didn't discuss your PR in Github issues there's a high chance it will not be merged.

The following links the related issue to the PR (https://docs.github.com/en/free-pro-team@latest/github/managing-your-work-on-github/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword)
-->

Fixes #12648
Fixes #11856

For reviewers: https://github.com/PyTorchLightning/pytorch-lightning/commit/a179fdce4117ad39af5f8cd0edc809c8d39198cc is the only commit that needs review. Rest is just moving tests from test_data_loading to test_data_connector since we have deprecated data_loading. Moved them since it becomes hard to find tests that belong to data_connector.


### Does your PR introduce any breaking changes? If yes, please list them.

<!-- FILL IN or None -->

## Before submitting

- [x] Was this **discussed/approved** via a GitHub issue? (not for typos and docs)
- [x] Did you read the [contributor guideline](https://github.com/PyTorchLightning/pytorch-lightning/blob/master/.github/CONTRIBUTING.md), **Pull Request** section?
- [x] Did you make sure your **PR does only one thing**, instead of bundling different changes together?
- [ ] Did you make sure to **update the documentation** with your changes? (if necessary)
- [x] Did you write any **new necessary tests**? (not for typos and docs)
- [x] Did you verify new and **existing tests pass** locally with your changes?
- [ ] Did you list all the **breaking changes** introduced by this pull request?
- [x] Did you **update the [CHANGELOG](https://github.com/PyTorchLightning/pytorch-lightning/blob/master/CHANGELOG.md)**? (not for typos, docs, test updates, or internal minor changes/refactorings)

<!-- In the CHANGELOG, separate each item in the unreleased section by a blank line to reduce collisions -->

## PR review

Anyone in the community is welcome to review the PR.
Before you start reviewing make sure you have read [Review guidelines](https://github.com/PyTorchLightning/pytorch-lightning/wiki/Review-guidelines). In short, see the following bullet-list:

- [x] Is this pull request ready for review? (if not, please submit in draft mode)
- [x] Check that all items from **Before submitting** are resolved
- [x] Make sure the title is self-explanatory and the description concisely explains the PR
- [x] Add labels and milestones (and optionally projects) to the PR so it can be classified

## Did you have fun?

Make sure you had fun coding 🙃


cc @borda @justusschock @awaelchli @ninginthecloud @rohitgr7

Problem statement:
Even when val DataLoader has shuffle=False, Lightning gives an incorrect warning that val Dataloader has shuffle=True
## 🐛 Bug
Even when DataLoader(self.valid_data, shuffle=False, ...), Lightning gives the following message:   UserWarning: Your `val_dataloader` has `shuffle=True`,.......

### To Reproduce
Assign shuffle=False in both the val and test dataloaders. Run the code and see the following two warning messages:

> .../.local/lib/python3.8/site-packages/pytorch_lightning/trainer/data_loading.py:659: UserWarning: Your `val_dataloader` has `shuffle=True`, it is strongly recommended that you turn this off for val/test/predict dataloaders.

> .local/lib/python3.8/site-packages/pytorch_lightning/trainer/data_loading.py:659: UserWarning: Your `test_dataloader` has `shuffle=True`, it is strongly recommended that you turn this off for val/test/predict dataloaders.

### Expected behavior
There should be no warning messages.

### Environment
- PyTorch Lightning Version (e.g., 1.5.0): 1.5.9
- PyTorch Version (e.g., 1.10): 1.10.2
- Python version (e.g., 3.9): 3.8.10
- OS (e.g., Linux): Linux (Ubuntu)
- CUDA/cuDNN version: cu113
- GPU models and configuration: GTX 1080
- How you installed PyTorch (`conda`, `pip`, source): pip

### Additional context
Previous versions of Lightning also gave those warning messages

cc @justusschock @awaelchli @ninginthecloud @tchaton @rohitgr7 @akihironitta

```_check_eval_shuffling``` issues warning when sampler is a NumPy array
## 🐛 Bug

The function ```pytorch_lightning.trainer.data_loading._check_eval_shuffling``` issues a warning if the sampler of a Map-style Dataset is defined and is not a SequentialSampler.
The warning however, says that the issue is that the val_dataset has ```shuffle=True```, which is not necessarily true.
For instance, if you use a simple NumPy array as sampler, the warning is issued but the data loader is completely deterministic.

### To Reproduce

```python
import os

import torch
from torch.utils.data import DataLoader, Dataset

from pytorch_lightning import LightningModule, Trainer


class RandomDataset(Dataset):
    def __init__(self, size, length):
        self.len = length
        self.data = torch.randn(length, size)

    def __getitem__(self, index):
        return self.data[index]

    def __len__(self):
        return self.len


class BoringModel(LightningModule):
    def __init__(self):
        super().__init__()
        self.layer = torch.nn.Linear(32, 2)

    def forward(self, x):
        return self.layer(x)

    def training_step(self, batch, batch_idx):
        loss = self(batch).sum()
        self.log("train_loss", loss)
        return {"loss": loss}

    def validation_step(self, batch, batch_idx):
        loss = self(batch).sum()
        self.log("valid_loss", loss)

    def test_step(self, batch, batch_idx):
        loss = self(batch).sum()
        self.log("test_loss", loss)

    def configure_optimizers(self):
        return torch.optim.SGD(self.layer.parameters(), lr=0.1)


def run():
    train_data = DataLoader(RandomDataset(32, 64), batch_size=2)
    val_data = DataLoader(RandomDataset(32, 64), batch_size=2, sampler=list(range(64)))
    test_data = DataLoader(RandomDataset(32, 64), batch_size=2)

    model = BoringModel()
    trainer = Trainer(
        default_root_dir=os.getcwd(),
        limit_train_batches=1,
        limit_val_batches=1,
        limit_test_batches=1,
        num_sanity_val_steps=0,
        max_epochs=1,
        enable_model_summary=False,
    )
    trainer.fit(model, train_dataloaders=train_data, val_dataloaders=val_data)
    trainer.test(model, dataloaders=test_data)


if __name__ == "__main__":
    run()
```

### Expected behavior

I see three possible solutions:
* The warning is not issued if the sampler is an Iterable whose order is fixed.
* The warning is clearer in stating what is triggering it.
* The warning specifies it can be ignored if you are using a list/array as sampler.

### Environment

- PyTorch Lightning Version: 1.5.9
- PyTorch Version: 1.10.2
- Python version: 3.8.3
- OS: MacOS
- How you installed PyTorch: poetry

cc @justusschock @awaelchli @ninginthecloud @rohitgr7 @carmocca @borda @ananthsub @jjenniferdai

Hints from issue discussion:
Can you reproduce it by modifying the following script?

https://github.com/PyTorchLightning/pytorch-lightning/blob/master/pl_examples/bug_report/bug_report_model.py
.local/lib/python3.8/site-packages/pytorch_lightning/trainer/data_loading.py:659 has a code as follows:

>     def _check_eval_shuffling(dataloader, mode):
        if (
            hasattr(dataloader, "sampler")
            and not isinstance(dataloader.sampler, SequentialSampler)
            and not isinstance(dataloader.dataset, IterableDataset)
        ):
            rank_zero_warn(
                f"Your `{mode.dataloader_prefix}_dataloader` has `shuffle=True`,"
                " it is strongly recommended that you turn this off for val/test/predict dataloaders.",
                category=UserWarning,
            )
**The above code does NOT check if shuffle == True, before giving a warning. I think that within the above "if statement" there should be a check for shuffle == True**

Following is my val dataloader code where shuffle=False but Lightning still gives a warning:
>     def val_dataloader(self) -> DataLoader:
        return DataLoader(self.valid_data,
                          batch_size=self.d_params['batch_size']['val'],
                          shuffle=False,
                          sampler=RandomSampler(self.valid_data),
                          batch_sampler=None,
                          num_workers=6,
                          collate_fn=self.gpt2_collater,
                          pin_memory=True,
                          drop_last=False,
                          timeout=0)
 

Just curious, why is shuffling recommended against for validation/testing anyway?
> Just curious, why is shuffling recommended against for validation/testing anyway?

The order in which you process the samples in validation and testing should not matter. Metrics are typically reported on the whole dataset, not per sample. As a result, shuffling the samples here is usually inefficient. 
> The order in which you process the samples in validation and testing should not matter.

This is precisely the reason that I want to shuffle. Because order doesn't matter, having a sampler that groups examples with similar lengths together would speed up validation/testing. Granted, sampler != shuffling, but that's the issue being raised here.
@ZhaofengWu to override this behavior today, you should be able to do the following: 

```py
from torch.utils.data.sampler import RandomSampler
dataloader = DataLoader(..., sampler=RandomSampler())
```

On the issue itself, I don't have a preference for Lightning but for personal use cases I would always set `shuffle=False` myself no matter what Lightning does. I usually want to log a few example predictions (e.g. images) during every validation run so I can compare them on a timeline in the logger and see visually the progression of quality in the predicted pixels. I wouldn't be able to compare them so easily from epoch to epoch if each time the sampling changes. I don't think this is an uncommon practice. 

I can also see the opposite argument though. Someone might want to disable shuffling in the validation loop because they are actually training a model there too (yes on the validation set, nothing wrong with that). 

Finally, IMO the torch DataLoader should have `shuffle=None` by default. Why? Because its value should be determined based on whether a sampler is given or not (my exact example above). If pytorch had that, our Trainer could know when the user explicitly sets `shuffle = False`, and only forces it to False if it previously was unset. 
@awaelchli I actually use huggingface's `LengthGroupedSampler` to group together examples with similar lengths, again to speed up validation/testing, so I can't use a `RandomSampler`. Although having a warning message is not that disruptive anyway.

And on the issue itself, I definitely agree that having a fixed order is nice for analysis, etc.  But when not doing analysis, when the validation set size is large, and/or when validation needs to be done frequently, having a faster validation run becomes important. So, personally, I think it'd be best if lightning does not recommend any particular choice over the other here.
I also have this issue but it only appears when I use a sampler, in my case a weighted sampler. If I do not use any sampling then no warning is produced.
just to clarify, this warning has false positives too since we could only check whether shuffled is enabled or not using the default samplers that PT adds when shuffle is set to True/False. But if a custom sampler is added, it is difficult to check this and thus this leads to false-positive warning here. I don't think this can be fixed, so I'd suggest considering it as a recommendation, and if you already have what you need, you can ignore it. improving the warning a little here: #12197
Hi, @LucaButera - Thank you for reporting this issue! I think it's a valid ask, the warning message should definitely be improved. However, I'm more inclined towards option 1 (warning not issued in this case). cc: @carmocca @awaelchli @Borda - What do you guys think?
it's WIP...  :)
just feeling lazy while adding test :p

Commit message: Merge branch 'master' into ref/eval_warning

Patch to review:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 4037dd3632379..6995bba82e9ad 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -91,6 +91,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 - Fixed an issue where incorrect type warnings appear when the overridden `LightningLite.run` method accepts user-defined arguments ([#12629](https://github.com/PyTorchLightning/pytorch-lightning/pull/12629))
 
 
+- Enforced eval shuffle warning only for default samplers in DataLoader ([#12653](https://github.com/PyTorchLightning/pytorch-lightning/pull/12653))
+
+
 - Fixed `rank_zero_only` decorator in LSF environments ([#12587](https://github.com/PyTorchLightning/pytorch-lightning/pull/12587))
 
 
diff --git a/pytorch_lightning/trainer/connectors/data_connector.py b/pytorch_lightning/trainer/connectors/data_connector.py
index f1943073da819..4d588cfeda39e 100644
--- a/pytorch_lightning/trainer/connectors/data_connector.py
+++ b/pytorch_lightning/trainer/connectors/data_connector.py
@@ -460,7 +460,12 @@ def replace_sampler(dataloader: DataLoader) -> DataLoader:
 
     @staticmethod
     def _check_eval_shuffling(dataloader, mode):
-        if _is_dataloader_shuffled(dataloader):
+        # limit this warning only for samplers assigned automatically when shuffle is set
+        if (
+            hasattr(dataloader, "sampler")
+            and isinstance(dataloader.sampler, (RandomSampler, SequentialSampler))
+            and _is_dataloader_shuffled(dataloader)
+        ):
             rank_zero_warn(
                 f"Your `{mode.dataloader_prefix}_dataloader`'s sampler has shuffling enabled,"
                 " it is strongly recommended that you turn shuffling off for val/test/predict dataloaders.",
diff --git a/tests/helpers/utils.py b/tests/helpers/utils.py
index 05dc566949a08..2296cad689178 100644
--- a/tests/helpers/utils.py
+++ b/tests/helpers/utils.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 import functools
 import os
+import re
 import traceback
 from contextlib import contextmanager
 from typing import Optional, Type
@@ -126,7 +127,7 @@ def no_warning_call(expected_warning: Type[Warning] = UserWarning, match: Option
             return
     else:
         for w in record.list:
-            if w.category is expected_warning and match in w.message.args[0]:
+            if w.category is expected_warning and re.compile(match).search(str(w.message.args[0])):
                 break
         else:
             return
diff --git a/tests/trainer/connectors/test_data_connector.py b/tests/trainer/connectors/test_data_connector.py
index e22e846600122..cea535838cb77 100644
--- a/tests/trainer/connectors/test_data_connector.py
+++ b/tests/trainer/connectors/test_data_connector.py
@@ -11,20 +11,388 @@
 # 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.
+from contextlib import redirect_stderr
+from io import StringIO
+from re import escape
 from unittest.mock import Mock
 
 import pytest
-from torch.utils.data import DataLoader
+from torch.utils.data import BatchSampler, DataLoader, DistributedSampler, Sampler, SequentialSampler
 
 from pytorch_lightning import Trainer
+from pytorch_lightning.strategies import DDPSpawnStrategy
 from pytorch_lightning.trainer.connectors.data_connector import _DataHookSelector, _DataLoaderSource, warning_cache
-from pytorch_lightning.trainer.states import TrainerFn
+from pytorch_lightning.trainer.states import RunningStage, TrainerFn
+from pytorch_lightning.trainer.supporters import CombinedLoader
+from pytorch_lightning.utilities.data import _update_dataloader
+from pytorch_lightning.utilities.exceptions import MisconfigurationException
 from pytorch_lightning.utilities.warnings import PossibleUserWarning
-from tests.helpers import BoringDataModule, BoringModel
-from tests.helpers.boring_model import RandomDataset
+from tests.helpers.boring_model import BoringDataModule, BoringModel, RandomDataset
+from tests.helpers.runif import RunIf
 from tests.helpers.utils import no_warning_call
 
 
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("mode", (1, 2))
+def test_replace_distributed_sampler(tmpdir, mode):
+    class IndexedRandomDataset(RandomDataset):
+        def __getitem__(self, index):
+            return self.data[index]
+
+    class CustomDataLoader(DataLoader):
+        def __init__(self, num_features, dataset, *args, **kwargs):
+            # argument `num_features` unused on purpose
+            # it gets automatically captured by _replace_dataloader_init_method()
+            super().__init__(dataset, *args, **kwargs)
+
+    class CustomBatchSampler(BatchSampler):
+        pass
+
+    class TestModel(BoringModel):
+        def __init__(self, numbers_test_dataloaders, mode):
+            super().__init__()
+            self._numbers_test_dataloaders = numbers_test_dataloaders
+            self._mode = mode
+
+        def test_step(self, batch, batch_idx, dataloader_idx=0):
+            return super().test_step(batch, batch_idx)
+
+        def on_test_start(self) -> None:
+            dataloader = self.trainer.test_dataloaders[0]
+            assert isinstance(dataloader, CustomDataLoader)
+            batch_sampler = dataloader.batch_sampler
+            if self._mode == 1:
+                assert isinstance(batch_sampler, CustomBatchSampler)
+                # the batch_size is set on the batch sampler
+                assert dataloader.batch_size is None
+            elif self._mode == 2:
+                assert type(batch_sampler) is BatchSampler
+                assert dataloader.batch_size == self._mode
+            assert batch_sampler.batch_size == self._mode
+            assert batch_sampler.drop_last
+            # the sampler has been replaced
+            assert isinstance(batch_sampler.sampler, DistributedSampler)
+
+        def create_dataset(self):
+            dataset = IndexedRandomDataset(32, 64)
+            if self._mode == 1:
+                # with a custom batch sampler
+                batch_sampler = CustomBatchSampler(SequentialSampler(dataset), batch_size=1, drop_last=True)
+                return CustomDataLoader(32, dataset, batch_sampler=batch_sampler)
+            elif self._mode == 2:
+                # with no batch sampler provided
+                return CustomDataLoader(32, dataset, batch_size=2, drop_last=True)
+
+        def test_dataloader(self):
+            return [self.create_dataset()] * self._numbers_test_dataloaders
+
+    model = TestModel(2, mode)
+    model.test_epoch_end = None
+
+    trainer = Trainer(
+        default_root_dir=tmpdir,
+        limit_test_batches=2,
+        accelerator="cpu",
+        devices=1,
+        strategy="ddp_find_unused_parameters_false",
+    )
+    trainer.test(model)
+
+
+class TestSpawnBoringModel(BoringModel):
+    def __init__(self, num_workers):
+        super().__init__()
+        self.num_workers = num_workers
+
+    def train_dataloader(self):
+        return DataLoader(RandomDataset(32, 64), num_workers=self.num_workers)
+
+    def on_fit_start(self):
+        self._resout = StringIO()
+        self.ctx = redirect_stderr(self._resout)
+        self.ctx.__enter__()
+
+    def on_train_end(self):
+        def _get_warning_msg():
+            dl = self.trainer.train_dataloader.loaders
+            if hasattr(dl, "persistent_workers"):
+                if self.num_workers == 0:
+                    warn_str = "Consider setting num_workers>0 and persistent_workers=True"
+                else:
+                    warn_str = "Consider setting persistent_workers=True"
+            else:
+                warn_str = "Consider setting strategy=ddp"
+
+            return warn_str
+
+        if self.trainer.is_global_zero:
+            self.ctx.__exit__(None, None, None)
+            msg = self._resout.getvalue()
+            warn_str = _get_warning_msg()
+            assert warn_str in msg
+
+
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("num_workers", [0, 1])
+def test_dataloader_warnings(tmpdir, num_workers):
+    trainer = Trainer(default_root_dir=tmpdir, accelerator="cpu", devices=2, strategy="ddp_spawn", fast_dev_run=4)
+    assert isinstance(trainer.strategy, DDPSpawnStrategy)
+    trainer.fit(TestSpawnBoringModel(num_workers))
+
+
+def test_update_dataloader_raises():
+    with pytest.raises(ValueError, match="needs to subclass `torch.utils.data.DataLoader"):
+        _update_dataloader(object(), object(), mode="fit")
+
+
+def test_dataloaders_with_missing_keyword_arguments():
+    ds = RandomDataset(10, 20)
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset, *args, **kwargs):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, *foo, **bar):
+            super().__init__(*foo, **bar)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, *args, shuffle=False):
+            self.num_feat = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, **kwargs):
+            self.feat_num = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+
+def test_update_dataloader_with_multiprocessing_context():
+    """This test verifies that replace_sampler conserves multiprocessing context."""
+    train = RandomDataset(32, 64)
+    context = "spawn"
+    train = DataLoader(train, batch_size=32, num_workers=2, multiprocessing_context=context, shuffle=True)
+    new_data_loader = _update_dataloader(train, SequentialSampler(train.dataset))
+    assert new_data_loader.multiprocessing_context == train.multiprocessing_context
+
+
+def test_dataloader_reinit_for_subclass():
+    class CustomDataLoader(DataLoader):
+        def __init__(
+            self,
+            dataset,
+            batch_size=1,
+            shuffle=False,
+            sampler=None,
+            batch_sampler=None,
+            num_workers=0,
+            collate_fn=None,
+            pin_memory=False,
+            drop_last=False,
+            timeout=0,
+            worker_init_fn=None,
+            dummy_kwarg=None,
+        ):
+            super().__init__(
+                dataset,
+                batch_size,
+                shuffle,
+                sampler,
+                batch_sampler,
+                num_workers,
+                collate_fn,
+                pin_memory,
+                drop_last,
+                timeout,
+                worker_init_fn,
+            )
+            self.dummy_kwarg = dummy_kwarg
+            self.something_unrelated = 1
+
+    trainer = Trainer(accelerator="cpu", devices=2, strategy="ddp_spawn")
+
+    class CustomDummyObj:
+        sampler = None
+
+    result = trainer._data_connector._prepare_dataloader(CustomDummyObj(), shuffle=True)
+    assert isinstance(result, CustomDummyObj), "Wrongly reinstantiated data loader"
+
+    dataset = list(range(10))
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    # Shuffled DataLoader should also work
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset, shuffle=True), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    class CustomSampler(Sampler):
+        pass
+
+    # Should raise an error if existing sampler is being replaced
+    dataloader = CustomDataLoader(dataset, sampler=CustomSampler(dataset))
+    with pytest.raises(MisconfigurationException, match="will be replaced by `DistributedSampler`"):
+        trainer._data_connector._prepare_dataloader(dataloader, shuffle=True)
+
+
+class LoaderTestModel(BoringModel):
+    def training_step(self, batch, batch_idx):
+        assert len(self.trainer.train_dataloader.loaders) == 10
+        return super().training_step(batch, batch_idx)
+
+    def validation_step(self, batch, batch_idx):
+        assert len(self.trainer.val_dataloaders[0]) == 10
+        return super().validation_step(batch, batch_idx)
+
+    def test_step(self, batch, batch_idx):
+        assert len(self.trainer.test_dataloaders[0]) == 10
+        return super().test_step(batch, batch_idx)
+
+    def predict_step(self, batch, batch_idx, dataloader_idx=0):
+        assert len(self.trainer.predict_dataloaders[0]) == 10
+        return super().predict_step(batch, batch_idx, dataloader_idx=dataloader_idx)
+
+
+def test_loader_detaching():
+    """Checks that the loader has been reset after the entrypoint."""
+
+    loader = DataLoader(RandomDataset(32, 10), batch_size=1)
+
+    model = LoaderTestModel()
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer = Trainer(fast_dev_run=1)
+    trainer.fit(model, loader, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.validate(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.predict(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.test(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+
+def test_pre_made_batches():
+    """Check that loader works with pre-made batches."""
+    loader = DataLoader(RandomDataset(32, 10), batch_size=None)
+    trainer = Trainer(fast_dev_run=1)
+    trainer.predict(LoaderTestModel(), loader)
+
+
+def test_error_raised_with_float_limited_eval_batches():
+    """Test that an error is raised if there are not enough batches when passed with float value of
+    limit_eval_batches."""
+    model = BoringModel()
+    dl_size = len(model.val_dataloader())
+    limit_val_batches = 1 / (dl_size + 2)
+    trainer = Trainer(limit_val_batches=limit_val_batches)
+    trainer._data_connector.attach_data(model)
+    with pytest.raises(
+        MisconfigurationException,
+        match=rf"{limit_val_batches} \* {dl_size} < 1. Please increase the `limit_val_batches`",
+    ):
+        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)
+
+
+@pytest.mark.parametrize(
+    "val_dl,raises_exception",
+    [
+        (DataLoader(dataset=RandomDataset(32, 64), shuffle=True), True),
+        (DataLoader(dataset=RandomDataset(32, 64), sampler=list(range(64))), False),
+        (CombinedLoader(DataLoader(dataset=RandomDataset(32, 64), shuffle=True)), True),
+        (
+            CombinedLoader(
+                [DataLoader(dataset=RandomDataset(32, 64)), DataLoader(dataset=RandomDataset(32, 64), shuffle=True)]
+            ),
+            True,
+        ),
+        (
+            CombinedLoader(
+                {
+                    "dl1": DataLoader(dataset=RandomDataset(32, 64)),
+                    "dl2": DataLoader(dataset=RandomDataset(32, 64), shuffle=True),
+                }
+            ),
+            True,
+        ),
+    ],
+)
+def test_non_sequential_sampler_warning_is_raised_for_eval_dataloader(val_dl, raises_exception):
+    trainer = Trainer()
+    model = BoringModel()
+    trainer._data_connector.attach_data(model, val_dataloaders=val_dl)
+    context = pytest.warns if raises_exception else no_warning_call
+    with context(PossibleUserWarning, match="recommended .* turn shuffling off for val/test/predict"):
+        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)
+
+
 class NoDataLoaderModel(BoringModel):
     def __init__(self):
         super().__init__()
diff --git a/tests/trainer/test_data_loading.py b/tests/trainer/test_data_loading.py
deleted file mode 100644
index 4455cd89e2104..0000000000000
--- a/tests/trainer/test_data_loading.py
+++ /dev/null
@@ -1,382 +0,0 @@
-# Copyright The PyTorch Lightning team.
-#
-# Licensed under the Apache License, Version 2.0 (the "License");
-# you may not use this file except in compliance with the License.
-# You may obtain a copy of the License at
-#
-#     http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# 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.
-from contextlib import redirect_stderr
-from io import StringIO
-from re import escape
-
-import pytest
-from torch.utils.data import BatchSampler, DataLoader, DistributedSampler, Sampler, SequentialSampler
-
-from pytorch_lightning import Trainer
-from pytorch_lightning.strategies import DDPSpawnStrategy
-from pytorch_lightning.trainer.states import RunningStage
-from pytorch_lightning.trainer.supporters import CombinedLoader
-from pytorch_lightning.utilities.data import _update_dataloader
-from pytorch_lightning.utilities.exceptions import MisconfigurationException
-from pytorch_lightning.utilities.warnings import PossibleUserWarning
-from tests.helpers import BoringModel, RandomDataset
-from tests.helpers.runif import RunIf
-
-
-@RunIf(skip_windows=True)
-@pytest.mark.parametrize("mode", (1, 2))
-def test_replace_distributed_sampler(tmpdir, mode):
-    class IndexedRandomDataset(RandomDataset):
-        def __getitem__(self, index):
-            return self.data[index]
-
-    class CustomDataLoader(DataLoader):
-        def __init__(self, num_features, dataset, *args, **kwargs):
-            # argument `num_features` unused on purpose
-            # it gets automatically captured by _replace_dataloader_init_method()
-            super().__init__(dataset, *args, **kwargs)
-
-    class CustomBatchSampler(BatchSampler):
-        pass
-
-    class TestModel(BoringModel):
-        def __init__(self, numbers_test_dataloaders, mode):
-            super().__init__()
-            self._numbers_test_dataloaders = numbers_test_dataloaders
-            self._mode = mode
-
-        def test_step(self, batch, batch_idx, dataloader_idx=0):
-            return super().test_step(batch, batch_idx)
-
-        def on_test_start(self) -> None:
-            dataloader = self.trainer.test_dataloaders[0]
-            assert isinstance(dataloader, CustomDataLoader)
-            batch_sampler = dataloader.batch_sampler
-            if self._mode == 1:
-                assert isinstance(batch_sampler, CustomBatchSampler)
-                # the batch_size is set on the batch sampler
-                assert dataloader.batch_size is None
-            elif self._mode == 2:
-                assert type(batch_sampler) is BatchSampler
-                assert dataloader.batch_size == self._mode
-            assert batch_sampler.batch_size == self._mode
-            assert batch_sampler.drop_last
-            # the sampler has been replaced
-            assert isinstance(batch_sampler.sampler, DistributedSampler)
-
-        def create_dataset(self):
-            dataset = IndexedRandomDataset(32, 64)
-            if self._mode == 1:
-                # with a custom batch sampler
-                batch_sampler = CustomBatchSampler(SequentialSampler(dataset), batch_size=1, drop_last=True)
-                return CustomDataLoader(32, dataset, batch_sampler=batch_sampler)
-            elif self._mode == 2:
-                # with no batch sampler provided
-                return CustomDataLoader(32, dataset, batch_size=2, drop_last=True)
-
-        def test_dataloader(self):
-            return [self.create_dataset()] * self._numbers_test_dataloaders
-
-    model = TestModel(2, mode)
-    model.test_epoch_end = None
-
-    trainer = Trainer(
-        default_root_dir=tmpdir,
-        limit_test_batches=2,
-        accelerator="cpu",
-        devices=1,
-        strategy="ddp_find_unused_parameters_false",
-    )
-    trainer.test(model)
-
-
-class TestSpawnBoringModel(BoringModel):
-    def __init__(self, num_workers):
-        super().__init__()
-        self.num_workers = num_workers
-
-    def train_dataloader(self):
-        return DataLoader(RandomDataset(32, 64), num_workers=self.num_workers)
-
-    def on_fit_start(self):
-        self._resout = StringIO()
-        self.ctx = redirect_stderr(self._resout)
-        self.ctx.__enter__()
-
-    def on_train_end(self):
-        def _get_warning_msg():
-            dl = self.trainer.train_dataloader.loaders
-            if hasattr(dl, "persistent_workers"):
-                if self.num_workers == 0:
-                    warn_str = "Consider setting num_workers>0 and persistent_workers=True"
-                else:
-                    warn_str = "Consider setting persistent_workers=True"
-            else:
-                warn_str = "Consider setting strategy=ddp"
-
-            return warn_str
-
-        if self.trainer.is_global_zero:
-            self.ctx.__exit__(None, None, None)
-            msg = self._resout.getvalue()
-            warn_str = _get_warning_msg()
-            assert warn_str in msg
-
-
-@RunIf(skip_windows=True)
-@pytest.mark.parametrize("num_workers", [0, 1])
-def test_dataloader_warnings(tmpdir, num_workers):
-    trainer = Trainer(default_root_dir=tmpdir, accelerator="cpu", devices=2, strategy="ddp_spawn", fast_dev_run=4)
-    assert isinstance(trainer.strategy, DDPSpawnStrategy)
-    trainer.fit(TestSpawnBoringModel(num_workers))
-
-
-def test_update_dataloader_raises():
-    with pytest.raises(ValueError, match="needs to subclass `torch.utils.data.DataLoader"):
-        _update_dataloader(object(), object(), mode="fit")
-
-
-def test_dataloaders_with_missing_keyword_arguments():
-    ds = RandomDataset(10, 20)
-
-    class TestDataLoader(DataLoader):
-        def __init__(self, dataset):
-            super().__init__(dataset)
-
-    loader = TestDataLoader(ds)
-    sampler = SequentialSampler(ds)
-    match = escape("missing arguments are ['batch_sampler', 'sampler', 'shuffle']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="fit")
-    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler', 'shuffle']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="predict")
-
-    class TestDataLoader(DataLoader):
-        def __init__(self, dataset, *args, **kwargs):
-            super().__init__(dataset)
-
-    loader = TestDataLoader(ds)
-    sampler = SequentialSampler(ds)
-    _update_dataloader(loader, sampler, mode="fit")
-    _update_dataloader(loader, sampler, mode="predict")
-
-    class TestDataLoader(DataLoader):
-        def __init__(self, *foo, **bar):
-            super().__init__(*foo, **bar)
-
-    loader = TestDataLoader(ds)
-    sampler = SequentialSampler(ds)
-    _update_dataloader(loader, sampler, mode="fit")
-    _update_dataloader(loader, sampler, mode="predict")
-
-    class TestDataLoader(DataLoader):
-        def __init__(self, num_feat, dataset, *args, shuffle=False):
-            self.num_feat = num_feat
-            super().__init__(dataset)
-
-    loader = TestDataLoader(1, ds)
-    sampler = SequentialSampler(ds)
-    match = escape("missing arguments are ['batch_sampler', 'sampler']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="fit")
-    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="predict")
-
-    class TestDataLoader(DataLoader):
-        def __init__(self, num_feat, dataset, **kwargs):
-            self.feat_num = num_feat
-            super().__init__(dataset)
-
-    loader = TestDataLoader(1, ds)
-    sampler = SequentialSampler(ds)
-    match = escape("missing attributes are ['num_feat']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="fit")
-    match = escape("missing attributes are ['num_feat']")
-    with pytest.raises(MisconfigurationException, match=match):
-        _update_dataloader(loader, sampler, mode="predict")
-
-
-def test_update_dataloader_with_multiprocessing_context():
-    """This test verifies that replace_sampler conserves multiprocessing context."""
-    train = RandomDataset(32, 64)
-    context = "spawn"
-    train = DataLoader(train, batch_size=32, num_workers=2, multiprocessing_context=context, shuffle=True)
-    new_data_loader = _update_dataloader(train, SequentialSampler(train.dataset))
-    assert new_data_loader.multiprocessing_context == train.multiprocessing_context
-
-
-def test_dataloader_reinit_for_subclass():
-    class CustomDataLoader(DataLoader):
-        def __init__(
-            self,
-            dataset,
-            batch_size=1,
-            shuffle=False,
-            sampler=None,
-            batch_sampler=None,
-            num_workers=0,
-            collate_fn=None,
-            pin_memory=False,
-            drop_last=False,
-            timeout=0,
-            worker_init_fn=None,
-            dummy_kwarg=None,
-        ):
-            super().__init__(
-                dataset,
-                batch_size,
-                shuffle,
-                sampler,
-                batch_sampler,
-                num_workers,
-                collate_fn,
-                pin_memory,
-                drop_last,
-                timeout,
-                worker_init_fn,
-            )
-            self.dummy_kwarg = dummy_kwarg
-            self.something_unrelated = 1
-
-    trainer = Trainer(accelerator="cpu", devices=2, strategy="ddp_spawn")
-
-    class CustomDummyObj:
-        sampler = None
-
-    result = trainer._data_connector._prepare_dataloader(CustomDummyObj(), shuffle=True)
-    assert isinstance(result, CustomDummyObj), "Wrongly reinstantiated data loader"
-
-    dataset = list(range(10))
-    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset), shuffle=True)
-    assert isinstance(result, DataLoader)
-    assert isinstance(result, CustomDataLoader)
-    assert result.dummy_kwarg is None
-
-    # Shuffled DataLoader should also work
-    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset, shuffle=True), shuffle=True)
-    assert isinstance(result, DataLoader)
-    assert isinstance(result, CustomDataLoader)
-    assert result.dummy_kwarg is None
-
-    class CustomSampler(Sampler):
-        pass
-
-    # Should raise an error if existing sampler is being replaced
-    dataloader = CustomDataLoader(dataset, sampler=CustomSampler(dataset))
-    with pytest.raises(MisconfigurationException, match="will be replaced by `DistributedSampler`"):
-        trainer._data_connector._prepare_dataloader(dataloader, shuffle=True)
-
-
-class LoaderTestModel(BoringModel):
-    def training_step(self, batch, batch_idx):
-        assert len(self.trainer.train_dataloader.loaders) == 10
-        return super().training_step(batch, batch_idx)
-
-    def validation_step(self, batch, batch_idx):
-        assert len(self.trainer.val_dataloaders[0]) == 10
-        return super().validation_step(batch, batch_idx)
-
-    def test_step(self, batch, batch_idx):
-        assert len(self.trainer.test_dataloaders[0]) == 10
-        return super().test_step(batch, batch_idx)
-
-    def predict_step(self, batch, batch_idx, dataloader_idx=0):
-        assert len(self.trainer.predict_dataloaders[0]) == 10
-        return super().predict_step(batch, batch_idx, dataloader_idx=dataloader_idx)
-
-
-def test_loader_detaching():
-    """Checks that the loader has been reset after the entrypoint."""
-
-    loader = DataLoader(RandomDataset(32, 10), batch_size=1)
-
-    model = LoaderTestModel()
-
-    assert len(model.train_dataloader()) == 64
-    assert len(model.val_dataloader()) == 64
-    assert len(model.predict_dataloader()) == 64
-    assert len(model.test_dataloader()) == 64
-
-    trainer = Trainer(fast_dev_run=1)
-    trainer.fit(model, loader, loader)
-
-    assert len(model.train_dataloader()) == 64
-    assert len(model.val_dataloader()) == 64
-    assert len(model.predict_dataloader()) == 64
-    assert len(model.test_dataloader()) == 64
-
-    trainer.validate(model, loader)
-
-    assert len(model.train_dataloader()) == 64
-    assert len(model.val_dataloader()) == 64
-    assert len(model.predict_dataloader()) == 64
-    assert len(model.test_dataloader()) == 64
-
-    trainer.predict(model, loader)
-
-    assert len(model.train_dataloader()) == 64
-    assert len(model.val_dataloader()) == 64
-    assert len(model.predict_dataloader()) == 64
-    assert len(model.test_dataloader()) == 64
-
-    trainer.test(model, loader)
-
-    assert len(model.train_dataloader()) == 64
-    assert len(model.val_dataloader()) == 64
-    assert len(model.predict_dataloader()) == 64
-    assert len(model.test_dataloader()) == 64
-
-
-def test_pre_made_batches():
-    """Check that loader works with pre-made batches."""
-    loader = DataLoader(RandomDataset(32, 10), batch_size=None)
-    trainer = Trainer(fast_dev_run=1)
-    trainer.predict(LoaderTestModel(), loader)
-
-
-def test_error_raised_with_float_limited_eval_batches():
-    """Test that an error is raised if there are not enough batches when passed with float value of
-    limit_eval_batches."""
-    model = BoringModel()
-    dl_size = len(model.val_dataloader())
-    limit_val_batches = 1 / (dl_size + 2)
-    trainer = Trainer(limit_val_batches=limit_val_batches)
-    trainer._data_connector.attach_data(model)
-    with pytest.raises(
-        MisconfigurationException,
-        match=rf"{limit_val_batches} \* {dl_size} < 1. Please increase the `limit_val_batches`",
-    ):
-        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)
-
-
-@pytest.mark.parametrize(
-    "val_dl",
-    [
-        DataLoader(dataset=RandomDataset(32, 64), shuffle=True),
-        CombinedLoader(DataLoader(dataset=RandomDataset(32, 64), shuffle=True)),
-        CombinedLoader(
-            [DataLoader(dataset=RandomDataset(32, 64)), DataLoader(dataset=RandomDataset(32, 64), shuffle=True)]
-        ),
-        CombinedLoader(
-            {
-                "dl1": DataLoader(dataset=RandomDataset(32, 64)),
-                "dl2": DataLoader(dataset=RandomDataset(32, 64), shuffle=True),
-            }
-        ),
-    ],
-)
-def test_non_sequential_sampler_warning_is_raised_for_eval_dataloader(val_dl):
-    trainer = Trainer()
-    model = BoringModel()
-    trainer._data_connector.attach_data(model, val_dataloaders=val_dl)
-    with pytest.warns(PossibleUserWarning, match="recommended .* turn shuffling off for val/test/predict"):
-        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)


Code review comments:
diff_hunk: @@ -460,7 +460,12 @@ def replace_sampler(dataloader: DataLoader) -> DataLoader:
 
     @staticmethod
     def _check_eval_shuffling(dataloader, mode):
-        if _is_dataloader_shuffled(dataloader):
+        # limit this warning only for samplers assigned automatically when shuffle is set
+        if (
+            hasattr(dataloader, "sampler")
+            and isinstance(dataloader.sampler, (RandomSampler, SequentialSampler))
path: pytorch_lightning/trainer/connectors/data_connector.py
text: @user1:
Why not move these conditions inside `is_dataloader_shuffled`?

@author:
`is_dataloader_shuffled` is a very loose check due to the behavior of how `shuffle` is treated in the `DataLoader`. I am trying to create a `we don't care situation` here if it's a custom sampler.

If we keep it inside `is_dataloader_shuffled` then it's hard to predict whether the sampler is working sequentially or not if it's a custom one. Right now it returns True if the sampler is not a sequential sampler, but if we move it there, then we need to return False to satisfy the condition here.

@user1:
I did not completely understand your argument here. However, I tried implementing my idea and it seems to be passing the tests added: https://github.com/PyTorchLightning/pytorch-lightning/pull/12653/commits/a030de37e303f8e4ffd973da8960e494fda83497

Can you double check it? Do we need to add a test case for an edge I missed?

Thank you!

@author:
okay.. I think this works too. For the custom sampler, it's going to return False now, and don't throw the warning. Thank you :)

diff_hunk: @@ -11,20 +11,388 @@
 # 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.
+from contextlib import redirect_stderr
+from io import StringIO
+from re import escape
 from unittest.mock import Mock
 
 import pytest
-from torch.utils.data import DataLoader
+from torch.utils.data import BatchSampler, DataLoader, DistributedSampler, Sampler, SequentialSampler
 
 from pytorch_lightning import Trainer
+from pytorch_lightning.strategies import DDPSpawnStrategy
 from pytorch_lightning.trainer.connectors.data_connector import _DataHookSelector, _DataLoaderSource, warning_cache
-from pytorch_lightning.trainer.states import TrainerFn
+from pytorch_lightning.trainer.states import RunningStage, TrainerFn
+from pytorch_lightning.trainer.supporters import CombinedLoader
+from pytorch_lightning.utilities.data import _update_dataloader
+from pytorch_lightning.utilities.exceptions import MisconfigurationException
 from pytorch_lightning.utilities.warnings import PossibleUserWarning
-from tests.helpers import BoringDataModule, BoringModel
-from tests.helpers.boring_model import RandomDataset
+from tests.helpers.boring_model import BoringDataModule, BoringModel, RandomDataset
+from tests.helpers.runif import RunIf
 from tests.helpers.utils import no_warning_call
 
 
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("mode", (1, 2))
+def test_replace_distributed_sampler(tmpdir, mode):
+    class IndexedRandomDataset(RandomDataset):
+        def __getitem__(self, index):
+            return self.data[index]
+
+    class CustomDataLoader(DataLoader):
+        def __init__(self, num_features, dataset, *args, **kwargs):
+            # argument `num_features` unused on purpose
+            # it gets automatically captured by _replace_dataloader_init_method()
+            super().__init__(dataset, *args, **kwargs)
+
+    class CustomBatchSampler(BatchSampler):
+        pass
+
+    class TestModel(BoringModel):
+        def __init__(self, numbers_test_dataloaders, mode):
+            super().__init__()
+            self._numbers_test_dataloaders = numbers_test_dataloaders
+            self._mode = mode
+
+        def test_step(self, batch, batch_idx, dataloader_idx=0):
+            return super().test_step(batch, batch_idx)
+
+        def on_test_start(self) -> None:
+            dataloader = self.trainer.test_dataloaders[0]
+            assert isinstance(dataloader, CustomDataLoader)
+            batch_sampler = dataloader.batch_sampler
+            if self._mode == 1:
+                assert isinstance(batch_sampler, CustomBatchSampler)
+                # the batch_size is set on the batch sampler
+                assert dataloader.batch_size is None
+            elif self._mode == 2:
+                assert type(batch_sampler) is BatchSampler
+                assert dataloader.batch_size == self._mode
+            assert batch_sampler.batch_size == self._mode
+            assert batch_sampler.drop_last
+            # the sampler has been replaced
+            assert isinstance(batch_sampler.sampler, DistributedSampler)
+
+        def create_dataset(self):
+            dataset = IndexedRandomDataset(32, 64)
+            if self._mode == 1:
+                # with a custom batch sampler
+                batch_sampler = CustomBatchSampler(SequentialSampler(dataset), batch_size=1, drop_last=True)
+                return CustomDataLoader(32, dataset, batch_sampler=batch_sampler)
+            elif self._mode == 2:
+                # with no batch sampler provided
+                return CustomDataLoader(32, dataset, batch_size=2, drop_last=True)
+
+        def test_dataloader(self):
+            return [self.create_dataset()] * self._numbers_test_dataloaders
+
+    model = TestModel(2, mode)
+    model.test_epoch_end = None
+
+    trainer = Trainer(
+        default_root_dir=tmpdir,
+        limit_test_batches=2,
+        accelerator="cpu",
+        devices=1,
+        strategy="ddp_find_unused_parameters_false",
+    )
+    trainer.test(model)
+
+
+class TestSpawnBoringModel(BoringModel):
+    def __init__(self, num_workers):
+        super().__init__()
+        self.num_workers = num_workers
+
+    def train_dataloader(self):
+        return DataLoader(RandomDataset(32, 64), num_workers=self.num_workers)
+
+    def on_fit_start(self):
+        self._resout = StringIO()
+        self.ctx = redirect_stderr(self._resout)
+        self.ctx.__enter__()
+
+    def on_train_end(self):
+        def _get_warning_msg():
+            dl = self.trainer.train_dataloader.loaders
+            if hasattr(dl, "persistent_workers"):
+                if self.num_workers == 0:
+                    warn_str = "Consider setting num_workers>0 and persistent_workers=True"
+                else:
+                    warn_str = "Consider setting persistent_workers=True"
+            else:
+                warn_str = "Consider setting strategy=ddp"
+
+            return warn_str
+
+        if self.trainer.is_global_zero:
+            self.ctx.__exit__(None, None, None)
+            msg = self._resout.getvalue()
+            warn_str = _get_warning_msg()
+            assert warn_str in msg
+
+
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("num_workers", [0, 1])
+def test_dataloader_warnings(tmpdir, num_workers):
+    trainer = Trainer(default_root_dir=tmpdir, accelerator="cpu", devices=2, strategy="ddp_spawn", fast_dev_run=4)
+    assert isinstance(trainer.strategy, DDPSpawnStrategy)
+    trainer.fit(TestSpawnBoringModel(num_workers))
+
+
+def test_update_dataloader_raises():
+    with pytest.raises(ValueError, match="needs to subclass `torch.utils.data.DataLoader"):
+        _update_dataloader(object(), object(), mode="fit")
+
+
+def test_dataloaders_with_missing_keyword_arguments():
+    ds = RandomDataset(10, 20)
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset, *args, **kwargs):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, *foo, **bar):
+            super().__init__(*foo, **bar)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, *args, shuffle=False):
+            self.num_feat = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, **kwargs):
+            self.feat_num = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+
+def test_update_dataloader_with_multiprocessing_context():
+    """This test verifies that replace_sampler conserves multiprocessing context."""
+    train = RandomDataset(32, 64)
+    context = "spawn"
+    train = DataLoader(train, batch_size=32, num_workers=2, multiprocessing_context=context, shuffle=True)
+    new_data_loader = _update_dataloader(train, SequentialSampler(train.dataset))
+    assert new_data_loader.multiprocessing_context == train.multiprocessing_context
+
+
+def test_dataloader_reinit_for_subclass():
+    class CustomDataLoader(DataLoader):
+        def __init__(
+            self,
+            dataset,
+            batch_size=1,
+            shuffle=False,
+            sampler=None,
+            batch_sampler=None,
+            num_workers=0,
+            collate_fn=None,
+            pin_memory=False,
+            drop_last=False,
+            timeout=0,
+            worker_init_fn=None,
+            dummy_kwarg=None,
+        ):
+            super().__init__(
+                dataset,
+                batch_size,
+                shuffle,
+                sampler,
+                batch_sampler,
+                num_workers,
+                collate_fn,
+                pin_memory,
+                drop_last,
+                timeout,
+                worker_init_fn,
+            )
+            self.dummy_kwarg = dummy_kwarg
+            self.something_unrelated = 1
+
+    trainer = Trainer(accelerator="cpu", devices=2, strategy="ddp_spawn")
+
+    class CustomDummyObj:
+        sampler = None
+
+    result = trainer._data_connector._prepare_dataloader(CustomDummyObj(), shuffle=True)
+    assert isinstance(result, CustomDummyObj), "Wrongly reinstantiated data loader"
+
+    dataset = list(range(10))
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    # Shuffled DataLoader should also work
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset, shuffle=True), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    class CustomSampler(Sampler):
+        pass
+
+    # Should raise an error if existing sampler is being replaced
+    dataloader = CustomDataLoader(dataset, sampler=CustomSampler(dataset))
+    with pytest.raises(MisconfigurationException, match="will be replaced by `DistributedSampler`"):
+        trainer._data_connector._prepare_dataloader(dataloader, shuffle=True)
+
+
+class LoaderTestModel(BoringModel):
+    def training_step(self, batch, batch_idx):
+        assert len(self.trainer.train_dataloader.loaders) == 10
+        return super().training_step(batch, batch_idx)
+
+    def validation_step(self, batch, batch_idx):
+        assert len(self.trainer.val_dataloaders[0]) == 10
+        return super().validation_step(batch, batch_idx)
+
+    def test_step(self, batch, batch_idx):
+        assert len(self.trainer.test_dataloaders[0]) == 10
+        return super().test_step(batch, batch_idx)
+
+    def predict_step(self, batch, batch_idx, dataloader_idx=0):
+        assert len(self.trainer.predict_dataloaders[0]) == 10
+        return super().predict_step(batch, batch_idx, dataloader_idx=dataloader_idx)
+
+
+def test_loader_detaching():
+    """Checks that the loader has been reset after the entrypoint."""
+
+    loader = DataLoader(RandomDataset(32, 10), batch_size=1)
+
+    model = LoaderTestModel()
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer = Trainer(fast_dev_run=1)
+    trainer.fit(model, loader, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.validate(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.predict(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.test(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+
+def test_pre_made_batches():
+    """Check that loader works with pre-made batches."""
+    loader = DataLoader(RandomDataset(32, 10), batch_size=None)
+    trainer = Trainer(fast_dev_run=1)
+    trainer.predict(LoaderTestModel(), loader)
+
+
+def test_error_raised_with_float_limited_eval_batches():
+    """Test that an error is raised if there are not enough batches when passed with float value of
+    limit_eval_batches."""
+    model = BoringModel()
+    dl_size = len(model.val_dataloader())
+    limit_val_batches = 1 / (dl_size + 2)
+    trainer = Trainer(limit_val_batches=limit_val_batches)
+    trainer._data_connector.attach_data(model)
+    with pytest.raises(
+        MisconfigurationException,
+        match=rf"{limit_val_batches} \* {dl_size} < 1. Please increase the `limit_val_batches`",
+    ):
+        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)
+
+
+@pytest.mark.parametrize(
+    "val_dl,raises_exception",
+    [
+        (DataLoader(dataset=RandomDataset(32, 64), shuffle=True), True),
+        (DataLoader(dataset=RandomDataset(32, 64), sampler=list(range(64))), False),
+        (CombinedLoader(DataLoader(dataset=RandomDataset(32, 64), shuffle=True)), True),
+        (
+            CombinedLoader(
+                [DataLoader(dataset=RandomDataset(32, 64)), DataLoader(dataset=RandomDataset(32, 64), shuffle=True)]
+            ),
+            True,
+        ),
+        (
+            CombinedLoader(
+                {
+                    "dl1": DataLoader(dataset=RandomDataset(32, 64)),
+                    "dl2": DataLoader(dataset=RandomDataset(32, 64), shuffle=True),
+                }
+            ),
+            True,
+        ),
+    ],
+)
+def test_non_sequential_sampler_warning_is_raised_for_eval_dataloader(val_dl, raises_exception):
+    trainer = Trainer()
+    model = BoringModel()
+    trainer._data_connector.attach_data(model, val_dataloaders=val_dl)
+    context = pytest.warns if raises_exception else no_warning_call
path: tests/trainer/connectors/test_data_connector.py
text: @author:
```suggestion
def test_non_sequential_sampler_warning_is_raised_for_eval_dataloader(val_dl, warns):
    trainer = Trainer()
    model = BoringModel()
    trainer._data_connector.attach_data(model, val_dataloaders=val_dl)
    context = pytest.warns if warns else no_warning_call
```

diff_hunk: @@ -126,7 +127,7 @@ def no_warning_call(expected_warning: Type[Warning] = UserWarning, match: Option
             return
     else:
         for w in record.list:
-            if w.category is expected_warning and match in w.message.args[0]:
+            if w.category is expected_warning and re.compile(match).search(str(w.message.args[0])):
path: tests/helpers/utils.py
text: @author:
```suggestion
            if w.category is expected_warning and re.compile(match).search(w.message.args[0]):
```

diff_hunk: @@ -126,7 +127,7 @@ def no_warning_call(expected_warning: Type[Warning] = UserWarning, match: Option
             return
     else:
         for w in record.list:
-            if w.category is expected_warning and match in w.message.args[0]:
+            if w.category is expected_warning and re.compile(match).search(str(w.message.args[0])):
path: tests/helpers/utils.py
text: @author:
this is a small fix to ensure we can match regex patterns.

diff_hunk: @@ -11,20 +11,388 @@
 # 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.
+from contextlib import redirect_stderr
+from io import StringIO
+from re import escape
 from unittest.mock import Mock
 
 import pytest
-from torch.utils.data import DataLoader
+from torch.utils.data import BatchSampler, DataLoader, DistributedSampler, Sampler, SequentialSampler
 
 from pytorch_lightning import Trainer
+from pytorch_lightning.strategies import DDPSpawnStrategy
 from pytorch_lightning.trainer.connectors.data_connector import _DataHookSelector, _DataLoaderSource, warning_cache
-from pytorch_lightning.trainer.states import TrainerFn
+from pytorch_lightning.trainer.states import RunningStage, TrainerFn
+from pytorch_lightning.trainer.supporters import CombinedLoader
+from pytorch_lightning.utilities.data import _update_dataloader
+from pytorch_lightning.utilities.exceptions import MisconfigurationException
 from pytorch_lightning.utilities.warnings import PossibleUserWarning
-from tests.helpers import BoringDataModule, BoringModel
-from tests.helpers.boring_model import RandomDataset
+from tests.helpers.boring_model import BoringDataModule, BoringModel, RandomDataset
+from tests.helpers.runif import RunIf
 from tests.helpers.utils import no_warning_call
 
 
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("mode", (1, 2))
+def test_replace_distributed_sampler(tmpdir, mode):
+    class IndexedRandomDataset(RandomDataset):
+        def __getitem__(self, index):
+            return self.data[index]
+
+    class CustomDataLoader(DataLoader):
+        def __init__(self, num_features, dataset, *args, **kwargs):
+            # argument `num_features` unused on purpose
+            # it gets automatically captured by _replace_dataloader_init_method()
+            super().__init__(dataset, *args, **kwargs)
+
+    class CustomBatchSampler(BatchSampler):
+        pass
+
+    class TestModel(BoringModel):
+        def __init__(self, numbers_test_dataloaders, mode):
+            super().__init__()
+            self._numbers_test_dataloaders = numbers_test_dataloaders
+            self._mode = mode
+
+        def test_step(self, batch, batch_idx, dataloader_idx=0):
+            return super().test_step(batch, batch_idx)
+
+        def on_test_start(self) -> None:
+            dataloader = self.trainer.test_dataloaders[0]
+            assert isinstance(dataloader, CustomDataLoader)
+            batch_sampler = dataloader.batch_sampler
+            if self._mode == 1:
+                assert isinstance(batch_sampler, CustomBatchSampler)
+                # the batch_size is set on the batch sampler
+                assert dataloader.batch_size is None
+            elif self._mode == 2:
+                assert type(batch_sampler) is BatchSampler
+                assert dataloader.batch_size == self._mode
+            assert batch_sampler.batch_size == self._mode
+            assert batch_sampler.drop_last
+            # the sampler has been replaced
+            assert isinstance(batch_sampler.sampler, DistributedSampler)
+
+        def create_dataset(self):
+            dataset = IndexedRandomDataset(32, 64)
+            if self._mode == 1:
+                # with a custom batch sampler
+                batch_sampler = CustomBatchSampler(SequentialSampler(dataset), batch_size=1, drop_last=True)
+                return CustomDataLoader(32, dataset, batch_sampler=batch_sampler)
+            elif self._mode == 2:
+                # with no batch sampler provided
+                return CustomDataLoader(32, dataset, batch_size=2, drop_last=True)
+
+        def test_dataloader(self):
+            return [self.create_dataset()] * self._numbers_test_dataloaders
+
+    model = TestModel(2, mode)
+    model.test_epoch_end = None
+
+    trainer = Trainer(
+        default_root_dir=tmpdir,
+        limit_test_batches=2,
+        accelerator="cpu",
+        devices=1,
+        strategy="ddp_find_unused_parameters_false",
+    )
+    trainer.test(model)
+
+
+class TestSpawnBoringModel(BoringModel):
+    def __init__(self, num_workers):
+        super().__init__()
+        self.num_workers = num_workers
+
+    def train_dataloader(self):
+        return DataLoader(RandomDataset(32, 64), num_workers=self.num_workers)
+
+    def on_fit_start(self):
+        self._resout = StringIO()
+        self.ctx = redirect_stderr(self._resout)
+        self.ctx.__enter__()
+
+    def on_train_end(self):
+        def _get_warning_msg():
+            dl = self.trainer.train_dataloader.loaders
+            if hasattr(dl, "persistent_workers"):
+                if self.num_workers == 0:
+                    warn_str = "Consider setting num_workers>0 and persistent_workers=True"
+                else:
+                    warn_str = "Consider setting persistent_workers=True"
+            else:
+                warn_str = "Consider setting strategy=ddp"
+
+            return warn_str
+
+        if self.trainer.is_global_zero:
+            self.ctx.__exit__(None, None, None)
+            msg = self._resout.getvalue()
+            warn_str = _get_warning_msg()
+            assert warn_str in msg
+
+
+@RunIf(skip_windows=True)
+@pytest.mark.parametrize("num_workers", [0, 1])
+def test_dataloader_warnings(tmpdir, num_workers):
+    trainer = Trainer(default_root_dir=tmpdir, accelerator="cpu", devices=2, strategy="ddp_spawn", fast_dev_run=4)
+    assert isinstance(trainer.strategy, DDPSpawnStrategy)
+    trainer.fit(TestSpawnBoringModel(num_workers))
+
+
+def test_update_dataloader_raises():
+    with pytest.raises(ValueError, match="needs to subclass `torch.utils.data.DataLoader"):
+        _update_dataloader(object(), object(), mode="fit")
+
+
+def test_dataloaders_with_missing_keyword_arguments():
+    ds = RandomDataset(10, 20)
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler', 'shuffle']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, dataset, *args, **kwargs):
+            super().__init__(dataset)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, *foo, **bar):
+            super().__init__(*foo, **bar)
+
+    loader = TestDataLoader(ds)
+    sampler = SequentialSampler(ds)
+    _update_dataloader(loader, sampler, mode="fit")
+    _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, *args, shuffle=False):
+            self.num_feat = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing arguments are ['batch_sampler', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing arguments are ['batch_sampler', 'batch_size', 'drop_last', 'sampler']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+    class TestDataLoader(DataLoader):
+        def __init__(self, num_feat, dataset, **kwargs):
+            self.feat_num = num_feat
+            super().__init__(dataset)
+
+    loader = TestDataLoader(1, ds)
+    sampler = SequentialSampler(ds)
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="fit")
+    match = escape("missing attributes are ['num_feat']")
+    with pytest.raises(MisconfigurationException, match=match):
+        _update_dataloader(loader, sampler, mode="predict")
+
+
+def test_update_dataloader_with_multiprocessing_context():
+    """This test verifies that replace_sampler conserves multiprocessing context."""
+    train = RandomDataset(32, 64)
+    context = "spawn"
+    train = DataLoader(train, batch_size=32, num_workers=2, multiprocessing_context=context, shuffle=True)
+    new_data_loader = _update_dataloader(train, SequentialSampler(train.dataset))
+    assert new_data_loader.multiprocessing_context == train.multiprocessing_context
+
+
+def test_dataloader_reinit_for_subclass():
+    class CustomDataLoader(DataLoader):
+        def __init__(
+            self,
+            dataset,
+            batch_size=1,
+            shuffle=False,
+            sampler=None,
+            batch_sampler=None,
+            num_workers=0,
+            collate_fn=None,
+            pin_memory=False,
+            drop_last=False,
+            timeout=0,
+            worker_init_fn=None,
+            dummy_kwarg=None,
+        ):
+            super().__init__(
+                dataset,
+                batch_size,
+                shuffle,
+                sampler,
+                batch_sampler,
+                num_workers,
+                collate_fn,
+                pin_memory,
+                drop_last,
+                timeout,
+                worker_init_fn,
+            )
+            self.dummy_kwarg = dummy_kwarg
+            self.something_unrelated = 1
+
+    trainer = Trainer(accelerator="cpu", devices=2, strategy="ddp_spawn")
+
+    class CustomDummyObj:
+        sampler = None
+
+    result = trainer._data_connector._prepare_dataloader(CustomDummyObj(), shuffle=True)
+    assert isinstance(result, CustomDummyObj), "Wrongly reinstantiated data loader"
+
+    dataset = list(range(10))
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    # Shuffled DataLoader should also work
+    result = trainer._data_connector._prepare_dataloader(CustomDataLoader(dataset, shuffle=True), shuffle=True)
+    assert isinstance(result, DataLoader)
+    assert isinstance(result, CustomDataLoader)
+    assert result.dummy_kwarg is None
+
+    class CustomSampler(Sampler):
+        pass
+
+    # Should raise an error if existing sampler is being replaced
+    dataloader = CustomDataLoader(dataset, sampler=CustomSampler(dataset))
+    with pytest.raises(MisconfigurationException, match="will be replaced by `DistributedSampler`"):
+        trainer._data_connector._prepare_dataloader(dataloader, shuffle=True)
+
+
+class LoaderTestModel(BoringModel):
+    def training_step(self, batch, batch_idx):
+        assert len(self.trainer.train_dataloader.loaders) == 10
+        return super().training_step(batch, batch_idx)
+
+    def validation_step(self, batch, batch_idx):
+        assert len(self.trainer.val_dataloaders[0]) == 10
+        return super().validation_step(batch, batch_idx)
+
+    def test_step(self, batch, batch_idx):
+        assert len(self.trainer.test_dataloaders[0]) == 10
+        return super().test_step(batch, batch_idx)
+
+    def predict_step(self, batch, batch_idx, dataloader_idx=0):
+        assert len(self.trainer.predict_dataloaders[0]) == 10
+        return super().predict_step(batch, batch_idx, dataloader_idx=dataloader_idx)
+
+
+def test_loader_detaching():
+    """Checks that the loader has been reset after the entrypoint."""
+
+    loader = DataLoader(RandomDataset(32, 10), batch_size=1)
+
+    model = LoaderTestModel()
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer = Trainer(fast_dev_run=1)
+    trainer.fit(model, loader, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.validate(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.predict(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+    trainer.test(model, loader)
+
+    assert len(model.train_dataloader()) == 64
+    assert len(model.val_dataloader()) == 64
+    assert len(model.predict_dataloader()) == 64
+    assert len(model.test_dataloader()) == 64
+
+
+def test_pre_made_batches():
+    """Check that loader works with pre-made batches."""
+    loader = DataLoader(RandomDataset(32, 10), batch_size=None)
+    trainer = Trainer(fast_dev_run=1)
+    trainer.predict(LoaderTestModel(), loader)
+
+
+def test_error_raised_with_float_limited_eval_batches():
+    """Test that an error is raised if there are not enough batches when passed with float value of
+    limit_eval_batches."""
+    model = BoringModel()
+    dl_size = len(model.val_dataloader())
+    limit_val_batches = 1 / (dl_size + 2)
+    trainer = Trainer(limit_val_batches=limit_val_batches)
+    trainer._data_connector.attach_data(model)
+    with pytest.raises(
+        MisconfigurationException,
+        match=rf"{limit_val_batches} \* {dl_size} < 1. Please increase the `limit_val_batches`",
+    ):
+        trainer._data_connector._reset_eval_dataloader(RunningStage.VALIDATING, model)
+
+
+@pytest.mark.parametrize(
+    "val_dl,raises_exception",
path: tests/trainer/connectors/test_data_connector.py
text: @user1:
```suggestion
    "val_dl,warns",
```