Function parse_create_dataflow

src/clusterd-test-driver/src/text.rs:375–424

Parse a `create-dataflow` body of `import`/`build`/`export` sub-commands. The directive's bare flags carry the dataflow-level options (`optimize`).

(
    args: &BTreeMap<String, String>,
    flags: &[String],
    body: &[Line],
)

/// Parse a `create-dataflow` body of `import`/`build`/`export` sub-commands. The
/// directive's bare flags carry the dataflow-level options (`optimize`).
fn parse_create_dataflow(
    args: &BTreeMap<String, String>,
    flags: &[String],
    body: &[Line],
) -> anyhow::Result<Command> {
    let name = opt_string(args, "name");
    let as_of = req_u64(args, "as-of")?;
    let optimize = flags.iter().any(|f| f == "optimize");
    let mut imports = Vec::new();
    let mut builds = Vec::new();
    let mut exports = Vec::new();
    for (header, sub_body) in group(body)? {
        let (verb, args, _flags) = parse_header(&header.text)?;
        match verb.as_str() {
            "import" => {
                if let Some(index_id) = args.get("index") {
                    imports.push(ImportSpec::Index {
                        index_id: index_id
                            .parse()
                            .with_context(|| format!("bad index id `{index_id}`"))?,
                    });
                } else {
                    imports.push(ImportSpec::Source {
                        id: req_u64(&args, "source")?,
                        shard: req(&args, "shard")?.to_string(),
                        schema: opt_string(&args, "schema"),
                        upper: req_u64(&args, "upper")?,
                    });
                }
            }
            "build" => {
                ensure!(!sub_body.is_empty(), "`build` needs a MIR body");
                builds.push(BuildSpec {
                    id: req_u64(&args, "id")?,
                    expr: body_text(sub_body),
                });
            }
            "export" => exports.push(parse_export(&args)?),
            other => bail!("unknown `create-dataflow` sub-command `{other}`"),
        }
    }
    Ok(Command::CreateDataflow {
        name,
        imports,
        builds,
        exports,
        as_of,
        optimize,
    })
}

/// Parse one command block (directive line plus indentation-structured body).
fn parse_command(input: &str) -> anyhow::Result<Command> {