Singleton splitter 1D split bug in state.py:_single_op_splits

[ibevers]

For the Bug Report,
Include this information:

• What version of Pydra are you using?
  0.23

• What were you trying to do?
  Split over a list of paths outputted by a task in another task.

• What did you expect will happen?
  The workflow would complete successfully.

• What actually happened?
  I got stopped at a PDB breakpoint (weird in production code). I continued, and I got this error:

Traceback (most recent call last):
  File "/Users/isaacbevers/sensein/b2ai-wrapper/b2aiprep/src/b2aiprep/summer_school_data_optimized.py", line 320, in <module>
    main()
  File "/Users/isaacbevers/sensein/b2ai-wrapper/b2aiprep/src/b2aiprep/summer_school_data_optimized.py", line 309, in main
    print(extract_features_workflow(args.bids_files_path, remove=False))
  File "/Users/isaacbevers/sensein/b2ai-wrapper/b2aiprep/src/b2aiprep/summer_school_data_optimized.py", line 215, in extract_features_workflow
    run(ef_wf)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/submitter.py", line 55, in __call__
    self.loop.run_until_complete(
  File "/Library/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/submitter.py", line 84, in submit_from_call
    await runnable._run(self, rerun=rerun)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/core.py", line 1310, in _run
    await self._run_task(submitter, rerun=rerun)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/core.py", line 1339, in _run_task
    await submitter.expand_workflow(self, rerun=rerun)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/submitter.py", line 242, in expand_workflow
    for fut in await self.expand_runnable(task, rerun=rerun):
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/submitter.py", line 129, in expand_runnable
    task_pkl = await prepare_runnable_with_state(runnable)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/submitter.py", line 342, in prepare_runnable_with_state
    runnable.state.prepare_states(runnable.inputs, cont_dim=runnable.cont_dim)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/state.py", line 778, in prepare_states
    self.prepare_states_ind()
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/state.py", line 807, in prepare_states_ind
    values_out_pr, keys_out_pr = self.splits(
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/state.py", line 980, in splits
    return self._single_op_splits(op_single)
  File "/Users/isaacbevers/sensein/pydra/pydra/engine/state.py", line 1080, in _single_op_splits
    inner_len = [shape[-1]] * reduce(lambda x, y: x * y, shape[:-1])
TypeError: reduce() of empty iterable with no initial value

• Can you replicate the behavior? If yes, how?
  Yes, I can run my workflow again with the current release of Pydra. I also fixed it 758.

• List the steps you performed that revealed the bug to you.
  I read the source code for _single_op_splits(op_single) in /Users/isaacbevers/sensein/pydra/pydra/engine/state.py. Then I removed the breakpoint, and updated /Users/isaacbevers/sensein/pydra/pydra/engine/state.py. After doing so, my code ran as expected, and the same tests passed as when I ran the current distribution.

• Include any code samples.
  Here is the workflow that was problematic:

    ef_wf = pydra.Workflow(
        name="ef_wf",
        input_spec=["bids_dir_path"],
        bids_dir_path=bids_dir_path
    )

    # Get subject paths.
    ef_wf.add(get_dir_paths(
            name="subject_paths",
            dir_path=ef_wf.lzin.bids_dir_path,
            id_prefix=SUBJECT_ID
        )
    )

    # Get session paths for each subject path.
    ef_wf.add(get_dir_paths(
            name="session_paths",
            dir_path=ef_wf.subject_paths.lzout.out,
            id_prefix=SESSION_ID
        ).split(
            "dir_path",
            dir_path=ef_wf.subject_paths.lzout.out
        )
    )

    # Get audio file paths for each session path.
    ef_wf.add(get_audio_files(
            name="audio_files",
            dir_path=ef_wf.session_paths.lzout.out,
            file_extension=AUDIO_FILE_EXTENSION
        ).split(
            "dir_path",
            dir_path=ef_wf.session_paths.lzout.out
        )
    )

Note that the following versions did not cause any errors:

    ef_wf = pydra.Workflow(
        name="ef_wf",
        input_spec=["bids_dir_path"],
        bids_dir_path=bids_dir_path
    )

    # Get subject paths.
    ef_wf.add(get_dir_paths(
            name="subject_paths",
            dir_path=ef_wf.lzin.bids_dir_path,
            id_prefix=SUBJECT_ID
        )
    )

    # Get session paths for each subject path.
    ef_wf.add(get_dir_paths(
            name="session_paths",
            dir_path=ef_wf.subject_paths.lzout.out,
            id_prefix=SESSION_ID
        ).split(
            "dir_path",
            dir_path=ef_wf.subject_paths.lzout.out
        )
    )

    ef_wf = pydra.Workflow(
        name="ef_wf",
        input_spec=["bids_dir_path"],
        bids_dir_path=bids_dir_path
    )

    # Get subject paths.
    ef_wf.add(get_dir_paths(
            name="subject_paths",
            dir_path=ef_wf.lzin.bids_dir_path,
            id_prefix=SUBJECT_ID
        )
    )

    # Get session paths for each subject path.
    ef_wf.add(get_dir_paths(
            name="session_paths",
            dir_path=ef_wf.subject_paths.lzout.out,
            id_prefix=SESSION_ID
        ).split(
            "dir_path",
            dir_path=ef_wf.subject_paths.lzout.out
        )
    )

    # Get audio file paths for each session path.
    ef_wf.add(get_audio_files(
            name="audio_files",
            dir_path=ef_wf.session_paths.lzout.out,
            file_extension=AUDIO_FILE_EXTENSION
        )

[tclose]

I believe this particular line of code has already been updated, but just in case all reduce calls in State methods have been given an initial value of 1 in #818