mz_clusterd_test_driver/
text.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! The text script format: a hand-writable, `datadriven`-style command file.
11//!
12//! A script is a sequence of stanzas, each a command and its expected output:
13//!
14//! ```text
15//! write-single-ts shard=data ts=0 count=5000
16//! ----
17//! wrote 5000
18//!
19//! count id=1001 ts=5
20//! ----
21//! 10000
22//! ```
23//!
24//! A stanza is a command (a directive line plus an optional indentation-structured
25//! body) up to a `----` separator, then the expected output up to a blank line. The
26//! `----` block is the assertion; `REWRITE=1` regenerates it (see [`crate::script`]).
27//! A `#` at column 0 is a comment; an indented `#0` is a column reference in MIR.
28//! Comments and blank lines are preserved across a rewrite.
29//!
30//! Command bodies are indentation-structured: `define-schema`/`write-rows`/`peek`
31//! carry rows or columns, and `define` carries `import`/`build`/`export`
32//! sub-commands, with a `build`'s MIR as its own deeper-indented sub-body.
33
34use std::collections::BTreeMap;
35
36use anyhow::{Context, anyhow, bail, ensure};
37use mz_repr::GlobalId;
38
39use crate::script::{BuildSpec, ColumnSpec, Command, ConfigSetting, ExportSpec, ImportSpec};
40
41/// One element of a parsed script file, retained so a `REWRITE` reproduces the
42/// file faithfully.
43pub enum Item {
44    /// A blank line or column-0 `#` comment, kept verbatim.
45    Verbatim(String),
46    /// A command and its expected output.
47    Stanza(Stanza),
48}
49
50/// A command stanza: the input block, the expected output, and the parsed command.
51pub struct Stanza {
52    /// The directive line plus body, verbatim (for rewrite).
53    pub input: String,
54    /// The expected output block (no trailing newline).
55    pub expected: String,
56    /// The parsed command.
57    pub command: Command,
58}
59
60/// Parse a script file into items: each command stanza plus the comments and blank
61/// lines around it.
62pub fn parse_file(content: &str) -> anyhow::Result<Vec<Item>> {
63    let lines: Vec<&str> = content.lines().collect();
64    let mut items = Vec::new();
65    let mut i = 0;
66    while i < lines.len() {
67        let line = lines[i];
68        // Blank lines and column-0 comments are preserved as-is.
69        if line.trim().is_empty() || line.starts_with('#') {
70            items.push(Item::Verbatim(line.to_string()));
71            i += 1;
72            continue;
73        }
74        // A stanza: slurp the input block up to the `----` separator.
75        let start = i;
76        while i < lines.len() && lines[i] != "----" {
77            i += 1;
78        }
79        ensure!(
80            i < lines.len(),
81            "stanza starting at line {} has no `----` separator",
82            start + 1
83        );
84        let input = lines[start..i].join("\n");
85        i += 1; // consume `----`
86        // The expected output runs to the next blank line (or end of file).
87        let exp_start = i;
88        while i < lines.len() && !lines[i].trim().is_empty() {
89            i += 1;
90        }
91        let expected = lines[exp_start..i].join("\n");
92        let command = parse_command(&input)
93            .with_context(|| format!("parsing stanza at line {}", start + 1))?;
94        items.push(Item::Stanza(Stanza {
95            input,
96            expected,
97            command,
98        }));
99    }
100    Ok(items)
101}
102
103/// Reproduce a script file with each stanza's expected output replaced by its
104/// actual output, for `REWRITE`. `actuals` has one entry per [`Item::Stanza`], in
105/// order.
106pub fn rewrite(items: &[Item], actuals: &[String]) -> String {
107    let mut out = String::new();
108    let mut next = 0;
109    for item in items {
110        match item {
111            Item::Verbatim(line) => {
112                out.push_str(line);
113                out.push('\n');
114            }
115            Item::Stanza(stanza) => {
116                let actual = &actuals[next];
117                next += 1;
118                out.push_str(&stanza.input);
119                out.push_str("\n----\n");
120                out.push_str(actual);
121                if !actual.is_empty() {
122                    out.push('\n');
123                }
124            }
125        }
126    }
127    out
128}
129
130/// A non-blank line of a command block, with its leading-space indentation.
131struct Line {
132    indent: usize,
133    text: String,
134}
135
136/// Split a command block into non-blank lines, recording each line's indentation.
137fn lex(block: &str) -> Vec<Line> {
138    block
139        .lines()
140        .filter_map(|raw| {
141            let trimmed = raw.trim_end();
142            if trimmed.trim().is_empty() {
143                return None;
144            }
145            let indent = trimmed.len() - trimmed.trim_start().len();
146            Some(Line {
147                indent,
148                text: trimmed.trim_start().to_string(),
149            })
150        })
151        .collect()
152}
153
154/// Split body lines into `(header, body)` groups: each header is a line at the
155/// minimum indentation, owning the following more-indented lines as its body.
156fn group(lines: &[Line]) -> anyhow::Result<Vec<(&Line, &[Line])>> {
157    if lines.is_empty() {
158        return Ok(vec![]);
159    }
160    let base = lines.iter().map(|l| l.indent).min().expect("non-empty");
161    let mut groups = Vec::new();
162    let mut i = 0;
163    while i < lines.len() {
164        ensure!(
165            lines[i].indent == base,
166            "inconsistent indentation: `{}`",
167            lines[i].text
168        );
169        let start = i + 1;
170        let mut j = start;
171        while j < lines.len() && lines[j].indent > base {
172            j += 1;
173        }
174        groups.push((&lines[i], &lines[start..j]));
175        i = j;
176    }
177    Ok(groups)
178}
179
180/// Render body lines as text, dedented to the body's minimum indentation so their
181/// relative structure (which MIR depends on) is preserved.
182fn body_text(lines: &[Line]) -> String {
183    if lines.is_empty() {
184        return String::new();
185    }
186    let base = lines.iter().map(|l| l.indent).min().expect("non-empty");
187    lines
188        .iter()
189        .map(|l| format!("{}{}", " ".repeat(l.indent - base), l.text))
190        .collect::<Vec<_>>()
191        .join("\n")
192}
193
194/// Split a header line into whitespace-separated tokens, keeping `"..."` strings
195/// and `[...]` lists intact.
196fn tokenize(s: &str) -> anyhow::Result<Vec<String>> {
197    let mut tokens = Vec::new();
198    let mut cur = String::new();
199    let mut depth = 0i32;
200    let mut in_quote = false;
201    for c in s.chars() {
202        match c {
203            '"' => {
204                in_quote = !in_quote;
205                cur.push(c);
206            }
207            '[' if !in_quote => {
208                depth += 1;
209                cur.push(c);
210            }
211            ']' if !in_quote => {
212                depth -= 1;
213                cur.push(c);
214            }
215            c if c.is_whitespace() && !in_quote && depth == 0 => {
216                if !cur.is_empty() {
217                    tokens.push(std::mem::take(&mut cur));
218                }
219            }
220            c => cur.push(c),
221        }
222    }
223    ensure!(!in_quote, "unterminated string in `{s}`");
224    ensure!(depth == 0, "unterminated list in `{s}`");
225    if !cur.is_empty() {
226        tokens.push(cur);
227    }
228    Ok(tokens)
229}
230
231/// Parse a header line into its verb, `key=value` arguments, and bare flags.
232fn parse_header(header: &str) -> anyhow::Result<(String, BTreeMap<String, String>, Vec<String>)> {
233    let mut tokens = tokenize(header)?.into_iter();
234    let verb = tokens.next().context("empty command")?;
235    let mut args = BTreeMap::new();
236    let mut flags = Vec::new();
237    for token in tokens {
238        match token.split_once('=') {
239            Some((key, value)) => {
240                args.insert(key.to_string(), value.to_string());
241            }
242            None => flags.push(token),
243        }
244    }
245    Ok((verb, args, flags))
246}
247
248fn req<'a>(args: &'a BTreeMap<String, String>, key: &str) -> anyhow::Result<&'a str> {
249    args.get(key)
250        .map(String::as_str)
251        .ok_or_else(|| anyhow!("missing argument `{key}`"))
252}
253
254fn req_u64(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<u64> {
255    req(args, key)?
256        .parse()
257        .with_context(|| format!("argument `{key}` is not an integer"))
258}
259
260fn opt_u64(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<Option<u64>> {
261    args.get(key)
262        .map(|v| {
263            v.parse()
264                .with_context(|| format!("argument `{key}` is not an integer"))
265        })
266        .transpose()
267}
268
269fn opt_usize(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<Option<usize>> {
270    args.get(key)
271        .map(|v| {
272            v.parse()
273                .with_context(|| format!("argument `{key}` is not an integer"))
274        })
275        .transpose()
276}
277
278/// Parse a global id argument: a bare `u64` (e.g. `1001`) is the user namespace,
279/// matching [`mz_repr::GlobalId`]'s `Display`/`FromStr` prefixes (`s`/`si`/`u`/`t`)
280/// for the other namespaces, e.g. `t7` is `GlobalId::Transient(7)`.
281fn parse_id(s: &str) -> anyhow::Result<GlobalId> {
282    if s.bytes().all(|b| b.is_ascii_digit()) {
283        let id: u64 = s.parse().with_context(|| format!("bad id `{s}`"))?;
284        Ok(GlobalId::User(id))
285    } else {
286        s.parse().with_context(|| format!("bad id `{s}`"))
287    }
288}
289
290fn req_id(args: &BTreeMap<String, String>, key: &str) -> anyhow::Result<GlobalId> {
291    parse_id(req(args, key)?)
292}
293
294fn opt_string(args: &BTreeMap<String, String>, key: &str) -> Option<String> {
295    args.get(key).cloned()
296}
297
298/// Parse a `[a,b,c]` list of `usize`s (`[]` is empty).
299fn parse_usize_list(s: &str) -> anyhow::Result<Vec<usize>> {
300    let inner = s
301        .strip_prefix('[')
302        .and_then(|s| s.strip_suffix(']'))
303        .ok_or_else(|| anyhow!("expected a list like `[0,1]`, got `{s}`"))?;
304    if inner.trim().is_empty() {
305        return Ok(vec![]);
306    }
307    inner
308        .split(',')
309        .map(|part| {
310            part.trim()
311                .parse()
312                .with_context(|| format!("bad list element `{part}`"))
313        })
314        .collect()
315}
316
317/// Parse an `export` sub-command into an [`ExportSpec`]. The `kind=` argument
318/// selects the variant (defaulting to `index`); each kind takes its own arguments.
319fn parse_export(args: &BTreeMap<String, String>) -> anyhow::Result<ExportSpec> {
320    let on_id = req_id(args, "on")?;
321    Ok(
322        match args.get("kind").map(String::as_str).unwrap_or("index") {
323            "index" => ExportSpec::Index {
324                index_id: req_id(args, "index")?,
325                on_id,
326                key: parse_usize_list(req(args, "key")?)?,
327            },
328            "materialized-view" => ExportSpec::MaterializedView {
329                sink_id: req_id(args, "sink")?,
330                on_id,
331                shard: req(args, "shard")?.to_string(),
332                schema: opt_string(args, "schema"),
333            },
334            "subscribe" => ExportSpec::Subscribe {
335                sink_id: req_id(args, "sink")?,
336                on_id,
337                schema: opt_string(args, "schema"),
338                up_to: opt_u64(args, "up-to")?,
339            },
340            "copy-to" => bail!("copy-to export is not implemented"),
341            other => bail!("unknown export kind `{other}`"),
342        },
343    )
344}
345
346/// Parse body lines as rows of raw space-separated value tokens (quotes intact).
347/// The tokens are typed against the schema server-side; see `cell_from_token`.
348fn rows_from_body(body: &[Line]) -> anyhow::Result<Vec<Vec<String>>> {
349    body.iter().map(|l| tokenize(&l.text)).collect()
350}
351
352/// Parse body lines as dyncfg settings: `name type value`.
353fn settings_from_body(body: &[Line]) -> anyhow::Result<Vec<ConfigSetting>> {
354    body.iter()
355        .map(|l| {
356            let tokens = tokenize(&l.text)?;
357            ensure!(
358                tokens.len() == 3,
359                "config setting needs `name type value`, got `{}`",
360                l.text
361            );
362            Ok(ConfigSetting {
363                name: tokens[0].clone(),
364                ty: tokens[1].clone(),
365                value: tokens[2].clone(),
366            })
367        })
368        .collect()
369}
370
371/// Parse body lines as column declarations: `name type [nullable]`.
372fn columns_from_body(body: &[Line]) -> anyhow::Result<Vec<ColumnSpec>> {
373    body.iter()
374        .map(|l| {
375            let tokens = tokenize(&l.text)?;
376            ensure!(
377                tokens.len() >= 2,
378                "column needs `name type [nullable]`, got `{}`",
379                l.text
380            );
381            Ok(ColumnSpec {
382                name: tokens[0].clone(),
383                ty: tokens[1].clone(),
384                nullable: tokens.get(2).is_some_and(|t| t == "nullable"),
385            })
386        })
387        .collect()
388}
389
390/// Parse a `create-dataflow` body of `import`/`build`/`export` sub-commands. The
391/// directive's bare flags carry the dataflow-level options (`optimize`).
392fn parse_create_dataflow(
393    args: &BTreeMap<String, String>,
394    flags: &[String],
395    body: &[Line],
396) -> anyhow::Result<Command> {
397    let name = opt_string(args, "name");
398    let as_of = req_u64(args, "as-of")?;
399    let optimize = flags.iter().any(|f| f == "optimize");
400    let mut imports = Vec::new();
401    let mut builds = Vec::new();
402    let mut exports = Vec::new();
403    for (header, sub_body) in group(body)? {
404        let (verb, args, _flags) = parse_header(&header.text)?;
405        match verb.as_str() {
406            "import" => {
407                if let Some(index_id) = args.get("index") {
408                    imports.push(ImportSpec::Index {
409                        index_id: parse_id(index_id)
410                            .with_context(|| format!("bad index id `{index_id}`"))?,
411                    });
412                } else {
413                    imports.push(ImportSpec::Source {
414                        id: req_id(&args, "source")?,
415                        shard: req(&args, "shard")?.to_string(),
416                        schema: opt_string(&args, "schema"),
417                        upper: req_u64(&args, "upper")?,
418                    });
419                }
420            }
421            "build" => {
422                ensure!(!sub_body.is_empty(), "`build` needs a MIR body");
423                builds.push(BuildSpec {
424                    id: req_id(&args, "id")?,
425                    expr: body_text(sub_body),
426                });
427            }
428            "export" => exports.push(parse_export(&args)?),
429            other => bail!("unknown `create-dataflow` sub-command `{other}`"),
430        }
431    }
432    Ok(Command::CreateDataflow {
433        name,
434        imports,
435        builds,
436        exports,
437        as_of,
438        optimize,
439    })
440}
441
442/// Parse one command block (directive line plus indentation-structured body).
443fn parse_command(input: &str) -> anyhow::Result<Command> {
444    let lines = lex(input);
445    let (header, body) = lines.split_first().context("empty command")?;
446    ensure!(header.indent == 0, "directive must not be indented");
447    let (verb, args, flags) = parse_header(&header.text)?;
448    let command = match verb.as_str() {
449        "define-schema" => Command::DefineSchema {
450            name: req(&args, "name")?.to_string(),
451            columns: columns_from_body(body)?,
452        },
453        "write-single-ts" => Command::WriteSingleTs {
454            shard: req(&args, "shard")?.to_string(),
455            schema: opt_string(&args, "schema"),
456            ts: req_u64(&args, "ts")?,
457            count: req_u64(&args, "count")?,
458            start: opt_u64(&args, "start")?.unwrap_or(0),
459            row_bytes: opt_usize(&args, "row-bytes")?,
460        },
461        "write-spread" => Command::WriteSpread {
462            shard: req(&args, "shard")?.to_string(),
463            schema: opt_string(&args, "schema"),
464            count: req_u64(&args, "count")?,
465            n_ts: req_u64(&args, "n-ts")?,
466            start: opt_u64(&args, "start")?.unwrap_or(0),
467            row_bytes: opt_usize(&args, "row-bytes")?,
468        },
469        "write-rows" => Command::WriteRows {
470            shard: req(&args, "shard")?.to_string(),
471            schema: opt_string(&args, "schema"),
472            ts: req_u64(&args, "ts")?,
473            rows: rows_from_body(body)?,
474        },
475        "define-index" => Command::DefineIndex {
476            source_id: req_id(&args, "source")?,
477            index_id: req_id(&args, "index")?,
478            shard: req(&args, "shard")?.to_string(),
479            schema: opt_string(&args, "schema"),
480            key: parse_usize_list(req(&args, "key")?)?,
481            as_of: req_u64(&args, "as-of")?,
482            upper: req_u64(&args, "upper")?,
483        },
484        "schedule" => Command::Schedule {
485            id: req_id(&args, "id")?,
486        },
487        "allow-compaction" => Command::AllowCompaction {
488            id: req_id(&args, "id")?,
489            frontier: req_u64(&args, "frontier")?,
490        },
491        "allow-writes" => Command::AllowWrites {
492            id: req_id(&args, "id")?,
493        },
494        "await-frontier" => Command::AwaitFrontier {
495            id: req_id(&args, "id")?,
496            ts: req_u64(&args, "ts")?,
497            timeout_secs: opt_u64(&args, "timeout-secs")?,
498            allow_timeout: flags.iter().any(|f| f == "allow-timeout"),
499        },
500        "count" => Command::Count {
501            id: req_id(&args, "id")?,
502            ts: req_u64(&args, "ts")?,
503        },
504        "peek" => Command::Peek {
505            id: req_id(&args, "id")?,
506            schema: opt_string(&args, "schema"),
507            ts: req_u64(&args, "ts")?,
508        },
509        "await-subscribe" => Command::AwaitSubscribe {
510            id: req_id(&args, "id")?,
511            up_to: req_u64(&args, "up-to")?,
512            timeout_secs: opt_u64(&args, "timeout-secs")?,
513        },
514        "create-dataflow" => parse_create_dataflow(&args, &flags, body)?,
515        "create-instance" => Command::CreateInstance {
516            expiration_offset: opt_string(&args, "expiration-offset"),
517            arrangement_dictionary_compression: args
518                .get("arrangement-dictionary-compression")
519                .map(|v| v.parse())
520                .transpose()
521                .context("argument `arrangement-dictionary-compression` is not a bool")?
522                .unwrap_or(false),
523            // Same `name type value` body rows as `update-configuration`.
524            initial_config: settings_from_body(body)?,
525        },
526        "update-configuration" => Command::UpdateConfiguration {
527            updates: settings_from_body(body)?,
528        },
529        "reconnect" => Command::Reconnect,
530        "initialization-complete" => Command::InitializationComplete,
531        other => bail!("unknown command `{other}`"),
532    };
533    Ok(command)
534}
535
536#[cfg(test)]
537mod tests {
538    use super::*;
539
540    /// A simple command parses its args; absent optionals default.
541    #[mz_ore::test]
542    fn parses_simple_command() {
543        let cmd = parse_command("write-single-ts shard=data ts=0 count=5000").unwrap();
544        assert_eq!(
545            cmd,
546            Command::WriteSingleTs {
547                shard: "data".to_string(),
548                schema: None,
549                ts: 0,
550                count: 5000,
551                start: 0,
552                row_bytes: None,
553            }
554        );
555    }
556
557    /// A flag and a list argument parse.
558    #[mz_ore::test]
559    fn parses_flag_and_list() {
560        let cmd =
561            parse_command("await-frontier id=1001 ts=1 timeout-secs=3 allow-timeout").unwrap();
562        assert_eq!(
563            cmd,
564            Command::AwaitFrontier {
565                id: GlobalId::User(1001),
566                ts: 1,
567                timeout_secs: Some(3),
568                allow_timeout: true,
569            }
570        );
571        let cmd =
572            parse_command("define-index source=1000 index=1001 shard=d key=[0] as-of=0 upper=1")
573                .unwrap();
574        assert_eq!(
575            cmd,
576            Command::DefineIndex {
577                source_id: GlobalId::User(1000),
578                index_id: GlobalId::User(1001),
579                shard: "d".to_string(),
580                schema: None,
581                key: vec![0],
582                as_of: 0,
583                upper: 1,
584            }
585        );
586    }
587
588    /// An id argument accepts a bare number (user namespace) or an explicit
589    /// `s`/`si`/`u`/`t` prefix selecting the namespace directly.
590    #[mz_ore::test]
591    fn parses_prefixed_ids() {
592        assert_eq!(
593            parse_command("schedule id=1001").unwrap(),
594            Command::Schedule {
595                id: GlobalId::User(1001)
596            }
597        );
598        assert_eq!(
599            parse_command("schedule id=u1001").unwrap(),
600            Command::Schedule {
601                id: GlobalId::User(1001)
602            }
603        );
604        assert_eq!(
605            parse_command("schedule id=t7").unwrap(),
606            Command::Schedule {
607                id: GlobalId::Transient(7)
608            }
609        );
610        assert_eq!(
611            parse_command("schedule id=s42").unwrap(),
612            Command::Schedule {
613                id: GlobalId::System(42)
614            }
615        );
616        assert!(parse_command("schedule id=bogus").is_err());
617    }
618
619    /// `define-schema` and `write-rows` parse their indented bodies, typing values.
620    #[mz_ore::test]
621    fn parses_bodies() {
622        let cmd = parse_command(
623            "define-schema name=events\n  key bigint\n  flag boolean\n  label text nullable",
624        )
625        .unwrap();
626        assert_eq!(
627            cmd,
628            Command::DefineSchema {
629                name: "events".to_string(),
630                columns: vec![
631                    ColumnSpec {
632                        name: "key".to_string(),
633                        ty: "bigint".to_string(),
634                        nullable: false
635                    },
636                    ColumnSpec {
637                        name: "flag".to_string(),
638                        ty: "boolean".to_string(),
639                        nullable: false
640                    },
641                    ColumnSpec {
642                        name: "label".to_string(),
643                        ty: "text".to_string(),
644                        nullable: true
645                    },
646                ],
647            }
648        );
649
650        let cmd = parse_command(
651            "write-rows shard=ev schema=events ts=1\n  1000 true alpha\n  1001 false null",
652        )
653        .unwrap();
654        assert_eq!(
655            cmd,
656            Command::WriteRows {
657                shard: "ev".to_string(),
658                schema: Some("events".to_string()),
659                ts: 1,
660                rows: vec![
661                    vec!["1000".to_string(), "true".to_string(), "alpha".to_string()],
662                    vec!["1001".to_string(), "false".to_string(), "null".to_string()],
663                ],
664            }
665        );
666    }
667
668    /// `create-dataflow` parses its sub-commands, keeping a `build`'s MIR (its
669    /// deeper body) with relative indentation preserved; the export kind defaults
670    /// to `index`.
671    #[mz_ore::test]
672    fn parses_create_dataflow_with_mir() {
673        let input = "create-dataflow name=count as-of=0\n  import index=1001\n  build id=2000\n    Reduce aggregates=[count(*)]\n      Get u1000\n  export index=2001 on=2000 key=[0]";
674        let cmd = parse_command(input).unwrap();
675        assert_eq!(
676            cmd,
677            Command::CreateDataflow {
678                name: Some("count".to_string()),
679                imports: vec![ImportSpec::Index {
680                    index_id: GlobalId::User(1001)
681                }],
682                builds: vec![BuildSpec {
683                    id: GlobalId::User(2000),
684                    expr: "Reduce aggregates=[count(*)]\n  Get u1000".to_string(),
685                }],
686                exports: vec![ExportSpec::Index {
687                    index_id: GlobalId::User(2001),
688                    on_id: GlobalId::User(2000),
689                    key: vec![0]
690                }],
691                as_of: 0,
692                optimize: false,
693            }
694        );
695
696        // The `optimize` flag on the directive line is picked up.
697        let optimized = parse_command(
698            "create-dataflow name=j as-of=0 optimize\n  import source=1000 shard=l upper=1\n  build id=2000\n    Get u1000\n  export index=2001 on=2000 key=[0]",
699        )
700        .unwrap();
701        assert!(matches!(
702            optimized,
703            Command::CreateDataflow { optimize: true, .. }
704        ));
705    }
706
707    /// `create-dataflow` parses the sink export kinds: a materialized-view sink with
708    /// a target shard, and a subscribe sink. The `copy-to` kind is rejected.
709    #[mz_ore::test]
710    fn parses_sink_export_kinds() {
711        let mv = "create-dataflow name=mv as-of=0\n  import source=1000 shard=r upper=1\n  build id=2000\n    Get u1000\n  export kind=materialized-view sink=2001 on=2000 shard=out schema=kv";
712        let Command::CreateDataflow { exports, .. } = parse_command(mv).unwrap() else {
713            panic!("expected create-dataflow");
714        };
715        assert_eq!(
716            exports,
717            vec![ExportSpec::MaterializedView {
718                sink_id: GlobalId::User(2001),
719                on_id: GlobalId::User(2000),
720                shard: "out".to_string(),
721                schema: Some("kv".to_string()),
722            }]
723        );
724
725        let sub = "create-dataflow name=sub as-of=0\n  import source=1000 shard=s upper=2\n  build id=2000\n    Get u1000\n  export kind=subscribe sink=2001 on=2000 up-to=2";
726        let Command::CreateDataflow { exports, .. } = parse_command(sub).unwrap() else {
727            panic!("expected create-dataflow");
728        };
729        assert_eq!(
730            exports,
731            vec![ExportSpec::Subscribe {
732                sink_id: GlobalId::User(2001),
733                on_id: GlobalId::User(2000),
734                schema: None,
735                up_to: Some(2),
736            }]
737        );
738
739        // copy-to is named but not implemented.
740        let copy = "create-dataflow name=c as-of=0\n  import source=1000 shard=s upper=1\n  build id=2000\n    Get u1000\n  export kind=copy-to sink=2001 on=2000";
741        assert!(parse_command(copy).is_err());
742    }
743
744    /// `create-instance` parses its optional knobs (defaulting), and
745    /// `update-configuration` parses a `name type value` table (empty when bodyless).
746    #[mz_ore::test]
747    fn parses_handshake_config() {
748        assert_eq!(
749            parse_command("create-instance").unwrap(),
750            Command::CreateInstance {
751                expiration_offset: None,
752                arrangement_dictionary_compression: false,
753                initial_config: vec![],
754            }
755        );
756        assert_eq!(
757            parse_command(
758                "create-instance expiration-offset=30s arrangement-dictionary-compression=true"
759            )
760            .unwrap(),
761            Command::CreateInstance {
762                expiration_offset: Some("30s".to_string()),
763                arrangement_dictionary_compression: true,
764                initial_config: vec![],
765            }
766        );
767        // The create-time snapshot is a `name type value` body, like `update-configuration`.
768        assert_eq!(
769            parse_command("create-instance\n  enable_my_flag bool true").unwrap(),
770            Command::CreateInstance {
771                expiration_offset: None,
772                arrangement_dictionary_compression: false,
773                initial_config: vec![ConfigSetting {
774                    name: "enable_my_flag".to_string(),
775                    ty: "bool".to_string(),
776                    value: "true".to_string(),
777                }],
778            }
779        );
780
781        assert_eq!(
782            parse_command("update-configuration").unwrap(),
783            Command::UpdateConfiguration { updates: vec![] }
784        );
785        assert_eq!(
786            parse_command("update-configuration\n  enable_my_flag bool true\n  my_dur duration 1s")
787                .unwrap(),
788            Command::UpdateConfiguration {
789                updates: vec![
790                    ConfigSetting {
791                        name: "enable_my_flag".to_string(),
792                        ty: "bool".to_string(),
793                        value: "true".to_string(),
794                    },
795                    ConfigSetting {
796                        name: "my_dur".to_string(),
797                        ty: "duration".to_string(),
798                        value: "1s".to_string(),
799                    },
800                ],
801            }
802        );
803    }
804
805    /// `await-subscribe` parses its arguments.
806    #[mz_ore::test]
807    fn parses_await_subscribe() {
808        let cmd = parse_command("await-subscribe id=2001 up-to=2 timeout-secs=5").unwrap();
809        assert_eq!(
810            cmd,
811            Command::AwaitSubscribe {
812                id: GlobalId::User(2001),
813                up_to: 2,
814                timeout_secs: Some(5),
815            }
816        );
817    }
818
819    /// A file splits into stanzas, preserving comments and blanks, and round-trips
820    /// through a rewrite when the actual output equals the expected.
821    #[mz_ore::test]
822    fn parses_and_rewrites_file() {
823        let content =
824            "# a comment\nschedule id=1001\n----\nok\n\ncount id=1001 ts=5\n----\n10000\n";
825        let items = parse_file(content).unwrap();
826        let stanzas: Vec<_> = items
827            .iter()
828            .filter_map(|i| match i {
829                Item::Stanza(s) => Some(s),
830                Item::Verbatim(_) => None,
831            })
832            .collect();
833        assert_eq!(stanzas.len(), 2);
834        assert_eq!(
835            stanzas[0].command,
836            Command::Schedule {
837                id: GlobalId::User(1001)
838            }
839        );
840        assert_eq!(stanzas[0].expected, "ok");
841        assert_eq!(stanzas[1].expected, "10000");
842
843        // Rewriting with the same outputs reproduces the file.
844        let actuals = vec!["ok".to_string(), "10000".to_string()];
845        assert_eq!(rewrite(&items, &actuals), content);
846    }
847}
mz_clusterd_test_driver/text.rs

mz_clusterd_test_driver/
text.rs
