Skip to content

Commit 90e684f

Browse files
committed
add store_inputs only if needed
1 parent f24cfca commit 90e684f

1 file changed

Lines changed: 32 additions & 11 deletions

File tree

src/jobflow/core/job.py

Lines changed: 32 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1359,7 +1359,7 @@ def prepare_replace(
13591359
"""
13601360
Prepare a replacement :obj:`Flow` or :obj:`Job`.
13611361
1362-
If the replacement is a ``Flow``, then an additional ``Job`` will be inserted
1362+
If the replacement is a ``Flow``, then an additional ``Job`` may be inserted
13631363
that maps the output id of the original job to outputs of the ``Flow``.
13641364
13651365
If the replacement is a ``Flow`` or a ``Job``, then this function pass on
@@ -1383,16 +1383,37 @@ def prepare_replace(
13831383
replace = Flow(jobs=replace)
13841384

13851385
if isinstance(replace, Flow) and replace.output is not None:
1386-
# add a job with same UUID as the current job to store the outputs of the
1387-
# flow; this job will inherit the metadata and output schema of the current
1388-
# job
1389-
store_output_job = store_inputs(replace.output)
1390-
store_output_job.set_uuid(current_job.uuid)
1391-
store_output_job.index = current_job.index + 1
1392-
store_output_job.metadata = current_job.metadata
1393-
store_output_job.output_schema = current_job.output_schema
1394-
store_output_job._kwargs = current_job._kwargs
1395-
replace.add_jobs(store_output_job)
1386+
leaf_nodes = [n for n, d in replace.graph.out_degree() if d == 0]
1387+
is_last_output_leaf = len(
1388+
leaf_nodes
1389+
) == 1 and replace.output == OutputReference(leaf_nodes[0])
1390+
if is_last_output_leaf:
1391+
# the last job of the replace inherits UUID and metadata from
1392+
# the original job
1393+
for j in replace.jobs:
1394+
if j.uuid == leaf_nodes[0]:
1395+
leaf_job = j
1396+
break
1397+
leaf_job.set_uuid(current_job.uuid)
1398+
leaf_job.index = current_job.index + 1
1399+
1400+
metadata = leaf_job.metadata
1401+
metadata.update(current_job.metadata)
1402+
leaf_job.metadata = metadata
1403+
1404+
if not leaf_job.output_schema:
1405+
leaf_job.output_schema = current_job.output_schema
1406+
else:
1407+
# add a job with same UUID as the current job to store the outputs of the
1408+
# flow; this job will inherit the metadata and output schema of the current
1409+
# job
1410+
store_output_job = store_inputs(replace.output)
1411+
store_output_job.set_uuid(current_job.uuid)
1412+
store_output_job.index = current_job.index + 1
1413+
store_output_job.metadata = current_job.metadata
1414+
store_output_job.output_schema = current_job.output_schema
1415+
store_output_job._kwargs = current_job._kwargs
1416+
replace.add_jobs(store_output_job)
13961417

13971418
elif isinstance(replace, Job):
13981419
# replace is a single Job

0 commit comments

Comments
 (0)