introspection: merge index+relation introspection and reintrospection

The path to multiSchema introspection we started on begins with a
refactoring aiming at merging introspection and reintrospection into one
step, and removing all DML usage from the introspection engine. This is
a part of it. See the following past PRs for more context:
#3338 and
#3333.

introspection-engine/connectors/sql-introspection-connector/src/calculate_datamodel.rs

@@ -1,16 +1,17 @@
-use crate::{introspection::introspect, SqlFamilyTrait, SqlIntrospectionResult};
+use crate::{
+    introspection::introspect, introspection_helpers::is_prisma_join_table, SqlFamilyTrait, SqlIntrospectionResult,
+};
 use introspection_connector::{IntrospectionContext, IntrospectionResult, Warning};
 use psl::{
     builtin_connectors::*,
     datamodel_connector::Connector,
     dml::Datamodel,
-    parser_database::{ast, walkers},
+    parser_database::{self, ast, walkers, RelationId},
     Configuration,
 };
 use quaint::prelude::SqlFamily;
 use sql_schema_describer as sql;
 use std::collections::HashMap;
-use tracing::debug;
 
 pub(crate) struct CalculateDatamodelContext<'a> {
     pub(crate) config: &'a Configuration,
@@ -21,6 +22,10 @@ pub(crate) struct CalculateDatamodelContext<'a> {
     pub(crate) warnings: &'a mut Vec<Warning>,
     pub(crate) previous_schema: &'a psl::ValidatedSchema,
     existing_enums: &'a HashMap<sql::EnumId, ast::EnumId>,
+    existing_models: &'a HashMap<sql::TableId, ast::ModelId>,
+    existing_scalar_fields: &'a HashMap<sql::ColumnId, (ast::ModelId, ast::FieldId)>,
+    existing_relations: &'a HashMap<sql::ForeignKeyId, parser_database::RelationId>,
+    existing_m2m_relations: &'a HashMap<sql::TableId, parser_database::RelationId>,
 }
 
 impl<'a> CalculateDatamodelContext<'a> {
@@ -55,6 +60,50 @@ impl<'a> CalculateDatamodelContext<'a> {
             .map(|enm| enm.name())
             .unwrap_or_else(|| self.schema.walk(id).name())
     }
+
+    pub(crate) fn existing_inline_relation(&self, id: sql::ForeignKeyId) -> Option<walkers::InlineRelationWalker<'a>> {
+        self.existing_relations
+            .get(&id)
+            .map(|relation_id| self.previous_schema.db.walk(*relation_id).refine().as_inline().unwrap())
+    }
+
+    pub(crate) fn existing_m2m_relation(
+        &self,
+        id: sql::TableId,
+    ) -> Option<walkers::ImplicitManyToManyRelationWalker<'a>> {
+        self.existing_m2m_relations.get(&id).map(|relation_id| {
+            self.previous_schema
+                .db
+                .walk(*relation_id)
+                .refine()
+                .as_many_to_many()
+                .unwrap()
+        })
+    }
+
+    pub(crate) fn existing_model(&self, id: sql::TableId) -> Option<walkers::ModelWalker<'a>> {
+        self.existing_models
+            .get(&id)
+            .map(|id| self.previous_schema.db.walk(*id))
+    }
+
+    pub(crate) fn existing_scalar_field(&self, id: sql::ColumnId) -> Option<walkers::ScalarFieldWalker<'a>> {
+        self.existing_scalar_fields
+            .get(&id)
+            .map(|(model_id, field_id)| self.previous_schema.db.walk(*model_id).scalar_field(*field_id))
+    }
+
+    pub(crate) fn column_prisma_name(&self, id: sql::ColumnId) -> &'a str {
+        self.existing_scalar_field(id)
+            .map(|sf| sf.name())
+            .unwrap_or_else(|| self.schema.walk(id).name())
+    }
+
+    pub(crate) fn model_prisma_name(&self, id: sql::TableId) -> &'a str {
+        self.existing_model(id)
+            .map(|model| model.name())
+            .unwrap_or_else(|| self.schema.walk(id).name())
+    }
 }
 
 /// Calculate a data model from a database schema.
@@ -96,6 +145,10 @@ pub fn calculate_datamodel(
             .collect()
     };
 
+    let existing_models = match_existing_models(schema, ctx);
+    let existing_scalar_fields = match_existing_scalar_fields(&existing_models, schema, ctx);
+    let existing_relations = match_existing_inline_relations(&existing_models, schema, ctx);
+    let existing_m2m_relations = match_existing_m2m_relations(schema, ctx);
     let mut warnings = Vec::new();
 
     let mut context = CalculateDatamodelContext {
@@ -107,16 +160,100 @@ pub fn calculate_datamodel(
         previous_schema: ctx.previous_schema(),
         warnings: &mut warnings,
         existing_enums: &existing_enums,
+        existing_models: &existing_models,
+        existing_scalar_fields: &existing_scalar_fields,
+        existing_relations: &existing_relations,
+        existing_m2m_relations: &existing_m2m_relations,
     };
 
     let (version, data_model, is_empty) = introspect(&mut context)?;
 
-    debug!("Done calculating datamodel.");
-
     Ok(IntrospectionResult {
         data_model,
         is_empty,
         warnings,
         version,
     })
 }
+
+fn match_existing_models(schema: &sql::SqlSchema, ctx: &IntrospectionContext) -> HashMap<sql::TableId, ast::ModelId> {
+    ctx.previous_schema()
+        .db
+        .walk_models()
+        .filter_map(|model| {
+            schema
+                .find_table(model.database_name())
+                .map(|sql_id| (sql_id, model.id))
+        })
+        .collect()
+}
+
+fn match_existing_scalar_fields(
+    existing_models: &HashMap<sql::TableId, ast::ModelId>,
+    schema: &sql::SqlSchema,
+    ctx: &IntrospectionContext,
+) -> HashMap<sql::ColumnId, (ast::ModelId, ast::FieldId)> {
+    schema
+        .walk_columns()
+        .filter_map(|col| {
+            let model_id = existing_models.get(&col.table().id)?;
+            let field_id = ctx
+                .previous_schema()
+                .db
+                .walk(*model_id)
+                .scalar_fields()
+                .find(|field| field.database_name() == col.name())
+                .map(|field| field.field_id())?;
+            Some((col.id, (*model_id, field_id)))
+        })
+        .collect()
+}
+
+fn match_existing_inline_relations(
+    existing_models: &HashMap<sql::TableId, ast::ModelId>,
+    schema: &sql::SqlSchema,
+    ctx: &IntrospectionContext,
+) -> HashMap<sql::ForeignKeyId, RelationId> {
+    schema
+        .walk_foreign_keys()
+        .filter_map(|fk| {
+            let referencing_model = *existing_models.get(&fk.table().id)?;
+            ctx.previous_schema()
+                .db
+                .walk(referencing_model)
+                .relations_from()
+                .filter_map(|rel| rel.refine().as_inline())
+                .find(|relation| {
+                    let referencing_fields = if let Some(fields) = relation.referencing_fields() {
+                        fields
+                    } else {
+                        return false;
+                    };
+                    let referencing_columns = fk.constrained_columns();
+                    referencing_fields.len() == referencing_columns.len()
+                        && referencing_fields
+                            .zip(referencing_columns)
+                            .all(|(field, col)| field.database_name() == col.name())
+                })
+                .map(|relation| (fk.id, relation.relation_id()))
+        })
+        .collect()
+}
+
+fn match_existing_m2m_relations(
+    schema: &sql::SqlSchema,
+    ctx: &IntrospectionContext,
+) -> HashMap<sql::TableId, RelationId> {
+    schema
+        .table_walkers()
+        .filter(|t| is_prisma_join_table(*t))
+        .filter_map(|table| {
+            ctx.previous_schema()
+                .db
+                .walk_relations()
+                .filter_map(|rel| rel.refine().as_many_to_many())
+                .find(|rel| rel.relation_name().to_string() == table.name()[1..])
+                .map(|rel| (table.id, rel.0.id))
+        })
+        .collect()
+}