Engine: fix bug when caching from process with nested outputs (#5538)

When the execution of a process is being skipped because it has a valid
cache source and caching is enabled, the process would except if the
process has nested output namespaces. Since the nesting cannot be
literally preserved in the database, the nesting is representing by
using double underscores between each level of nesting in the link
label.

The problem occurs when `Process._create_and_setup_db_record` would be
called which would create the outputs of the cloned node, however, it
would use these "collapsed" link labels containing the double
underscores. Passing these to `self.out` would cause the validation to
fail because the process port validation expects the actual nesting.

The solution is to restore the database nesting representation to the
one expected by `Process.out`, which amounts to replacing the namespace
separators.

aiida/engine/processes/process.py

@@ -582,11 +582,13 @@ def _create_and_setup_db_record(self) -> Union[int, UUID]:
                     for entry in self.node.base.links.get_outgoing(link_type=LinkType.RETURN):
                         if entry.link_label.endswith(f'_{entry.node.pk}'):
                             continue
-                        self.out(entry.link_label, entry.node)
+                        label = entry.link_label.replace(PORT_NAMESPACE_SEPARATOR, self.spec().namespace_separator)
+                        self.out(label, entry.node)
                     # This is needed for CalcJob. In that case, the outputs are
                     # returned regardless of whether they end in '_pk'
                     for entry in self.node.base.links.get_outgoing(link_type=LinkType.CREATE):
-                        self.out(entry.link_label, entry.node)
+                        label = entry.link_label.replace(PORT_NAMESPACE_SEPARATOR, self.spec().namespace_separator)
+                        self.out(label, entry.node)
             except exceptions.ModificationNotAllowed:
                 # The calculation was already stored
                 pass

tests/engine/processes/test_caching.py

@@ -0,0 +1,37 @@
+# -*- coding: utf-8 -*-
+"""Test the caching functionality for a :class:`aiida.engine.processes.process.Process`."""
+from aiida.engine import Process, run
+from aiida.manage import enable_caching
+from aiida.orm import CalcJobNode, Int
+
+
+class NestedOutputsProcess(Process):
+    """Process with dynamic nested output namespace."""
+
+    _node_class = CalcJobNode
+
+    @classmethod
+    def define(cls, spec):
+        super().define(spec)
+        spec.input('a')
+        spec.output_namespace('nested', dynamic=True)
+
+    def run(self):
+        self.out('nested', {'a': self.inputs.a + 2})
+
+
+def test_caching_nested_output_namespace():
+    """Test that caching from a process with a nested output namespace works."""
+    _, node_original = run.get_node(NestedOutputsProcess, a=Int(1))
+    assert not node_original.is_created_from_cache
+
+    with enable_caching():
+        _, node_clone = run.get_node(NestedOutputsProcess, a=Int(1))
+
+    assert node_clone.is_created_from_cache
+    assert node_clone.get_cache_source() == node_original.uuid
+
+    outputs = node_clone.get_outgoing().nested()
+    assert list(outputs.keys()) == ['nested']
+    assert list(outputs['nested'].keys()) == ['a']
+    assert isinstance(outputs['nested']['a'], Int)