Trait mz_storage::source::types::SourceRender

pub trait SourceRender {
    type Time: SourceTimestamp;

    const STATUS_NAMESPACE: StatusNamespace;

    // Required method
    fn render<G: Scope<Timestamp = Self::Time>>(
        self,
        scope: &mut G,
        config: RawSourceCreationConfig,
        resume_uppers: impl Stream<Item = Antichain<Self::Time>> + 'static,
        start_signal: impl Future<Output = ()> + 'static
    ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>);
}

Describes a source that can render itself in a timely scope.

Required Associated Types

type Time: SourceTimestamp

Required Associated Constants

const STATUS_NAMESPACE: StatusNamespace

Required Methods

fn render<G: Scope<Timestamp = Self::Time>>( self, scope: &mut G, config: RawSourceCreationConfig, resume_uppers: impl Stream<Item = Antichain<Self::Time>> + 'static, start_signal: impl Future<Output = ()> + 'static ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>)

Renders the source in the provided timely scope.

The resume_uppers stream can be used by the source to observe the overall progress of the ingestion. When a frontier appears in this stream the source implementation can be certain that future ingestion instances will request to read the external data only at times beyond that frontier. Therefore, the source implementation can react to this stream by e.g committing offsets upstream or advancing the LSN of a replication slot. It is safe to ignore this argument.

Rendering a source is expected to return four things.

First, a source must produce a collection that is produced by the rendered dataflow and must contain definite¹ data for all times beyond the resumption frontier.

Second, a source may produce an optional progress stream that will be used to drive reclocking. This is useful for sources that can query the highest offsets of the external source before reading the data for those offsets. In those cases it is preferable to produce this additional stream.

Third, a source must produce a stream of health status updates.

Finally, the source is expected to return an opaque token that when dropped will cause the source to immediately drop all capabilities and advance its frontier to the empty antichain.

¹ https://github.com/MaterializeInc/materialize/blob/main/doc/developer/design/20210831_correctness.md#describing-definite-data

---

1.  ↩

Object Safety

This trait is not object safe.

Implementations on Foreign Types

impl SourceRender for KafkaSourceConnection

type Time = Partitioned<RangeBound<i32>, MzOffset>

const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::Kafka

fn render<G: Scope<Timestamp = Partitioned<RangeBound<PartitionId>, MzOffset>>>( self, scope: &mut G, config: RawSourceCreationConfig, resume_uppers: impl Stream<Item = Antichain<Partitioned<RangeBound<PartitionId>, MzOffset>>> + 'static, start_signal: impl Future<Output = ()> + 'static ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>)

impl SourceRender for LoadGeneratorSourceConnection

type Time = MzOffset

const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::Generator

fn render<G: Scope<Timestamp = MzOffset>>( self, scope: &mut G, config: RawSourceCreationConfig, committed_uppers: impl Stream<Item = Antichain<MzOffset>> + 'static, start_signal: impl Future<Output = ()> + 'static ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>)

impl SourceRender for MySqlSourceConnection

fn render<G: Scope<Timestamp = GtidPartition>>( self, scope: &mut G, config: RawSourceCreationConfig, resume_uppers: impl Stream<Item = Antichain<GtidPartition>> + 'static, _start_signal: impl Future<Output = ()> + 'static ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>)

Render the ingestion dataflow. This function only connects things together and contains no actual processing logic.

type Time = Partitioned<Uuid, GtidState>

const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::MySql

impl SourceRender for PostgresSourceConnection

fn render<G: Scope<Timestamp = MzOffset>>( self, scope: &mut G, config: RawSourceCreationConfig, resume_uppers: impl Stream<Item = Antichain<MzOffset>> + 'static, _start_signal: impl Future<Output = ()> + 'static ) -> (Collection<G, (usize, Result<SourceMessage, SourceReaderError>), Diff>, Option<Stream<G, Infallible>>, Stream<G, HealthStatusMessage>, Stream<G, ProgressStatisticsUpdate>, Vec<PressOnDropButton>)

Render the ingestion dataflow. This function only connects things together and contains no actual processing logic.

type Time = MzOffset

const STATUS_NAMESPACE: StatusNamespace = StatusNamespace::Postgres

Implementors