mz_compute_types/
sinks.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Use of this software is governed by the Business Source License
4// included in the LICENSE file.
5//
6// As of the Change Date specified in that file, in accordance with
7// the Business Source License, use of this software will be governed
8// by the Apache License, Version 2.0.
9
10//! Types for describing dataflow sinks.
11
12use mz_expr::{ColumnOrder, MfpPlan};
13use mz_repr::refresh_schedule::RefreshSchedule;
14use mz_repr::{CatalogItemId, GlobalId, RelationDesc, Timestamp};
15use mz_storage_types::connections::aws::AwsConnection;
16use mz_storage_types::sinks::S3UploadInfo;
17use serde::{Deserialize, Serialize};
18use timely::progress::Antichain;
19
20use crate::plan::scalar::LirScalarExpr;
21
22/// A sink for updates to a relational collection.
23#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
24pub struct ComputeSinkDesc<S: 'static = ()> {
25    /// TODO(database-issues#7533): Add documentation.
26    pub from: GlobalId,
27    /// TODO(database-issues#7533): Add documentation.
28    pub from_desc: RelationDesc,
29    /// TODO(database-issues#7533): Add documentation.
30    pub connection: ComputeSinkConnection<S>,
31    /// TODO(database-issues#7533): Add documentation.
32    pub with_snapshot: bool,
33    /// TODO(database-issues#7533): Add documentation.
34    pub up_to: Antichain<Timestamp>,
35    /// TODO(database-issues#7533): Add documentation.
36    pub non_null_assertions: Vec<usize>,
37    /// TODO(database-issues#7533): Add documentation.
38    pub refresh_schedule: Option<RefreshSchedule>,
39    /// The arrangement of `from` the renderer consumes, and how to read rows from it.
40    ///
41    /// `None` means consume the unarranged collection. `Some((key, mfp))` names an arrangement
42    /// of `from` by its key, and `mfp` reconstructs full rows of `from_desc` from the
43    /// arrangement's key and value.
44    ///
45    /// Chosen during LIR lowering, where the available arrangements of `from` are known. It is
46    /// `None` before lowering runs.
47    pub from_arrangement: Option<(Vec<LirScalarExpr>, MfpPlan<LirScalarExpr>)>,
48}
49
50/// TODO(database-issues#7533): Add documentation.
51#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
52pub enum ComputeSinkConnection<S: 'static = ()> {
53    /// TODO(database-issues#7533): Add documentation.
54    Subscribe(SubscribeSinkConnection),
55    /// TODO(database-issues#7533): Add documentation.
56    MaterializedView(MaterializedViewSinkConnection<S>),
57    /// A compute sink to do a oneshot copy to s3.
58    CopyToS3Oneshot(CopyToS3OneshotSinkConnection),
59    /// A compute sink that writes rows into the in-process Prometheus metrics registry.
60    MetricSink(MetricSinkConnection),
61}
62
63impl<S> ComputeSinkConnection<S> {
64    /// Returns the name of the sink connection.
65    pub fn name(&self) -> &'static str {
66        match self {
67            ComputeSinkConnection::Subscribe(_) => "subscribe",
68            ComputeSinkConnection::MaterializedView(_) => "materialized_view",
69            ComputeSinkConnection::CopyToS3Oneshot(_) => "copy_to_s3_oneshot",
70            ComputeSinkConnection::MetricSink(_) => "metric_sink",
71        }
72    }
73
74    /// True if the sink is a subscribe, which is differently recoverable than other sinks.
75    pub fn is_subscribe(&self) -> bool {
76        if let ComputeSinkConnection::Subscribe(_) = self {
77            true
78        } else {
79            false
80        }
81    }
82}
83
84/// TODO(database-issues#7533): Add documentation.
85#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
86pub struct SubscribeSinkConnection {
87    /// An ordering for the data in the subscribe.
88    pub output: Vec<ColumnOrder>,
89}
90
91/// Connection for a sink that publishes rows into the in-process Prometheus metrics registry.
92///
93/// The sink does not write to persist, so there is no storage metadata to parameterize over.
94#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
95pub struct MetricSinkConnection {
96    /// Value of the `sink` label on the sink's health gauges. A user sink passes its `GlobalId`,
97    /// which is durable. A coordinator-installed curated sink passes its definition name, because
98    /// its `GlobalId` is transient and would churn the label on every boot.
99    pub label: String,
100}
101
102/// Connection attributes required to do a oneshot copy to s3.
103#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
104pub struct CopyToS3OneshotSinkConnection {
105    /// Information specific to the upload.
106    pub upload_info: S3UploadInfo,
107    /// The AWS connection information to do the writes.
108    pub aws_connection: AwsConnection,
109    /// The ID of the Connection object, used to generate the External ID when
110    /// using AssumeRole with AWS connection.
111    pub connection_id: CatalogItemId,
112    /// The number of batches the COPY TO output will be divided into
113    /// where each worker will process 0 or more batches of data.
114    pub output_batch_count: u64,
115}
116
117/// TODO(database-issues#7533): Add documentation.
118#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
119pub struct MaterializedViewSinkConnection<S> {
120    /// TODO(database-issues#7533): Add documentation.
121    pub value_desc: RelationDesc,
122    /// TODO(database-issues#7533): Add documentation.
123    pub storage_metadata: S,
124}
mz_compute_types/sinks.rs

mz_compute_types/
sinks.rs
