Adds support for thread interruption in compilation and execution (#1211)

* Adds support for thread interruption in compilation and execution

* Adds support for CLI users to use CTRL-C

Author: johnedquinn

CHANGELOG.md

@@ -29,6 +29,10 @@ Thank you to all who have contributed!
 ## [Unreleased]
 
 ### Added
+- Adds `isInterruptible` property to `CompileOptions`. The default value is `false`. Please see the KDocs for more information.
+- Adds support for thread interruption in compilation and execution. If you'd like to opt-in to this addition, please see
+  the `isInterruptible` addition above for more information.
+- Adds support for CLI users to use CTRL-C to cancel long-running compilation/execution of queries
 
 ### Changed
 

partiql-cli/src/main/kotlin/org/partiql/cli/pipeline/AbstractPipeline.kt

@@ -145,6 +145,7 @@ internal sealed class AbstractPipeline(open val options: PipelineOptions) {
             projectionIteration(options.projectionIterationBehavior)
             undefinedVariable(options.undefinedVariableBehavior)
             typingMode(options.typingMode)
+            isInterruptible(true)
         }
 
         private val compilerPipeline = CompilerPipeline.build {

partiql-cli/src/main/kotlin/org/partiql/cli/shell/RunnablePipeline.kt

@@ -0,0 +1,56 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates.  All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at:
+ *
+ *      http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific
+ * language governing permissions and limitations under the License.
+ */
+
+package org.partiql.cli.shell
+
+import org.partiql.cli.pipeline.AbstractPipeline
+import org.partiql.lang.eval.EvaluationSession
+import org.partiql.lang.eval.PartiQLResult
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * A wrapper over [AbstractPipeline]. It constantly grabs input queries from [inputs] and places the [PartiQLResult]
+ * in [results]. When it is done compiling a single statement, it sets [doneCompiling] to true.
+ */
+internal class RunnablePipeline(
+    private val inputs: BlockingQueue<Input>,
+    private val results: BlockingQueue<PartiQLResult>,
+    val pipeline: AbstractPipeline,
+    private val doneCompiling: AtomicBoolean
+) : Runnable {
+    /**
+     * When the Thread running this [Runnable] is interrupted, the underlying [AbstractPipeline] should catch the
+     * interruption and fail with some exception. Then, this will break out of [run].
+     */
+    override fun run() {
+        while (true) {
+            val input = inputs.poll(3, TimeUnit.SECONDS)
+            if (input != null) {
+                val result = pipeline.compile(input.input, input.session)
+                results.put(result)
+                doneCompiling.set(true)
+            }
+        }
+    }
+
+    /**
+     * Represents a PartiQL statement ([input]) and the [EvaluationSession] to evaluate with.
+     */
+    internal data class Input(
+        val input: String,
+        val session: EvaluationSession
+    )
+}

partiql-cli/src/main/kotlin/org/partiql/cli/shell/RunnableWriter.kt

@@ -0,0 +1,53 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates.  All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at:
+ *
+ *      http://aws.amazon.com/apache2.0/
+ *
+ * or in the "license" file accompanying this file. This file is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific
+ * language governing permissions and limitations under the License.
+ */
+
+package org.partiql.cli.shell
+
+import org.partiql.lang.eval.ExprValue
+import org.partiql.lang.util.ExprValueFormatter
+import java.io.PrintStream
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * A wrapper over [ExprValueFormatter]. It constantly grabs [ExprValue]s from [values] and writes them to [out].
+ * When it is done printing a single item, it sets [donePrinting] to true.
+ */
+internal class RunnableWriter(
+    private val out: PrintStream,
+    private val formatter: ExprValueFormatter,
+    private val values: BlockingQueue<ExprValue>,
+    private val donePrinting: AtomicBoolean
+) : Runnable {
+
+    /**
+     * When the Thread running this [Runnable] is interrupted, the underlying formatter should check the interruption
+     * flag and fail with some exception. The formatter itself doesn't do this, but, since [ExprValue]s are lazily created,
+     * the creation of the [ExprValue] (by means of the thunks produced by the EvaluatingCompiler) should throw an exception
+     * when the thread is interrupted. Then, this will break out of [run].
+     */
+    override fun run() {
+        while (true) {
+            val value = values.poll(3, TimeUnit.SECONDS)
+            if (value != null) {
+                out.info(BAR_1)
+                formatter.formatTo(value, out)
+                out.println()
+                out.info(BAR_2)
+                donePrinting.set(true)
+            }
+        }
+    }
+}