WIP to add 'EXCLUDE' to physical plan compiler

docs/wiki/documentation/Functions.md

@@ -477,20 +477,20 @@ EXTRACT(TIMEZONE_MINUTE FROM TIME WITH TIME ZONE '23:12:59-08:30')  -- -30
 ### `FILTER_DISTINCT` -- since v0.7.0
 
 Signature
-: `FILTER_DISTINCT: Container -> Bag`
+: `FILTER_DISTINCT: Container -> Bag|List`
 
 Header
 : `FILTER_DISTINCT(c)`
 
 Purpose
-: Returns a bag of distinct values contained within a bag, list, sexp, or struct.  If the container is a struct,
-the field names are not considered.
+: Returns a bag or list of distinct values contained within a bag, list, sexp, or struct.  If the container is a struct,
+the field names are not considered. A list will be returned if and only if the input is a list.
 
 Examples
 :
 
 ```sql
-FILTER_DISTINCT([0, 0, 1])                  -- <<0, 1>>
+FILTER_DISTINCT([0, 0, 1])                  -- [0, 1]
 FILTER_DISTINCT(<<0, 0, 1>>)                -- <<0, 1>>
 FILTER_DISTINCT(SEXP(0, 0, 1))              -- <<0, 1>>
 FILTER_DISTINCT({'a': 0, 'b': 0, 'c': 1})   -- <<0, 1>>

partiql-ast/src/main/pig/partiql.ion

@@ -799,6 +799,9 @@ may then be further optimized by selecting better implementations of each operat
 
                 // For every row of `source`, adds each specified `let_binding`.
                 (let source::bexpr bindings::(* let_binding 1))
+
+                // For every row of `source`, omits the values specified by `exclude_expr`s
+                (exclude_clause source::bexpr exprs::(* exclude_expr 1))
             )
         )
 
@@ -872,6 +875,7 @@ may then be further optimized by selecting better implementations of each operat
             column_component
             returning_mapping
             assignment
+            exclude_op
         )
     )
 )
@@ -934,6 +938,9 @@ may then be further optimized by selecting better implementations of each operat
             )
         )
 
+        (exclude exclude_expr)
+        (include (product exclude_expr root::int steps::(* exclude_step 1)))
+
         // Replace statement.dml.target with statement.dml.uniqueId (the "resolved" corollary).
         (with statement
             (exclude dml)
@@ -1007,6 +1014,8 @@ may then be further optimized by selecting better implementations of each operat
                 // Notice that the physical window operator contains a list of window expression
                 // That is because, we want to combine the window functions that are operating on the same window to a single window operator
                 (window i::impl source:: bexpr window_specification:: over window_expression_list:: (* window_expression 1))
+
+                (exclude_clause i::impl source::bexpr exprs::(* exclude_expr 1))
             )
         )
     )

partiql-eval/src/main/kotlin/org/partiql/lang/compiler/PartiQLCompilerBuilder.kt

@@ -24,6 +24,7 @@ import org.partiql.lang.eval.builtins.storedprocedure.StoredProcedure
 import org.partiql.lang.eval.internal.builtins.SCALAR_BUILTINS_DEFAULT
 import org.partiql.lang.eval.internal.builtins.definitionalBuiltins
 import org.partiql.lang.eval.physical.operators.AggregateOperatorFactoryDefault
+import org.partiql.lang.eval.physical.operators.ExcludeRelationalOperatorFactoryDefault
 import org.partiql.lang.eval.physical.operators.FilterRelationalOperatorFactoryDefault
 import org.partiql.lang.eval.physical.operators.JoinRelationalOperatorFactoryDefault
 import org.partiql.lang.eval.physical.operators.LetRelationalOperatorFactoryDefault
@@ -87,6 +88,7 @@ class PartiQLCompilerBuilder private constructor() {
             // Notice here we will not propagate the optin requirement to the user
             @OptIn(ExperimentalWindowFunctions::class)
             WindowRelationalOperatorFactoryDefault,
+            ExcludeRelationalOperatorFactoryDefault,
         )
 
         @JvmStatic

partiql-eval/src/main/kotlin/org/partiql/lang/eval/internal/StructExprValue.kt

@@ -34,7 +34,7 @@ internal enum class StructOrdering {
  * Provides a [ExprValueType.STRUCT] implementation lazily backed by a sequence.
  */
 internal open class StructExprValue(
-    private val ordering: StructOrdering,
+    internal val ordering: StructOrdering,
     private val sequence: Sequence<ExprValue>
 ) : BaseExprValue() {
 

partiql-eval/src/main/kotlin/org/partiql/lang/eval/internal/builtins/ScalarBuiltinsExt.kt

@@ -110,31 +110,33 @@ internal object ExprFunctionUtcNow : ExprFunction {
 }
 
 /**
- * Returns a bag of distinct values contained within a bag, list, sexp, or struct.
+ * Returns a bag or list of distinct values contained within a bag, list, sexp, or struct.
  * If the container is a struct, the field names are not considered.
  */
 internal object ExprFunctionFilterDistinct : ExprFunction {
 
     override val signature = FunctionSignature(
         name = "filter_distinct",
         requiredParameters = listOf(unionOf(StaticType.BAG, StaticType.LIST, StaticType.SEXP, StaticType.STRUCT)),
-        returnType = StaticType.BAG
+        returnType = unionOf(StaticType.BAG, StaticType.LIST)
     )
 
     override fun callWithRequired(session: EvaluationSession, required: List<ExprValue>): ExprValue {
         val argument = required.first()
         // We cannot use a [HashSet] here because [ExprValue] does not implement .equals() and .hashCode()
         val encountered = TreeSet(DEFAULT_COMPARATOR)
-        return ExprValue.newBag(
-            sequence {
-                argument.asSequence().forEach {
-                    if (!encountered.contains(it)) {
-                        encountered.add(it.unnamedValue())
-                        yield(it)
-                    }
+        val seq = sequence {
+            argument.asSequence().forEach {
+                if (!encountered.contains(it)) {
+                    encountered.add(it.unnamedValue())
+                    yield(it)
                 }
             }
-        )
+        }
+        return when (argument.type) {
+            ExprValueType.LIST -> ExprValue.newList(seq)
+            else -> ExprValue.newBag(seq)
+        }
     }
 }
 

partiql-eval/src/main/kotlin/org/partiql/lang/eval/physical/PhysicalBexprToThunkConverter.kt

@@ -14,6 +14,7 @@ import org.partiql.lang.eval.physical.operators.CompiledAggregateFunction
 import org.partiql.lang.eval.physical.operators.CompiledGroupKey
 import org.partiql.lang.eval.physical.operators.CompiledSortKey
 import org.partiql.lang.eval.physical.operators.CompiledWindowFunction
+import org.partiql.lang.eval.physical.operators.ExcludeRelationalOperatorFactory
 import org.partiql.lang.eval.physical.operators.FilterRelationalOperatorFactory
 import org.partiql.lang.eval.physical.operators.JoinRelationalOperatorFactory
 import org.partiql.lang.eval.physical.operators.LetRelationalOperatorFactory
@@ -28,6 +29,7 @@ import org.partiql.lang.eval.physical.operators.ScanRelationalOperatorFactory
 import org.partiql.lang.eval.physical.operators.SortOperatorFactory
 import org.partiql.lang.eval.physical.operators.UnpivotOperatorFactory
 import org.partiql.lang.eval.physical.operators.WindowRelationalOperatorFactory
+import org.partiql.lang.eval.physical.operators.compileExcludeClause
 import org.partiql.lang.eval.physical.operators.valueExpression
 import org.partiql.lang.eval.physical.window.createBuiltinWindowFunction
 import org.partiql.lang.util.toIntExact
@@ -324,6 +326,21 @@ internal class PhysicalBexprToThunkConverter(
         // wrap in thunk
         return bindingsExpr.toRelationThunk(node.metas)
     }
+
+    override fun convertExcludeClause(node: PartiqlPhysical.Bexpr.ExcludeClause): RelationThunkEnv {
+        // recurse into children
+        val sourceBexpr = this.convert(node.source)
+        val compiledBindings = compileExcludeClause(node)
+
+        // locate operator factory
+        val factory = findOperatorFactory<ExcludeRelationalOperatorFactory>(RelationalOperatorKind.EXCLUDE, node.i.name.text)
+
+        // create operator implementation
+        val bindingsExpr = factory.create(node.i, sourceBexpr, compiledBindings)
+
+        // wrap in thunk
+        return bindingsExpr.toRelationThunk(node.metas)
+    }
 }
 
 private fun PartiqlPhysical.Expr.isLitTrue() =