feat(schema-engine): prepare skeleton for implicit many-to-many table logical replication on Postgres

libs/test-setup/src/capabilities.rs

@@ -3,8 +3,9 @@
 #[derive(Copy, Clone, Debug, PartialEq)]
 #[repr(u8)]
 pub enum Capabilities {
-    ScalarLists = 0b0001,
-    Enums = 0b0010,
-    Json = 0b0100,
-    CreateDatabase = 0b1000,
+    ScalarLists = 1,
+    Enums = 1 << 1,
+    Json = 1 << 2,
+    CreateDatabase = 1 << 3,
+    LogicalReplication = 1 << 4,
 }

schema-engine/connectors/sql-schema-connector/src/apply_migration.rs

@@ -145,6 +145,12 @@ fn render_raw_sql(
             renderer.render_drop_table(table.namespace(), table.name())
         }
         SqlMigrationStep::RedefineIndex { index } => renderer.render_drop_and_recreate_index(schemas.walk(*index)),
+        SqlMigrationStep::ImplicitManyToManyRefinement { index, table_id } => {
+            vec![renderer.refine_implicit_many_to_many_table(
+                schemas.next.walk(*table_id).name(),
+                schemas.next.walk(*index).name(),
+            )]
+        }
         SqlMigrationStep::AddForeignKey { foreign_key_id } => {
             let foreign_key = schemas.next.walk(*foreign_key_id);
             vec![renderer.render_add_foreign_key(foreign_key)]

schema-engine/connectors/sql-schema-connector/src/flavour/postgres.rs

@@ -149,13 +149,19 @@ impl PostgresFlavour {
         }
     }
 
-    fn circumstances(&self) -> Option<BitFlags<Circumstances>> {
+    pub(crate) fn circumstances(&self) -> Option<BitFlags<Circumstances>> {
         match &self.state {
             State::Initial | State::WithParams(_) => None,
             State::Connected(_, (circ, _)) => Some(*circ),
         }
     }
 
+    pub(crate) fn can_replicate_logically(&self) -> bool {
+        self.circumstances()
+            .map(|c| c.contains(Circumstances::CanReplicateLogically))
+            .unwrap_or(false)
+    }
+
     pub(crate) fn is_cockroachdb(&self) -> bool {
         self.provider == PostgresProvider::CockroachDb
             || self
@@ -657,6 +663,7 @@ pub(crate) enum Circumstances {
     IsCockroachDb,
     CockroachWithPostgresNativeTypes, // FIXME: we should really break and remove this
     CanPartitionTables,
+    CanReplicateLogically,
 }
 
 fn disable_postgres_statement_cache(url: &mut Url) -> ConnectorResult<()> {
@@ -678,6 +685,48 @@ fn disable_postgres_statement_cache(url: &mut Url) -> ConnectorResult<()> {
     Ok(())
 }
 
+fn infer_circumstances(
+    circumstances: &mut BitFlags<Circumstances, u8>,
+    provider: PostgresProvider,
+    version: Option<(String, i64)>,
+) -> ConnectorResult<()> {
+    match version {
+        Some((version, version_num)) => {
+            let db_is_cockroach = version.contains("CockroachDB");
+
+            if db_is_cockroach {
+                if provider == PostgresProvider::PostgreSql {
+                    let msg = "You are trying to connect to a CockroachDB database, but the provider in your Prisma schema is `postgresql`. Please change it to `cockroachdb`.";
+
+                    return Err(ConnectorError::from_msg(msg.to_owned()));
+                }
+
+                *circumstances |= Circumstances::IsCockroachDb;
+                return Ok(());
+            }
+
+            if provider == PostgresProvider::CockroachDb {
+                let msg = "You are trying to connect to a PostgreSQL database, but the provider in your Prisma schema is `cockroachdb`. Please change it to `postgresql`.";
+
+                return Err(ConnectorError::from_msg(msg.to_owned()));
+            }
+
+            if version_num >= 100000 {
+                // https://www.postgresql.org/docs/10/ddl-partitioning.html
+                *circumstances |= Circumstances::CanPartitionTables;
+
+                // https://www.postgresql.org/docs/10/logical-replication.html
+                *circumstances |= Circumstances::CanReplicateLogically;
+            }
+        }
+        None => {
+            tracing::warn!("Could not determine the version of the database.");
+        }
+    };
+
+    Ok(())
+}
+
 fn with_connection<'a, O, F, C>(flavour: &'a mut PostgresFlavour, f: C) -> BoxFuture<'a, ConnectorResult<O>>
 where
     O: 'a,
@@ -718,32 +767,10 @@ where
                           .and_then(|row| row.at(1).and_then(|ver_str| row.at(2).map(|ver_num| (ver_str, ver_num))))
                           .and_then(|(ver_str,ver_num)| ver_str.to_string().and_then(|version| ver_num.as_integer().map(|version_number| (version, version_number))));
 
-                    match version {
-                        Some((version, version_num)) => {
-                            let db_is_cockroach = version.contains("CockroachDB");
-
-                            if db_is_cockroach && provider == PostgresProvider::PostgreSql  {
-                                let msg = "You are trying to connect to a CockroachDB database, but the provider in your Prisma schema is `postgresql`. Please change it to `cockroachdb`.";
-
-                                return Err(ConnectorError::from_msg(msg.to_owned()));
-                            }
-
-                            if !db_is_cockroach && provider == PostgresProvider::CockroachDb {
-                                let msg = "You are trying to connect to a PostgreSQL database, but the provider in your Prisma schema is `cockroachdb`. Please change it to `postgresql`.";
-
-                                return Err(ConnectorError::from_msg(msg.to_owned()));
-                            }
+                    infer_circumstances(&mut circumstances, provider, version)?;
 
-                            if db_is_cockroach {
-                                circumstances |= Circumstances::IsCockroachDb;
-                                connection.raw_cmd(COCKROACHDB_PRELUDE, &params.url).await?;
-                            } else if version_num >= 100000 {
-                                circumstances |= Circumstances:: CanPartitionTables;
-                            }
-                        }
-                        None => {
-                            tracing::warn!("Could not determine the version of the database.")
-                        }
+                    if circumstances.contains(Circumstances::IsCockroachDb) {
+                        connection.raw_cmd(COCKROACHDB_PRELUDE, &params.url).await?;
                     }
 
                     if let Some(true) = schema_exists_result

schema-engine/connectors/sql-schema-connector/src/flavour/postgres/connection.rs

@@ -83,6 +83,9 @@ impl Connection {
         namespaces: Option<Namespaces>,
     ) -> ConnectorResult<SqlSchema> {
         use sql_schema_describer::{postgres as describer, DescriberErrorKind, SqlSchemaDescriberBackend};
+
+        // TODO: `sql_schema_connector::flavour::postgres::Circumstances` and `sql_schema_describer::Circumstances` are identitcal.
+        // What follows is a redundant and confusing mapping.
         let mut describer_circumstances: BitFlags<describer::Circumstances> = Default::default();
 
         if circumstances.contains(super::Circumstances::IsCockroachDb) {
@@ -97,6 +100,10 @@ impl Connection {
             describer_circumstances |= describer::Circumstances::CanPartitionTables;
         }
 
+        if circumstances.contains(super::Circumstances::CanReplicateLogically) {
+            describer_circumstances |= describer::Circumstances::CanReplicateLogically;
+        }
+
         let namespaces_vec = Namespaces::to_vec(namespaces, String::from(params.url.schema()));
         let namespaces_str: Vec<&str> = namespaces_vec.iter().map(AsRef::as_ref).collect();
 

schema-engine/connectors/sql-schema-connector/src/sql_migration.rs

@@ -55,6 +55,7 @@ impl SqlMigration {
             RedefinedTable,
             ChangedEnum,
             ChangedTable,
+            AlterManyToManyTable,
         }
 
         // (sort key, item name, step index)
@@ -158,6 +159,10 @@ impl SqlMigration {
                         idx,
                     ));
                 }
+                SqlMigrationStep::ImplicitManyToManyRefinement { table_id, .. } => {
+                    let table = self.schemas().next.walk(*table_id).name();
+                    drift_items.insert((DriftType::AlterManyToManyTable, table, idx));
+                }
                 SqlMigrationStep::CreateExtension(create_extension) => {
                     let ext: &PostgresSchemaExt = self.schemas().next.downcast_connector_data();
                     let extension = ext.get_extension(create_extension.id);
@@ -200,6 +205,11 @@ impl SqlMigration {
                         out.push_str(item_name);
                         out.push_str("`\n")
                     }
+                    DriftType::AlterManyToManyTable => {
+                        out.push_str("\n[*] Altered many-to-many table `");
+                        out.push_str(item_name);
+                        out.push_str("`\n")
+                    }
                     DriftType::ChangedEnum => {
                         out.push_str("\n[*] Changed the `");
                         out.push_str(item_name);
@@ -349,6 +359,11 @@ impl SqlMigration {
                     out.push_str(self.schemas().next.walk(*table_id).name());
                     out.push('\n');
                 }
+                SqlMigrationStep::ImplicitManyToManyRefinement { table_id, .. } => {
+                    out.push_str("  [*] Refined implicit many-to-many table `");
+                    out.push_str(self.schemas().next.walk(*table_id).name());
+                    out.push_str("`\n");
+                }
                 SqlMigrationStep::RedefineTables(_) => {}
                 SqlMigrationStep::RenameForeignKey { foreign_key_id } => {
                     let fks = self.schemas().walk(*foreign_key_id);
@@ -503,6 +518,10 @@ pub(crate) enum SqlMigrationStep {
     RedefineIndex {
         index: MigrationPair<IndexId>,
     },
+    ImplicitManyToManyRefinement {
+        table_id: TableId,
+        index: IndexId,
+    },
 }
 
 impl SqlMigrationStep {
@@ -523,6 +542,7 @@ impl SqlMigrationStep {
             SqlMigrationStep::DropTable { .. } => "DropTable",
             SqlMigrationStep::DropUserDefinedType(_) => "DropUserDefinedType",
             SqlMigrationStep::DropView(_) => "DropView",
+            SqlMigrationStep::ImplicitManyToManyRefinement { .. } => "ImplicitManyToManyRefinement",
             SqlMigrationStep::RedefineIndex { .. } => "RedefineIndex",
             SqlMigrationStep::RedefineTables { .. } => "RedefineTables",
             SqlMigrationStep::RenameForeignKey { .. } => "RenameForeignKey",

schema-engine/connectors/sql-schema-connector/src/sql_renderer.rs

@@ -70,6 +70,10 @@ pub(crate) trait SqlRenderer {
         unreachable!("unreachable render_drop_and_recreate_index")
     }
 
+    fn refine_implicit_many_to_many_table(&self, _table_name: &str, _index_name: &str) -> String {
+        unreachable!("unreachable refine_implicit_many_to_many_table")
+    }
+
     /// Render a `DropEnum` step.
     fn render_drop_enum(&self, dropped_enum: EnumWalker<'_>) -> Vec<String>;
 

schema-engine/connectors/sql-schema-connector/src/sql_renderer/postgres_renderer.rs

@@ -408,7 +408,12 @@ impl SqlRenderer for PostgresFlavour {
             String::new()
         };
 
-        format!("CREATE TABLE {table_name} (\n{columns}{pk}\n)")
+        let refinement_str = match table.is_implicit_m2m() {
+            true => "-- Implicitly many-to-many\n",
+            false => "",
+        };
+
+        format!("{}CREATE TABLE {table_name} (\n{columns}{pk}\n)", refinement_str)
     }
 
     fn render_drop_enum(&self, dropped_enum: EnumWalker<'_>) -> Vec<String> {
@@ -503,6 +508,10 @@ impl SqlRenderer for PostgresFlavour {
             for fk in tables.next.foreign_keys() {
                 result.push(self.render_add_foreign_key(fk));
             }
+
+            if tables.next.is_implicit_m2m() {
+                result.push(format!("-- {} is implicitly many-to-many", &tables.next.name()));
+            }
         }
 
         result
@@ -528,6 +537,11 @@ impl SqlRenderer for PostgresFlavour {
             next = self.quote(fks.next.constraint_name().unwrap()),
         )
     }
+
+    fn refine_implicit_many_to_many_table(&self, table_name: &str, index_name: &str) -> String {
+        // `REPLICA IDENTITY`: https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY
+        format!(r#"ALTER TABLE {table_name} REPLICA IDENTITY USING INDEX {index_name}"#,)
+    }
 }
 
 fn render_column_type(col: TableColumnWalker<'_>, flavour: &PostgresFlavour) -> Cow<'static, str> {

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ.rs

@@ -39,6 +39,7 @@ pub(crate) fn calculate_steps(
 
     flavour.push_enum_steps(&mut steps, &db);
     flavour.push_alter_sequence_steps(&mut steps, &db);
+    flavour.push_implicit_many_to_many_refinement_steps(&mut steps, &db);
 
     steps.sort();
 
@@ -55,12 +56,14 @@ fn push_created_table_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDataba
     for table in db.created_tables() {
         steps.push(SqlMigrationStep::CreateTable { table_id: table.id });
 
+        // Initial `-- AddForeignKey` steps
         if db.flavour.should_push_foreign_keys_from_created_tables() {
             for fk in table.foreign_keys() {
                 steps.push(SqlMigrationStep::AddForeignKey { foreign_key_id: fk.id });
             }
         }
 
+        // Initial `-- CreateIndex` steps
         if db.flavour.should_create_indexes_from_created_tables() {
             let create_indexes_from_created_tables = table
                 .indexes()
@@ -73,6 +76,20 @@ fn push_created_table_steps(steps: &mut Vec<SqlMigrationStep>, db: &DifferDataba
 
             steps.extend(create_indexes_from_created_tables);
         }
+
+        // Initial `-- ImplicitManyToManyRefinement` steps
+        if db.flavour.should_refine_implicit_many_to_many_tables() && table.is_implicit_m2m() {
+            let index = table
+                .indexes()
+                .find(|index| index.columns().len() == 2)
+                .expect("Implicit many-to-many tables must have a two-column index")
+                .id;
+
+            steps.push(SqlMigrationStep::ImplicitManyToManyRefinement {
+                table_id: table.id,
+                index,
+            });
+        }
     }
 }
 
@@ -529,13 +546,7 @@ fn next_column_has_virtual_default(column_id: TableColumnId, db: &DifferDatabase
 }
 
 fn is_prisma_implicit_m2m_fk(fk: ForeignKeyWalker<'_>) -> bool {
-    let table = fk.table();
-
-    if table.columns().count() != 2 {
-        return false;
-    }
-
-    table.column("A").is_some() && table.column("B").is_some()
+    fk.table().is_implicit_m2m()
 }
 
 fn all_match<T: PartialEq>(a: &mut dyn ExactSizeIterator<Item = T>, b: &mut dyn ExactSizeIterator<Item = T>) -> bool {

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ/sql_schema_differ_flavour.rs

@@ -53,6 +53,14 @@ pub(crate) trait SqlSchemaDifferFlavour {
     /// Push AlterExtension steps.
     fn push_extension_steps(&self, _steps: &mut Vec<SqlMigrationStep>, _db: &DifferDatabase<'_>) {}
 
+    /// Push implicit many-to-many relation refinement steps.
+    fn push_implicit_many_to_many_refinement_steps(
+        &self,
+        _steps: &mut Vec<SqlMigrationStep>,
+        _db: &DifferDatabase<'_>,
+    ) {
+    }
+
     /// Define database-specific extension modules.
     fn define_extensions(&self, _db: &mut DifferDatabase<'_>) {}
 
@@ -108,6 +116,11 @@ pub(crate) trait SqlSchemaDifferFlavour {
         true
     }
 
+    /// Whether the implicit many-to-many tables should be refined with further steps.
+    fn should_refine_implicit_many_to_many_tables(&self) -> bool {
+        false
+    }
+
     /// Whether to skip diffing JSON defaults.
     fn should_ignore_json_defaults(&self) -> bool {
         false

schema-engine/connectors/sql-schema-connector/src/sql_schema_differ/sql_schema_differ_flavour/postgres.rs

@@ -36,6 +36,10 @@
 static EXTENSION_VIEWS: Lazy<RegexSet> = Lazy::new(|| RegexSet::new(["(?i)^pg_buffercache$"]).unwrap());
 
 impl SqlSchemaDifferFlavour for PostgresFlavour {
+    fn should_refine_implicit_many_to_many_tables(&self) -> bool {
+        self.can_replicate_logically()
+    }
+
     fn can_alter_primary_keys(&self) -> bool {
         self.is_cockroachdb()
     }
@@ -97,6 +101,30 @@
         }
     }
 
+    fn push_implicit_many_to_many_refinement_steps(&self, steps: &mut Vec<SqlMigrationStep>, db: &DifferDatabase<'_>) {
+        // This stuff should only be applied if something changed in the schema.
+
+        // if self.is_cockroachdb() {
+        //     return;
+        // }
+
+        // let implicit_many_to_many_tables = db.table_pairs().map(|differ| differ.tables).filter_map(|table| {
+        //     if table.next.is_implicit_m2m() {
+        //         Some(table)
+        //     } else {
+        //         None
+        //     }
+        // });
+
+        // for table in implicit_many_to_many_tables {
+        //     let index = table.next.indexes().find(|idx| idx.is_unique()).unwrap().id;
+        //     let table_id = table.next.id;
+
+        //     println!("  steps++");
+        //     steps.push(SqlMigrationStep::ImplicitManyToManyRefinement { index, table_id });
+        // }
+    }
+
     fn push_alter_sequence_steps(&self, steps: &mut Vec<SqlMigrationStep>, db: &DifferDatabase<'_>) {
         if !self.is_cockroachdb() {
             return;