feat: Add concat operation to constraint streams

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/BavetConstraintSessionFactory.java

@@ -11,12 +11,15 @@
 import java.util.SortedMap;
 import java.util.TreeMap;
 
+import ai.timefold.solver.constraint.streams.bavet.common.AbstractConcatNode;
 import ai.timefold.solver.constraint.streams.bavet.common.AbstractIfExistsNode;
 import ai.timefold.solver.constraint.streams.bavet.common.AbstractJoinNode;
 import ai.timefold.solver.constraint.streams.bavet.common.AbstractNode;
 import ai.timefold.solver.constraint.streams.bavet.common.BavetAbstractConstraintStream;
+import ai.timefold.solver.constraint.streams.bavet.common.BavetConcatConstraintStream;
 import ai.timefold.solver.constraint.streams.bavet.common.BavetIfExistsConstraintStream;
 import ai.timefold.solver.constraint.streams.bavet.common.BavetJoinConstraintStream;
+import ai.timefold.solver.constraint.streams.bavet.common.BavetStreamBinaryOperation;
 import ai.timefold.solver.constraint.streams.bavet.common.NodeBuildHelper;
 import ai.timefold.solver.constraint.streams.bavet.common.PropagationQueue;
 import ai.timefold.solver.constraint.streams.bavet.common.Propagator;
@@ -134,24 +137,28 @@ private long determineLayerIndex(AbstractNode node, NodeBuildHelper<Score_> buil
         if (node instanceof AbstractForEachUniNode<?>) { // ForEach nodes, and only they, are in layer 0.
             return 0;
         } else if (node instanceof AbstractJoinNode<?, ?, ?> joinNode) {
-            var nodeCreator = (BavetJoinConstraintStream<?>) buildHelper.getNodeCreatingStream(joinNode);
-            var leftParent = nodeCreator.getLeftParent();
-            var rightParent = nodeCreator.getRightParent();
-            var leftParentNode = buildHelper.findParentNode(leftParent);
-            var rightParentNode = buildHelper.findParentNode(rightParent);
-            return Math.max(leftParentNode.getLayerIndex(), rightParentNode.getLayerIndex()) + 1;
+            return determineLayerIndexOfBinaryOperation(
+                    (BavetJoinConstraintStream<?>) buildHelper.getNodeCreatingStream(joinNode), buildHelper);
+        } else if (node instanceof AbstractConcatNode<?> concatNode) {
+            return determineLayerIndexOfBinaryOperation(
+                    (BavetConcatConstraintStream<?>) buildHelper.getNodeCreatingStream(concatNode), buildHelper);
         } else if (node instanceof AbstractIfExistsNode<?, ?> ifExistsNode) {
-            var nodeCreator = (BavetIfExistsConstraintStream<?>) buildHelper.getNodeCreatingStream(ifExistsNode);
-            var leftParent = nodeCreator.getLeftParent();
-            var rightParent = nodeCreator.getRightParent();
-            var leftParentNode = buildHelper.findParentNode(leftParent);
-            var rightParentNode = buildHelper.findParentNode(rightParent);
-            return Math.max(leftParentNode.getLayerIndex(), rightParentNode.getLayerIndex()) + 1;
+            return determineLayerIndexOfBinaryOperation(
+                    (BavetIfExistsConstraintStream<?>) buildHelper.getNodeCreatingStream(ifExistsNode), buildHelper);
         } else {
             var nodeCreator = (BavetAbstractConstraintStream<?>) buildHelper.getNodeCreatingStream(node);
             var parentNode = buildHelper.findParentNode(nodeCreator.getParent());
             return parentNode.getLayerIndex() + 1;
         }
     }
 
+    private long determineLayerIndexOfBinaryOperation(BavetStreamBinaryOperation<?> nodeCreator,
+            NodeBuildHelper<Score_> buildHelper) {
+        var leftParent = nodeCreator.getLeftParent();
+        var rightParent = nodeCreator.getRightParent();
+        var leftParentNode = buildHelper.findParentNode(leftParent);
+        var rightParentNode = buildHelper.findParentNode(rightParent);
+        return Math.max(leftParentNode.getLayerIndex(), rightParentNode.getLayerIndex()) + 1;
+    }
+
 }

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/bi/BavetAbstractBiConstraintStream.java

@@ -359,6 +359,19 @@ public BiConstraintStream<A, B> distinct() {
         }
     }
 
+    @Override
+    public BiConstraintStream<A, B> concat(BiConstraintStream<A, B> otherStream) {
+        var other = (BavetAbstractBiConstraintStream<Solution_, A, B>) otherStream;
+        var leftBridge = new BavetForeBridgeBiConstraintStream<>(constraintFactory, this);
+        var rightBridge = new BavetForeBridgeBiConstraintStream<>(constraintFactory, other);
+        var concatStream = new BavetConcatBiConstraintStream<>(constraintFactory, leftBridge, rightBridge);
+        return constraintFactory.share(concatStream, concatStream_ -> {
+            // Connect the bridges upstream
+            getChildStreamList().add(leftBridge);
+            other.getChildStreamList().add(rightBridge);
+        });
+    }
+
     @Override
     public <ResultA_> UniConstraintStream<ResultA_> map(BiFunction<A, B, ResultA_> mapping) {
         var stream = shareAndAddChild(new BavetUniMapBiConstraintStream<>(constraintFactory, this, mapping));

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/bi/BavetBiConcatNode.java

@@ -0,0 +1,25 @@
+package ai.timefold.solver.constraint.streams.bavet.bi;
+
+import ai.timefold.solver.constraint.streams.bavet.common.AbstractConcatNode;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.BiTuple;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleLifecycle;
+
+public final class BavetBiConcatNode<A, B> extends AbstractConcatNode<BiTuple<A, B>> {
+
+    BavetBiConcatNode(TupleLifecycle<BiTuple<A, B>> nextNodesTupleLifecycle, int inputStoreIndexLeftOutTupleList,
+            int inputStoreIndexRightOutTupleList,
+            int outputStoreSize) {
+        super(nextNodesTupleLifecycle, inputStoreIndexLeftOutTupleList, inputStoreIndexRightOutTupleList, outputStoreSize);
+    }
+
+    @Override
+    protected BiTuple<A, B> getOutTuple(BiTuple<A, B> inTuple) {
+        return new BiTuple<>(inTuple.factA, inTuple.factB, outputStoreSize);
+    }
+
+    @Override
+    protected void updateOutTuple(BiTuple<A, B> inTuple, BiTuple<A, B> outTuple) {
+        outTuple.factA = inTuple.factA;
+        outTuple.factB = inTuple.factB;
+    }
+}

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/bi/BavetConcatBiConstraintStream.java

@@ -0,0 +1,104 @@
+package ai.timefold.solver.constraint.streams.bavet.bi;
+
+import java.util.Objects;
+import java.util.Set;
+
+import ai.timefold.solver.constraint.streams.bavet.BavetConstraintFactory;
+import ai.timefold.solver.constraint.streams.bavet.common.BavetAbstractConstraintStream;
+import ai.timefold.solver.constraint.streams.bavet.common.BavetConcatConstraintStream;
+import ai.timefold.solver.constraint.streams.bavet.common.NodeBuildHelper;
+import ai.timefold.solver.constraint.streams.bavet.common.bridge.BavetForeBridgeBiConstraintStream;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.BiTuple;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleLifecycle;
+import ai.timefold.solver.core.api.score.Score;
+
+public final class BavetConcatBiConstraintStream<Solution_, A, B> extends BavetAbstractBiConstraintStream<Solution_, A, B>
+        implements BavetConcatConstraintStream<Solution_> {
+
+    private final BavetForeBridgeBiConstraintStream<Solution_, A, B> leftParent;
+    private final BavetForeBridgeBiConstraintStream<Solution_, A, B> rightParent;
+
+    public BavetConcatBiConstraintStream(BavetConstraintFactory<Solution_> constraintFactory,
+            BavetForeBridgeBiConstraintStream<Solution_, A, B> leftParent,
+            BavetForeBridgeBiConstraintStream<Solution_, A, B> rightParent) {
+        super(constraintFactory, leftParent.getRetrievalSemantics());
+        this.leftParent = leftParent;
+        this.rightParent = rightParent;
+    }
+
+    @Override
+    public boolean guaranteesDistinct() {
+        return false;
+    }
+
+    // ************************************************************************
+    // Node creation
+    // ************************************************************************
+
+    @Override
+    public void collectActiveConstraintStreams(Set<BavetAbstractConstraintStream<Solution_>> constraintStreamSet) {
+        leftParent.collectActiveConstraintStreams(constraintStreamSet);
+        rightParent.collectActiveConstraintStreams(constraintStreamSet);
+        constraintStreamSet.add(this);
+    }
+
+    @Override
+    public <Score_ extends Score<Score_>> void buildNode(NodeBuildHelper<Score_> buildHelper) {
+        TupleLifecycle<BiTuple<A, B>> downstream = buildHelper.getAggregatedTupleLifecycle(childStreamList);
+        int leftCloneStoreIndex = buildHelper.reserveTupleStoreIndex(leftParent.getTupleSource());
+        int rightCloneStoreIndex = buildHelper.reserveTupleStoreIndex(rightParent.getTupleSource());
+        int outputStoreSize = buildHelper.extractTupleStoreSize(this);
+        var node = new BavetBiConcatNode<>(downstream,
+                leftCloneStoreIndex,
+                rightCloneStoreIndex,
+                outputStoreSize);
+        buildHelper.addNode(node, this, leftParent, rightParent);
+    }
+
+    // ************************************************************************
+    // Equality for node sharing
+    // ************************************************************************
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        BavetConcatBiConstraintStream<?, ?, ?> other = (BavetConcatBiConstraintStream<?, ?, ?>) o;
+        /*
+         * Bridge streams do not implement equality because their equals() would have to point back to this stream,
+         * resulting in StackOverflowError.
+         * Therefore we need to check bridge parents to see where this concat node comes from.
+         */
+        return Objects.equals(leftParent.getParent(), other.leftParent.getParent())
+                && Objects.equals(rightParent.getParent(), other.rightParent.getParent());
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(leftParent.getParent(), rightParent.getParent());
+    }
+
+    @Override
+    public String toString() {
+        return "Concat() with " + childStreamList.size() + " children";
+    }
+
+    // ************************************************************************
+    // Getters/setters
+    // ************************************************************************
+
+    @Override
+    public BavetAbstractConstraintStream<Solution_> getLeftParent() {
+        return leftParent;
+    }
+
+    @Override
+    public BavetAbstractConstraintStream<Solution_> getRightParent() {
+        return rightParent;
+    }
+
+}

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/common/AbstractConcatNode.java

@@ -0,0 +1,133 @@
+package ai.timefold.solver.constraint.streams.bavet.common;
+
+import static ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleState.ABORTING;
+import static ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleState.CREATING;
+import static ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleState.DYING;
+
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.AbstractTuple;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.LeftTupleLifecycle;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.RightTupleLifecycle;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleLifecycle;
+import ai.timefold.solver.constraint.streams.bavet.common.tuple.TupleState;
+
+/**
+ * Implements the concat operation. Concat cannot be implemented as a pass-through operation because of two caveats:
+ *
+ * <ul>
+ * <li>It is possible to have the same {@link TupleSource} for both parent streams,
+ * in which case the exact same tuple can be inserted twice. Such a tuple
+ * should be counted twice downstream, and thus need to be cloned.
+ * </li>
+ *
+ * <li>Because concat has two parent nodes, it must be a {@link TupleSource} (since
+ * all nodes have exactly one {@link TupleSource}, and the source tuple can come from
+ * either parent). {@link TupleSource} must produce new tuples and not reuse them, since
+ * if tuples are reused, the stores inside them get corrupted.
+ * </li>
+ * </ul>
+ *
+ * The {@link AbstractConcatNode} works by creating a copy of the source tuple and putting it into
+ * the tuple's store. If the same tuple is inserted twice (i.e. when the left and right parent
+ * have the same {@link TupleSource}), it creates another clone.
+ */
+public abstract class AbstractConcatNode<Tuple_ extends AbstractTuple>
+        extends AbstractNode
+        implements LeftTupleLifecycle<Tuple_>, RightTupleLifecycle<Tuple_> {
+    private final int leftSourceTupleCloneStoreIndex;
+    private final int rightSourceTupleCloneStoreIndex;
+    protected final int outputStoreSize;
+    private final StaticPropagationQueue<Tuple_> propagationQueue;
+
+    protected AbstractConcatNode(TupleLifecycle<Tuple_> nextNodesTupleLifecycle,
+            int leftSourceTupleCloneStoreIndex,
+            int rightSourceTupleCloneStoreIndex,
+            int outputStoreSize) {
+        this.propagationQueue = new StaticPropagationQueue<>(nextNodesTupleLifecycle);
+        this.leftSourceTupleCloneStoreIndex = leftSourceTupleCloneStoreIndex;
+        this.rightSourceTupleCloneStoreIndex = rightSourceTupleCloneStoreIndex;
+        this.outputStoreSize = outputStoreSize;
+    }
+
+    protected abstract Tuple_ getOutTuple(Tuple_ inTuple);
+
+    protected abstract void updateOutTuple(Tuple_ inTuple, Tuple_ outTuple);
+
+    @Override
+    public final void insertLeft(Tuple_ tuple) {
+        Tuple_ outTuple = getOutTuple(tuple);
+        tuple.setStore(leftSourceTupleCloneStoreIndex, outTuple);
+        propagationQueue.insert(outTuple);
+    }
+
+    @Override
+    public final void updateLeft(Tuple_ tuple) {
+        Tuple_ outTuple = tuple.getStore(leftSourceTupleCloneStoreIndex);
+        if (outTuple != null) {
+            updateOutTuple(tuple, outTuple);
+            propagationQueue.update(outTuple);
+        } else {
+            // this can happen when left and right have the same TupleSource,
+            // and it was inserted on the right (but not the left).
+            insertLeft(tuple);
+        }
+    }
+
+    @Override
+    public final void retractLeft(Tuple_ tuple) {
+        Tuple_ outTuple = tuple.getStore(leftSourceTupleCloneStoreIndex);
+        if (outTuple == null) {
+            // this can happen when left and right have the same TupleSource,
+            // and it was inserted on the right (but not the left).
+            return;
+        }
+
+        TupleState state = outTuple.state;
+        if (!state.isActive()) {
+            throw new IllegalStateException("Impossible state: The tuple (" + outTuple.state + ") in node (" + this
+                    + ") is in an unexpected state (" + outTuple.state + ").");
+        }
+        propagationQueue.retract(outTuple, state == CREATING ? ABORTING : DYING);
+    }
+
+    @Override
+    public final void insertRight(Tuple_ tuple) {
+        Tuple_ outTuple = getOutTuple(tuple);
+        tuple.setStore(rightSourceTupleCloneStoreIndex, outTuple);
+        propagationQueue.insert(outTuple);
+    }
+
+    @Override
+    public final void updateRight(Tuple_ tuple) {
+        Tuple_ outTuple = tuple.getStore(rightSourceTupleCloneStoreIndex);
+        if (outTuple != null) {
+            updateOutTuple(tuple, outTuple);
+            propagationQueue.update(outTuple);
+        } else {
+            // this can happen when left and right have the same TupleSource,
+            // and it was inserted on the left (but not the right).
+            insertRight(tuple);
+        }
+    }
+
+    @Override
+    public final void retractRight(Tuple_ tuple) {
+        Tuple_ outTuple = tuple.getStore(rightSourceTupleCloneStoreIndex);
+        if (outTuple == null) {
+            // this can happen when left and right have the same TupleSource,
+            // and it was inserted on the left (but not the right).
+            return;
+        }
+
+        TupleState state = outTuple.state;
+        if (!state.isActive()) {
+            throw new IllegalStateException("Impossible state: The tuple (" + outTuple.state + ") in node (" + this
+                    + ") is in an unexpected state (" + outTuple.state + ").");
+        }
+        propagationQueue.retract(outTuple, state == CREATING ? ABORTING : DYING);
+    }
+
+    @Override
+    public Propagator getPropagator() {
+        return propagationQueue;
+    }
+}

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/common/BavetConcatConstraintStream.java

@@ -0,0 +1,6 @@
+package ai.timefold.solver.constraint.streams.bavet.common;
+
+public interface BavetConcatConstraintStream<Solution_>
+        extends BavetStreamBinaryOperation<Solution_>, TupleSource {
+
+}

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/common/BavetIfExistsConstraintStream.java

@@ -1,15 +1,5 @@
 package ai.timefold.solver.constraint.streams.bavet.common;
 
-import ai.timefold.solver.constraint.streams.bavet.common.bridge.BavetForeBridgeUniConstraintStream;
-
-public interface BavetIfExistsConstraintStream<Solution_> {
-
-    BavetAbstractConstraintStream<Solution_> getLeftParent();
-
-    /**
-     *
-     * @return An instance of {@link BavetForeBridgeUniConstraintStream}.
-     */
-    BavetAbstractConstraintStream<Solution_> getRightParent();
+public interface BavetIfExistsConstraintStream<Solution_> extends BavetStreamBinaryOperation<Solution_> {
 
 }

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/common/BavetJoinConstraintStream.java

@@ -3,7 +3,7 @@
 import ai.timefold.solver.constraint.streams.bavet.common.bridge.BavetForeBridgeUniConstraintStream;
 
 public interface BavetJoinConstraintStream<Solution_>
-        extends TupleSource {
+        extends BavetStreamBinaryOperation<Solution_>, TupleSource {
 
     /**
      *

core/constraint-streams/src/main/java/ai/timefold/solver/constraint/streams/bavet/common/BavetStreamBinaryOperation.java

@@ -0,0 +1,16 @@
+package ai.timefold.solver.constraint.streams.bavet.common;
+
+import ai.timefold.solver.constraint.streams.bavet.common.bridge.BavetForeBridgeUniConstraintStream;
+
+public interface BavetStreamBinaryOperation<Solution_> {
+    /**
+     * @return An instance of {@link BavetForeBridgeUniConstraintStream}.
+     */
+    BavetAbstractConstraintStream<Solution_> getLeftParent();
+
+    /**
+     * @return An instance of {@link BavetForeBridgeUniConstraintStream}.
+     */
+    BavetAbstractConstraintStream<Solution_> getRightParent();
+
+}