diff --git a/doc/developer/design/20260612_headless_clusterd_test_driver.md b/doc/developer/design/20260612_headless_clusterd_test_driver.md index eb78c22be505d..08ae6daf47e87 100644 --- a/doc/developer/design/20260612_headless_clusterd_test_driver.md +++ b/doc/developer/design/20260612_headless_clusterd_test_driver.md @@ -281,6 +281,10 @@ The ephemeral dataflows take global ids from a high reserved range so they never `create-dataflow` is the generic abstraction behind index / materialized-view / subscribe / copy-to: a command carrying `imports`, `builds`, `exports`, an `as_of`, and an optional `optimize` flag over the `DataflowBuilder` API, with `define-index` as sugar over it. The `optimize` flag runs the MIR optimizer before lowering; it is what lets a `Join` lower (`join.spec` joins two sources on a key), since a raw join's `implementation` is `Unimplemented` and the lowering rejects it otherwise. +The `explain` verb renders a dataflow's lowered LIR plan (the `EXPLAIN PHYSICAL PLAN` form, via a no-catalog `DummyHumanizer` so ids render as `u` and columns as `#n`) as its golden output, instead of submitting. +It takes the dataflow either inline (the same body as `create-dataflow`) or by reference — `explain ref=` renders a dataflow a prior `create-dataflow name=` declared, without repeating its body (the recorded spec is re-lowered, so the plan matches what was submitted). +With `optimize`, this asserts the optimizer's plan shape, so subtle optimizer or lowering drift that a result-only assertion would miss is caught (`join.spec` declares the join, then `explain ref=join` asserts the differential-join plan alongside the count). +Because the render spans multiple objects separated by blank lines, an `explain` golden uses the `datadriven` doubled-`----` block form (see `text`), which `REWRITE` emits automatically. Each export selects a `kind`: `index` (an arrangement), `materialized-view` (a persist sink), and `subscribe` (a sink streaming changes back) are implemented; `copy-to` (matching the last `ComputeSinkConnection` variant) is rejected as unimplemented. A sink export declares its output `schema`, which the builder validates against the exported object's column types before submission. A subscribe also takes an optional `up-to` (the exclusive upper at which it completes); the materialized-view sink does not support `UP TO` (the real optimizer leaves it empty too), so the driver always passes an empty one. The augment step splices each materialized-view sink's target `CollectionMetadata` into its connection — the same fill-in `compute-client`'s `Instance::create_dataflow` does — while subscribe carries no storage metadata. A materialized-view sink begins read-only and writes nothing until `allow-writes`; indexes, subscribes, and peeks need no such permission. diff --git a/src/clusterd-test-driver/src/dataflow.rs b/src/clusterd-test-driver/src/dataflow.rs index 2cce5f00968fb..6be0d664457ba 100644 --- a/src/clusterd-test-driver/src/dataflow.rs +++ b/src/clusterd-test-driver/src/dataflow.rs @@ -24,6 +24,7 @@ //! [`ComputeCommand::CreateDataflow`]: mz_compute_client::protocol::command::ComputeCommand::CreateDataflow use std::collections::BTreeMap; +use std::time::Duration; use mz_compute_types::dataflows::{ BuildDesc, DataflowDescription, IndexDesc, IndexImport, SourceImport, @@ -35,10 +36,12 @@ use mz_compute_types::sinks::{ SubscribeSinkConnection, }; use mz_compute_types::sources::SourceInstanceDesc; +use mz_expr::explain::ExplainContext; use mz_expr::{ AggregateExpr, AggregateFunc, MirRelationExpr, MirScalarExpr, OptimizedMirRelationExpr, }; use mz_persist_types::{PersistLocation, ShardId}; +use mz_repr::explain::{DummyHumanizer, Explain, ExplainConfig, ExplainFormat, UsedIndexes}; use mz_repr::optimize::OptimizerFeatures; use mz_repr::{GlobalId, RelationDesc, ReprRelationType, Timestamp}; use mz_storage_types::controller::CollectionMetadata; @@ -443,34 +446,73 @@ impl DataflowBuilder { /// column out of range, or an unbalanced object graph), so a caller driving /// this from external input — notably the script reader — can surface a clean /// error instead of crashing the process. - pub fn finish(mut self) -> anyhow::Result> { + pub fn finish(self) -> anyhow::Result> { let features = OptimizerFeatures::default(); + let lowered = Self::lower(self.mir, self.optimize, &features)?; + augment(lowered, &self.sources, &self.sinks) + } + + /// Render the lowered dataflow as `EXPLAIN PHYSICAL PLAN`-style text — the LIR + /// the dataflow ships — so a script can golden-assert the optimized + /// plan shape and catch optimizer (or lowering) drift, which a result-only + /// assertion misses. + /// + /// Honors [`Self::optimize`] exactly like [`Self::finish`], so the explained + /// plan is the one that would be shipped. A no-catalog [`DummyHumanizer`] + /// renders ids as `u123` and columns as `#n` — stable and matching the `.spec` + /// MIR vocabulary, with no catalog to thread in. + pub fn explain(self) -> anyhow::Result { + let features = OptimizerFeatures::default(); + let mut lowered = Self::lower(self.mir, self.optimize, &features)?; + let config = ExplainConfig::default(); + let context = ExplainContext { + config: &config, + features: &features, + humanizer: &DummyHumanizer, + cardinality_stats: BTreeMap::new(), + used_indexes: UsedIndexes::default(), + finishing: None, + duration: Duration::default(), + target_cluster: None, + optimizer_notices: Vec::new(), + }; + lowered + .explain(&ExplainFormat::Text, &context) + .map_err(|e| anyhow::anyhow!("explaining dataflow failed: {e}")) + } + + /// Optionally run the MIR dataflow optimizer, then lower MIR to LIR. + /// Shared by [`Self::finish`] (which augments the result with persist metadata) + /// and [`Self::explain`] (which renders it). Deterministic and self-contained. + fn lower( + mut mir: DataflowDescription, + optimize: bool, + features: &OptimizerFeatures, + ) -> anyhow::Result> { // Optionally run the MIR dataflow optimizer first (e.g. to fill a `Join`'s // implementation). The index oracle is built from this dataflow's own // `index_imports`, so the optimizer recognizes imported arrangements and // plans `Get`s over them as arrangement reads (not persist reads); the // statistics oracle is empty — no catalog stats — so join planning falls // back to a differential join, which lowers. - if self.optimize { - let indexes = ImportedIndexOracle::new(&self.mir.index_imports); + if optimize { + let indexes = ImportedIndexOracle::new(&mir.index_imports); let typecheck_ctx = empty_typechecking_context(); let mut df_meta = DataflowMetainfo::default(); let mut ctx = TransformCtx::global( &indexes, &EmptyStatisticsOracle, - &features, + features, &typecheck_ctx, &mut df_meta, None, ); - optimize_dataflow(&mut self.mir, &mut ctx, false) + optimize_dataflow(&mut mir, &mut ctx, false) .map_err(|e| anyhow::anyhow!("optimizing dataflow failed: {e}"))?; } // Lower MIR -> LIR. Deterministic and self-contained. - let lowered: DataflowDescription = - LirRelationExpr::finalize_dataflow(self.mir, &features, None) - .map_err(|e| anyhow::anyhow!("lowering dataflow failed: {e}"))?; - augment(lowered, &self.sources, &self.sinks) + LirRelationExpr::finalize_dataflow(mir, features, None) + .map_err(|e| anyhow::anyhow!("lowering dataflow failed: {e}")) } } @@ -848,6 +890,52 @@ mod tests { assert!(assemble(true).is_ok()); } + /// `explain` renders the lowered LIR plan as text, so a script can assert the + /// optimized plan shape. Build the optimized two-source join and confirm the + /// rendered plan mentions a `Join` (the operator the optimizer selected). + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux` + fn explain_join_renders_plan() { + let loc = PersistLocation { + blob_uri: "mem://".parse().unwrap(), + consensus_uri: "mem://".parse().unwrap(), + }; + let mut builder = DataflowBuilder::new("headless-explain-test"); + let left = builder.import_persist( + GlobalId::User(1000), + PersistSource { + shard: ShardId::new(), + location: loc.clone(), + desc: crate::data::sample_desc(), + upper: Timestamp::from(1), + }, + ); + let right = builder.import_persist( + GlobalId::User(1001), + PersistSource { + shard: ShardId::new(), + location: loc.clone(), + desc: crate::data::sample_desc(), + upper: Timestamp::from(1), + }, + ); + let join = MirRelationExpr::join_scalars( + vec![left.get(), right.get()], + vec![vec![MirScalarExpr::column(0), MirScalarExpr::column(2)]], + ); + builder.build(GlobalId::User(2000), join); + builder.optimize(); + builder.as_of(Timestamp::from(0)); + builder.export_index(GlobalId::User(2001), GlobalId::User(2000), vec![0]); + let text = builder.explain().unwrap(); + // Print so the rendered shape is visible under `--nocapture`. + println!("{text}"); + assert!( + text.contains("Join"), + "explain output missing Join:\n{text}" + ); + } + /// A single dataflow can export both an index and a materialized view over the /// same built object (binding). Both exports reference that object; the index /// arranges it and the MV sink writes it to a target shard. diff --git a/src/clusterd-test-driver/src/script.rs b/src/clusterd-test-driver/src/script.rs index ee9a261074111..fe51bba5bd357 100644 --- a/src/clusterd-test-driver/src/script.rs +++ b/src/clusterd-test-driver/src/script.rs @@ -172,6 +172,41 @@ pub enum ExportSpec { }, } +/// What an `explain` command renders: a dataflow given inline, or a reference to one +/// a prior `create-dataflow` declared by name. The reference form avoids repeating a +/// dataflow's body just to assert its plan. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ExplainTarget { + /// A dataflow declared inline, with the same body as `create-dataflow`. + Inline { + /// Debug name for the dataflow; defaults to `headless-create-dataflow`. + #[serde(default)] + name: Option, + /// Collections to import (persist sources and/or existing indexes). + #[serde(default)] + imports: Vec, + /// MIR objects to compute, each bound to an id. + #[serde(default)] + builds: Vec, + /// Exports over imported or built ids. + #[serde(default)] + exports: Vec, + /// The dataflow's `as_of`. + as_of: u64, + /// Run the MIR optimizer before lowering. Off by default. + #[serde(default)] + optimize: bool, + }, + /// A dataflow a prior `create-dataflow name=` declared, rendered without + /// repeating its body. Reuses the recorded spec, so the plan matches what was + /// submitted. + Reference { + /// The `create-dataflow` name to render. + name: String, + }, +} + /// Map a JSON type name to a [`SqlScalarType`]. The supported set is intentionally /// small and matches [`crate::data::Cell`]; extend both together. fn scalar_type_from_str(s: &str) -> anyhow::Result { @@ -494,6 +529,16 @@ pub enum Command { #[serde(default)] optimize: bool, }, + /// Render a dataflow's lowered LIR plan as text, the output assertion being the + /// plan shape itself. It submits nothing and records no index, subscribe, or + /// materialized-view output. With `optimize`, this asserts the optimizer's plan + /// so subtle optimizer (or lowering) drift is caught, which a result-only assertion + /// misses. The dataflow is given either inline (the `create-dataflow` body) or by + /// reference to one a prior `create-dataflow` declared (see [`ExplainTarget`]). + Explain { + /// What to explain: an inline dataflow or a reference to a declared one. + target: ExplainTarget, + }, /// Peek `id` at `ts` and emit the returned rows (sorted, one per line). The /// generic output assertion: the script's `----` block holds the expected rows. Peek { @@ -562,6 +607,33 @@ struct IndexEntry { on_type: ReprRelationType, } +/// Side-effect registrations a successful `create-dataflow` submit must apply: +/// exported indexes (for later import / count), subscribe sinks (response buffers), +/// and materialized-view outputs (for persist peeks). `explain` builds the same +/// dataflow but discards these, since it submits nothing. +#[derive(Default)] +struct PendingRegistrations { + /// Exported indexes, by global id. + indexes: Vec<(GlobalId, IndexEntry)>, + /// Subscribe sink ids needing a response buffer. + subscribes: Vec, + /// Materialized-view outputs: sink id and its target shard metadata. + mv_outputs: Vec<(GlobalId, CollectionMetadata)>, +} + +/// The `create-dataflow` body recorded under a dataflow's name, so a later +/// `explain ref=` can re-render its plan without repeating the body. Holds the +/// parsed spec rather than a built plan: lowering is deterministic, so re-running it +/// yields the same plan that was submitted. +#[derive(Clone)] +struct DataflowSpec { + imports: Vec, + builds: Vec, + exports: Vec, + as_of: u64, + optimize: bool, +} + /// The base for ephemeral global ids the count sugar allocates. Far above any /// id a script would use, so its dataflows never collide with user objects. const INTERNAL_ID_BASE: u64 = u64::MAX / 2; @@ -580,6 +652,9 @@ pub struct ScriptState { /// Materialized-view sink outputs, by sink global id: the target shard's /// metadata, so a `peek` of the sink id reads its shard via a persist peek. mv_outputs: BTreeMap, + /// `create-dataflow` specs by name, so `explain ref=` can render a declared + /// dataflow's plan without repeating its body. + dataflows: BTreeMap, /// Next ephemeral id for the count sugar's dataflows. next_internal: u64, } @@ -597,6 +672,7 @@ impl ScriptState { shards: BTreeMap::new(), indexes: BTreeMap::new(), mv_outputs: BTreeMap::new(), + dataflows: BTreeMap::new(), next_internal: INTERNAL_ID_BASE, }) } @@ -713,6 +789,157 @@ impl ScriptState { Ok(()) } + /// Build (but do not submit) a [`DataflowBuilder`] from a `create-dataflow` / + /// `explain` body: import sources and existing indexes, build the MIR objects, and + /// wire the exports, setting `as_of`. Returns the configured builder plus the + /// [`PendingRegistrations`] a successful submit must apply — `create-dataflow` + /// applies them, `explain` discards them (it submits nothing). + fn configure_dataflow( + &mut self, + name: Option, + imports: Vec, + builds: Vec, + exports: Vec, + as_of: u64, + optimize: bool, + ) -> anyhow::Result<(DataflowBuilder, PendingRegistrations)> { + let mut builder = + DataflowBuilder::new(name.unwrap_or_else(|| "headless-create-dataflow".to_string())); + if optimize { + builder.optimize(); + } + // The parser's catalog resolves `Get u` leaves by name; it assigns its own + // global ids, so we keep a name->our-id map and remap the parsed `Get`s back to + // the script's ids afterwards. + let mut catalog = TestCatalog::default(); + let mut name_to_id: BTreeMap = BTreeMap::new(); + for import in imports { + match import { + ImportSpec::Source { + id, + shard, + schema, + upper, + } => { + let desc = self.resolve_schema(&schema)?; + register_catalog_object(&mut catalog, &mut name_to_id, id, desc.typ().clone())?; + let shard = self.shard_id(&shard); + builder.import_persist( + id, + PersistSource { + shard, + location: self.loc.clone(), + desc, + upper: Timestamp::from(upper), + }, + ); + } + ImportSpec::Index { index_id } => { + let entry = self.indexes.get(&index_id).ok_or_else(|| { + anyhow::anyhow!("unknown index {index_id}; define it before importing it") + })?; + let on_id = entry.on_id; + let key = entry.key.clone(); + let on_type = entry.on_type.clone(); + register_catalog_object( + &mut catalog, + &mut name_to_id, + on_id, + SqlRelationType::from_repr(&on_type), + )?; + builder.import_index(index_id, on_id, key, on_type, false); + } + } + } + for build in builds { + // Parse the pretty MIR spec against the catalog, then remap its + // catalog-assigned `Get` ids to the script's ids. + let mut expr = try_parse_mir(&catalog, &build.expr) + .map_err(|e| anyhow::anyhow!("parsing MIR for object {}: {e}", build.id))?; + remap_gets(&mut expr, &catalog, &name_to_id)?; + let id = build.id; + // Register the built object so later builds can `get` it. + register_catalog_object( + &mut catalog, + &mut name_to_id, + id, + SqlRelationType::from_repr(&expr.typ()), + )?; + builder.build(id, expr); + } + // Wire each export onto the builder. Index exports are captured for later + // import / count assertions; sink exports route their output either to a target + // shard (materialized view) or back as responses (subscribe). Sink output + // schemas must match the exported object's type, validated here so a mismatch + // fails before submission. + let mut registrations = PendingRegistrations::default(); + for export in exports { + match export { + ExportSpec::Index { + index_id, + on_id, + key, + } => { + let on_type = builder.get(on_id)?.typ(); + builder.export_index(index_id, on_id, key.clone()); + registrations.indexes.push(( + index_id, + IndexEntry { + on_id, + key, + on_type, + }, + )); + } + ExportSpec::MaterializedView { + sink_id, + on_id, + shard, + schema, + } => { + let desc = self.resolve_schema(&schema)?; + self.check_sink_schema(&builder, on_id, &desc)?; + let shard = self.shard_id(&shard); + let location = self.loc.clone(); + builder.export_materialized_view( + sink_id, + on_id, + desc.clone(), + PersistSink { + shard, + location: location.clone(), + }, + ); + // Record the target shard so a later `peek` of the sink id reads it + // via a persist peek (the `SELECT * FROM mv` path), with no separate + // read-back command. + registrations.mv_outputs.push(( + sink_id, + CollectionMetadata { + persist_location: location, + data_shard: shard, + relation_desc: desc, + txns_shard: None, + }, + )); + } + ExportSpec::Subscribe { + sink_id, + on_id, + schema, + up_to, + } => { + let desc = self.resolve_schema(&schema)?; + self.check_sink_schema(&builder, on_id, &desc)?; + builder.export_subscribe(sink_id, on_id, desc, up_to_antichain(up_to)); + registrations.subscribes.push(sink_id); + } + } + } + builder.as_of(Timestamp::from(as_of)); + Ok((builder, registrations)) + } + /// Execute a single command, returning its golden output text. pub async fn execute(&mut self, cmd: Command) -> anyhow::Result { match cmd { @@ -860,165 +1087,79 @@ impl ScriptState { as_of, optimize, } => { - let mut builder = DataflowBuilder::new( - name.unwrap_or_else(|| "headless-create-dataflow".to_string()), - ); - if optimize { - builder.optimize(); - } - // The parser's catalog resolves `Get u` leaves by name; it - // assigns its own global ids, so we keep a name->our-id map and - // remap the parsed `Get`s back to the script's ids afterwards. - let mut catalog = TestCatalog::default(); - let mut name_to_id: BTreeMap = BTreeMap::new(); - for import in imports { - match import { - ImportSpec::Source { - id, - shard, - schema, - upper, - } => { - let desc = self.resolve_schema(&schema)?; - register_catalog_object( - &mut catalog, - &mut name_to_id, - id, - desc.typ().clone(), - )?; - let shard = self.shard_id(&shard); - builder.import_persist( - id, - PersistSource { - shard, - location: self.loc.clone(), - desc, - upper: Timestamp::from(upper), - }, - ); - } - ImportSpec::Index { index_id } => { - let entry = self.indexes.get(&index_id).ok_or_else(|| { - anyhow::anyhow!( - "unknown index {index_id}; define it before importing it" - ) - })?; - let on_id = entry.on_id; - let key = entry.key.clone(); - let on_type = entry.on_type.clone(); - register_catalog_object( - &mut catalog, - &mut name_to_id, - on_id, - SqlRelationType::from_repr(&on_type), - )?; - builder.import_index(index_id, on_id, key, on_type, false); - } - } - } - for build in builds { - // Parse the pretty MIR spec against the catalog, then remap - // its catalog-assigned `Get` ids to the script's ids. - let mut expr = try_parse_mir(&catalog, &build.expr) - .map_err(|e| anyhow::anyhow!("parsing MIR for object {}: {e}", build.id))?; - remap_gets(&mut expr, &catalog, &name_to_id)?; - let id = build.id; - // Register the built object so later builds can `get` it. - register_catalog_object( - &mut catalog, - &mut name_to_id, - id, - SqlRelationType::from_repr(&expr.typ()), - )?; - builder.build(id, expr); - } - // Wire each export onto the builder. Index exports are captured for - // later import / count assertions; sink exports route their output - // either to a target shard (materialized view) or back as responses - // (subscribe). Sink output schemas must match the exported object's - // type, validated here so a mismatch fails before submission. - let mut new_indexes = Vec::new(); - let mut new_subscribes = Vec::new(); - let mut new_mv_outputs = Vec::new(); - for export in exports { - match export { - ExportSpec::Index { - index_id, - on_id, - key, - } => { - let on_type = builder.get(on_id)?.typ(); - builder.export_index(index_id, on_id, key.clone()); - new_indexes.push((index_id, on_id, key, on_type)); - } - ExportSpec::MaterializedView { - sink_id, - on_id, - shard, - schema, - } => { - let desc = self.resolve_schema(&schema)?; - self.check_sink_schema(&builder, on_id, &desc)?; - let shard = self.shard_id(&shard); - let location = self.loc.clone(); - builder.export_materialized_view( - sink_id, - on_id, - desc.clone(), - PersistSink { - shard, - location: location.clone(), - }, - ); - // Record the target shard so a later `peek` of the sink - // id reads it via a persist peek (the `SELECT * FROM mv` - // path), with no separate read-back command. - new_mv_outputs.push(( - sink_id, - CollectionMetadata { - persist_location: location, - data_shard: shard, - relation_desc: desc, - txns_shard: None, - }, - )); - } - ExportSpec::Subscribe { - sink_id, - on_id, - schema, - up_to, - } => { - let desc = self.resolve_schema(&schema)?; - self.check_sink_schema(&builder, on_id, &desc)?; - builder.export_subscribe(sink_id, on_id, desc, up_to_antichain(up_to)); - new_subscribes.push(sink_id); - } - } + // Record the spec under its name so `explain ref=` can render + // this dataflow's plan later without repeating the body. + if let Some(name) = &name { + self.dataflows.insert( + name.clone(), + DataflowSpec { + imports: imports.clone(), + builds: builds.clone(), + exports: exports.clone(), + as_of, + optimize, + }, + ); } - builder.as_of(Timestamp::from(as_of)); + let (builder, registrations) = + self.configure_dataflow(name, imports, builds, exports, as_of, optimize)?; let df = builder.finish()?; self.driver.submit_dataflow(df)?; // Register only after a successful submit, so a rejected dataflow // leaves no dangling index entry or subscribe buffer. - for (index_id, on_id, key, on_type) in new_indexes { - self.indexes.insert( - index_id, - IndexEntry { - on_id, - key, - on_type, - }, - ); + for (index_id, entry) in registrations.indexes { + self.indexes.insert(index_id, entry); } - for sink_id in new_subscribes { + for sink_id in registrations.subscribes { self.driver.register_subscribe(sink_id); } - for (sink_id, metadata) in new_mv_outputs { + for (sink_id, metadata) in registrations.mv_outputs { self.mv_outputs.insert(sink_id, metadata); } Ok("ok".to_string()) } + Command::Explain { target } => { + // Resolve the target to a dataflow body: either given inline, or the + // spec a prior `create-dataflow name=` recorded. + let (name, imports, builds, exports, as_of, optimize) = match target { + ExplainTarget::Inline { + name, + imports, + builds, + exports, + as_of, + optimize, + } => (name, imports, builds, exports, as_of, optimize), + ExplainTarget::Reference { name } => { + let spec = self.dataflows.get(&name).ok_or_else(|| { + anyhow::anyhow!( + "unknown dataflow {name:?}; declare it with \ + create-dataflow name={name} first" + ) + })?; + ( + Some(name.clone()), + spec.imports.clone(), + spec.builds.clone(), + spec.exports.clone(), + spec.as_of, + spec.optimize, + ) + } + }; + // Build the same dataflow as `create-dataflow`, but render its lowered + // LIR plan instead of submitting it. The registrations are discarded: + // explain has no side effects, so it neither installs a dataflow nor + // records an index / subscribe / materialized-view output. + let (builder, _registrations) = + self.configure_dataflow(name, imports, builds, exports, as_of, optimize)?; + // The LIR render separates objects with blank lines; the `----` block + // preserves them via the doubled-separator form (see `crate::text`). + // Trim the trailing newline so the golden matches like every other + // command's (none emit a trailing newline). + let plan = builder.explain()?; + Ok(plan.trim_end().to_string()) + } Command::Peek { id, schema, ts } => { let desc = self.resolve_schema(&schema)?; // A materialized-view sink id resolves to a persist peek of its diff --git a/src/clusterd-test-driver/src/text.rs b/src/clusterd-test-driver/src/text.rs index 0b42d81dd6040..3d3395f266782 100644 --- a/src/clusterd-test-driver/src/text.rs +++ b/src/clusterd-test-driver/src/text.rs @@ -27,6 +27,11 @@ //! A `#` at column 0 is a comment; an indented `#0` is a column reference in MIR. //! Comments and blank lines are preserved across a rewrite. //! +//! Output that itself contains blank lines (notably an `explain` plan render) uses +//! the `datadriven` doubled-separator form: the directive, then `----`, then `----`, +//! then the expected output, closed by a `----`/`----` pair. `REWRITE` emits this +//! form automatically when the output contains a blank line. +//! //! Command bodies are indentation-structured: `define-schema`/`write-rows`/`peek` //! carry rows or columns, and `define` carries `import`/`build`/`export` //! sub-commands, with a `build`'s MIR as its own deeper-indented sub-body. @@ -36,7 +41,9 @@ use std::collections::BTreeMap; use anyhow::{Context, anyhow, bail, ensure}; use mz_repr::GlobalId; -use crate::script::{BuildSpec, ColumnSpec, Command, ConfigSetting, ExportSpec, ImportSpec}; +use crate::script::{ + BuildSpec, ColumnSpec, Command, ConfigSetting, ExplainTarget, ExportSpec, ImportSpec, +}; /// One element of a parsed script file, retained so a `REWRITE` reproduces the /// file faithfully. @@ -83,12 +90,36 @@ pub fn parse_file(content: &str) -> anyhow::Result> { ); let input = lines[start..i].join("\n"); i += 1; // consume `----` - // The expected output runs to the next blank line (or end of file). - let exp_start = i; - while i < lines.len() && !lines[i].trim().is_empty() { + // A second `----` opens "blank-line mode" (the `datadriven` convention): the + // expected output may contain blank lines and runs until a closing `----` + // `----` pair, instead of ending at the first blank line. Used for the + // multi-object `explain` plan render. + let blank_mode = i < lines.len() && lines[i] == "----"; + if blank_mode { i += 1; } - let expected = lines[exp_start..i].join("\n"); + let exp_start = i; + let expected = if blank_mode { + while i < lines.len() + && !(lines[i] == "----" && i + 1 < lines.len() && lines[i + 1] == "----") + { + i += 1; + } + ensure!( + i < lines.len(), + "stanza starting at line {} has an unclosed `----`/`----` block", + start + 1 + ); + let expected = lines[exp_start..i].join("\n"); + i += 2; // consume the closing `----` `----` + expected + } else { + // The expected output runs to the next blank line (or end of file). + while i < lines.len() && !lines[i].trim().is_empty() { + i += 1; + } + lines[exp_start..i].join("\n") + }; let command = parse_command(&input) .with_context(|| format!("parsing stanza at line {}", start + 1))?; items.push(Item::Stanza(Stanza { @@ -116,10 +147,18 @@ pub fn rewrite(items: &[Item], actuals: &[String]) -> String { let actual = &actuals[next]; next += 1; out.push_str(&stanza.input); - out.push_str("\n----\n"); - out.push_str(actual); - if !actual.is_empty() { - out.push('\n'); + if actual.contains("\n\n") { + // Blank lines in the output need the doubled-`----` form, else the + // first blank line would truncate the block on the next parse. + out.push_str("\n----\n----\n"); + out.push_str(actual); + out.push_str("\n----\n----\n"); + } else { + out.push_str("\n----\n"); + out.push_str(actual); + if !actual.is_empty() { + out.push('\n'); + } } } } @@ -387,13 +426,24 @@ fn columns_from_body(body: &[Line]) -> anyhow::Result> { .collect() } -/// 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( +/// The parsed parts of a `create-dataflow` / `explain` body, shared by both verbs. +struct DataflowBody { + name: Option, + imports: Vec, + builds: Vec, + exports: Vec, + as_of: u64, + optimize: bool, +} + +/// Parse a dataflow body of `import`/`build`/`export` sub-commands, shared by +/// `create-dataflow` and `explain`. The directive's bare flags carry the +/// dataflow-level options (`optimize`). +fn parse_dataflow_body( args: &BTreeMap, flags: &[String], body: &[Line], -) -> anyhow::Result { +) -> anyhow::Result { let name = opt_string(args, "name"); let as_of = req_u64(args, "as-of")?; let optimize = flags.iter().any(|f| f == "optimize"); @@ -426,10 +476,10 @@ fn parse_create_dataflow( }); } "export" => exports.push(parse_export(&args)?), - other => bail!("unknown `create-dataflow` sub-command `{other}`"), + other => bail!("unknown dataflow sub-command `{other}`"), } } - Ok(Command::CreateDataflow { + Ok(DataflowBody { name, imports, builds, @@ -511,7 +561,53 @@ fn parse_command(input: &str) -> anyhow::Result { up_to: req_u64(&args, "up-to")?, timeout_secs: opt_u64(&args, "timeout-secs")?, }, - "create-dataflow" => parse_create_dataflow(&args, &flags, body)?, + "create-dataflow" => { + let DataflowBody { + name, + imports, + builds, + exports, + as_of, + optimize, + } = parse_dataflow_body(&args, &flags, body)?; + Command::CreateDataflow { + name, + imports, + builds, + exports, + as_of, + optimize, + } + } + "explain" => { + // `explain ref=` renders a previously declared dataflow; otherwise + // the dataflow is given inline with the `create-dataflow` body. + let target = if let Some(reference) = opt_string(&args, "ref") { + ensure!( + body.is_empty(), + "`explain ref=...` takes no body; it renders the declared dataflow" + ); + ExplainTarget::Reference { name: reference } + } else { + let DataflowBody { + name, + imports, + builds, + exports, + as_of, + optimize, + } = parse_dataflow_body(&args, &flags, body)?; + ExplainTarget::Inline { + name, + imports, + builds, + exports, + as_of, + optimize, + } + }; + Command::Explain { target } + } "create-instance" => Command::CreateInstance { expiration_offset: opt_string(&args, "expiration-offset"), arrangement_dictionary_compression: args @@ -704,6 +800,63 @@ mod tests { )); } + /// Inline `explain` shares the `create-dataflow` body grammar but parses into + /// `Command::Explain` with an `Inline` target carrying the same body. + #[mz_ore::test] + fn parses_explain_inline() { + let input = "explain name=j as-of=0 optimize\n import source=1000 shard=l upper=1\n import source=1001 shard=r upper=1\n build id=2000\n Join on=(#0 = #2)\n Get u1000\n Get u1001\n export index=2001 on=2000 key=[0]"; + let cmd = parse_command(input).unwrap(); + assert_eq!( + cmd, + Command::Explain { + target: ExplainTarget::Inline { + name: Some("j".to_string()), + imports: vec![ + ImportSpec::Source { + id: GlobalId::User(1000), + shard: "l".to_string(), + schema: None, + upper: 1, + }, + ImportSpec::Source { + id: GlobalId::User(1001), + shard: "r".to_string(), + schema: None, + upper: 1, + }, + ], + builds: vec![BuildSpec { + id: GlobalId::User(2000), + expr: "Join on=(#0 = #2)\n Get u1000\n Get u1001".to_string(), + }], + exports: vec![ExportSpec::Index { + index_id: GlobalId::User(2001), + on_id: GlobalId::User(2000), + key: vec![0], + }], + as_of: 0, + optimize: true, + } + } + ); + } + + /// `explain ref=` parses into a `Reference` target and takes no body. + #[mz_ore::test] + fn parses_explain_ref() { + let cmd = parse_command("explain ref=join").unwrap(); + assert_eq!( + cmd, + Command::Explain { + target: ExplainTarget::Reference { + name: "join".to_string(), + }, + } + ); + // A body is rejected: the declared dataflow supplies it. + assert!(parse_command("explain ref=join\n import source=1 shard=s upper=1").is_err()); + } + /// `create-dataflow` parses the sink export kinds: a materialized-view sink with /// a target shard, and a subscribe sink. The `copy-to` kind is rejected. #[mz_ore::test] @@ -844,4 +997,26 @@ mod tests { let actuals = vec!["ok".to_string(), "10000".to_string()]; assert_eq!(rewrite(&items, &actuals), content); } + + /// Output with blank lines uses the doubled-`----` form: it parses with the + /// blanks intact, and rewriting an actual that contains a blank line emits that + /// form (so it round-trips). + #[mz_ore::test] + fn blank_mode_round_trips() { + let content = "explain name=j as-of=0\n----\n----\nu2001:\n →Arrange (#0)\n\nu2000:\n →Stream u1000\n----\n----\n"; + let items = parse_file(content).unwrap(); + let Item::Stanza(stanza) = &items[0] else { + panic!("expected a stanza"); + }; + // The blank line between the two objects is preserved in the expected block. + assert_eq!( + stanza.expected, + "u2001:\n →Arrange (#0)\n\nu2000:\n →Stream u1000" + ); + // Rewriting with the same (blank-containing) output reproduces the file. + assert_eq!( + rewrite(&items, std::slice::from_ref(&stanza.expected)), + content + ); + } } diff --git a/test/clusterd-test-driver/scripts/join.spec b/test/clusterd-test-driver/scripts/join.spec index 4046592d12529..6571758c8dc05 100644 --- a/test/clusterd-test-driver/scripts/join.spec +++ b/test/clusterd-test-driver/scripts/join.spec @@ -53,6 +53,29 @@ create-dataflow name=join as-of=0 optimize ---- ok +# Assert the optimized plan shape, not just the result: `optimize` must pick a +# (differential) join, so optimizer or lowering drift that changed the plan would +# surface here. `explain ref=join` renders the dataflow declared above without +# repeating its body, and submits nothing. +explain ref=join +---- +---- +u2001: + →Arrange (#0) + →Stream u2000 + +u2000: + →Differential Join %0:u1000[#0] » %1:u1001[#0] + →Arrange (#0) + →Stream u1000 + →Arrange (#0) + →Stream u1001 + +Source u1000 +Source u1001 +---- +---- + schedule id=2001 ---- ok