Repository: Lightning-AI/pytorch-lightning
Pull request: Error checking for non-iterables
## What does this PR do?

Fixes https://github.com/Lightning-AI/lightning/issues/16912

cc @borda @justusschock @awaelchli

Problem statement:
CombinedLoader: NoneType object is not iterable
### Bug description

When the `LM.val_dataloader` method returns None, two unexpected behaviors take place:

1. There is a warning that shouldn't show:

```
UserWarning: Total length of `NoneType` across ranks is zero. Please make sure this was your intention.
```

2. There is an error, because the loop tries to run validation with `None` as the dataloader:

```
TypeError: 'NoneType' object is not iterable
```

### How to reproduce the bug

```python
import os

import torch
from torch.utils.data import DataLoader, Dataset

from lightning.pytorch 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 train_dataloader(self):
        return DataLoader(RandomDataset(32, 64), batch_size=2)

    def val_dataloader(self):
        return None

    def configure_optimizers(self):
        return torch.optim.SGD(self.layer.parameters(), lr=0.1)


def run():
    model = BoringModel()
    trainer = Trainer(
        default_root_dir=os.getcwd(),
        limit_train_batches=1,
        limit_val_batches=1,
        num_sanity_val_steps=0,
        max_epochs=2,
        enable_model_summary=False,
    )
    trainer.fit(model)


if __name__ == "__main__":
    run()
```


### Error messages and logs

```
  File "/Users/adrian/repositories/lightning/examples/pl_bug_report/bug_report_model.py", line 62, in <module>
    run()
  File "/Users/adrian/repositories/lightning/examples/pl_bug_report/bug_report_model.py", line 58, in run
    trainer.fit(model)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/trainer.py", line 517, in fit
    call._call_and_handle_interrupt(
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/call.py", line 44, in _call_and_handle_interrupt
    return trainer_fn(*args, **kwargs)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/trainer.py", line 556, in _fit_impl
    self._run(model, ckpt_path=ckpt_path)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/trainer.py", line 928, in _run
    results = self._run_stage()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/trainer.py", line 967, in _run_stage
    self._run_train()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/trainer/trainer.py", line 988, in _run_train
    self.fit_loop.run()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/fit_loop.py", line 192, in run
    self.advance()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/fit_loop.py", line 365, in advance
    self.epoch_loop.run(self._data_fetcher)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/training_epoch_loop.py", line 134, in run
    self.on_advance_end()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/training_epoch_loop.py", line 248, in on_advance_end
    self.val_loop.run()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/utilities.py", line 167, in _decorator
    return loop_run(self, *args, **kwargs)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/evaluation_loop.py", line 93, in run
    self.reset()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/evaluation_loop.py", line 184, in reset
    iter(data_fetcher)  # creates the iterator inside the fetcher
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/fetchers.py", line 104, in __iter__
    super().__iter__()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/loops/fetchers.py", line 54, in __iter__
    self.dataloader_iter = iter(self.dataloader)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/utilities/combined_loader.py", line 242, in __iter__
    iter(iterator)
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/utilities/combined_loader.py", line 121, in __iter__
    super().__iter__()
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/utilities/combined_loader.py", line 35, in __iter__
    self.iterators = [iter(iterable) for iterable in self.iterables]
  File "/Users/adrian/repositories/lightning/src/lightning/pytorch/utilities/combined_loader.py", line 35, in <listcomp>
    self.iterators = [iter(iterable) for iterable in self.iterables]
TypeError: 'NoneType' object is not iterable
```


### Environment

<details>
  <summary>Current environment</summary>

```
#- PyTorch Lightning Version (e.g., 1.5.0): 2.0.0rc0
#- PyTorch Version (e.g., 2.0): 1.13.1
#- Python version (e.g., 3.9): 3.10
#- OS (e.g., Linux): MacOS
```

</details>


### More info

_No response_

cc @justusschock @awaelchli @borda

Hints from issue discussion:
@carmocca Do you agree that this was an unintended change from 1.9.x?
I believe the use case is valid: You may want to conditionally return a val loader from a DataModule or LightningModule depending on availability of validation data or not. 
This breaking change was intended: https://github.com/Lightning-AI/lightning/pull/16800#discussion_r1110256813
It is only in master

Skipping None could be error prone and lead to silent bugs. If you had no data, then you'd return an empty iterable
And what is the argument against converting the original warning to an error? Don't you think it is strange to get the warning "Total length of `NoneType` across ranks is zero". It doesn't mention that his is about a dataloader, so it is unclear to the user what should be done.
Yes. I'll add error checks in the loops for this

Commit message: Fix test

Patch to review:
diff --git a/src/lightning/pytorch/CHANGELOG.md b/src/lightning/pytorch/CHANGELOG.md
index 945f6380fad7b..1a258d24b352d 100644
--- a/src/lightning/pytorch/CHANGELOG.md
+++ b/src/lightning/pytorch/CHANGELOG.md
@@ -62,6 +62,9 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/).
 
 - Added `DDPStrategy(start_method=...)` argument, defaulting to 'popen' ([#16809](https://github.com/Lightning-AI/lightning/pull/16809))
 
+
+- Added checks for whether the iterables used by the loops are valid ([#17007](https://github.com/Lightning-AI/lightning/pull/17007))
+
 ### Changed
 
 
diff --git a/src/lightning/pytorch/loops/evaluation_loop.py b/src/lightning/pytorch/loops/evaluation_loop.py
index faf56b0648e33..424255c2a7189 100644
--- a/src/lightning/pytorch/loops/evaluation_loop.py
+++ b/src/lightning/pytorch/loops/evaluation_loop.py
@@ -29,6 +29,7 @@
 from lightning.pytorch.loops.utilities import _no_grad_context, _select_data_fetcher, _verify_dataloader_idx_requirement
 from lightning.pytorch.trainer import call
 from lightning.pytorch.trainer.connectors.data_connector import (
+    _check_dataloader_iterable,
     _DataLoaderSource,
     _parse_num_batches,
     _process_dataloader,
@@ -123,15 +124,15 @@ def run(self) -> List[_OUT_DICT]:
     def setup_data(self) -> None:
         trainer = self.trainer
         trainer_fn = trainer.state.fn
+        assert trainer_fn is not None
 
         if self._combined_loader is not None and trainer_fn == "fit" and not self._should_reload_val_dl:
             return
 
-        source = self._data_source
         pl_module = trainer.lightning_module
         limit_batches = trainer.limit_test_batches if trainer.testing else trainer.limit_val_batches
         hook_name = "test_step" if trainer.testing else "validation_step"
-        if not source.is_defined() or limit_batches == 0 or not is_overridden(hook_name, pl_module):
+        if limit_batches == 0 or not is_overridden(hook_name, pl_module):
             return
 
         # store epoch of dataloader reset for reload_dataloaders_every_n_epochs
@@ -145,6 +146,7 @@ def setup_data(self) -> None:
         stage = trainer.state.stage
         assert stage is not None
 
+        source = self._data_source
         dataloaders = _request_dataloader(source)
         trainer.strategy.barrier(f"{stage.dataloader_prefix}_dataloader()")
 
@@ -163,6 +165,7 @@ def setup_data(self) -> None:
         dataloaders = []
         self._max_batches = []
         for dl in combined_loader.flattened:
+            _check_dataloader_iterable(dl, source, trainer_fn)
             dl = _process_dataloader(trainer, dl)
             dataloaders.append(dl)
 
diff --git a/src/lightning/pytorch/loops/fit_loop.py b/src/lightning/pytorch/loops/fit_loop.py
index a42a6e1b28edb..529b366668ef5 100644
--- a/src/lightning/pytorch/loops/fit_loop.py
+++ b/src/lightning/pytorch/loops/fit_loop.py
@@ -23,6 +23,7 @@
 from lightning.pytorch.loops.utilities import _is_max_limit_reached, _select_data_fetcher
 from lightning.pytorch.trainer import call
 from lightning.pytorch.trainer.connectors.data_connector import (
+    _check_dataloader_iterable,
     _DataLoaderSource,
     _parse_num_batches,
     _process_dataloader,
@@ -30,7 +31,7 @@
     _resolve_overfit_batches,
 )
 from lightning.pytorch.trainer.connectors.logger_connector.result import _ResultCollection
-from lightning.pytorch.trainer.states import RunningStage
+from lightning.pytorch.trainer.states import RunningStage, TrainerFn
 from lightning.pytorch.utilities.combined_loader import CombinedLoader
 from lightning.pytorch.utilities.data import has_len_all_ranks
 from lightning.pytorch.utilities.exceptions import MisconfigurationException, SIGTERMException
@@ -208,13 +209,13 @@ def setup_data(self) -> None:
             return
 
         trainer = self.trainer
-        source = self._data_source
         pl_module = trainer.lightning_module
-        if not source.is_defined() or trainer.limit_train_batches == 0 or not is_overridden("training_step", pl_module):
+        if trainer.limit_train_batches == 0 or not is_overridden("training_step", pl_module):
             return
 
         log.debug(f"{self.__class__.__name__}: resetting train dataloader")
 
+        source = self._data_source
         train_dataloader = _request_dataloader(source)
         trainer.strategy.barrier("train_dataloader()")
 
@@ -226,7 +227,12 @@ def setup_data(self) -> None:
         if trainer.overfit_batches > 0:
             _resolve_overfit_batches(combined_loader, mode=RunningStage.TRAINING)
 
-        dataloaders = [_process_dataloader(trainer, dl) for dl in combined_loader.flattened]
+        trainer_fn = TrainerFn.FITTING
+        dataloaders = []
+        for dl in combined_loader.flattened:
+            _check_dataloader_iterable(dl, source, trainer_fn)
+            dl = _process_dataloader(trainer, dl)
+            dataloaders.append(dl)
         combined_loader.flattened = dataloaders
         self._combined_loader = combined_loader
 
diff --git a/src/lightning/pytorch/loops/prediction_loop.py b/src/lightning/pytorch/loops/prediction_loop.py
index 4057a853c528d..cdce413e211fc 100644
--- a/src/lightning/pytorch/loops/prediction_loop.py
+++ b/src/lightning/pytorch/loops/prediction_loop.py
@@ -28,12 +28,13 @@
 from lightning.pytorch.strategies.launchers import _MultiProcessingLauncher
 from lightning.pytorch.trainer import call
 from lightning.pytorch.trainer.connectors.data_connector import (
+    _check_dataloader_iterable,
     _DataLoaderSource,
     _parse_num_batches,
     _process_dataloader,
     _request_dataloader,
 )
-from lightning.pytorch.trainer.states import RunningStage
+from lightning.pytorch.trainer.states import RunningStage, TrainerFn
 from lightning.pytorch.utilities.combined_loader import _Sequential, CombinedLoader
 from lightning.pytorch.utilities.data import has_len_all_ranks
 from lightning.pytorch.utilities.exceptions import MisconfigurationException
@@ -118,11 +119,11 @@ def run(self) -> Optional[_PREDICT_OUTPUT]:
 
     def setup_data(self) -> None:
         trainer = self.trainer
-        source = self._data_source
-        # a dfault `predict_step` exists in the LightningModule, so no need to check if it's overridden
-        if not source.is_defined() or trainer.limit_predict_batches == 0:
+        # a default `predict_step` exists in the LightningModule, so no need to check if it's overridden
+        if trainer.limit_predict_batches == 0:
             return
 
+        source = self._data_source
         dataloaders = _request_dataloader(source)
         trainer.strategy.barrier("predict_dataloader()")
 
@@ -135,10 +136,12 @@ def setup_data(self) -> None:
         if trainer.datamodule is not None:
             allow_zero_length |= trainer.datamodule.allow_zero_length_dataloader_with_multiple_devices
 
+        trainer_fn = TrainerFn.PREDICTING
         stage = RunningStage.PREDICTING
         dataloaders = []
         self.max_batches = []
         for dl in combined_loader.flattened:
+            _check_dataloader_iterable(dl, source, trainer_fn)
             dl = _process_dataloader(trainer, dl)
             dataloaders.append(dl)
 
diff --git a/src/lightning/pytorch/trainer/connectors/data_connector.py b/src/lightning/pytorch/trainer/connectors/data_connector.py
index 99bc55a5e0ce9..1bddeac0870d7 100644
--- a/src/lightning/pytorch/trainer/connectors/data_connector.py
+++ b/src/lightning/pytorch/trainer/connectors/data_connector.py
@@ -118,21 +118,8 @@ def attach_data(
         )
         self.attach_datamodule(model, datamodule=datamodule)
 
-        trainer = self.trainer
-        fn = trainer.state.fn
-        # Validate that the required data sources are available
-        if fn == TrainerFn.FITTING:
-            _check_dataloader_none(train_dataloaders, trainer.fit_loop._data_source, fn)
-            # TODO(carmocca): fit's validation dataloaders should be checked too
-        elif fn == TrainerFn.VALIDATING:
-            _check_dataloader_none(val_dataloaders, trainer.validate_loop._data_source, fn)
-        elif fn == TrainerFn.TESTING:
-            _check_dataloader_none(test_dataloaders, trainer.test_loop._data_source, fn)
-        elif fn == TrainerFn.PREDICTING:
-            _check_dataloader_none(predict_dataloaders, trainer.predict_loop._data_source, fn)
-
         # Attach the trainer to the LightningModule
-        model.trainer = trainer
+        model.trainer = self.trainer
 
     def attach_dataloaders(
         self,
@@ -264,7 +251,9 @@ def _get_distributed_sampler(
 
 
 def _resolve_overfit_batches(combined_loader: CombinedLoader, mode: RunningStage) -> None:
-    all_have_sequential_sampler = all(isinstance(dl.sampler, SequentialSampler) for dl in combined_loader.flattened)
+    all_have_sequential_sampler = all(
+        isinstance(dl.sampler, SequentialSampler) for dl in combined_loader.flattened if hasattr(dl, "sampler")
+    )
     if all_have_sequential_sampler:
         return
     rank_zero_warn(
@@ -272,7 +261,8 @@ def _resolve_overfit_batches(combined_loader: CombinedLoader, mode: RunningStage
         f" We are turning off the {mode.dataloader_prefix} dataloader shuffling for you."
     )
     updated = [
-        _update_dataloader(dl, sampler=SequentialSampler(dl.dataset), mode=mode) for dl in combined_loader.flattened
+        _update_dataloader(dl, sampler=SequentialSampler(dl.dataset), mode=mode) if hasattr(dl, "dataset") else dl
+        for dl in combined_loader.flattened
     ]
     combined_loader.flattened = updated
 
@@ -303,11 +293,9 @@ def dataloader(self) -> Union[TRAIN_DATALOADERS, EVAL_DATALOADERS]:
         """
         if isinstance(self.instance, pl.LightningModule):
             return call._call_lightning_module_hook(self.instance.trainer, self.name, pl_module=self.instance)
-
         if isinstance(self.instance, pl.LightningDataModule):
-            method = getattr(self.instance, self.name)
-            return method()
-
+            assert self.instance.trainer is not None
+            return call._call_lightning_datamodule_hook(self.instance.trainer, self.name)
         assert self.instance is not None
         return self.instance
 
@@ -386,18 +374,30 @@ def get_instance(self, hook_name: str) -> Union["pl.LightningModule", "pl.Lightn
         return self.model
 
 
-def _check_dataloader_none(
-    dataloader: Optional[Union[TRAIN_DATALOADERS, EVAL_DATALOADERS]],
-    dataloader_source: _DataLoaderSource,
+def _check_dataloader_iterable(
+    dataloader: object,
+    source: _DataLoaderSource,
     trainer_fn: TrainerFn,
 ) -> None:
-    # A prefix in the message to disambiguate between the train- and (optional) val dataloader that .fit() accepts
-    prefix = "train_" if trainer_fn == TrainerFn.FITTING else ""
-    if dataloader is None and not dataloader_source.is_defined():
-        raise ValueError(
-            f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`."
-            f" Either pass the dataloader to the `.{trainer_fn}()` method OR implement"
-            f" `def {dataloader_source.name}(self):` in your LightningModule/LightningDataModule."
+    try:
+        iter(dataloader)  # type: ignore[call-overload]
+    except TypeError:
+        # A prefix in the message to disambiguate between the train- and (optional) val dataloader that .fit() accepts
+        prefix = "train_" if trainer_fn == TrainerFn.FITTING else ""
+        if source.is_module():
+            if not is_overridden(source.name, source.instance):
+                raise TypeError(
+                    f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`."
+                    f" Found {dataloader}."
+                    f" Either pass the dataloader to the `.{trainer_fn}()` method OR implement"
+                    f" `def {source.name}(self):` in your LightningModule/LightningDataModule."
+                )
+            raise TypeError(
+                f"An invalid dataloader was returned from `{type(source.instance).__name__}.{source.name}()`."
+                f" Found {dataloader}."
+            )
+        raise TypeError(
+            f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`. Found {dataloader}."
         )
 
 
diff --git a/tests/tests_pytorch/loops/test_loops.py b/tests/tests_pytorch/loops/test_loops.py
index ab32150d368cc..0d10af82bf7d5 100644
--- a/tests/tests_pytorch/loops/test_loops.py
+++ b/tests/tests_pytorch/loops/test_loops.py
@@ -762,19 +762,39 @@ def test_workers_are_shutdown(tmpdir, should_fail, persistent_workers):
     # `num_workers == 1` uses `_MultiProcessingDataLoaderIter`
     # `persistent_workers` makes sure `self._iterator` gets set on the `DataLoader` instance
 
+    class TestCallback(Callback):
+        def on_train_epoch_end(self, trainer, *_):
+            if trainer.current_epoch == 1:
+                raise CustomException
+
+    max_epochs = 3
+
+    model = BoringModel()
+    trainer = Trainer(
+        default_root_dir=tmpdir,
+        limit_train_batches=2,
+        limit_val_batches=2,
+        max_epochs=max_epochs,
+        callbacks=TestCallback() if should_fail else None,
+        enable_checkpointing=False,
+        enable_model_summary=False,
+        enable_progress_bar=False,
+        logger=False,
+    )
+
     class _TestMultiProcessingDataLoaderIter(_MultiProcessingDataLoaderIter):
         def __init__(self, *args, dataloader, **kwargs):
             super().__init__(*args, **kwargs)
             self.dataloader = dataloader
 
         def _shutdown_workers(self):
-            self.dataloader.count_shutdown_workers += 1
+            self.dataloader.shutdown_workers_epochs.append(trainer.current_epoch)
             super()._shutdown_workers()
 
     class TestDataLoader(DataLoader):
         def __init__(self, *args, **kwargs):
             super().__init__(*args, **kwargs)
-            self.count_shutdown_workers = 0
+            self.shutdown_workers_epochs = []
 
         def _get_iterator(self):
             if self.num_workers == 0:
@@ -786,29 +806,58 @@ def _get_iterator(self):
     train_dataloader = TestDataLoader(RandomDataset(32, 64), num_workers=1, persistent_workers=persistent_workers)
     val_dataloader = TestDataLoader(RandomDataset(32, 64), num_workers=1, persistent_workers=persistent_workers)
 
-    class TestCallback(Callback):
-        def on_train_epoch_end(self, trainer, *_):
-            if trainer.current_epoch == 1:
-                raise CustomException
-
-    max_epochs = 3
-
-    model = BoringModel()
-    trainer = Trainer(
-        default_root_dir=tmpdir,
-        limit_train_batches=2,
-        limit_val_batches=2,
-        max_epochs=max_epochs,
-        callbacks=TestCallback() if should_fail else None,
-    )
-
     if should_fail:
         with pytest.raises(CustomException):
             trainer.fit(model, train_dataloader, val_dataloader)
     else:
         trainer.fit(model, train_dataloader, val_dataloader)
 
-    assert train_dataloader.count_shutdown_workers == 2 if should_fail else (2 if persistent_workers else max_epochs)
-    # on sanity checking end, the workers are being deleted too.
-    expected = 2 if persistent_workers else (3 if should_fail else max_epochs + 1)
-    assert val_dataloader.count_shutdown_workers == expected
+    if persistent_workers:
+        expected = [trainer.current_epoch, trainer.current_epoch]  # once epoch end, once on teardown
+    elif should_fail:
+        expected = [
+            # iterable check
+            0,
+            # epoch ends
+            1,
+            # teardown
+            1,
+        ]
+    else:
+        expected = [
+            # iterable check
+            0,
+            # epoch ends
+            1,
+            2,
+            # teardown
+            3,
+        ]
+    assert train_dataloader.shutdown_workers_epochs == expected
+
+    if persistent_workers:
+        expected = [trainer.current_epoch, trainer.current_epoch]  # once epoch end, once on teardown
+    elif should_fail:
+        expected = [
+            # sanity check
+            0,
+            # iterable check
+            0,
+            # epoch ends
+            1,
+            # teardown
+            1,
+        ]
+    else:
+        expected = [
+            # sanity check
+            0,
+            # iterable check
+            0,
+            # epoch ends
+            1,
+            2,
+            # teardown
+            3,
+        ]
+    assert val_dataloader.shutdown_workers_epochs == expected
diff --git a/tests/tests_pytorch/trainer/connectors/test_data_connector.py b/tests/tests_pytorch/trainer/connectors/test_data_connector.py
index fd1acbfbabc70..86b12aab8a18d 100644
--- a/tests/tests_pytorch/trainer/connectors/test_data_connector.py
+++ b/tests/tests_pytorch/trainer/connectors/test_data_connector.py
@@ -614,8 +614,34 @@ def test_attach_data_input_validation_with_none_dataloader(trainer_fn_name, data
     datamodule.test_dataloader = None
     datamodule.predict_dataloader = None
 
-    with pytest.raises(ValueError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
+    with pytest.raises(TypeError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
         trainer_fn(model, **{dataloader_name: None}, datamodule=datamodule)
 
-    with pytest.raises(ValueError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
+    with pytest.raises(TypeError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
         trainer_fn(model, **{dataloader_name: None}, datamodule=None)
+
+
+@pytest.mark.parametrize(
+    "trainer_fn_name, dataloader_name, stage",
+    [
+        ("fit", "train_dataloaders", RunningStage.TRAINING),
+        ("validate", "dataloaders", RunningStage.VALIDATING),
+        ("test", "dataloaders", RunningStage.TESTING),
+        ("predict", "dataloaders", RunningStage.PREDICTING),
+    ],
+)
+@pytest.mark.parametrize("dataloader", [object(), [1, object()]])
+def test_non_iterables_raise(tmp_path, trainer_fn_name, dataloader_name, stage, dataloader):
+    model = BoringModel()
+    trainer = Trainer(default_root_dir=tmp_path, fast_dev_run=1)
+    trainer_fn = getattr(trainer, trainer_fn_name)
+
+    with pytest.raises(
+        TypeError, match=rf"invalid dataloader was passed to `Trainer.{trainer_fn_name}\({dataloader_name}"
+    ):
+        trainer_fn(model, **{dataloader_name: dataloader})
+
+    dl_method = stage.dataloader_prefix + "_dataloader"
+    setattr(model, dl_method, lambda: dataloader)
+    with pytest.raises(TypeError, match=f"invalid dataloader was returned from `BoringModel.{dl_method}"):
+        trainer_fn(model)


Code review comments:
diff_hunk: @@ -386,18 +374,30 @@ def get_instance(self, hook_name: str) -> Union["pl.LightningModule", "pl.Lightn
         return self.model
 
 
-def _check_dataloader_none(
-    dataloader: Optional[Union[TRAIN_DATALOADERS, EVAL_DATALOADERS]],
-    dataloader_source: _DataLoaderSource,
+def _check_dataloader_iterable(
+    dataloader: object,
+    source: _DataLoaderSource,
     trainer_fn: TrainerFn,
 ) -> None:
-    # A prefix in the message to disambiguate between the train- and (optional) val dataloader that .fit() accepts
-    prefix = "train_" if trainer_fn == TrainerFn.FITTING else ""
-    if dataloader is None and not dataloader_source.is_defined():
-        raise ValueError(
-            f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`."
-            f" Either pass the dataloader to the `.{trainer_fn}()` method OR implement"
-            f" `def {dataloader_source.name}(self):` in your LightningModule/LightningDataModule."
+    try:
+        iter(dataloader)  # type: ignore[call-overload]
+    except TypeError:
+        # A prefix in the message to disambiguate between the train- and (optional) val dataloader that .fit() accepts
+        prefix = "train_" if trainer_fn == TrainerFn.FITTING else ""
+        if source.is_module():
+            if not is_overridden(source.name, source.instance):
+                raise TypeError(
+                    f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`."
+                    f" Found {dataloader}."
+                    f" Either pass the dataloader to the `.{trainer_fn}()` method OR implement"
+                    f" `def {source.name}(self):` in your LightningModule/LightningDataModule."
+                )
+            raise TypeError(
+                f"An invalid dataloader was returned from `{type(source.instance).__name__}.{source.name}()`."
+                f" Found {dataloader}."
+            )
+        raise TypeError(
+            f"An invalid dataloader was passed to `Trainer.{trainer_fn}({prefix}dataloaders=...)`. Found {dataloader}."
         )
path: src/lightning/pytorch/trainer/connectors/data_connector.py
text: @user1:
Nit: You could avoid the extra indentation by doing

```py
if not source.is_module():
    raise ...
if not is_overridden()
    raise ...
raise ...
```

diff_hunk: @@ -614,8 +614,34 @@ def test_attach_data_input_validation_with_none_dataloader(trainer_fn_name, data
     datamodule.test_dataloader = None
     datamodule.predict_dataloader = None
 
-    with pytest.raises(ValueError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
+    with pytest.raises(TypeError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
         trainer_fn(model, **{dataloader_name: None}, datamodule=datamodule)
 
-    with pytest.raises(ValueError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
+    with pytest.raises(TypeError, match=f"An invalid .*dataloader was passed to `Trainer.{trainer_fn_name}"):
         trainer_fn(model, **{dataloader_name: None}, datamodule=None)
+
+
+@pytest.mark.parametrize(
+    "trainer_fn_name, dataloader_name, stage",
+    [
+        ("fit", "train_dataloaders", RunningStage.TRAINING),
+        ("validate", "dataloaders", RunningStage.VALIDATING),
+        ("test", "dataloaders", RunningStage.TESTING),
+        ("predict", "dataloaders", RunningStage.PREDICTING),
+    ],
+)
+@pytest.mark.parametrize("dataloader", [object(), [1, object()]])
path: tests/tests_pytorch/trainer/connectors/test_data_connector.py
text: @user2:
```suggestion
@user1.mark.parametrize("dataloader", [None, object(), [1, object()]])
```