iceberg/spec/
view_metadata_builder.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::collections::{HashMap, HashSet};
19use std::sync::Arc;
20
21use chrono::Utc;
22use itertools::Itertools;
23use uuid::Uuid;
24
25use super::{
26    DEFAULT_SCHEMA_ID, INITIAL_VIEW_VERSION_ID, ONE_MINUTE_MS, Schema, SchemaId,
27    TableMetadataBuilder, VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED,
28    VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED_DEFAULT, VIEW_PROPERTY_VERSION_HISTORY_SIZE,
29    VIEW_PROPERTY_VERSION_HISTORY_SIZE_DEFAULT, ViewFormatVersion, ViewMetadata,
30    ViewRepresentation, ViewVersion, ViewVersionLog, ViewVersionRef,
31};
32use crate::ViewCreation;
33use crate::catalog::ViewUpdate;
34use crate::error::{Result, invalid_data};
35use crate::io::is_truthy;
36
37/// Manipulating view metadata.
38///
39/// For this builder the order of called functions matters.
40/// All operations applied to the `ViewMetadata` are tracked in `changes` as  a chronologically
41/// ordered vec of `ViewUpdate`.
42/// If an operation does not lead to a change of the `ViewMetadata`, the corresponding update
43/// is omitted from `changes`.
44#[derive(Debug, Clone)]
45pub struct ViewMetadataBuilder {
46    metadata: ViewMetadata,
47    changes: Vec<ViewUpdate>,
48    last_added_schema_id: Option<SchemaId>,
49    last_added_version_id: Option<SchemaId>,
50    history_entry: Option<ViewVersionLog>,
51    // Previous view version is only used during build to check
52    // weather dialects are dropped or not.
53    previous_view_version: Option<ViewVersionRef>,
54}
55
56#[derive(Debug, Clone, PartialEq)]
57/// Result of modifying or creating a `ViewMetadata`.
58pub struct ViewMetadataBuildResult {
59    /// The new `ViewMetadata`.
60    pub metadata: ViewMetadata,
61    /// The changes that were applied to the metadata.
62    pub changes: Vec<ViewUpdate>,
63}
64
65impl ViewMetadataBuilder {
66    const LAST_ADDED: i32 = TableMetadataBuilder::LAST_ADDED;
67
68    /// Creates a new view metadata builder.
69    pub fn new(
70        location: String,
71        schema: Schema,
72        view_version: ViewVersion,
73        format_version: ViewFormatVersion,
74        properties: HashMap<String, String>,
75    ) -> Result<Self> {
76        let builder = Self {
77            metadata: ViewMetadata {
78                format_version,
79                view_uuid: Uuid::now_v7(),
80                location: "".to_string(), // Overwritten immediately by set_location
81                current_version_id: -1,   // Overwritten immediately by set_current_version,
82                versions: HashMap::new(), // Overwritten immediately by set_current_version
83                version_log: Vec::new(),
84                schemas: HashMap::new(), // Overwritten immediately by set_current_version
85                properties: HashMap::new(), // Overwritten immediately by set_properties
86            },
87            changes: vec![],
88            last_added_schema_id: None, // Overwritten immediately by set_current_version
89            last_added_version_id: None, // Overwritten immediately by set_current_version
90            history_entry: None,
91            previous_view_version: None, // This is a new view
92        };
93
94        builder
95            .set_location(location)
96            .set_current_version(view_version, schema)?
97            .set_properties(properties)
98    }
99
100    /// Creates a new view metadata builder from the given metadata to modify it.
101    #[must_use]
102    pub fn new_from_metadata(previous: ViewMetadata) -> Self {
103        let previous_view_version = previous.current_version().clone();
104        Self {
105            metadata: previous,
106            changes: Vec::default(),
107            last_added_schema_id: None,
108            last_added_version_id: None,
109            history_entry: None,
110            previous_view_version: Some(previous_view_version),
111        }
112    }
113
114    /// Creates a new view metadata builder from the given view creation.
115    pub fn from_view_creation(view_creation: ViewCreation) -> Result<Self> {
116        let ViewCreation {
117            location,
118            schema,
119            properties,
120            name: _,
121            representations,
122            default_catalog,
123            default_namespace,
124            summary,
125        } = view_creation;
126        let version = ViewVersion::builder()
127            .with_default_catalog(default_catalog)
128            .with_default_namespace(default_namespace)
129            .with_representations(representations)
130            .with_schema_id(schema.schema_id())
131            .with_summary(summary)
132            .with_timestamp_ms(Utc::now().timestamp_millis())
133            .with_version_id(INITIAL_VIEW_VERSION_ID)
134            .build();
135
136        Self::new(location, schema, version, ViewFormatVersion::V1, properties)
137    }
138
139    /// Upgrade `FormatVersion`. Downgrades are not allowed.
140    ///
141    /// # Errors
142    /// - Cannot downgrade to older format versions.
143    pub fn upgrade_format_version(self, format_version: ViewFormatVersion) -> Result<Self> {
144        if format_version < self.metadata.format_version {
145            return Err(invalid_data!(
146                "Cannot downgrade ViewFormatVersion from {} to {}",
147                self.metadata.format_version,
148                format_version
149            ));
150        }
151
152        if format_version != self.metadata.format_version {
153            match format_version {
154                ViewFormatVersion::V1 => {
155                    // No changes needed for V1
156                }
157            }
158        }
159
160        Ok(self)
161    }
162
163    /// Set the location of the view, stripping any trailing slashes.
164    pub fn set_location(mut self, location: String) -> Self {
165        let location = location.trim_end_matches('/').to_string();
166        if self.metadata.location != location {
167            self.changes.push(ViewUpdate::SetLocation {
168                location: location.clone(),
169            });
170            self.metadata.location = location;
171        }
172
173        self
174    }
175
176    /// Set an existing view version as the current version.
177    ///
178    /// # Errors
179    /// - The specified `version_id` does not exist.
180    /// - The specified `version_id` is `-1` but no version has been added.
181    pub fn set_current_version_id(mut self, mut version_id: i32) -> Result<Self> {
182        if version_id == Self::LAST_ADDED {
183            let Some(last_added_id) = self.last_added_version_id else {
184                return Err(invalid_data!(
185                    "Cannot set current version id to last added version: no version has been added."
186                ));
187            };
188            version_id = last_added_id;
189        }
190
191        let version_id = version_id; // make immutable
192
193        if version_id == self.metadata.current_version_id {
194            return Ok(self);
195        }
196
197        let version = self.metadata.versions.get(&version_id).ok_or_else(|| {
198            invalid_data!("Cannot set current version to unknown version with id: {version_id}")
199        })?;
200
201        self.metadata.current_version_id = version_id;
202
203        if self.last_added_version_id == Some(version_id) {
204            self.changes.push(ViewUpdate::SetCurrentViewVersion {
205                view_version_id: Self::LAST_ADDED,
206            });
207        } else {
208            self.changes.push(ViewUpdate::SetCurrentViewVersion {
209                view_version_id: version_id,
210            });
211        }
212
213        // Use the timestamp of the snapshot if it was added in this set of changes,
214        // otherwise use a current timestamp for the log. The view version was added
215        // by a past transaction.
216        let version_added_in_this_changes = self
217            .changes
218            .iter()
219            .any(|update| matches!(update, ViewUpdate::AddViewVersion { view_version } if view_version.version_id() == version_id));
220
221        let mut log = version.log();
222        if !version_added_in_this_changes {
223            log.set_timestamp_ms(Utc::now().timestamp_millis());
224        }
225
226        self.history_entry = Some(log);
227
228        Ok(self)
229    }
230
231    /// Add a new view version and set it as current.
232    pub fn set_current_version(
233        mut self,
234        view_version: ViewVersion,
235        schema: Schema,
236    ) -> Result<Self> {
237        let schema_id = self.add_schema_internal(schema);
238        let view_version = view_version.with_schema_id(schema_id);
239        let view_version_id = self.add_version_internal(view_version)?;
240        self.set_current_version_id(view_version_id)
241    }
242
243    /// Add a new version to the view.
244    ///
245    /// # Errors
246    /// - The schema ID of the version is set to `-1`, but no schema has been added.
247    /// - The schema ID of the specified version is unknown.
248    /// - Multiple queries for the same dialect are added.
249    pub fn add_version(mut self, view_version: ViewVersion) -> Result<Self> {
250        self.add_version_internal(view_version)?;
251
252        Ok(self)
253    }
254
255    fn add_version_internal(&mut self, view_version: ViewVersion) -> Result<i32> {
256        let version_id = self.reuse_or_create_new_view_version_id(&view_version);
257        let view_version = view_version.with_version_id(version_id);
258
259        if self.metadata.versions.contains_key(&version_id) {
260            // ToDo Discuss: Similar to TableMetadata sort-order, Java does not add changes
261            // in this case. I prefer to add changes as the state of the builder is
262            // potentially mutated (`last_added_version_id`), thus we should record the change.
263            if self.last_added_version_id != Some(version_id) {
264                self.changes
265                    .push(ViewUpdate::AddViewVersion { view_version });
266                self.last_added_version_id = Some(version_id);
267            }
268            return Ok(version_id);
269        }
270
271        let view_version = if view_version.schema_id() == Self::LAST_ADDED {
272            let last_added_schema_id = self.last_added_schema_id.ok_or_else(|| {
273                invalid_data!("Cannot set last added schema: no schema has been added")
274            })?;
275            view_version.with_schema_id(last_added_schema_id)
276        } else {
277            view_version
278        };
279
280        if !self
281            .metadata
282            .schemas
283            .contains_key(&view_version.schema_id())
284        {
285            return Err(invalid_data!(
286                "Cannot add version with unknown schema: {}",
287                view_version.schema_id()
288            ));
289        }
290
291        require_unique_dialects(&view_version)?;
292
293        // The `TableMetadataBuilder` uses these checks in multiple places - also in Java.
294        // If we think delayed requests are a problem, I think we should also add it here.
295        if let Some(last) = self.metadata.version_log.last() {
296            // commits can happen concurrently from different machines.
297            // A tolerance helps us avoid failure for small clock skew
298            if view_version.timestamp_ms() - last.timestamp_ms() < -ONE_MINUTE_MS {
299                return Err(invalid_data!(
300                    "Invalid snapshot timestamp {}: before last snapshot timestamp {}",
301                    view_version.timestamp_ms(),
302                    last.timestamp_ms()
303                ));
304            }
305        }
306
307        self.metadata
308            .versions
309            .insert(version_id, Arc::new(view_version.clone()));
310
311        let view_version = if let Some(last_added_schema_id) = self.last_added_schema_id {
312            if view_version.schema_id() == last_added_schema_id {
313                view_version.with_schema_id(Self::LAST_ADDED)
314            } else {
315                view_version
316            }
317        } else {
318            view_version
319        };
320        self.changes
321            .push(ViewUpdate::AddViewVersion { view_version });
322
323        self.last_added_version_id = Some(version_id);
324
325        Ok(version_id)
326    }
327
328    fn reuse_or_create_new_view_version_id(&self, new_view_version: &ViewVersion) -> i32 {
329        self.metadata
330            .versions
331            .iter()
332            .find_map(|(id, other_version)| {
333                new_view_version
334                    .behaves_identical_to(other_version)
335                    .then_some(*id)
336            })
337            .unwrap_or_else(|| {
338                self.get_highest_view_version_id()
339                    .map(|id| id + 1)
340                    .unwrap_or(INITIAL_VIEW_VERSION_ID)
341            })
342    }
343
344    fn get_highest_view_version_id(&self) -> Option<i32> {
345        self.metadata.versions.keys().max().copied()
346    }
347
348    /// Add a new schema to the view.
349    pub fn add_schema(mut self, schema: Schema) -> Self {
350        self.add_schema_internal(schema);
351
352        self
353    }
354
355    fn add_schema_internal(&mut self, schema: Schema) -> SchemaId {
356        let schema_id = self.reuse_or_create_new_schema_id(&schema);
357
358        if self.metadata.schemas.contains_key(&schema_id) {
359            // ToDo Discuss: Java does not add changes in this case. I prefer to add changes
360            // as the state of the builder is potentially mutated (`last_added_schema_id`),
361            // thus we should record the change.
362            if self.last_added_schema_id != Some(schema_id) {
363                self.changes.push(ViewUpdate::AddSchema {
364                    schema: schema.clone().with_schema_id(schema_id),
365                    last_column_id: None,
366                });
367                self.last_added_schema_id = Some(schema_id);
368            }
369            return schema_id;
370        }
371
372        let schema = schema.with_schema_id(schema_id);
373
374        self.metadata
375            .schemas
376            .insert(schema_id, Arc::new(schema.clone()));
377        let last_column_id = schema.highest_field_id();
378        self.changes.push(ViewUpdate::AddSchema {
379            schema,
380            last_column_id: Some(last_column_id),
381        });
382
383        self.last_added_schema_id = Some(schema_id);
384
385        schema_id
386    }
387
388    fn reuse_or_create_new_schema_id(&self, new_schema: &Schema) -> SchemaId {
389        self.metadata
390            .schemas
391            .iter()
392            .find_map(|(id, schema)| new_schema.is_same_schema(schema).then_some(*id))
393            .unwrap_or_else(|| {
394                self.get_highest_schema_id()
395                    .map(|id| id + 1)
396                    .unwrap_or(DEFAULT_SCHEMA_ID)
397            })
398    }
399
400    fn get_highest_schema_id(&self) -> Option<SchemaId> {
401        self.metadata.schemas.keys().max().copied()
402    }
403
404    /// Update properties of the view.
405    pub fn set_properties(mut self, updates: HashMap<String, String>) -> Result<Self> {
406        if updates.is_empty() {
407            return Ok(self);
408        }
409
410        let num_versions_to_keep = updates
411            .get(VIEW_PROPERTY_VERSION_HISTORY_SIZE)
412            .and_then(|v| v.parse::<i64>().ok())
413            .unwrap_or(1);
414        if num_versions_to_keep < 0 {
415            return Err(invalid_data!(
416                "{VIEW_PROPERTY_VERSION_HISTORY_SIZE} must be positive but was {num_versions_to_keep}"
417            ));
418        }
419
420        self.metadata.properties.extend(updates.clone());
421        self.changes.push(ViewUpdate::SetProperties { updates });
422
423        Ok(self)
424    }
425
426    /// Remove properties from the view
427    pub fn remove_properties(mut self, removals: &[String]) -> Self {
428        if removals.is_empty() {
429            return self;
430        }
431
432        for property in removals {
433            self.metadata.properties.remove(property);
434        }
435
436        self.changes.push(ViewUpdate::RemoveProperties {
437            removals: removals.to_vec(),
438        });
439
440        self
441    }
442
443    /// Assign a new UUID to the view.
444    pub fn assign_uuid(mut self, uuid: Uuid) -> Self {
445        if self.metadata.view_uuid != uuid {
446            self.metadata.view_uuid = uuid;
447            self.changes.push(ViewUpdate::AssignUuid { uuid });
448        }
449
450        self
451    }
452
453    /// Build the `ViewMetadata` from the changes.
454    pub fn build(mut self) -> Result<ViewMetadataBuildResult> {
455        if let Some(history_entry) = self.history_entry.take() {
456            self.metadata.version_log.push(history_entry);
457        }
458
459        // We should run validate before `self.metadata.current_version()` below,
460        // as it might panic if the metadata is invalid.
461        self.metadata.validate()?;
462
463        if let Some(previous) = self.previous_view_version.take()
464            && !allow_replace_drop_dialects(&self.metadata.properties)
465        {
466            require_no_dialect_dropped(&previous, self.metadata.current_version())?;
467        }
468
469        let _expired_versions = self.expire_versions();
470        self.metadata.version_log = update_version_log(
471            self.metadata.version_log,
472            self.metadata.versions.keys().copied().collect(),
473        );
474
475        Ok(ViewMetadataBuildResult {
476            metadata: self.metadata,
477            changes: self.changes,
478        })
479    }
480
481    /// Removes expired versions from the view and returns them.
482    fn expire_versions(&mut self) -> Vec<ViewVersionRef> {
483        let num_versions_to_keep = self
484            .metadata
485            .properties
486            .get(VIEW_PROPERTY_VERSION_HISTORY_SIZE)
487            .and_then(|v| v.parse::<usize>().ok())
488            .unwrap_or(VIEW_PROPERTY_VERSION_HISTORY_SIZE_DEFAULT)
489            .max(1);
490
491        // expire old versions, but keep at least the versions added in this builder
492        let num_added_versions = self
493            .changes
494            .iter()
495            .filter(|update| matches!(update, ViewUpdate::AddViewVersion { .. }))
496            .count();
497        let num_versions_to_keep = num_added_versions.max(num_versions_to_keep);
498
499        if self.metadata.versions.len() > num_versions_to_keep {
500            // version ids are assigned sequentially. keep the latest versions by ID.
501            let mut versions_to_keep = self
502                .metadata
503                .versions
504                .keys()
505                .copied()
506                .sorted()
507                .rev()
508                .take(num_versions_to_keep)
509                .collect::<HashSet<_>>();
510
511            // always retain current version
512            if !versions_to_keep.contains(&self.metadata.current_version_id) {
513                // Remove the lowest ID
514                if num_versions_to_keep > num_added_versions {
515                    let lowest_id = versions_to_keep.iter().min().copied();
516                    lowest_id.map(|id| versions_to_keep.remove(&id));
517                }
518                // Add the current version ID
519                versions_to_keep.insert(self.metadata.current_version_id);
520            }
521
522            let mut expired_versions = Vec::new();
523            // remove all versions which are not in versions_to_keep from the metadata
524            // and add them to the expired_versions list
525            self.metadata.versions.retain(|id, version| {
526                if versions_to_keep.contains(id) {
527                    true
528                } else {
529                    expired_versions.push(version.clone());
530                    false
531                }
532            });
533
534            expired_versions
535        } else {
536            Vec::new()
537        }
538    }
539}
540
541/// Expire version log entries that are no longer relevant.
542/// Returns the history entries to retain.
543fn update_version_log(
544    version_log: Vec<ViewVersionLog>,
545    ids_to_keep: HashSet<i32>,
546) -> Vec<ViewVersionLog> {
547    let mut retained_history = Vec::new();
548    for log_entry in version_log {
549        if ids_to_keep.contains(&log_entry.version_id()) {
550            retained_history.push(log_entry);
551        } else {
552            retained_history.clear();
553        }
554    }
555    retained_history
556}
557
558fn allow_replace_drop_dialects(properties: &HashMap<String, String>) -> bool {
559    properties
560        .get(VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED)
561        .map_or(
562            VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED_DEFAULT,
563            |value| is_truthy(value),
564        )
565}
566
567fn require_no_dialect_dropped(previous: &ViewVersion, current: &ViewVersion) -> Result<()> {
568    let base_dialects = lowercase_sql_dialects_for(previous);
569    let updated_dialects = lowercase_sql_dialects_for(current);
570
571    if !updated_dialects.is_superset(&base_dialects) {
572        return Err(invalid_data!(
573            "Cannot replace view due to loss of view dialects: \nPrevious dialects: {:?}\nNew dialects: {:?}\nSet {} to true to allow dropping dialects.",
574            Vec::from_iter(base_dialects),
575            Vec::from_iter(updated_dialects),
576            VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED
577        ));
578    }
579
580    Ok(())
581}
582
583fn lowercase_sql_dialects_for(view_version: &ViewVersion) -> HashSet<String> {
584    view_version
585        .representations()
586        .iter()
587        .map(|repr| match repr {
588            ViewRepresentation::Sql(sql_repr) => sql_repr.dialect.to_lowercase(),
589        })
590        .collect()
591}
592
593pub(super) fn require_unique_dialects(view_version: &ViewVersion) -> Result<()> {
594    let mut seen_dialects = HashSet::with_capacity(view_version.representations().len());
595    for repr in view_version.representations().iter() {
596        match repr {
597            ViewRepresentation::Sql(sql_repr) => {
598                if !seen_dialects.insert(sql_repr.dialect.to_lowercase()) {
599                    return Err(invalid_data!(
600                        "Invalid view version: Cannot add multiple queries for dialect {}",
601                        sql_repr.dialect
602                    ));
603                }
604            }
605        }
606    }
607    Ok(())
608}
609
610#[cfg(test)]
611mod test {
612    use super::super::view_metadata::tests::get_test_view_metadata;
613    use super::*;
614    use crate::NamespaceIdent;
615    use crate::spec::{
616        NestedField, PrimitiveType, SqlViewRepresentation, Type, ViewRepresentations,
617    };
618
619    fn new_view_version(id: usize, schema_id: SchemaId, sql: &str) -> ViewVersion {
620        new_view_version_with_dialect(id, schema_id, sql, vec!["spark"])
621    }
622
623    fn new_view_version_with_dialect(
624        id: usize,
625        schema_id: SchemaId,
626        sql: &str,
627        dialects: Vec<&str>,
628    ) -> ViewVersion {
629        ViewVersion::builder()
630            .with_version_id(id as i32)
631            .with_schema_id(schema_id)
632            .with_timestamp_ms(1573518431300)
633            .with_default_catalog(Some("prod".to_string()))
634            .with_summary(HashMap::from_iter(vec![(
635                "user".to_string(),
636                "some-user".to_string(),
637            )]))
638            .with_representations(ViewRepresentations(
639                dialects
640                    .iter()
641                    .map(|dialect| {
642                        ViewRepresentation::Sql(SqlViewRepresentation {
643                            dialect: dialect.to_string(),
644                            sql: sql.to_string(),
645                        })
646                    })
647                    .collect(),
648            ))
649            .with_default_namespace(NamespaceIdent::new("default".to_string()))
650            .build()
651    }
652
653    fn builder_without_changes() -> ViewMetadataBuilder {
654        ViewMetadataBuilder::new_from_metadata(get_test_view_metadata("ViewMetadataV1Valid.json"))
655    }
656
657    #[test]
658    fn test_minimal_builder() {
659        let location = "s3://bucket/table".to_string();
660        let schema = Schema::builder()
661            .with_schema_id(1)
662            .with_fields(vec![])
663            .build()
664            .unwrap();
665        // Version ID and schema should be re-assigned
666        let version = new_view_version(20, 21, "select 1 as count");
667        let format_version = ViewFormatVersion::V1;
668        let properties = HashMap::from_iter(vec![("key".to_string(), "value".to_string())]);
669
670        let build_result = ViewMetadataBuilder::new(
671            location.clone(),
672            schema.clone(),
673            version.clone(),
674            format_version,
675            properties.clone(),
676        )
677        .unwrap()
678        .build()
679        .unwrap();
680
681        let metadata = build_result.metadata;
682        assert_eq!(metadata.location, location);
683        assert_eq!(metadata.current_version_id, INITIAL_VIEW_VERSION_ID);
684        assert_eq!(metadata.format_version, format_version);
685        assert_eq!(metadata.properties, properties);
686        assert_eq!(metadata.versions.len(), 1);
687        assert_eq!(metadata.schemas.len(), 1);
688        assert_eq!(metadata.version_log.len(), 1);
689        assert_eq!(
690            Arc::unwrap_or_clone(metadata.versions[&INITIAL_VIEW_VERSION_ID].clone()),
691            version
692                .clone()
693                .with_version_id(INITIAL_VIEW_VERSION_ID)
694                .with_schema_id(0)
695        );
696
697        let changes = build_result.changes;
698        assert_eq!(changes.len(), 5);
699        assert!(changes.contains(&ViewUpdate::SetLocation { location }));
700        assert!(
701            changes.contains(&ViewUpdate::AddViewVersion {
702                view_version: version
703                    .with_version_id(INITIAL_VIEW_VERSION_ID)
704                    .with_schema_id(-1)
705            })
706        );
707        assert!(changes.contains(&ViewUpdate::SetCurrentViewVersion {
708            view_version_id: -1
709        }));
710        assert!(changes.contains(&ViewUpdate::AddSchema {
711            schema: schema.clone().with_schema_id(0),
712            last_column_id: Some(0)
713        }));
714        assert!(changes.contains(&ViewUpdate::SetProperties {
715            updates: properties
716        }));
717    }
718
719    #[test]
720    fn test_version_expiration() {
721        let v1 = new_view_version(0, 1, "select 1 as count");
722        let v2 = new_view_version(0, 1, "select count(1) as count from t2");
723        let v3 = new_view_version(0, 1, "select count from t1");
724
725        let builder = builder_without_changes()
726            .add_version(v1)
727            .unwrap()
728            .add_version(v2)
729            .unwrap()
730            .add_version(v3)
731            .unwrap();
732        let builder_without_changes = builder.clone().build().unwrap().metadata.into_builder();
733
734        // No limit on versions
735        let metadata = builder.clone().build().unwrap().metadata;
736        assert_eq!(
737            metadata.versions.keys().cloned().collect::<HashSet<_>>(),
738            HashSet::from_iter(vec![1, 2, 3, 4])
739        );
740
741        // Limit to 2 versions, we still want to keep 3 versions as 3 where added during this build
742        // Plus the current version
743        let metadata = builder
744            .clone()
745            .set_properties(HashMap::from_iter(vec![(
746                VIEW_PROPERTY_VERSION_HISTORY_SIZE.to_string(),
747                "2".to_string(),
748            )]))
749            .unwrap()
750            .build()
751            .unwrap()
752            .metadata;
753        assert_eq!(
754            metadata.versions.keys().cloned().collect::<HashSet<_>>(),
755            HashSet::from_iter(vec![1, 2, 3, 4])
756        );
757        assert_eq!(metadata.version_log.len(), 1);
758
759        // Limit to 2 versions in new build, only keep 2.
760        // One of them should be the current
761        let metadata = builder_without_changes
762            .clone()
763            .set_properties(HashMap::from_iter(vec![(
764                VIEW_PROPERTY_VERSION_HISTORY_SIZE.to_string(),
765                "2".to_string(),
766            )]))
767            .unwrap()
768            .build()
769            .unwrap()
770            .metadata;
771        assert_eq!(
772            metadata.versions.keys().cloned().collect::<HashSet<_>>(),
773            HashSet::from_iter(vec![1, 4])
774        );
775
776        // Keep at least 1 version irrespective of the limit.
777        // This is the current version
778        let metadata = builder_without_changes
779            .set_properties(HashMap::from_iter(vec![(
780                VIEW_PROPERTY_VERSION_HISTORY_SIZE.to_string(),
781                "0".to_string(),
782            )]))
783            .unwrap()
784            .build()
785            .unwrap()
786            .metadata;
787        assert_eq!(
788            metadata.versions.keys().cloned().collect::<HashSet<_>>(),
789            HashSet::from_iter(vec![1])
790        );
791    }
792
793    #[test]
794    fn test_update_version_log() {
795        let v1 = new_view_version(1, 1, "select 1 as count");
796        let v2 = new_view_version(2, 1, "select count(1) as count from t2");
797        let v3 = new_view_version(3, 1, "select count from t1");
798
799        let one = ViewVersionLog::new(1, v1.timestamp_ms());
800        let two = ViewVersionLog::new(2, v2.timestamp_ms());
801        let three = ViewVersionLog::new(3, v3.timestamp_ms());
802
803        assert_eq!(
804            update_version_log(
805                vec![one.clone(), two.clone(), three.clone()],
806                HashSet::from_iter(vec![1, 2, 3])
807            ),
808            vec![one.clone(), two.clone(), three.clone()]
809        );
810
811        // one was an invalid entry in the history, so all previous elements are removed
812        assert_eq!(
813            update_version_log(
814                vec![
815                    three.clone(),
816                    two.clone(),
817                    one.clone(),
818                    two.clone(),
819                    three.clone()
820                ],
821                HashSet::from_iter(vec![2, 3])
822            ),
823            vec![two.clone(), three.clone()]
824        );
825
826        // two was an invalid entry in the history, so all previous elements are removed
827        assert_eq!(
828            update_version_log(
829                vec![
830                    one.clone(),
831                    two.clone(),
832                    three.clone(),
833                    one.clone(),
834                    three.clone()
835                ],
836                HashSet::from_iter(vec![1, 3])
837            ),
838            vec![three.clone(), one.clone(), three.clone()]
839        );
840    }
841
842    #[test]
843    fn test_use_previously_added_version() {
844        let v2 = new_view_version(2, 1, "select 1 as count");
845        let v3 = new_view_version(3, 1, "select count(1) as count from t2");
846        let schema = Schema::builder().build().unwrap();
847
848        let log_v2 = ViewVersionLog::new(2, v2.timestamp_ms());
849        let log_v3 = ViewVersionLog::new(3, v3.timestamp_ms());
850
851        let metadata_v2 = builder_without_changes()
852            .set_current_version(v2.clone(), schema.clone())
853            .unwrap()
854            .build()
855            .unwrap()
856            .metadata;
857
858        // Log should use the exact timestamp of v1
859        assert_eq!(metadata_v2.version_log.last().unwrap(), &log_v2);
860
861        // Add second version, should use exact timestamp of v2
862        let metadata_v3 = metadata_v2
863            .into_builder()
864            .set_current_version(v3.clone(), schema)
865            .unwrap()
866            .build()
867            .unwrap()
868            .metadata;
869
870        assert_eq!(metadata_v3.version_log[1..], vec![
871            log_v2.clone(),
872            log_v3.clone()
873        ]);
874
875        // Re-use Version 1, add a new log entry with a new timestamp
876        let metadata_v4 = metadata_v3
877            .into_builder()
878            .set_current_version_id(2)
879            .unwrap()
880            .build()
881            .unwrap()
882            .metadata;
883
884        // Last entry should be equal to v2 but with an updated timestamp
885        let entry = metadata_v4.version_log.last().unwrap();
886        assert_eq!(entry.version_id(), 2);
887        assert!(entry.timestamp_ms() > v2.timestamp_ms());
888    }
889
890    #[test]
891    fn test_assign_uuid() {
892        let builder = builder_without_changes();
893        let uuid = Uuid::now_v7();
894        let build_result = builder.clone().assign_uuid(uuid).build().unwrap();
895        assert_eq!(build_result.metadata.view_uuid, uuid);
896        assert_eq!(build_result.changes, vec![ViewUpdate::AssignUuid { uuid }]);
897    }
898
899    #[test]
900    fn test_set_location() {
901        let builder = builder_without_changes();
902        let location = "s3://bucket/table".to_string();
903        let build_result = builder
904            .clone()
905            .set_location(location.clone())
906            .build()
907            .unwrap();
908        assert_eq!(build_result.metadata.location, location);
909        assert_eq!(build_result.changes, vec![ViewUpdate::SetLocation {
910            location
911        }]);
912    }
913
914    #[test]
915    fn test_set_and_remove_properties() {
916        let builder = builder_without_changes();
917        let properties = HashMap::from_iter(vec![
918            ("key1".to_string(), "value1".to_string()),
919            ("key2".to_string(), "value2".to_string()),
920        ]);
921        let build_result = builder
922            .clone()
923            .set_properties(properties.clone())
924            .unwrap()
925            .remove_properties(&["key2".to_string(), "key3".to_string()])
926            .build()
927            .unwrap();
928        assert_eq!(
929            build_result.metadata.properties.get("key1"),
930            Some(&"value1".to_string())
931        );
932        assert_eq!(build_result.metadata.properties.get("key2"), None);
933        assert_eq!(build_result.changes, vec![
934            ViewUpdate::SetProperties {
935                updates: properties
936            },
937            ViewUpdate::RemoveProperties {
938                removals: vec!["key2".to_string(), "key3".to_string()]
939            }
940        ]);
941    }
942
943    #[test]
944    fn test_add_schema() {
945        let builder = builder_without_changes();
946        let schema = Schema::builder()
947            .with_schema_id(1)
948            .with_fields(vec![])
949            .build()
950            .unwrap();
951        let build_result = builder.clone().add_schema(schema.clone()).build().unwrap();
952        assert_eq!(build_result.metadata.schemas.len(), 2);
953        assert_eq!(build_result.changes, vec![ViewUpdate::AddSchema {
954            schema: schema.clone().with_schema_id(2),
955            last_column_id: Some(0)
956        }]);
957
958        // Add schema again - id is reused
959        let build_result = builder.clone().add_schema(schema.clone()).build().unwrap();
960        assert_eq!(build_result.metadata.schemas.len(), 2);
961        assert_eq!(build_result.changes, vec![ViewUpdate::AddSchema {
962            schema: schema.clone().with_schema_id(2),
963            last_column_id: Some(0)
964        }]);
965    }
966
967    #[test]
968    fn test_add_and_set_current_version() {
969        let builder = builder_without_changes();
970        let v1 = new_view_version(2, 1, "select 1 as count");
971        let v2 = new_view_version(3, 2, "select count(1) as count from t2");
972        let v2_schema = Schema::builder()
973            .with_schema_id(2)
974            .with_fields(vec![])
975            .build()
976            .unwrap();
977
978        let build_result = builder
979            .clone()
980            .add_version(v1.clone())
981            .unwrap()
982            .add_schema(v2_schema.clone())
983            .add_version(v2.clone())
984            .unwrap()
985            .set_current_version_id(3)
986            .unwrap()
987            .build()
988            .unwrap();
989
990        assert_eq!(build_result.metadata.current_version_id, 3);
991        assert_eq!(build_result.metadata.versions.len(), 3);
992        assert_eq!(build_result.metadata.schemas.len(), 2);
993        assert_eq!(build_result.metadata.version_log.len(), 2);
994        assert_eq!(
995            Arc::unwrap_or_clone(build_result.metadata.versions[&2].clone()),
996            v1.clone().with_version_id(2).with_schema_id(1)
997        );
998        assert_eq!(
999            Arc::unwrap_or_clone(build_result.metadata.versions[&3].clone()),
1000            v2.clone().with_version_id(3).with_schema_id(2)
1001        );
1002        assert_eq!(build_result.changes.len(), 4);
1003        assert_eq!(build_result.changes, vec![
1004            ViewUpdate::AddViewVersion {
1005                view_version: v1.clone().with_version_id(2).with_schema_id(1)
1006            },
1007            ViewUpdate::AddSchema {
1008                schema: v2_schema.clone().with_schema_id(2),
1009                last_column_id: Some(0)
1010            },
1011            ViewUpdate::AddViewVersion {
1012                view_version: v2.clone().with_version_id(3).with_schema_id(-1)
1013            },
1014            ViewUpdate::SetCurrentViewVersion {
1015                view_version_id: -1
1016            }
1017        ]);
1018        assert_eq!(
1019            build_result
1020                .metadata
1021                .version_log
1022                .iter()
1023                .map(|v| v.version_id())
1024                .collect::<Vec<_>>(),
1025            vec![1, 3]
1026        );
1027    }
1028
1029    #[test]
1030    fn test_schema_and_version_id_reassignment() {
1031        let builder = builder_without_changes();
1032        let v1 = new_view_version(0, 1, "select 1 as count");
1033        let v2 = new_view_version(0, 2, "select count(1) as count from t2");
1034        let v2_schema = Schema::builder()
1035            .with_schema_id(0)
1036            .with_fields(vec![])
1037            .build()
1038            .unwrap();
1039
1040        let build_result = builder
1041            .clone()
1042            .add_version(v1.clone())
1043            .unwrap()
1044            .set_current_version(v2.clone(), v2_schema.clone())
1045            .unwrap()
1046            .build()
1047            .unwrap();
1048
1049        assert_eq!(build_result.metadata.current_version_id, 3);
1050        assert_eq!(build_result.metadata.versions.len(), 3);
1051        assert_eq!(build_result.metadata.schemas.len(), 2);
1052        assert_eq!(build_result.metadata.version_log.len(), 2);
1053        assert_eq!(
1054            Arc::unwrap_or_clone(build_result.metadata.versions[&2].clone()),
1055            v1.clone().with_version_id(2).with_schema_id(1)
1056        );
1057        assert_eq!(
1058            Arc::unwrap_or_clone(build_result.metadata.versions[&3].clone()),
1059            v2.clone().with_version_id(3).with_schema_id(2)
1060        );
1061        assert_eq!(build_result.changes.len(), 4);
1062        assert_eq!(build_result.changes, vec![
1063            ViewUpdate::AddViewVersion {
1064                view_version: v1.clone().with_version_id(2).with_schema_id(1)
1065            },
1066            ViewUpdate::AddSchema {
1067                schema: v2_schema.clone().with_schema_id(2),
1068                last_column_id: Some(0)
1069            },
1070            ViewUpdate::AddViewVersion {
1071                view_version: v2.clone().with_version_id(3).with_schema_id(-1)
1072            },
1073            ViewUpdate::SetCurrentViewVersion {
1074                view_version_id: -1
1075            }
1076        ]);
1077        assert_eq!(
1078            build_result
1079                .metadata
1080                .version_log
1081                .iter()
1082                .map(|v| v.version_id())
1083                .collect::<Vec<_>>(),
1084            vec![1, 3]
1085        );
1086    }
1087
1088    #[test]
1089    fn test_view_version_deduplication() {
1090        let builder = builder_without_changes();
1091        let v1 = new_view_version(0, 1, "select * from ns.tbl");
1092
1093        assert_eq!(builder.metadata.versions.len(), 1);
1094        let build_result = builder
1095            .clone()
1096            .add_version(v1.clone())
1097            .unwrap()
1098            .add_version(v1)
1099            .unwrap()
1100            .build()
1101            .unwrap();
1102
1103        assert_eq!(build_result.metadata.versions.len(), 2);
1104        assert_eq!(build_result.metadata.schemas.len(), 1);
1105    }
1106
1107    #[test]
1108    fn test_view_version_and_schema_deduplication() {
1109        let schema_one = Schema::builder()
1110            .with_schema_id(5)
1111            .with_fields(vec![
1112                NestedField::required(1, "x", Type::Primitive(PrimitiveType::Long)).into(),
1113            ])
1114            .build()
1115            .unwrap();
1116        let schema_two = Schema::builder()
1117            .with_schema_id(7)
1118            .with_fields(vec![
1119                NestedField::required(1, "y", Type::Primitive(PrimitiveType::Long)).into(),
1120            ])
1121            .build()
1122            .unwrap();
1123        let schema_three = Schema::builder()
1124            .with_schema_id(9)
1125            .with_fields(vec![
1126                NestedField::required(1, "z", Type::Primitive(PrimitiveType::Long)).into(),
1127            ])
1128            .build()
1129            .unwrap();
1130
1131        let v1 = new_view_version(1, 5, "select * from ns.tbl");
1132        let v2 = new_view_version(1, 7, "select count(*) from ns.tbl");
1133        let v3 = new_view_version(1, 9, "select count(*) as count from ns.tbl");
1134
1135        let build_result = builder_without_changes()
1136            .add_schema(schema_one.clone())
1137            .add_schema(schema_two.clone())
1138            .add_schema(schema_three.clone())
1139            .set_current_version(v1.clone(), schema_one.clone())
1140            .unwrap()
1141            .set_current_version(v2.clone(), schema_two.clone())
1142            .unwrap()
1143            .set_current_version(v3.clone(), schema_three.clone())
1144            .unwrap()
1145            .set_current_version(v3.clone(), schema_three.clone())
1146            .unwrap()
1147            .set_current_version(v2.clone(), schema_two.clone())
1148            .unwrap()
1149            .set_current_version(v1.clone(), schema_one.clone())
1150            .unwrap()
1151            .build()
1152            .unwrap();
1153
1154        assert_eq!(
1155            Arc::unwrap_or_clone(build_result.metadata.current_version().clone()),
1156            v1.clone().with_version_id(2).with_schema_id(2)
1157        );
1158        assert_eq!(build_result.metadata.versions.len(), 4);
1159        assert_eq!(
1160            build_result.metadata.versions[&2],
1161            Arc::new(v1.clone().with_version_id(2).with_schema_id(2))
1162        );
1163        assert_eq!(
1164            build_result.metadata.versions[&3],
1165            Arc::new(v2.clone().with_version_id(3).with_schema_id(3))
1166        );
1167        assert_eq!(
1168            build_result.metadata.versions[&4],
1169            Arc::new(v3.clone().with_version_id(4).with_schema_id(4))
1170        );
1171        assert_eq!(
1172            // Remove schema_id 1 and get struct only
1173            build_result
1174                .metadata
1175                .schemas_iter()
1176                .filter(|s| s.schema_id() != 1)
1177                .sorted_by_key(|s| s.schema_id())
1178                .map(|s| s.as_struct())
1179                .collect::<Vec<_>>(),
1180            vec![
1181                schema_one.as_struct(),
1182                schema_two.as_struct(),
1183                schema_three.as_struct()
1184            ]
1185        )
1186    }
1187
1188    #[test]
1189    fn test_error_on_missing_schema() {
1190        let builder = builder_without_changes();
1191        // Missing schema
1192        assert!(
1193            builder
1194                .clone()
1195                .add_version(new_view_version(0, 10, "SELECT * FROM foo"))
1196                .unwrap_err()
1197                .to_string()
1198                .contains("Cannot add version with unknown schema: 10")
1199        );
1200
1201        // Missing last added schema
1202        assert!(
1203            builder
1204                .clone()
1205                .add_version(new_view_version(0, -1, "SELECT * FROM foo"))
1206                .unwrap_err()
1207                .to_string()
1208                .contains("Cannot set last added schema: no schema has been added")
1209        );
1210    }
1211
1212    #[test]
1213    fn test_error_on_missing_current_version() {
1214        let builder = builder_without_changes();
1215        assert!(builder
1216            .clone()
1217            .set_current_version_id(-1)
1218            .unwrap_err()
1219            .to_string()
1220            .contains(
1221                "Cannot set current version id to last added version: no version has been added."
1222            ));
1223        assert!(
1224            builder
1225                .clone()
1226                .set_current_version_id(10)
1227                .unwrap_err()
1228                .to_string()
1229                .contains("Cannot set current version to unknown version with id: 10")
1230        );
1231    }
1232
1233    #[test]
1234    fn test_set_current_version_to_last_added() {
1235        let builder = builder_without_changes();
1236        let v1 = new_view_version(2, 1, "select * from ns.tbl");
1237        let v2 = new_view_version(3, 1, "select a,b from ns.tbl");
1238        let meta = builder
1239            .clone()
1240            .add_version(v1)
1241            .unwrap()
1242            .add_version(v2)
1243            .unwrap()
1244            .set_current_version_id(-1)
1245            .unwrap()
1246            .build()
1247            .unwrap();
1248        assert_eq!(meta.metadata.current_version_id, 3);
1249    }
1250
1251    #[test]
1252    fn test_error_when_setting_negative_version_history_size() {
1253        let builder = builder_without_changes();
1254        assert!(
1255            builder
1256                .clone()
1257                .set_properties(HashMap::from_iter(vec![(
1258                    VIEW_PROPERTY_VERSION_HISTORY_SIZE.to_string(),
1259                    "-1".to_string(),
1260                )]))
1261                .unwrap_err()
1262                .to_string()
1263                .contains("version.history.num-entries must be positive but was -1")
1264        );
1265    }
1266
1267    #[test]
1268    fn test_view_version_changes() {
1269        let builder = builder_without_changes();
1270
1271        let v1 = new_view_version(2, 1, "select 1 as count");
1272        let v2 = new_view_version(3, 1, "select count(1) as count from t2");
1273
1274        let changes = builder
1275            .clone()
1276            .add_version(v1.clone())
1277            .unwrap()
1278            .add_version(v2.clone())
1279            .unwrap()
1280            .build()
1281            .unwrap()
1282            .changes;
1283
1284        assert_eq!(changes.len(), 2);
1285        assert_eq!(changes, vec![
1286            ViewUpdate::AddViewVersion {
1287                view_version: v1.clone()
1288            },
1289            ViewUpdate::AddViewVersion {
1290                view_version: v2.clone()
1291            }
1292        ]);
1293    }
1294
1295    #[test]
1296    fn test_dropping_dialect_fails_by_default() {
1297        let builder = builder_without_changes();
1298
1299        let spark = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark"]);
1300        let spark_trino =
1301            new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark", "trino"]);
1302        let schema = Schema::builder()
1303            .with_schema_id(0)
1304            .with_fields(vec![])
1305            .build()
1306            .unwrap();
1307
1308        let err = builder
1309            .set_current_version(spark_trino, schema.clone())
1310            .unwrap()
1311            .build()
1312            .unwrap()
1313            .metadata
1314            .into_builder()
1315            .set_current_version(spark, schema)
1316            .unwrap()
1317            .build()
1318            .unwrap_err();
1319
1320        assert!(
1321            err.to_string()
1322                .contains("Cannot replace view due to loss of view dialects")
1323        );
1324    }
1325
1326    #[test]
1327    fn test_dropping_dialects_does_not_fail_when_allowed() {
1328        let builder = builder_without_changes();
1329
1330        let spark = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark"]);
1331        let spark_trino =
1332            new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark", "trino"]);
1333        let schema = Schema::builder()
1334            .with_schema_id(0)
1335            .with_fields(vec![])
1336            .build()
1337            .unwrap();
1338
1339        let build_result = builder
1340            .set_properties(HashMap::from_iter(vec![(
1341                VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED.to_string(),
1342                "true".to_string(),
1343            )]))
1344            .unwrap()
1345            .set_current_version(spark_trino, schema.clone())
1346            .unwrap()
1347            .build()
1348            .unwrap()
1349            .metadata
1350            .into_builder()
1351            .set_current_version(spark.clone(), schema)
1352            .unwrap()
1353            .build()
1354            .unwrap();
1355
1356        assert_eq!(
1357            Arc::unwrap_or_clone(build_result.metadata.current_version().clone()),
1358            spark.with_version_id(3).with_schema_id(2)
1359        );
1360    }
1361
1362    #[test]
1363    fn test_can_add_dialects_by_default() {
1364        let builder = builder_without_changes();
1365
1366        let spark = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark"]);
1367        let spark_trino =
1368            new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark", "trino"]);
1369
1370        let schema = Schema::builder()
1371            .with_schema_id(0)
1372            .with_fields(vec![])
1373            .build()
1374            .unwrap();
1375
1376        let build_result = builder
1377            .set_current_version(spark.clone(), schema.clone())
1378            .unwrap()
1379            .build()
1380            .unwrap()
1381            .metadata
1382            .into_builder()
1383            .set_current_version(spark_trino.clone(), schema.clone())
1384            .unwrap()
1385            .build()
1386            .unwrap();
1387
1388        assert_eq!(
1389            Arc::unwrap_or_clone(build_result.metadata.current_version().clone()),
1390            spark_trino.with_version_id(3).with_schema_id(2)
1391        );
1392    }
1393
1394    #[test]
1395    fn test_can_update_dialect_by_default() {
1396        let builder = builder_without_changes();
1397
1398        let spark_v1 = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark"]);
1399        let spark_v2 = new_view_version_with_dialect(0, 0, "SELECT * FROM bar", vec!["spark"]);
1400
1401        let schema = Schema::builder()
1402            .with_schema_id(0)
1403            .with_fields(vec![])
1404            .build()
1405            .unwrap();
1406
1407        let build_result = builder
1408            .set_current_version(spark_v1.clone(), schema.clone())
1409            .unwrap()
1410            .build()
1411            .unwrap()
1412            .metadata
1413            .into_builder()
1414            .set_current_version(spark_v2.clone(), schema.clone())
1415            .unwrap()
1416            .build()
1417            .unwrap();
1418
1419        assert_eq!(
1420            Arc::unwrap_or_clone(build_result.metadata.current_version().clone()),
1421            spark_v2.with_version_id(3).with_schema_id(2)
1422        );
1423    }
1424
1425    #[test]
1426    fn test_dropping_dialects_allowed_and_then_disallowed() {
1427        let builder = builder_without_changes();
1428
1429        let spark = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["spark"]);
1430        let trino = new_view_version_with_dialect(0, 0, "SELECT * FROM foo", vec!["trino"]);
1431
1432        let schema = Schema::builder()
1433            .with_schema_id(0)
1434            .with_fields(vec![])
1435            .build()
1436            .unwrap();
1437
1438        let updated = builder
1439            .set_current_version(spark.clone(), schema.clone())
1440            .unwrap()
1441            .build()
1442            .unwrap()
1443            .metadata
1444            .into_builder()
1445            .set_current_version(trino.clone(), schema.clone())
1446            .unwrap()
1447            .set_properties(HashMap::from_iter(vec![(
1448                VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED.to_string(),
1449                "true".to_string(),
1450            )]))
1451            .unwrap()
1452            .build()
1453            .unwrap();
1454
1455        assert_eq!(
1456            Arc::unwrap_or_clone(updated.metadata.current_version().clone()),
1457            trino.with_version_id(3).with_schema_id(2)
1458        );
1459
1460        let err = updated
1461            .metadata
1462            .into_builder()
1463            .set_current_version(spark.clone(), schema.clone())
1464            .unwrap()
1465            .set_properties(HashMap::from_iter(vec![(
1466                VIEW_PROPERTY_REPLACE_DROP_DIALECT_ALLOWED.to_string(),
1467                "false".to_string(),
1468            )]))
1469            .unwrap()
1470            .build()
1471            .unwrap_err();
1472
1473        assert!(
1474            err.to_string()
1475                .contains("Cannot replace view due to loss of view dialects")
1476        );
1477    }
1478
1479    #[test]
1480    fn test_require_no_dialect_dropped() {
1481        let previous = ViewVersion::builder()
1482            .with_version_id(0)
1483            .with_schema_id(0)
1484            .with_timestamp_ms(0)
1485            .with_representations(ViewRepresentations(vec![
1486                ViewRepresentation::Sql(SqlViewRepresentation {
1487                    dialect: "trino".to_string(),
1488                    sql: "SELECT * FROM foo".to_string(),
1489                }),
1490                ViewRepresentation::Sql(SqlViewRepresentation {
1491                    dialect: "spark".to_string(),
1492                    sql: "SELECT * FROM bar".to_string(),
1493                }),
1494            ]))
1495            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1496            .build();
1497
1498        let current = ViewVersion::builder()
1499            .with_version_id(0)
1500            .with_schema_id(0)
1501            .with_timestamp_ms(0)
1502            .with_representations(ViewRepresentations(vec![ViewRepresentation::Sql(
1503                SqlViewRepresentation {
1504                    dialect: "trino".to_string(),
1505                    sql: "SELECT * FROM foo".to_string(),
1506                },
1507            )]))
1508            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1509            .build();
1510
1511        assert!(require_no_dialect_dropped(&previous, &current).is_err());
1512
1513        let current = ViewVersion::builder()
1514            .with_version_id(0)
1515            .with_schema_id(0)
1516            .with_timestamp_ms(0)
1517            .with_representations(ViewRepresentations(vec![
1518                ViewRepresentation::Sql(SqlViewRepresentation {
1519                    dialect: "spark".to_string(),
1520                    sql: "SELECT * FROM bar".to_string(),
1521                }),
1522                ViewRepresentation::Sql(SqlViewRepresentation {
1523                    dialect: "trino".to_string(),
1524                    sql: "SELECT * FROM foo".to_string(),
1525                }),
1526            ]))
1527            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1528            .build();
1529
1530        assert!(require_no_dialect_dropped(&previous, &current).is_ok());
1531    }
1532
1533    #[test]
1534    fn test_allow_replace_drop_dialects() {
1535        use std::collections::HashMap;
1536
1537        use super::allow_replace_drop_dialects;
1538
1539        let mut properties = HashMap::new();
1540        assert!(!allow_replace_drop_dialects(&properties));
1541
1542        properties.insert(
1543            "replace.drop-dialect.allowed".to_string(),
1544            "true".to_string(),
1545        );
1546        assert!(allow_replace_drop_dialects(&properties));
1547
1548        properties.insert(
1549            "replace.drop-dialect.allowed".to_string(),
1550            "false".to_string(),
1551        );
1552        assert!(!allow_replace_drop_dialects(&properties));
1553
1554        properties.insert(
1555            "replace.drop-dialect.allowed".to_string(),
1556            "TRUE".to_string(),
1557        );
1558        assert!(allow_replace_drop_dialects(&properties));
1559
1560        properties.insert(
1561            "replace.drop-dialect.allowed".to_string(),
1562            "FALSE".to_string(),
1563        );
1564        assert!(!allow_replace_drop_dialects(&properties));
1565    }
1566
1567    #[test]
1568    fn test_lowercase_sql_dialects_for() {
1569        let view_version = ViewVersion::builder()
1570            .with_version_id(0)
1571            .with_schema_id(0)
1572            .with_timestamp_ms(0)
1573            .with_representations(ViewRepresentations(vec![
1574                ViewRepresentation::Sql(SqlViewRepresentation {
1575                    dialect: "STARROCKS".to_string(),
1576                    sql: "SELECT * FROM foo".to_string(),
1577                }),
1578                ViewRepresentation::Sql(SqlViewRepresentation {
1579                    dialect: "trino".to_string(),
1580                    sql: "SELECT * FROM bar".to_string(),
1581                }),
1582                ViewRepresentation::Sql(SqlViewRepresentation {
1583                    dialect: "Spark".to_string(),
1584                    sql: "SELECT * FROM bar".to_string(),
1585                }),
1586            ]))
1587            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1588            .build();
1589
1590        let dialects = lowercase_sql_dialects_for(&view_version);
1591        assert_eq!(dialects.len(), 3);
1592        assert!(dialects.contains("trino"));
1593        assert!(dialects.contains("spark"));
1594        assert!(dialects.contains("starrocks"));
1595    }
1596
1597    #[test]
1598    fn test_require_unique_dialects() {
1599        let view_version = ViewVersion::builder()
1600            .with_version_id(0)
1601            .with_schema_id(0)
1602            .with_timestamp_ms(0)
1603            .with_representations(ViewRepresentations(vec![
1604                ViewRepresentation::Sql(SqlViewRepresentation {
1605                    dialect: "trino".to_string(),
1606                    sql: "SELECT * FROM foo".to_string(),
1607                }),
1608                ViewRepresentation::Sql(SqlViewRepresentation {
1609                    dialect: "trino".to_string(),
1610                    sql: "SELECT * FROM bar".to_string(),
1611                }),
1612            ]))
1613            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1614            .build();
1615
1616        assert!(require_unique_dialects(&view_version).is_err());
1617
1618        let view_version = ViewVersion::builder()
1619            .with_version_id(0)
1620            .with_schema_id(0)
1621            .with_timestamp_ms(0)
1622            .with_representations(ViewRepresentations(vec![
1623                ViewRepresentation::Sql(SqlViewRepresentation {
1624                    dialect: "trino".to_string(),
1625                    sql: "SELECT * FROM foo".to_string(),
1626                }),
1627                ViewRepresentation::Sql(SqlViewRepresentation {
1628                    dialect: "spark".to_string(),
1629                    sql: "SELECT * FROM bar".to_string(),
1630                }),
1631            ]))
1632            .with_default_namespace(NamespaceIdent::new("default".to_string()))
1633            .build();
1634
1635        assert!(require_unique_dialects(&view_version).is_ok());
1636    }
1637}
iceberg/spec/view_metadata_builder.rs

iceberg/spec/
view_metadata_builder.rs
