Ensure plan enumeration & channel conversion are deterministic. Use ordered sets for scope/slot tracking, add stable cost tiebreaker, and cover with determinism tests.

novatechflow · novatechflow · commit 93fbadb35a11 · 2025-12-16T08:41:55.000+01:00
diff --git a/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/channels/ChannelConversionGraph.java b/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/channels/ChannelConversionGraph.java
@@ -47,15 +47,16 @@
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
 import java.util.function.ToDoubleFunction;
 import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
 
 /**
  * This graph contains a set of {@link ChannelConversion}s.
@@ -168,7 +169,7 @@ private Tree mergeTrees(Collection<Tree> trees) {
         final Tree firstTree = iterator.next();
         Bitmask combinationSettledIndices = new Bitmask(firstTree.settledDestinationIndices);
         int maxSettledIndices = combinationSettledIndices.cardinality();
-        final HashSet<ChannelDescriptor> employedChannelDescriptors = new HashSet<>(firstTree.employedChannelDescriptors);
+        final LinkedHashSet<ChannelDescriptor> employedChannelDescriptors = new LinkedHashSet<>(firstTree.employedChannelDescriptors);
         int maxVisitedChannelDescriptors = employedChannelDescriptors.size();
         double costs = firstTree.costs;
         TreeVertex newRoot = new TreeVertex(firstTree.root.channelDescriptor, firstTree.root.settledIndices);
@@ -222,7 +223,11 @@ public static class CostbasedTreeSelectionStrategy implements TreeSelectionStrat
 
         @Override
         public Tree select(Tree t1, Tree t2) {
-            return t1.costs <= t2.costs ? t1 : t2;
+            int cmp = Double.compare(t1.costs, t2.costs);
+            if (cmp == 0) {
+                cmp = t1.getDeterministicSignature().compareTo(t2.getDeterministicSignature());
+            }
+            return cmp <= 0 ? t1 : t2;
         }
 
     }
@@ -381,7 +386,7 @@ private ShortestTreeSearcher(OutputSlot<?> sourceOutput,
                 this.existingDestinationChannelIndices = new Bitmask();
 
                 this.collectExistingChannels(sourceChannel);
-                this.openChannelDescriptors = new HashSet<>(openChannels.size());
+                this.openChannelDescriptors = new LinkedHashSet<>(openChannels.size());
                 for (Channel openChannel : openChannels) {
                     this.openChannelDescriptors.add(openChannel.getDescriptor());
                 }
@@ -477,7 +482,9 @@ private Set<ChannelDescriptor> resolveSupportedChannels(final InputSlot<?> input
             final List<ChannelDescriptor> supportedInputChannels = owner.getSupportedInputChannels(input.getIndex());
             if (input.isLoopInvariant()) {
                 // Loop input is needed in several iterations and must therefore be reusable.
-                return supportedInputChannels.stream().filter(ChannelDescriptor::isReusable).collect(Collectors.toSet());
+                return supportedInputChannels.stream()
+                        .filter(ChannelDescriptor::isReusable)
+                        .collect(Collectors.toCollection(LinkedHashSet::new));
             } else {
                 return WayangCollections.asSet(supportedInputChannels);
             }
@@ -546,7 +553,7 @@ private void kernelizeChannelRequests() {
                 }
                 if (channelDescriptors.size() - numReusableChannels == 1) {
                     iterator.remove();
-                    channelDescriptors = new HashSet<>(channelDescriptors);
+                    channelDescriptors = new LinkedHashSet<>(channelDescriptors);
                     channelDescriptors.removeIf(channelDescriptor -> !channelDescriptor.isReusable());
                     kernelDestChannelDescriptorSetsToIndicesUpdates.add(new Tuple<>(channelDescriptors, indices));
                 }
@@ -575,7 +582,7 @@ private void kernelizeChannelRequests() {
          */
         private Tree searchTree() {
             // Prepare the recursive traversal.
-            final HashSet<ChannelDescriptor> visitedChannelDescriptors = new HashSet<>(16);
+            final LinkedHashSet<ChannelDescriptor> visitedChannelDescriptors = new LinkedHashSet<>(16);
             visitedChannelDescriptors.add(this.sourceChannelDescriptor);
 
             // Perform the traversal.
@@ -777,7 +784,7 @@ private Set<ChannelDescriptor> getSuccessorChannelDescriptors(ChannelDescriptor
             final Channel channel = this.existingChannels.get(descriptor);
             if (channel == null || this.openChannelDescriptors.contains(descriptor)) return null;
 
-            Set<ChannelDescriptor> result = new HashSet<>();
+            Set<ChannelDescriptor> result = new LinkedHashSet<>();
             for (ExecutionTask consumer : channel.getConsumers()) {
                 if (!consumer.getOperator().isAuxiliary()) continue;
                 for (Channel successorChannel : consumer.getOutputChannels()) {
@@ -988,7 +995,12 @@ private static class Tree {
          *
          * @see TreeVertex#channelDescriptor
          */
-        private final Set<ChannelDescriptor> employedChannelDescriptors = new HashSet<>();
+        private final Set<ChannelDescriptor> employedChannelDescriptors = new LinkedHashSet<>();
+
+        /**
+         * Cached deterministic signature for tie-breaking.
+         */
+        private String deterministicSignature;
 
         /**
          * The sum of the costs of all {@link TreeEdge}s of this instance.
@@ -1010,6 +1022,7 @@ static Tree singleton(ChannelDescriptor channelDescriptor, Bitmask settledIndice
             this.root = root;
             this.settledDestinationIndices = settledDestinationIndices;
             this.employedChannelDescriptors.add(root.channelDescriptor);
+            this.deterministicSignature = null;
         }
 
         /**
@@ -1033,6 +1046,21 @@ void reroot(ChannelDescriptor newRootChannelDescriptor,
             this.employedChannelDescriptors.add(newRootChannelDescriptor);
             this.settledDestinationIndices.orInPlace(newRootSettledIndices);
             this.costs += edge.costEstimate;
+            this.deterministicSignature = null;
+        }
+
+        private String getDeterministicSignature() {
+            if (this.deterministicSignature == null) {
+                final String descriptorSignature = this.employedChannelDescriptors.stream()
+                        .map(Object::toString)
+                        .sorted()
+                        .collect(Collectors.joining("|"));
+                final String indexSignature = StreamSupport.stream(this.settledDestinationIndices.spliterator(), false)
+                        .map(String::valueOf)
+                        .collect(Collectors.joining(","));
+                this.deterministicSignature = descriptorSignature + "#" + indexSignature;
+            }
+            return this.deterministicSignature;
         }
 
         @Override
@@ -1090,7 +1118,7 @@ private void copyEdgesFrom(TreeVertex that) {
          * @return a {@link Set} of said {@link ChannelConversion}s
          */
         private Set<ChannelConversion> getChildChannelConversions() {
-            Set<ChannelConversion> channelConversions = new HashSet<>();
+            Set<ChannelConversion> channelConversions = new LinkedHashSet<>();
             for (TreeEdge edge : this.outEdges) {
                 channelConversions.add(edge.channelConversion);
                 channelConversions.addAll(edge.destination.getChildChannelConversions());
diff --git a/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/enumeration/PlanEnumeration.java b/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/enumeration/PlanEnumeration.java
@@ -42,7 +42,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -91,7 +91,7 @@ public class PlanEnumeration {
      * Creates a new instance.
      */
     public PlanEnumeration() {
-        this(new HashSet<>(), new HashSet<>(), new HashSet<>());
+        this(new LinkedHashSet<>(), new LinkedHashSet<>(), new LinkedHashSet<>());
     }
 
     /**
diff --git a/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/enumeration/PlanImplementation.java b/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/optimizer/enumeration/PlanImplementation.java
@@ -53,6 +53,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.LinkedHashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -255,7 +256,7 @@ Collection<InputSlot<?>> findExecutionOperatorInputs(final InputSlot<?> someInpu
 
             // Discern LoopHeadOperator InputSlots and loop body InputSlots.
             final List<LoopImplementation.IterationImplementation> iterationImpls = loopImplementation.getIterationImplementations();
-            final Collection<InputSlot<?>> collector = new HashSet<>(innerInputs.size());
+            final Collection<InputSlot<?>> collector = new LinkedHashSet<>(innerInputs.size());
             for (InputSlot<?> innerInput : innerInputs) {
                 if (innerInput.getOwner() == loopSubplan.getLoopHead()) {
                     final LoopImplementation.IterationImplementation initialIterationImpl = iterationImpls.get(0);
@@ -329,7 +330,7 @@ Collection<Tuple<OutputSlot<?>, PlanImplementation>> findExecutionOperatorOutput
             // For all the iterations, return the potential OutputSlots.
             final List<LoopImplementation.IterationImplementation> iterationImpls =
                     loopImplementation.getIterationImplementations();
-            final Set<Tuple<OutputSlot<?>, PlanImplementation>> collector = new HashSet<>(iterationImpls.size());
+            final Set<Tuple<OutputSlot<?>, PlanImplementation>> collector = new LinkedHashSet<>(iterationImpls.size());
             for (LoopImplementation.IterationImplementation iterationImpl : iterationImpls) {
                 final Collection<Tuple<OutputSlot<?>, PlanImplementation>> outputsWithContext =
                         iterationImpl.getBodyImplementation().findExecutionOperatorOutputWithContext(innerOutput);
@@ -695,8 +696,8 @@ public double getSquashedCostEstimate() {
 
     private Tuple<List<ProbabilisticDoubleInterval>, List<Double>> getParallelOperatorJunctionAllCostEstimate(Operator operator) {
 
-        Set<Operator> inputOperators = new HashSet<>();
-        Set<Junction> inputJunction = new HashSet<>();
+        Set<Operator> inputOperators = new LinkedHashSet<>();
+        Set<Junction> inputJunction = new LinkedHashSet<>();
 
         List<ProbabilisticDoubleInterval> probalisticCost = new ArrayList<>();
         List<Double> squashedCost = new ArrayList<>();
diff --git a/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/util/WayangCollections.java b/wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/util/WayangCollections.java
@@ -24,7 +24,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
+import java.util.LinkedHashSet;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
@@ -59,7 +59,7 @@ public static <T> Set<T> asSet(Collection<T> collection) {
         if (collection instanceof Set<?>) {
             return (Set<T>) collection;
         }
-        return new HashSet<>(collection);
+        return new LinkedHashSet<>(collection);
     }
 
     /**
@@ -69,7 +69,7 @@ public static <T> Set<T> asSet(Iterable<T> iterable) {
         if (iterable instanceof Set<?>) {
             return (Set<T>) iterable;
         }
-        Set<T> set = new HashSet<>();
+        Set<T> set = new LinkedHashSet<>();
         for (T t : iterable) {
             set.add(t);
         }
@@ -80,7 +80,7 @@ public static <T> Set<T> asSet(Iterable<T> iterable) {
      * Provides the given {@code values} as {@link Set}.
      */
     public static <T> Set<T> asSet(T... values) {
-        Set<T> set = new HashSet<>(values.length);
+        Set<T> set = new LinkedHashSet<>(values.length);
         for (T value : values) {
             set.add(value);
         }
diff --git a/wayang-commons/wayang-core/src/test/java/org/apache/wayang/core/optimizer/channels/ChannelConversionGraphDeterminismTest.java b/wayang-commons/wayang-core/src/test/java/org/apache/wayang/core/optimizer/channels/ChannelConversionGraphDeterminismTest.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.wayang.core.optimizer.channels;
+
+import org.apache.wayang.core.api.Configuration;
+import org.apache.wayang.core.api.Job;
+import org.apache.wayang.core.optimizer.DefaultOptimizationContext;
+import org.apache.wayang.core.optimizer.OptimizationContext;
+import org.apache.wayang.core.optimizer.OptimizationUtils;
+import org.apache.wayang.core.optimizer.cardinality.CardinalityEstimate;
+import org.apache.wayang.core.plan.executionplan.Channel;
+import org.apache.wayang.core.plan.executionplan.ExecutionTask;
+import org.apache.wayang.core.plan.wayangplan.ExecutionOperator;
+import org.apache.wayang.core.plan.wayangplan.InputSlot;
+import org.apache.wayang.core.plan.wayangplan.OutputSlot;
+import org.apache.wayang.core.platform.ChannelDescriptor;
+import org.apache.wayang.core.platform.Junction;
+import org.apache.wayang.core.test.DummyExecutionOperator;
+import org.apache.wayang.core.test.DummyExternalReusableChannel;
+import org.apache.wayang.core.test.DummyNonReusableChannel;
+import org.apache.wayang.core.test.DummyReusableChannel;
+import org.apache.wayang.core.test.MockFactory;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ChannelConversionGraphDeterminismTest {
+
+    private static Supplier<ExecutionOperator> createDummyExecutionOperatorFactory(ChannelDescriptor channelDescriptor) {
+        return () -> {
+            ExecutionOperator execOp = new DummyExecutionOperator(1, 1, false);
+            execOp.getSupportedOutputChannels(0).add(channelDescriptor);
+            return execOp;
+        };
+    }
+
+    private static DefaultChannelConversion conversion(ChannelDescriptor source, ChannelDescriptor target) {
+        return new DefaultChannelConversion(source, target, createDummyExecutionOperatorFactory(target));
+    }
+
+    @Test
+    void channelConversionSelectionIsStable() {
+        List<String> first = computeJunctionFingerprint();
+        List<String> second = computeJunctionFingerprint();
+        assertEquals(first, second, "Channel conversion choices must be deterministic.");
+    }
+
+    private static List<String> computeJunctionFingerprint() {
+        Configuration configuration = new Configuration();
+        ChannelConversionGraph graph = new ChannelConversionGraph(configuration);
+        graph.add(conversion(DummyReusableChannel.DESCRIPTOR, DummyNonReusableChannel.DESCRIPTOR));
+        graph.add(conversion(DummyReusableChannel.DESCRIPTOR, DummyExternalReusableChannel.DESCRIPTOR));
+        graph.add(conversion(DummyExternalReusableChannel.DESCRIPTOR, DummyNonReusableChannel.DESCRIPTOR));
+        graph.add(conversion(DummyNonReusableChannel.DESCRIPTOR, DummyReusableChannel.DESCRIPTOR));
+
+        Job job = MockFactory.createJob(configuration);
+        OptimizationContext optimizationContext = new DefaultOptimizationContext(job);
+
+        DummyExecutionOperator sourceOperator = new DummyExecutionOperator(0, 1, false);
+        sourceOperator.getSupportedOutputChannels(0).add(DummyReusableChannel.DESCRIPTOR);
+        optimizationContext.addOneTimeOperator(sourceOperator)
+                .setOutputCardinality(0, new CardinalityEstimate(1000, 1000, 1d));
+
+        DummyExecutionOperator destOperator0 = new DummyExecutionOperator(1, 1, false);
+        destOperator0.getSupportedInputChannels(0).add(DummyNonReusableChannel.DESCRIPTOR);
+
+        DummyExecutionOperator destOperator1 = new DummyExecutionOperator(1, 1, false);
+        destOperator1.getSupportedInputChannels(0).add(DummyExternalReusableChannel.DESCRIPTOR);
+
+        Junction junction = graph.findMinimumCostJunction(
+                sourceOperator.getOutput(0),
+                Arrays.asList(destOperator0.getInput(0), destOperator1.getInput(0)),
+                optimizationContext,
+                false
+        );
+
+        return describeJunction(junction);
+    }
+
+    private static List<String> describeJunction(Junction junction) {
+        List<String> descriptorList = new ArrayList<>();
+        descriptorList.add(describeChannel(junction.getSourceChannel(), true));
+        for (int i = 0; i < junction.getNumTargets(); i++) {
+            descriptorList.add(describeChannel(junction.getTargetChannel(i), false));
+        }
+        return descriptorList;
+    }
+
+    private static String describeChannel(Channel channel, boolean isSourceChannel) {
+        if (channel == null) {
+            return "null";
+        }
+        List<String> descriptors = new ArrayList<>();
+        Channel cursor = channel;
+        while (cursor != null) {
+            descriptors.add(cursor.getDescriptor().toString() + (cursor.isCopy() ? ":copy" : ":orig"));
+            ExecutionTask producer = cursor.getProducer();
+            if (producer == null || producer.getNumInputChannels() == 0) {
+                break;
+            }
+            // If we are describing the top-level source channel (junction entry), stop once we reach the producer that
+            // has no inputs. For target channels, follow until the conversion tree ends.
+            if (isSourceChannel) {
+                cursor = producer.getNumInputChannels() == 0 ? null : producer.getInputChannel(0);
+            } else if (producer.getNumInputChannels() == 0) {
+                cursor = null;
+            } else {
+                cursor = producer.getInputChannel(0);
+            }
+        }
+        Collections.reverse(descriptors);
+        return descriptors.stream().collect(Collectors.joining("->"));
+    }
+}
diff --git a/wayang-commons/wayang-core/src/test/java/org/apache/wayang/core/optimizer/enumeration/PlanEnumerationDeterminismTest.java b/wayang-commons/wayang-core/src/test/java/org/apache/wayang/core/optimizer/enumeration/PlanEnumerationDeterminismTest.java
