guarantee table names are unique (#656)

<!-- The PR description should answer 2 (maybe 3) important questions:
-->

### What

<!-- What is this PR trying to accomplish (and why, if it's not
obvious)? -->

Table names may conflict with scalar type names.


<!-- Consider: do we need to add a changelog entry? -->

### How

<!-- How is it trying to accomplish it (what are the implementation
steps)? -->

 Prevent conflicts by:

-- assigning unique aliases to tables during introspection, if needed
-- preserving customized aliases during update operations

changelog.md

@@ -6,10 +6,13 @@
 
 ### Changed
 
+- When updating configuration, if collection or field name is customized, keep the customized name.
+
 ### Fixed
 
 - Predicates in relationships using in ordering (usually supplied by the engine's permission system) would fail to join the related tables correctly if the predicate was null.
   [#655](https://github.com/hasura/ndc-postgres/pull/655)
+- Table names that conflict with scalar type names will now be aliased, with a suffix starting with `_table`, and from then `_table_n` where `n` is an incrementing integer, until a unique name is found. [hasura/graphql-engine#10570](https://github.com/hasura/graphql-engine/issues/10570)
 
 ## [v1.2.0] - 2024-10-25
 

crates/configuration/src/version3/mod.rs

@@ -5,9 +5,9 @@ pub mod connection_settings;
 pub mod metadata;
 pub(crate) mod options;
 
-use ndc_models as models;
+use ndc_models::{self as models, CollectionName, TypeName};
 use std::borrow::Cow;
-use std::collections::{BTreeMap, BTreeSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::path::Path;
 
 use schemars::JsonSchema;
@@ -182,6 +182,16 @@ pub async fn introspect(
     let relevant_type_representations =
         filter_type_representations(&scalar_types, type_representations);
 
+    // build a list of names to ensure they are unique. We assume scalar type names + composite types are already a unique set.
+    let mut type_names: HashSet<TypeName> = scalar_types
+        .into_iter()
+        .map(ndc_models::ScalarTypeName::into_inner)
+        .collect();
+
+    type_names.extend(composite_types.0.keys().cloned());
+
+    let tables = get_aliased_tables(type_names, tables, &args.metadata.tables);
+
     Ok(RawConfiguration {
         version: Version::This,
         schema: args.schema,
@@ -199,6 +209,85 @@ pub async fn introspect(
     })
 }
 
+/// given scalar type names already in use, introspected tables, and optionally any existing table configuration:
+/// get collections with names guaranteed unique, preserving customized collection and field names if any
+fn get_aliased_tables(
+    type_names: HashSet<TypeName>,
+    tables: metadata::TablesInfo,
+    old_tables: &metadata::TablesInfo,
+) -> metadata::TablesInfo {
+    let mut type_names = type_names;
+    let mut mapped_tables = BTreeMap::new();
+
+    for (collection_name, table_info) in tables.0 {
+        let old_config = old_tables.0.iter().find(|(_, old_table_info)| {
+            old_table_info.table_name == table_info.table_name
+                && old_table_info.schema_name == table_info.schema_name
+        });
+
+        // use the old collection alias if one exists
+        let collection_name = old_config
+            .map_or(&collection_name, |(collection_name, _)| collection_name)
+            .to_owned();
+
+        // add a suffix to the collection name if needed
+        let collection_name = get_unique_collection_name(collection_name, &type_names);
+
+        type_names.insert(collection_name.clone().into_inner().into());
+
+        // if a column has a customized field name, keep it
+        let table_info = metadata::TableInfo {
+            columns: table_info
+                .columns
+                .into_iter()
+                .map(|(field_name, column_info)| {
+                    let field_name = old_config
+                        .and_then(|(_, table_info)| {
+                            table_info
+                                .columns
+                                .iter()
+                                .find(|(_, old_column_info)| {
+                                    old_column_info.name == column_info.name
+                                })
+                                .map(|(field_name, _)| field_name.to_owned())
+                        })
+                        .unwrap_or(field_name);
+
+                    (field_name, column_info)
+                })
+                .collect(),
+            ..table_info
+        };
+
+        mapped_tables.insert(collection_name, table_info);
+    }
+
+    metadata::TablesInfo(mapped_tables)
+}
+
+/// given a collection name and a list of already used type names, get a unique name by adding a suffix if needed
+fn get_unique_collection_name(
+    collection_name: CollectionName,
+    type_names: &HashSet<TypeName>,
+) -> CollectionName {
+    let mut collection_name = collection_name;
+
+    if type_names.contains(collection_name.as_ref()) {
+        let mut aliased_collection_name: CollectionName = format!("{collection_name}_table").into();
+
+        for counter in 1.. {
+            if !type_names.contains(aliased_collection_name.as_ref()) {
+                collection_name = aliased_collection_name;
+                break;
+            }
+
+            aliased_collection_name = format!("{collection_name}_table_{counter}").into();
+        }
+    }
+
+    collection_name
+}
+
 /// Merge the type representations currenting defined in the user's configuration with
 /// our base defaults. User configuration takes precedence.
 fn base_type_representations(

crates/configuration/src/version4/mod.rs

@@ -8,8 +8,9 @@ pub mod options;
 mod to_runtime_configuration;
 mod upgrade_from_v3;
 
+use ndc_models::{CollectionName, TypeName};
 use std::borrow::Cow;
-use std::collections::HashSet;
+use std::collections::{BTreeMap, HashSet};
 use std::path::Path;
 pub use to_runtime_configuration::make_runtime_configuration;
 pub use upgrade_from_v3::upgrade_from_v3;
@@ -173,6 +174,17 @@ pub async fn introspect(
     .instrument(info_span!("Decode introspection result"))
     .await?;
 
+    // build a list of names to ensure they are unique. We assume scalar type names + composite types are already a unique set.
+    let mut type_names: HashSet<TypeName> = scalar_types
+        .0
+        .keys()
+        .map(|t| t.clone().into_inner())
+        .collect();
+
+    type_names.extend(composite_types.0.keys().cloned());
+
+    let tables = get_aliased_tables(type_names, tables, &args.metadata.tables);
+
     Ok(ParsedConfiguration {
         version: Version::This,
         schema: args.schema,
@@ -188,6 +200,85 @@ pub async fn introspect(
     })
 }
 
+/// given scalar type names already in use, introspected tables, and optionally any existing table configuration:
+/// get collections with names guaranteed unique, preserving customized collection and field names if any
+fn get_aliased_tables(
+    type_names: HashSet<TypeName>,
+    tables: metadata::TablesInfo,
+    old_tables: &metadata::TablesInfo,
+) -> metadata::TablesInfo {
+    let mut type_names = type_names;
+    let mut mapped_tables = BTreeMap::new();
+
+    for (collection_name, table_info) in tables.0 {
+        let old_config = old_tables.0.iter().find(|(_, old_table_info)| {
+            old_table_info.table_name == table_info.table_name
+                && old_table_info.schema_name == table_info.schema_name
+        });
+
+        // use the old collection alias if one exists
+        let collection_name = old_config
+            .map_or(&collection_name, |(collection_name, _)| collection_name)
+            .to_owned();
+
+        // add a suffix to the collection name if needed
+        let collection_name = get_unique_collection_name(collection_name, &type_names);
+
+        type_names.insert(collection_name.clone().into_inner().into());
+
+        // if a column has a customized field name, keep it
+        let table_info = metadata::TableInfo {
+            columns: table_info
+                .columns
+                .into_iter()
+                .map(|(field_name, column_info)| {
+                    let field_name = old_config
+                        .and_then(|(_, table_info)| {
+                            table_info
+                                .columns
+                                .iter()
+                                .find(|(_, old_column_info)| {
+                                    old_column_info.name == column_info.name
+                                })
+                                .map(|(field_name, _)| field_name.to_owned())
+                        })
+                        .unwrap_or(field_name);
+
+                    (field_name, column_info)
+                })
+                .collect(),
+            ..table_info
+        };
+
+        mapped_tables.insert(collection_name, table_info);
+    }
+
+    metadata::TablesInfo(mapped_tables)
+}
+
+/// given a collection name and a list of already used type names, get a unique name by adding a suffix if needed
+fn get_unique_collection_name(
+    collection_name: CollectionName,
+    type_names: &HashSet<TypeName>,
+) -> CollectionName {
+    let mut collection_name = collection_name;
+
+    if type_names.contains(collection_name.as_ref()) {
+        let mut aliased_collection_name: CollectionName = format!("{collection_name}_table").into();
+
+        for counter in 1.. {
+            if !type_names.contains(aliased_collection_name.as_ref()) {
+                collection_name = aliased_collection_name;
+                break;
+            }
+
+            aliased_collection_name = format!("{collection_name}_table_{counter}").into();
+        }
+    }
+
+    collection_name
+}
+
 /// Parse the configuration format from a directory.
 pub async fn parse_configuration(
     configuration_dir: impl AsRef<Path>,

crates/configuration/src/version5/mod.rs

@@ -8,7 +8,8 @@ mod options;
 mod to_runtime_configuration;
 mod upgrade_from_v4;
 
-use std::collections::HashSet;
+use ndc_models::{CollectionName, TypeName};
+use std::collections::{BTreeMap, HashSet};
 use std::path::Path;
 pub use to_runtime_configuration::make_runtime_configuration;
 pub use upgrade_from_v4::upgrade_from_v4;
@@ -182,6 +183,17 @@ pub async fn introspect(
     .instrument(info_span!("Decode introspection result"))
     .await?;
 
+    // build a list of names to ensure they are unique. We assume scalar type names + composite types are already a unique set.
+    let mut type_names: HashSet<TypeName> = scalar_types
+        .0
+        .keys()
+        .map(|t| t.clone().into_inner())
+        .collect();
+
+    type_names.extend(composite_types.0.keys().cloned());
+
+    let tables = get_aliased_tables(type_names, tables, &args.metadata.tables);
+
     Ok(ParsedConfiguration {
         version: Version::This,
         schema: args.schema,
@@ -200,6 +212,85 @@ pub async fn introspect(
     })
 }
 
+/// given scalar type names already in use, introspected tables, and optionally any existing table configuration:
+/// get collections with names guaranteed unique, preserving customized collection and field names if any
+fn get_aliased_tables(
+    type_names: HashSet<TypeName>,
+    tables: metadata::TablesInfo,
+    old_tables: &metadata::TablesInfo,
+) -> metadata::TablesInfo {
+    let mut type_names = type_names;
+    let mut mapped_tables = BTreeMap::new();
+
+    for (collection_name, table_info) in tables.0 {
+        let old_config = old_tables.0.iter().find(|(_, old_table_info)| {
+            old_table_info.table_name == table_info.table_name
+                && old_table_info.schema_name == table_info.schema_name
+        });
+
+        // use the old collection alias if one exists
+        let collection_name = old_config
+            .map_or(&collection_name, |(collection_name, _)| collection_name)
+            .to_owned();
+
+        // add a suffix to the collection name if needed
+        let collection_name = get_unique_collection_name(collection_name, &type_names);
+
+        type_names.insert(collection_name.clone().into_inner().into());
+
+        // if a column has a customized field name, keep it
+        let table_info = metadata::TableInfo {
+            columns: table_info
+                .columns
+                .into_iter()
+                .map(|(field_name, column_info)| {
+                    let field_name = old_config
+                        .and_then(|(_, table_info)| {
+                            table_info
+                                .columns
+                                .iter()
+                                .find(|(_, old_column_info)| {
+                                    old_column_info.name == column_info.name
+                                })
+                                .map(|(field_name, _)| field_name.to_owned())
+                        })
+                        .unwrap_or(field_name);
+
+                    (field_name, column_info)
+                })
+                .collect(),
+            ..table_info
+        };
+
+        mapped_tables.insert(collection_name, table_info);
+    }
+
+    metadata::TablesInfo(mapped_tables)
+}
+
+/// given a collection name and a list of already used type names, get a unique name by adding a suffix if needed
+fn get_unique_collection_name(
+    collection_name: CollectionName,
+    type_names: &HashSet<TypeName>,
+) -> CollectionName {
+    let mut collection_name = collection_name;
+
+    if type_names.contains(collection_name.as_ref()) {
+        let mut aliased_collection_name: CollectionName = format!("{collection_name}_table").into();
+
+        for counter in 1.. {
+            if !type_names.contains(aliased_collection_name.as_ref()) {
+                collection_name = aliased_collection_name;
+                break;
+            }
+
+            aliased_collection_name = format!("{collection_name}_table_{counter}").into();
+        }
+    }
+
+    collection_name
+}
+
 /// Parse the configuration format from a directory.
 pub async fn parse_configuration(
     configuration_dir: impl AsRef<Path>,

crates/tests/databases-tests/src/postgres/snapshots/databases_tests__postgres__cli_version3_tests__postgres_current_only_configure_initial_configuration_is_unchanged.snap

@@ -1,6 +1,7 @@
 ---
 source: crates/tests/databases-tests/src/postgres/cli_version3_tests.rs
 expression: default_configuration
+snapshot_kind: text
 ---
 {
   "version": "3",
@@ -1070,6 +1071,31 @@ expression: default_configuration
         "foreignRelations": {},
         "description": null
       },
+      "text_table": {
+        "schemaName": "public",
+        "tableName": "text",
+        "columns": {
+          "content": {
+            "name": "content",
+            "type": {
+              "scalarType": "text"
+            },
+            "nullable": "nullable",
+            "description": null
+          },
+          "id": {
+            "name": "id",
+            "type": {
+              "scalarType": "int4"
+            },
+            "nullable": "nullable",
+            "description": null
+          }
+        },
+        "uniquenessConstraints": {},
+        "foreignRelations": {},
+        "description": null
+      },
       "topology_layer": {
         "schemaName": "topology",
         "tableName": "layer",