1use super::relationship::{
4 Cardinality, MediaHandler, MediaHandlerMap, Relationship, RelationshipsMap,
5};
6use super::routine::{FuncVolatility, RetType, Routine, RoutineMap, RoutineParam};
7use super::table::{Column, ColumnMap, Table, TablesMap};
8use crate::api_request::QualifiedIdentifier;
9use crate::error::{Error, Result};
10use indexmap::IndexMap;
11use sqlx::{PgPool, Row};
12use std::collections::{HashMap, HashSet};
13use std::sync::LazyLock;
14
15pub async fn get_pg_version(pool: &PgPool) -> Result<i32> {
17 let row = sqlx::query("SELECT current_setting('server_version_num')::int as version")
18 .fetch_one(pool)
19 .await
20 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
21
22 Ok(row.get("version"))
23}
24
25fn acl_grants(acl: &str, privileges: &str) -> String {
41 format!(
42 "EXISTS (SELECT 1 FROM aclexplode({acl}) g \
43 WHERE g.privilege_type = ANY ('{{{privileges}}}'::text[]) \
44 AND {grantee})",
45 grantee = can_become("g.grantee"),
46 )
47}
48
49fn can_become(role: &str) -> String {
55 format!("CASE WHEN {role} = 0 THEN true ELSE pg_has_role({role}, 'MEMBER') END")
56}
57
58const PREDEFINED_DATA_ROLE: &str = "EXISTS (SELECT 1 FROM pg_roles pr \
66 WHERE pr.rolname IN ('pg_read_all_data', 'pg_write_all_data') \
67 AND pg_has_role(pr.oid, 'MEMBER'))";
68
69const RELATION_ACL: &str = "COALESCE(c.relacl, acldefault('r', c.relowner))";
71
72fn relation_visible() -> String {
77 format!(
78 "(pg_has_role(c.relowner, 'MEMBER') \
79 OR {PREDEFINED_DATA_ROLE} \
80 OR {table} \
81 OR EXISTS (SELECT 1 FROM pg_attribute ca \
82 WHERE ca.attrelid = c.oid AND ca.attnum > 0 \
83 AND NOT ca.attisdropped AND ca.attacl IS NOT NULL \
84 AND {column}))",
85 table = acl_grants(
86 RELATION_ACL,
87 "SELECT,INSERT,UPDATE,DELETE,TRUNCATE,REFERENCES,TRIGGER"
88 ),
89 column = acl_grants("ca.attacl", "SELECT,INSERT,UPDATE,REFERENCES"),
90 )
91}
92
93fn privilege_granted(privilege: &str) -> String {
98 format!(
99 "(c.relkind IN ('r', 'v', 'f', 'p') AND EXISTS (\
100 SELECT 1 FROM aclexplode({RELATION_ACL}) g \
101 WHERE g.privilege_type = '{privilege}' \
102 AND ({grantee} OR pg_has_role(g.grantor, 'MEMBER'))))",
103 grantee = can_become("g.grantee"),
104 )
105}
106
107pub async fn load_tables(pool: &PgPool, schemas: &[String]) -> Result<TablesMap> {
109 let mut tables = HashMap::new();
110
111 static SQL: LazyLock<String> = LazyLock::new(|| {
116 format!(
117 r#"
118 SELECT
119 t.table_schema,
120 t.table_name,
121 t.table_type,
122 pg_catalog.obj_description(t.oid, 'pg_class') as description,
123 COALESCE(
124 (SELECT array_agg(a.attname::text ORDER BY array_position(i.indkey, a.attnum))
125 FROM pg_index i
126 JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
127 WHERE i.indrelid = t.oid
128 AND i.indisprimary),
129 ARRAY[]::text[]
130 ) as pk_cols,
131 -- Being granted INSERT on a view does not make the view
132 -- insertable. A view is written through only where it is simple
133 -- enough for PostgreSQL to write through automatically, or where
134 -- an INSTEAD OF trigger says how -- `projects_view_without_triggers`
135 -- carries every grant and accepts nothing. Both halves have to
136 -- hold. A table is not a view, so the grant is the whole answer
137 -- there.
138 --
139 -- The view half is `information_schema.views`' own: the bits of
140 -- `pg_relation_is_updatable` it reads, and the INSTEAD OF trigger
141 -- types (`tgtype` 69 insert, 81 update, 73 delete).
142 {insert} AND (
143 t.relkind <> 'v'
144 OR (pg_relation_is_updatable(t.oid, false) & 8) = 8
145 OR EXISTS (SELECT 1 FROM pg_trigger tg
146 WHERE tg.tgrelid = t.oid AND (tg.tgtype::int & 69) = 69)
147 ) as insertable,
148 {update} AND (
149 t.relkind <> 'v'
150 OR (pg_relation_is_updatable(t.oid, false) & 20) = 20
151 OR EXISTS (SELECT 1 FROM pg_trigger tg
152 WHERE tg.tgrelid = t.oid AND (tg.tgtype::int & 81) = 81)
153 ) as updatable,
154 -- Updatable rather than a deletable bit, as `information_schema`
155 -- has it: a view PostgreSQL can write through automatically is
156 -- one it can delete through as well. Only the trigger half is
157 -- asked separately.
158 {delete} AND (
159 t.relkind <> 'v'
160 OR (pg_relation_is_updatable(t.oid, false) & 20) = 20
161 OR EXISTS (SELECT 1 FROM pg_trigger tg
162 WHERE tg.tgrelid = t.oid AND (tg.tgtype::int & 73) = 73)
163 ) as deletable,
164 t.relkind = 'p' as is_partitioned
165 FROM (
166 -- A foreign table is a table: it has columns, it can be selected
167 -- from and written to, and which server stores the rows is not
168 -- the API's business. Leaving it out meant a schema that exposes
169 -- one answered "no such table".
170 --
171 -- A materialized view is a view that cannot be written to, which
172 -- is what the privilege tests report for it -- they cover only
173 -- the relation kinds `information_schema.table_privileges` does,
174 -- and this is not one of them.
175 SELECT c.oid, c.relkind, c.relispartition,
176 n.nspname::text AS table_schema,
177 c.relname::text AS table_name,
178 CASE WHEN c.relkind IN ('v', 'm') THEN 'VIEW'
179 WHEN c.relkind = 'f' THEN 'FOREIGN'
180 ELSE 'BASE TABLE' END AS table_type,
181 {visible} OR c.relkind = 'm' AS visible
182 FROM pg_class c
183 JOIN pg_namespace n ON n.oid = c.relnamespace
184 WHERE c.relkind IN ('r', 'v', 'f', 'p', 'm')
185 AND NOT pg_is_other_temp_schema(n.oid)
186 AND n.oid <> pg_my_temp_schema()
187 ) t
188 JOIN pg_class c ON c.oid = t.oid
189 WHERE t.table_schema = ANY($1)
190 AND t.visible
191 -- A partition is reached through the table it partitions, not on its
192 -- own. Listing every one of them gave a schema with a year of
193 -- monthly partitions twelve endpoints PostgREST does not have -- and
194 -- a request naming one was answered as though the table were real,
195 -- which made a missing relationship read as a missing column. The
196 -- partitioned parent itself (`relkind = 'p'`) is not a partition and
197 -- stays.
198 AND NOT t.relispartition
199 ORDER BY t.table_schema, t.table_name
200 "#,
201 insert = privilege_granted("INSERT"),
202 update = privilege_granted("UPDATE"),
203 delete = privilege_granted("DELETE"),
204 visible = relation_visible(),
205 )
206 });
207 let rows = sqlx::query(&SQL)
208 .bind(schemas)
209 .fetch_all(pool)
210 .await
211 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
212
213 for row in rows {
214 let schema: String = row.get("table_schema");
215 let name: String = row.get("table_name");
216 let qi = QualifiedIdentifier::new(&schema, &name);
217
218 let table_type: String = row.get("table_type");
219 let pk_cols: Vec<String> = row.get("pk_cols");
220
221 let table = Table {
222 schema: schema.clone(),
223 name: name.clone(),
224 description: row.get("description"),
225 is_view: table_type == "VIEW",
226 is_partitioned: row.get("is_partitioned"),
227 insertable: row.get("insertable"),
228 updatable: row.get("updatable"),
229 deletable: row.get("deletable"),
230 pk_cols: pk_cols.clone(),
231 unique_constraints: Vec::new(),
232 columns: load_columns(pool, &schema, &name, &pk_cols).await?,
233 computed_columns: Default::default(),
234 };
235
236 tables.insert(qi, table);
237 }
238
239 Ok(tables)
240}
241
242async fn load_columns(
244 pool: &PgPool,
245 schema: &str,
246 table: &str,
247 pk_cols: &[String],
248) -> Result<ColumnMap> {
249 let mut columns = IndexMap::new();
250
251 static SQL: LazyLock<String> = LazyLock::new(|| {
257 format!(
258 r#"
259 SELECT
260 c.column_name,
261 c.ordinal_position,
262 c.is_nullable,
263 c.data_type,
264 c.udt_name,
265 -- A domain's own name. `information_schema` reports the type
266 -- underneath it in `data_type` and `udt_name`, so this is the only
267 -- column that says a value is a `color` rather than an integer --
268 -- and the domain is what a data representation is declared on.
269 c.domain_name,
270 c.character_maximum_length,
271 -- An identity column has no `column_default`: `information_schema`
272 -- describes it as generated rather than defaulted. What it would
273 -- take if a row left it out is the next value of the sequence the
274 -- column owns, which is the answer `Prefer: missing=default`
275 -- needs.
276 COALESCE(
277 c.column_default,
278 (SELECT format('nextval(%L)', s.oid::regclass)
279 FROM pg_class rel
280 JOIN pg_namespace ns ON ns.oid = rel.relnamespace
281 JOIN pg_attribute att ON att.attrelid = rel.oid
282 AND att.attname = c.column_name
283 AND att.attidentity = 'd'
284 JOIN pg_depend dep ON dep.refobjid = rel.oid
285 AND dep.refobjsubid = att.attnum
286 AND dep.deptype = 'i'
287 JOIN pg_class s ON s.oid = dep.objid AND s.relkind = 'S'
288 WHERE ns.nspname = c.table_schema
289 AND rel.relname = c.table_name
290 LIMIT 1)
291 ) AS column_default,
292 -- What a write may not name at all. `GENERATED ALWAYS AS
293 -- IDENTITY` and a stored generated column both belong to
294 -- PostgreSQL, which answers `cannot insert a non-DEFAULT value`
295 -- to anything that names one. `BY DEFAULT` is deliberately not
296 -- here: it has a default, and a value given for it is taken.
297 COALESCE(
298 (c.is_identity = 'YES' AND c.identity_generation = 'ALWAYS')
299 OR c.is_generated = 'ALWAYS',
300 false
301 ) AS always_generated,
302 -- By OID: naming the table and casting to `regclass` needs USAGE
303 -- on its schema, which a NOINHERIT authenticator does not have.
304 pg_catalog.col_description(c.relid, c.ordinal_position) as description,
305 CASE WHEN e.enumtypid IS NOT NULL
306 THEN array_agg(e.enumlabel ORDER BY e.enumsortorder)
307 ELSE ARRAY[]::text[]
308 END as enum_values
309 FROM (
310 SELECT
311 c.oid AS relid,
312 nc.nspname::text AS table_schema,
313 c.relname::text AS table_name,
314 a.attname::text AS column_name,
315 a.attnum::int AS ordinal_position,
316 CASE WHEN a.attgenerated = '' THEN pg_get_expr(ad.adbin, ad.adrelid) END
317 AS column_default,
318 CASE WHEN a.attnotnull OR (t.typtype = 'd' AND t.typnotnull)
319 THEN 'NO' ELSE 'YES' END AS is_nullable,
320 CASE WHEN t.typtype = 'd' THEN
321 CASE WHEN bt.typelem <> 0 AND bt.typlen = -1 THEN 'ARRAY'
322 WHEN nbt.nspname = 'pg_catalog' THEN format_type(t.typbasetype, NULL)
323 ELSE 'USER-DEFINED' END
324 ELSE
325 CASE WHEN t.typelem <> 0 AND t.typlen = -1 THEN 'ARRAY'
326 WHEN nt.nspname = 'pg_catalog' THEN format_type(a.atttypid, NULL)
327 ELSE 'USER-DEFINED' END
328 END AS data_type,
329 information_schema._pg_char_max_length(
330 information_schema._pg_truetypid(a.*, t.*),
331 information_schema._pg_truetypmod(a.*, t.*)
332 )::int AS character_maximum_length,
333 CASE WHEN t.typtype = 'd' THEN t.typname::text END AS domain_name,
334 COALESCE(bt.typname, t.typname)::text AS udt_name,
335 -- The same type by OID, for its enum labels: two schemas may
336 -- each have an enum of one name.
337 COALESCE(bt.oid, t.oid) AS udt_oid,
338 CASE WHEN a.attidentity IN ('a', 'd') THEN 'YES' ELSE 'NO' END AS is_identity,
339 CASE a.attidentity WHEN 'a' THEN 'ALWAYS' WHEN 'd' THEN 'BY DEFAULT' END
340 AS identity_generation,
341 CASE WHEN a.attgenerated <> '' THEN 'ALWAYS' ELSE 'NEVER' END AS is_generated
342 FROM pg_attribute a
343 LEFT JOIN pg_attrdef ad ON a.attrelid = ad.adrelid AND a.attnum = ad.adnum
344 JOIN pg_class c ON a.attrelid = c.oid
345 JOIN pg_namespace nc ON c.relnamespace = nc.oid
346 JOIN pg_type t ON a.atttypid = t.oid
347 JOIN pg_namespace nt ON t.typnamespace = nt.oid
348 LEFT JOIN (pg_type bt JOIN pg_namespace nbt ON bt.typnamespace = nbt.oid)
349 ON t.typtype = 'd' AND t.typbasetype = bt.oid
350 WHERE nc.nspname = $1 AND c.relname = $2
351 AND NOT pg_is_other_temp_schema(nc.oid)
352 AND a.attnum > 0 AND NOT a.attisdropped
353 AND c.relkind IN ('r', 'v', 'f', 'p')
354 AND (pg_has_role(c.relowner, 'MEMBER')
355 OR {predefined}
356 OR {table_grant}
357 OR (a.attacl IS NOT NULL AND {column_grant}))
358 ) c
359 LEFT JOIN pg_enum e ON e.enumtypid = c.udt_oid
360 WHERE c.table_schema = $1 AND c.table_name = $2
361 GROUP BY c.relid, c.table_schema, c.table_name, c.column_name, c.ordinal_position, c.is_nullable,
362 c.data_type, c.udt_name, c.domain_name, c.character_maximum_length,
363 c.column_default, c.is_identity, c.identity_generation, c.is_generated,
364 c.udt_oid, e.enumtypid
365 ORDER BY c.ordinal_position
366 "#,
367 predefined = PREDEFINED_DATA_ROLE,
368 table_grant = acl_grants(RELATION_ACL, "SELECT,INSERT,UPDATE,REFERENCES"),
369 column_grant = acl_grants("a.attacl", "SELECT,INSERT,UPDATE,REFERENCES"),
370 )
371 });
372 let rows = sqlx::query(&SQL)
373 .bind(schema)
374 .bind(table)
375 .fetch_all(pool)
376 .await
377 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
378
379 let rows = match rows.is_empty() {
382 true => load_catalog_columns(pool, schema, table).await?,
383 false => rows,
384 };
385
386 for row in rows {
387 let name: String = row.get("column_name");
388 let is_nullable: String = row.get("is_nullable");
389 let data_type: String = row.get("data_type");
390 let udt_name: String = row.get("udt_name");
391 let max_len: Option<i32> = row.get("character_maximum_length");
392 let enum_values: Vec<String> = row.get("enum_values");
393 let position: i32 = row.get("ordinal_position");
394
395 let column = Column {
396 name: name.clone(),
397 description: row.get("description"),
398 nullable: is_nullable == "YES",
399 data_type,
400 nominal_type: udt_name,
401 domain_type: row.get("domain_name"),
402 max_len,
403 default: row.get("column_default"),
404 enum_values,
405 is_pk: pk_cols.contains(&name),
406 position,
407 always_generated: row.get("always_generated"),
408 };
409
410 columns.insert(name, column);
411 }
412
413 Ok(columns)
414}
415
416pub async fn load_relationships(pool: &PgPool, schemas: &[String]) -> Result<RelationshipsMap> {
418 let mut relationships: RelationshipsMap = HashMap::new();
419
420 let rows = sqlx::query(
421 r#"
422 SELECT
423 c.conname as constraint_name,
424 ns1.nspname as table_schema,
425 t1.relname as table_name,
426 ns2.nspname as foreign_table_schema,
427 t2.relname as foreign_table_name,
428 -- Each key's columns are read in their own subquery, by ordinal
429 -- position. Joining pg_attribute twice in the outer query instead
430 -- would multiply the two keys together: a two-column foreign key
431 -- came back as four pairs, of which two paired a column with the
432 -- wrong one on the other side.
433 (SELECT array_agg(a.attname ORDER BY k.ord)
434 FROM unnest(c.conkey) WITH ORDINALITY AS k(attnum, ord)
435 JOIN pg_attribute a
436 ON a.attrelid = c.conrelid AND a.attnum = k.attnum) as columns,
437 (SELECT array_agg(a.attname ORDER BY k.ord)
438 FROM unnest(c.confkey) WITH ORDINALITY AS k(attnum, ord)
439 JOIN pg_attribute a
440 ON a.attrelid = c.confrelid AND a.attnum = k.attnum) as foreign_columns,
441 t1.relkind = 'v' as table_is_view,
442 t2.relkind = 'v' as foreign_table_is_view,
443 -- The key is unique when a unique index is *covered by* it, not
444 -- when the index merely contains it. Asked the other way round,
445 -- any key that is part of a wider unique index looked unique --
446 -- so a foreign key on two of a three-column primary key became a
447 -- one-to-one, and the far side embedded a single object where it
448 -- should have embedded an array.
449 EXISTS (
450 SELECT 1 FROM pg_index i
451 WHERE i.indrelid = c.conrelid
452 AND i.indisunique
453 AND c.conkey::int[] @> i.indkey::int[]
454 ) as is_unique
455 FROM pg_constraint c
456 JOIN pg_class t1 ON t1.oid = c.conrelid
457 JOIN pg_namespace ns1 ON ns1.oid = t1.relnamespace
458 JOIN pg_class t2 ON t2.oid = c.confrelid
459 JOIN pg_namespace ns2 ON ns2.oid = t2.relnamespace
460 WHERE c.contype = 'f'
461 AND ns1.nspname = ANY($1)
462 -- Ordered, because the order reaches the client. Where two foreign
463 -- keys join the same pair of tables the embed is ambiguous, and the
464 -- PGRST201 that says so lists the candidates -- in this order, both in
465 -- `details` and in the `hint` that tells the caller what to write
466 -- instead. `pg_constraint` unordered is heap order, which differs
467 -- between two databases holding the same schema and can differ between
468 -- two reads of one database. That made the error message arbitrary and
469 -- the conformance cases covering it flip between runs.
470 ORDER BY ns1.nspname, t1.relname, c.conname
471 "#,
472 )
473 .bind(schemas)
474 .fetch_all(pool)
475 .await
476 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
477
478 for row in rows {
479 let table_schema: String = row.get("table_schema");
480 let table_name: String = row.get("table_name");
481 let foreign_schema: String = row.get("foreign_table_schema");
482 let foreign_name: String = row.get("foreign_table_name");
483 let constraint_name: String = row.get("constraint_name");
484 let columns: Vec<String> = row.get("columns");
485 let foreign_columns: Vec<String> = row.get("foreign_columns");
486 let table_is_view: bool = row.get("table_is_view");
487 let foreign_is_view: bool = row.get("foreign_table_is_view");
488 let is_unique: bool = row.get("is_unique");
489
490 let table_qi = QualifiedIdentifier::new(&table_schema, &table_name);
491 let foreign_qi = QualifiedIdentifier::new(&foreign_schema, &foreign_name);
492
493 let column_pairs: Vec<(String, String)> =
494 columns.into_iter().zip(foreign_columns).collect();
495
496 let is_self = table_qi == foreign_qi;
497
498 let cardinality = if is_unique {
500 Cardinality::O2O {
501 constraint: constraint_name.clone(),
502 columns: column_pairs.clone(),
503 is_parent: false,
504 }
505 } else {
506 Cardinality::M2O {
507 constraint: constraint_name.clone(),
508 columns: column_pairs.clone(),
509 }
510 };
511
512 let rel = Relationship::ForeignKey {
513 table: table_qi.clone(),
514 foreign_table: foreign_qi.clone(),
515 is_self,
516 cardinality,
517 table_is_view,
518 foreign_table_is_view: foreign_is_view,
519 constraint_name: constraint_name.clone(),
520 };
521
522 relationships
523 .entry((table_qi.clone(), table_schema.clone()))
524 .or_default()
525 .push(rel);
526
527 let reverse_columns: Vec<(String, String)> = column_pairs
529 .iter()
530 .map(|(a, b)| (b.clone(), a.clone()))
531 .collect();
532
533 let reverse_cardinality = if is_unique {
534 Cardinality::O2O {
535 constraint: constraint_name.clone(),
536 columns: reverse_columns.clone(),
537 is_parent: true,
538 }
539 } else {
540 Cardinality::O2M {
545 constraint: constraint_name.clone(),
546 columns: reverse_columns,
547 }
548 };
549
550 let reverse_rel = Relationship::ForeignKey {
551 table: foreign_qi.clone(),
552 foreign_table: table_qi,
553 is_self,
554 cardinality: reverse_cardinality,
555 table_is_view: foreign_is_view,
556 foreign_table_is_view: table_is_view,
557 constraint_name,
558 };
559
560 relationships
561 .entry((foreign_qi, foreign_schema))
562 .or_default()
563 .push(reverse_rel);
564 }
565
566 load_computed_relationships(pool, schemas, &mut relationships).await?;
567
568 Ok(relationships)
569}
570
571async fn load_computed_relationships(
583 pool: &PgPool,
584 schemas: &[String],
585 relationships: &mut RelationshipsMap,
586) -> Result<()> {
587 let rows = sqlx::query(
588 r#"
589 SELECT
590 pn.nspname AS function_schema,
591 p.proname AS function_name,
592 tn.nspname AS table_schema,
593 t.relname AS table_name,
594 fn.nspname AS foreign_table_schema,
595 f.relname AS foreign_table_name,
596 (NOT p.proretset) OR p.prorows = 1 AS to_one,
597 (SELECT p.proargnames[i]
598 FROM generate_series(1, p.pronargs) AS i
599 WHERE p.proargtypes[i - 1] = argt.oid
600 LIMIT 1) AS row_argument
601 FROM pg_proc p
602 JOIN pg_namespace pn ON pn.oid = p.pronamespace
603 -- one argument is a table's composite type. Usually the only one, and
604 -- then the call is positional; a function that also takes a search
605 -- term or the caller's session has it somewhere among the rest, and is
606 -- called by name.
607 JOIN pg_type argt ON argt.oid = ANY (p.proargtypes::oid[])
608 JOIN pg_class t ON t.oid = argt.typrelid
609 JOIN pg_namespace tn ON tn.oid = t.relnamespace
610 -- and the return type is a table as well
611 JOIN pg_type rett ON rett.oid = p.prorettype
612 JOIN pg_class f ON f.oid = rett.typrelid
613 JOIN pg_namespace fn ON fn.oid = f.relnamespace
614 -- Every argument named, since one that is not cannot be given by name
615 -- and a call with more than the row has to be.
616 WHERE (p.pronargs = 1 OR (SELECT count(*)
617 FROM generate_series(1, p.pronargs) AS i
618 WHERE COALESCE(p.proargnames[i], '') = '') = 0)
619 AND pn.nspname = ANY($1)
620 AND tn.nspname = ANY($1)
621 AND fn.nspname = ANY($1)
622 -- relkind 'c' would be a standalone composite type, not a relation
623 AND t.relkind = ANY (ARRAY['r','v','m','f','p'])
624 AND f.relkind = ANY (ARRAY['r','v','m','f','p'])
625 "#,
626 )
627 .bind(schemas)
628 .fetch_all(pool)
629 .await
630 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
631
632 for row in rows {
633 let function = QualifiedIdentifier::new(
634 row.get::<String, _>("function_schema"),
635 row.get::<String, _>("function_name"),
636 );
637 let table_schema: String = row.get("table_schema");
638 let table = QualifiedIdentifier::new(&table_schema, row.get::<String, _>("table_name"));
639 let foreign_table = QualifiedIdentifier::new(
640 row.get::<String, _>("foreign_table_schema"),
641 row.get::<String, _>("foreign_table_name"),
642 );
643 let is_self = table == foreign_table;
644
645 relationships
646 .entry((table.clone(), table_schema))
647 .or_default()
648 .push(Relationship::Computed {
649 function,
650 table: table.clone(),
651 foreign_table: foreign_table.clone(),
652 table_alias: table,
653 to_one: row.get("to_one"),
654 is_self,
655 row_argument: row.get::<Option<String>, _>("row_argument"),
656 });
657 }
658
659 Ok(())
660}
661
662fn parse_routine_params(value: serde_json::Value) -> Vec<RoutineParam> {
668 let Some(items) = value.as_array() else {
669 return Vec::new();
670 };
671
672 items
673 .iter()
674 .filter_map(|item| {
675 let param_type = item.get("type")?.as_str()?.to_string();
676 Some(RoutineParam {
677 name: item
678 .get("name")
679 .and_then(|n| n.as_str())
680 .unwrap_or_default()
681 .to_string(),
682 type_max_length: item
688 .get("cast_type")
689 .and_then(|t| t.as_str())
690 .unwrap_or(¶m_type)
691 .to_string(),
692 param_type,
693 required: item
694 .get("required")
695 .and_then(|r| r.as_bool())
696 .unwrap_or(true),
697 variadic: item
698 .get("variadic")
699 .and_then(|v| v.as_bool())
700 .unwrap_or(false),
701 })
702 })
703 .collect()
704}
705
706pub async fn load_routines(pool: &PgPool, schemas: &[String]) -> Result<RoutineMap> {
708 let mut routines: RoutineMap = HashMap::new();
709
710 let rows = sqlx::query(
711 r#"
712 -- What a type is once every domain over it is stripped away.
713 --
714 -- `RETURNS SETOF projects_domain` returns rows of `projects`, and a
715 -- domain is transparent to everything that matters here: whether the
716 -- result is composite, and which table's columns it has. Reading the
717 -- declared type alone finds a type name that names no table.
718 WITH RECURSIVE base_types AS (
719 SELECT oid, typbasetype, oid AS base_type FROM pg_catalog.pg_type
720 UNION
721 SELECT t.oid, b.typbasetype, b.oid AS base_type
722 FROM base_types t
723 JOIN pg_catalog.pg_type b ON b.oid = t.typbasetype
724 ),
725 resolved AS (
726 SELECT oid, base_type FROM base_types WHERE typbasetype = 0
727 )
728 SELECT
729 n.nspname as schema,
730 p.proname as name,
731 pg_catalog.obj_description(p.oid) as description,
732 p.provolatile::text as volatility,
733 p.provariadic <> 0 as has_variadic,
734 EXISTS (
735 SELECT 1 FROM unnest(COALESCE(p.proargmodes, '{}'::"char"[])) AS mode
736 WHERE mode IN ('o', 't', 'b')
737 ) as has_out_params,
738 p.prokind = 'p' as is_procedure,
739 pg_get_function_identity_arguments(p.oid) as args,
740 -- Input parameters, in declaration order. Built from the catalog
741 -- rather than parsed out of the identity-arguments string, whose
742 -- types can themselves contain commas (`numeric(10,2)`).
743 -- `proargtypes` covers the IN parameters only and is 0-indexed;
744 -- `proargnames` is 1-indexed and lists IN names first, so the two
745 -- line up over 1..pronargs. A parameter is optional when it falls
746 -- inside the trailing run that has defaults.
747 COALESCE((
748 SELECT json_agg(
749 json_build_object(
750 'name', COALESCE(p.proargnames[i], ''),
751 'type', pg_catalog.format_type(p.proargtypes[i - 1], NULL),
752 -- What an argument is cast to. `character` and `bit`
753 -- name themselves without a length, and casting to
754 -- either truncates to one character; a parameter's
755 -- length is not enforced anyway, so the varying
756 -- spelling is what the value should reach.
757 'cast_type', CASE p.proargtypes[i - 1]
758 WHEN 'character'::regtype THEN 'character varying'
759 WHEN 'character[]'::regtype THEN 'character varying[]'
760 WHEN 'bit'::regtype THEN 'bit varying'
761 WHEN 'bit[]'::regtype THEN 'bit varying[]'
762 ELSE pg_catalog.format_type(p.proargtypes[i - 1], NULL)
763 END,
764 'required', i <= (p.pronargs - p.pronargdefaults),
765 'variadic', p.provariadic <> 0 AND i = p.pronargs
766 ) ORDER BY i
767 )
768 FROM generate_series(1, p.pronargs) AS i
769 ), '[]'::json) as params,
770 -- The columns the function's rows have, where it names them:
771 -- `RETURNS TABLE(a text, b colour)` declares them as OUT
772 -- parameters, and they are the only account of the result's shape
773 -- -- there is no table to read it off. `proallargtypes` is a
774 -- plain 1-indexed array and lines up with `proargnames`.
775 COALESCE((
776 SELECT json_agg(
777 json_build_object(
778 'name', COALESCE(p.proargnames[i], ''),
779 'type', pg_catalog.format_type(p.proallargtypes[i], NULL)
780 ) ORDER BY i
781 )
782 FROM generate_series(
783 1, COALESCE(array_length(p.proallargtypes, 1), 0)) AS i
784 WHERE p.proargmodes[i] IN ('o', 't', 'b')
785 ), '[]'::json) as out_params,
786 CASE
787 WHEN p.proretset THEN 'SETOF ' || pg_catalog.format_type(rt.base_type, NULL)
788 ELSE pg_catalog.format_type(rt.base_type, NULL)
789 END as return_type,
790 p.proretset as returns_set,
791 (SELECT t.typtype::text FROM pg_catalog.pg_type t WHERE t.oid = rt.base_type)
792 as ret_typtype,
793 -- A domain named after a media type says the value already is
794 -- that media type: `returns "image/png"` is a PNG, not a row to
795 -- serialise. Only a single value can be one, so a set-returning
796 -- function is not it.
797 (SELECT lower(dt.typname) FROM pg_catalog.pg_type dt
798 WHERE dt.oid = p.prorettype
799 AND dt.typbasetype <> 0
800 AND NOT p.proretset
801 AND (dt.typname ~ '^[A-Za-z0-9.-]+/[A-Za-z0-9.+-]+$'
802 OR dt.typname = '*/*')) as media_type,
803 -- What that domain is a domain *over*. A value whose type this
804 -- side does not recognise arrives as PostgreSQL's text rendering
805 -- of it, and a `bytea` renders as `\x` and hex -- which is a
806 -- description of the bytes, not the bytes. Knowing the base type
807 -- is what tells the two apart.
808 (SELECT lower(bt.typname) FROM pg_catalog.pg_type dt
809 JOIN pg_catalog.pg_type bt ON bt.oid = dt.typbasetype
810 WHERE dt.oid = p.prorettype
811 AND NOT p.proretset
812 AND (dt.typname ~ '^[A-Za-z0-9.-]+/[A-Za-z0-9.+-]+$'
813 OR dt.typname = '*/*')) as media_base_type
814 FROM pg_proc p
815 JOIN pg_namespace n ON n.oid = p.pronamespace
816 JOIN resolved rt ON rt.oid = p.prorettype
817 WHERE n.nspname = ANY($1)
818 AND p.prokind IN ('f', 'p')
819 -- A parameter with no name cannot be given an argument by name, so
820 -- a function that has one is not callable over the API at all --
821 -- and leaving it in the cache makes it a candidate that no request
822 -- can ever match, and a hint that sends the client nowhere.
823 --
824 -- The exception is the single unnamed parameter that takes the
825 -- whole request body, which is how a function is called with one
826 -- json, text, xml or bytea argument and no name for it.
827 AND (
828 (SELECT count(*) FROM generate_series(1, p.pronargs) AS i
829 WHERE COALESCE(p.proargnames[i], '') = '') = 0
830 OR (
831 p.pronargs = 1
832 AND COALESCE(p.proargnames[1], '') = ''
833 AND p.proargtypes[0] IN ('bytea'::regtype, 'json'::regtype,
834 'jsonb'::regtype, 'text'::regtype,
835 'xml'::regtype)
836 )
837 )
838 ORDER BY n.nspname, p.proname
839 "#,
840 )
841 .bind(schemas)
842 .fetch_all(pool)
843 .await
844 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
845
846 for row in rows {
847 let schema: String = row.get("schema");
848 let name: String = row.get("name");
849 let qi = QualifiedIdentifier::new(&schema, &name);
850
851 let volatility: String = row.get("volatility");
852 let return_type_str: String = row.get("return_type");
853 let returns_set: bool = row.get("returns_set");
854
855 let return_type = if return_type_str == "void" {
856 RetType::Void
857 } else if returns_set {
858 RetType::SetOf(return_type_str.replace("SETOF ", ""))
859 } else {
860 RetType::Single(return_type_str)
861 };
862
863 let ret_typtype: Option<String> = row.get("ret_typtype");
869 let has_out_params: bool = row.get("has_out_params");
874 let returns_composite = !matches!(return_type, RetType::Void)
875 && (ret_typtype.as_deref() == Some("c")
876 || (ret_typtype.as_deref() == Some("p") && has_out_params));
877
878 let routine = Routine {
879 schema,
880 name,
881 description: row.get("description"),
882 params: parse_routine_params(row.get("params")),
883 return_type,
884 returns_composite,
885 volatility: FuncVolatility::from_char(volatility.chars().next().unwrap_or('v')),
886 has_variadic: row.get("has_variadic"),
887 isolation_level: None,
888 settings: vec![],
889 is_procedure: row.get("is_procedure"),
890 media_type: row.get("media_type"),
891 media_base_type: row.get("media_base_type"),
892 output_columns: parse_routine_params(row.get("out_params"))
893 .into_iter()
894 .filter(|param| !param.name.is_empty())
895 .map(|param| (param.name, param.param_type))
896 .collect(),
897 };
898
899 routines.entry(qi).or_default().push(routine);
900 }
901
902 for overloads in routines.values_mut() {
908 overloads.sort_by(|a, b| {
909 let key = |r: &Routine| {
910 (
911 r.params.len(),
912 r.params
913 .iter()
914 .map(|p| (p.name.clone(), p.param_type.clone()))
915 .collect::<Vec<_>>(),
916 )
917 };
918 key(a).cmp(&key(b))
919 });
920 }
921
922 Ok(routines)
923}
924
925pub async fn load_timezones(pool: &PgPool) -> Result<HashSet<String>> {
927 let rows = sqlx::query("SELECT name FROM pg_timezone_names")
928 .fetch_all(pool)
929 .await
930 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
931
932 Ok(rows.iter().map(|r| r.get("name")).collect())
933}
934
935pub async fn load_media_handlers(pool: &PgPool, schemas: &[String]) -> Result<MediaHandlerMap> {
945 let rows = sqlx::query(
946 r#"
947 SELECT
948 n.nspname AS agg_schema,
949 p.proname AS agg_name,
950 t.typname AS media_type,
951 bt.typname AS media_base_type,
952 argn.nspname AS arg_schema,
953 argc.relname AS arg_table
954 FROM pg_aggregate a
955 JOIN pg_proc p ON p.oid = a.aggfnoid
956 JOIN pg_namespace n ON n.oid = p.pronamespace
957 JOIN pg_type t ON t.oid = a.aggtranstype
958 JOIN pg_type bt ON bt.oid = t.typbasetype
959 LEFT JOIN pg_type argt ON argt.oid = p.proargtypes[0]
960 LEFT JOIN pg_class argc ON argc.oid = argt.typrelid
961 LEFT JOIN pg_namespace argn ON argn.oid = argc.relnamespace
962 WHERE n.nspname = ANY($1)
963 AND t.typtype = 'd'
964 AND t.typname LIKE '%/%'
965 "#,
966 )
967 .bind(schemas)
968 .fetch_all(pool)
969 .await
970 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
971
972 let mut handlers: MediaHandlerMap = HashMap::new();
973 for row in rows {
974 let agg_schema: String = row.get("agg_schema");
975 let media_type: String = row.get("media_type");
976 let table = match (
977 row.get::<Option<String>, _>("arg_schema"),
978 row.get::<Option<String>, _>("arg_table"),
979 ) {
980 (Some(schema), Some(name)) => Some(QualifiedIdentifier::new(schema, name)),
981 _ => None,
982 };
983
984 handlers
985 .entry((agg_schema.clone(), media_type))
986 .or_default()
987 .push(MediaHandler {
988 aggregate: QualifiedIdentifier::new(agg_schema, row.get::<String, _>("agg_name")),
989 table,
990 base_type: row.get("media_base_type"),
991 });
992 }
993
994 Ok(handlers)
995}
996
997async fn load_catalog_columns(
1013 pool: &PgPool,
1014 schema: &str,
1015 table: &str,
1016) -> Result<Vec<sqlx::postgres::PgRow>> {
1017 sqlx::query(
1018 r#"
1019 SELECT
1020 a.attname AS column_name,
1021 a.attnum::int AS ordinal_position,
1022 CASE WHEN a.attnotnull THEN 'NO' ELSE 'YES' END AS is_nullable,
1023 pg_catalog.format_type(a.atttypid, NULL) AS data_type,
1024 t.typname AS udt_name,
1025 CASE WHEN t.typtype = 'd' THEN t.typname END AS domain_name,
1026 NULL::int AS character_maximum_length,
1027 pg_catalog.pg_get_expr(d.adbin, d.adrelid) AS column_default,
1028 (a.attidentity = 'a' OR a.attgenerated <> '') AS always_generated,
1029 pg_catalog.col_description(c.oid, a.attnum) AS description,
1030 COALESCE(
1031 (SELECT array_agg(e.enumlabel ORDER BY e.enumsortorder)
1032 FROM pg_enum e WHERE e.enumtypid = t.oid),
1033 ARRAY[]::text[]
1034 ) AS enum_values
1035 FROM pg_class c
1036 JOIN pg_namespace n ON n.oid = c.relnamespace
1037 JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped
1038 JOIN pg_type t ON t.oid = a.atttypid
1039 LEFT JOIN pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum
1040 WHERE n.nspname = $1 AND c.relname = $2
1041 ORDER BY a.attnum
1042 "#,
1043 )
1044 .bind(schema)
1045 .bind(table)
1046 .fetch_all(pool)
1047 .await
1048 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))
1049}
1050
1051pub async fn load_representations(
1063 pool: &PgPool,
1064) -> Result<std::collections::HashMap<(String, String), String>> {
1065 static SQL: LazyLock<String> = LazyLock::new(|| {
1071 format!(
1072 r#"
1073 SELECT c.castsource::regtype::text AS source,
1074 c.casttarget::regtype::text AS target,
1075 c.castfunc::regproc::text AS function
1076 FROM pg_catalog.pg_cast c
1077 JOIN pg_catalog.pg_type src ON src.oid = c.castsource
1078 JOIN pg_catalog.pg_type dst ON dst.oid = c.casttarget
1079 JOIN pg_catalog.pg_proc p ON p.oid = c.castfunc
1080 WHERE c.castcontext = 'i'
1081 AND c.castmethod = 'f'
1082 AND (pg_has_role(p.proowner, 'MEMBER') OR {execute})
1083 AND ((src.typtype = 'd' AND c.casttarget IN ('json'::regtype, 'text'::regtype))
1084 OR (dst.typtype = 'd' AND c.castsource IN ('json'::regtype, 'text'::regtype)))
1085 "#,
1086 execute = acl_grants("COALESCE(p.proacl, acldefault('f', p.proowner))", "EXECUTE"),
1087 )
1088 });
1089 let rows = sqlx::query(&SQL)
1090 .fetch_all(pool)
1091 .await
1092 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
1093
1094 Ok(rows
1095 .into_iter()
1096 .map(|row| {
1097 (
1098 (
1099 row.get::<String, _>("source"),
1100 row.get::<String, _>("target"),
1101 ),
1102 row.get::<String, _>("function"),
1103 )
1104 })
1105 .collect())
1106}
1107
1108pub async fn load_computed_columns(
1124 pool: &PgPool,
1125 schemas: &[String],
1126 function_schemas: &[String],
1127) -> Result<Vec<ComputedColumnRow>> {
1128 let rows = sqlx::query(
1136 r#"
1137 SELECT tn.nspname AS table_schema,
1138 t.relname AS table_name,
1139 fn.nspname AS function_schema,
1140 p.proname AS function_name,
1141 pg_catalog.format_type(p.prorettype, null) AS return_type,
1142 pg_catalog.obj_description(p.oid, 'pg_proc') AS description,
1143 (SELECT p.proargnames[i]
1144 FROM generate_series(1, p.pronargs) AS i
1145 WHERE p.proargtypes[i - 1] = t.reltype
1146 LIMIT 1) AS row_argument,
1147 (SELECT p.proargnames[i]
1148 FROM generate_series(1, p.pronargs) AS i
1149 WHERE p.proargnames[i] = 'hasura_session'
1150 LIMIT 1) AS session_argument,
1151 p.pronargs > 1 AS takes_arguments
1152 FROM pg_proc p
1153 JOIN pg_namespace fn ON fn.oid = p.pronamespace
1154 JOIN pg_class t ON t.reltype = ANY (p.proargtypes::oid[])
1155 JOIN pg_namespace tn ON tn.oid = t.relnamespace
1156 WHERE (p.pronargs = 1 OR (SELECT count(*)
1157 FROM generate_series(1, p.pronargs) AS i
1158 WHERE COALESCE(p.proargnames[i], '') = '') = 0)
1159 AND NOT p.proretset
1160 AND p.prokind = 'f'
1161 AND p.prorettype <> 'pg_catalog.trigger'::pg_catalog.regtype
1162 AND t.relkind = ANY (ARRAY['r', 'v', 'm', 'p', 'f'])
1163 AND fn.nspname = ANY($2)
1164 AND tn.nspname = ANY($1)
1165 "#,
1166 )
1167 .bind(schemas)
1168 .bind(function_schemas)
1169 .fetch_all(pool)
1170 .await
1171 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
1172
1173 Ok(rows
1174 .into_iter()
1175 .map(|row| ComputedColumnRow {
1176 table: QualifiedIdentifier::new(
1177 row.get::<String, _>("table_schema"),
1178 row.get::<String, _>("table_name"),
1179 ),
1180 name: row.get::<String, _>("function_name"),
1181 function: QualifiedIdentifier::new(
1182 row.get::<String, _>("function_schema"),
1183 row.get::<String, _>("function_name"),
1184 ),
1185 return_type: row.get::<String, _>("return_type"),
1186 description: row.get::<Option<String>, _>("description"),
1187 row_argument: row.get::<Option<String>, _>("row_argument"),
1188 session_argument: row.get::<Option<String>, _>("session_argument"),
1189 takes_arguments: row.get::<bool, _>("takes_arguments"),
1190 })
1191 .collect())
1192}
1193
1194#[derive(Clone, Debug)]
1196pub struct ComputedColumnRow {
1197 pub table: QualifiedIdentifier,
1199 pub name: String,
1201 pub function: QualifiedIdentifier,
1203 pub return_type: String,
1205 pub description: Option<String>,
1207 pub row_argument: Option<String>,
1209 pub session_argument: Option<String>,
1211 pub takes_arguments: bool,
1213}
1214
1215#[derive(Clone, Debug)]
1217pub struct ViewColumn {
1218 pub view: QualifiedIdentifier,
1220 pub base: QualifiedIdentifier,
1222 pub base_column: String,
1224 pub view_column: String,
1226}
1227
1228pub async fn load_view_columns(pool: &PgPool, schemas: &[String]) -> Result<Vec<ViewColumn>> {
1245 let rows = sqlx::query(
1246 r#"
1247 WITH RECURSIVE
1248 views AS (
1249 SELECT c.oid AS view_id, c.relnamespace AS view_schema_id,
1250 n.nspname AS view_schema, c.relname AS view_name,
1251 r.ev_action AS view_definition
1252 FROM pg_class c
1253 JOIN pg_namespace n ON n.oid = c.relnamespace
1254 JOIN pg_rewrite r ON r.ev_class = c.oid
1255 WHERE c.relkind IN ('v', 'm')
1256 ),
1257 transform_json AS (
1258 SELECT view_id, view_schema_id, view_schema, view_name,
1259 replace(
1260 replace(
1261 replace(
1262 replace(
1263 replace(
1264 replace(
1265 replace(
1266 regexp_replace(
1267 replace(
1268 replace(
1269 replace(
1270 replace(
1271 replace(
1272 replace(
1273 replace(
1274 replace(
1275 replace(
1276 replace(
1277 replace(
1278 view_definition::text,
1279 '<>', '()'
1280 ), ',', ''
1281 ), E'\\{', ''
1282 ), E'\\}', ''
1283 ), ' :targetList ', ',"targetList":'
1284 ), ' :resno ', ',"resno":'
1285 ), ' :resorigtbl ', ',"resorigtbl":'
1286 ), ' :resorigcol ', ',"resorigcol":'
1287 ), '{', '{ :'
1288 ), '((', '{(('
1289 ), '({', '{({'
1290 ), ' :[^}{,]+', ',"":', 'g'
1291 ), ',"":}', '}'
1292 ), ',"":,', ','
1293 ), '{(', '('
1294 ), '{,', '{'
1295 ), '(', '['
1296 ), ')', ']'
1297 ), ' ', ','
1298 )::json AS view_definition
1299 FROM views
1300 ),
1301 results AS (
1302 SELECT view_id, view_schema_id, view_schema, view_name,
1303 (entry->>'resno')::int AS view_column,
1304 (entry->>'resorigtbl')::oid AS resorigtbl,
1305 (entry->>'resorigcol')::int AS resorigcol
1306 FROM (
1307 SELECT view_id, view_schema_id, view_schema, view_name,
1308 json_array_elements(view_definition->0->'targetList') AS entry
1309 FROM transform_json
1310 ) target_entries
1311 ),
1312 -- A view over a view is followed down to the base table. `path`
1313 -- carries the tables already visited so a cyclic definition stops
1314 -- rather than recursing forever.
1315 recursion(view_id, view_schema, view_name, view_column,
1316 resorigtbl, resorigcol, is_cycle, path) AS (
1317 SELECT r.view_id, r.view_schema, r.view_name, r.view_column,
1318 r.resorigtbl, r.resorigcol, false, ARRAY[r.resorigtbl]
1319 FROM results r
1320 WHERE r.view_schema = ANY($1)
1321 UNION ALL
1322 SELECT v.view_id, v.view_schema, v.view_name, v.view_column,
1323 t.resorigtbl, t.resorigcol,
1324 t.resorigtbl = ANY(v.path), v.path || t.resorigtbl
1325 FROM recursion v
1326 JOIN results t ON v.resorigtbl = t.view_id AND v.resorigcol = t.view_column
1327 WHERE NOT v.is_cycle
1328 )
1329 SELECT DISTINCT
1330 rec.view_schema, rec.view_name,
1331 sch.nspname AS base_schema, tbl.relname AS base_name,
1332 col.attname AS base_column, vcol.attname AS view_column
1333 FROM recursion rec
1334 JOIN pg_attribute vcol ON vcol.attrelid = rec.view_id
1335 AND vcol.attnum = rec.view_column
1336 AND NOT vcol.attisdropped
1337 JOIN pg_class tbl ON tbl.oid = rec.resorigtbl
1338 JOIN pg_namespace sch ON sch.oid = tbl.relnamespace
1339 JOIN pg_attribute col ON col.attrelid = tbl.oid
1340 AND col.attnum = rec.resorigcol
1341 AND NOT col.attisdropped
1342 WHERE rec.resorigtbl <> 0
1343 "#,
1344 )
1345 .bind(schemas)
1346 .fetch_all(pool)
1347 .await
1348 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
1349
1350 Ok(rows
1351 .into_iter()
1352 .map(|row| ViewColumn {
1353 view: QualifiedIdentifier::new(
1354 row.get::<String, _>("view_schema"),
1355 row.get::<String, _>("view_name"),
1356 ),
1357 base: QualifiedIdentifier::new(
1358 row.get::<String, _>("base_schema"),
1359 row.get::<String, _>("base_name"),
1360 ),
1361 base_column: row.get::<String, _>("base_column"),
1362 view_column: row.get::<String, _>("view_column"),
1363 })
1364 .collect())
1365}
1366
1367pub async fn load_unique_constraints(
1383 pool: &PgPool,
1384 schemas: &[String],
1385) -> Result<HashMap<QualifiedIdentifier, Vec<(String, Vec<String>)>>> {
1386 let rows = sqlx::query(
1387 r#"
1388 SELECT
1389 n.nspname AS table_schema,
1390 c.relname AS table_name,
1391 con.conname AS constraint_name,
1392 (SELECT array_agg(a.attname ORDER BY k.ord)
1393 FROM unnest(con.conkey) WITH ORDINALITY AS k(attnum, ord)
1394 JOIN pg_attribute a
1395 ON a.attrelid = c.oid AND a.attnum = k.attnum) AS columns
1396 FROM pg_constraint con
1397 JOIN pg_class c ON c.oid = con.conrelid
1398 JOIN pg_namespace n ON n.oid = c.relnamespace
1399 WHERE con.contype IN ('p', 'u')
1400 AND n.nspname = ANY($1)
1401 "#,
1402 )
1403 .bind(schemas)
1404 .fetch_all(pool)
1405 .await
1406 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
1407
1408 let mut by_table: HashMap<QualifiedIdentifier, Vec<(String, Vec<String>)>> = HashMap::new();
1409 for row in rows {
1410 let Some(columns): Option<Vec<String>> = row.get("columns") else {
1411 continue;
1412 };
1413 by_table
1414 .entry(QualifiedIdentifier::new(
1415 row.get::<String, _>("table_schema"),
1416 row.get::<String, _>("table_name"),
1417 ))
1418 .or_default()
1419 .push((row.get::<String, _>("constraint_name"), columns));
1420 }
1421 for constraints in by_table.values_mut() {
1422 constraints.sort();
1423 }
1424 Ok(by_table)
1425}
1426
1427pub async fn load_primary_keys(
1428 pool: &PgPool,
1429 schemas: &[String],
1430) -> Result<HashMap<QualifiedIdentifier, Vec<String>>> {
1431 let rows = sqlx::query(
1432 r#"
1433 SELECT
1434 n.nspname AS table_schema,
1435 c.relname AS table_name,
1436 (SELECT array_agg(a.attname ORDER BY k.ord)
1437 FROM unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord)
1438 JOIN pg_attribute a
1439 ON a.attrelid = c.oid AND a.attnum = k.attnum) AS pk_cols
1440 FROM pg_index i
1441 JOIN pg_class c ON c.oid = i.indrelid
1442 JOIN pg_namespace n ON n.oid = c.relnamespace
1443 WHERE i.indisprimary
1444 AND n.nspname = ANY($1)
1445 "#,
1446 )
1447 .bind(schemas)
1448 .fetch_all(pool)
1449 .await
1450 .map_err(|e| Error::SchemaCacheLoadFailed(e.to_string()))?;
1451
1452 Ok(rows
1453 .into_iter()
1454 .filter_map(|row| {
1455 let columns: Option<Vec<String>> = row.get("pk_cols");
1456 Some((
1457 QualifiedIdentifier::new(
1458 row.get::<String, _>("table_schema"),
1459 row.get::<String, _>("table_name"),
1460 ),
1461 columns?,
1462 ))
1463 })
1464 .collect())
1465}
1466
1467pub async fn enum_values(
1476 pool: &PgPool,
1477 schema: &str,
1478 table: &str,
1479 key_column: &str,
1480 has_comment: bool,
1481) -> Result<Vec<(String, Option<String>)>> {
1482 let failed = |e: sqlx::Error| Error::SchemaCacheLoadFailed(e.to_string());
1483 let mut tx = pool.begin().await.map_err(failed)?;
1484
1485 let reader: Option<String> = sqlx::query_scalar(
1486 r#"
1487 SELECT r.rolname::text
1488 FROM pg_class c
1489 JOIN pg_namespace n ON n.oid = c.relnamespace
1490 JOIN pg_roles r ON pg_has_role(r.oid, 'MEMBER')
1491 WHERE n.nspname = $1 AND c.relname = $2
1492 AND has_schema_privilege(r.oid, n.oid, 'USAGE')
1493 AND has_table_privilege(r.oid, c.oid, 'SELECT')
1494 ORDER BY r.rolname = current_user DESC, r.rolname
1495 LIMIT 1
1496 "#,
1497 )
1498 .bind(schema)
1499 .bind(table)
1500 .fetch_optional(&mut *tx)
1501 .await
1502 .map_err(failed)?;
1503 let current: String = sqlx::query_scalar("SELECT current_user::text")
1504 .fetch_one(&mut *tx)
1505 .await
1506 .map_err(failed)?;
1507 if let Some(reader) = reader.filter(|reader| *reader != current) {
1510 sqlx::query(&format!(
1511 "SET LOCAL ROLE {}",
1512 postrust_sql::escape_ident(&reader)
1513 ))
1514 .execute(&mut *tx)
1515 .await
1516 .map_err(failed)?;
1517 }
1518
1519 let comment = match has_comment {
1520 true => format!("{}::text", postrust_sql::escape_ident("comment")),
1521 false => "NULL::text".to_string(),
1522 };
1523 let sql = format!(
1524 "SELECT {}::text, {} FROM {}.{} ORDER BY 1",
1525 postrust_sql::escape_ident(key_column),
1526 comment,
1527 postrust_sql::escape_ident(schema),
1528 postrust_sql::escape_ident(table)
1529 );
1530 let rows: Vec<(Option<String>, Option<String>)> = sqlx::query_as(&sql)
1531 .fetch_all(&mut *tx)
1532 .await
1533 .map_err(failed)?;
1534 tx.rollback().await.map_err(failed)?;
1535
1536 Ok(rows
1537 .into_iter()
1538 .filter_map(|(value, comment)| value.map(|v| (v, comment)))
1539 .collect())
1540}