MemoryDataset: _EMPTY sentinel does not survive pickling, so load() returns a bare object() instead of raising DatasetError

Description

MemoryDataset uses a bare object() sentinel to mean "nothing has been saved yet" (kedro/io/memory_dataset.py:10, _EMPTY = object()). Pickle cannot preserve the identity of a bare object(), so once a MemoryDataset — or a catalogue holding one — has been pickled and unpickled, the is _EMPTY identity checks all fail and the dataset silently misreports its own state.

Steps to Reproduce

import pickle

[repro_memorydataset_sentinel.py](https://github.com/user-attachments/files/32282584/repro_memorydataset_sentinel.py)

from kedro.io import MemoryDataset

# the sentinel itself
from kedro.io.memory_dataset import _EMPTY
print(_EMPTY is pickle.loads(pickle.dumps(_EMPTY)))
# False

# an unsaved dataset, before and after a pickle round trip
MemoryDataset().load()
# DatasetError: Data for MemoryDataset has not been saved yet.

pickle.loads(pickle.dumps(MemoryDataset())).load()
# <object object at 0x...>        <-- no exception, a bare object() is returned

A pickle round trip is not an exotic path. It is what happens to the data catalogue whenever a runner serialises it into another process or another machine — including the DaskRunner recipe in the Dask deployment guide, which does client.submit(DaskRunner._run_node, node, catalog, ...).

Expected Result

An unsaved MemoryDataset should raise DatasetError regardless of whether it has been pickled, and _exists() should return False.

Actual Result

Three observable consequences, all from the same root cause. Measured, not inferred:

fresh     = MemoryDataset()
roundtrip = pickle.loads(pickle.dumps(MemoryDataset()))

fresh._exists()        # 0
roundtrip._exists()    # True

repr(fresh)            # kedro.io.memory_dataset.MemoryDataset()
repr(roundtrip)        # kedro.io.memory_dataset.MemoryDataset(data='<object>')

fresh.load()           # DatasetError: Data for MemoryDataset has not been saved yet.
roundtrip.load()       # <object object at 0x...>

So load() stops raising, _exists() starts reporting a dataset that was never written, and __repr__ presents the sentinel as real data. The second row matters beyond MemoryDataset itself: logic that asks the catalogue whether a dataset exists, such as the missing-output handling in the runner, will be told a dataset is present when nothing has been written.

Context

I hit this while working on the Dask deployment guide. The documented custom runner ships the catalogue to the workers, so a downstream node in another worker holds an unpickled catalogue. Instead of a clear DatasetError saying the upstream node's output was never saved, the node receives a bare object() and carries on with it. It took a while to trace, because the failure looks like a data problem rather than a serialisation problem.

I have not checked whether any test constructs a MemoryDataset through pickle; if not, that would explain why this has gone unnoticed.

Possible fix

Make the sentinel a singleton that pickles back to itself, for example:

class _Empty:
    __slots__ = ()

    def __reduce__(self):
        return (_get_empty, ())


def _get_empty():
    return _EMPTY


_EMPTY = _Empty()

__reduce__ makes every unpickled copy resolve to the module-level _EMPTY, so the is comparisons keep working and no call site needs to change. A module-level Enum member would work equally well.

Happy to open a PR with this plus a regression test that round-trips a MemoryDataset and a catalogue through pickle, if you are happy with the approach.