1use crate::project::ir::object_id::ObjectId;
18use crate::types::{ColumnType, ObjectKind};
19use rusqlite::{Connection, OpenFlags, params};
20use std::collections::{BTreeMap, BTreeSet};
21use std::path::Path;
22
23#[derive(Debug, Clone)]
25pub struct CachedObject {
26 pub fqn: String,
27 pub database: String,
28 pub schema: String,
29 pub name: String,
30 pub kind: ObjectKind,
31 pub cluster: Option<String>,
32 pub file_path: String,
33 pub sql_text: String,
34 pub comments: Vec<CachedComment>,
35 pub indexes: Vec<CachedIndex>,
36 pub grants: Vec<CachedGrant>,
37 pub aliases: BTreeMap<String, String>,
38 pub infrastructure: Option<CachedInfrastructure>,
39}
40
41#[derive(Debug, Clone)]
43pub struct CachedObjectSummary {
44 pub fqn: String,
45 pub database: String,
46 pub schema: String,
47 pub name: String,
48 pub kind: ObjectKind,
49 pub cluster: Option<String>,
50 pub file_path: String,
51}
52
53#[derive(Debug, Clone)]
55pub struct CachedDatabase {
56 pub name: String,
57 pub schemas: Vec<CachedSchema>,
58}
59
60#[derive(Debug, Clone)]
62pub struct CachedSchema {
63 pub name: String,
64 pub schema_type: String,
65 pub objects: Vec<CachedObject>,
66}
67
68#[derive(Debug, Clone)]
70pub struct CachedComment {
71 pub comment_type: String,
72 pub target_column: Option<String>,
73 pub text: String,
74 pub sql_text: String,
75}
76
77#[derive(Debug, Clone)]
79pub struct CachedIndex {
80 pub name: String,
81 pub cluster: Option<String>,
82 pub columns: String,
83 pub sql_text: String,
84}
85
86#[derive(Debug, Clone)]
88pub struct CachedGrant {
89 pub privilege: String,
90 pub grantee: String,
91 pub sql_text: String,
92}
93
94#[derive(Debug, Clone)]
96pub struct CachedInfrastructure {
97 pub infra_type: String,
98 pub connector_type: Option<String>,
99 pub connection_ref: Option<String>,
100 pub source_ref: Option<String>,
101 pub external_reference: Option<String>,
102 pub properties: Vec<CachedProperty>,
103}
104
105#[derive(Debug, Clone)]
107pub struct CachedProperty {
108 pub key: String,
109 pub value: String,
110 pub secret_ref: Option<String>,
111 pub object_ref: Option<String>,
112}
113
114#[derive(Debug, Clone)]
116pub struct CachedTest {
117 pub name: String,
118 pub sql_text: String,
119}
120
121pub struct ProjectCache {
126 conn: Connection,
127}
128
129impl ProjectCache {
130 pub fn open(
135 directory: &Path,
136 profile: &str,
137 profile_suffix: Option<&str>,
138 variables: &BTreeMap<String, String>,
139 ) -> Result<Option<Self>, super::CacheError> {
140 let path = super::db_path(directory, profile, profile_suffix, variables);
141 if !path.exists() {
142 return Ok(None);
143 }
144 let conn = Connection::open_with_flags(
145 &path,
146 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
147 )
148 .map_err(|source| super::CacheError::DatabaseOpenFailed {
149 path: path.clone(),
150 source,
151 })?;
152 Ok(Some(Self { conn }))
153 }
154
155 fn query_vec<T, P, F>(&self, sql: &str, params: P, map: F) -> Vec<T>
158 where
159 P: rusqlite::Params,
160 F: FnMut(&rusqlite::Row<'_>) -> rusqlite::Result<T>,
161 {
162 let Ok(mut stmt) = self.conn.prepare(sql) else {
163 return Vec::new();
164 };
165 let Ok(rows) = stmt.query_map(params, map) else {
166 return Vec::new();
167 };
168 rows.filter_map(|r| r.ok()).collect()
169 }
170
171 pub fn get_columns(&self, id: &ObjectId) -> Option<BTreeMap<String, ColumnType>> {
173 let mut stmt = self
174 .conn
175 .prepare(
176 "SELECT column_name, column_type, nullable, position \
177 FROM typecheck_columns WHERE object_key = ?1",
178 )
179 .ok()?;
180 let rows = stmt
181 .query_map(params![id.to_string()], |row| {
182 Ok((
183 row.get::<_, String>(0)?,
184 ColumnType {
185 r#type: super::decode_column_type(row.get(1)?)?,
186 nullable: row.get::<_, i32>(2)? != 0,
187 position: usize::try_from(row.get::<_, i64>(3)?).unwrap_or(0),
188 comment: None,
189 },
190 ))
191 })
192 .ok()?;
193 let mut columns = BTreeMap::new();
194 for row in rows {
195 let (name, col_type) = row.ok()?;
196 columns.insert(name, col_type);
197 }
198 if columns.is_empty() {
199 None
200 } else {
201 Some(columns)
202 }
203 }
204
205 pub fn get_kind(&self, id: &ObjectId) -> Option<ObjectKind> {
207 self.conn
208 .query_row(
209 "SELECT object_kind FROM typecheck_objects WHERE object_key = ?1",
210 params![id.to_string()],
211 |row| {
212 let kind_str: String = row.get(0)?;
213 Ok(ObjectKind::from_db_str(&kind_str))
214 },
215 )
216 .ok()
217 }
218
219 pub fn get_column_names(&self, ids: &[&ObjectId]) -> BTreeMap<String, BTreeSet<String>> {
224 if ids.is_empty() {
225 return BTreeMap::new();
226 }
227 let placeholders: Vec<String> = (1..=ids.len()).map(|i| format!("?{}", i)).collect();
228 let sql = format!(
229 "SELECT object_key, column_name FROM typecheck_columns WHERE object_key IN ({})",
230 placeholders.join(", ")
231 );
232 let mut stmt = match self.conn.prepare(&sql) {
233 Ok(s) => s,
234 Err(_) => return BTreeMap::new(),
235 };
236 let key_strings: Vec<String> = ids.iter().map(|id| id.to_string()).collect();
237 let params: Vec<&dyn rusqlite::ToSql> = key_strings
238 .iter()
239 .map(|s| -> &dyn rusqlite::ToSql { s })
240 .collect();
241 let rows = match stmt.query_map(params.as_slice(), |row| {
242 Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
243 }) {
244 Ok(r) => r,
245 Err(_) => return BTreeMap::new(),
246 };
247 let mut result: BTreeMap<String, BTreeSet<String>> = BTreeMap::new();
248 for row in rows {
249 if let Ok((key, col)) = row {
250 result
251 .entry(key.to_lowercase())
252 .or_default()
253 .insert(col.to_lowercase());
254 }
255 }
256 result
257 }
258
259 pub fn get_object(&self, id: &ObjectId) -> Option<CachedObject> {
261 let fqn = id.to_string();
262 let row = self
264 .conn
265 .query_row(
266 "SELECT database, schema, name, object_kind, cluster, \
267 file_path, sql_text \
268 FROM project_objects WHERE object_key = ?1",
269 params![fqn],
270 |row| {
271 Ok((
272 row.get::<_, String>(0)?,
273 row.get::<_, String>(1)?,
274 row.get::<_, String>(2)?,
275 row.get::<_, String>(3)?,
276 row.get::<_, Option<String>>(4)?,
277 row.get::<_, String>(5)?,
278 row.get::<_, String>(6)?,
279 ))
280 },
281 )
282 .ok()?;
283
284 let (database, schema, name, kind_str, cluster, file_path, sql_text) = row;
285 let kind = ObjectKind::from_db_str(&kind_str);
286
287 let comments = self.query_comments(&fqn);
288 let indexes = self.query_indexes(&fqn);
289 let grants = self.query_grants(&fqn);
290 let aliases = self.query_aliases(&fqn);
291 let infrastructure = self.query_infrastructure(&fqn);
292
293 Some(CachedObject {
294 fqn,
295 database,
296 schema,
297 name,
298 kind,
299 cluster,
300 file_path,
301 sql_text,
302 comments,
303 indexes,
304 grants,
305 aliases,
306 infrastructure,
307 })
308 }
309
310 pub fn get_object_by_path(&self, file_path: &str) -> Option<CachedObject> {
312 let fqn: String = self
313 .conn
314 .query_row(
315 "SELECT object_key FROM project_objects WHERE file_path = ?1",
316 params![file_path],
317 |row| row.get(0),
318 )
319 .ok()?;
320 let id = fqn.parse::<ObjectId>().ok()?;
321 self.get_object(&id)
322 }
323
324 pub fn list_objects(&self) -> Vec<CachedObjectSummary> {
326 self.query_vec(
327 "SELECT object_key, database, schema, name, object_kind, cluster, \
328 file_path FROM project_objects",
329 [],
330 |row| {
331 Ok(CachedObjectSummary {
332 fqn: row.get(0)?,
333 database: row.get(1)?,
334 schema: row.get(2)?,
335 name: row.get(3)?,
336 kind: ObjectKind::from_db_str(&row.get::<_, String>(4)?),
337 cluster: row.get(5)?,
338 file_path: row.get(6)?,
339 })
340 },
341 )
342 }
343
344 pub fn list_databases_with_objects(&self) -> Vec<CachedDatabase> {
347 let db_names: Vec<String> =
348 self.query_vec("SELECT name FROM project_databases", [], |row| row.get(0));
349
350 db_names
351 .into_iter()
352 .map(|db_name| {
353 let schema_rows: Vec<(String, String)> = self.query_vec(
354 "SELECT name, schema_type FROM project_schemas WHERE database = ?1",
355 params![&db_name],
356 |row| Ok((row.get(0)?, row.get(1)?)),
357 );
358 let schemas = schema_rows
359 .into_iter()
360 .map(|(schema_name, schema_type)| {
361 let object_ids = self.query_object_keys_in_schema(&db_name, &schema_name);
362 let objects = object_ids
363 .iter()
364 .filter_map(|id| self.get_object(id))
365 .collect();
366 CachedSchema {
367 name: schema_name,
368 schema_type,
369 objects,
370 }
371 })
372 .collect();
373 CachedDatabase {
374 name: db_name,
375 schemas,
376 }
377 })
378 .collect()
379 }
380
381 pub fn list_external_dependencies(&self) -> Vec<ObjectId> {
383 self.query_vec(
384 "SELECT object_key FROM project_external_dependencies",
385 [],
386 |row| row.get::<_, String>(0),
387 )
388 .into_iter()
389 .filter_map(|s| s.parse().ok())
390 .collect()
391 }
392
393 pub fn get_dependencies(&self, id: &ObjectId) -> Vec<ObjectId> {
395 self.query_vec(
396 "SELECT dependency_key FROM project_dependencies WHERE object_key = ?1",
397 params![id.to_string()],
398 |row| row.get::<_, String>(0),
399 )
400 .into_iter()
401 .filter_map(|s| s.parse().ok())
402 .collect()
403 }
404
405 pub fn get_dependents(&self, id: &ObjectId) -> Vec<ObjectId> {
407 self.query_vec(
408 "SELECT object_key FROM project_dependencies WHERE dependency_key = ?1",
409 params![id.to_string()],
410 |row| row.get::<_, String>(0),
411 )
412 .into_iter()
413 .filter_map(|s| s.parse().ok())
414 .collect()
415 }
416
417 pub fn get_tests(&self, id: &ObjectId) -> Vec<CachedTest> {
419 self.query_vec(
420 "SELECT test_name, sql_text FROM project_tests WHERE object_key = ?1",
421 params![id.to_string()],
422 |row| {
423 Ok(CachedTest {
424 name: row.get(0)?,
425 sql_text: row.get(1)?,
426 })
427 },
428 )
429 }
430
431 pub fn get_mod_statements(&self, database: &str, schema: Option<&str>) -> Vec<String> {
433 self.query_vec(
434 "SELECT sql_text FROM project_mod_statements \
435 WHERE database = ?1 AND schema IS ?2 \
436 ORDER BY position",
437 params![database, schema],
438 |row| row.get(0),
439 )
440 }
441
442 fn query_comments(&self, object_key: &str) -> Vec<CachedComment> {
443 self.query_vec(
444 "SELECT comment_type, target_column, comment_text, sql_text \
445 FROM project_comments WHERE object_key = ?1",
446 params![object_key],
447 |row| {
448 Ok(CachedComment {
449 comment_type: row.get(0)?,
450 target_column: row.get(1)?,
451 text: row.get(2)?,
452 sql_text: row.get(3)?,
453 })
454 },
455 )
456 }
457
458 fn query_indexes(&self, object_key: &str) -> Vec<CachedIndex> {
459 self.query_vec(
460 "SELECT index_name, cluster, columns, sql_text \
461 FROM project_indexes WHERE object_key = ?1",
462 params![object_key],
463 |row| {
464 Ok(CachedIndex {
465 name: row.get::<_, Option<String>>(0)?.unwrap_or_default(),
466 cluster: row.get(1)?,
467 columns: row.get(2)?,
468 sql_text: row.get(3)?,
469 })
470 },
471 )
472 }
473
474 fn query_grants(&self, object_key: &str) -> Vec<CachedGrant> {
475 self.query_vec(
476 "SELECT privilege, grantee, sql_text \
477 FROM project_grants WHERE object_key = ?1",
478 params![object_key],
479 |row| {
480 Ok(CachedGrant {
481 privilege: row.get(0)?,
482 grantee: row.get(1)?,
483 sql_text: row.get(2)?,
484 })
485 },
486 )
487 }
488
489 fn query_infrastructure(&self, object_key: &str) -> Option<CachedInfrastructure> {
490 let row = self
491 .conn
492 .query_row(
493 "SELECT infra_type, connector_type, connection_ref, source_ref, external_reference \
494 FROM project_infrastructure WHERE object_key = ?1",
495 params![object_key],
496 |row| {
497 Ok((
498 row.get::<_, String>(0)?,
499 row.get::<_, Option<String>>(1)?,
500 row.get::<_, Option<String>>(2)?,
501 row.get::<_, Option<String>>(3)?,
502 row.get::<_, Option<String>>(4)?,
503 ))
504 },
505 )
506 .ok()?;
507
508 let (infra_type, connector_type, connection_ref, source_ref, external_reference) = row;
509 let properties = self.query_infrastructure_properties(object_key);
510
511 Some(CachedInfrastructure {
512 infra_type,
513 connector_type,
514 connection_ref,
515 source_ref,
516 external_reference,
517 properties,
518 })
519 }
520
521 fn query_infrastructure_properties(&self, object_key: &str) -> Vec<CachedProperty> {
522 self.query_vec(
523 "SELECT property_key, property_value, secret_ref, object_ref \
524 FROM project_infrastructure_properties WHERE object_key = ?1",
525 params![object_key],
526 |row| {
527 Ok(CachedProperty {
528 key: row.get(0)?,
529 value: row.get(1)?,
530 secret_ref: row.get(2)?,
531 object_ref: row.get(3)?,
532 })
533 },
534 )
535 }
536
537 fn query_aliases(&self, object_key: &str) -> BTreeMap<String, String> {
538 self.query_vec(
539 "SELECT alias, target_fqn FROM project_aliases WHERE object_key = ?1",
540 params![object_key],
541 |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
542 )
543 .into_iter()
544 .collect()
545 }
546
547 fn query_object_keys_in_schema(&self, database: &str, schema: &str) -> Vec<ObjectId> {
548 self.query_vec(
549 "SELECT object_key FROM project_objects WHERE database = ?1 AND schema = ?2",
550 params![database, schema],
551 |row| row.get::<_, String>(0),
552 )
553 .into_iter()
554 .filter_map(|s| s.parse().ok())
555 .collect()
556 }
557}
558
559#[cfg(test)]
560mod tests {
561 use super::*;
562 use crate::types::DataType;
563 use rusqlite::Connection;
564 use std::collections::BTreeMap;
565
566 fn create_test_db(path: &Path) -> Connection {
568 let conn = Connection::open(path).unwrap();
569 conn.execute_batch(
570 "
571 CREATE TABLE IF NOT EXISTS typecheck_objects (
572 object_key TEXT PRIMARY KEY,
573 object_kind TEXT NOT NULL
574 );
575 CREATE TABLE IF NOT EXISTS typecheck_columns (
576 object_key TEXT NOT NULL,
577 column_name TEXT NOT NULL,
578 column_type TEXT NOT NULL,
579 nullable INTEGER NOT NULL,
580 position INTEGER NOT NULL,
581 PRIMARY KEY (object_key, column_name),
582 FOREIGN KEY (object_key) REFERENCES typecheck_objects(object_key)
583 );
584 CREATE TABLE IF NOT EXISTS project_databases (
585 name TEXT PRIMARY KEY
586 );
587 CREATE TABLE IF NOT EXISTS project_schemas (
588 database TEXT NOT NULL,
589 name TEXT NOT NULL,
590 schema_type TEXT NOT NULL,
591 PRIMARY KEY (database, name)
592 );
593 CREATE TABLE IF NOT EXISTS project_objects (
594 object_key TEXT PRIMARY KEY,
595 database TEXT NOT NULL,
596 schema TEXT NOT NULL,
597 name TEXT NOT NULL,
598 object_kind TEXT NOT NULL,
599 cluster TEXT,
600 file_path TEXT NOT NULL,
601 sql_text TEXT NOT NULL
602 );
603 CREATE TABLE IF NOT EXISTS project_dependencies (
604 object_key TEXT NOT NULL,
605 dependency_key TEXT NOT NULL,
606 PRIMARY KEY (object_key, dependency_key)
607 );
608 CREATE TABLE IF NOT EXISTS project_external_dependencies (
609 object_key TEXT NOT NULL PRIMARY KEY
610 );
611 CREATE TABLE IF NOT EXISTS project_comments (
612 object_key TEXT NOT NULL,
613 comment_type TEXT NOT NULL,
614 target_column TEXT,
615 comment_text TEXT NOT NULL,
616 sql_text TEXT NOT NULL,
617 PRIMARY KEY (object_key, comment_type, target_column)
618 );
619 CREATE TABLE IF NOT EXISTS project_indexes (
620 object_key TEXT NOT NULL,
621 index_name TEXT,
622 cluster TEXT,
623 columns TEXT NOT NULL,
624 sql_text TEXT NOT NULL,
625 PRIMARY KEY (object_key, index_name)
626 );
627 CREATE TABLE IF NOT EXISTS project_grants (
628 object_key TEXT NOT NULL,
629 privilege TEXT NOT NULL,
630 grantee TEXT NOT NULL,
631 sql_text TEXT NOT NULL,
632 PRIMARY KEY (object_key, privilege, grantee)
633 );
634 CREATE TABLE IF NOT EXISTS project_tests (
635 object_key TEXT NOT NULL,
636 test_name TEXT NOT NULL,
637 sql_text TEXT NOT NULL,
638 PRIMARY KEY (object_key, test_name)
639 );
640 CREATE TABLE IF NOT EXISTS project_infrastructure (
641 object_key TEXT NOT NULL PRIMARY KEY,
642 infra_type TEXT NOT NULL,
643 connector_type TEXT,
644 connection_ref TEXT,
645 source_ref TEXT,
646 external_reference TEXT
647 );
648 CREATE TABLE IF NOT EXISTS project_infrastructure_properties (
649 object_key TEXT NOT NULL,
650 property_key TEXT NOT NULL,
651 property_value TEXT NOT NULL,
652 secret_ref TEXT,
653 object_ref TEXT,
654 PRIMARY KEY (object_key, property_key)
655 );
656 CREATE TABLE IF NOT EXISTS project_aliases (
657 object_key TEXT NOT NULL,
658 alias TEXT NOT NULL,
659 target_fqn TEXT NOT NULL,
660 PRIMARY KEY (object_key, alias)
661 );
662 CREATE TABLE IF NOT EXISTS project_mod_statements (
663 database TEXT NOT NULL,
664 schema TEXT,
665 position INTEGER NOT NULL,
666 sql_text TEXT NOT NULL,
667 PRIMARY KEY (database, schema, position)
668 );
669 ",
670 )
671 .unwrap();
672 conn
673 }
674
675 fn open_cache(path: &Path) -> ProjectCache {
676 ProjectCache {
677 conn: Connection::open_with_flags(
678 path,
679 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
680 )
681 .unwrap(),
682 }
683 }
684
685 #[cfg_attr(miri, ignore)] #[mz_ore::test]
687 fn test_open_returns_none_when_no_db() {
688 let dir = tempfile::tempdir().unwrap();
689 let result = ProjectCache::open(dir.path(), "default", None, &BTreeMap::new());
690 assert!(result.is_ok());
691 assert!(result.unwrap().is_none());
692 }
693
694 #[cfg_attr(miri, ignore)] #[mz_ore::test]
696 fn test_get_columns_found() {
697 let dir = tempfile::tempdir().unwrap();
698 let db_path = dir.path().join("test.db");
699 let conn = create_test_db(&db_path);
700 conn.execute(
701 "INSERT INTO typecheck_objects (object_key, object_kind) VALUES (?1, ?2)",
702 params!["db.schema.my_view", "view"],
703 )
704 .unwrap();
705 conn.execute(
706 "INSERT INTO typecheck_columns (object_key, column_name, column_type, nullable, position) \
707 VALUES (?1, ?2, ?3, ?4, ?5)",
708 params![
709 "db.schema.my_view",
710 "id",
711 DataType::named("integer").to_json(),
712 0,
713 1
714 ],
715 )
716 .unwrap();
717 conn.execute(
718 "INSERT INTO typecheck_columns (object_key, column_name, column_type, nullable, position) \
719 VALUES (?1, ?2, ?3, ?4, ?5)",
720 params![
721 "db.schema.my_view",
722 "name",
723 DataType::named("text").to_json(),
724 1,
725 2
726 ],
727 )
728 .unwrap();
729 drop(conn);
730
731 let cache = ProjectCache {
732 conn: Connection::open_with_flags(
733 &db_path,
734 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
735 )
736 .unwrap(),
737 };
738 let columns = cache
739 .get_columns(&"db.schema.my_view".parse::<ObjectId>().unwrap())
740 .unwrap();
741 assert_eq!(columns.len(), 2);
742
743 let id_col = &columns["id"];
744 assert_eq!(id_col.r#type, DataType::named("integer"));
745 assert!(!id_col.nullable);
746 assert_eq!(id_col.position, 1);
747
748 let name_col = &columns["name"];
749 assert_eq!(name_col.r#type, DataType::named("text"));
750 assert!(name_col.nullable);
751 assert_eq!(name_col.position, 2);
752 }
753
754 #[cfg_attr(miri, ignore)] #[mz_ore::test]
756 fn test_get_columns_not_found() {
757 let dir = tempfile::tempdir().unwrap();
758 let db_path = dir.path().join("test.db");
759 let _conn = create_test_db(&db_path);
760
761 let cache = ProjectCache {
762 conn: Connection::open_with_flags(
763 &db_path,
764 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
765 )
766 .unwrap(),
767 };
768 assert!(
769 cache
770 .get_columns(&"nonexistent.object.x".parse::<ObjectId>().unwrap())
771 .is_none()
772 );
773 }
774
775 #[cfg_attr(miri, ignore)] #[mz_ore::test]
777 fn test_get_kind_found() {
778 let dir = tempfile::tempdir().unwrap();
779 let db_path = dir.path().join("test.db");
780 let conn = create_test_db(&db_path);
781 conn.execute(
782 "INSERT INTO typecheck_objects (object_key, object_kind) VALUES (?1, ?2)",
783 params!["db.schema.my_mv", "materialized-view"],
784 )
785 .unwrap();
786 drop(conn);
787
788 let cache = ProjectCache {
789 conn: Connection::open_with_flags(
790 &db_path,
791 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
792 )
793 .unwrap(),
794 };
795 assert_eq!(
796 cache.get_kind(&"db.schema.my_mv".parse::<ObjectId>().unwrap()),
797 Some(ObjectKind::MaterializedView)
798 );
799 }
800
801 #[cfg_attr(miri, ignore)] #[mz_ore::test]
803 fn test_get_kind_not_found() {
804 let dir = tempfile::tempdir().unwrap();
805 let db_path = dir.path().join("test.db");
806 let _conn = create_test_db(&db_path);
807
808 let cache = ProjectCache {
809 conn: Connection::open_with_flags(
810 &db_path,
811 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
812 )
813 .unwrap(),
814 };
815 assert!(
816 cache
817 .get_kind(&"nonexistent.object.x".parse::<ObjectId>().unwrap())
818 .is_none()
819 );
820 }
821
822 #[cfg_attr(miri, ignore)] #[mz_ore::test]
824 fn test_get_column_names_batch() {
825 let dir = tempfile::tempdir().unwrap();
826 let db_path = dir.path().join("test.db");
827 let conn = create_test_db(&db_path);
828
829 conn.execute(
830 "INSERT INTO typecheck_objects (object_key, object_kind) VALUES (?1, ?2)",
831 params!["db.schema.obj_a", "view"],
832 )
833 .unwrap();
834 conn.execute(
835 "INSERT INTO typecheck_objects (object_key, object_kind) VALUES (?1, ?2)",
836 params!["db.schema.obj_b", "table"],
837 )
838 .unwrap();
839 conn.execute(
840 "INSERT INTO typecheck_columns (object_key, column_name, column_type, nullable, position) \
841 VALUES (?1, ?2, ?3, ?4, ?5)",
842 params!["db.schema.obj_a", "Col_X", "integer", 0, 1],
843 )
844 .unwrap();
845 conn.execute(
846 "INSERT INTO typecheck_columns (object_key, column_name, column_type, nullable, position) \
847 VALUES (?1, ?2, ?3, ?4, ?5)",
848 params!["db.schema.obj_b", "Col_Y", "text", 1, 1],
849 )
850 .unwrap();
851 drop(conn);
852
853 let cache = ProjectCache {
854 conn: Connection::open_with_flags(
855 &db_path,
856 OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
857 )
858 .unwrap(),
859 };
860
861 let id_a: ObjectId = "db.schema.obj_a".parse().unwrap();
862 let id_b: ObjectId = "db.schema.obj_b".parse().unwrap();
863 let result = cache.get_column_names(&[&id_a, &id_b]);
864 assert_eq!(result.len(), 2);
865 assert!(result["db.schema.obj_a"].contains("col_x"));
866 assert!(result["db.schema.obj_b"].contains("col_y"));
867 }
868
869 fn insert_sample_project(conn: &Connection) {
871 conn.execute(
872 "INSERT INTO project_databases (name) VALUES (?1)",
873 params!["mydb"],
874 )
875 .unwrap();
876 conn.execute(
877 "INSERT INTO project_schemas (database, name, schema_type) VALUES (?1, ?2, ?3)",
878 params!["mydb", "public", "user"],
879 )
880 .unwrap();
881 conn.execute(
882 "INSERT INTO project_objects (object_key, database, schema, name, object_kind, cluster, file_path, sql_text) \
883 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
884 params![
885 "mydb.public.orders",
886 "mydb",
887 "public",
888 "orders",
889 "materialized-view",
890 "compute",
891 "sql/orders.sql",
892 "CREATE MATERIALIZED VIEW orders AS SELECT 1",
893 ],
894 )
895 .unwrap();
896 conn.execute(
897 "INSERT INTO project_objects (object_key, database, schema, name, object_kind, cluster, file_path, sql_text) \
898 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
899 params![
900 "mydb.public.users",
901 "mydb",
902 "public",
903 "users",
904 "view",
905 None::<String>,
906 "sql/users.sql",
907 "CREATE VIEW users AS SELECT 1",
908 ],
909 )
910 .unwrap();
911 conn.execute(
912 "INSERT INTO project_comments (object_key, comment_type, target_column, comment_text, sql_text) \
913 VALUES (?1, ?2, ?3, ?4, ?5)",
914 params![
915 "mydb.public.orders",
916 "object",
917 None::<String>,
918 "Order data",
919 "COMMENT ON MATERIALIZED VIEW orders IS 'Order data'"
920 ],
921 )
922 .unwrap();
923 conn.execute(
924 "INSERT INTO project_indexes (object_key, index_name, cluster, columns, sql_text) \
925 VALUES (?1, ?2, ?3, ?4, ?5)",
926 params![
927 "mydb.public.orders",
928 "orders_id_idx",
929 "compute",
930 "id",
931 "CREATE INDEX orders_id_idx ON orders (id)"
932 ],
933 )
934 .unwrap();
935 conn.execute(
936 "INSERT INTO project_grants (object_key, privilege, grantee, sql_text) \
937 VALUES (?1, ?2, ?3, ?4)",
938 params![
939 "mydb.public.orders",
940 "SELECT",
941 "reader_role",
942 "GRANT SELECT ON orders TO reader_role"
943 ],
944 )
945 .unwrap();
946 conn.execute(
947 "INSERT INTO project_aliases (object_key, alias, target_fqn) VALUES (?1, ?2, ?3)",
948 params!["mydb.public.orders", "raw_orders", "ext.public.raw_orders"],
949 )
950 .unwrap();
951 conn.execute(
952 "INSERT INTO project_aliases (object_key, alias, target_fqn) VALUES (?1, ?2, ?3)",
953 params![
954 "mydb.public.orders",
955 "order_items",
956 "ext.public.order_items"
957 ],
958 )
959 .unwrap();
960 conn.execute(
961 "INSERT INTO project_infrastructure (object_key, infra_type, connector_type, connection_ref, source_ref, external_reference) \
962 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
963 params![
964 "mydb.public.orders",
965 "source",
966 "postgres",
967 "mydb.public.pg_conn",
968 None::<String>,
969 None::<String>
970 ],
971 )
972 .unwrap();
973 conn.execute(
974 "INSERT INTO project_infrastructure_properties (object_key, property_key, property_value, secret_ref, object_ref) \
975 VALUES (?1, ?2, ?3, ?4, ?5)",
976 params![
977 "mydb.public.orders",
978 "PUBLICATION",
979 "mz_source",
980 None::<String>,
981 None::<String>
982 ],
983 )
984 .unwrap();
985 conn.execute(
986 "INSERT INTO project_dependencies (object_key, dependency_key) VALUES (?1, ?2)",
987 params!["mydb.public.orders", "mydb.public.users"],
988 )
989 .unwrap();
990 conn.execute(
991 "INSERT INTO project_external_dependencies (object_key) VALUES (?1)",
992 params!["ext.public.raw_data"],
993 )
994 .unwrap();
995 conn.execute(
996 "INSERT INTO project_tests (object_key, test_name, sql_text) VALUES (?1, ?2, ?3)",
997 params![
998 "mydb.public.orders",
999 "test_orders_not_empty",
1000 "SELECT count(*) > 0 FROM orders"
1001 ],
1002 )
1003 .unwrap();
1004 conn.execute(
1005 "INSERT INTO project_mod_statements (database, schema, position, sql_text) \
1006 VALUES (?1, ?2, ?3, ?4)",
1007 params!["mydb", None::<String>, 0, "CREATE DATABASE mydb"],
1008 )
1009 .unwrap();
1010 conn.execute(
1011 "INSERT INTO project_mod_statements (database, schema, position, sql_text) \
1012 VALUES (?1, ?2, ?3, ?4)",
1013 params!["mydb", "public", 0, "CREATE SCHEMA public"],
1014 )
1015 .unwrap();
1016 }
1017
1018 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1020 fn test_get_object_full_metadata() {
1021 let dir = tempfile::tempdir().unwrap();
1022 let db_path = dir.path().join("test.db");
1023 let conn = create_test_db(&db_path);
1024 insert_sample_project(&conn);
1025 drop(conn);
1026
1027 let cache = open_cache(&db_path);
1028 let obj = cache
1029 .get_object(&"mydb.public.orders".parse::<ObjectId>().unwrap())
1030 .unwrap();
1031
1032 assert_eq!(obj.fqn, "mydb.public.orders");
1033 assert_eq!(obj.database, "mydb");
1034 assert_eq!(obj.schema, "public");
1035 assert_eq!(obj.name, "orders");
1036 assert_eq!(obj.kind, ObjectKind::MaterializedView);
1037 assert_eq!(obj.cluster.as_deref(), Some("compute"));
1038 assert_eq!(obj.file_path, "sql/orders.sql");
1039
1040 assert_eq!(obj.comments.len(), 1);
1041 assert_eq!(obj.comments[0].comment_type, "object");
1042 assert_eq!(obj.comments[0].text, "Order data");
1043
1044 assert_eq!(obj.indexes.len(), 1);
1045 assert_eq!(obj.indexes[0].name, "orders_id_idx");
1046
1047 assert_eq!(obj.grants.len(), 1);
1048 assert_eq!(obj.grants[0].privilege, "SELECT");
1049 assert_eq!(obj.grants[0].grantee, "reader_role");
1050
1051 assert_eq!(obj.aliases.len(), 2);
1052 assert_eq!(obj.aliases["order_items"], "ext.public.order_items");
1053 assert_eq!(obj.aliases["raw_orders"], "ext.public.raw_orders");
1054
1055 let infra = obj.infrastructure.unwrap();
1056 assert_eq!(infra.infra_type, "source");
1057 assert_eq!(infra.connector_type.as_deref(), Some("postgres"));
1058 assert_eq!(infra.connection_ref.as_deref(), Some("mydb.public.pg_conn"));
1059 assert_eq!(infra.properties.len(), 1);
1060 assert_eq!(infra.properties[0].key, "PUBLICATION");
1061 }
1062
1063 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1065 fn test_get_object_not_found() {
1066 let dir = tempfile::tempdir().unwrap();
1067 let db_path = dir.path().join("test.db");
1068 let conn = create_test_db(&db_path);
1069 insert_sample_project(&conn);
1070 drop(conn);
1071
1072 let cache = open_cache(&db_path);
1073 assert!(
1074 cache
1075 .get_object(&"nonexistent.x.y".parse::<ObjectId>().unwrap())
1076 .is_none()
1077 );
1078 }
1079
1080 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1082 fn test_get_object_by_path() {
1083 let dir = tempfile::tempdir().unwrap();
1084 let db_path = dir.path().join("test.db");
1085 let conn = create_test_db(&db_path);
1086 insert_sample_project(&conn);
1087 drop(conn);
1088
1089 let cache = open_cache(&db_path);
1090 let obj = cache.get_object_by_path("sql/orders.sql").unwrap();
1091 assert_eq!(obj.fqn, "mydb.public.orders");
1092 }
1093
1094 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1096 fn test_list_objects() {
1097 let dir = tempfile::tempdir().unwrap();
1098 let db_path = dir.path().join("test.db");
1099 let conn = create_test_db(&db_path);
1100 insert_sample_project(&conn);
1101 drop(conn);
1102
1103 let cache = open_cache(&db_path);
1104 let objects = cache.list_objects();
1105 assert_eq!(objects.len(), 2);
1106
1107 let fqns: Vec<&str> = objects.iter().map(|o| o.fqn.as_str()).collect();
1108 assert!(fqns.contains(&"mydb.public.orders"));
1109 assert!(fqns.contains(&"mydb.public.users"));
1110 }
1111
1112 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1114 fn test_list_databases() {
1115 let dir = tempfile::tempdir().unwrap();
1116 let db_path = dir.path().join("test.db");
1117 let conn = create_test_db(&db_path);
1118 insert_sample_project(&conn);
1119 drop(conn);
1120
1121 let cache = open_cache(&db_path);
1122 let databases = cache.list_databases_with_objects();
1123 assert_eq!(databases.len(), 1);
1124 assert_eq!(databases[0].name, "mydb");
1125 assert_eq!(databases[0].schemas.len(), 1);
1126 assert_eq!(databases[0].schemas[0].name, "public");
1127 assert_eq!(databases[0].schemas[0].schema_type, "user");
1128 assert_eq!(databases[0].schemas[0].objects.len(), 2);
1129
1130 let fqns: Vec<&str> = databases[0].schemas[0]
1131 .objects
1132 .iter()
1133 .map(|o| o.fqn.as_str())
1134 .collect();
1135 assert!(fqns.contains(&"mydb.public.orders"));
1136 assert!(fqns.contains(&"mydb.public.users"));
1137 }
1138
1139 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1141 fn test_get_dependencies_and_dependents() {
1142 let dir = tempfile::tempdir().unwrap();
1143 let db_path = dir.path().join("test.db");
1144 let conn = create_test_db(&db_path);
1145 insert_sample_project(&conn);
1146 drop(conn);
1147
1148 let cache = open_cache(&db_path);
1149
1150 let orders: ObjectId = "mydb.public.orders".parse().unwrap();
1151 let users: ObjectId = "mydb.public.users".parse().unwrap();
1152
1153 let deps = cache.get_dependencies(&orders);
1154 assert_eq!(deps, vec![users.clone()]);
1155
1156 let dependents = cache.get_dependents(&users);
1157 assert_eq!(dependents, vec![orders.clone()]);
1158
1159 assert!(cache.get_dependencies(&users).is_empty());
1160 }
1161
1162 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1164 fn test_list_external_dependencies() {
1165 let dir = tempfile::tempdir().unwrap();
1166 let db_path = dir.path().join("test.db");
1167 let conn = create_test_db(&db_path);
1168 insert_sample_project(&conn);
1169 drop(conn);
1170
1171 let cache = open_cache(&db_path);
1172 let ext = cache.list_external_dependencies();
1173 assert_eq!(
1174 ext,
1175 vec!["ext.public.raw_data".parse::<ObjectId>().unwrap()]
1176 );
1177 }
1178
1179 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1181 fn test_get_tests() {
1182 let dir = tempfile::tempdir().unwrap();
1183 let db_path = dir.path().join("test.db");
1184 let conn = create_test_db(&db_path);
1185 insert_sample_project(&conn);
1186 drop(conn);
1187
1188 let cache = open_cache(&db_path);
1189 let orders: ObjectId = "mydb.public.orders".parse().unwrap();
1190 let users: ObjectId = "mydb.public.users".parse().unwrap();
1191
1192 let tests = cache.get_tests(&orders);
1193 assert_eq!(tests.len(), 1);
1194 assert_eq!(tests[0].name, "test_orders_not_empty");
1195
1196 assert!(cache.get_tests(&users).is_empty());
1197 }
1198
1199 #[cfg_attr(miri, ignore)] #[mz_ore::test]
1201 fn test_get_mod_statements() {
1202 let dir = tempfile::tempdir().unwrap();
1203 let db_path = dir.path().join("test.db");
1204 let conn = create_test_db(&db_path);
1205 insert_sample_project(&conn);
1206 drop(conn);
1207
1208 let cache = open_cache(&db_path);
1209
1210 let db_mods = cache.get_mod_statements("mydb", None);
1211 assert_eq!(db_mods, vec!["CREATE DATABASE mydb"]);
1212
1213 let schema_mods = cache.get_mod_statements("mydb", Some("public"));
1214 assert_eq!(schema_mods, vec!["CREATE SCHEMA public"]);
1215
1216 assert!(cache.get_mod_statements("unknown", None).is_empty());
1217 }
1218}