feat: return a can_assume_strict_equality for in-memory joins (#5785)

[TML-1868](https://linear.app/prisma-company/issue/TML-1868/fix-mysql-bigint-relation-issue)

/prisma-branch fix/strict-equality-joins

Client PR: prisma/prisma#29251

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Joins now carry a flag to enable stricter equality assumptions when
supported, improving cross-database join handling.
* Validation expressions include richer context for clearer error
reporting.

* **Chores**
* Connector implementations updated to advertise support for
strict-equality joins where applicable, enabling optimized query
generation.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: jacek-prisma

psl/psl-core/src/builtin_connectors/cockroach_datamodel_connector.rs

@@ -355,6 +355,10 @@ impl Connector for CockroachDatamodelConnector {
             None => self.parse_json_bytes(str, Some(NativeTypeInstance::new::<CockroachType>(BYTES_DEFAULT))),
         }
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        true
+    }
 }
 
 /// An `@default(sequence())` function.

psl/psl-core/src/builtin_connectors/mongodb.rs

@@ -167,4 +167,8 @@ impl Connector for MongoDbDatamodelConnector {
     fn flavour(&self) -> Flavour {
         Flavour::Mongo
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        true
+    }
 }

psl/psl-core/src/builtin_connectors/mssql_datamodel_connector.rs

@@ -300,6 +300,10 @@ impl Connector for MsSqlDatamodelConnector {
     fn does_manage_udts(&self) -> bool {
         true
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        true
+    }
 }
 
 /// A collection of types stored outside of the row to the heap, having

psl/psl-core/src/builtin_connectors/postgres_datamodel_connector.rs

@@ -659,6 +659,10 @@ impl Connector for PostgresDatamodelConnector {
             None => self.parse_json_bytes(str, Some(NativeTypeInstance::new(PostgresType::Known(BYTES_DEFAULT)))),
         }
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        true
+    }
 }
 
 fn allowed_index_operator_classes(algo: IndexAlgorithm, field: walkers::ScalarFieldWalker<'_>) -> Vec<OperatorClass> {

psl/psl-core/src/builtin_connectors/sqlite_datamodel_connector.rs

@@ -126,4 +126,8 @@ impl Connector for SqliteDatamodelConnector {
     fn flavour(&self) -> Flavour {
         Flavour::Sqlite
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        true
+    }
 }

psl/psl-core/src/datamodel_connector.rs

@@ -309,6 +309,10 @@ pub trait Connector: Send + Sync {
     fn does_manage_udts(&self) -> bool {
         false
     }
+
+    fn can_assume_strict_equality_in_joins(&self) -> bool {
+        false
+    }
 }
 
 #[derive(Copy, Clone, Debug, PartialEq)]

query-compiler/query-compiler/src/expression.rs

@@ -83,9 +83,11 @@ pub enum Expression {
     Required(Box<Expression>),
 
     /// Application-level join.
+    #[serde(rename_all = "camelCase")]
     Join {
         parent: Box<Expression>,
         children: Vec<JoinExpression>,
+        can_assume_strict_equality: bool,
     },
 
     /// Get a field from a record or records. If the argument is a list of records,
@@ -103,6 +105,7 @@ pub enum Expression {
     },
 
     /// Validates the expression according to the data rule and throws an error if it doesn't match.
+    #[serde(rename_all = "camelCase")]
     Validate {
         expr: Box<Expression>,
         rules: Vec<DataRule>,
@@ -194,7 +197,7 @@ impl Expression {
             Expression::Required(expr) => {
                 expr.simplify();
             }
-            Expression::Join { parent, children } => {
+            Expression::Join { parent, children, .. } => {
                 parent.simplify();
                 children.iter_mut().for_each(|child| child.child.simplify());
             }

query-compiler/query-compiler/src/expression/format.rs

@@ -59,7 +59,7 @@ where
             Expression::Concat(vec) => self.function("concat", vec),
             Expression::Unique(expression) => self.unary_function("unique", expression),
             Expression::Required(expression) => self.unary_function("required", expression),
-            Expression::Join { parent, children } => self.join(parent, children),
+            Expression::Join { parent, children, .. } => self.join(parent, children),
             Expression::MapField { field, records } => self.map_field(field, records),
             Expression::Transaction(expression) => self.transaction(expression),
             Expression::DataMap { expr, structure, enums } => self.data_map(expr, structure, enums),

query-compiler/query-compiler/src/translate/query/read.rs

@@ -143,6 +143,9 @@ pub(super) fn add_inmemory_join(
         })
         .collect();
 
+    let can_assume_strict_equality = nested
+        .iter()
+        .all(|nested| nested.model().dm.schema.connector.can_assume_strict_equality_in_joins());
     let join_expressions = nested
         .into_iter()
         .filter_map(|nested| match nested {
@@ -197,6 +200,7 @@ pub(super) fn add_inmemory_join(
                     name: binding::join_parent(),
                 }),
                 children: join_expressions,
+                can_assume_strict_equality,
             }),
         }),
     })