The DependencyTracker makes sure that prefetched DependencyInfos have…

… the right partition id when we try to match up query invocations at run time.
apavlo · May 30, 2013 · 3bc4fbe · 3bc4fbe
1 parent 1906946
commit 3bc4fbe
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 6 deletions.
diff --git a/src/frontend/edu/brown/hstore/callbacks/LocalInitQueueCallback.java b/src/frontend/edu/brown/hstore/callbacks/LocalInitQueueCallback.java
@@ -62,7 +62,10 @@ public void run(int partition) {
         if (trace.val)
             LOG.trace(String.format("%s - Prefetch=%s / HasPrefetchFragments=%s",
                       this.ts, this.prefetch, this.ts.hasPrefetchFragments()));
-        if (this.prefetch && partition != this.ts.getBasePartition() && this.ts.hasPrefetchFragments()) {
+        if (this.prefetch && 
+                this.ts.hasPrefetchFragments() &&
+                partition != this.ts.getBasePartition() &&
+                hstore_site.isLocalPartition(partition)) {
             if (debug.val)
                 LOG.debug(String.format("%s - Checking for prefetch queries at partition %d",
                           this.ts, partition));

diff --git a/src/frontend/edu/brown/hstore/txns/DependencyTracker.java b/src/frontend/edu/brown/hstore/txns/DependencyTracker.java
@@ -424,6 +424,7 @@ private DependencyInfo getOrCreateDependencyInfo(LocalTransaction ts,
      * @param state
      * @param round
      * @param stmtCounter
+     * @param partitionId TODO
      * @param paramsHash
      * @param fragmentId
      * @param dependencyId
@@ -433,6 +434,7 @@ private DependencyInfo getPrefetchDependencyInfo(TransactionState state,
                                                      int round,
                                                      int stmtCounter,
                                                      int stmtIndex,
+                                                     int partitionId,
                                                      int paramsHash,
                                                      int fragmentId,
                                                      int dependencyId) {
@@ -462,6 +464,14 @@ private DependencyInfo getPrefetchDependencyInfo(TransactionState state,
                           dinfo.getParameterSetHash(), paramsHash));
             return (null);
         }
+        else if (dinfo.getExpectedPartitions().contains(partitionId) == false) {
+            if (trace.val)
+                LOG.trace(String.format("%s - Invalid prefetch query for %s. " +
+                          "Partition mismatch [%d != %d]",
+                          state, TransactionUtil.debugStmtDep(stmtCounter, dependencyId),
+                          partitionId, dinfo.getExpectedPartitions()));
+            return (null);
+        }
 
         // IMPORTANT: We have to update this DependencyInfo's output id 
         // so that the blocked WorkFragment can retrieve it properly when it
@@ -611,6 +621,7 @@ public boolean addWorkFragment(LocalTransaction ts, WorkFragment.Builder fragmen
         int output_dep_id, input_dep_id;
         int ignore_ctr = 0;
         for (int i = 0; i < num_fragments; i++) {
+            int partitionId = fragment.getPartitionId();
             int fragmentId = fragment.getFragmentId(i);
             int stmtCounter = fragment.getStmtCounter(i);
             int stmtIndex = fragment.getStmtIndex(i);
@@ -626,8 +637,8 @@ public boolean addWorkFragment(LocalTransaction ts, WorkFragment.Builder fragmen
                 // this same query invocation.
                 if (state.prefetch_ctr > 0) {
                     dinfo = this.getPrefetchDependencyInfo(state, currentRound,
-                                                           stmtCounter, stmtIndex, paramsHash,
-                                                           fragmentId, output_dep_id);
+                                                           stmtCounter, stmtIndex, partitionId,
+                                                           paramsHash, fragmentId, output_dep_id);
                     prefetch = (dinfo != null);
 
                 }
@@ -690,8 +701,8 @@ public boolean addWorkFragment(LocalTransaction ts, WorkFragment.Builder fragmen
                     // generate this result for us.
                     if (state.prefetch_ctr > 0) {
                         dinfo = this.getPrefetchDependencyInfo(state, currentRound,
-                                                               stmtCounter, stmtIndex, paramsHash,
-                                                               fragmentId, input_dep_id);
+                                                               stmtCounter, stmtIndex, partitionId,
+                                                               paramsHash, fragmentId, input_dep_id);
                     }
                     if (dinfo == null) {
                         dinfo = this.getOrCreateDependencyInfo(ts, state, currentRound,
@@ -1038,7 +1049,7 @@ public void addPrefetchWorkFragment(LocalTransaction ts, WorkFragment.Builder fr
 
             if (debug.val) {
                 String msg = String.format("%s - Adding prefetch %s %s at partition %d for %s",
-                                           ts, dinfo.getClass().getSimpleName(),
+                                           ts, dinfo,
                                            TransactionUtil.debugStmtDep(stmtCounter, output_dep_id), partition,
                                            CatalogUtil.getPlanFragment(catalogContext.catalog, fragment.getFragmentId(i)).fullName());
                 if (trace.val)