Adds UNPIVOT, LIMIT, and OFFSET to partiql-eval

partiql-eval/src/main/kotlin/org/partiql/eval/internal/Compiler.kt

@@ -9,12 +9,14 @@ import org.partiql.eval.internal.operator.rel.RelJoinInner
 import org.partiql.eval.internal.operator.rel.RelJoinLeft
 import org.partiql.eval.internal.operator.rel.RelJoinOuterFull
 import org.partiql.eval.internal.operator.rel.RelJoinRight
+import org.partiql.eval.internal.operator.rel.RelLimit
 import org.partiql.eval.internal.operator.rel.RelProject
 import org.partiql.eval.internal.operator.rel.RelScan
 import org.partiql.eval.internal.operator.rel.RelScanIndexed
 import org.partiql.eval.internal.operator.rel.RelScanIndexedPermissive
 import org.partiql.eval.internal.operator.rel.RelScanPermissive
 import org.partiql.eval.internal.operator.rel.RelSort
+import org.partiql.eval.internal.operator.rel.RelUnpivot
 import org.partiql.eval.internal.operator.rex.ExprCallDynamic
 import org.partiql.eval.internal.operator.rex.ExprCallStatic
 import org.partiql.eval.internal.operator.rex.ExprCase
@@ -197,6 +199,26 @@ internal class Compiler(
         }
     }
 
+    override fun visitRelOpUnpivot(node: Rel.Op.Unpivot, ctx: StaticType?): Operator {
+        val expr = visitRex(node.rex, ctx)
+        return when (session.mode) {
+            PartiQLEngine.Mode.PERMISSIVE -> RelUnpivot.Permissive(expr)
+            PartiQLEngine.Mode.STRICT -> RelUnpivot.Strict(expr)
+        }
+    }
+
+    override fun visitRelOpLimit(node: Rel.Op.Limit, ctx: StaticType?): Operator {
+        val input = visitRel(node.input, ctx)
+        val limit = visitRex(node.limit, ctx)
+        return RelLimit(input, limit)
+    }
+
+    override fun visitRelOpOffset(node: Rel.Op.Offset, ctx: StaticType?): Operator {
+        val input = visitRel(node.input, ctx)
+        val offset = visitRex(node.offset, ctx)
+        return RelLimit(input, offset)
+    }
+
     override fun visitRexOpTupleUnion(node: Rex.Op.TupleUnion, ctx: StaticType?): Operator {
         val args = node.args.map { visitRex(it, ctx) }.toTypedArray()
         return ExprTupleUnion(args)

partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/rel/RelLimit.kt

@@ -1,24 +1,37 @@
 package org.partiql.eval.internal.operator.rel
 
+import org.partiql.errors.TypeCheckException
 import org.partiql.eval.internal.Record
 import org.partiql.eval.internal.operator.Operator
+import org.partiql.value.NumericValue
+import org.partiql.value.PartiQLValueExperimental
 
+@OptIn(PartiQLValueExperimental::class)
 internal class RelLimit(
     private val input: Operator.Relation,
-    private val limit: Long,
+    private val limit: Operator.Expr,
 ) : Operator.Relation {
 
-    private var seen = 0
+    private var _seen: Long = 0
+    private var _limit: Long = 0
 
     override fun open() {
         input.open()
-        seen = 0
+        _seen = 0
+
+        // TODO pass outer scope to limit expression
+        val l = limit.eval(Record.empty)
+        if (l is NumericValue<*>) {
+            _limit = l.toInt64().value ?: 0L
+        } else {
+            throw TypeCheckException()
+        }
     }
 
     override fun next(): Record? {
-        if (seen < limit) {
+        if (_seen < _limit) {
             val row = input.next() ?: return null
-            seen += 1
+            _seen += 1
             return row
         }
         return null

partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/rel/RelOffset.kt

@@ -1,27 +1,40 @@
 package org.partiql.eval.internal.operator.rel
 
+import org.partiql.errors.TypeCheckException
 import org.partiql.eval.internal.Record
 import org.partiql.eval.internal.operator.Operator
+import org.partiql.value.NumericValue
+import org.partiql.value.PartiQLValueExperimental
 
+@OptIn(PartiQLValueExperimental::class)
 internal class RelOffset(
     private val input: Operator.Relation,
-    private val offset: Long,
+    private val offset: Operator.Expr,
 ) : Operator.Relation {
 
     private var init = false
-    private var seen = 0
+    private var _seen: Long = 0
+    private var _offset: Long = 0
 
     override fun open() {
         input.open()
         init = false
-        seen = 0
+        _seen = 0
+
+        // TODO pass outer scope to offset expression
+        val o = offset.eval(Record.empty)
+        if (o is NumericValue<*>) {
+            _offset = o.toInt64().value ?: 0L
+        } else {
+            throw TypeCheckException()
+        }
     }
 
     override fun next(): Record? {
         if (!init) {
-            while (seen < offset) {
+            while (_seen < _offset) {
                 input.next() ?: return null
-                seen += 1
+                _seen += 1
             }
             init = true
         }

partiql-eval/src/main/kotlin/org/partiql/eval/internal/operator/rel/RelUnpivot.kt

@@ -0,0 +1,84 @@
+package org.partiql.eval.internal.operator.rel
+
+import org.partiql.errors.TypeCheckException
+import org.partiql.eval.internal.Record
+import org.partiql.eval.internal.operator.Operator
+import org.partiql.value.MissingValue
+import org.partiql.value.PartiQLValue
+import org.partiql.value.PartiQLValueExperimental
+import org.partiql.value.StructValue
+import org.partiql.value.stringValue
+import org.partiql.value.structValue
+
+/**
+ * The unpivot operator produces a bag of records from a struct.
+ *
+ *  Input:   { k_0: v_0, ..., k_i: v_i }
+ *  Output:  [ k_0, v_0 ] ... [ k_i, v_i ]
+ */
+@OptIn(PartiQLValueExperimental::class)
+internal sealed class RelUnpivot : Operator.Relation {
+
+    /**
+     * Iterator of the struct fields.
+     */
+    private lateinit var _iterator: Iterator<Pair<String, PartiQLValue>>
+
+    /**
+     * Each mode overrides.
+     */
+    abstract fun struct(): StructValue<*>
+
+    /**
+     * Initialize the _iterator from the concrete implementation's struct()
+     */
+    override fun open() {
+        _iterator = struct().entries.iterator()
+    }
+
+    override fun next(): Record? {
+        if (!_iterator.hasNext()) {
+            return null
+        }
+        val f = _iterator.next()
+        val k = stringValue(f.first)
+        val v = f.second
+        return Record.of(k, v)
+    }
+
+    override fun close() {}
+
+    /**
+     * In strict mode, the UNPIVOT operator raises an error on mistyped input.
+     *
+     * @property expr
+     */
+    class Strict(private val expr: Operator.Expr) : RelUnpivot() {
+
+        override fun struct(): StructValue<*> {
+            val v = expr.eval(Record.empty)
+            if (v !is StructValue<*>) {
+                throw TypeCheckException()
+            }
+            return v
+        }
+    }
+
+    /**
+     * In permissive mode, the UNPIVOT operator coerces the input (v) to a struct.
+     *
+     *  1. If v is a struct, return it.
+     *  2. If v is MISSING, return { }.
+     *  3. Else, return { '_1': v }.
+     *
+     * @property expr
+     */
+    class Permissive(private val expr: Operator.Expr) : RelUnpivot() {
+
+        override fun struct(): StructValue<*> = when (val v = expr.eval(Record.empty)) {
+            is StructValue<*> -> v
+            is MissingValue -> structValue<PartiQLValue>()
+            else -> structValue("_1" to v)
+        }
+    }
+}

partiql-planner/src/main/kotlin/org/partiql/planner/internal/typer/PlanTyper.kt

@@ -34,7 +34,6 @@ import org.partiql.planner.internal.ir.relBinding
 import org.partiql.planner.internal.ir.relOpAggregate
 import org.partiql.planner.internal.ir.relOpAggregateCallUnresolved
 import org.partiql.planner.internal.ir.relOpDistinct
-import org.partiql.planner.internal.ir.relOpErr
 import org.partiql.planner.internal.ir.relOpExclude
 import org.partiql.planner.internal.ir.relOpExcludePath
 import org.partiql.planner.internal.ir.relOpFilter
@@ -173,26 +172,23 @@ internal class PlanTyper(
             // descend, with GLOBAL resolution strategy
             val rex = node.rex.type(outer, Scope.GLOBAL)
 
-            // only UNPIVOT a struct
-            if (rex.type !is StructType) {
-                handleUnexpectedType(rex.type, expected = setOf(StaticType.STRUCT))
-                return rel(ctx!!, relOpErr("UNPIVOT on non-STRUCT type ${rex.type}"))
-            }
+            // key type, always a string.
+            val kType = STRING
 
-            // compute element type
-            val t = rex.type
-            val e = if (t.contentClosed) {
-                unionOf(t.fields.map { it.value }.toSet()).flatten()
-            } else {
-                ANY
+            // value type, possibly coerced.
+            val vType = when (val t = rex.type) {
+                is StructType -> {
+                    if (t.contentClosed || t.constraints.contains(TupleConstraint.Open(false))) {
+                        unionOf(t.fields.map { it.value }.toSet()).flatten()
+                    } else {
+                        ANY
+                    }
+                }
+                else -> t
             }
 
-            // compute rel type
-            val kType = STRING
-            val vType = e
-            val type = ctx!!.copyWithSchema(listOf(kType, vType))
-
             // rewrite
+            val type = ctx!!.copyWithSchema(listOf(kType, vType))
             val op = relOpUnpivot(rex)
             return rel(type, op)
         }