diff --git a/interactive/examples/import_bench.rs b/interactive/examples/import_bench.rs
new file mode 100644
index 000000000..351239998
--- /dev/null
+++ b/interactive/examples/import_bench.rs
@@ -0,0 +1,147 @@
+//! What an importer pays for a shared trace: one producer publishes `edges` (keyed by
+//! source), then `K` consumers install, each a two-hop join from its own query roots over
+//! the imported `edges` (or, with `BENCH_QUERY=reach`, reachability: a join inside a loop).
+//! Reports each consumer's install-to-answer time and the process's memory, then the
+//! per-tick time under edge churn with all consumers live, and the published trace's size.
+//!
+//! ```text
+//! cargo run --release --example import_bench -- [edges] [consumers] [churn] [ticks] [nodes]
+//! DDIR_ARRANGED_IMPORTS=0 cargo run --release --example import_bench   # importers re-arrange
+//! DDIR_BACKEND=corgi cargo run --release --example import_bench         # columnar exports
+//! BENCH_QUERY=reach cargo run --release --example import_bench          # joins inside a loop
+//! ```
+
+use std::time::Instant;
+
+use interactive::ir::Value;
+use interactive::server::Server;
+use interactive::{lower, parse};
+
+fn install(server: &mut Server, worker: &mut timely::worker::Worker, name: &str, src: &str) {
+    let mut program = lower::lower_tree(parse::pipe::parse(src));
+    program.optimize();
+    server.install(worker, name, &program).unwrap();
+}
+
+/// This process's memory in MiB: on macOS its `footprint` (which, unlike RSS, counts compressed
+/// pages, so it does not shrink under memory pressure), elsewhere its RSS. `None` if unavailable.
+fn memory_mib() -> Option<f64> {
+    let out = std::process::Command::new("footprint").args(["-p", &std::process::id().to_string()]).output();
+    if let Ok(out) = out {
+        let text = String::from_utf8_lossy(&out.stdout).into_owned();
+        let mut words = text.split("Footprint:").nth(1)?.split_whitespace();
+        let n: f64 = words.next()?.parse().ok()?;
+        return match words.next()? {
+            "B" => Some(n / (1024.0 * 1024.0)),
+            "KB" => Some(n / 1024.0),
+            "MB" => Some(n),
+            "GB" => Some(n * 1024.0),
+            _ => None,
+        };
+    }
+    let status = std::fs::read_to_string("/proc/self/status").ok()?;
+    let kib: f64 = status.lines().find(|l| l.starts_with("VmRSS:"))?.split_whitespace().nth(1)?.parse().ok()?;
+    Some(kib / 1024.0)
+}
+
+/// [`memory_mib`], formatted.
+fn mem() -> String {
+    memory_mib().map_or_else(|| "n/a".to_string(), |m| format!("{m:>7.1} MiB"))
+}
+
+fn main() {
+    let args: Vec<u64> = std::env::args().skip(1).map(|a| a.parse().expect("integer argument")).collect();
+    let edges = args.first().copied().unwrap_or(1_000_000);
+    let consumers = args.get(1).copied().unwrap_or(8);
+    let churn = args.get(2).copied().unwrap_or(1_000);
+    let ticks = args.get(3).copied().unwrap_or(20);
+    let nodes = args.get(4).copied().unwrap_or(edges / 2);
+    let arranged = std::env::var("DDIR_ARRANGED_IMPORTS").map_or(true, |v| v != "0");
+    // BENCH_QUERY=reach: each consumer computes reachability, joining `edges` inside a loop.
+    let reach = std::env::var("BENCH_QUERY").as_deref() == Ok("reach");
+    let roots = if reach { 1 } else { 100 };
+
+    timely::execute_directly(move |worker| {
+        let backend = std::env::var("DDIR_BACKEND").unwrap_or("vec".into()).parse().expect("DDIR_BACKEND");
+        let mut server = Server::with_backend(backend);
+        server.set_arranged_imports(arranged);
+        println!("backend {backend:?}, arranged imports: {arranged}; {edges} edges over {nodes} nodes, {consumers} {} consumers, churn {churn}", if reach { "reach" } else { "two-hop" });
+
+        install(&mut server, worker, "prod", r#"export "edges" = input 0 | key($0[0] ; $0[1]);"#);
+        server.load(worker, "prod", 0, &format!("random:nodes={nodes},edges={edges},churn={churn}")).unwrap();
+        let loaded = Instant::now();
+        server.tick(worker);
+        println!("producer:    {:>8.1} ms  mem {}", loaded.elapsed().as_secs_f64() * 1e3, mem());
+
+        for c in 0..consumers {
+            let name = format!("q{c}");
+            let src = if reach {
+                // Reachability: the join with `edges` is inside an iterative scope.
+                format!(
+                    r#"
+                    let edges = import "edges";
+                    let roots = input 0 | key($0[0] ;);
+                    r: {{
+                        let step = reach | join(edges, ($2 ;));
+                        var reach = roots + step | distinct;
+                    }}
+                    export "reach{c}" = r::reach | key(;) | count;
+                    "#
+                )
+            } else {
+                format!(
+                    r#"
+                    let edges = import "edges";
+                    let roots = input 0 | key($0[0] ;);
+                    let hop1 = roots | join(edges, ($2 ; $0));
+                    let hop2 = hop1 | join(edges, ($2 ; $1));
+                    export "reach{c}" = hop2 | key($1 ;) | count;
+                    "#
+                )
+            };
+            let installed = Instant::now();
+            install(&mut server, worker, &name, &src);
+            for r in 0..roots {
+                let root = (interactive::hash_u64(c * 1000 + r) % nodes) as i64;
+                server.feed(&name, 0, Value::Tuple(vec![Value::Int(root)]), Value::unit(), None, 1).unwrap();
+            }
+            server.tick(worker);
+            println!("consumer {c:>2}: {:>8.1} ms  mem {}", installed.elapsed().as_secs_f64() * 1e3, mem());
+        }
+
+        let churned = Instant::now();
+        for _ in 0..ticks { server.tick(worker); }
+        println!(
+            "churn:       {:>8.2} ms/tick over {ticks} ticks ({churn} edges replaced per tick)  mem {}",
+            churned.elapsed().as_secs_f64() * 1e3 / ticks as f64, mem(),
+        );
+        let answer: i64 = (0..consumers)
+            .map(|c| server.snapshot(worker, &format!("reach{c}")).unwrap().iter().map(|(_, v, d)| ints(v) * d).sum::<i64>())
+            .sum();
+        println!("checksum (sum of consumer answers): {answer}");
+        {
+            use differential_dataflow::trace::TraceReader;
+            use differential_dataflow::trace::implementations::spine_fueled::SpineBatch;
+            use differential_dataflow::trace::chunk::Chunk;
+            let (mut updates, mut spans) = (0, 0);
+            match server.published("edges") {
+                Some(interactive::server::Published::Rows(trace)) => {
+                    trace.map_spans(|span| { spans += 1; if let Some(b) = &span.inner { updates += b.len(); } });
+                }
+                Some(interactive::server::Published::Columnar(trace)) => {
+                    trace.map_spans(|span| { spans += 1; if let Some(b) = &span.inner { updates += b.chunks.iter().map(Chunk::len).sum::<usize>(); } });
+                }
+                None => {}
+            }
+            println!("published edges trace: {updates} updates in {spans} spans");
+        }
+    });
+}
+
+fn ints(v: &Value) -> i64 {
+    match v {
+        Value::Tuple(xs) => xs.iter().map(ints).sum(),
+        Value::Int(n) => *n,
+        _ => 0,
+    }
+}
diff --git a/interactive/server/src/loop_.rs b/interactive/server/src/loop_.rs
index 0fb2ae8ec..7d583dcba 100644
--- a/interactive/server/src/loop_.rs
+++ b/interactive/server/src/loop_.rs
@@ -148,6 +148,9 @@ pub fn run_worker(
 
     let mut server = Server::with_backend(backend);
     server.set_columnar_exports(columnar_exports);
+    // Published traces reach importers as arrangements; DDIR_ARRANGED_IMPORTS=0 sends rows
+    // instead (each importer re-arranges what it joins).
+    server.set_arranged_imports(std::env::var("DDIR_ARRANGED_IMPORTS").map_or(true, |v| v != "0"));
     let mut tails: HashMap<TailKey, Tail> = HashMap::new();
     let mut responses: HashMap<u64, Sender<String>> = HashMap::new();
     let mut next_token = 0u64;
diff --git a/interactive/src/backend/corgi.rs b/interactive/src/backend/corgi.rs
index b0badeb57..62955b49b 100644
--- a/interactive/src/backend/corgi.rs
+++ b/interactive/src/backend/corgi.rs
@@ -11,11 +11,15 @@
 use timely::dataflow::Scope;
 use timely::dataflow::channels::pact::Pipeline;
 use timely::dataflow::operators::generic::Operator;
+use timely::progress::Antichain;
 
 use differential_dataflow::AsCollection;
 use differential_dataflow::Collection;
-use differential_dataflow::operators::join::join_with_tactic;
-use differential_dataflow::operators::reduce::reduce_with_tactic;
+use differential_dataflow::operators::join::{Fresh, JoinTactic, join_with_tactic};
+use differential_dataflow::operators::reduce::{ReduceTactic, reduce_with_tactic};
+use differential_dataflow::trace::Span;
+use differential_dataflow::trace::chunk::ChunkBatch;
+use differential_dataflow::trace::wrappers::enter::{BatchEnter, TraceEnter};
 use differential_dataflow::operators::arrange::arrangement::arrange_core;
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
 use differential_dataflow::trace::chunk::{Chunk, ChunkBatcher};
@@ -191,6 +195,101 @@ fn apply_ops(mut c: CC, ops: &[LinearOp], level: usize, plans: &mut [Plan]) -> C
     c
 }
 
+/// A batch of the program's own corgi arrangements (`CTrace`'s batch).
+type NativeBatch = std::rc::Rc<ChunkBatch<CorgiChunk<Time, Diff>>>;
+/// A batch of a published (host-time) corgi export (`ExportTrace`'s batch).
+type HostBatch = std::rc::Rc<ChunkBatch<CorgiChunk<u64, Diff>>>;
+/// A published corgi export entered into a program's scope: the producer's own chunks.
+pub type ImportArr<'scope> = Arranged<'scope, TraceEnter<ExportTrace, Time>>;
+
+/// The corgi backend's arrangement: one it built, or a published export it imported.
+#[derive(Clone)]
+pub enum Arr<'scope> {
+    Local(Arranged<'scope, TraceAgent<CTrace>>),
+    Imported(ImportArr<'scope>),
+}
+
+/// A batch as the corgi tactics read it. An imported batch crosses into the program as its
+/// producer's chunks *retimed*: the same `Rc`s, read with each host epoch `e` as `(e, [])`. The
+/// lanes are identical, so the crossing costs one `Rc` per chunk and copies no data.
+trait Native {
+    fn native(self) -> NativeBatch;
+}
+impl Native for NativeBatch {
+    fn native(self) -> NativeBatch { self }
+}
+impl Native for BatchEnter<HostBatch, Time> {
+    fn native(self) -> NativeBatch {
+        std::rc::Rc::new(ChunkBatch { chunks: self.inner().chunks.iter().map(CorgiChunk::retime).collect() })
+    }
+}
+
+/// A join tactic over native batches, accepting either input imported: the bridge where an
+/// imported batch's times change to the program's.
+struct RetimeJoin<J>(J);
+impl<B0: Native, B1: Native, J: JoinTactic<Time, NativeBatch, NativeBatch, CC>> JoinTactic<Time, B0, B1, CC> for RetimeJoin<J> {
+    fn prep(&mut self, input0: Vec<B0>, input1: Vec<B1>, fresh: Fresh, meet: Time) -> Box<dyn Iterator<Item = CC>> {
+        let input0 = input0.into_iter().map(Native::native).collect();
+        let input1 = input1.into_iter().map(Native::native).collect();
+        self.0.prep(input0, input1, fresh, meet)
+    }
+}
+
+/// A reduce tactic over native batches, accepting an imported input. (Its output is its own.)
+struct RetimeReduce<R>(R);
+impl<B1: Native, R: ReduceTactic<Time, NativeBatch, NativeBatch>> ReduceTactic<Time, B1, NativeBatch> for RetimeReduce<R> {
+    fn retire(
+        &mut self,
+        source_batches: Vec<B1>,
+        output_batches: Vec<NativeBatch>,
+        input_batches: Vec<B1>,
+        lower: &Antichain<Time>,
+        upper: &Antichain<Time>,
+        held: &Antichain<Time>,
+    ) -> (Option<Span<Time, NativeBatch>>, Antichain<Time>) {
+        let source_batches = source_batches.into_iter().map(Native::native).collect();
+        let input_batches = input_batches.into_iter().map(Native::native).collect();
+        self.0.retire(source_batches, output_batches, input_batches, lower, upper, held)
+    }
+}
+
+/// The updates of a stream of columnar batches, as corgi containers.
+///
+/// Each chunk already IS a columnar container: its key/val columns clone by Arc bump,
+/// so a chunk becomes a `CorgiContainer` for the price of copying its time lanes and
+/// its diffs. One container per chunk — no concatenation, no gather, no
+/// columns→rows→columns round-trip. `chunks` presents a batch's chunks at the stream's time,
+/// which retimes an imported batch on the way.
+fn chunk_containers<'s, T: ColTime, B: Clone + 'static>(
+    stream: timely::dataflow::Stream<'s, T, Vec<Span<T, B>>>,
+    name: &str,
+    chunks: impl Fn(B) -> std::rc::Rc<ChunkBatch<CorgiChunk<T, Diff>>> + 'static,
+) -> Collection<'s, T, CorgiContainer<T, Diff>> {
+    stream
+        .unary(Pipeline, name, move |_, _| {
+            move |input, output| {
+                input.for_each(|cap, data| {
+                    let mut session = output.session(&cap);
+                    for batch in data.iter() {
+                        let Some(payload) = batch.inner.clone() else { continue };
+                        for ch in chunks(payload).chunks.iter().filter(|c| c.len() > 0) {
+                            let mut c = CorgiContainer {
+                                // Drop the arrangement's leading identifier lane: edges carry
+                                // the key the program wrote, so `$0` indexes what it always did.
+                                keys: recover_key(ch.keys()),
+                                vals: ch.vals().clone(),
+                                times: ch.times().clone(),
+                                diffs: ch.diffs().to_vec(),
+                            };
+                            session.give_container(&mut c);
+                        }
+                    }
+                });
+            }
+        })
+        .as_collection()
+}
+
 /// The corgi rendering substrate. An uninhabited type used only as a type-level tag: it
 /// carries the [`Backend`] impl (a namespace of rendering functions selected by type) and is
 /// never a value — rendering goes through `render_tree::<CorgiBackend>`. The empty enum (vs a
@@ -199,7 +298,7 @@ pub enum CorgiBackend {}
 
 impl Backend for CorgiBackend {
     type Container = CC;
-    type Arr<'scope> = Arranged<'scope, TraceAgent<CTrace>>;
+    type Arr<'scope> = Arr<'scope>;
 
     fn linear<'s>(c: Collection<'s, Time, CC>, ops: Vec<LinearOp>, level: usize, shape: Option<st::RowShape>) -> Collection<'s, Time, CC> {
         // Container-level: fold the LinearOp chain over each corgi batch (no inter-op transcode).
@@ -235,20 +334,19 @@ impl Backend for CorgiBackend {
         // Column-native ingest: `CorgiChunker` sort-consolidates each input `CorgiContainer`'s
         // columns straight into a `CorgiChunk` (no drain-to-rows), then the standard chunk batcher +
         // builder. No columns→rows→columns round-trip at the arrangement boundary.
-        arrange_core::<_, CC, _, CTrace>(
+        Arr::Local(arrange_core::<_, CC, _, CTrace>(
             c.inner,
             CorgiPact,
             "CorgiArrange",
             ChunkBatcher::<CorgiChunker<Time, Diff>, _>::new,
-        )
-    }
-
-    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: timely::dataflow::Scope<'r, Time>) -> Self::Arr<'r> {
-        a.enter_region(region)
+        ))
     }
 
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, CC> {
-        chunk_containers(a, "CorgiAsCollection")
+        match a {
+            Arr::Local(a) => chunk_containers(a.stream, "CorgiAsCollection", Native::native),
+            Arr::Imported(a) => chunk_containers(a.stream, "CorgiAsCollection", Native::native),
+        }
     }
 
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, CC> {
@@ -257,14 +355,22 @@ impl Backend for CorgiBackend {
         // are known and otherwise on its first output, and emits corgi columns directly as
         // `CorgiContainer`s — column-native, no row round-trip.
         let shapes = shapes.map(|((k, v0), (_, v1))| (k, v0, v1));
-        let tactic = ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone(), shapes));
-        join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic).as_collection()
+        let tactic = || RetimeJoin(ProxyJoinTactic::new(CorgiJoinBackend::new(projection.key.clone(), projection.val.clone(), shapes.clone())));
+        match (l, r) {
+            (Arr::Local(l), Arr::Local(r)) => join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic()).as_collection(),
+            (Arr::Local(l), Arr::Imported(r)) => join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic()).as_collection(),
+            (Arr::Imported(l), Arr::Local(r)) => join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic()).as_collection(),
+            (Arr::Imported(l), Arr::Imported(r)) => join_with_tactic::<_, _, _, CC>(l, r, "Join", tactic()).as_collection(),
+        }
     }
 
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         // Amortize columnar corrections while reusing sweep scratch across wide presentations.
-        let tactic = ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())).with_key_batch_size(256);
-        reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", tactic)
+        let tactic = RetimeReduce(ProxyReduceTactic::new(CorgiReduceBackend::new(reducer.clone())).with_key_batch_size(256));
+        Arr::Local(match a {
+            Arr::Local(a) => reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", tactic),
+            Arr::Imported(a) => reduce_with_tactic::<_, CTrace, _>(a, "CorgiReduce", tactic),
+        })
     }
 
     fn inspect<'s>(c: Collection<'s, Time, CC>, label: String) -> Collection<'s, Time, CC> {
@@ -329,41 +435,12 @@ impl Backend for CorgiBackend {
 
         stream.as_collection()
     }
-}
-
-/// The updates of a columnar arrangement as a collection of corgi containers.
-///
-/// Each chunk already IS a columnar container: its key/val columns clone by Arc bump,
-/// so a chunk becomes a `CorgiContainer` for the price of copying its time lanes and
-/// its diffs. One container per chunk — no concatenation, no gather, no
-/// columns→rows→columns round-trip.
-fn chunk_containers<'s, T: ColTime>(
-    a: Arranged<'s, TraceAgent<differential_dataflow::trace::chunk::ChunkSpine<CorgiChunk<T, Diff>>>>,
-    name: &str,
-) -> Collection<'s, T, CorgiContainer<T, Diff>> {
-    a.stream
-        .unary(Pipeline, name, |_, _| {
-            |input, output| {
-                input.for_each(|cap, data| {
-                    let mut session = output.session(&cap);
-                    for batch in data.iter() {
-                        let Some(payload) = batch.inner.as_ref() else { continue };
-                        for ch in payload.chunks.iter().filter(|c| c.len() > 0) {
-                            let mut c = CorgiContainer {
-                                // Drop the arrangement's leading identifier lane: edges carry
-                                // the key the program wrote, so `$0` indexes what it always did.
-                                keys: recover_key(ch.keys()),
-                                vals: ch.vals().clone(),
-                                times: ch.times().clone(),
-                                diffs: ch.diffs().to_vec(),
-                            };
-                            session.give_container(&mut c);
-                        }
-                    }
-                });
-            }
-        })
-        .as_collection()
+    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: Scope<'r, Time>) -> Self::Arr<'r> {
+        match a {
+            Arr::Local(a) => Arr::Local(a.enter_region(region)),
+            Arr::Imported(a) => Arr::Imported(a.enter_region(region)),
+        }
+    }
 }
 
 /// Convert a collection of rows to corgi containers, at `shape` if it is known and otherwise at
@@ -424,6 +501,7 @@ pub fn render_tree<'s>(
     imports: Vec<Collection<'s, Time, CC>>,
     shapes: Option<&crate::shapes::ScopeShapes>,
 ) -> Vec<Collection<'s, Time, CC>> {
+    let imports = imports.into_iter().map(crate::backend::Rendered::Collection).collect();
     crate::backend::render_tree::<CorgiBackend>(s, scope, depth, imports, shapes)
 }
 
@@ -443,16 +521,42 @@ pub fn render_tree_corgi<'s>(
         .into_iter()
         .zip(&s.imports)
         .enumerate()
-        .map(|(k, (c, import))| {
-            // The declared shape, or the one inferred at install (an imported trace's).
-            let shape = import.shape.clone()
-                .or_else(|| shapes.and_then(|sh| sh.imports[k].clone()).map(|c| (c.key, c.val)));
-            rows_to_corgi(c, shape)
-        })
+        .map(|(k, (c, import))| rows_to_corgi(c, import_shape(k, import, shapes)))
         .collect();
     render_tree(s, scope, depth, corgi_imports, shapes)
 }
 
+/// Import `k`'s declared shape, or the one inferred at install (an imported trace's).
+fn import_shape(k: usize, import: &st::Import, shapes: Option<&crate::shapes::ScopeShapes>) -> Option<st::RowShape> {
+    import.shape.clone().or_else(|| shapes.and_then(|sh| sh.imports[k].clone()).map(|c| (c.key, c.val)))
+}
+
+/// A root import for [`render_tree_corgi_arranged`]: corgi containers, or a published columnar
+/// export.
+#[allow(clippy::large_enum_variant)] // One per import, consumed as the program renders.
+pub enum CorgiImport<'s> {
+    Columns(Collection<'s, Time, CC>),
+    Export(ImportArr<'s>),
+}
+
+/// As [`render_tree`], but a published columnar export arrives as the producer's arrangement,
+/// which the program's `arrange`s, joins, and reduces of it read in place. Shape ascriptions are
+/// the driver's to check ([`assert_shape`], [`assert_export_shape`]).
+pub fn render_tree_corgi_arranged<'s>(
+    s: &st::Scope,
+    scope: Scope<'s, Time>,
+    depth: usize,
+    imports: Vec<CorgiImport<'s>>,
+    shapes: Option<&crate::shapes::ScopeShapes>,
+) -> Vec<Collection<'s, Time, CC>> {
+    assert_eq!(s.imports.len(), imports.len());
+    let imports = imports.into_iter().map(|imp| match imp {
+        CorgiImport::Columns(c) => crate::backend::Rendered::Collection(c),
+        CorgiImport::Export(a) => crate::backend::Rendered::Arrangement(Arr::Imported(a)),
+    }).collect();
+    crate::backend::render_tree::<CorgiBackend>(s, scope, depth, imports, shapes)
+}
+
 /// A columnar export: the program's corgi collection, left to the host time and arranged as
 /// corgi chunks. Readers convert to rows only when they read ([`export_rows`]).
 pub type ExportTrace = TraceAgent<differential_dataflow::trace::chunk::ChunkSpine<CorgiChunk<u64, Diff>>>;
@@ -529,7 +633,25 @@ pub fn tap_export<'s>(a: &Arranged<'s, ExportTrace>, tap: ExportTap) -> timely::
 
 /// The updates of an imported columnar export, as corgi containers.
 pub fn export_containers<'s>(a: Arranged<'s, ExportTrace>) -> Collection<'s, u64, CorgiContainer<u64, Diff>> {
-    chunk_containers(a, "ExportContainers")
+    chunk_containers(a.stream, "ExportContainers", |batch| batch)
+}
+
+/// Assert that every chunk of an imported export has `shape`, an import's ascription, when the
+/// export's shape was not known at install. A side operator over the batches: the program reads
+/// the arrangement itself, and nothing is copied.
+pub fn assert_export_shape(a: &Arranged<'_, ExportTrace>, shape: st::RowShape) {
+    use timely::dataflow::operators::generic::Operator;
+    a.stream.clone().sink(Pipeline, "AssertShape", move |(input, _frontier)| {
+        input.for_each(|_cap, data| {
+            for batch in data.iter() {
+                let Some(payload) = batch.inner.as_ref() else { continue };
+                for ch in payload.chunks.iter().filter(|c| c.len() > 0) {
+                    let (k, v) = (corgi::shape_of_value(&recover_key(ch.keys())), corgi::shape_of_value(ch.vals()));
+                    assert!(k == shape.0 && v == shape.1, "input does not match its shape ascription");
+                }
+            }
+        });
+    });
 }
 
 /// The rows of an imported columnar export, as `((key, val), time, diff)` updates.
diff --git a/interactive/src/backend/mod.rs b/interactive/src/backend/mod.rs
index 0d0f8d745..28eaa8351 100644
--- a/interactive/src/backend/mod.rs
+++ b/interactive/src/backend/mod.rs
@@ -55,27 +55,46 @@ pub trait Backend {
 }
 
 /// A rendered item's value: a collection, or an arrangement.
-enum Rendered<'s, B: Backend> {
+///
+/// Imports are rendered values too, so a driver can hand a scope an arrangement it already
+/// maintains (a published trace) and the scope's `arrange`s and joins of it reuse it as is.
+pub enum Rendered<'s, B: Backend> {
     Collection(Collection<'s, Time, B::Container>),
     Arrangement(B::Arr<'s>),
+    /// An imported arrangement that the scope also reads as rows: both forms, so the rows
+    /// are flattened out of it once rather than once per use.
+    Both(B::Arr<'s>, Collection<'s, Time, B::Container>),
+}
+
+impl<'s, B: Backend> Clone for Rendered<'s, B> {
+    fn clone(&self) -> Self {
+        match self {
+            Rendered::Collection(c) => Rendered::Collection(c.clone()),
+            Rendered::Arrangement(a) => Rendered::Arrangement(a.clone()),
+            Rendered::Both(a, c) => Rendered::Both(a.clone(), c.clone()),
+        }
+    }
 }
 
 impl<'s, B: Backend> Rendered<'s, B> {
+    /// Bring the value into a child region. An arrangement enters as the same trace, and rows
+    /// already flattened from one enter with it.
     fn enter<'r>(&self, region: Scope<'r, Time>) -> Rendered<'r, B> {
         match self {
             Rendered::Collection(c) => Rendered::Collection(c.clone().enter(region)),
             Rendered::Arrangement(a) => Rendered::Arrangement(B::enter_region(a.clone(), region)),
+            Rendered::Both(a, c) => Rendered::Both(B::enter_region(a.clone(), region), c.clone().enter(region)),
         }
     }
-    fn collection(&self) -> Collection<'s, Time, B::Container> {
+    pub fn collection(&self) -> Collection<'s, Time, B::Container> {
         match self {
-            Rendered::Collection(c) => c.clone(),
+            Rendered::Collection(c) | Rendered::Both(_, c) => c.clone(),
             Rendered::Arrangement(a) => B::as_collection(a.clone()),
         }
     }
     fn arrange(&self) -> B::Arr<'s> {
         match self {
-            Rendered::Arrangement(a) => a.clone(),
+            Rendered::Arrangement(a) | Rendered::Both(a, _) => a.clone(),
             Rendered::Collection(c) => B::arrange(c.clone()),
         }
     }
@@ -96,14 +115,10 @@ fn resolve<'s, B: Backend>(
 ) -> Rendered<'s, B> {
     match r {
         st::Ref::Local(i) => match &items[*i] {
-            RItem::Op(Rendered::Collection(c)) => Rendered::Collection(c.clone()),
-            RItem::Op(Rendered::Arrangement(a)) => Rendered::Arrangement(a.clone()),
+            RItem::Op(rendered) => rendered.clone(),
             RItem::Sub(_) => panic!("Ref::Local points at a child scope"),
         },
-        st::Ref::Import(i) => match &imports[*i] {
-            Rendered::Collection(c) => Rendered::Collection(c.clone()),
-            Rendered::Arrangement(a) => Rendered::Arrangement(a.clone()),
-        },
+        st::Ref::Import(i) => imports[*i].clone(),
         st::Ref::Var(i) => Rendered::Collection(var_cols[*i].clone()),
         st::Ref::ChildExport(i, j) => match &items[*i] {
             RItem::Sub(exports) => Rendered::Collection(exports[*j].clone()),
@@ -112,24 +127,24 @@ fn resolve<'s, B: Backend>(
     }
 }
 
+/// Whether scope `s` reads import `k` as rows: as a linear op's, concat's, or inspect's input,
+/// or as a bind or export. A child's import is not a read: the arrangement enters the child.
+fn reads_rows(s: &st::Scope, k: usize) -> bool {
+    let is_k = |r: &st::Ref| matches!(r, st::Ref::Import(i) if *i == k);
+    s.items.iter().any(|item| match item {
+        st::Item::Op(st::Node::Linear { input, .. } | st::Node::Inspect { input, .. }) => is_k(input),
+        st::Item::Op(st::Node::Concat(refs)) => refs.iter().any(is_k),
+        st::Item::Op(st::Node::Arrange(_) | st::Node::Join { .. } | st::Node::Reduce { .. }) => false,
+        st::Item::Sub(_) => false,
+    }) || s.binds.iter().any(|b| is_k(&b.value)) || s.exports.iter().any(|e| is_k(&e.value))
+}
+
 /// Render one scope at `depth` (root = 0): feedback vars first (they're listed,
 /// not scanned for), then items in order, then binds close the loops, then the
 /// exports are surrendered. The returned collections are at this scope's depth;
 /// popping the coordinate (`leave_dynamic`) is the caller's job. `shapes`, where
 /// inference ran, gives the operators their inputs' shapes ahead of the data.
 pub fn render_tree<'s, B: Backend>(
-    s: &st::Scope,
-    scope: Scope<'s, Time>,
-    depth: usize,
-    imports: Vec<Collection<'s, Time, B::Container>>,
-    shapes: Option<&ScopeShapes>,
-) -> Vec<Collection<'s, Time, B::Container>> {
-    render_scope::<B>(s, scope, depth, imports.into_iter().map(Rendered::Collection).collect(), shapes)
-}
-
-/// [`render_tree`] at any depth. A child's imports are the parent's items as rendered, so an
-/// arrangement crosses into the child as the same trace, not as rows to arrange again.
-fn render_scope<'s, B: Backend>(
     s: &st::Scope,
     scope: Scope<'s, Time>,
     depth: usize,
@@ -137,6 +152,14 @@ fn render_scope<'s, B: Backend>(
     shapes: Option<&ScopeShapes>,
 ) -> Vec<Collection<'s, Time, B::Container>> {
     let shape_of = |r: &st::Ref| shapes.and_then(|sh| sh.of(r)).map(|c| (c.key, c.val));
+    // An arranged import that is also read as rows gets its rows flattened once, here.
+    let imports: Vec<Rendered<'s, B>> = imports.into_iter().enumerate().map(|(k, imp)| match imp {
+        Rendered::Arrangement(a) if reads_rows(s, k) => {
+            let rows = B::as_collection(a.clone());
+            Rendered::Both(a, rows)
+        }
+        other => other,
+    }).collect();
     let mut var_handles: Vec<Option<Variable<'s, Time, B::Container>>> = Vec::new();
     let mut var_cols: Vec<Collection<'s, Time, B::Container>> = Vec::new();
     for _ in &s.vars {
@@ -186,9 +209,9 @@ fn render_scope<'s, B: Backend>(
                 // the child renders inside, exports leave it structurally —
                 // and then pop the child's dynamic coordinate.
                 let exported = scope.region_named(&child.name, |region| {
-                    let entered: Vec<_> = child_imports.iter().map(|c| c.enter(region)).collect();
+                    let entered: Vec<_> = child_imports.iter().map(|imp| imp.enter(region)).collect();
                     let child_shapes = shapes.and_then(|sh| match sh.items.get(index) { Some(ItemShapes::Sub(c)) => Some(c), _ => None });
-                    let exports = render_scope::<B>(child, region, depth + 1, entered, child_shapes);
+                    let exports = render_tree::<B>(child, region, depth + 1, entered, child_shapes);
                     exports.into_iter().map(|c| c.leave(scope)).collect::<Vec<_>>()
                 });
                 let left: Vec<Collection<'s, Time, B::Container>> = exported.into_iter().map(|c| B::leave_dynamic(c, depth + 1)).collect();
diff --git a/interactive/src/backend/vec.rs b/interactive/src/backend/vec.rs
index 547814df7..8ef431415 100644
--- a/interactive/src/backend/vec.rs
+++ b/interactive/src/backend/vec.rs
@@ -14,10 +14,11 @@ use differential_dataflow::{Collection, VecCollection};
 use differential_dataflow::dynamic::pointstamp::PointStamp;
 use differential_dataflow::trace::implementations::{ValSpine, ValBuilder};
 use differential_dataflow::operators::arrange::{Arranged, TraceAgent};
+use differential_dataflow::trace::wrappers::enter::TraceEnter;
 use smallvec::SmallVec;
 use smallvec::smallvec as svec;
 
-use crate::backend::Backend;
+use crate::backend::{Backend, Rendered};
 use crate::scope_ir as st;
 use crate::ir::{LinearOp, Diff, Projection, Reducer, Time, Value, eval};
 
@@ -31,19 +32,35 @@ pub type Col<'scope> = VecCollection<'scope, Time, (Row, Row), Diff>;
 /// server on either backend. This is not transactional feed admission or
 /// per-program failure isolation.
 pub(crate) fn check_import_shapes<'s>(s: &st::Scope, imports: Vec<Col<'s>>) -> Vec<Col<'s>> {
+    assert_eq!(s.imports.len(), imports.len());
+    imports.into_iter().zip(&s.imports).map(|(c, import)| check_shape(c, import)).collect()
+}
+
+/// Check one import's rows against its shape ascription, if it has one.
+pub(crate) fn check_shape<'s>(c: Col<'s>, import: &st::Import) -> Col<'s> {
     use differential_dataflow::AsCollection;
     use timely::dataflow::operators::core::Map;
-    assert_eq!(s.imports.len(), imports.len());
-    imports.into_iter().zip(&s.imports).map(|(c, import)| {
-        if let Some((key, val)) = import.shape.clone() {
-            c.inner.map(move |row| {
-                assert!(row.0.0.has_shape(&key) && row.0.1.has_shape(&val), "input does not match its shape ascription");
-                row
-            }).as_collection()
-        } else { c }
-    }).collect()
+    if let Some((key, val)) = import.shape.clone() {
+        c.inner.map(move |row| {
+            assert!(row.0.0.has_shape(&key) && row.0.1.has_shape(&val), "input does not match its shape ascription");
+            row
+        }).as_collection()
+    } else { c }
+}
+
+/// An arrangement this program built, at the renderer's time.
+pub type LocalArr<'scope> = Arranged<'scope, TraceAgent<ValSpine<Row, Row, Time, Diff>>>;
+/// A published trace (arranged at the server's host time, `u64`), entered into the program's
+/// scope: the producer's own batches, read with each host time `t` as `(t, [])`.
+pub type ImportArr<'scope> = Arranged<'scope, TraceEnter<TraceAgent<ValSpine<Row, Row, u64, Diff>>, Time>>;
+
+/// The vec backend's arrangement: one it built, or one it imported. Join and reduce accept
+/// either (they read batches through cursors, and an entered batch's cursor lifts its times).
+#[derive(Clone)]
+pub enum Arr<'scope> {
+    Local(LocalArr<'scope>),
+    Imported(ImportArr<'scope>),
 }
-type Arr<'scope> = Arranged<'scope, TraceAgent<ValSpine<Row, Row, Time, Diff>>>;
 
 /// Append the user-iter coordinate to a value: extend a `Tuple` in place, or
 /// wrap any other value as `(value, iter)`.
@@ -135,14 +152,13 @@ impl Backend for VecBackend {
         render_linear(c, ops, level)
     }
     fn arrange<'s>(c: Collection<'s, Time, Self::Container>) -> Self::Arr<'s> {
-        c.arrange_by_key()
-    }
-    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: timely::dataflow::Scope<'r, Time>) -> Self::Arr<'r> {
-        a.enter_region(region)
+        Arr::Local(c.arrange_by_key())
     }
-
     fn as_collection<'s>(a: Self::Arr<'s>) -> Collection<'s, Time, Self::Container> {
-        a.as_collection(|k, v| (k.clone(), v.clone()))
+        match a {
+            Arr::Local(a) => a.as_collection(|k, v| (k.clone(), v.clone())),
+            Arr::Imported(a) => a.as_collection(|k, v| (k.clone(), v.clone())),
+        }
     }
     fn join<'s>(l: Self::Arr<'s>, r: Self::Arr<'s>, projection: &Projection, _shapes: Option<(st::RowShape, st::RowShape)>) -> Collection<'s, Time, Self::Container> {
         let proj = projection.clone();
@@ -153,7 +169,12 @@ impl Backend for VecBackend {
                 let v = eval(&proj.val, &mut env);
                 svec![(k, v)]
             });
-        l.join_core(r, move |k, v1, v2| f(k, v1, v2))
+        match (l, r) {
+            (Arr::Local(l), Arr::Local(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
+            (Arr::Local(l), Arr::Imported(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
+            (Arr::Imported(l), Arr::Local(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
+            (Arr::Imported(l), Arr::Imported(r)) => l.join_core(r, move |k, v1, v2| f(k, v1, v2)),
+        }
     }
     fn reduce<'s>(a: Self::Arr<'s>, reducer: &Reducer) -> Self::Arr<'s> {
         let f: Arc<dyn Fn(&Row, &[(&Row, Diff)], &mut Vec<(Row, Diff)>) + Send + Sync> = match reducer {
@@ -170,11 +191,20 @@ impl Backend for VecBackend {
                 output.push((Value::List(items), 1));
             }),
         };
-        a.reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine<_, _, _, _>, _, _>(
-            "Reduce",
-            move |k, v, o| f(k, v, o),
-            |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
-        )
+        // The same reduction over either arrangement; its output is always local.
+        macro_rules! reduce {
+            ($a:expr) => {
+                $a.reduce_abelian::<_, ValBuilder<_, _, _, _>, ValSpine<_, _, _, _>, _, _>(
+                    "Reduce",
+                    move |k, v, o| f(k, v, o),
+                    |vec, key, upds| { vec.clear(); vec.extend(upds.drain(..).map(|(v, t, r)| ((key.clone(), v), t, r))); },
+                )
+            };
+        }
+        Arr::Local(match a {
+            Arr::Local(a) => reduce!(a),
+            Arr::Imported(a) => reduce!(a),
+        })
     }
     fn inspect<'s>(c: Collection<'s, Time, Self::Container>, label: String) -> Collection<'s, Time, Self::Container> {
         use std::fmt::Write;
@@ -189,6 +219,12 @@ impl Backend for VecBackend {
     fn leave_dynamic<'s>(c: Collection<'s, Time, Self::Container>, depth: usize) -> Collection<'s, Time, Self::Container> {
         c.leave_dynamic(depth)
     }
+    fn enter_region<'s, 'r>(a: Self::Arr<'s>, region: Scope<'r, Time>) -> Self::Arr<'r> {
+        match a {
+            Arr::Local(a) => Arr::Local(a.enter_region(region)),
+            Arr::Imported(a) => Arr::Imported(a.enter_region(region)),
+        }
+    }
 }
 
 /// Render `s` with the vec substrate. See [`crate::backend::render_tree`].
@@ -199,5 +235,25 @@ pub fn render_tree<'s>(
     imports: Vec<Col<'s>>,
     shapes: Option<&crate::shapes::ScopeShapes>,
 ) -> Vec<Col<'s>> {
-    crate::backend::render_tree::<VecBackend>(s, scope, depth, check_import_shapes(s, imports), shapes)
+    let imports = check_import_shapes(s, imports).into_iter().map(Rendered::Collection).collect();
+    crate::backend::render_tree::<VecBackend>(s, scope, depth, imports, shapes)
+}
+
+/// As [`render_tree`], but an import may arrive already arranged (a published trace), in
+/// which case the program's `arrange`s and joins of it read the producer's arrangement
+/// instead of building their own. Shape ascriptions are checked here for imports that
+/// arrive as rows; checking an arranged import's ascription is the driver's job.
+pub fn render_tree_arranged<'s>(
+    s: &st::Scope,
+    scope: Scope<'s, Time>,
+    depth: usize,
+    imports: Vec<Rendered<'s, VecBackend>>,
+    shapes: Option<&crate::shapes::ScopeShapes>,
+) -> Vec<Col<'s>> {
+    assert_eq!(s.imports.len(), imports.len());
+    let imports = imports.into_iter().zip(&s.imports).map(|(rendered, import)| match rendered {
+        Rendered::Collection(c) if import.shape.is_some() => Rendered::Collection(check_shape(c, import)),
+        other => other,
+    }).collect();
+    crate::backend::render_tree::<VecBackend>(s, scope, depth, imports, shapes)
 }
diff --git a/interactive/src/corgi/chunk.rs b/interactive/src/corgi/chunk.rs
index 98b3a15e5..eb50b47d9 100644
--- a/interactive/src/corgi/chunk.rs
+++ b/interactive/src/corgi/chunk.rs
@@ -54,6 +54,10 @@ const INGEST: usize = 1 << 24;
 /// Same payload as [`CorgiContainer`](crate::corgi::container::CorgiContainer), and the two
 /// should eventually be ONE type: both hold times as a `ColTimes`, and they differ only in
 /// invariants (sorted+consolidated+shared here, raw+owned there).
+///
+/// `repr(C)`, and `T` appears only in `ColTimes<T>` (itself `repr(C)`, `T` in `PhantomData`), so
+/// `Inner<T, R>` and `Inner<U, R>` share a layout and drop glue: what [`CorgiChunk::retime`] needs.
+#[repr(C)]
 struct Inner<T, R> {
     /// Key column (corgi), aligned with `vals`/`times`/`diffs`, sorted by `(key, val, time)`.
     keys: CValue,
@@ -96,6 +100,17 @@ impl<T, R> CorgiChunk<T, R> {
         let (keys, vals) = split_kv(kv);
         Self::from_parts(keys, vals, times, diffs)
     }
+    /// The same chunk, shared rather than copied, read at time type `U`: a row's `U` time has the
+    /// row's lane coordinates followed by zeros. For a host-time (`u64`) chunk read as a program
+    /// time (`(u64, PointStamp)`), each epoch `e` reads as `(e, [])` — the time it has on entering
+    /// the program's scope — and the order and lattice operations, being lane-wise, agree.
+    pub fn retime<U>(&self) -> CorgiChunk<U, R> {
+        let raw = Rc::into_raw(Rc::clone(&self.0)).cast::<Inner<U, R>>();
+        // SAFETY: `Inner` is `repr(C)` and mentions `T`/`U` only inside the `repr(C)` `ColTimes`,
+        // in a `PhantomData`, so `Inner<T, R>` and `Inner<U, R>` have the same size, alignment,
+        // field offsets, and drop glue; `raw` came from `Rc::into_raw` of the former.
+        CorgiChunk(unsafe { Rc::from_raw(raw) })
+    }
     pub fn keys(&self) -> &CValue { &self.0.keys }
     pub fn vals(&self) -> &CValue { &self.0.vals }
     pub fn times(&self) -> &ColTimes<T> { &self.0.times }
@@ -950,6 +965,23 @@ mod test {
         }
     }
 
+    #[test]
+    fn retime_reads_host_epochs_as_program_times_without_copying() {
+        use differential_dataflow::dynamic::pointstamp::PointStamp;
+        use timely::order::Product;
+        type Program = Product<u64, PointStamp<u64>>;
+        let host = chunk(&[((1, 10), 0, 1), ((1, 11), 3, -1), ((2, 10), 7, 2)]);
+        let program: CorgiChunk<Program, i64> = host.retime();
+        // The same allocation, read at the program's time: epoch `e` as `(e, [])`.
+        assert!(std::ptr::eq(host.keys(), program.keys()));
+        let expect: Vec<Program> = host.times().to_vec().into_iter().map(|e| Product::new(e, PointStamp::default())).collect();
+        assert_eq!(program.times().to_vec(), expect);
+        assert_eq!(program.diffs(), host.diffs());
+        // Either handle may be the last to drop.
+        drop(host);
+        assert_eq!(program.times().len(), 3);
+    }
+
     /// Build a single sorted+consolidated CorgiChunk from u64 (key,val,time,diff) rows.
     fn chunk(rows: &[((u64, u64), u64, i64)]) -> CorgiChunk<u64, i64> {
         // Sort + consolidate by ((k,v),t) so a chunk is a legal sorted run.
diff --git a/interactive/src/corgi/col_times.rs b/interactive/src/corgi/col_times.rs
index b13890444..d5613c7aa 100644
--- a/interactive/src/corgi/col_times.rs
+++ b/interactive/src/corgi/col_times.rs
@@ -77,6 +77,10 @@ impl<B: LaneSummary> LaneSummary for Product<u64, B> {
 }
 
 /// A column of times as lanes: `lanes[j][i]` is coordinate `j` of row `i`.
+///
+/// `repr(C)` with `T` only in `PhantomData`, so every `ColTimes<T>` has one layout: a chunk's times
+/// can be read at another time type with the same lanes (see [`CorgiChunk::retime`](crate::corgi::chunk::CorgiChunk::retime)).
+#[repr(C)]
 pub struct ColTimes<T> {
     lanes: Vec<Vec<u64>>,
     len: usize,
diff --git a/interactive/src/server.rs b/interactive/src/server.rs
index cd6a899b4..b3ce26cdb 100644
--- a/interactive/src/server.rs
+++ b/interactive/src/server.rs
@@ -61,7 +61,7 @@ use timely::progress::Antichain;
 use timely::worker::Worker;
 
 use crate::backend::corgi::render_tree_rows;
-use crate::backend::vec::render_tree;
+use crate::backend::vec::render_tree_arranged;
 use crate::ir::{Diff, Value};
 use crate::scope_ir as st;
 
@@ -108,6 +108,18 @@ impl std::str::FromStr for RenderBackend {
 /// arranged by key at the host time so any later install can `import` it.
 pub type ServerTrace = TraceAgent<ValSpine<Value, Value, OuterTime, Diff>>;
 
+/// A root import at the host time, before it enters a program's scope.
+enum OuterImport<'s> {
+    /// An input, or a published trace read as rows.
+    Rows(VecCollection<'s, OuterTime, (Value, Value), Diff>),
+    /// A published row trace, for a vec program to read in place.
+    Trace(differential_dataflow::operators::arrange::Arranged<'s, ServerTrace>),
+    /// For a columnar program: an input, or a published trace read as corgi containers.
+    Columns(differential_dataflow::Collection<'s, OuterTime, HostContainer>),
+    /// A published columnar export, for a columnar program to read in place.
+    Export(differential_dataflow::operators::arrange::Arranged<'s, crate::backend::corgi::ExportTrace>),
+}
+
 /// A published export as its readers see it: a row trace, or with columnar exports
 /// ([`Server::set_columnar_exports`]) a trace of corgi chunks. Either reads as rows.
 #[derive(Clone)]
@@ -610,6 +622,9 @@ pub struct Server {
     epoch: OuterTime,
     /// Rendering substrate for subsequently installed programs.
     backend: RenderBackend,
+    /// Hand published traces to importing programs as arrangements (vec row traces, corgi
+    /// columnar exports), so an importer shares the producer's trace instead of copying it.
+    arranged_imports: bool,
 }
 
 impl Server {
@@ -632,6 +647,7 @@ impl Server {
             bindings: Vec::new(),
             epoch: 0,
             backend,
+            arranged_imports: true,
         }
     }
 
@@ -659,6 +675,15 @@ impl Server {
         self.backend == RenderBackend::Corgi && self.columnar_exports
     }
 
+    /// Import published traces into programs as arrangements (default on): row traces into vec
+    /// programs, columnar exports into corgi programs with columnar exports. Joins, reduces, and
+    /// `arrange`s of an import then read the producer's trace, at the root and inside iterative
+    /// scopes, rather than each importer rebuilding it. Off, imports arrive as rows, or in
+    /// columnar mode as containers. Affects programs installed afterwards.
+    pub fn set_arranged_imports(&mut self, on: bool) {
+        self.arranged_imports = on;
+    }
+
     /// With columnar exports, also keep each export's new batches for [`Server::take_changes`]
     /// and [`Server::for_each_change_batch`] (default off). Affects programs installed afterwards.
     pub fn set_export_taps(&mut self, on: bool) {
@@ -798,6 +823,7 @@ impl Server {
         let columnar = self.columnar();
         let traces = &mut self.traces;
         let backend = self.backend;
+        let arranged_imports = self.arranged_imports;
         let tapped = columnar && self.export_taps;
         let ctraces = &mut self.ctraces;
 
@@ -809,13 +835,13 @@ impl Server {
             worker.dataflow::<OuterTime, _, _>(|outer| {
                 let mut inputs: Vec<(usize, ServerInput)> = Vec::new();
 
-                // Columnar: one outer (host-time) collection of corgi containers per root import,
-                // rendered without rows. A columnar trace is read as its chunks, and an input
-                // converts rows to columns as they are fed.
+                // Columnar: one outer (host-time) value per root import, rendered without rows. An
+                // input converts rows to columns as they are fed, and a columnar trace is read as
+                // its chunks, or with arranged imports as the producer's arrangement itself.
                 let (leaved, cleaved) = if columnar {
                     use timely::dataflow::operators::core::Input as _;
-                    use crate::backend::corgi::{assert_shape, export_containers, rows_to_corgi};
-                    let outer_cols: Vec<differential_dataflow::Collection<OuterTime, HostContainer>> = root
+                    use crate::backend::corgi::{assert_export_shape, assert_shape, export_containers, rows_to_corgi, CorgiImport};
+                    let outer_imports: Vec<OuterImport> = root
                         .imports
                         .iter()
                         .enumerate()
@@ -823,29 +849,34 @@ impl Server {
                             // The declared shape, or the one inferred at install (an imported trace's).
                             let shape = imp.shape.clone()
                                 .or_else(|| scope_shapes.imports[k].clone().map(|c| (c.key, c.val)));
+                            let checked = |cols| match unchecked[k].clone() {
+                                Some(declared) => assert_shape(cols, declared),
+                                None => cols,
+                            };
                             match &imp.from {
                                 st::Source::Input(n) => {
                                     let (handle, stream) = outer.new_input::<HostContainer>();
                                     inputs.push((*n, ServerInput::Columns(ColumnInput::new(handle, shape, imp.shape.is_some()))));
-                                    differential_dataflow::AsCollection::as_collection(stream)
+                                    OuterImport::Columns(differential_dataflow::AsCollection::as_collection(stream))
                                 }
                                 st::Source::Trace(t) => {
                                     // The first binding point: resolve a named trace by importing it.
                                     let key = canonical_source_name(t);
-                                    let cols = match ctraces.get_mut(&key) {
-                                        Some(ctrace) => export_containers(ctrace.import(outer.clone())),
-                                        None => {
-                                            let arranged = traces
-                                                .get_mut(&key)
-                                                .expect("validated above")
-                                                .import(outer.clone());
-                                            rows_to_corgi(arranged.as_collection(|k, v| (k.clone(), v.clone())), shape)
+                                    if let Some(ctrace) = ctraces.get_mut(&key) {
+                                        let arranged = ctrace.import(outer.clone());
+                                        if arranged_imports {
+                                            if let Some(declared) = unchecked[k].clone() {
+                                                assert_export_shape(&arranged, declared);
+                                            }
+                                            return OuterImport::Export(arranged);
                                         }
-                                    };
-                                    match unchecked[k].clone() {
-                                        Some(declared) => assert_shape(cols, declared),
-                                        None => cols,
+                                        return OuterImport::Columns(checked(export_containers(arranged)));
                                     }
+                                    let arranged = traces
+                                        .get_mut(&key)
+                                        .expect("validated above")
+                                        .import(outer.clone());
+                                    OuterImport::Columns(checked(rows_to_corgi(arranged.as_collection(|k, v| (k.clone(), v.clone())), shape)))
                                 }
                                 st::Source::Parent(_) => unreachable!("root import from a parent scope"),
                             }
@@ -855,23 +886,31 @@ impl Server {
                     // Render the program body in its own iterative scope, then bring
                     // every export back out to the host time.
                     let exports = outer.iterative::<PointStamp<OuterTime>, _, _>(|inner| {
-                        let entered: Vec<_> = outer_cols.iter().map(|c| c.clone().enter(inner)).collect();
-                        crate::backend::corgi::render_tree(root, inner.clone(), 0, entered, Some(&scope_shapes))
+                        let imports = outer_imports
+                            .iter()
+                            .map(|imp| match imp {
+                                OuterImport::Columns(c) => CorgiImport::Columns(c.clone().enter(inner)),
+                                OuterImport::Export(a) => CorgiImport::Export(a.clone().enter(inner)),
+                                OuterImport::Rows(_) | OuterImport::Trace(_) => unreachable!("a columnar program imports columns"),
+                            })
+                            .collect();
+                        crate::backend::corgi::render_tree_corgi_arranged(root, inner.clone(), 0, imports, Some(&scope_shapes))
                             .into_iter()
                             .map(|c| c.leave(outer))
                             .collect::<Vec<_>>()
                     });
                     (Vec::new(), exports)
                 } else {
-                    // One outer (host-time) collection per root import.
-                    let outer_cols: Vec<VecCollection<OuterTime, (Value, Value), Diff>> = root
+                    // One outer (host-time) value per root import: rows, or with arranged imports
+                    // a published row trace itself, for a vec program to read in place.
+                    let outer_imports: Vec<OuterImport> = root
                         .imports
                         .iter()
                         .map(|imp| match &imp.from {
                             st::Source::Input(n) => {
                                 let (handle, col) = outer.new_collection::<(Value, Value), Diff>();
                                 inputs.push((*n, ServerInput::Rows(handle)));
-                                col
+                                OuterImport::Rows(col)
                             }
                             st::Source::Trace(t) => {
                                 // The first binding point: resolve a named trace by importing it.
@@ -879,13 +918,25 @@ impl Server {
                                 // A columnar export is read as rows by its importers.
                                 if let Some(ctrace) = ctraces.get_mut(&key) {
                                     let rows = crate::backend::corgi::export_rows(ctrace.import(outer.clone()));
-                                    return differential_dataflow::AsCollection::as_collection(rows);
+                                    return OuterImport::Rows(differential_dataflow::AsCollection::as_collection(rows));
                                 }
                                 let arranged = traces
                                     .get_mut(&key)
                                     .expect("validated above")
                                     .import(outer.clone());
-                                arranged.as_collection(|k, v| (k.clone(), v.clone()))
+                                if arranged_imports && backend == RenderBackend::Vec {
+                                    // An ascription is checked against the trace's batches by a side
+                                    // operator (no output, no copy); the program reads the trace itself.
+                                    if let Some((ks, vs)) = imp.shape.clone() {
+                                        let _ = arranged.clone().flat_map_ref(move |k, v| {
+                                            assert!(k.has_shape(&ks) && v.has_shape(&vs), "input does not match its shape ascription");
+                                            None::<()>
+                                        });
+                                    }
+                                    OuterImport::Trace(arranged)
+                                } else {
+                                    OuterImport::Rows(arranged.as_collection(|k, v| (k.clone(), v.clone())))
+                                }
                             }
                             st::Source::Parent(_) => unreachable!("root import from a parent scope"),
                         })
@@ -894,11 +945,27 @@ impl Server {
                     // Render the program body in its own iterative scope, then bring
                     // every export back out to the host time.
                     let exports = outer.iterative::<PointStamp<OuterTime>, _, _>(|inner| {
-                        let entered: Vec<_> =
-                            outer_cols.iter().map(|c| c.clone().enter(inner)).collect();
                         let exports = match backend {
-                            RenderBackend::Vec => render_tree(root, inner.clone(), 0, entered, Some(&scope_shapes)),
+                            RenderBackend::Vec => {
+                                use crate::backend::{Rendered, vec::Arr};
+                                let rendered = outer_imports
+                                    .iter()
+                                    .map(|imp| match imp {
+                                        OuterImport::Rows(c) => Rendered::Collection(c.clone().enter(inner)),
+                                        OuterImport::Trace(a) => Rendered::Arrangement(Arr::Imported(a.clone().enter(inner))),
+                                        OuterImport::Columns(_) | OuterImport::Export(_) => unreachable!("a row program imports rows"),
+                                    })
+                                    .collect();
+                                render_tree_arranged(root, inner.clone(), 0, rendered, Some(&scope_shapes))
+                            }
                             RenderBackend::Corgi => {
+                                let entered = outer_imports
+                                    .iter()
+                                    .map(|imp| match imp {
+                                        OuterImport::Rows(c) => c.clone().enter(inner),
+                                        _ => unreachable!("a row-exporting corgi program imports rows"),
+                                    })
+                                    .collect();
                                 render_tree_rows(root, inner.clone(), 0, entered, Some(&scope_shapes))
                             }
                         };
diff --git a/interactive/tests/server_arranged_imports.rs b/interactive/tests/server_arranged_imports.rs
new file mode 100644
index 000000000..e7161f12d
--- /dev/null
+++ b/interactive/tests/server_arranged_imports.rs
@@ -0,0 +1,127 @@
+//! Imports that arrive as the producer's arrangement (`Server::set_arranged_imports`) answer
+//! exactly as imports that arrive as rows, on both backends: a join and a reduce of an import
+//! at the root, a join inside an iterative scope, and the import read as rows, through
+//! additions and retractions of the shared trace, including for an importer installed late.
+
+use interactive::ir::Value;
+use interactive::server::{RenderBackend, Server};
+use interactive::{lower, parse};
+
+type Snapshot = Vec<(Value, Value, i64)>;
+
+const PRODUCER: &str = r#"export "edges" = input 0 | key($0[0] ; $0[1]);"#;
+
+/// A consumer of `edges`, its exports suffixed by `tag`.
+fn consumer(tag: &str) -> String {
+    format!(
+        r#"
+        let edges = import "edges";
+        let roots = input 0 | key($0[0] ;);
+        r: {{
+            let step = reach | join(edges, ($2 ;));
+            var reach = roots + step | distinct;
+        }}
+        export "hop{tag}" = roots | join(edges, ($2 ; $0));
+        export "degree{tag}" = edges | count;
+        export "reach{tag}" = r::reach;
+        export "flipped{tag}" = edges | map($1 ; $0);
+        "#
+    )
+}
+
+fn install(server: &mut Server, worker: &mut timely::worker::Worker, name: &str, src: &str) {
+    let mut program = lower::lower_tree(parse::pipe::parse(src));
+    program.optimize();
+    server.install(worker, name, &program).unwrap();
+}
+
+fn edge(server: &mut Server, src: i64, dst: i64, diff: i64) {
+    let row = Value::Tuple(vec![Value::Int(src), Value::Int(dst)]);
+    server.feed("prod", 0, row, Value::unit(), None, diff).unwrap();
+}
+
+fn root(server: &mut Server, name: &str, node: i64) {
+    server.feed(name, 0, Value::Tuple(vec![Value::Int(node)]), Value::unit(), None, 1).unwrap();
+}
+
+/// Snapshot each named export, each expected to be non-empty.
+fn take(server: &mut Server, worker: &mut timely::worker::Worker, names: &[String], snapshots: &mut Vec<Snapshot>) {
+    for name in names {
+        let rows = server.snapshot(worker, name).unwrap();
+        assert!(!rows.is_empty(), "export {name} is empty");
+        snapshots.push(rows);
+    }
+}
+
+/// Run the scenario, snapshotting every consumer export after each tick.
+fn scenario(backend: RenderBackend, arranged: bool) -> Vec<Snapshot> {
+    timely::execute_directly(move |worker| {
+        let mut snapshots = Vec::new();
+        let mut server = Server::with_backend(backend);
+        server.set_arranged_imports(arranged);
+        install(&mut server, worker, "prod", PRODUCER);
+        for (s, d) in [(1, 2), (2, 3), (3, 4), (5, 6), (2, 7)] { edge(&mut server, s, d, 1); }
+        server.tick(worker);
+
+        let mut exports: Vec<String> = Vec::new();
+
+        // An importer installed over an existing trace.
+        install(&mut server, worker, "early", &consumer("_early"));
+        exports.extend(["hop_early", "degree_early", "reach_early", "flipped_early"].map(String::from));
+        root(&mut server, "early", 1);
+        server.tick(worker);
+        take(&mut server, worker, &exports, &mut snapshots);
+
+        // The shared trace changes: an addition, and a retraction that cuts reachability.
+        edge(&mut server, 4, 5, 1);
+        edge(&mut server, 2, 3, -1);
+        server.tick(worker);
+        take(&mut server, worker, &exports, &mut snapshots);
+
+        // An importer installed after the trace has history.
+        install(&mut server, worker, "late", &consumer("_late"));
+        exports.extend(["hop_late", "degree_late", "reach_late", "flipped_late"].map(String::from));
+        root(&mut server, "late", 2);
+        edge(&mut server, 2, 3, 1);
+        server.tick(worker);
+        take(&mut server, worker, &exports, &mut snapshots);
+
+        for _ in 0..3 {
+            edge(&mut server, 6, 1, 1);
+            edge(&mut server, 6, 1, -1);
+            server.tick(worker);
+        }
+        take(&mut server, worker, &exports, &mut snapshots);
+        snapshots
+    })
+}
+
+#[test]
+fn arranged_imports_answer_as_rows_do_vec() {
+    assert_eq!(scenario(RenderBackend::Vec, true), scenario(RenderBackend::Vec, false));
+}
+
+#[test]
+fn arranged_imports_answer_as_rows_do_corgi() {
+    assert_eq!(scenario(RenderBackend::Corgi, true), scenario(RenderBackend::Corgi, false));
+}
+
+#[test]
+fn backends_agree_with_arranged_imports() {
+    assert_eq!(scenario(RenderBackend::Vec, true), scenario(RenderBackend::Corgi, true));
+}
+
+/// A declared import of an export whose shape was unknown at install is checked as its data
+/// arrives, also when the program reads the export's arrangement in place.
+#[test]
+#[should_panic(expected = "does not match its shape ascription")]
+fn an_arranged_import_is_checked_against_its_declaration() {
+    timely::execute_directly(|worker| {
+        let mut server = Server::with_backend(RenderBackend::Corgi);
+        install(&mut server, worker, "loose", r#"export "loose" = input 0;"#);
+        install(&mut server, worker, "declaring", r#"export "same" = import "loose" : ((int) ; ()) | arrange;"#);
+        let row = Value::Tuple(vec![Value::Int(1), Value::Int(2)]);
+        server.feed("loose", 0, row, Value::unit(), None, 1).unwrap();
+        server.tick(worker);
+    });
+}