mz_postgres_util/
desc.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//! Descriptions of PostgreSQL objects.
11
12use std::collections::{BTreeMap, BTreeSet};
13
14use anyhow::bail;
15use mz_proto::{IntoRustIfSome, RustType, TryFromProtoError};
16use proptest::prelude::any;
17use proptest_derive::Arbitrary;
18use serde::{Deserialize, Serialize};
19use tokio_postgres::types::Oid;
20use tracing::warn;
21
22include!(concat!(env!("OUT_DIR"), "/mz_postgres_util.desc.rs"));
23
24/// Describes a schema in a PostgreSQL database.
25///
26/// <https://www.postgresql.org/docs/current/catalog-pg-namespace.html>
27#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
28pub struct PostgresSchemaDesc {
29    /// The OID of the schema.
30    pub oid: Oid,
31    /// The name of the schema.
32    pub name: String,
33    /// Owner of the namespace
34    pub owner: Oid,
35}
36
37/// Describes a table in a PostgreSQL database.
38#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize, Arbitrary)]
39pub struct PostgresTableDesc {
40    /// The OID of the table.
41    pub oid: Oid,
42    /// The name of the schema that the table belongs to.
43    pub namespace: String,
44    /// The name of the table.
45    pub name: String,
46    /// The description of each column, in order of their position in the table.
47    #[proptest(strategy = "proptest::collection::vec(any::<PostgresColumnDesc>(), 1..4)")]
48    pub columns: Vec<PostgresColumnDesc>,
49    /// Applicable keys for this table (i.e. primary key and unique
50    /// constraints).
51    #[proptest(strategy = "proptest::collection::btree_set(any::<PostgresKeyDesc>(), 1..4)")]
52    pub keys: BTreeSet<PostgresKeyDesc>,
53}
54
55impl PostgresTableDesc {
56    /// Determines if two `PostgresTableDesc` are compatible with one another in
57    /// a way that Materialize can handle.
58    ///
59    /// Currently this means that the values are equal except for the following
60    /// exceptions:
61    /// - `self`'s columns are a compatible prefix of `other`'s columns.
62    ///   Compatibility is defined as returning `true` for
63    ///   `PostgresColumnDesc::is_compatible`.
64    /// - `self`'s keys are all present in `other`
65    pub fn determine_compatibility(
66        &self,
67        other: &PostgresTableDesc,
68        allow_type_to_change_by_col_num: &BTreeSet<u16>,
69    ) -> Result<(), anyhow::Error> {
70        if self == other {
71            return Ok(());
72        }
73
74        let PostgresTableDesc {
75            oid: other_oid,
76            namespace: other_namespace,
77            name: other_name,
78            columns: other_cols,
79            keys: other_keys,
80        } = other;
81
82        let other_cols_by_name = BTreeMap::from_iter(other_cols.iter().map(|c| (&c.name, c)));
83        let columns_compatible =
84            self.columns
85                .iter()
86                .all(|info| match other_cols_by_name.get(&info.name) {
87                    Some(other_info) => {
88                        let allow_type_change =
89                            allow_type_to_change_by_col_num.contains(&info.col_num);
90                        info.is_compatible(other_info, allow_type_change)
91                    }
92                    None => false,
93                });
94
95        if columns_compatible
96            && &self.name == other_name
97            && &self.oid == other_oid
98            && &self.namespace == other_namespace
99            // Our keys are all still present in exactly the same shape.
100            && self.keys.difference(other_keys).next().is_none()
101        {
102            Ok(())
103        } else {
104            warn!(
105                "Error validating table in publication. Expected: {:?} Actual: {:?}",
106                &self, other
107            );
108            bail!(
109                "source table {} with oid {} has been altered",
110                self.name,
111                self.oid
112            )
113        }
114    }
115}
116
117impl RustType<ProtoPostgresTableDesc> for PostgresTableDesc {
118    fn into_proto(&self) -> ProtoPostgresTableDesc {
119        ProtoPostgresTableDesc {
120            oid: self.oid,
121            namespace: self.namespace.clone(),
122            name: self.name.clone(),
123            columns: self.columns.iter().map(|c| c.into_proto()).collect(),
124            keys: self.keys.iter().map(PostgresKeyDesc::into_proto).collect(),
125        }
126    }
127
128    fn from_proto(proto: ProtoPostgresTableDesc) -> Result<Self, TryFromProtoError> {
129        Ok(PostgresTableDesc {
130            oid: proto.oid,
131            namespace: proto.namespace.clone(),
132            name: proto.name.clone(),
133            columns: proto
134                .columns
135                .into_iter()
136                .map(PostgresColumnDesc::from_proto)
137                .collect::<Result<_, _>>()?,
138            keys: proto
139                .keys
140                .into_iter()
141                .map(PostgresKeyDesc::from_proto)
142                .collect::<Result<_, _>>()?,
143        })
144    }
145}
146
147/// Describes a column in a [`PostgresTableDesc`].
148#[derive(
149    Debug,
150    Clone,
151    Eq,
152    PartialEq,
153    Ord,
154    PartialOrd,
155    Serialize,
156    Deserialize,
157    Arbitrary
158)]
159pub struct PostgresColumnDesc {
160    /// The name of the column.
161    pub name: String,
162    /// The column's monotonic position in its table, i.e. "this was the _i_th
163    /// column created" irrespective of the current number of columns.
164    pub col_num: u16,
165    /// The OID of the column's type.
166    pub type_oid: Oid,
167    /// The modifier for the column's type.
168    pub type_mod: i32,
169    /// True if the column lacks a `NOT NULL` constraint.
170    pub nullable: bool,
171}
172
173impl PostgresColumnDesc {
174    /// Determines if data a relation with a structure of `other` can be treated
175    /// the same as `self`.
176    ///
177    /// Note that this function somewhat unnecessarily errors if the names
178    /// differ; this is negotiable but we want users to understand the fixedness
179    /// of names in our schemas.
180    fn is_compatible(&self, other: &PostgresColumnDesc, allow_type_change: bool) -> bool {
181        self.name == other.name
182            && self.col_num == other.col_num
183            && (self.type_oid == other.type_oid || allow_type_change)
184            && (self.type_mod == other.type_mod || allow_type_change)
185            // Columns are compatible if:
186            // - self is nullable; introducing a not null constraint doesn't
187            //   change this column's behavior.
188            // - self and other are both not nullable
189            && (self.nullable || self.nullable == other.nullable)
190    }
191}
192
193impl RustType<ProtoPostgresColumnDesc> for PostgresColumnDesc {
194    fn into_proto(&self) -> ProtoPostgresColumnDesc {
195        ProtoPostgresColumnDesc {
196            name: self.name.clone(),
197            col_num: Some(self.col_num.into()),
198            type_oid: self.type_oid,
199            type_mod: self.type_mod,
200            nullable: self.nullable,
201        }
202    }
203
204    fn from_proto(proto: ProtoPostgresColumnDesc) -> Result<Self, TryFromProtoError> {
205        let col_num_u32: u32 = proto
206            .col_num
207            .into_rust_if_some("ProtoPostgresColumnDesc::col_num")?;
208        // `col_num` is `u16` on the Rust side. Reject u32 values that don't fit
209        // instead of panicking. This is reachable from untrusted proto bytes.
210        let col_num = u16::try_from(col_num_u32)
211            .map_err(|e| TryFromProtoError::InvalidFieldError(e.to_string()))?;
212        Ok(PostgresColumnDesc {
213            name: proto.name,
214            col_num,
215            type_oid: proto.type_oid,
216            type_mod: proto.type_mod,
217            nullable: proto.nullable,
218        })
219    }
220}
221
222/// Describes a key in a [`PostgresTableDesc`].
223#[derive(
224    Debug,
225    Clone,
226    Eq,
227    PartialEq,
228    Serialize,
229    Deserialize,
230    PartialOrd,
231    Ord,
232    Arbitrary
233)]
234pub struct PostgresKeyDesc {
235    /// This key is derived from the `pg_constraint` with this OID.
236    pub oid: Oid,
237    /// The name of the constraints.
238    pub name: String,
239    /// The `attnum` of the columns comprising the key. `attnum` is a unique identifier for a column
240    /// in a PG table; see <https://www.postgresql.org/docs/current/catalog-pg-attribute.html>
241    #[proptest(strategy = "proptest::collection::vec(any::<u16>(), 0..4)")]
242    pub cols: Vec<u16>,
243    /// Whether or not this key is the primary key.
244    pub is_primary: bool,
245    /// If this constraint was generated with NULLS NOT DISTINCT; see
246    /// <https://www.postgresql.org/about/featurematrix/detail/392/>
247    pub nulls_not_distinct: bool,
248}
249
250impl RustType<ProtoPostgresKeyDesc> for PostgresKeyDesc {
251    fn into_proto(&self) -> ProtoPostgresKeyDesc {
252        ProtoPostgresKeyDesc {
253            oid: self.oid,
254            name: self.name.clone(),
255            cols: self.cols.clone().into_iter().map(u32::from).collect(),
256            is_primary: self.is_primary,
257            nulls_not_distinct: self.nulls_not_distinct,
258        }
259    }
260
261    fn from_proto(proto: ProtoPostgresKeyDesc) -> Result<Self, TryFromProtoError> {
262        // `cols` is `Vec<u16>` on the Rust side but `Vec<u32>` on the wire;
263        // a u32 value above 65535 used to panic via `.expect`, which is
264        // reachable from untrusted proto bytes.
265        let cols = proto
266            .cols
267            .into_iter()
268            .map(|c| {
269                u16::try_from(c).map_err(|e| TryFromProtoError::InvalidFieldError(e.to_string()))
270            })
271            .collect::<Result<Vec<_>, _>>()?;
272        Ok(PostgresKeyDesc {
273            oid: proto.oid,
274            name: proto.name,
275            cols,
276            is_primary: proto.is_primary,
277            nulls_not_distinct: proto.nulls_not_distinct,
278        })
279    }
280}
mz_postgres_util/desc.rs

mz_postgres_util/
desc.rs
