mz_adapter/catalog/command/source/
alter.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
10use std::collections::BTreeSet;
11
12use mz_catalog::memory::objects::{DataSourceDesc, Source};
13use mz_ore::collections::CollectionExt;
14use mz_repr::CatalogItemId;
15use mz_sql::catalog::CatalogItem;
16use mz_sql::names::Aug;
17use mz_sql::plan::{Params, Plan};
18use mz_sql_parser::ast::{
19    AlterSourceAddSubsourceOption, CreateSourceConnection, MySqlConfigOption,
20    MySqlConfigOptionName, PgConfigOption, PgConfigOptionName, SqlServerConfigOption,
21    SqlServerConfigOptionName, Statement, UnresolvedItemName, WithOptionValue,
22};
23use mz_storage_types::AlterCompatible;
24use mz_storage_types::connections::inline::IntoInlineConnection;
25use mz_storage_types::controller::StorageError;
26
27use crate::AdapterError;
28use crate::catalog::ConnCatalog;
29
30/// Rebuilds the current source definition with the supplied export options.
31pub(crate) fn rebuild_source(
32    catalog: &ConnCatalog<'_>,
33    item_id: CatalogItemId,
34    options: Vec<AlterSourceAddSubsourceOption<Aug>>,
35) -> Result<Source, AdapterError> {
36    let state = catalog.state();
37    let cur_entry = state.get_entry(&item_id);
38    let cur_source = cur_entry.source().expect("known to be source");
39    const ALTER_SOURCE: &str = "ALTER SOURCE...ADD SUBSOURCES";
40
41    let mz_sql::plan::AlterSourceAddSubsourceOptionExtracted {
42        text_columns: mut new_text_columns,
43        exclude_columns: mut new_exclude_columns,
44        ..
45    } = options.try_into()?;
46
47    let create_source_stmt = match mz_sql::parse::parse(cur_entry.create_sql())
48        .expect("invalid create sql persisted to catalog")
49        .into_element()
50        .ast
51    {
52        Statement::CreateSource(stmt) => stmt,
53        _ => unreachable!("proved type is source"),
54    };
55    let (mut create_source_stmt, resolved_ids) =
56        mz_sql::names::resolve(&state.for_system_session(), create_source_stmt)
57            .map_err(|e| AdapterError::internal(ALTER_SOURCE, e))?;
58
59    let curr_references: BTreeSet<_> = state
60        .get_entry(&item_id)
61        .used_by()
62        .into_iter()
63        .filter_map(|subsource| {
64            state
65                .get_entry(subsource)
66                .subsource_details()
67                .map(|(_id, reference, _details)| reference)
68        })
69        .collect();
70
71    let purification_err =
72        || AdapterError::internal(ALTER_SOURCE, "error in subsource purification");
73
74    match &mut create_source_stmt.connection {
75        CreateSourceConnection::Postgres {
76            options: curr_options,
77            ..
78        } => {
79            let mz_sql::plan::PgConfigOptionExtracted {
80                mut text_columns, ..
81            } = curr_options.clone().try_into()?;
82
83            curr_options.retain(|o| !matches!(o.name, PgConfigOptionName::TextColumns));
84
85            text_columns.retain(|column_qualified_reference| {
86                mz_ore::soft_assert_eq_or_log!(
87                    column_qualified_reference.0.len(),
88                    4,
89                    "all TEXT COLUMNS values must be column-qualified references"
90                );
91                let mut table = column_qualified_reference.clone();
92                table.0.truncate(3);
93                curr_references.contains(&table)
94            });
95
96            new_text_columns.extend(text_columns);
97
98            if !new_text_columns.is_empty() {
99                new_text_columns.sort();
100                let new_text_columns = new_text_columns
101                    .into_iter()
102                    .map(WithOptionValue::UnresolvedItemName)
103                    .collect();
104
105                curr_options.push(PgConfigOption {
106                    name: PgConfigOptionName::TextColumns,
107                    value: Some(WithOptionValue::Sequence(new_text_columns)),
108                });
109            }
110        }
111        CreateSourceConnection::MySql {
112            options: curr_options,
113            ..
114        } => {
115            let mz_sql::plan::MySqlConfigOptionExtracted {
116                mut text_columns,
117                mut exclude_columns,
118                ..
119            } = curr_options.clone().try_into()?;
120
121            curr_options.retain(|o| {
122                !matches!(
123                    o.name,
124                    MySqlConfigOptionName::TextColumns | MySqlConfigOptionName::ExcludeColumns
125                )
126            });
127
128            let column_referenced = |column_qualified_reference: &UnresolvedItemName| {
129                mz_ore::soft_assert_eq_or_log!(
130                    column_qualified_reference.0.len(),
131                    3,
132                    "all TEXT COLUMNS & EXCLUDE COLUMNS values must be column-qualified references"
133                );
134                let mut table = column_qualified_reference.clone();
135                table.0.truncate(2);
136                curr_references.contains(&table)
137            };
138            text_columns.retain(column_referenced);
139            exclude_columns.retain(column_referenced);
140
141            new_text_columns.extend(text_columns);
142            new_exclude_columns.extend(exclude_columns);
143
144            if !new_text_columns.is_empty() {
145                new_text_columns.sort();
146                let new_text_columns = new_text_columns
147                    .into_iter()
148                    .map(WithOptionValue::UnresolvedItemName)
149                    .collect();
150
151                curr_options.push(MySqlConfigOption {
152                    name: MySqlConfigOptionName::TextColumns,
153                    value: Some(WithOptionValue::Sequence(new_text_columns)),
154                });
155            }
156            if !new_exclude_columns.is_empty() {
157                new_exclude_columns.sort();
158                let new_exclude_columns = new_exclude_columns
159                    .into_iter()
160                    .map(WithOptionValue::UnresolvedItemName)
161                    .collect();
162
163                curr_options.push(MySqlConfigOption {
164                    name: MySqlConfigOptionName::ExcludeColumns,
165                    value: Some(WithOptionValue::Sequence(new_exclude_columns)),
166                });
167            }
168        }
169        CreateSourceConnection::SqlServer {
170            options: curr_options,
171            ..
172        } => {
173            let mz_sql::plan::SqlServerConfigOptionExtracted {
174                mut text_columns,
175                mut exclude_columns,
176                ..
177            } = curr_options.clone().try_into()?;
178
179            curr_options.retain(|o| {
180                !matches!(
181                    o.name,
182                    SqlServerConfigOptionName::TextColumns
183                        | SqlServerConfigOptionName::ExcludeColumns
184                )
185            });
186
187            // SQL Server text/exclude column refs are 3-part (schema.table.col),
188            // which truncate to 2-part (schema.table). But external references
189            // are 3-part (database.schema.table). Use suffix matching since
190            // a SQL Server source connects to a single database.
191            let column_referenced = |column_qualified_reference: &UnresolvedItemName| {
192                mz_ore::soft_assert_eq_or_log!(
193                    column_qualified_reference.0.len(),
194                    3,
195                    "all TEXT COLUMNS & EXCLUDE COLUMNS values must be column-qualified references"
196                );
197                let mut table = column_qualified_reference.clone();
198                table.0.truncate(2);
199                curr_references.iter().any(|r| r.0.ends_with(&table.0))
200            };
201            text_columns.retain(column_referenced);
202            exclude_columns.retain(column_referenced);
203
204            new_text_columns.extend(text_columns);
205            new_exclude_columns.extend(exclude_columns);
206
207            if !new_text_columns.is_empty() {
208                new_text_columns.sort();
209                let new_text_columns = new_text_columns
210                    .into_iter()
211                    .map(WithOptionValue::UnresolvedItemName)
212                    .collect();
213
214                curr_options.push(SqlServerConfigOption {
215                    name: SqlServerConfigOptionName::TextColumns,
216                    value: Some(WithOptionValue::Sequence(new_text_columns)),
217                });
218            }
219            if !new_exclude_columns.is_empty() {
220                new_exclude_columns.sort();
221                let new_exclude_columns = new_exclude_columns
222                    .into_iter()
223                    .map(WithOptionValue::UnresolvedItemName)
224                    .collect();
225
226                curr_options.push(SqlServerConfigOption {
227                    name: SqlServerConfigOptionName::ExcludeColumns,
228                    value: Some(WithOptionValue::Sequence(new_exclude_columns)),
229                });
230            }
231        }
232        _ => return Err(purification_err()),
233    };
234
235    let mut catalog = state.for_system_session();
236    catalog.mark_id_unresolvable_for_replanning(cur_entry.id());
237
238    let planned = mz_sql::plan::plan(
239        None,
240        &catalog,
241        Statement::CreateSource(create_source_stmt),
242        &Params::empty(),
243        &resolved_ids,
244    )
245    .map_err(|e| AdapterError::internal(ALTER_SOURCE, e))?;
246    let plan = match planned {
247        (Plan::CreateSource(plan), _sql_impl_ids) => plan,
248        (p, _) => {
249            unreachable!("create source plan is only valid response, got {:?}", p)
250        }
251    };
252
253    let source = Source::new(
254        plan,
255        cur_source.global_id,
256        resolved_ids,
257        cur_source.custom_logical_compaction_window,
258        cur_source.is_retained_metrics_object,
259    );
260
261    let old_desc = cur_entry
262        .source_desc()
263        .map_err(|error| AdapterError::internal(ALTER_SOURCE, error))?
264        .expect("ingestion source")
265        .clone()
266        .into_inline_connection(state);
267    let new_desc = match &source.data_source {
268        DataSourceDesc::Ingestion { desc, .. }
269        | DataSourceDesc::OldSyntaxIngestion { desc, .. } => {
270            desc.clone().into_inline_connection(state)
271        }
272        _ => unreachable!("replanned ingestion source"),
273    };
274    old_desc
275        .alter_compatible(cur_source.global_id(), &new_desc)
276        .map_err(StorageError::from)
277        .map_err(|error| AdapterError::internal(ALTER_SOURCE, error))?;
278    Ok(source)
279}
mz_adapter/catalog/command/source/alter.rs

mz_adapter/catalog/command/source/
alter.rs
