diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index e9d336bbcf6..b699558cae5 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -3181,6 +3181,25 @@ class HistoryDatasetAssociation(DatasetInstance, HasTags, Dictifiable, UsesAnnot
             self.version = self.version + 1 if self.version else 1
             session.add(past_hda)
 
+    def copy_from(self, other_hda):
+        # This deletes the old dataset, so make sure to only call this on new things
+        # in the history (e.g. during job finishing).
+        old_dataset = self.dataset
+        self._metadata = None
+        self.metadata = other_hda.metadata
+        self.info = other_hda.info
+        self.blurb = other_hda.blurb
+        self.peek = other_hda.peek
+        self.extension = other_hda.extension
+        self.designation = other_hda.designation
+        self.deleted = other_hda.deleted
+        self.visible = other_hda.visible
+        self.validated_state = other_hda.validated_state
+        self.validated_state_message = other_hda.validated_state_message
+        self.copy_tags_from(self.history.user, other_hda)
+        self.dataset = other_hda.dataset
+        old_dataset.full_delete()
+
     def copy(self, parent_id=None, copy_tags=None, flush=True, copy_hid=True, new_name=None):
         """
         Create a copy of this HDA.
diff --git a/lib/galaxy/tool_util/parser/output_objects.py b/lib/galaxy/tool_util/parser/output_objects.py
index f57d04f6abf..679cf947592 100644
--- a/lib/galaxy/tool_util/parser/output_objects.py
+++ b/lib/galaxy/tool_util/parser/output_objects.py
@@ -7,13 +7,14 @@ from .output_collection_def import dataset_collector_descriptions_from_output_di
 
 class ToolOutputBase(Dictifiable):
 
-    def __init__(self, name, label=None, filters=None, hidden=False):
+    def __init__(self, name, label=None, filters=None, hidden=False, from_expression=None):
         super().__init__()
         self.name = name
         self.label = label
         self.filters = filters or []
         self.hidden = hidden
         self.collection = False
+        self.from_expression = from_expression
 
     def to_dict(self, view='collection', value_mapper=None, app=None):
         return super().to_dict(view=view, value_mapper=value_mapper)
@@ -32,8 +33,8 @@ class ToolOutput(ToolOutputBase):
 
     def __init__(self, name, format=None, format_source=None, metadata_source=None,
                  parent=None, label=None, filters=None, actions=None, hidden=False,
-                 implicit=False):
-        super().__init__(name, label=label, filters=filters, hidden=hidden)
+                 implicit=False, from_expression=None):
+        super().__init__(name, label=label, filters=filters, hidden=hidden, from_expression=from_expression)
         self.output_type = "data"
         self.format = format
         self.format_source = format_source
diff --git a/lib/galaxy/tool_util/parser/xml.py b/lib/galaxy/tool_util/parser/xml.py
index 23665a131dd..2940b5e13bc 100644
--- a/lib/galaxy/tool_util/parser/xml.py
+++ b/lib/galaxy/tool_util/parser/xml.py
@@ -398,7 +398,8 @@ class XmlToolSource(ToolSource):
         default_metadata_source="",
         expression_type=None,
     ):
-        output = ToolOutput(data_elem.get("name"))
+        from_expression = data_elem.get("from")
+        output = ToolOutput(data_elem.get("name"), from_expression=from_expression)
         output_format = data_elem.get("format", default_format)
         auto_format = string_as_bool(data_elem.get("auto_format", "false"))
         if auto_format and output_format != "data":
diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py
index 694d6720659..43cdda9ee6e 100644
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -2483,6 +2483,22 @@ class ExpressionTool(Tool):
         with open(expression_inputs_path, "w") as f:
             json.dump(expression_inputs, f)
 
+    def exec_after_process(self, app, inp_data, out_data, param_dict, job=None, **kwds):
+        for key, val in self.outputs.items():
+            if val.output_type == "data":
+                with open(out_data[key].file_name, "r") as f:
+                    src = json.load(f)
+                assert isinstance(src, dict)
+                dataset_id = src["id"]
+                copy_object = None
+                for input_dataset in inp_data.values():
+                    if input_dataset.id == dataset_id:
+                        copy_object = input_dataset
+                        break
+                if copy_object is None:
+                    raise Exception("Failed to find dataset output.")
+                out_data[key].copy_from(copy_object)
+
     def parse_environment_variables(self, tool_source):
         """ Setup environment variable for inputs file.
         """
diff --git a/lib/galaxy/tools/parameters/wrapped_json.py b/lib/galaxy/tools/parameters/wrapped_json.py
index 70120ed8d1f..9ae1882e751 100644
--- a/lib/galaxy/tools/parameters/wrapped_json.py
+++ b/lib/galaxy/tools/parameters/wrapped_json.py
@@ -123,8 +123,10 @@ def _hda_to_object(hda):
 
     return {
         'file_ext': hda_dict['file_ext'],
+        'file_size': hda_dict['file_size'],
         'name': hda_dict['name'],
         'metadata': metadata_dict,
+        'src': {'src': 'hda', 'id': hda.id},
     }
 
 
diff --git a/test/functional/tools/expression_pick_larger_file.xml b/test/functional/tools/expression_pick_larger_file.xml
new file mode 100644
index 00000000000..eca011bb213
--- /dev/null
+++ b/test/functional/tools/expression_pick_larger_file.xml
@@ -0,0 +1,44 @@
+<tool name="expression_pick_larger_file" id="expression_pick_larger_file"
+      version="0.1.0" tool_type="expression">
+    <expression type="ecma5.1"><![CDATA[
+        {
+            var input1 = $job.input1;
+            var input2 = $job.input2;
+            var input1_size = input1 && input1.file_size;
+            var input2_size = input2 && input2.file_size;
+            var output;
+            if( !input2 || !input2_size || input1_size >= input2_size ) {
+                output = input1.src;
+            } else {
+                output = input2.src;
+            }
+            return {'output': output};
+        }
+    ]]></expression>
+    <inputs>
+        <param type="data" label="First file." optional="true" name="input1" />
+        <param type="data" label="Second file." optional="true" name="input2" />
+    </inputs>
+    <outputs>
+        <output type="data" name="larger_file" from="output" />
+    </outputs>
+    <help>
+    </help>
+    <tests>
+        <test>
+            <param name="input1" value="simple_line.txt" />
+            <param name="input2" value="simple_line_alternative.txt" />
+            <output name="larger_file" file="simple_line_alternative.txt"/>
+        </test>
+        <test>
+            <param name="input1" value="simple_line_alternative.txt" />
+            <param name="input2" value_json="null" />
+            <output name="larger_file" file="simple_line_alternative.txt"/>
+        </test>
+        <test>
+            <param name="input1" value_json="null" />
+            <param name="input2" value_json="simple_line.txt" />
+            <output name="larger_file" file="simple_line.txt"/>
+        </test>
+    </tests>
+</tool>
diff --git a/test/functional/tools/samples_tool_conf.xml b/test/functional/tools/samples_tool_conf.xml
index 05a1ed6efd7..a634fef6194 100644
--- a/test/functional/tools/samples_tool_conf.xml
+++ b/test/functional/tools/samples_tool_conf.xml
@@ -171,6 +171,7 @@
   <tool file="collection_cat_group_tag_multiple.xml" />
   <tool file="discover_sort_by.xml" />
   <tool file="expression_forty_two.xml" />
+  <tool file="expression_pick_larger_file.xml" />
   <tool file="expression_parse_int.xml" />
   <tool file="expression_log_line_count.xml" />
   <tool file="expression_null_handling_boolean.xml" />
diff --git a/test/unit/tool_util/test_parsing.py b/test/unit/tool_util/test_parsing.py
index 44880871db4..6f02a7cd122 100644
--- a/test/unit/tool_util/test_parsing.py
+++ b/test/unit/tool_util/test_parsing.py
@@ -618,6 +618,18 @@ class ExpressionTestToolLoaderTestCase(BaseLoaderTestCase):
         assert output0['attributes']['object'] is None
 
 
+class ExpressionOutputDataToolLoaderTestCase(BaseLoaderTestCase):
+    source_file_name = os.path.join(galaxy_directory(), "test/functional/tools/expression_pick_larger_file.xml")
+    source_contents = None
+
+    def test_output_parsing(self):
+        outputs, _ = self._tool_source.parse_outputs(None)
+        assert 'larger_file' in outputs
+        tool_output = outputs['larger_file']
+        assert tool_output.format == "data"
+        assert tool_output.from_expression == "output"
+
+
 class SpecialToolLoaderTestCase(BaseLoaderTestCase):
     source_file_name = os.path.join(galaxy_directory(), "lib/galaxy/tools/imp_exp/exp_history_to_archive.xml")
     source_contents = None