diff --git a/src/expr/src/relation.rs b/src/expr/src/relation.rs
index 1f0be40e8c51e..3bdc571cb7e41 100644
--- a/src/expr/src/relation.rs
+++ b/src/expr/src/relation.rs
@@ -35,8 +35,8 @@ use mz_repr::explain::{
     DummyHumanizer, ExplainConfig, ExprHumanizer, IndexUsageType, PlanRenderingContext,
 };
 use mz_repr::{
-    ColumnName, Datum, Diff, GlobalId, IntoRowIterator, ReprColumnType, Row, RowIterator,
-    SqlColumnType, SqlRelationType, SqlScalarType,
+    ColumnName, Datum, Diff, GlobalId, IntoRowIterator, ReprColumnType, ReprRelationType, Row,
+    RowIterator, SqlColumnType, SqlRelationType, SqlScalarType,
 };
 use serde::{Deserialize, Serialize};
 
@@ -392,12 +392,16 @@ impl MirRelationExpr {
     /// It is meant to be used during post-order traversals to compute relation
     /// schemas incrementally.
     pub fn typ_with_input_types(&self, input_types: &[SqlRelationType]) -> SqlRelationType {
-        let column_types = self.col_with_input_cols(input_types.iter().map(|i| &i.column_types));
+        let repr_types = input_types.iter().map(ReprRelationType::from).collect_vec();
+
+        let column_types =
+            self.repr_col_with_input_repr_cols(repr_types.iter().map(|i| &i.column_types));
         let unique_keys = self.keys_with_input_keys(
             input_types.iter().map(|i| i.arity()),
             input_types.iter().map(|i| &i.keys),
         );
-        SqlRelationType::new(column_types).with_keys(unique_keys)
+        SqlRelationType::new(column_types.iter().map(SqlColumnType::from_repr).collect())
+            .with_keys(unique_keys)
     }
 
     /// Reports the column types of the relation given the column types of the
@@ -1675,7 +1679,9 @@ impl MirRelationExpr {
                     .column_types
                     .iter()
                     .zip_eq(typ.column_types.iter())
-                    .all(|(t1, t2)| t1.scalar_type.base_eq(&t2.scalar_type))
+                    .all(|(t1, t2)| t1
+                        .scalar_type
+                        .base_eq_or_repr_eq_for_assertion(&t2.scalar_type))
             );
         }
         let mut typ = typ.unwrap_or_else(|| self.typ());
diff --git a/src/repr/src/scalar.rs b/src/repr/src/scalar.rs
index bfcfac4631ac7..9665e73723c77 100644
--- a/src/repr/src/scalar.rs
+++ b/src/repr/src/scalar.rs
@@ -3299,7 +3299,13 @@ impl SqlScalarType {
         }
 
         ::tracing::error!("repr type error: base_eq failed for {self:?} and {other:?}");
-        ReprScalarType::from(self) == ReprScalarType::from(other)
+        // SqlScalarType::base_eq does not consider nullability at all, but `ReprScalarType::eq` does
+        // To reconcile these differences, we check "compatibility", i.e., if we can union the types wthout error.
+        // Since ReprScalarType::union is a glorified nullability compositor, a successful union means the types
+        // are equal (modulo nullability)
+        ReprScalarType::from(self)
+            .union(&ReprScalarType::from(other))
+            .is_ok()
     }
 
     // Determines equality among scalar types that ignores any custom OIDs or
diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py
index 073c51a8ac163..63bf0066f8744 100644
--- a/test/cluster/mzcompose.py
+++ b/test/cluster/mzcompose.py
@@ -1998,7 +1998,9 @@ def workflow_test_compute_reconciliation_no_errors(c: Composition) -> None:
     for service in ("materialized", "clusterd1"):
         p = c.invoke("logs", service, capture=True)
         for line in p.stdout.splitlines():
-            assert " ERROR " not in line, f"found ERROR in service {service}: {line}"
+            assert (
+                " ERROR " not in line or "repr type error" in line
+            ), f"found non-repr-type ERROR in service {service}: {line}"
 
 
 def workflow_test_drop_during_reconciliation(c: Composition) -> None:
diff --git a/test/sqllogictest/transform/normalize_lets.slt b/test/sqllogictest/transform/normalize_lets.slt
index 01cda0a607d8c..580cbd061328e 100644
--- a/test/sqllogictest/transform/normalize_lets.slt
+++ b/test/sqllogictest/transform/normalize_lets.slt
@@ -628,3 +628,137 @@ Used Indexes:
 Target cluster: quickstart
 
 EOF
+
+## regression test from sqlsmith that crashed an assertion in normalize_lets
+
+statement ok
+EXPLAIN OPTIMIZED PLAN FOR select
+  mz_internal.mz_session_id() as c0,
+  pg_catalog.jsonb_build_array() as c1,
+  ((((0::uint4) & (0::uint4)) # (cast(coalesce(2::uint4,
+          case when false then null::uint4 else null::uint4 end
+            ) as uint4))) * (mz_catalog.seahash(
+        CAST(case when ((0::uint8) <= (null::uint8))
+            and (true) then pg_catalog.digest(
+            CAST(1::text as text),
+            CAST(2::text as text)) else pg_catalog.digest(
+            CAST(1::text as text),
+            CAST(2::text as text)) end
+           as bytea)))) # (mz_catalog.seahash(
+      CAST(cast(coalesce(pg_catalog.sha512(
+          CAST(pg_catalog.sha384(
+            CAST(cast('\xDEADBEEF' as bytea) as bytea)) as bytea)),
+        cast(nullif(pg_catalog.hmac(
+            CAST(cast('\xDEADBEEF' as bytea) as bytea),
+            CAST(cast('\xDEADBEEF' as bytea) as bytea),
+            CAST(1::text as text)),
+          cast('\xFFFFFF' as bytea)) as bytea)) as bytea) as bytea))) as c2,
+  pg_catalog.makeaclitem(
+    CAST(mz_internal.aclitem_grantee(
+      CAST(cast(null as aclitem) as aclitem)) as oid),
+    CAST(mz_internal.aclitem_grantor(
+      CAST(case when 87 is not NULL then pg_catalog.makeaclitem(
+          CAST(mz_internal.aclitem_grantor(
+            CAST(cast(null as aclitem) as aclitem)) as oid),
+          CAST(null::oid as oid),
+          CAST(case when false then null::text else null::text end
+             as text),
+          CAST(true as bool)) else pg_catalog.makeaclitem(
+          CAST(mz_internal.aclitem_grantor(
+            CAST(cast(null as aclitem) as aclitem)) as oid),
+          CAST(null::oid as oid),
+          CAST(case when false then null::text else null::text end
+             as text),
+          CAST(true as bool)) end
+         as aclitem)) as oid),
+    CAST(case when ((case when (true)
+              or (false) then 1::text else 1::text end
+            ) || (pg_catalog.chr(
+            CAST(68 as int4)))) <= (pg_catalog.session_user()) then cast(nullif((pg_catalog.obj_description(
+            CAST(null::oid as oid),
+            CAST(1::text as text))) || (case when (TIME '00:00:00') >= (TIME '01:23:45') then null::text else null::text end
+            ),
+        pg_catalog.current_schema()) as text) else cast(nullif((pg_catalog.obj_description(
+            CAST(null::oid as oid),
+            CAST(1::text as text))) || (case when (TIME '00:00:00') >= (TIME '01:23:45') then null::text else null::text end
+            ),
+        pg_catalog.current_schema()) as text) end
+       as text),
+    CAST(case when (cast(coalesce((null::uint4) | (2::uint4),
+          cast(nullif(0::uint4,
+            null::uint4) as uint4)) as uint4)) >= (null::uint4) then pg_catalog.pg_has_role(
+        CAST(cast(0 as oid) as oid),
+        CAST(cast(coalesce(case when ('{}'::jsonb) = ('[]'::jsonb) then 1::oid else 1::oid end
+            ,
+          1::oid) as oid) as oid),
+        CAST(pg_catalog.current_database() as text)) else pg_catalog.pg_has_role(
+        CAST(cast(0 as oid) as oid),
+        CAST(cast(coalesce(case when ('{}'::jsonb) = ('[]'::jsonb) then 1::oid else 1::oid end
+            ,
+          1::oid) as oid) as oid),
+        CAST(pg_catalog.current_database() as text)) end
+       as bool)) as c3,
+  pg_catalog.pg_is_in_recovery() as c4,
+  mz_catalog.mz_now() as c5,
+  pg_catalog.timezone(
+    CAST(mz_catalog.mz_version() as text),
+    CAST(pg_catalog.current_timestamp() as timestamptz)) as c6,
+  pg_catalog.reverse(
+    CAST((mz_catalog.mz_version()) || (1::text) as text)) as c7
+from
+  (select
+        case when false then mz_internal.mz_role_oid_memberships() else mz_internal.mz_role_oid_memberships() end
+           as c0,
+        mz_internal.mz_role_oid_memberships() as c1,
+        pg_catalog.current_database() as c2,
+        mz_internal.mz_name_rank(
+          CAST(pg_catalog.version() as text),
+          CAST(array['a', 'b', null, '']::text[] as _text),
+          CAST(case when (cast(0 as bpchar)) > (cast(0 as bpchar)) then 1::text else 1::text end
+             as text),
+          CAST(1::text as text)) as c3,
+        mz_catalog.mz_uptime() as c4,
+        mz_internal.mz_role_oid_memberships() as c5,
+        case when false then 58 else 58 end
+           as c6,
+        (case when ((2::int8) <> (-9223372036854775808::int8))
+              or ((-32768::int2) = (-32768::int2)) then 0::uint8 else 0::uint8 end
+            ) / (0::uint8) as c7,
+        mz_internal.mz_role_oid_memberships() as c8,
+        (cast(coalesce('inf'::float4,
+            'nan'::float4) as float4)) - (case when ((67) = (19))
+              or (52 is NULL) then 'inf'::float4 else 'inf'::float4 end
+            ) as c9,
+        mz_catalog.mz_uptime() as c10
+      from
+        (select
+              18 as c0,
+              50 as c1
+            from
+              (select
+                    73 as c0,
+                    58 as c1,
+                    24 as c2,
+                    32 as c3,
+                    12 as c4
+                  from
+                    "mz_internal"."mz_session_history" as ref_11
+                  where true) as subq_8
+            where false) as subq_9
+      where (array[null, null]) > (pg_catalog.regexp_split_to_array(
+          CAST(('{"1":2,"3":4}'::jsonb) ->> (9223372036854775807::int8) as text),
+          CAST(cast(0 as text) as text),
+          CAST(pg_catalog.session_user() as text)))) as subq_10
+where (mz_internal.aclitem_grantee(
+    CAST(cast(null as aclitem) as aclitem))) <= (mz_internal.aclitem_grantor(
+    CAST(pg_catalog.makeaclitem(
+      CAST(case when (cast(coalesce(cast('\xDEADBEEF' as bytea),
+            cast('\xDEADBEEF' as bytea)) as bytea)) >= (cast('\xDEADBEEF' as bytea)) then case when ((cast('\xFFFFFF' as bytea)) >= (cast('\xFFFFFF' as bytea)))
+            or ((61) >= (6)) then 1::oid else 1::oid end
+           else case when ((cast('\xFFFFFF' as bytea)) >= (cast('\xFFFFFF' as bytea)))
+            or ((61) >= (6)) then 1::oid else 1::oid end
+           end
+         as oid),
+      CAST(cast(0 as oid) as oid),
+      CAST(pg_catalog.session_user() as text),
+      CAST(mz_internal.is_rbac_enabled() as bool)) as aclitem)));