Method export_subscribe

src/clusterd-test-driver/src/dataflow.rs:363–383

Export a subscribe sink `sink_id` streaming changes of the collection `from_id` back as `ComputeResponse::SubscribeResponse` batches. Unlike a materialized view, a subscribe writes no shard, so it needs no storage metadata. `value_desc` is the output schema (must match `from_id`'s type); `up_to` is the exclusive upper at which the subscribe completes. The empty `output` ordering leaves intra-time

(
        &mut self,
        sink_id: GlobalId,
        from_id: GlobalId,
        value_desc: RelationDesc,
        up_to: Antichain<Timestamp>,
    )

    /// empty `output` ordering leaves intra-timestamp order unconstrained — the
    /// driver consolidates and sorts the updates for a deterministic golden.
    pub fn export_subscribe(
        &mut self,
        sink_id: GlobalId,
        from_id: GlobalId,
        value_desc: RelationDesc,
        up_to: Antichain<Timestamp>,
    ) -> &mut Self {
        let desc = ComputeSinkDesc {
            from: from_id,
            from_desc: value_desc,
            connection: ComputeSinkConnection::Subscribe(SubscribeSinkConnection {
                output: vec![],
            }),
            with_snapshot: true,
            up_to,
            non_null_assertions: vec![],
            refresh_schedule: None,
        };
        self.mir.export_sink(sink_id, desc);
        self
    }

    /// Set the dataflow's `as_of` (the read frontier hydration starts from).
    pub fn as_of(&mut self, t: Timestamp) -> &mut Self {