DAG-based mapping and execution

[WarmCyan]

Closes #29

The intent of this PR is to make curifactory primarily determine need-to-execute for each stage based on the DAG and where that stage's outputs are needed (based on cache status of later stage outputs and overwrite conditions etc.)

This allows skipping loading an artifact into memory if it won't ever be needed by an executed stage

[WarmCyan]

Unclear on best terminology for the "soft inputs" that aggregate stages can now take. Currently it's expected_state. Something like inputs would make it consistent with a regular stage, but since it's not actually handled the same way (since they're not directly passed in to the function), it seems odd to call it the same. Alternatively, we could make it so that 'inputs' really do get passed (as lists of values or as a list of tuples representing each 'call') into the function itself

[WarmCyan]

The basic concepts have been implemented - the idea is that the full run_experiment code is run in map_mode as a form of forward-pass, meaning every stage short-circuits before running or loading anything (essentially just collecting info about the stage and building up 'pseudo-records'). After this forward pass completes, the manager populates the DAG with a copy of these records and all of the pseudo artifacts (a modified ArtifactRepresentation which will load metadata if the underlying artifact was found/cached etc.)

The DAG then analyzes all the records and artifact metadata to gather execution trees for each leaf stage (stage whose outputs are not used as inputs in any other stage throughout the experiment), and then looks through each tree and determines which stages need to execute based on cache and overwrite status (specifying --overwrite-stage will now also automatically apply to impacted later stages.)

This allows looking at the "execution plan" of an experiment without actually running it, e.g. with the new --map-only flag on the CLI, which will tell you what values are cached, which run they were cached from, and the list of stages that will actually be executed.

An example is below, this is the newsgroups example experiment showing that a couple of the records' parametersets had been run previously (newsgroups_210 and newsgroups_212), so those values will be re-used from cache.

Farther up (this output will be cleaned up at some point), the full execution tree is shown, which is a rough tree-view of each stage and it's indented dependency stages. Following that is the execution list which is a list of tuples of combination record indices and stage names, representing the stages that should actually execute during the experiment run.

Something this should allow - we might be able to automatically unload values from state (conditioned on this not being in interactive mode, and prob some other flag that specifies no_unload or something - by checking the dag's is_output_used_anywhere at the end of a stage, and deleting from state if not.

An interesting side effect and improvement from adding the explicit expected_state is that stage maps now directly show which outputs go into an aggregate stage: (rather than previously, which would just show an arrow from the entire record to the aggregate stage.)

[jasonmhite]

Nathan and I discussed briefly at a high level and this approach seems sound to me.

[stewartsl]

I have no problems with the overall approach.

[mbadams5]

Looks good. Just a few minor comments/questions and some code suggestions.

[WarmCyan]

After discussion with Mark, we figured that a better way of handling expected_state for @aggregate would be to make it more consistent with @stage, naming it inputs and actually passing in the associated state where possible into the function itself as an argument.

The idea here would be to pass in the state objects via dictionaries that are keyed by the associated records in the passed records list. Records that don't have the specified name in state would simply not be part of the argument-specific dictionary.

As an example, where a previous aggregate stage might have looked like:

@aggregate(outputs=["final_results"])
def combine_results(record: Record, records: list[Record]):
    final_results = {}
    for record in records:
        if "results" in record.state:
            final_results[record.args.name] = record.state["results"]
    return final_results

could now become:

@aggregate(inputs=["results"], outputs=["final_results"])
def combine_results(record: Record, records: list[Record], results: dict[Record, float]):
    final_results = {}
    for record, result in results.items():
        final_results[record.args.name] = result
    return final_results

This eliminates the need to directly access state on records and simplifies having to check whether a given state has the artifact at all, while retaining the benefits above from adding in an expected_state/inputs