Extensible query planner & relation evaluation fixes #139

* The Rete recipe compiler is now extensible through the use of
ExternalQueryPlanProvider.
* JoinCandidate visibility is changed, so that
ExternalQueryPlanProviders can reuse the class.
* The communication tracker correctly considers the dependencies between
relation evaluator nodes, batching receivers, and the batching
receivers' transitive sources.
* Small performance improvement in RelationEvaluatorNode to guard
against spurious batchUpdateCompleted calls as much as possible.

Signed-off-by: Tamas Szabo <szabta89@github.com>
Change-Id: I948e80a3057f63416a78c072073107960678adeb

query/plugins/org.eclipse.viatra.query.runtime.matchers/src/org/eclipse/viatra/query/runtime/matchers/backend/CommonQueryHintOptions.java

@@ -8,6 +8,7 @@
  *******************************************************************************/
 package org.eclipse.viatra.query.runtime.matchers.backend;
 
+import org.eclipse.viatra.query.runtime.matchers.psystem.rewriters.ExternalQueryPlanProvider;
 import org.eclipse.viatra.query.runtime.matchers.psystem.rewriters.IRewriterTraceCollector;
 import org.eclipse.viatra.query.runtime.matchers.psystem.rewriters.NopTraceCollector;
 
@@ -28,6 +29,15 @@ private CommonQueryHintOptions() {
     public static final QueryHintOption<IRewriterTraceCollector> normalizationTraceCollector = 
             hintOption("normalizationTraceCollector", NopTraceCollector.INSTANCE);
 
+    /**
+     * This hint allows to plug in an external query planner.
+     * 
+     * @since 2.8
+     */
+    public static final QueryHintOption<ExternalQueryPlanProvider> externalQueryPlanProvider = 
+            hintOption("externalQueryPlanProvider", null);
+
+
     // internal helper for conciseness
     private static <T> QueryHintOption<T> hintOption(String hintKeyLocalName, T defaultValue) {
         return new QueryHintOption<>(CommonQueryHintOptions.class, hintKeyLocalName, defaultValue);

query/plugins/org.eclipse.viatra.query.runtime.matchers/src/org/eclipse/viatra/query/runtime/matchers/psystem/rewriters/ExternalQueryPlanProvider.java

@@ -0,0 +1,26 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2017, Grill Balázs, IncQueryLabs
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-v20.html.
+ * 
+ * SPDX-License-Identifier: EPL-2.0
+ *******************************************************************************/
+package org.eclipse.viatra.query.runtime.matchers.psystem.rewriters;
+
+import org.eclipse.viatra.query.runtime.matchers.backend.CommonQueryHintOptions;
+import org.eclipse.viatra.query.runtime.matchers.planning.SubPlan;
+import org.eclipse.viatra.query.runtime.matchers.psystem.PBody;
+
+/**
+ * An implementation of this interface can provide an execution plan for a {@link PBody}.
+ * The plan provider can be registered via a query hint (see {@link CommonQueryHintOptions#externalQueryPlanProvider}), 
+ * and it will be used by the Rete recipe compiler. 
+ * 
+ * @since 2.8
+ */
+public interface ExternalQueryPlanProvider {
+
+    public SubPlan getPlan(PBody body);
+
+}

query/plugins/org.eclipse.viatra.query.runtime.rete/src/org/eclipse/viatra/query/runtime/rete/construction/quasitree/JoinCandidate.java

@@ -30,7 +30,7 @@
  * @author Gabor Bergmann
  * 
  */
-class JoinCandidate {
+public class JoinCandidate {
     private QueryAnalyzer analyzer;
 
     SubPlan primary;
@@ -44,7 +44,7 @@ class JoinCandidate {
     List<PConstraint> consSecondary;
 
 
-    JoinCandidate(SubPlan primary, SubPlan secondary, QueryAnalyzer analyzer) {
+    public JoinCandidate(SubPlan primary, SubPlan secondary, QueryAnalyzer analyzer) {
         super();
         this.primary = primary;
         this.secondary = secondary;

query/plugins/org.eclipse.viatra.query.runtime.rete/src/org/eclipse/viatra/query/runtime/rete/eval/RelationEvaluatorNode.java

@@ -9,11 +9,13 @@
 package org.eclipse.viatra.query.runtime.rete.eval;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.stream.IntStream;
 
 import org.eclipse.viatra.query.runtime.matchers.psystem.IRelationEvaluator;
 import org.eclipse.viatra.query.runtime.matchers.psystem.basicdeferred.RelationEvaluation;
@@ -28,6 +30,7 @@
 import org.eclipse.viatra.query.runtime.rete.network.ReteContainer;
 import org.eclipse.viatra.query.runtime.rete.network.StandardNode;
 import org.eclipse.viatra.query.runtime.rete.network.Supplier;
+import org.eclipse.viatra.query.runtime.rete.network.communication.CommunicationTracker;
 import org.eclipse.viatra.query.runtime.rete.network.communication.Timestamp;
 import org.eclipse.viatra.query.runtime.rete.single.AbstractUniquenessEnforcerNode;
 
@@ -46,20 +49,29 @@ public class RelationEvaluatorNode extends StandardNode implements Supplier, Cle
     private Supplier[] inputSuppliers;
     private BatchingReceiver[] inputReceivers;
 
-    public RelationEvaluatorNode(final ReteContainer container, final IRelationEvaluator evaluator) {
+    public RelationEvaluatorNode(final ReteContainer container, final IRelationEvaluator evaluator,
+            final BatchingReceiver[] inputReceivers) {
         super(container);
         this.evaluator = evaluator;
+        this.inputReceivers = inputReceivers;
+        for (final BatchingReceiver inputReceiver : inputReceivers) {
+            inputReceiver.setContainerNode(this);
+        }
         this.reteContainer.registerClearable(this);
     }
-
+
+    public RelationEvaluatorNode(final ReteContainer container, final IRelationEvaluator evaluator) {
+        this(container, evaluator, IntStream.range(0, evaluator.getInputArities().size())
+                .mapToObj(i -> new BatchingReceiver(container)).toArray(BatchingReceiver[]::new));
+    }
+
     @Override
     public void clear() {
         this.cachedOutputs.clear();
     }
 
     public void connectToParents(final List<Supplier> inputSuppliers) {
         this.inputSuppliers = new Supplier[inputSuppliers.size()];
-        this.inputReceivers = new BatchingReceiver[inputSuppliers.size()];
 
         final List<Integer> inputArities = evaluator.getInputArities();
 
@@ -85,10 +97,10 @@ public void connectToParents(final List<Supplier> inputSuppliers) {
                         evaluator.toString() + " expects input arity " + currentExpectedInputArity + " at position " + i
                                 + " but got " + currentActualInputArity + "!");
             }
-            final BatchingReceiver inputReceiver = new BatchingReceiver((ProductionNode) inputSupplier,
-                    this.reteContainer);
+            final BatchingReceiver inputReceiver = this.inputReceivers[i];
+            inputReceiver.setSourceNode((ProductionNode) inputSupplier);
+
             this.inputSuppliers[i] = inputSupplier;
-            this.inputReceivers[i] = inputReceiver;
             this.reteContainer.connectAndSynchronize(inputSupplier, inputReceiver);
             reteContainer.getCommunicationTracker().registerDependency(inputReceiver, this);
         }
@@ -153,16 +165,32 @@ private void batchUpdateCompleted() {
         this.cachedOutputs = newOutputs;
     }
 
-    public class BatchingReceiver extends SimpleReceiver {
-        private final ProductionNode source;
+    public static class BatchingReceiver extends SimpleReceiver {
+        private ProductionNode sourceNode;
+        private RelationEvaluatorNode containerNode;
 
-        private BatchingReceiver(final ProductionNode source, final ReteContainer container) {
+        private BatchingReceiver(final ReteContainer container) {
             super(container);
-            this.source = source;
+        }
+
+        public void setSourceNode(final ProductionNode sourceNode) {
+            this.sourceNode = sourceNode;
+        }
+
+        public ProductionNode getSourceNode() {
+            return this.sourceNode;
+        }
+
+        public void setContainerNode(final RelationEvaluatorNode containerNode) {
+            this.containerNode = containerNode;
+        }
+
+        public RelationEvaluatorNode getContainerNode() {
+            return this.containerNode;
         }
 
         private Set<Tuple> getTuples() {
-            return ((AbstractUniquenessEnforcerNode) this.source).getTuples();
+            return ((AbstractUniquenessEnforcerNode) this.sourceNode).getTuples();
         }
 
         @Override
@@ -173,9 +201,25 @@ public void update(final Direction direction, final Tuple updateElement, final T
         @Override
         public void batchUpdate(final Collection<Entry<Tuple, Integer>> updates, final Timestamp timestamp) {
             assert Timestamp.ZERO.equals(timestamp);
-            // there is nothing to do here because the source production node has already updated itself
-            // the only thing we need to do is to issue the callback
-            RelationEvaluatorNode.this.batchUpdateCompleted();
+            // The source production node has already updated itself, so there is no need to do anything with the input
+            // updates. We will just use the tuples maintained in the memory of the production node.
+            // However, we should guard against spurious calls to the evaluation logic as much as possible, and only
+            // really issue the call if this is the "last" batchUpdate call among all the batching receivers of this
+            // relation evaluator node. It can happen that certain batching receivers are "lacking behind" because
+            // their ancestor may not have processed their updates yet. In such cases, the batchUpdateCompleted will
+            // be called potentially unnecessarily again, which is an issue for performance but not an issue for
+            // correctness.
+            final CommunicationTracker tracker = this.containerNode.getCommunicationTracker();
+            if (Arrays.stream(this.containerNode.inputReceivers).noneMatch(receiver -> {
+                return tracker.isEnqueued(receiver);
+            })) {
+                this.containerNode.batchUpdateCompleted();
+            }
+        }
+
+        @Override
+        protected String getTraceInfoPatternsEnumerated() {
+            return this.getContainerNode().toString();
         }
 
     }

query/plugins/org.eclipse.viatra.query.runtime.rete/src/org/eclipse/viatra/query/runtime/rete/network/communication/CommunicationGroup.java

@@ -97,7 +97,7 @@ public boolean equals(final Object obj) {
 
     @Override
     public int compareTo(final CommunicationGroup that) {
-        return this.identifier - that.identifier;
+        return Integer.compare(this.identifier, that.identifier);
     }
 
 }

query/plugins/org.eclipse.viatra.query.runtime.rete/src/org/eclipse/viatra/query/runtime/rete/network/communication/CommunicationTracker.java

@@ -98,6 +98,13 @@ public CommunicationTracker() {
         this.groupQueue = new PriorityQueue<CommunicationGroup>();
         this.groupMap = new HashMap<Node, CommunicationGroup>();
     }
+
+    /**
+     * @since 2.8
+     */
+    public boolean isEnqueued(final Node node) {
+        return this.groupQueue.contains(this.getGroup(node));
+    }
 
     public Graph<Node> getDependencyGraph() {
         return dependencyGraph;