Implement async JSONL streaming support and add corresponding tests

Author: kzu

.github/product.md

@@ -206,21 +206,30 @@ Remaining parameterized builtins:
 - Destructuring alternative operator `?//`
 - SQL-style operators: `INDEX`, `IN`, `JOIN`
 
-### - [ ] Phase 16 — I/O
+### - [x] Phase 16 - Streaming
 
-- `input`, `inputs`
-- `debug`, `debug(msgs)`, `stderr`
-- `input_filename`, `input_line_number`
+Implement support for JSONL/NDJSON async streaming
 
-### - [ ] Phase 17 — Streaming
-
-- `tostream`, `fromstream(stream_expression)`, `truncate_stream(stream_expression)`
-
-### - [ ] Phase 18 — Modules
+### - [ ] Phase 16 — Modules
 
 - `import RelativePathString as NAME [<metadata>];`
 - `include RelativePathString [<metadata>];`
 - `import RelativePathString as $NAME [<metadata>];`
 - `module <metadata>;`
 - `modulemeta`
 - Module search paths
+
+## Out of Scope
+
+Features from the jq manual we will not implement.
+
+### I/O
+
+- `input`, `inputs`
+- `debug`, `debug(msgs)`, `stderr`
+- `input_filename`, `input_line_number`
+
+### Streaming
+
+- `tostream`, `fromstream(stream_expression)`, `truncate_stream(stream_expression)`
+

src/JQSharp/Jq.cs

@@ -1,3 +1,4 @@
+using System.Runtime.CompilerServices;
 using System.Text.Json;
 
 namespace Devlooped;
@@ -35,4 +36,45 @@ public static JqExpression Parse(string expression)
     /// <exception cref="JqException">Thrown when the expression is empty, invalid, or causes an error during evaluation.</exception>
     public static IEnumerable<JsonElement> Evaluate(string expression, JsonElement input)
         => Parse(expression).Evaluate(input);
+
+    /// <summary>
+    /// Evaluates a jq filter expression against each element in an asynchronous sequence and yields the matching results.
+    /// </summary>
+    /// <param name="expression">The jq filter expression to evaluate.</param>
+    /// <param name="input">The asynchronous sequence of <see cref="JsonElement"/> values to evaluate the filter against.</param>
+    /// <param name="cancellationToken">A token to cancel the asynchronous iteration.</param>
+    /// <returns>
+    /// An asynchronous sequence of <see cref="JsonElement"/> values produced by applying the filter
+    /// to each element in <paramref name="input"/>.
+    /// </returns>
+    /// <exception cref="JqException">Thrown when the expression is empty, invalid, or causes an error during evaluation.</exception>
+    public static async IAsyncEnumerable<JsonElement> EvaluateAsync(string expression, IAsyncEnumerable<JsonElement> input, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        var parsed = Parse(expression);
+        await foreach (var element in EvaluateAsync(parsed, input, cancellationToken))
+            yield return element;
+    }
+
+    /// <summary>
+    /// Evaluates a parsed <see cref="JqExpression"/> against each element in an asynchronous sequence and yields the matching results.
+    /// </summary>
+    /// <param name="expression">The pre-parsed <see cref="JqExpression"/> to evaluate.</param>
+    /// <param name="input">The asynchronous sequence of <see cref="JsonElement"/> values to evaluate the expression against.</param>
+    /// <param name="cancellationToken">A token to cancel the asynchronous iteration.</param>
+    /// <returns>
+    /// An asynchronous sequence of <see cref="JsonElement"/> values produced by applying the expression
+    /// to each element in <paramref name="input"/>.
+    /// </returns>
+    /// <remarks>
+    /// Prefer this overload when evaluating the same expression against multiple streams, as it avoids
+    /// re-parsing the filter on every call.
+    /// </remarks>
+    public static async IAsyncEnumerable<JsonElement> EvaluateAsync(JqExpression expression, IAsyncEnumerable<JsonElement> input, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        await foreach (var element in input.WithCancellation(cancellationToken))
+        {
+            foreach (var result in expression.Evaluate(element))
+                yield return result;
+        }
+    }
 }

src/Tests/JqJsonlStreamingTests.cs

@@ -0,0 +1,86 @@
+using System.Runtime.CompilerServices;
+using System.Text.Json;
+using Devlooped;
+
+namespace Devlooped.Tests;
+
+public class JqJsonlStreamingTests
+{
+    static async IAsyncEnumerable<JsonElement> ParseJsonLines(
+        string jsonl,
+        [EnumeratorCancellation] CancellationToken cancellation = default)
+    {
+        using var reader = new StringReader(jsonl);
+        while (await reader.ReadLineAsync(cancellation).ConfigureAwait(false) is { } line)
+        {
+            line = line.Trim();
+            if (line.Length == 0)
+                continue;
+
+            yield return JsonDocument.Parse(line).RootElement;
+        }
+    }
+
+    [Fact]
+    public async Task Jsonl_stream_extracts_field_from_each_element()
+    {
+        var jsonl = """
+            {"name":"Alice","age":30}
+            {"name":"Bob","age":25}
+            {"name":"Charlie","age":35}
+            """;
+
+        var results = new List<string>();
+        await foreach (var element in Jq.EvaluateAsync(".name", ParseJsonLines(jsonl)))
+            results.Add(element.GetString()!);
+
+        Assert.Equal(["Alice", "Bob", "Charlie"], results);
+    }
+
+    [Fact]
+    public async Task Jsonl_stream_transforms_each_element()
+    {
+        var jsonl = """
+            {"x":1}
+            {"x":2}
+            {"x":3}
+            """;
+
+        var results = new List<int>();
+        await foreach (var element in Jq.EvaluateAsync(".x * 2", ParseJsonLines(jsonl)))
+            results.Add(element.GetInt32());
+
+        Assert.Equal([2, 4, 6], results);
+    }
+
+    [Fact]
+    public async Task Jsonl_stream_expands_array_elements_across_inputs()
+    {
+        var jsonl = """
+            {"items":[1,2]}
+            {"items":[3,4]}
+            """;
+
+        var results = new List<int>();
+        await foreach (var element in Jq.EvaluateAsync(".items[]", ParseJsonLines(jsonl)))
+            results.Add(element.GetInt32());
+
+        Assert.Equal([1, 2, 3, 4], results);
+    }
+
+    [Fact]
+    public async Task Jsonl_stream_filters_elements_with_select()
+    {
+        var jsonl = """
+            {"name":"Alice","active":true}
+            {"name":"Bob","active":false}
+            {"name":"Charlie","active":true}
+            """;
+
+        var results = new List<string>();
+        await foreach (var element in Jq.EvaluateAsync("select(.active) | .name", ParseJsonLines(jsonl)))
+            results.Add(element.GetString()!);
+
+        Assert.Equal(["Alice", "Charlie"], results);
+    }
+}