feat: ORM-1110 externally managed tables - migrate dev (#5529)

This PR implements the functionality for filtering out tables marked as
"externally" managed by the coming `tables.external` `prisma.config.ts`
config option. It focuses on providing this functionality for the
`migrate dev` command.

The filtering happens during the actual diff computation. This is
necessary because we still want to create "outgoing" foreign key
constraints to external tables. For this we still need to know about
them during diff creation. The goal is for the diff creation to just not
create the SQL to create/update/delete external tables. ([also see slack
about
this](https://prisma-company.slack.com/archives/C0350LQSUF2/p1752229610438489))

Note: This PR comes with a minor change of behavior for schema creation.
Previously we would issue `CREATE SCHEMA` statements for a schema if it
just appears in the list of given schemas. Now that schema needs to have
a non-external model or enum assigned to it that we would consider
issuing a `CREATE SCHEMA` statement. This is to ensure that we do not
try to create externally managed schemas.

Implementation for `db push`, `db pull`, `migrate diff`, and the actual
config attribute will follow in dedicated PRs.

Author: FGoessler

query-engine/connector-test-kit-rs/qe-setup/src/lib.rs

@@ -18,7 +18,7 @@ use driver_adapters::DriverAdapter;
 use enumflags2::BitFlags;
 use providers::Provider;
 use psl::{builtin_connectors::*, Datasource};
-use schema_core::schema_connector::{ConnectorResult, SchemaConnector, SchemaDialect};
+use schema_core::schema_connector::{ConnectorResult, SchemaConnector, SchemaDialect, SchemaFilter};
 use std::env;
 
 #[derive(Debug, serde::Deserialize, PartialEq)]
@@ -151,7 +151,7 @@ pub async fn teardown(prisma_schema: &str, db_schemas: &[&str]) -> ConnectorResu
 pub(crate) async fn diff(schema: &str, dialect: &dyn SchemaDialect) -> ConnectorResult<String> {
     let from = dialect.empty_database_schema();
     let to = dialect.schema_from_datamodel(vec![("schema.prisma".to_string(), schema.into())])?;
-    let migration = dialect.diff(from, to);
+    let migration = dialect.diff(from, to, &SchemaFilter::default());
     dialect.render_script(&migration, &Default::default())
 }
 

schema-engine/commands/src/commands/create_migration.rs

@@ -37,13 +37,13 @@ pub async fn create_migration(
 
     let from = migration_schema_cache
         .get_or_insert(&input.migrations_list.migration_directories, || async {
+            // We pass the namespaces here, because we want to describe all of the namespaces we know about from the "to" schema.
             let namespaces = dialect.extract_namespaces(&to);
-            // We pass the namespaces here, because we want to describe all of these namespaces.
             connector.schema_from_migrations(&previous_migrations, namespaces).await
         })
         .await?;
 
-    let migration = dialect.diff(from, to);
+    let migration = dialect.diff(from, to, &input.filters.into());
 
     let extension = dialect.migration_file_extension().to_owned();
 

schema-engine/commands/src/commands/dev_diagnostic.rs

@@ -24,6 +24,7 @@ pub async fn dev_diagnostic(
     let diagnose_input = DiagnoseMigrationHistoryInput {
         migrations_list: input.migrations_list,
         opt_in_to_shadow_database: true,
+        schema_filter: input.schema_filter,
     };
 
     let diagnose_migration_history_output = diagnose_migration_history(

schema-engine/commands/src/commands/diagnose_migration_history.rs

@@ -146,13 +146,15 @@ pub async fn diagnose_migration_history(
                 })
                 .await;
             let to = connector.schema_from_database(namespaces.clone()).await;
-            let drift = match from.and_then(|from| to.map(|to| dialect.diff(from, to))).map(|mig| {
-                if dialect.migration_is_empty(&mig) {
-                    None
-                } else {
-                    Some(mig)
-                }
-            }) {
+            let drift = match from
+                .and_then(|from| to.map(|to| dialect.diff(from, to, &input.schema_filter.into())))
+                .map(|mig| {
+                    if dialect.migration_is_empty(&mig) {
+                        None
+                    } else {
+                        Some(mig)
+                    }
+                }) {
                 Ok(Some(drift)) => Some(DriftDiagnostic::DriftDetected {
                     summary: dialect.migration_summary(&drift),
                 }),

schema-engine/commands/src/commands/diff.rs

@@ -11,7 +11,7 @@ use json_rpc::types::MigrationList;
 use psl::SourceFile;
 use quaint::connector::ExternalConnectorFactory;
 use schema_connector::{
-    ConnectorError, DatabaseSchema, ExternalShadowDatabase, Namespaces, SchemaConnector, SchemaDialect,
+    ConnectorError, DatabaseSchema, ExternalShadowDatabase, Namespaces, SchemaConnector, SchemaDialect, SchemaFilter,
 };
 
 pub async fn diff(
@@ -49,7 +49,8 @@ pub async fn diff(
     let from = schema_from.unwrap_or_else(|| dialect.empty_database_schema());
     let to = schema_to.unwrap_or_else(|| dialect.empty_database_schema());
 
-    let migration = dialect.diff(from, to);
+    // TODO:(schema-filter) get filter from params and prisma config
+    let migration = dialect.diff(from, to, &SchemaFilter::default());
 
     let mut stdout = if params.script {
         dialect.render_script(&migration, &Default::default())?

schema-engine/commands/src/commands/evaluate_data_loss.rs

@@ -31,7 +31,7 @@ pub async fn evaluate_data_loss(
         })
         .await?;
 
-    let migration = dialect.diff(from, to);
+    let migration = dialect.diff(from, to, &input.filters.into());
 
     let migration_steps = dialect.migration_len(&migration) as u32;
     let diagnostics = connector.destructive_change_checker().check(&migration).await?;

schema-engine/commands/src/commands/schema_push.rs

@@ -1,5 +1,5 @@
 use crate::{CoreResult, SchemaContainerExt, json_rpc::types::*, parse_schema_multi};
-use schema_connector::{ConnectorError, SchemaConnector};
+use schema_connector::{ConnectorError, SchemaConnector, SchemaFilter};
 use tracing_futures::Instrument;
 
 /// Command to bring the local database in sync with the prisma schema, without
@@ -31,7 +31,8 @@ pub async fn schema_push(input: SchemaPushInput, connector: &mut dyn SchemaConne
         .schema_from_database(namespaces)
         .instrument(tracing::info_span!("Calculate from database"))
         .await?;
-    let database_migration = dialect.diff(from, to);
+    // TODO:(schema-filter) get filter from prisma config
+    let database_migration = dialect.diff(from, to, &SchemaFilter::default());
 
     tracing::debug!(migration = dialect.migration_summary(&database_migration).as_str());
 

schema-engine/connectors/mongodb-schema-connector/src/lib.rs

@@ -54,7 +54,7 @@ impl MongoDbSchemaConnector {
 }
 
 impl SchemaDialect for MongoDbSchemaDialect {
-    fn diff(&self, from: DatabaseSchema, to: DatabaseSchema) -> Migration {
+    fn diff(&self, from: DatabaseSchema, to: DatabaseSchema, _filter: &SchemaFilter) -> Migration {
         let from: Box<MongoSchema> = from.downcast();
         let to: Box<MongoSchema> = to.downcast();
         Migration::new(differ::diff(from, to))

schema-engine/connectors/mongodb-schema-connector/tests/migrations/test_api.rs

@@ -3,7 +3,7 @@ use enumflags2::BitFlags;
 use futures::TryStreamExt;
 use mongodb_schema_connector::MongoDbSchemaConnector;
 use psl::{parser_database::SourceFile, PreviewFeature};
-use schema_connector::{ConnectorParams, SchemaConnector};
+use schema_connector::{ConnectorParams, SchemaConnector, SchemaFilter};
 use std::{
     collections::BTreeMap,
     fmt::Write as _,
@@ -184,7 +184,7 @@ pub(crate) fn test_scenario(scenario_name: &str) {
         let to = dialect
             .schema_from_datamodel(vec![("schema.prisma".to_string(), schema.clone())])
             .unwrap();
-        let migration = dialect.diff(from, to);
+        let migration = dialect.diff(from, to, &SchemaFilter::default());
 
         connector.apply_migration(&migration).await.unwrap();
 
@@ -222,7 +222,7 @@ Snapshot comparison failed. Run the test again with UPDATE_EXPECT=1 in the envir
         let to = dialect
             .schema_from_datamodel(vec![("schema.prisma".to_string(), schema.clone())])
             .unwrap();
-        let migration = dialect.diff(from, to);
+        let migration = dialect.diff(from, to, &SchemaFilter::default());
 
         assert!(
             dialect.migration_is_empty(&migration),

schema-engine/connectors/schema-connector/src/filter.rs

@@ -1,14 +1,38 @@
 /// Configuration of entities in the schema/database to be included or excluded from an operation.
 #[derive(Debug, Default)]
 pub struct SchemaFilter {
-    /// Tables that shall be considered 'externally" managed. As per prisma.config.ts > tables.external.
+    /// Tables that shall be considered "externally" managed. As per prisma.config.ts > tables.external.
+    /// Prisma will not consider those tables during diffing operations, migration creation, or introspection.
+    /// They are still available for querying at runtime.
     pub external_tables: Vec<String>,
 }
 
+impl SchemaFilter {
+    /// Check if the given table name is in the list of external tables.
+    /// `external_tables` can contain fully qualified table names with namespace
+    /// (e.g. "auth.user") or just the table name.
+    pub fn is_table_external(&self, namespace: Option<&str>, table_name: &str) -> bool {
+        if let Some(namespace) = namespace {
+            self.external_tables.contains(&format!("{namespace}.{table_name}"))
+                || self.external_tables.contains(&table_name.to_string())
+        } else {
+            self.external_tables.contains(&table_name.to_string())
+        }
+    }
+}
+
 impl From<json_rpc::types::SchemaFilter> for SchemaFilter {
     fn from(filter: json_rpc::types::SchemaFilter) -> Self {
         Self {
             external_tables: filter.external_tables,
         }
     }
 }
+
+impl From<Option<json_rpc::types::SchemaFilter>> for SchemaFilter {
+    fn from(filter: Option<json_rpc::types::SchemaFilter>) -> Self {
+        Self {
+            external_tables: filter.map(|f| f.external_tables).unwrap_or_default(),
+        }
+    }
+}