OXY-17: Add IN / NOT IN value-list predicates to the sql query DSL

docs/docs/sql/queries.md

@@ -124,6 +124,7 @@ input makes it a `QueryO`/`Query`. Pass `debug = true` (`@compile(debug = true)`
 | `join[A] if <cond>` / `leftJoin[A] if <cond>` | inner / left join (`leftJoin` yields `Option[A]`) |
 | `where if <cond>` | filter |
 | `s.like(p)` / `s.ilike(p)` / `s.notLike(p)` / `s.notILike(p)` | string pattern match (see below) |
+| `col.in(coll)` / `col.notIn(coll)` | `col = ANY(?)` / `col <> ALL(?)` over a runtime `Seq` / `Set` (static SQL, one array bind) |
 | `orderBy(a.field.asc, …)`, `limit(n)`, `offset(n)` | ordering / paging |
 | `Q.insert[A]` / `Q.update[A]` / `Q.delete[A]` | begin an insert / update / delete |
 | `set(_.field := value)` | assignment in an update |
@@ -240,6 +241,36 @@ Notes:
 - Currently supported as the root `FROM` source; `JOIN UNNEST(?)` (unnest as a non-root join item)
   is not yet supported.
 
+### `IN` / `NOT IN` (value lists)
+
+Filter a single column against a runtime collection with `col.in(coll)` / `col.notIn(coll)`. The
+collection is a `Seq[A]` supplied as a normal `input`. The generated SQL is **100% static** — the
+whole collection binds as a single Postgres array parameter, so the SQL text is identical regardless
+of the list size: `col = ANY(?)` for `in`, `col <> ALL(?)` for `notIn`:
+
+```scala
+@compile
+val peopleByIds: QueryIO[Seq[UUID], Person] =
+  for {
+    ids <- input[Seq[UUID]]
+    p   <- select[Person]
+    _   <- where if p.id.in(ids)   // -> WHERE p.id = ANY(?)
+  } yield p
+```
+
+- Composes with other predicates: `where if p.groupId == groupId && p.id.in(ids)`.
+- Empty list is handled natively by `ANY` / `ALL` (never the illegal `IN ()`): `col = ANY('{}')` is
+  `FALSE` (matches nothing) and `col <> ALL('{}')` is `TRUE` (matches everything) — exactly the
+  truth table of SQL `IN ()` / `NOT IN ()`.
+- `= ANY` / `<> ALL` preserve the same three-valued logic as `IN` / `NOT IN`, including the classic
+  `NOT IN`-with-`NULL` footgun (a `NULL` element makes `NOT IN` yield no rows). Here it is unreachable:
+  array columns are non-nullable and the value-list `input` must be non-optional.
+- Because the SQL and the `?` count are static, the same prepared statement is reused across
+  executions and the JDBC per-statement parameter limit no longer applies to large lists. The
+  value-list must be a runtime `input` (not `const`).
+- This shares the array-bind machinery (`RowRepr.ArrayRepr` / `ArraySeqEncoder`) introduced for
+  array input under OXY-6 / OXY-18; `in` / `notIn` are effectively sugar over `= ANY(?)`.
+
 ## A real repo method
 
 ```scala

modules/sql/core/src/main/scala/oxygen/sql/generic/generation/DecoderBuilder.scala

@@ -17,6 +17,7 @@ final class DecoderBuilder {
         case (queryExpr: QueryExpr.InputVariableReferenceLike, Some(parentContext)) => convert.input(queryExpr, parentContext)
         case (queryExpr: QueryExpr.QueryVariableReferenceLike, _)                   => convert.query(queryExpr)
         case (_: QueryExpr.ArrayContains, _)                         => ParseResult.success(GeneratedResultDecoder.single(TypeclassExpr.RowRepr.boolean.resultDecoder, TypeRepr.of[Boolean]))
+        case (queryExpr: QueryExpr.InList, _)                        => ParseResult.error(queryExpr.fullTerm, "`in`/`notIn` is a predicate and can not be used as a returned/output value")
         case (queryExpr: QueryExpr.Binary, _)                        => convert.binary(queryExpr)
         case (queryExpr: QueryExpr.BuiltIn, _)                       => convert.builtIn(queryExpr)
         case (queryExpr: QueryExpr.Composite, _)                     => convert.composite(queryExpr, parentContext)

modules/sql/core/src/main/scala/oxygen/sql/generic/generation/FragmentBuilder.scala

@@ -90,6 +90,7 @@ final case class FragmentBuilder(inputs: List[InputPart])(using Quotes) {
         case (queryExpr: QueryExpr.InputVariableReferenceLike, Some(parentContext)) => queryExprToFragment.constOrInput(queryExpr, parentContext)
         case (queryExpr: QueryExpr.QueryVariableReferenceLike, _)                   => queryExprToFragment.query(queryExpr)
         case (queryExpr: QueryExpr.ArrayContains, _)                                => queryExprToFragment.arrayContains(queryExpr)
+        case (queryExpr: QueryExpr.InList, _)                                       => queryExprToFragment.inList(queryExpr)
         case (queryExpr: QueryExpr.Binary, _)                                       => queryExprToFragment.binary(queryExpr)
         case (queryExpr: QueryExpr.BuiltIn, _)                                      => queryExprToFragment.builtIn(queryExpr)
         case (queryExpr: QueryExpr.Composite, _)                                    => queryExprToFragment.composite(queryExpr, parentContext)
@@ -179,6 +180,72 @@ final case class FragmentBuilder(inputs: List[InputPart])(using Quotes) {
         )
       }
 
+    /**
+      * `col IN (coll)` / `col NOT IN (coll)` where the value-list is a runtime collection (OXY-17).
+      *
+      * The SQL emitted is 100% STATIC -- a single array bind parameter, never an N-placeholder list:
+      *   - `col IN (coll)`     -> `col = ANY(?)`
+      *   - `col NOT IN (coll)` -> `col <> ALL(?)`
+      *
+      * The whole collection binds as one `java.sql.Array` (reusing OXY-6/OXY-18's
+      * [[oxygen.sql.schema.RowRepr.ArrayRepr]] / `ArraySeqEncoder` / `unsafeWriteArray` machinery), so
+      * the `?` count is constant regardless of the runtime list size -- no sentinel token, no
+      * per-execution SQL rewrite. `= ANY` / `<> ALL` reproduce the exact 3-valued-logic truth table of
+      * SQL `IN` / `NOT IN`, including empty-collection handling (`= ANY('{}')` -> FALSE,
+      * `<> ALL('{}')` -> TRUE) and the classic `NOT IN`-with-NULL footgun (here unreachable: array
+      * columns are non-nullable and optional inputs are rejected below).
+      */
+    def inList(queryExpr: QueryExpr.InList)(using ParseContext, GenerationContext, Quotes): ParseResult[GeneratedFragment] = {
+      import oxygen.sql.schema.RowRepr as SRowRepr
+
+      val lhs: QueryExpr.QueryVariableReferenceLike = queryExpr.lhs
+      val rhs: QueryExpr.InputVariableReferenceLike = queryExpr.rhs
+      val notIn: Boolean = queryExpr.notIn
+
+      val elemTpe: TypeRepr = lhs.fullTerm.tpe.widen // single column element type
+      val collTpe: TypeRepr = rhs.outTpe // declared collection type, e.g. Seq[A]
+
+      for {
+        transform <- inputSymToInputRepr.get(rhs.queryRef.param.sym) match {
+          case Some(InputRepr.NonConst(inputTransformer, false)) =>
+            (inputTransformer >>> rhs) match {
+              case TermTransformer.Die            => ParseResult.error(rhs.fullTerm, "unexpected non-input transform for `in`/`notIn` value-list")
+              case t: TermTransformer.SimpleValid => ParseResult.success(t)
+            }
+          case Some(InputRepr.NonConst(_, true)) => ParseResult.error(rhs.fullTerm, "optional input is not supported as an `in`/`notIn` value-list")
+          case Some(InputRepr.Const(_))          => ParseResult.error(rhs.fullTerm, "const input is not supported as an `in`/`notIn` value-list; use `input[Seq[A]]`")
+          case None                              => ParseResult.error(rhs.rootIdent, "Not able to find in symMap?")
+        }
+
+        colFrag <- GenerationContext.updated(query = GenerationContext.Parens.Never) { queryExprToFragment.query(lhs) }
+
+        // Bind the entire collection as ONE `java.sql.Array` param via the column's own RowRepr,
+        // reusing the array machinery from OXY-6/OXY-18. `ArraySeq.untagged.from` adapts the runtime
+        // `Seq[A]` (order preserved, dups kept) without needing a `ClassTag`.
+        arrayEnc = {
+          type A
+          type Coll
+          given Type[A] = elemTpe.asTypeOf
+          given Type[Coll] = collTpe.asTypeOf
+          TypeclassExpr.InputEncoder {
+            '{
+              SRowRepr
+                .ArrayRepr[A](${ lhs.rowRepr.expr.asExprOf[SRowRepr[A]] })
+                .encoder
+                .contramap[Coll]((c: Coll) => scala.collection.immutable.ArraySeq.untagged.from(c.asInstanceOf[IterableOnce[A]]))
+            }
+          }
+        }
+        baseEnc = GeneratedInputEncoder.nonConst(arrayEnc, collTpe)
+        enc = transform match {
+          case TermTransformer.Id                   => baseEnc
+          case transform: TermTransformer.Transform => baseEnc.contramap(transform)
+        }
+
+        opSql = if notIn then " <> ALL(?)" else " = ANY(?)"
+      } yield GeneratedFragment.both(colFrag.generatedSql ++ GeneratedSql.single(opSql), enc)
+    }
+
     def binary(queryExpr: QueryExpr.Binary)(using ParseContext, GenerationContext, Quotes): ParseResult[GeneratedFragment] =
       queryExpr match {
         // standard and/or

modules/sql/core/src/main/scala/oxygen/sql/generic/generation/GeneratedFragment.scala

@@ -42,7 +42,10 @@ object GeneratedFragment {
     fragment.getOrElse(GeneratedFragment.empty)
 
   def flatten[S[_]: SeqOps](all: S[GeneratedFragment]): GeneratedFragment =
-    GeneratedFragment(GeneratedSql.flatten(all.map(_.generatedSql)), GeneratedInputEncoder.flatten(all.map(_.generatedInputEncoder)))
+    GeneratedFragment(
+      GeneratedSql.flatten(all.map(_.generatedSql)),
+      GeneratedInputEncoder.flatten(all.map(_.generatedInputEncoder)),
+    )
 
   def indented(inner: GeneratedFragment, indent: String): GeneratedFragment =
     GeneratedFragment(GeneratedSql.indented(inner.generatedSql, indent), inner.generatedInputEncoder)

modules/sql/core/src/main/scala/oxygen/sql/generic/model/QueryExpr.scala

@@ -36,6 +36,7 @@ private[generic] sealed trait QueryExpr {
       case unary: QueryExpr.QueryVariableReferenceLike => ParseResult.success(unary.rowRepr)
       case _: QueryExpr.Binary                         => ParseResult.success(TypeclassExpr.RowRepr.boolean)
       case _: QueryExpr.ArrayContains                  => ParseResult.success(TypeclassExpr.RowRepr.boolean)
+      case _: QueryExpr.InList                         => ParseResult.success(TypeclassExpr.RowRepr.boolean)
       case _                                           => ParseResult.error(fullTerm, "Unable to extract RowRepr")
     }
 
@@ -314,6 +315,25 @@ private[generic] object QueryExpr extends Parser[RawQueryExpr, QueryExpr] {
     override def show(using Quotes): String = s"${arrInput.show}.contains(${queryCol.show})"
   }
 
+  //////////////////////////////////////////////////////////////////////////////////////////////////////
+  //      InList
+  //////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  /**
+    * `col IN (?, ?, ...)` / `col NOT IN (?, ?, ...)` where the value-list is a runtime collection
+    * (OXY-17). The LHS must be a single query column; the RHS must be a non-optional runtime input
+    * collection (`Seq[A]`). Kept deliberately distinct from OXY-6's array `= ANY(?)` / `UNNEST`.
+    */
+  final case class InList(
+      fullTerm: Term,
+      lhs: QueryExpr.QueryVariableReferenceLike,
+      notIn: Boolean,
+      rhs: QueryExpr.InputVariableReferenceLike,
+  ) extends QueryExpr {
+    override def queryRefs: Growable[VariableReference] = lhs.queryRefs ++ rhs.queryRefs
+    override def show(using Quotes): String = s"${lhs.show} ${(if notIn then "NOT IN" else "IN").hexFg("#E6C120")} ( ${rhs.show} )"
+  }
+
   //////////////////////////////////////////////////////////////////////////////////////////////////////
   //      BuiltIn
   //////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -405,6 +425,19 @@ private[generic] object QueryExpr extends Parser[RawQueryExpr, QueryExpr] {
           case inner: QueryExpr.QueryVariableReferenceLike => ParseResult.Success(QueryExpr.CountWithArg(fullTerm, inner))
           case inner                                       => ParseResult.error(inner.fullTerm, "can only count( _ ) a unary query")
         }
+      case RawQueryExpr.InList(fullTerm, lhs, notIn, rhs) =>
+        for {
+          lhs <- parse(lhs)
+          lhs <- lhs match {
+            case lhs: QueryExpr.QueryVariableReferenceLike => ParseResult.Success(lhs)
+            case _                                         => ParseResult.error(lhs.fullTerm, "left-hand side of `in`/`notIn` must be a single query column")
+          }
+          rhs <- parse(rhs)
+          rhs <- rhs match {
+            case rhs: QueryExpr.InputVariableReferenceLike => ParseResult.Success(rhs)
+            case _                                         => ParseResult.error(rhs.fullTerm, "right-hand side of `in`/`notIn` must be a runtime input collection (e.g. `input[Seq[A]]`)")
+          }
+        } yield QueryExpr.InList(fullTerm, lhs, notIn, rhs)
     }
 
 }

modules/sql/core/src/main/scala/oxygen/sql/generic/parsing/RawQueryExpr.scala

@@ -43,6 +43,7 @@ private[generic] sealed trait RawQueryExpr {
     case RawQueryExpr.InstantNow(_)                                 => "Instant.now()"
     case RawQueryExpr.OptionApply(_, inner)                         => s"Option(${inner.show})"
     case RawQueryExpr.StringConcat(_, args)                         => args.map(_.show).mkString("CONCAT(", ", ", ")")
+    case RawQueryExpr.InList(_, lhs, notIn, rhs)                    => s"${lhs.show} ${if notIn then "NOT IN" else "IN"} (${rhs.show})"
   }
 
 }
@@ -227,6 +228,29 @@ private[generic] object RawQueryExpr extends Parser[(Term, RefMap), RawQueryExpr
 
   }
 
+  //////////////////////////////////////////////////////////////////////////////////////////////////////
+  //      InList
+  //////////////////////////////////////////////////////////////////////////////////////////////////////
+
+  final case class InList(fullTerm: Term, lhs: RawQueryExpr, notIn: Boolean, rhs: RawQueryExpr) extends RawQueryExpr
+  object InList extends Parser[(Term, RefMap), InList] {
+
+    override def parse(input: (Term, RefMap))(using ParseContext, Quotes): ParseResult[InList] = {
+      val (rootTerm, refs) = input
+      rootTerm match {
+        // `col.in(values)` / `col.notIn(values)` extension-method call -> `in(col)(values)`
+        case singleApply(singleApply(Ident(op @ ("in" | "notIn")), lhs), rhs) =>
+          for {
+            lhs <- RawQueryExpr.parse((lhs, refs))
+            rhs <- RawQueryExpr.parse((rhs, refs))
+          } yield InList(rootTerm, lhs, op == "notIn", rhs)
+        case _ =>
+          ParseResult.unknown(rootTerm, "not an in/notIn")
+      }
+    }
+
+  }
+
   //////////////////////////////////////////////////////////////////////////////////////////////////////
   //      BuiltIn
   //////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -396,6 +420,7 @@ private[generic] object RawQueryExpr extends Parser[(Term, RefMap), RawQueryExpr
     case OptionGet.optional(res)           => res
     case OptionNullability.optional(res)   => res
     case SelectProductField.optional(res)  => res
+    case InList.optional(res)              => res
     case Binary.optional(res)              => res
     case InstantiateTable.optional(res)    => res
     case RandomUUID.optional(res)          => res

modules/sql/core/src/main/scala/oxygen/sql/query/dsl/Q.scala

@@ -90,6 +90,10 @@ object Q {
     def <+>(value: A): Double = macroOnly // L1 (Manhattan) distance
     def @>(value: A): Boolean = macroOnly // is ancestor of (ltree)
     def <@(value: A): Boolean = macroOnly // is descendant of (ltree)
+    def in(values: Seq[A]): Boolean = macroOnly // SQL `col = ANY(?)`  -- static, single array bind (OXY-17)
+    def notIn(values: Seq[A]): Boolean = macroOnly // SQL `col <> ALL(?)` -- static, single array bind (OXY-17)
+    def in(values: Set[A]): Boolean = macroOnly // SQL `col = ANY(?)`  -- static, single array bind (OXY-17)
+    def notIn(values: Set[A]): Boolean = macroOnly // SQL `col <> ALL(?)` -- static, single array bind (OXY-17)
 
   extension (self: String)
     def like(pattern: String): Boolean = macroOnly // SQL `LIKE` (case-sensitive pattern match)

modules/sql/core/src/main/scala/oxygen/sql/query/query.scala

@@ -160,7 +160,8 @@ final class QueryI[I](
   ): QueryResult.Update[QueryError] =
     self.execute(ev((i1, i2, i3, i4, i5, i6, i7, i8)))
 
-  def contramap[I2](f: I2 => I): QueryI[I2] = QueryI[I2](self.ctx, self.encoder.contramap(f))
+  def contramap[I2](f: I2 => I): QueryI[I2] =
+    QueryI[I2](self.ctx, self.encoder.contramap(f))
   def transformIn[I2](using t: Transform[I2, I]): QueryI[I2] = contramap(t.transform)
 
 }
@@ -356,7 +357,8 @@ object QueryIO {
           } yield o,
       )
 
-    override def contramap[I2](f: I2 => I): QueryIO[I2, O] = QueryIO.Simple[I2, O](self.ctx, self.encoder.contramap(f), self.decoder)
+    override def contramap[I2](f: I2 => I): QueryIO[I2, O] =
+      QueryIO.Simple[I2, O](self.ctx, self.encoder.contramap(f), self.decoder)
     override def map[O2](f: O => O2): QueryIO[I, O2] = QueryIO.Simple[I, O2](self.ctx, self.encoder, self.decoder.map(f))
     override def mapOrFail[O2](f: O => Either[String, O2]): QueryIO[I, O2] = QueryIO.Simple[I, O2](self.ctx, self.encoder, self.decoder.mapOrFail(f))
 

modules/sql/it-test/src/test/scala/oxygen/sql/CustomQuerySpec.scala

@@ -28,6 +28,68 @@ object CustomQuerySpec extends OxygenSpec[Database] {
           getP2s == p2s.toSet,
         )
       },
+      test("in / notIn (OXY-17)") {
+        for {
+          groupId1 <- Random.nextUUID
+          groupId2 <- Random.nextUUID
+          ps <- Person.generate(groupId1)().replicateZIO(6).map(_.toList)
+          others <- Person.generate(groupId2)().replicateZIO(3).map(_.toList)
+
+          _ <- Person.insert.batched(ps ++ others).unit
+
+          allIds = ps.map(_.id)
+          subset = ps.take(3)
+          subsetIds = subset.map(_.id)
+
+          // non-empty list
+          resSubset <- queries.selectByIds(subsetIds).to[Set]
+          // single-element list
+          resSingle <- queries.selectByIds(List(ps.head.id)).to[Set]
+          // empty list -> `= ANY('{}')` -> FALSE -> no rows
+          resEmpty <- queries.selectByIds(Nil).to[Set]
+          // large list (still a single array bind, static SQL)
+          resAll <- queries.selectByIds(allIds).to[Set]
+
+          // NOT IN: everything in group NOT among the excluded subset
+          allGroup1 <- queries.selectByGroupId(groupId1).to[Set]
+          resNotIn <- queries.selectByGroupAndIds2NotIn(groupId1, subsetIds).to[Set]
+          // NOT IN empty -> WHERE TRUE -> all group-1 rows
+          resNotInEmpty <- queries.selectByGroupAndIds2NotIn(groupId1, Nil).to[Set]
+
+          // composition with another predicate
+          resComposed <- queries.selectByGroupAndIds(groupId1, allIds).to[Set]
+          resComposedWrongGroup <- queries.selectByGroupAndIds(groupId2, allIds).to[Set]
+
+          // `Set` value-list (same `= ANY(?)` / `<> ALL(?)` path)
+          resSubsetSet <- queries.selectByIdsSet(subsetIds.toSet).to[Set]
+          resSetNotIn <- queries.selectByIdsSetNotIn(subsetIds.toSet).to[Set]
+
+          // delete by IN
+          deleted <- queries.deleteByIds(subsetIds).to[Set]
+          afterDelete <- queries.selectByGroupId(groupId1).to[Set]
+        } yield assertTrue(
+          resSubset == subset.toSet,
+          resSingle == Set(ps.head),
+          resEmpty.isEmpty,
+          resAll == ps.toSet,
+          resSubsetSet == subset.toSet,
+          // NOT IN over a Set: all inserted people (both groups) except the excluded subset
+          resSetNotIn == ((ps ++ others).toSet -- subset.toSet),
+          queries.selectByIdsSet.ctx.sql.contains("p.id = ANY(?)"),
+          queries.selectByIdsSetNotIn.ctx.sql.contains("p.id <> ALL(?)"),
+          resNotIn == (allGroup1 -- subset.toSet),
+          resNotInEmpty == allGroup1,
+          resComposed == ps.toSet,
+          resComposedWrongGroup.isEmpty,
+          deleted == subset.toSet,
+          afterDelete == (ps.toSet -- subset.toSet),
+          // SQL is 100% STATIC: a single array bind param (`= ANY(?)` / `<> ALL(?)`), never an
+          // N-placeholder `IN (?, ?, ...)` list -- so the text is identical regardless of list size.
+          queries.selectByIds.ctx.sql.contains("p.id = ANY(?)"),
+          !queries.selectByIds.ctx.sql.contains("IN ("),
+          queries.selectByGroupAndIds2NotIn.ctx.sql.contains("p.id <> ALL(?)"),
+        )
+      },
       test("setAgeTo0") {
         for {
           groupId <- Random.nextUUID