[ES|QL] Refactor MV_UNION and MV_INTERSECTION to use shared set operation helper

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvIntersection.java

@@ -38,7 +38,6 @@
 
 import java.io.IOException;
 import java.util.HashSet;
-import java.util.LinkedHashSet;
 import java.util.Objects;
 import java.util.Set;
 import java.util.function.BiFunction;
@@ -286,33 +285,37 @@ static <T> void processIntersectionSet(
     ) {
         int firstValueCount = field1.getValueCount(position);
         int secondValueCount = field2.getValueCount(position);
+
         if (firstValueCount == 0 || secondValueCount == 0) {
-            // if either block has no values, there will be no intersection
+            // If either block has no values, there will be no intersection
             builder.appendNull();
             return;
         }
 
+        // Extract values from first field into OperationalSet (preserves order)
+        MvSetOperationHelper.OperationalSet<T> firstSet = new MvSetOperationHelper.OperationalSet<>();
         int firstValueIndex = field1.getFirstValueIndex(position);
-        int secondValueIndex = field2.getFirstValueIndex(position);
-
-        Set<T> values = new LinkedHashSet<>();
         for (int i = 0; i < firstValueCount; i++) {
-            values.add(getValueFunction.apply(firstValueIndex + i, field1));
+            firstSet.add(getValueFunction.apply(firstValueIndex + i, field1));
         }
 
-        Set<T> secondValues = new HashSet<>();
+        // Extract values from second field (HashSet - order doesn't matter for lookup)
+        Set<T> secondSet = new HashSet<>();
+        int secondValueIndex = field2.getFirstValueIndex(position);
         for (int i = 0; i < secondValueCount; i++) {
-            secondValues.add(getValueFunction.apply(secondValueIndex + i, field2));
+            secondSet.add(getValueFunction.apply(secondValueIndex + i, field2));
         }
 
-        values.retainAll(secondValues);
-        if (values.isEmpty()) {
+        // Compute intersection in-place
+        Set<T> result = firstSet.intersect(secondSet);
+
+        if (result.isEmpty()) {
             builder.appendNull();
             return;
         }
 
         builder.beginPositionEntry();
-        for (T value : values) {
+        for (T value : result) {
             addValueFunction.accept(value);
         }
         builder.endPositionEntry();

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSetOperationHelper.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.expression.function.scalar.multivalue;
+
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * Shared set operations for MV functions.
+ * Preserves insertion order using LinkedHashSet.
+ */
+public final class MvSetOperationHelper {
+
+    private MvSetOperationHelper() {}
+
+    public static class OperationalSet<E> extends LinkedHashSet<E> {
+
+        /**
+         * Performs an in-place union with the given set.
+         * Adds all elements from the given set to this set.
+         * @param set the set to union with
+         * @return this set after the union operation
+         */
+        public Set<E> union(Set<E> set) {
+            this.addAll(set);
+            return this;
+        }
+
+        /**
+         * Performs an in-place intersection with the given set.
+         * Retains only elements that are present in both sets.
+         * @param set the set to intersect with
+         * @return this set after the intersection operation
+         */
+        public Set<E> intersect(Set<E> set) {
+            this.retainAll(set);
+            return this;
+        }
+    }
+}

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvUnion.java

@@ -299,31 +299,32 @@ static <T> void processUnionSet(
         int firstValueCount = field1.getValueCount(position);
         int secondValueCount = field2.getValueCount(position);
 
-        // If both field has no values, return null
+        // If both fields have no values, return null
         if (firstValueCount == 0 && secondValueCount == 0) {
             builder.appendNull();
             return;
         }
 
+        // Extract values from first field into OperationalSet
+        MvSetOperationHelper.OperationalSet<T> firstSet = new MvSetOperationHelper.OperationalSet<>();
         int firstValueIndex = field1.getFirstValueIndex(position);
-        int secondValueIndex = field2.getFirstValueIndex(position);
-
-        // Use LinkedHashSet to maintain insertion order
-        Set<T> values = new LinkedHashSet<>();
-
-        // Add all values from first field
         for (int i = 0; i < firstValueCount; i++) {
-            values.add(getValueFunction.apply(firstValueIndex + i, field1));
+            firstSet.add(getValueFunction.apply(firstValueIndex + i, field1));
         }
 
-        // Add all values from second field (duplicates automatically ignored by Set)
+        // Extract values from second field
+        Set<T> secondSet = new LinkedHashSet<>();
+        int secondValueIndex = field2.getFirstValueIndex(position);
         for (int i = 0; i < secondValueCount; i++) {
-            values.add(getValueFunction.apply(secondValueIndex + i, field2));
+            secondSet.add(getValueFunction.apply(secondValueIndex + i, field2));
         }
 
+        // Compute union in-place
+        Set<T> result = firstSet.union(secondSet);
+
         // Build result
         builder.beginPositionEntry();
-        for (T value : values) {
+        for (T value : result) {
             addValueFunction.accept(value);
         }
         builder.endPositionEntry();