Merge pull request #23120 from mvdbeek/tool-data-followup

Relativize bundle paths from the known job output root

Author: mvdbeek

lib/galaxy/tool_shed/galaxy_install/client.py

@@ -1,4 +1,5 @@
 import threading
+from collections.abc import Mapping
 from typing import (
     Any,
     runtime_checkable,
@@ -27,7 +28,9 @@ class DataManagerInterface(Protocol):
 
     def process_result(self, out_data): ...
 
-    def write_bundle(self, out: dict[str, OutputDataset]) -> dict[str, OutputDataset]: ...
+    def write_bundle(
+        self, out: dict[str, OutputDataset], source_extra_files_paths: Mapping[str, str] | None = None
+    ) -> dict[str, OutputDataset]: ...
 
 
 class DataManagersInterface(Protocol):

lib/galaxy/tool_shed/unittest_utils/__init__.py

@@ -144,7 +144,7 @@ class DummyDataManager(DataManagerInterface):
     def process_result(self, out_data):
         return None
 
-    def write_bundle(self, out) -> dict[str, OutputDataset]:
+    def write_bundle(self, out, source_extra_files_paths=None) -> dict[str, OutputDataset]:
         return {}
 
 

lib/galaxy/tool_util/data/__init__.py

@@ -17,6 +17,7 @@
 from collections.abc import (
     Callable,
     Iterator,
+    Mapping,
 )
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -1341,6 +1342,7 @@ def write_bundle(
         out_data: dict[str, OutputDataset],
         bundle_description: DataTableBundleProcessorDescription,
         repo_info: RepoInfo | None,
+        source_extra_files_paths: Mapping[str, str] | None = None,
     ) -> dict[str, OutputDataset]:
         """Writes bundle and returns bundle path."""
         data_manager_dict = _data_manager_dict(out_data, ensure_single_output=True)
@@ -1350,8 +1352,9 @@ def write_bundle(
                 continue
 
             extra_files_path = dataset.extra_files_path
+            source_extra_files_path = (source_extra_files_paths or {}).get(output_name, extra_files_path)
             _relativize_bundle_data_table_paths(
-                data_manager_dict.get("data_tables", {}), extra_files_path, bundle_description
+                data_manager_dict.get("data_tables", {}), source_extra_files_path, bundle_description
             )
             bundle = DataTableBundle(
                 data_tables=data_manager_dict.get("data_tables", {}),
@@ -1398,13 +1401,13 @@ def _relativize_bundle_data_table_paths(
     extra_files_path: str,
     bundle_description: DataTableBundleProcessorDescription,
 ) -> None:
-    """Rewrite absolute path values under ``extra_files_path`` to be relative to it.
+    """Rewrite absolute path values under the compute-side extra-files dir.
 
     Some data managers record the absolute path inside the (transient) job
     directory (e.g. MetaPhlAn's ``$out_file.extra_files_path/$index``) rather
     than the expected relative path; that absolute path no longer resolves once
     the bundle is staged elsewhere, so imports and chained data managers fail.
-    Paths already relative, or outside ``extra_files_path``, are left untouched.
+    Paths already relative, or outside that directory, are left untouched.
     """
     for data_table_name, data_table_values in data_tables.items():
         path_headers = get_path_headers(bundle_description, data_table_name)
@@ -1441,9 +1444,6 @@ def _data_manager_dict(out_data: dict[str, OutputDataset], ensure_single_output:
     return data_manager_dict
 
 
-from collections.abc import Mapping
-
-
 def _process_bundle(
     out_data: Mapping[str, HasExtraFiles],
     bundle: DataTableBundle,

lib/galaxy/tools/__init__.py

@@ -43,6 +43,7 @@
     ToolInputsNotReadyException,
 )
 from galaxy.job_execution import output_collect
+from galaxy.job_execution.compute_environment import dataset_path_to_extra_path
 from galaxy.job_execution.output_collect import (
     BaseJobContext,
     MetadataSourceProvider,
@@ -202,6 +203,7 @@
 from galaxy.tools.parameters.workflow_utils import workflow_building_modes
 from galaxy.tools.parameters.wrapped_json import json_wrap
 from galaxy.util import (
+    asbool,
     in_directory,
     Params,
     parse_xml_string,
@@ -3752,6 +3754,36 @@ def job_failed(self, job_wrapper, message, exception=False):
         self.__remove_interactivetool_by_job(job)
 
 
+def _data_manager_bundle_source_extra_files_paths(app, job, out_data: dict[str, Any]) -> dict[str, str]:
+    """Return the compute-side extra-files roots for bundle outputs.
+
+    Data managers running with ``outputs_to_working_directory`` receive an
+    extra-files path derived from the job working directory, not the path the
+    handler later obtains from the object store. Reconstruct that same path
+    here, using the rules in ``OutputsToWorkingDirectoryPathRewriter``.
+    """
+    outputs_to_working_directory = asbool(
+        job.get_destination_configuration({}, app.config, "outputs_to_working_directory", False)
+    )
+    if not outputs_to_working_directory:
+        return {}
+
+    job_working_directory = app.object_store.get_filename(job, base_dir="job_work", dir_only=True, obj_dir=True)
+    outputs_directory = None if Version(job.galaxy_version or "19.05") < Version("20.01") else "outputs"
+    if outputs_directory is not None:
+        job_working_directory = os.path.join(job_working_directory, outputs_directory)
+
+    source_extra_files_paths = {}
+    for output_name, dataset in out_data.items():
+        if dataset.ext != "data_manager_json":
+            continue
+        hda = cast(HistoryDatasetAssociation, dataset)
+        assert hda.dataset is not None
+        output_path = os.path.join(job_working_directory, f"dataset_{hda.dataset.uuid}.dat")
+        source_extra_files_paths[output_name] = dataset_path_to_extra_path(output_path)
+    return source_extra_files_paths
+
+
 class DataManagerTool(OutputParameterJSONTool):
     tool_type = "manage_data"
     default_tool_action = DataManagerToolAction
@@ -3782,7 +3814,8 @@ def exec_after_process(self, app, inp_data, out_data, param_dict, job, final_job
         elif data_manager_mode == "dry_run":
             pass
         elif data_manager_mode == "bundle":
-            for bundle_path, dataset in data_manager.write_bundle(out_data).items():
+            source_extra_files_paths = _data_manager_bundle_source_extra_files_paths(app, job, out_data)
+            for bundle_path, dataset in data_manager.write_bundle(out_data, source_extra_files_paths).items():
                 hda = cast(HistoryDatasetAssociation, dataset)
                 assert hda.dataset is not None
                 assert hda.dataset.object_store is not None

lib/galaxy/tools/data_manager/manager.py

@@ -1,6 +1,7 @@
 import errno
 import logging
 import os
+from collections.abc import Mapping
 from typing import (
     Optional,
 )
@@ -242,12 +243,14 @@ def process_result(self, out_data: dict[str, OutputDataset]) -> None:
     def write_bundle(
         self,
         out_data: dict[str, OutputDataset],
+        source_extra_files_paths: Mapping[str, str] | None = None,
     ) -> dict[str, OutputDataset]:
         tool_data_tables = self.data_managers.app.tool_data_tables
         return tool_data_tables.write_bundle(
             out_data,
             self.processor_description,
             self.repo_info,
+            source_extra_files_paths,
         )
 
     @property

test/unit/tool_util/data/test_tool_data_bundles.py

@@ -309,6 +309,150 @@ def test_import_bundle_with_absolute_recorded_path(tdt_manager, tmp_path, table_
     assert os.path.exists(os.path.join(loc_target, f"{index}.pkl"))
 
 
+def prepare_split_prefix_output_and_description(
+    tmp_path,
+    table_name="testalpha",
+    path_column="path",
+    store_leaf=None,
+    compute_leaf=None,
+    recorded_relative_path=None,
+):
+    """Model the object-store vs job-working-dir prefix split.
+
+    In production ``write_bundle`` runs on the job handler, where
+    ``dataset.extra_files_path`` resolves to the object-store extra-files dir,
+    while the data manager baked its absolute ``path`` on the compute node under
+    the transient job working dir. The directories differ in every prefix
+    component, so a plain ``os.path.relpath`` escapes with ``..``. They may even
+    differ in the ``dataset_<key>_files`` leaf itself: the compute side is
+    uuid-keyed under outputs_to_working_directory while the store may be
+    id-keyed — pass distinct ``store_leaf``/``compute_leaf`` to model that. The
+    recorded absolute path deliberately does NOT exist under
+    ``extra_files_path`` here, so the fix must be structural (no existence check).
+    """
+    index = "mpa_toy"
+    compute_leaf = compute_leaf or "dataset_00000000-0000-0000-0000-000000000000_files"
+    store_leaf = store_leaf or compute_leaf
+
+    # Object-store extra-files dir: where the bundle index actually gets written.
+    extra_files_path = tmp_path / "objectstore" / "000" / store_leaf
+    extra_files_path.mkdir(parents=True)
+
+    # Absolute path as recorded by the DM under the (now-irrelevant) job working dir.
+    job_dir = tmp_path / "jobs" / "001" / "outputs" / compute_leaf
+    recorded_path = str(job_dir / (index if recorded_relative_path is None else recorded_relative_path))
+
+    output = {"data_tables": {table_name: [{"value": index, "name": "toy", path_column: recorded_path}]}}
+    output_dataset_path = tmp_path / "output.dat"
+    output_dataset_path.write_text(json.dumps(output))
+    output_dataset = OutputDataset(output_dataset_path, extra_files_path)
+    out_data = {"out1": output_dataset}
+    data_table = {
+        "name": table_name,
+        "output": {
+            "columns": [
+                {"name": "value"},
+                {"name": "name"},
+                {
+                    "name": path_column,
+                    "output_ref": "out1",
+                    "moves": [
+                        {
+                            "type": "directory",
+                            "relativize_symlinks": False,
+                            "source_value": "${" + path_column + "}",
+                            "target_value": "metaphlan/data/${value}",
+                            "target_base": "${GALAXY_DATA_MANAGER_DATA_PATH}",
+                        }
+                    ],
+                    "value_translations": [
+                        {"value": "${GALAXY_DATA_MANAGER_DATA_PATH}/metaphlan/data/${value}", "type": "template"},
+                        {"value": "abspath", "type": "function"},
+                    ],
+                },
+            ]
+        },
+    }
+    process_description = DataTableBundleProcessorDescription(
+        **{"undeclared_tables": False, "data_tables": [data_table]}
+    )
+    return index, extra_files_path, job_dir, out_data, process_description
+
+
+def test_write_bundle_relativizes_split_prefix_path_from_compute_root(tdt_manager, tmp_path):
+    """A data manager's compute-side path is made relative to its known job-dir root."""
+    index, extra_files_path, compute_extra_files_path, out_data, process_description = (
+        prepare_split_prefix_output_and_description(tmp_path)
+    )
+    tdt_manager.write_bundle(
+        out_data,
+        process_description,
+        None,
+        source_extra_files_paths={"out1": str(compute_extra_files_path)},
+    )
+
+    bundle_index = json.loads((extra_files_path / BUNDLE_INDEX_FILE_NAME).read_text())
+    stored_path = bundle_index["data_tables"]["testalpha"][0]["path"]
+    assert stored_path == index
+    assert not os.path.isabs(stored_path)
+
+
+def test_write_bundle_relativizes_uuid_keyed_job_path_on_id_keyed_store(tdt_manager, tmp_path):
+    """A uuid-keyed compute root works even when the object store is id-keyed."""
+    index, extra_files_path, compute_extra_files_path, out_data, process_description = (
+        prepare_split_prefix_output_and_description(
+            tmp_path,
+            store_leaf="dataset_42_files",
+            compute_leaf="dataset_00000000-0000-0000-0000-000000000000_files",
+        )
+    )
+    tdt_manager.write_bundle(
+        out_data,
+        process_description,
+        None,
+        source_extra_files_paths={"out1": str(compute_extra_files_path)},
+    )
+
+    bundle_index = json.loads((extra_files_path / BUNDLE_INDEX_FILE_NAME).read_text())
+    stored_path = bundle_index["data_tables"]["testalpha"][0]["path"]
+    assert stored_path == index
+    assert not os.path.isabs(stored_path)
+
+
+def test_write_bundle_relativizes_compute_extra_files_root(tdt_manager, tmp_path):
+    _, extra_files_path, compute_extra_files_path, out_data, process_description = (
+        prepare_split_prefix_output_and_description(tmp_path, recorded_relative_path="")
+    )
+    tdt_manager.write_bundle(
+        out_data,
+        process_description,
+        None,
+        source_extra_files_paths={"out1": str(compute_extra_files_path)},
+    )
+
+    bundle_index = json.loads((extra_files_path / BUNDLE_INDEX_FILE_NAME).read_text())
+    assert bundle_index["data_tables"]["testalpha"][0]["path"] == os.curdir
+
+
+def test_write_bundle_preserves_nested_dataset_named_directory(tdt_manager, tmp_path):
+    store_leaf = "dataset_42_files"
+    relative_path = os.path.join("mpa_toy", store_leaf, "index")
+    _, extra_files_path, compute_extra_files_path, out_data, process_description = (
+        prepare_split_prefix_output_and_description(
+            tmp_path, store_leaf=store_leaf, recorded_relative_path=relative_path
+        )
+    )
+    tdt_manager.write_bundle(
+        out_data,
+        process_description,
+        None,
+        source_extra_files_paths={"out1": str(compute_extra_files_path)},
+    )
+
+    bundle_index = json.loads((extra_files_path / BUNDLE_INDEX_FILE_NAME).read_text())
+    assert bundle_index["data_tables"]["testalpha"][0]["path"] == relative_path
+
+
 def test_path_headers_from_move_and_abspath():
     """A column is a path column when it has a move or an abspath translation, whatever its name."""
     description = DataTableBundleProcessorDescription(