Repository: kestra-io/kestra
Issue: Allow to iterate with an Each with object or KeyValue for example
Labels: backend

For now, we can only iterate through a `List<String>` : 
https://github.com/kestra-io/kestra/blob/6d33b89025b32becd1905c1b6ba5d1b3f4ec10b5/core/src/main/java/org/kestra/core/runners/FlowableUtils.java#L163
Some normal case can be to iterate through a `List<Object>` no matter the type, for example a Key Value and or query output with multiple values. 

Issue comments:

Pull request review comments:

Patch that resolved the issue:
diff --git a/core/src/main/java/org/kestra/core/runners/FlowableUtils.java b/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
index be07caa3ca..b191a2aa0c 100644
--- a/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
+++ b/core/src/main/java/org/kestra/core/runners/FlowableUtils.java
@@ -1,6 +1,7 @@
 package org.kestra.core.runners;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.extern.slf4j.Slf4j;
 import org.kestra.core.exceptions.IllegalVariableEvaluationException;
@@ -10,6 +11,7 @@
 import org.kestra.core.models.flows.State;
 import org.kestra.core.models.tasks.ResolvedTask;
 import org.kestra.core.models.tasks.Task;
+import org.kestra.core.serializers.JacksonMapper;
 
 import java.util.*;
 import java.util.stream.Collectors;
@@ -173,42 +175,56 @@ public static List<NextTaskRun> resolveParallelNexts(
         return new ArrayList<>();
     }
 
-    public static List<ResolvedTask> resolveEachTasks(RunContext runContext, TaskRun parentTaskRun, List<Task> tasks, String value) throws IllegalVariableEvaluationException {
-        ObjectMapper mapper = new ObjectMapper();
-
-        String[] values;
+    private final static TypeReference<List<Object>> TYPE_REFERENCE = new TypeReference<>() {};
+    private final static ObjectMapper MAPPER = JacksonMapper.ofJson();
 
+    public static List<ResolvedTask> resolveEachTasks(RunContext runContext, TaskRun parentTaskRun, List<Task> tasks, String value) throws IllegalVariableEvaluationException {
         String renderValue = runContext.render(value);
+
+        List<Object> values;
         try {
-            values = mapper.readValue(renderValue, String[].class);
+            values = MAPPER.readValue(renderValue, TYPE_REFERENCE);
         } catch (JsonProcessingException e) {
             throw new IllegalVariableEvaluationException(e);
         }
 
-        long nullCount = Arrays
-            .stream(values)
+        List<Object> distinctValue = values
+            .stream()
+            .distinct()
+            .collect(Collectors.toList());
+
+
+        long nullCount = distinctValue
+            .stream()
             .filter(Objects::isNull)
             .count();
 
         if (nullCount > 0) {
             throw new IllegalVariableEvaluationException("Found '" + nullCount + "' null values on Each, " +
-                "with values=" + Arrays.toString(values)
+                "with values=" + Arrays.toString(values.toArray())
             );
         }
 
-        return Arrays
-            .stream(values)
-            .distinct()
-            .flatMap(v -> tasks
-                .stream()
-                .map(task -> ResolvedTask.builder()
-                    .task(task)
-                    .value(v)
-                    .parentId(parentTaskRun.getId())
-                    .build()
-                )
-            )
-            .collect(Collectors.toList());
+        ArrayList<ResolvedTask> result = new ArrayList<>();
+
+        for (Object current: distinctValue) {
+            for( Task task: tasks) {
+                try {
+                    String resolvedValue = current instanceof String ? (String)current : MAPPER.writeValueAsString(current);
+
+                    result.add(ResolvedTask.builder()
+                        .task(task)
+                        .value(resolvedValue)
+                        .parentId(parentTaskRun.getId())
+                        .build()
+                    );
+                } catch (JsonProcessingException e) {
+                    throw new IllegalVariableEvaluationException(e);
+                }
+            }
+        }
+
+        return result;
     }
 
     public static boolean isTaskRunFor(ResolvedTask resolvedTask, TaskRun taskRun, TaskRun parentTaskRun) {
diff --git a/core/src/main/java/org/kestra/core/runners/handlebars/helpers/JsonHelper.java b/core/src/main/java/org/kestra/core/runners/handlebars/helpers/JsonHelper.java
index 971745dc6a..3ade42e6be 100644
--- a/core/src/main/java/org/kestra/core/runners/handlebars/helpers/JsonHelper.java
+++ b/core/src/main/java/org/kestra/core/runners/handlebars/helpers/JsonHelper.java
@@ -24,6 +24,27 @@ public CharSequence apply(final Object value, final Options options) {
                 throw new RuntimeException(e);
             }
         }
+    },
+
+    /**
+     * <p>
+     * Usage:
+     * </p>
+     *
+     * <pre>
+     *    {{json object}}
+     * </pre>
+     */
+    isJson {
+        @Override
+        public Boolean apply(final Object value, final Options options) {
+            try {
+                JacksonMapper.ofJson().readTree((String) value);
+                return true;
+            } catch (JsonProcessingException e) {
+                return false;
+            }
+        }
     }
 }
 
diff --git a/core/src/main/java/org/kestra/core/tasks/flows/EachParallel.java b/core/src/main/java/org/kestra/core/tasks/flows/EachParallel.java
index 22256e9665..4a6e32ca02 100644
--- a/core/src/main/java/org/kestra/core/tasks/flows/EachParallel.java
+++ b/core/src/main/java/org/kestra/core/tasks/flows/EachParallel.java
@@ -39,7 +39,7 @@
 @Schema(
     title = "Execute a tasks for a list of value in parallel.",
     description = "For each `value`, `tasks` will be executed\n" +
-        "The value must be valid json string representing an arrays, like `[\"value1\", \"value2\"]` and must be a string\n" +
+        "The value must be valid json string representing an arrays, like `[\"value1\", \"value2\"]` or `[{\"key\":\"value1\"}, {\"key\":\"value2\"}]` \n" +
         "The current value is available on vars `{{ taskrun.value }}`.\n" +
         "The task list will be executed in parallel, for example if you have a 3 value with each one 2 tasks, all the " +
         "6 tasks will be computed in parallel with out any garantee on the order.\n" +
diff --git a/core/src/main/java/org/kestra/core/tasks/flows/EachSequential.java b/core/src/main/java/org/kestra/core/tasks/flows/EachSequential.java
index 5699a92ca5..709e3d8967 100644
--- a/core/src/main/java/org/kestra/core/tasks/flows/EachSequential.java
+++ b/core/src/main/java/org/kestra/core/tasks/flows/EachSequential.java
@@ -39,7 +39,7 @@
 @Schema(
     title = "Execute a tasks for a list of value sequentially",
     description = "For each `value`, `tasks` will be executed\n" +
-        "The value must be valid json string representing an arrays, like `[\"value1\", \"value2\"]` and must be a string\n" +
+        "The value must be valid json string representing an arrays, like `[\"value1\", \"value2\"]` or `[{\"key\":\"value1\"}, {\"key\":\"value2\"}]` \n" +
         "The current value is available on vars `{{ taskrun.value }}`."
 )
 @Plugin(


Test patch:
diff --git a/core/src/test/java/org/kestra/core/Helpers.java b/core/src/test/java/org/kestra/core/Helpers.java
index 1be6e6e1a1..0e5f2ec5eb 100644
--- a/core/src/test/java/org/kestra/core/Helpers.java
+++ b/core/src/test/java/org/kestra/core/Helpers.java
@@ -17,7 +17,7 @@
 import java.util.function.BiConsumer;
 
 public class Helpers {
-    public static long FLOWS_COUNT = 39;
+    public static long FLOWS_COUNT = 40;
 
     public static ApplicationContext applicationContext() throws URISyntaxException {
         return applicationContext(Paths.get(Objects.requireNonNull(Helpers.class.getClassLoader().getResource("plugins")).toURI()));
diff --git a/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java b/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
index efec7479bb..a67f5f882d 100644
--- a/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
+++ b/core/src/test/java/org/kestra/core/tasks/flows/EachSequentialTest.java
@@ -36,6 +36,15 @@ void sequential() throws TimeoutException {
         assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
     }
 
+    @Test
+    void object() throws TimeoutException {
+        Execution execution = runnerUtils.runOne("org.kestra.tests", "each-object");
+
+        assertThat(execution.getTaskRunList(), hasSize(8));
+        assertThat(execution.getState().getCurrent(), is(State.Type.SUCCESS));
+        assertThat((String) execution.getTaskRunList().get(6).getOutputs().get("value"), containsString("json > JSON > [\"my-complex\"]"));
+    }
+
     @Test
     void sequentialNested() throws TimeoutException, InternalException {
         Execution execution = runnerUtils.runOne("org.kestra.tests", "each-sequential-nested");
@@ -78,7 +87,7 @@ public static void eachNullTest(RunnerUtils runnerUtils, QueueInterface<LogEntry
 
         assertThat(execution.getTaskRunList(), hasSize(1));
         assertThat(execution.getState().getCurrent(), is(State.Type.FAILED));
-        assertThat(logs.get(1).getMessage(), Matchers.containsString("Found '1' null values on Each, with values=[1, null, 2]"));
+        assertThat(logs.get(1).getMessage(), Matchers.containsString("Found '1' null values on Each, with values=[1, null, {key=my-key, value=my-value}]"));
     }
 
     @Test
diff --git a/core/src/test/resources/flows/valids/each-null.yaml b/core/src/test/resources/flows/valids/each-null.yaml
index 90a6dde537..b6b6cdf19b 100644
--- a/core/src/test/resources/flows/valids/each-null.yaml
+++ b/core/src/test/resources/flows/valids/each-null.yaml
@@ -4,7 +4,7 @@ namespace: org.kestra.tests
 tasks:
   - id: 1_each
     type: org.kestra.core.tasks.flows.EachSequential
-    value: '[1, null, 2]'
+    value: '[1, null, {"key": "my-key", "value": "my-value"}]'
     tasks:
       - id: 1-1
         type: org.kestra.core.tasks.debugs.Return
diff --git a/core/src/test/resources/flows/valids/each-object.yaml b/core/src/test/resources/flows/valids/each-object.yaml
new file mode 100644
index 0000000000..ff92f71c92
--- /dev/null
+++ b/core/src/test/resources/flows/valids/each-object.yaml
@@ -0,0 +1,30 @@
+id: each-object
+namespace: org.kestra.tests
+
+tasks:
+  - id: 1_each
+    type: org.kestra.core.tasks.flows.EachSequential
+    value: |
+      [
+        "value 1",
+        {"key": "my-key", "value": "my-value"},
+        {"key": "my-complex", "value": {"sub": 1, "bool": true}}
+      ]
+
+    tasks:
+      - id: is-json
+        type: org.kestra.core.tasks.flows.Switch
+        value: "{{isJson taskrun.value}}"
+        cases:
+          "false":
+            - id: not-json
+              type: org.kestra.core.tasks.debugs.Return
+              format: "{{task.id}} > STRING > {{parent.taskrun.value}}"
+        defaults:
+        - id: json
+          type: org.kestra.core.tasks.debugs.Return
+          format: "{{task.id}} > JSON > {{jq parent.taskrun.value '.key'}} > {{jq parent.taskrun.value '.value'}}"
+
+  - id: 2_end
+    type: org.kestra.core.tasks.debugs.Return
+    format: "{{task.id}} > {{taskrun.startDate}}"