Don't drop rows for a non-null field under a nullable parent

build.sbt

@@ -36,7 +36,7 @@ ThisBuild / scalaVersion := Scala2
 ThisBuild / crossScalaVersions := Seq(Scala2, Scala3)
 ThisBuild / tlJdkRelease := Some(11)
 
-ThisBuild / tlBaseVersion := "0.29"
+ThisBuild / tlBaseVersion := "0.30"
 ThisBuild / startYear := Some(2019)
 ThisBuild / licenses := Seq(License.Apache2)
 ThisBuild / developers := List(

modules/doobie-mssql/src/test/scala/DoobieMSSqlSuites.scala

@@ -191,6 +191,10 @@ final class NestedEffectsSuite extends DoobieMSSqlDatabaseSuite with SqlNestedEf
     }
 }
 
+final class NullableParentSuite extends DoobieMSSqlDatabaseSuite with SqlNullableParentSuite {
+  lazy val mapping = new DoobieMSSqlTestMapping(transactor) with SqlNullableParentMapping[IO]
+}
+
 final class NullOrderingSuite extends DoobieMSSqlDatabaseSuite with SqlNullOrderingSuite {
   lazy val mapping = new DoobieMSSqlTestMapping(transactor) with SqlNullOrderingMapping[IO]
 }

modules/doobie-oracle/src/test/scala/DoobieOracleSuites.scala

@@ -197,6 +197,10 @@ final class NestedEffectsSuite extends DoobieOracleDatabaseSuite with SqlNestedE
     }
 }
 
+final class NullableParentSuite extends DoobieOracleDatabaseSuite with SqlNullableParentSuite {
+  lazy val mapping = new DoobieOracleTestMapping(transactor) with SqlNullableParentMapping[IO]
+}
+
 final class NullOrderingSuite extends DoobieOracleDatabaseSuite with SqlNullOrderingSuite {
   lazy val mapping = new DoobieOracleTestMapping(transactor) with SqlNullOrderingMapping[IO]
 }

modules/doobie-pg/src/test/scala/DoobiePgSuites.scala

@@ -193,6 +193,10 @@ final class NestedEffectsSuite extends DoobiePgDatabaseSuite with SqlNestedEffec
     }
 }
 
+final class NullableParentSuite extends DoobiePgDatabaseSuite with SqlNullableParentSuite {
+  lazy val mapping = new DoobiePgTestMapping(transactor) with SqlNullableParentMapping[IO]
+}
+
 final class NullOrderingSuite extends DoobiePgDatabaseSuite with SqlNullOrderingSuite {
   lazy val mapping = new DoobiePgTestMapping(transactor) with SqlNullOrderingMapping[IO]
 }

modules/skunk/js-jvm/src/test/scala/SkunkSuites.scala

@@ -198,6 +198,10 @@ final class NestedEffectsSuite extends SkunkDatabaseSuite with SqlNestedEffectsS
     }
 }
 
+final class NullableParentSuite extends SkunkDatabaseSuite with SqlNullableParentSuite {
+  lazy val mapping = new SkunkTestMapping(pool) with SqlNullableParentMapping[IO]
+}
+
 final class NullOrderingSuite extends SkunkDatabaseSuite with SqlNullOrderingSuite {
   lazy val mapping = new SkunkTestMapping(pool) with SqlNullOrderingMapping[IO]
 }

modules/sql-core/src/main/scala/SqlMapping.scala

@@ -1385,7 +1385,19 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
     trait Laterality {
       def toFragment: Fragment
       def joinToFragment(join: SqlJoin, subquery: SubqueryRef): Aliased[Fragment]
-      def joinPredicates(join: SqlJoin): List[Predicate]
+
+      /**
+       * Distribute the conditions of `join` between the join itself and the enclosing select.
+       *
+       * Yields a possibly rewritten join, together with the predicates which must be added to
+       * the enclosing select's WHERE clause for the join to have its intended semantics.
+       *
+       * Lateralities which render the join with an `ON` clause need neither, and so yield the
+       * join unchanged and no predicates.
+       */
+      def distributeJoinConditions(
+          join: SqlJoin,
+          subquery: SubqueryRef): (SqlJoin, List[Predicate])
     }
     object Laterality {
       def apply(lateral: Boolean, inner: Boolean): Laterality =
@@ -1395,21 +1407,95 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
         def toFragment: Fragment = Fragments.empty
         def joinToFragment(join: SqlJoin, subquery: SubqueryRef): Aliased[Fragment] =
           join.toFragmentWithoutLaterality
-        def joinPredicates(join: SqlJoin): List[Predicate] = Nil
+        def distributeJoinConditions(
+            join: SqlJoin,
+            subquery: SubqueryRef): (SqlJoin, List[Predicate]) = (join, Nil)
       }
       case object Lateral extends Laterality {
         def toFragment: Fragment = Fragments.const("LATERAL ")
         def joinToFragment(join: SqlJoin, subquery: SubqueryRef): Aliased[Fragment] =
           join.toFragmentWithoutLaterality
-        def joinPredicates(join: SqlJoin): List[Predicate] = Nil
+        def distributeJoinConditions(
+            join: SqlJoin,
+            subquery: SubqueryRef): (SqlJoin, List[Predicate]) = (join, Nil)
       }
       case class Apply(inner: Boolean) extends Laterality {
         def toFragment: Fragment =
           Fragments.const(if (inner) "CROSS " else "OUTER ") |+| Fragments.const("APPLY ")
         def joinToFragment(join: SqlJoin, subquery: SubqueryRef): Aliased[Fragment] =
           subquery.toDefFragment
-        def joinPredicates(join: SqlJoin): List[Predicate] =
+
+        /**
+         * An `APPLY` join is rendered without an `ON` clause, so its conditions have to be
+         * expressed elsewhere. For `CROSS APPLY` they are lifted into the enclosing select's
+         * WHERE clause, which for an inner join is equivalent to an `ON` clause.
+         *
+         * For `OUTER APPLY` lifting them would discard precisely the null padded rows the outer
+         * join produced, the WHERE clause being applied after the join. Its subquery is instead
+         * wrapped in a correlating select, which is legal because `APPLY` is lateral and is
+         * what makes `OUTER APPLY` equivalent to `LEFT JOIN LATERAL`. A subquery that can't be
+         * correlated has no correct rendering, so `correlate` failing is treated as a bug.
+         */
+        def distributeJoinConditions(
+            join: SqlJoin,
+            subquery: SubqueryRef): (SqlJoin, List[Predicate]) =
+          if (inner || join.isPredicate) (join, liftedPredicates(join))
+          else
+            // Every OUTER APPLY reachable here can be correlated; a None is a bug, not a query.
+            correlate(join, subquery)
+              .map((_, Nil))
+              .getOrElse(throw new SqlMappingException(
+                s"OUTER APPLY subquery '${subquery.name}' could not be correlated"))
+
+        /**
+         * The join's conditions expressed as predicates of the enclosing select.
+         */
+        private def liftedPredicates(join: SqlJoin): List[Predicate] =
           if (!join.isPredicate) join.on.map { case (p, c) => Eql(p.toTerm, c.toTerm) } else Nil
+
+        /**
+         * Yields a copy of `join` with its conditions moved inside its subquery.
+         *
+         * The subquery is wrapped in a select which applies the conditions to its result, so
+         * that any LIMIT, OFFSET or DISTINCT within is evaluated first, as it would be were
+         * they in the ON clause of a `LEFT JOIN LATERAL`, and so that only the subquery's
+         * exposed columns are in scope where they are applied — a table of the enclosing select
+         * can otherwise be shadowed by a same named table within.
+         *
+         * Yields `None` only for a subquery shape it doesn't handle; no `OUTER APPLY` produces
+         * one, so the caller treats a `None` as a bug rather than lifting. A subquery already
+         * such a wrapper is yielded unchanged, so re-nesting doesn't correlate it twice.
+         */
+        private def correlate(join: SqlJoin, subquery: SubqueryRef): Option[SqlJoin] =
+          subquery.subquery match {
+            case sel: SqlSelect if sel.withs.isEmpty =>
+              sel.table match {
+                case wrapped: SubqueryRef if wrapped.correlated && sel.wheres.nonEmpty =>
+                  Some(join)
+                case table: TableRef =>
+                  val exposed =
+                    join.on.traverse {
+                      case (p, c) => sel.cols.find(_ == c.subst(subquery, table)).map((p, _))
+                    }
+                  exposed.map { exposed0 =>
+                    val wrapper =
+                      sel.toSubquery(correlationName(subquery), NotLateral, correlated = true)
+                    val wheres =
+                      exposed0.map {
+                        case (p, c) => Eql(p.toTerm, c.derive(wrapper.table).toTerm)
+                      }
+                    join.copy(child = subquery.copy(subquery = wrapper.copy(wheres = wheres)))
+                  }
+                case _ => None
+              }
+            case _ => None
+          }
+
+        /**
+         * The alias given to the correlating select `correlate` wraps around `subquery`. Only
+         * cosmetic: the wrapper is recognised by its `correlated` flag, not by this name.
+         */
+        private def correlationName(subquery: SubqueryRef): String = subquery.name + "_corr"
       }
     }
 
@@ -1420,7 +1506,8 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
         context: Context,
         name: String,
         subquery: SqlQuery,
-        laterality: Laterality)
+        laterality: Laterality,
+        correlated: Boolean = false)
         extends TableExpr {
       def owns(col: SqlColumn): Boolean = col.owner.isSameOwner(this) || subquery.owns(col)
       def contains(other: ColumnOwner): Boolean = isSameOwner(other) || subquery.contains(other)
@@ -2377,18 +2464,35 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
         parentTableForType(parentContext).flatMap { parentTable =>
           val inner = !context.tpe.isNullable && !context.tpe.isList
 
+          // A nested select may be flattened into this select's join chain only if doing so
+          // can't cost this select rows.
+          //
+          // Flattening moves the nested select's joins into this chain, below the join which
+          // attaches it. Where that join is LEFT and finds no match it still yields a row
+          // padded with NULLs, and an INNER join below discards it — losing a row which
+          // should have been returned with the nested field simply null or empty. Every join
+          // in `nested.joins` is downstream of that select's own table, so all of them land
+          // below the attaching join and see its padded columns, directly or transitively.
+          // Flattening is safe, then, if this join is INNER or the nested select has no
+          // INNER joins.
+          def mergePreservesRows(nested: SqlSelect): Boolean =
+            inner || !nested.joins.exists(_.inner)
+
           def mkJoins(joins: List[Join]): SqlSelect = {
             def mkSubquery(
                 multiTable: Boolean,
                 nested: SqlSelect,
                 joinCols: List[SqlColumn],
                 suffix: String): SqlSelect = {
               def isMergeable: Boolean =
-                !multiTable && !nested.joins.exists(_.isPredicate) && nested
-                  .wheres
-                  .isEmpty && nested.orders.isEmpty && nested.offset.isEmpty && nested
-                  .limit
-                  .isEmpty && !nested.isDistinct
+                mergePreservesRows(nested) &&
+                  !multiTable &&
+                  !nested.joins.exists(_.isPredicate) &&
+                  nested.wheres.isEmpty &&
+                  nested.orders.isEmpty &&
+                  nested.offset.isEmpty &&
+                  nested.limit.isEmpty &&
+                  !nested.isDistinct
 
               if (isMergeable) nested
               else {
@@ -2410,22 +2514,23 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
                 cols: List[SqlColumn],
                 wheres: List[Predicate],
                 joins: List[SqlJoin]): SqlSelect = {
-              val extraWheres =
-                if (!outer) Nil
-                else
-                  joins.flatMap { join =>
-                    join.child match {
-                      case sq: SubqueryRef => sq.laterality.joinPredicates(join)
-                      case _ => Nil
-                    }
+              val distributed =
+                joins.map { join =>
+                  join.child match {
+                    case sq: SubqueryRef if outer =>
+                      sq.laterality.distributeJoinConditions(join, sq)
+                    case _ => (join, Nil)
                   }
+                }
+              val joins0 = distributed.map(_._1)
+              val extraWheres = distributed.flatMap(_._2)
 
               SqlSelect(
                 context = parentContext,
                 withs = withs,
                 table = table,
                 cols = cols,
-                joins = joins,
+                joins = joins0,
                 wheres = wheres ++ extraWheres,
                 orders = Nil,
                 offset = None,
@@ -2458,6 +2563,10 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
                   true)
                 val assocJoin = lastJoin.toSqlJoin(lastJoinParentTable, assocTable, inner)
 
+                // This join is on the key between a table and itself, so it always matches and
+                // an outer join is as good as an inner one. `assocJoin` carries the join's real
+                // innerness, and it's that which decides how a subquery beneath it distributes
+                // its conditions.
                 val finalJoin =
                   SqlJoin(
                     assocTable,
@@ -3143,8 +3252,11 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
       def toSubquery(name: String): Result[SqlSelect] =
         toSubquery(name, Laterality.NotLateral).success
 
-      def toSubquery(name: String, lateral: Laterality): SqlSelect = {
-        val ref = SubqueryRef(context, name, this, lateral)
+      def toSubquery(
+          name: String,
+          lateral: Laterality,
+          correlated: Boolean = false): SqlSelect = {
+        val ref = SubqueryRef(context, name, this, lateral, correlated)
         SqlSelect(
           context,
           Nil,
@@ -3165,7 +3277,7 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
        */
       def subqueryToWithQuery: SqlSelect = {
         table match {
-          case SubqueryRef(_, name, sq, _) =>
+          case SubqueryRef(_, name, sq, _, _) =>
             val with0 = WithRef(context, name + "_base", sq)
             val ref = TableExpr.DerivedTableRef(context, Some(name), with0, true)
             copy(withs = with0 :: withs, table = ref)
@@ -3465,7 +3577,7 @@ trait SqlMappingLike[F[_]] extends CirceMappingLike[F] with SqlModule[F] { self
        */
       def isPredicate: Boolean =
         child match {
-          case SubqueryRef(_, _, sq: SqlSelect, _) => sq.predicate
+          case SubqueryRef(_, _, sq: SqlSelect, _, _) => sq.predicate
           case _ => false
         }