make interface changes backwards compatible

src/coffea/dataset_tools/apply_processor.py

@@ -86,7 +86,11 @@ def apply_to_dataset(
     metadata: dict[Hashable, Any] = {},
     uproot_options: dict[str, Any] = {},
     parallelize_with_dask: bool = False,
-) -> DaskOutputType | tuple[DaskOutputType, dask_awkward.Array]:
+) -> (
+    DaskOutputType
+    | tuple[DaskOutputType, dask_awkward.Array]
+    | tuple[dask_awkward.Array, DaskOutputType, dask_awkward.Array]
+):
     """
     Apply the supplied function or processor to the supplied dataset.
     Parameters
@@ -103,6 +107,8 @@ def apply_to_dataset(
             Options to pass to uproot. Pass at least {"allow_read_errors_with_report": True} to turn on file access reports.
         parallelize_with_dask: bool, default False
             Create dask.delayed objects that will return the the computable dask collections for the analysis when computed.
+        return_events: bool, default True
+            Return the created events object, or not.
 
     Returns
     -------
@@ -165,7 +171,16 @@ def apply_to_fileset(
     uproot_options: dict[str, Any] = {},
     parallelize_with_dask: bool = False,
     scheduler: Callable | str | None = None,
-) -> dict[str, DaskOutputType] | tuple[dict[str, DaskOutputType], dask_awkward.Array]:
+    return_events: bool = False,
+) -> (
+    dict[str, DaskOutputType]
+    | tuple[dict[str, DaskOutputType], dict[str, dask_awkward.Array]]
+    | tuple[
+        dict[str, dask_awkward.Array],
+        dict[str, DaskOutputType],
+        dict[str, dask_awkward.Array],
+    ]
+):
     """
     Apply the supplied function or processor to the supplied fileset (set of datasets).
     Parameters
@@ -242,8 +257,8 @@ def apply_to_fileset(
             out[name] = out[name][0]
 
     if len(report) > 0:
-        return events, out, report
-    return events, out
+        return (events, out, report) if return_events else (out, report)
+    return (events, out) if return_events else out
 
 
 def save_taskgraph(filename, events, *data_products, optimize_graph=False):

tests/test_dataset_tools.py

@@ -216,6 +216,7 @@ def test_tuple_data_manipulation_output(allow_read_errors_with_report):
         _my_analysis_output_2,
         _runnable_result,
         uproot_options={"allow_read_errors_with_report": allow_read_errors_with_report},
+        return_events=True,
     )
 
     if allow_read_errors_with_report:
@@ -253,6 +254,7 @@ def test_tuple_data_manipulation_output(allow_read_errors_with_report):
         _my_analysis_output_3,
         _runnable_result,
         uproot_options={"allow_read_errors_with_report": allow_read_errors_with_report},
+        return_events=True,
     )
 
     if allow_read_errors_with_report:
@@ -301,6 +303,7 @@ def test_apply_to_fileset(proc_and_schema, delayed_taskgraph_calc):
             _runnable_result,
             schemaclass=schemaclass,
             parallelize_with_dask=delayed_taskgraph_calc,
+            return_events=True,
         )
         out = dask.compute(to_compute)[0]
 
@@ -314,6 +317,7 @@ def test_apply_to_fileset(proc_and_schema, delayed_taskgraph_calc):
             max_chunks(_runnable_result, 1),
             schemaclass=schemaclass,
             parallelize_with_dask=delayed_taskgraph_calc,
+            return_events=True,
         )
         out = dask.compute(to_compute)[0]
 
@@ -338,6 +342,7 @@ def test_apply_to_fileset_hinted_form():
             NanoEventsProcessor(),
             dataset_runnable,
             schemaclass=NanoAODSchema,
+            return_events=True,
         )
         out = dask.compute(to_compute)[0]
 
@@ -554,6 +559,7 @@ def test_recover_failed_chunks(delayed_taskgraph_calc):
             schemaclass=NanoAODSchema,
             uproot_options={"allow_read_errors_with_report": True},
             parallelize_with_dask=delayed_taskgraph_calc,
+            return_events=True,
         )
         out, reports = dask.compute(to_compute, reports)
 
@@ -597,6 +603,7 @@ def test_task_graph_serialization(proc_and_schema, with_report):
             schemaclass=schemaclass,
             parallelize_with_dask=False,
             uproot_options={"allow_read_errors_with_report": with_report},
+            return_events=True,
         )
 
         events = output[0]