diff --git a/core/src/main/java/io/kestra/core/models/Label.java b/core/src/main/java/io/kestra/core/models/Label.java
new file mode 100644
index 00000000000..290c2631da1
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/models/Label.java
@@ -0,0 +1,3 @@
+package io.kestra.core.models;
+
+public record Label(String key, String value) {}
diff --git a/core/src/main/java/io/kestra/core/models/executions/Execution.java b/core/src/main/java/io/kestra/core/models/executions/Execution.java
index f1a933a55ca..caa744509f9 100644
--- a/core/src/main/java/io/kestra/core/models/executions/Execution.java
+++ b/core/src/main/java/io/kestra/core/models/executions/Execution.java
@@ -5,8 +5,13 @@
 import ch.qos.logback.classic.spi.ThrowableProxy;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Streams;
+import io.kestra.core.models.Label;
+import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
+import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import lombok.Builder;
 import lombok.Value;
 import lombok.With;
@@ -53,7 +58,9 @@ public class Execution implements DeletedInterface {
     Map<String, Object> inputs;
 
     @With
-    Map<String, String> labels;
+    @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
+    @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
+    List<Label> labels;
 
     @With
     Map<String, Object> variables;
diff --git a/core/src/main/java/io/kestra/core/models/flows/Flow.java b/core/src/main/java/io/kestra/core/models/flows/Flow.java
index e0a5237adf5..66304e7e324 100644
--- a/core/src/main/java/io/kestra/core/models/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/models/flows/Flow.java
@@ -3,10 +3,13 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
 import com.fasterxml.jackson.databind.introspect.JacksonAnnotationIntrospector;
 import io.kestra.core.exceptions.InternalException;
 import io.kestra.core.models.DeletedInterface;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.listeners.Listener;
 import io.kestra.core.models.tasks.FlowableTask;
@@ -14,6 +17,8 @@
 import io.kestra.core.models.triggers.AbstractTrigger;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.serializers.JacksonMapper;
+import io.kestra.core.serializers.ListOrMapOfLabelDeserializer;
+import io.kestra.core.serializers.ListOrMapOfLabelSerializer;
 import io.kestra.core.services.FlowService;
 import io.kestra.core.validations.FlowValidation;
 import io.micronaut.core.annotation.Introspected;
@@ -63,7 +68,9 @@ public boolean hasIgnoreMarker(final AnnotatedMember m) {
 
     String description;
 
-    Map<String, String> labels;
+    @JsonSerialize(using = ListOrMapOfLabelSerializer.class)
+    @JsonDeserialize(using = ListOrMapOfLabelDeserializer.class)
+    List<Label> labels;
 
     @Valid
     List<Input<?>> inputs;
diff --git a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
index 1aa9ba91b00..3122591335a 100644
--- a/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
+++ b/core/src/main/java/io/kestra/core/runners/RunnerUtils.java
@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.collect.ImmutableMap;
 import io.kestra.core.exceptions.MissingRequiredInput;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.Input;
@@ -283,7 +284,7 @@ public Execution runOne(String namespace, String flowId, Integer revision, BiFun
         return this.runOne(namespace, flowId, revision, inputs, duration, null);
     }
 
-    public Execution runOne(String namespace, String flowId, Integer revision, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration, Map<String, String> labels) throws TimeoutException {
+    public Execution runOne(String namespace, String flowId, Integer revision, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration, List<Label> labels) throws TimeoutException {
         return this.runOne(
             flowRepository
                 .findById(namespace, flowId, revision != null ? Optional.of(revision) : Optional.empty())
@@ -301,7 +302,7 @@ public Execution runOne(Flow flow, BiFunction<Flow, Execution, Map<String, Objec
         return this.runOne(flow, inputs, duration, null);
     }
 
-    public Execution runOne(Flow flow, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration, Map<String, String> labels) throws TimeoutException {
+    public Execution runOne(Flow flow, BiFunction<Flow, Execution, Map<String, Object>> inputs, Duration duration, List<Label> labels) throws TimeoutException {
         if (duration == null) {
             duration = Duration.ofSeconds(15);
         }
@@ -381,7 +382,7 @@ private Predicate<Execution> isTerminatedChildExecution(Execution parentExecutio
         return e -> e.getParentId() != null && e.getParentId().equals(parentExecution.getId()) && conditionService.isTerminatedWithListeners(flow, e);
     }
 
-    public Execution newExecution(Flow flow, BiFunction<Flow, Execution, Map<String, Object>> inputs, Map<String, String> labels) {
+    public Execution newExecution(Flow flow, BiFunction<Flow, Execution, Map<String, Object>> inputs, List<Label> labels) {
         Execution execution = Execution.builder()
             .id(IdUtils.create())
             .namespace(flow.getNamespace())
@@ -394,12 +395,12 @@ public Execution newExecution(Flow flow, BiFunction<Flow, Execution, Map<String,
             execution = execution.withInputs(inputs.apply(flow, execution));
         }
 
-        Map<String, String> executionLabels = new HashMap<>();
+        List<Label> executionLabels = new ArrayList<>();
         if (flow.getLabels() != null) {
-            executionLabels.putAll(flow.getLabels());
+            executionLabels.addAll(flow.getLabels());
         }
         if (labels != null) {
-            executionLabels.putAll(labels);
+            executionLabels.addAll(labels);
         }
         if (!executionLabels.isEmpty()) {
             execution = execution.withLabels(executionLabels);
diff --git a/core/src/main/java/io/kestra/core/runners/Worker.java b/core/src/main/java/io/kestra/core/runners/Worker.java
index e8bbe025900..9bf2f6316f1 100644
--- a/core/src/main/java/io/kestra/core/runners/Worker.java
+++ b/core/src/main/java/io/kestra/core/runners/Worker.java
@@ -9,6 +9,7 @@
 import com.google.common.hash.Hashing;
 import io.kestra.core.exceptions.TimeoutExceededException;
 import io.kestra.core.metrics.MetricRegistry;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.*;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.models.tasks.Output;
@@ -200,8 +201,8 @@ private void handleTrigger(WorkerTrigger workerTrigger) {
                         var flowLabels = workerTrigger.getConditionContext().getFlow().getLabels();
                         if (flowLabels != null) {
                             evaluate = evaluate.map( execution -> {
-                                    Map<String, String> executionLabels = execution.getLabels() != null ? execution.getLabels() : new HashMap<>();
-                                    executionLabels.putAll(flowLabels);
+                                    List<Label> executionLabels = execution.getLabels() != null ? execution.getLabels() : new ArrayList<>();
+                                    executionLabels.addAll(flowLabels);
                                     return execution.withLabels(executionLabels);
                                 }
                             );
diff --git a/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelDeserializer.java b/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelDeserializer.java
new file mode 100644
index 00000000000..1759a284bc4
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelDeserializer.java
@@ -0,0 +1,41 @@
+package io.kestra.core.serializers;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JsonDeserializer;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.deser.ResolvableDeserializer;
+import io.kestra.core.models.Label;
+import io.kestra.core.utils.LabelUtils;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This deserializer is for historical purpose, labels was first a map but has been updated to a List of Label so
+ * this deserializer allows using both types.
+ */
+public class ListOrMapOfLabelDeserializer extends JsonDeserializer<List<Label>> implements ResolvableDeserializer {
+    @Override
+    public List<Label> deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
+        if (p.hasToken(JsonToken.VALUE_NULL)) {
+            return null;
+        }
+        else if (p.hasToken(JsonToken.START_ARRAY)) {
+            // deserialize as list
+            List<Map<String, String>> ret = ctxt.readValue(p, List.class);
+            return ret.stream().map(map -> new Label(map.get("key"), map.get("value"))).toList();
+        }
+        else if (p.hasToken(JsonToken.START_OBJECT)) {
+            // deserialize as map
+            Map<String, String> ret = ctxt.readValue(p, Map.class);
+            return LabelUtils.from(ret);
+        }
+        throw new IllegalArgumentException("Unable to deserialize value as it's neither an object neither an array");
+    }
+
+    @Override
+    public void resolve(DeserializationContext ctxt) throws JsonMappingException {}
+}
diff --git a/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelSerializer.java b/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelSerializer.java
new file mode 100644
index 00000000000..47096293b60
--- /dev/null
+++ b/core/src/main/java/io/kestra/core/serializers/ListOrMapOfLabelSerializer.java
@@ -0,0 +1,31 @@
+package io.kestra.core.serializers;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This serializer is for historical purpose, labels was first a map but has been updated to a List of Label so
+ * this serializer allows using both types.
+ */
+public class ListOrMapOfLabelSerializer extends JsonSerializer<Object> {
+    @Override
+    public void serialize(Object value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+        if (value == null) {
+            gen.writeNull();
+        }
+        else if (value instanceof List) {
+            serializers.findValueSerializer(List.class).serialize(value, gen, serializers);
+        }
+        else if (value instanceof Map) {
+            serializers.findValueSerializer(Map.class).serialize(value, gen, serializers);
+        }
+        else {
+            throw new IllegalArgumentException("Unable to serialize value as it's neither a map nor a list");
+        }
+    }
+}
diff --git a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
index bcea3e58ed2..1b12d9e2e9c 100644
--- a/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
+++ b/core/src/main/java/io/kestra/core/tasks/flows/Flow.java
@@ -1,6 +1,7 @@
 package io.kestra.core.tasks.flows;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
@@ -17,7 +18,9 @@
 
 import javax.annotation.Nullable;
 import javax.validation.constraints.NotNull;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -118,10 +121,10 @@ public Execution createExecution(RunContext runContext, FlowExecutorInterface fl
             }
         }
 
-        Map<String, String> labels = new HashMap<>();
+        List<Label> labels = new ArrayList<>();
         if (this.labels != null) {
             for (Map.Entry<String, String> entry: this.labels.entrySet()) {
-                labels.put(entry.getKey(), runContext.render(entry.getValue()));
+                labels.add(new Label(entry.getKey(), runContext.render(entry.getValue())));
             }
         }
 
diff --git a/core/src/test/java/io/kestra/core/models/executions/ExecutionTest.java b/core/src/test/java/io/kestra/core/models/executions/ExecutionTest.java
index caf5bebe1bc..8fef39433c2 100644
--- a/core/src/test/java/io/kestra/core/models/executions/ExecutionTest.java
+++ b/core/src/test/java/io/kestra/core/models/executions/ExecutionTest.java
@@ -1,10 +1,12 @@
 package io.kestra.core.models.executions;
 
+import io.kestra.core.models.Label;
 import io.kestra.core.utils.IdUtils;
 import org.junit.jupiter.api.Test;
 import io.kestra.core.models.flows.State;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -153,10 +155,10 @@ void originalId() {
     @Test
     void labels() {
         final Execution execution = Execution.builder()
-            .labels(Map.of("test", "test-value"))
+            .labels(List.of(new Label("test", "test-value")))
             .build();
 
         assertThat(execution.getLabels().size(), is(1));
-        assertThat(execution.getLabels().get("test"), is("test-value"));
+        assertThat(execution.getLabels().get(0), is(new Label("test", "test-value")));
     }
 }
diff --git a/core/src/test/java/io/kestra/core/models/triggers/types/FlowTest.java b/core/src/test/java/io/kestra/core/models/triggers/types/FlowTest.java
index 0c74cda30ad..5ec7e6ec3e1 100644
--- a/core/src/test/java/io/kestra/core/models/triggers/types/FlowTest.java
+++ b/core/src/test/java/io/kestra/core/models/triggers/types/FlowTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers.types;
 
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.State;
 import io.kestra.core.runners.RunContextFactory;
@@ -10,6 +11,7 @@
 import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -28,9 +30,9 @@ void success() {
             .namespace("io.kestra.unittest")
             .revision(1)
             .labels(
-                Map.of(
-                    "flow-label-1", "flow-label-1",
-                    "flow-label-2", "flow-label-2")
+                List.of(
+                    new Label("flow-label-1", "flow-label-1"),
+                    new Label("flow-label-2", "flow-label-2"))
             )
             .tasks(Collections.singletonList(Return.builder()
                 .id("test")
@@ -58,7 +60,7 @@ void success() {
 
         assertThat(evaluate.isPresent(), is(true));
         assertThat(evaluate.get().getFlowId(), is("flow-with-flow-trigger"));
-        assertThat(evaluate.get().getLabels().get("flow-label-1"), is("flow-label-1"));
-        assertThat(evaluate.get().getLabels().get("flow-label-2"), is("flow-label-2"));
+        assertThat(evaluate.get().getLabels().get(0), is(new Label("flow-label-1", "flow-label-1")));
+        assertThat(evaluate.get().getLabels().get(1), is(new Label("flow-label-2", "flow-label-2")));;
     }
 }
diff --git a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
index 77b18967267..96773c8aa24 100644
--- a/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
+++ b/core/src/test/java/io/kestra/core/models/triggers/types/ScheduleTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.models.triggers.types;
 
+import io.kestra.core.models.Label;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.conditions.types.DateTimeBetweenCondition;
 import io.kestra.core.models.conditions.types.DayWeekInMonthCondition;
@@ -91,8 +92,8 @@ void success() throws Exception {
         assertThat(dateFromVars(vars.get("date"), date), is(date));
         assertThat(dateFromVars(vars.get("next"), date), is(date.plusMonths(1)));
         assertThat(dateFromVars(vars.get("previous"), date), is(date.minusMonths(1)));
-        assertThat(evaluate.get().getLabels().get("flow-label-1"), is("flow-label-1"));
-        assertThat(evaluate.get().getLabels().get("flow-label-2"), is("flow-label-2"));
+        assertThat(evaluate.get().getLabels().get(0), is(new Label("flow-label-1", "flow-label-1")));
+        assertThat(evaluate.get().getLabels().get(1), is(new Label("flow-label-2", "flow-label-2")));
     }
 
     @SuppressWarnings("unchecked")
@@ -396,9 +397,9 @@ private ConditionContext conditionContext(AbstractTrigger trigger) {
             .id(IdUtils.create())
             .namespace("io.kestra.tests")
             .labels(
-                Map.of(
-                    "flow-label-1", "flow-label-1",
-                    "flow-label-2", "flow-label-2")
+                List.of(
+                    new Label("flow-label-1", "flow-label-1"),
+                    new Label("flow-label-2", "flow-label-2"))
             )
             .build();
 
diff --git a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
index 9fad2456eeb..d1c28a512b9 100644
--- a/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/AbstractSchedulerTest.java
@@ -1,6 +1,7 @@
 package io.kestra.core.schedulers;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.ExecutionTrigger;
@@ -62,9 +63,9 @@ protected static Flow createFlow(List<AbstractTrigger> triggers, List<TaskDefaul
             ))
             .revision(1)
             .labels(
-                Map.of(
-                    "flow-label-1", "flow-label-1",
-                    "flow-label-2", "flow-label-2")
+                List.of(
+                    new Label("flow-label-1", "flow-label-1"),
+                    new Label("flow-label-2", "flow-label-2"))
             )
             .triggers(triggers)
             .tasks(Collections.singletonList(Return.builder()
diff --git a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
index 1ce374da864..c323d565a6d 100644
--- a/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
+++ b/core/src/test/java/io/kestra/core/schedulers/SchedulerThreadTest.java
@@ -1,5 +1,6 @@
 package io.kestra.core.schedulers;
 
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.flows.Flow;
 import io.kestra.core.models.flows.State;
@@ -93,8 +94,8 @@ void thread() throws Exception {
 
             assertThat(last.get().getTrigger().getVariables().get("defaultInjected"), is("done"));
             assertThat(last.get().getTrigger().getVariables().get("counter"), is(3));
-            assertThat(last.get().getLabels().get("flow-label-1"), is("flow-label-1"));
-            assertThat(last.get().getLabels().get("flow-label-2"), is("flow-label-2"));
+            assertThat(last.get().getLabels().get(0), is(new Label("flow-label-1", "flow-label-1")));
+            assertThat(last.get().getLabels().get(1), is(new Label("flow-label-2", "flow-label-2")));
             AbstractSchedulerTest.COUNTER = 0;
         }
     }
diff --git a/jdbc-h2/src/main/java/io/kestra/repository/h2/H2ExecutionRepositoryService.java b/jdbc-h2/src/main/java/io/kestra/repository/h2/H2ExecutionRepositoryService.java
index 720d56d5380..b4e4a90822b 100644
--- a/jdbc-h2/src/main/java/io/kestra/repository/h2/H2ExecutionRepositoryService.java
+++ b/jdbc-h2/src/main/java/io/kestra/repository/h2/H2ExecutionRepositoryService.java
@@ -20,12 +20,14 @@ public static Condition findCondition(AbstractJdbcRepository<Execution> jdbcRepo
 
         if (labels != null) {
             labels.forEach((key, value) -> {
-                Field<String> field = DSL.field("JQ_STRING(\"value\", '.labels." + key + "')", String.class);
+                Field<String> keyField = DSL.field("JQ_STRING(\"value\", '.labels[].key')", String.class);
+                conditions.add(keyField.eq(key));
 
+                Field<String> valueField = DSL.field("JQ_STRING(\"value\", '.labels[].value')", String.class);
                 if (value == null) {
-                    conditions.add(field.isNotNull());
+                    conditions.add(valueField.isNotNull());
                 } else {
-                    conditions.add(field.eq(value));
+                    conditions.add(valueField.eq(value));
                 }
             });
         }
diff --git a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2FunctionsTest.java b/jdbc-h2/src/test/java/io/kestra/runner/h2/H2FunctionsTest.java
index 5a6554d90d2..a6ac3378de9 100644
--- a/jdbc-h2/src/test/java/io/kestra/runner/h2/H2FunctionsTest.java
+++ b/jdbc-h2/src/test/java/io/kestra/runner/h2/H2FunctionsTest.java
@@ -20,6 +20,14 @@ public void jqString() {
         assertThat(jqString, is("b"));
     }
 
+    @Test
+    public void jqStringWithArray() {
+        String jqString = H2Functions.jqString("""
+            {"a": [{"b": "c", "d": "e"}]}
+            """, ".a[].b");
+        assertThat(jqString, is("c"));
+    }
+
     @Test
     public void jqBoolean() {
         Boolean jqString = H2Functions.jqBoolean("{\"a\": true}", ".a");
diff --git a/jdbc-mysql/src/main/java/io/kestra/repository/mysql/MysqlExecutionRepositoryService.java b/jdbc-mysql/src/main/java/io/kestra/repository/mysql/MysqlExecutionRepositoryService.java
index 367450e30e0..1d4dbd0474d 100644
--- a/jdbc-mysql/src/main/java/io/kestra/repository/mysql/MysqlExecutionRepositoryService.java
+++ b/jdbc-mysql/src/main/java/io/kestra/repository/mysql/MysqlExecutionRepositoryService.java
@@ -21,12 +21,14 @@ public static Condition findCondition(AbstractJdbcRepository<Execution> jdbcRepo
 
         if (labels != null) {
             labels.forEach((key, value) -> {
-                Field<String> field = DSL.field("JSON_VALUE(value, '$.labels." + key + "' NULL ON EMPTY)", String.class);
+                Field<String> keyField = DSL.field("JSON_VALUE(value, '$.labels[*].key' NULL ON EMPTY)", String.class);
+                conditions.add(keyField.eq(key));
 
+                Field<String> valueField = DSL.field("JSON_VALUE(value, '$.labels[*].value' NULL ON EMPTY)", String.class);
                 if (value == null) {
-                    conditions.add(field.isNotNull());
+                    conditions.add(valueField.isNotNull());
                 } else {
-                    conditions.add(field.eq(value));
+                    conditions.add(valueField.eq(value));
                 }
             });
         }
diff --git a/jdbc-postgres/src/main/java/io/kestra/repository/postgres/PostgresExecutionRepositoryService.java b/jdbc-postgres/src/main/java/io/kestra/repository/postgres/PostgresExecutionRepositoryService.java
index d864b77216c..1728ace9582 100644
--- a/jdbc-postgres/src/main/java/io/kestra/repository/postgres/PostgresExecutionRepositoryService.java
+++ b/jdbc-postgres/src/main/java/io/kestra/repository/postgres/PostgresExecutionRepositoryService.java
@@ -21,13 +21,8 @@ public static Condition findCondition(AbstractJdbcRepository<Execution> jdbcRepo
 
         if (labels != null)  {
             labels.forEach((key, value) -> {
-                Field<String> field = DSL.field("value #>> '{labels, " + key + "}'", String.class);
-
-                if (value == null) {
-                    conditions.add(field.isNotNull());
-                } else {
-                    conditions.add(field.eq(value));
-                }
+                String sql = "value -> 'labels' @> '[{\"key\":\"" + key + "\", \"value\":\"" + value + "\"}]'";
+                conditions.add(DSL.condition(sql));
             });
         }
 
diff --git a/ui/src/components/layout/Labels.vue b/ui/src/components/layout/Labels.vue
index fbe706ad2d2..c912c1e33a0 100644
--- a/ui/src/components/layout/Labels.vue
+++ b/ui/src/components/layout/Labels.vue
@@ -2,7 +2,7 @@
     <span v-if="labels">
         <!-- 'el-check-tag' would be a better fit but it currently lacks customization -->
         <el-tag
-            v-for="(value, key) in labels"
+            v-for="(value, key) in labelMap"
             :key="key"
             :type="checked(key) ? 'info' : ''"
             class="me-1 labels"
@@ -29,6 +29,22 @@
                 default: true
             }
         },
+        // this is needed as flows uses a Map and Execution a List of Labels.
+        // if we align both of them this can be removed
+        mounted() {
+            if (Array.isArray(this.labels)) {
+                const labelMap = {};
+                this.labels.forEach(label => labelMap[label.key] = label.value)
+                this.labelMap = labelMap;
+            } else {
+                this.labelMap = this.labels;
+            }
+        },
+        data() {
+            return {
+                labelMap: {}
+            }
+        },
         methods: {
             getLabelsFromQuery() {
                 const labels = new Map();
diff --git a/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java b/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
index 6a9410510c0..ce338bf641f 100644
--- a/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
+++ b/webserver/src/main/java/io/kestra/webserver/controllers/ExecutionController.java
@@ -1,6 +1,7 @@
 package io.kestra.webserver.controllers;
 
 import io.kestra.core.exceptions.InternalException;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.tasks.Task;
 import io.kestra.core.models.validations.ManualConstraintViolation;
 import io.kestra.core.runners.FlowableUtils;
@@ -15,6 +16,7 @@
 import io.micronaut.data.model.Pageable;
 import io.micronaut.http.*;
 import io.micronaut.http.annotation.*;
+import io.micronaut.http.exceptions.HttpStatusException;
 import io.micronaut.http.multipart.StreamingFileUpload;
 import io.micronaut.http.server.types.files.StreamedFile;
 import io.micronaut.http.sse.Event;
@@ -462,7 +464,7 @@ public Execution trigger(
         Execution current = runnerUtils.newExecution(
             find.get(),
             (flow, execution) -> runnerUtils.typedInputs(flow, execution, inputs, files),
-            RequestUtils.toMap(labels)
+            parseLabels(labels)
         );
 
         executionQueue.emit(current);
@@ -496,6 +498,18 @@ public Execution trigger(
             .blockingGet();
     }
 
+    private List<Label> parseLabels(List<String> labels) {
+        return labels == null ? null : labels.stream()
+            .map(label ->  {
+                String[] split = label.split(":");
+                if (split.length != 2) {
+                    throw new HttpStatusException(HttpStatus.UNPROCESSABLE_ENTITY, "Invalid labels parameter");
+                }
+                return new Label(split[0], split[1]);
+            })
+            .toList();
+    }
+
     protected <T> HttpResponse<T> validateFile(String executionId, URI path, String redirect) {
         Optional<Execution> execution = executionRepository.findById(executionId);
         if (execution.isEmpty()) {
diff --git a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
index 19b28103219..cb1b41c280c 100644
--- a/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
+++ b/webserver/src/test/java/io/kestra/webserver/controllers/ExecutionControllerTest.java
@@ -1,6 +1,7 @@
 package io.kestra.webserver.controllers;
 
 import com.google.common.collect.ImmutableMap;
+import io.kestra.core.models.Label;
 import io.kestra.core.models.executions.Execution;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.flows.Flow;
@@ -130,10 +131,11 @@ void trigger() {
         assertThat(result.getInputs().get("file").toString(), startsWith("kestra:///io/kestra/tests/inputs/executions/"));
         assertThat(result.getInputs().containsKey("bool"), is(true));
         assertThat(result.getInputs().get("bool"), nullValue());
-        assertThat(result.getLabels().get("a"), is("label-1"));
-        assertThat(result.getLabels().get("b"), is("label-2"));
-        assertThat(result.getLabels().get("flow-label-1"), is("flow-label-1"));
-        assertThat(result.getLabels().get("flow-label-2"), is("flow-label-2"));
+        assertThat(result.getLabels().size(), is(4));
+        assertThat(result.getLabels().get(0), is(new Label("flow-label-1", "flow-label-1")));
+        assertThat(result.getLabels().get(1), is(new Label("flow-label-2", "flow-label-2")));
+        assertThat(result.getLabels().get(2), is(new Label("a", "label-1")));
+        assertThat(result.getLabels().get(3), is(new Label("b", "label-2")));
     }
 
     @Test
@@ -467,8 +469,8 @@ void webhook() {
         assertThat(((Map<String, Object>) execution.getTrigger().getVariables().get("body")).get("b"), is(true));
         assertThat(((Map<String, Object>) execution.getTrigger().getVariables().get("parameters")).get("name"), is(List.of("john")));
         assertThat(((Map<String, List<Integer>>) execution.getTrigger().getVariables().get("parameters")).get("age"), containsInAnyOrder("12", "13"));
-        assertThat(execution.getLabels().get("flow-label-1"), is("flow-label-1"));
-        assertThat(execution.getLabels().get("flow-label-2"), is("flow-label-2"));
+        assertThat(execution.getLabels().get(0), is(new Label("flow-label-1", "flow-label-1")));
+        assertThat(execution.getLabels().get(1), is(new Label("flow-label-2", "flow-label-2")));
 
         execution = client.toBlocking().retrieve(
             HttpRequest