From cb6e3263460a3a667e0820c23bd20ad359ec25a6 Mon Sep 17 00:00:00 2001 From: jymaire Date: Thu, 30 Jul 2026 15:53:06 +0200 Subject: [PATCH 1/3] fix(jsonata): bound maxDepth by expression nesting, not input size MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Frame.setRuntimeBounds installs a Timebox whose entry and exit callbacks both skip accounting when the frame carries isParallelCall. The two sides do not observe the same flag for a given evaluate() call, so the counter gains one level per input item and never unwinds: maxDepth ended up capping how many records could be processed rather than how deeply the expression nests. A 1000-item batch failed on the default maxDepth=1000, reporting non-terminating recursion for expressions containing none. Replace it with EvaluationBounds, which counts entry and exit unconditionally — safe because the Java port evaluates path steps sequentially. Depth is now constant in input size (4 levels for a $map projection at any batch size), and runaway recursion and timeouts are still caught, with actionable messages. The companion defect in #102 — $map with a lambda being quadratic — is upstream in dashjoin/jsonata-java and unfixable from here; the plugin doc now points at the equivalent path projection, which is linear. Refs #102 Co-Authored-By: Claude Opus 5 --- .../transform/jsonata/EvaluationBounds.java | 59 +++++++++++++ .../transform/jsonata/JSONataInterface.java | 10 ++- .../plugin/transform/jsonata/Transform.java | 2 +- .../doc/io.kestra.plugin.transform.jsonata.md | 16 ++++ .../transform/jsonata/TransformItemsTest.java | 48 +++++++++++ .../transform/jsonata/TransformValueTest.java | 82 +++++++++++++++++++ 6 files changed, 214 insertions(+), 3 deletions(-) create mode 100644 plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java new file mode 100644 index 0000000..c91f0b0 --- /dev/null +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java @@ -0,0 +1,59 @@ +package io.kestra.plugin.transform.jsonata; + +import com.dashjoin.jsonata.JException; +import com.dashjoin.jsonata.Jsonata; + +/** + * Depth and timeout guard for a single JSONata evaluation. + * + *

Replaces {@link Jsonata.Frame#setRuntimeBounds(long, int)}, whose {@code Timebox} skips the + * increment and the decrement whenever the frame carries {@code isParallelCall}. Entry and exit do + * not observe the same flag for a given {@code evaluate()} call, so the counter gains one per input + * item and never unwinds — making {@code maxDepth} bound input size instead of recursion depth + * (kestra-io/plugin-transform#102). Counting unconditionally keeps the two sides symmetric, which is + * safe because the Java port evaluates path steps sequentially. + */ +final class EvaluationBounds { + + private final int maxDepth; + private final long timeoutMillis; + private final long startedAt; + + private int depth; + + static void register(Jsonata.Frame frame, int maxDepth, long timeoutMillis) { + var bounds = new EvaluationBounds(maxDepth, timeoutMillis); + frame.setEvaluateEntryCallback((expression, input, environment) -> bounds.enter()); + frame.setEvaluateExitCallback((expression, input, environment, result) -> bounds.exit()); + } + + private EvaluationBounds(int maxDepth, long timeoutMillis) { + this.maxDepth = maxDepth; + this.timeoutMillis = timeoutMillis; + this.startedAt = System.currentTimeMillis(); + } + + private void enter() { + if (++depth > maxDepth) { + throw new JException( + "JSONata expression exceeded maxDepth=" + maxDepth + " nested evaluation levels. " + + "Raise maxDepth if the expression legitimately recurses that deep, otherwise check " + + "for a recursive function that never reaches its terminating case.", + -1 + ); + } + + if (System.currentTimeMillis() - startedAt > timeoutMillis) { + throw new JException( + "JSONata evaluation exceeded timeout=" + timeoutMillis + "ms. " + + "Raise timeout for genuinely long transformations, otherwise check for a " + + "non-terminating expression.", + -1 + ); + } + } + + private void exit() { + depth--; + } +} diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java index dd1fba0..76406bc 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java @@ -13,8 +13,14 @@ public interface JSONataInterface { Property getExpression(); @Schema( - title = "The maximum number of recursive calls allowed for the JSONata transformation", - description = "Limits recursive JSONata function call depth. Each recursive call adds a frame to the chain traversed by variable lookup, so high values can cause a JVM StackOverflowError on platforms with small default thread stacks (e.g. Windows ~256 KB vs Linux ~512 KB). Raise only for expressions with proven deep recursion needs." + title = "The maximum number of nested evaluation levels allowed for the JSONata expression", + description = """ + Bounds how deeply the expression may nest while it is evaluated, which is what caps runaway \ + recursive functions. The limit depends only on the expression, never on how many records are \ + processed: the default of 1000 is far above what ordinary expressions reach, and a batch of \ + 1,000,000 records needs no more than a batch of 10. Raise it only for expressions with proven \ + deep recursion needs — each level adds frames to the chain traversed by variable lookup, so very \ + high values trade the clean error for a JVM StackOverflowError.""" ) @NotNull @PluginProperty(group = "main") diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java index d117376..39b0504 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/Transform.java @@ -108,7 +108,7 @@ protected JsonNode evaluateExpression(RunContext runContext, JsonNode jsonNode) var data = MAPPER.convertValue(jsonNode, Object.class); var frame = this.parsedExpression.createFrame(); - frame.setRuntimeBounds(timeoutInMilli, rMaxDepth); + EvaluationBounds.register(frame, rMaxDepth, timeoutInMilli); var resultRef = new AtomicReference(); var errorRef = new AtomicReference(); diff --git a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md index f7c362c..77fd8c5 100644 --- a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md +++ b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md @@ -13,3 +13,19 @@ Query and transform JSON data using [JSONata](https://jsonata.org/) expressions: Expressions follow the standard JSONata syntax. See the [JSONata documentation](https://docs.jsonata.org/) for operators, built-in functions, and examples. `TransformValue` works on a value already in the flow, while `TransformItems` streams over a file in internal storage record by record. + +`maxDepth` bounds how deeply an expression nests during evaluation, which is what stops a runaway recursive function. It does not depend on how much data you process, so the default of 1000 needs no adjustment as batches grow. + +### Prefer path projection over `$map` with a lambda + +Mapping a batch with a user-defined function is quadratic in the underlying engine, so it degrades sharply with batch size — around 39 s for 1000 items where the equivalent path projection takes 10 ms. Both forms produce the same output, so prefer the projection: + +``` +# slow — avoid +$append([], $map($, function($r) { { "id": $r.eventId, "amount": $r.value } })) + +# fast — same result +[ $.{ "id": eventId, "amount": value } ] +``` + +The cost is in [`dashjoin/jsonata-java`](https://github.com/dashjoin/jsonata-java), not in this plugin. Reach for `$map` with a lambda only when the transformation genuinely needs a function value, such as passing it to `$reduce` or `$sort`. diff --git a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java index f36cf47..18505e9 100644 --- a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java +++ b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformItemsTest.java @@ -10,6 +10,8 @@ import jakarta.inject.Inject; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; import reactor.core.publisher.Flux; import java.io.InputStream; @@ -223,6 +225,52 @@ void shouldHandleLargeDatasetWithFlatFieldLookupOnConstrainedStack() throws Exce Assertions.assertEquals(recordCount, output.getProcessedItemsTotal()); } + @ParameterizedTest + @ValueSource(ints = {100, 5_000}) + void shouldCollapseBatchIntoOneRecordOnDefaultMaxDepth(int itemCount) throws Exception { + // Regression: kestra-io/plugin-transform#102. One record holding a batch of N items, folded into a + // single object — the reported shape. Frame.setRuntimeBounds leaked depth per item, so the default + // maxDepth=1000 failed above roughly 300 items here with "Depth=1001 max=1000" while passing at 100. + // The nested object constructor is what leaked; the projection form keeps the test linear in time. + RunContext runContext = runContextFactory.of(); + final Path outputFilePath = runContext.workingDir().createTempFile(".ion"); + + List> batch = new ArrayList<>(itemCount); + for (int i = 0; i < itemCount; i++) { + batch.add(Map.of("eventId", "e" + i, "value", i, "currency", "USD")); + } + + try (Writer writer = new OutputStreamWriter(Files.newOutputStream(outputFilePath))) { + FileSerde.writeAll(writer, Flux.just(batch)).block(); + writer.flush(); + } + URI uri = runContext.storage().putFile(outputFilePath.toFile()); + + TransformItems task = TransformItems.builder() + .from(Property.ofValue(uri.toString())) + .explodeArray(Property.ofValue(false)) + .expression(Property.ofValue(""" + { + "items": [ $.{ + "eventId": eventId, + "deposit": [ { "value": value, "currency": currency } ] + } ] + } + """)) + .build(); + + TransformItems.Output output = task.run(runContext); + + Assertions.assertEquals(1, output.getProcessedItemsTotal()); + + InputStream is = runContext.storage().getFile(output.getUri()); + List transformationResult = FileSerde.readAll(new InputStreamReader(is), new TypeReference() { + }).collectList().block(); + + Assertions.assertEquals(1, transformationResult.size()); + Assertions.assertEquals(itemCount, ((List) transformationResult.getFirst().get("items")).size()); + } + @Test void shouldTransformJsonInputWithDefaultIonMapper() throws Exception { // Given diff --git a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java index 4bc0b62..cd41f11 100644 --- a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java +++ b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java @@ -13,6 +13,10 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; +import java.time.Duration; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -206,6 +210,84 @@ void shouldContinueWorkingAfterStackOverflowError() throws Exception { assertThat(output.getValue().toString()).isEqualTo("42"); } + // Regression: kestra-io/plugin-transform#102. + // Frame.setRuntimeBounds counted one un-unwound level per input item, so maxDepth silently bounded + // input size — a 1000-item batch failed on the default maxDepth=1000 with a message blaming + // recursion the expression never contained. EvaluationBounds counts entry/exit symmetrically, so the + // budget a given expression needs is a property of the expression alone. + + @ParameterizedTest + @ValueSource(ints = {10, 400}) + void shouldRequireTheSameMaxDepthRegardlessOfInputSize(int itemCount) throws Exception { + // maxDepth=12 leaves ample headroom over the 4 levels this expression actually nests, while staying + // far below the itemCount + 3 the leaking counter reached — so it fails pre-fix at both sizes and + // passes post-fix at both. 400 is the upper size because $map with a lambda is still quadratic + // upstream (defect 2 of #102); the size-independence claim is carried to 5000 items by + // TransformItemsTest#shouldCollapseBatchIntoOneRecordOnDefaultMaxDepth, which uses the linear form. + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue(jsonArrayOfItems(itemCount))) + .expression(Property.ofValue(""" + { + "items": $append([], $map($, function($r) { + { + "eventId": $r.eventId, + "deposit": [ { "value": $r.value, "currency": $r.currency } ] + } + })) + } + """)) + .maxDepth(Property.ofValue(12)) + .build(); + + TransformValue.Output output = task.run(runContext); + + assertThat(output.getValue()).isNotNull(); + JsonNode result = new ObjectMapper().valueToTree(output.getValue()); + assertThat(result.get("items").size()).isEqualTo(itemCount); + } + + @Test + void shouldReportMaxDepthBreachWithActionableMessage() throws Exception { + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue("{}")) + .expression(Property.ofValue("($f := function($n) { $n > 0 ? $f($n - 1) + 0 : 0 }; $f(500))")) + .maxDepth(Property.ofValue(100)) + .build(); + + assertThatThrownBy(() -> task.run(runContext)) + .isInstanceOf(RuntimeException.class) + .hasCauseInstanceOf(JException.class) + .cause() + .hasMessageContaining("exceeded maxDepth=100"); + } + + @Test + void shouldStillEnforceTimeout() throws Exception { + // Guards that replacing setRuntimeBounds did not drop timeout enforcement along with the depth + // counter. The expression is tail-recursive, so TCO holds it at ~5 nested levels: it runs long + // purely through iteration count and cannot be stopped by the depth bound instead. + RunContext runContext = runContextFactory.of(); + TransformValue task = TransformValue.builder() + .from(Property.ofValue("{}")) + .expression(Property.ofValue("($f := function($n) { $n > 0 ? $f($n - 1) : 0 }; $f(5000000))")) + .timeout(Property.ofValue(Duration.ofMillis(200))) + .build(); + + assertThatThrownBy(() -> task.run(runContext)) + .isInstanceOf(RuntimeException.class) + .hasCauseInstanceOf(JException.class) + .cause() + .hasMessageContaining("exceeded timeout=200ms"); + } + + private static String jsonArrayOfItems(int itemCount) { + return IntStream.range(0, itemCount) + .mapToObj(i -> "{\"eventId\":\"e%d\",\"value\":%d,\"currency\":\"USD\"}".formatted(i, i)) + .collect(Collectors.joining(",", "[", "]")); + } + @Test void shouldIsolateStackOverflowInEvalThreadWhenMaxDepthExceedsStackCapacity() throws Exception { // User sets maxDepth high enough that bounds check never fires before stack exhaustion. From c8c170610c3dc5176ef62d1577f5163425a8cb98 Mon Sep 17 00:00:00 2001 From: jymaire Date: Thu, 30 Jul 2026 16:29:19 +0200 Subject: [PATCH 2/3] docs(jsonata): narrow the $map performance caveat to its real trigger Isolating the shapes shows the quadratic cost needs both a user-defined lambda in $map and an object constructor consuming its result: the same wrapper around a builtin function is flat (10 ms at 1000 items), and the lambda without the wrapper is too (103 ms). The previous wording blamed $map with a lambda in general, which overstates it. Also corrects the figures to the reported expression measured end to end: 18 s as written against 18 ms rewritten, not 39 s against 10 ms. Refs #102 Co-Authored-By: Claude Opus 5 --- .../resources/doc/io.kestra.plugin.transform.jsonata.md | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md index 77fd8c5..584c014 100644 --- a/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md +++ b/plugin-transform-json/src/main/resources/doc/io.kestra.plugin.transform.jsonata.md @@ -18,14 +18,14 @@ Expressions follow the standard JSONata syntax. See the [JSONata documentation]( ### Prefer path projection over `$map` with a lambda -Mapping a batch with a user-defined function is quadratic in the underlying engine, so it degrades sharply with batch size — around 39 s for 1000 items where the equivalent path projection takes 10 ms. Both forms produce the same output, so prefer the projection: +Building an object around a batch mapped through a user-defined function is quadratic in the underlying engine, so it degrades sharply with batch size — around 18 s for 1000 items where the equivalent path projection takes 18 ms. Both forms produce the same output, so prefer the projection: ``` # slow — avoid -$append([], $map($, function($r) { { "id": $r.eventId, "amount": $r.value } })) +{ "items": $map($, function($r) { { "id": $r.eventId, "amount": $r.value } }) } # fast — same result -[ $.{ "id": eventId, "amount": value } ] +{ "items": [ $.{ "id": eventId, "amount": value } ] } ``` -The cost is in [`dashjoin/jsonata-java`](https://github.com/dashjoin/jsonata-java), not in this plugin. Reach for `$map` with a lambda only when the transformation genuinely needs a function value, such as passing it to `$reduce` or `$sort`. +It takes both ingredients to trigger: the same object wrapper around a builtin function (`$map($, $string)`) stays fast, and so does the lambda on its own without the wrapper. The cost is in [`dashjoin/jsonata-java`](https://github.com/dashjoin/jsonata-java), not in this plugin. Reach for `$map` with a lambda only when the transformation genuinely needs a function value, such as passing it to `$reduce` or `$sort`. From e16d9022075e0476119e503c0ba06ae34c0428a1 Mon Sep 17 00:00:00 2001 From: jymaire Date: Fri, 31 Jul 2026 10:53:01 +0200 Subject: [PATCH 3/3] refactor(jsonata): address review feedback on evaluation bounds Review notes on #103, all non-blocking: - Disambiguate the maxDepth schema text. "a batch of 1,000,000 records needs no more than a batch of 10" elided its subject and first-read as a comparison of batch sizes rather than of the maxDepth each needs. - Split enter() into checkDepth()/checkTimeout(), and record why the timeout is checked on entry only where Timebox checked both sides. - Use {@code} rather than {@link} for setRuntimeBounds: the link resolves against 0.9.10 but would fail -Xdoclint on a dependency bump that renames it. - Note in shouldStillEnforceTimeout that its margin is deliberate and should be widened rather than the test deleted if CI ever flakes it. Co-Authored-By: Claude Opus 5 --- .../plugin/transform/jsonata/EvaluationBounds.java | 14 +++++++++++++- .../plugin/transform/jsonata/JSONataInterface.java | 6 +++--- .../transform/jsonata/TransformValueTest.java | 3 +++ 3 files changed, 19 insertions(+), 4 deletions(-) diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java index c91f0b0..f7a9edf 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/EvaluationBounds.java @@ -6,7 +6,7 @@ /** * Depth and timeout guard for a single JSONata evaluation. * - *

Replaces {@link Jsonata.Frame#setRuntimeBounds(long, int)}, whose {@code Timebox} skips the + *

Replaces {@code Jsonata.Frame#setRuntimeBounds(long, int)}, whose {@code Timebox} skips the * increment and the decrement whenever the frame carries {@code isParallelCall}. Entry and exit do * not observe the same flag for a given {@code evaluate()} call, so the counter gains one per input * item and never unwinds — making {@code maxDepth} bound input size instead of recursion depth @@ -34,6 +34,11 @@ private EvaluationBounds(int maxDepth, long timeoutMillis) { } private void enter() { + checkDepth(); + checkTimeout(); + } + + private void checkDepth() { if (++depth > maxDepth) { throw new JException( "JSONata expression exceeded maxDepth=" + maxDepth + " nested evaluation levels. " @@ -42,7 +47,14 @@ private void enter() { -1 ); } + } + /** + * Checked on entry only, unlike {@code Timebox.checkRunnaway()} which also checked on exit. Every + * node that can extend the evaluation enters before it runs, so the entry check bounds the overrun + * to a single node; neither side can interrupt an in-flight call anyway. + */ + private void checkTimeout() { if (System.currentTimeMillis() - startedAt > timeoutMillis) { throw new JException( "JSONata evaluation exceeded timeout=" + timeoutMillis + "ms. " diff --git a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java index 76406bc..00dff34 100644 --- a/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java +++ b/plugin-transform-json/src/main/java/io/kestra/plugin/transform/jsonata/JSONataInterface.java @@ -18,9 +18,9 @@ public interface JSONataInterface { Bounds how deeply the expression may nest while it is evaluated, which is what caps runaway \ recursive functions. The limit depends only on the expression, never on how many records are \ processed: the default of 1000 is far above what ordinary expressions reach, and a batch of \ - 1,000,000 records needs no more than a batch of 10. Raise it only for expressions with proven \ - deep recursion needs — each level adds frames to the chain traversed by variable lookup, so very \ - high values trade the clean error for a JVM StackOverflowError.""" + 1,000,000 records needs no higher a value than a batch of 10 does. Raise it only for \ + expressions with proven deep recursion needs — each level adds frames to the chain traversed \ + by variable lookup, so very high values trade the clean error for a JVM StackOverflowError.""" ) @NotNull @PluginProperty(group = "main") diff --git a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java index cd41f11..5b14531 100644 --- a/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java +++ b/plugin-transform-json/src/test/java/io/kestra/plugin/transform/jsonata/TransformValueTest.java @@ -268,6 +268,9 @@ void shouldStillEnforceTimeout() throws Exception { // Guards that replacing setRuntimeBounds did not drop timeout enforcement along with the depth // counter. The expression is tail-recursive, so TCO holds it at ~5 nested levels: it runs long // purely through iteration count and cannot be stopped by the depth bound instead. + // Wall-clock assertion by nature: 5M interpreted iterations against a 200ms budget leaves a wide + // margin. Widen it further (more iterations, lower timeout) if CI contention ever makes it flake — + // this is the only guard that timeout enforcement survived the move off Timebox. RunContext runContext = runContextFactory.of(); TransformValue task = TransformValue.builder() .from(Property.ofValue("{}"))