check point

Author: Ultra-Seven

pom.xml

@@ -118,5 +118,10 @@
 	    <artifactId>affinity</artifactId>
 	    <version>3.2.2</version>
 	</dependency>
+      <dependency>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-simple</artifactId>
+          <version>2.0.0-alpha1</version>
+      </dependency>
   </dependencies>
 </project>

src/config/ParallelConfig.java

@@ -24,7 +24,7 @@ public class ParallelConfig {
 	/**
 	 * Whether to collect statistics of all constraints.
 	 */
-	public final static boolean CONSTRAINTS = true;
+	public final static boolean CONSTRAINTS = false;
 	/**
 	 * Maximal number of tuples per batch during execution.
 	 */
@@ -36,15 +36,23 @@ public class ParallelConfig {
 	/**
 	 * The minimal size of sparse columns.
 	 */
-	public static int SPARSE_KEY_SIZE = 10000;
+	public final static int SPARSE_KEY_SIZE = 10000;
 	/**
 	 * The minimal size of sparse columns.
 	 */
-	public static int SPARSE_FILTER_SIZE = 100;
+	public final static int SPARSE_FILTER_SIZE = 100;
 	/**
 	 * The minimum size of partitioned table
 	 */
-	public static int PARTITION_SIZE = 50000;
+	public final static int PARTITION_SIZE = 50000;
+	/**
+	 * The maximum size of statistics
+	 */
+	public final static int STATISTICS_SIZE = 20;
+	/**
+	 * Whether to assign constraint per thread.
+	 */
+	public final static boolean CONSTRAINT_PER_THREAD = true;
 	/**
 	 * The base of round counts to assign a new best join order
 	 * to executor thread.

src/indexing/Index.java

@@ -38,10 +38,6 @@ public abstract class Index {
 	 * Whether it is unique key.
 	 */
 	public boolean sorted = true;
-	/**
-	 * Whether it is unique key.
-	 */
-	public boolean filterSorted = true;
 	/**
 	 * Initialize for given cardinality of indexed table.
 	 * 

src/joining/parallel/indexing/IntPartitionIndex.java

@@ -175,6 +175,7 @@ else if (policy == IndexPolicy.Sparse) {
                 }
             });
         }
+        int nrPos = positions == null ? 0 : positions.length;
     }
 
     /**

src/joining/parallel/join/DPJoin.java

@@ -86,6 +86,10 @@ public abstract class DPJoin {
      * Last large table.
      */
     public int largeTable;
+    /**
+     * Last large table.
+     */
+    public int deepIndex;
     /**
      * Whether we have progress on the split table.
      */

src/joining/parallel/join/JoinDoublePartitionWrapper.java

@@ -27,6 +27,11 @@ public JoinDoublePartitionWrapper(ExpressionInfo equiPred, int[] order) {
         nextDoubleIndex = (DoublePartitionIndex)nextIndex;
     }
 
+    @Override
+    public void reset(int[] tupleIndices) {
+
+    }
+
     @Override
     public int nextIndex(int[] tupleIndices, int[] nextSize) {
         int priorTuple = tupleIndices[priorTable];
@@ -35,6 +40,16 @@ public int nextIndex(int[] tupleIndices, int[] nextSize) {
         return nextDoubleIndex.nextTuple(priorVal, curTuple, nextTable, nextSize);
     }
 
+    @Override
+    public int nextIndexFromLast(int[] tupleIndices, int[] nextSize) {
+        return 0;
+    }
+
+    @Override
+    public int nextIndexFromLast(int[] tupleIndices, int[] nextSize, int tid) {
+        return 0;
+    }
+
     @Override
     public int nextIndexInScope(int[] tupleIndices, int tid, int[] nextSize) {
         int priorTuple = tupleIndices[priorTable];

src/joining/parallel/join/JoinIntPartitionWrapper.java

@@ -27,6 +27,19 @@ public JoinIntPartitionWrapper(ExpressionInfo equiPred, int[] order) throws Exce
         nextIntIndex = (IntPartitionIndex)nextIndex;
     }
 
+    @Override
+    public void reset(int[] tupleIndices) {
+        int priorTuple = tupleIndices[priorTable];
+        int priorVal = priorIntData.data[priorTuple];
+        int prevTuple = tupleIndices[nextTable];
+        if (!(lastValue == priorVal &&
+                lastPositionsStart >= 0 &&
+                nextIntIndex.positions[lastPositionsStart] <= prevTuple)) {
+            lastPositionsStart = -1;
+            lastPositionsEnd = -1;
+        }
+    }
+
     @Override
     public int nextIndex(int[] tupleIndices, int[] nextSize) {
         int priorTuple = tupleIndices[priorTable];
@@ -35,6 +48,64 @@ public int nextIndex(int[] tupleIndices, int[] nextSize) {
         return nextIntIndex.nextTuple(priorVal, curTuple, nextTable, nextSize);
     }
 
+    @Override
+    public int nextIndexFromLast(int[] tupleIndices, int[] nextSize) {
+        return 0;
+    }
+
+    @Override
+    public int nextIndexFromLast(int[] tupleIndices, int[] nextSize, int tid) {
+        int priorTuple = tupleIndices[priorTable];
+        int priorVal = priorIntData.data[priorTuple];
+        int prevTuple = tupleIndices[nextTable];
+        int cardinality = nextIntIndex.cardinality;
+        if (nextIntIndex.unique) {
+            int onlyRow = nextIntIndex.keyToPositions.getOrDefault(priorVal, cardinality);
+            return onlyRow > prevTuple ? onlyRow : cardinality;
+        }
+        else {
+            int[] positions = nextIntIndex.positions;
+            int nextIndex = -1;
+            if (lastPositionsStart >= 0) {
+                for (int i = lastPositionsStart + 1; i <= lastPositionsEnd; i++) {
+                    nextIndex = positions[i];
+                    if (nextIndex > prevTuple) {
+                        lastPositionsStart = i;
+                        return nextIndex;
+                    }
+                }
+            }
+            else {
+                int firstPos = lastValue == priorVal ?
+                        lastFirst : nextIntIndex.keyToPositions.getOrDefault(priorVal, -1);
+                lastValue = priorVal;
+                lastFirst = firstPos;
+                if (firstPos < 0) {
+                    return cardinality;
+                }
+                // Get number of indexed values
+                int nrVals = nextIntIndex.positions[firstPos];
+                // Can we return first indexed value?
+                int firstTuple = nextIntIndex.positions[firstPos + 1];
+                if (firstTuple > prevTuple) {
+                    lastPositionsStart = firstPos + 1;
+                    lastPositionsEnd = firstPos + nrVals;
+                    return firstTuple;
+                }
+                for (int i = 2; i <= nrVals; i++) {
+                    nextIndex = positions[firstPos + i];
+                    if (nextIndex > prevTuple) {
+                        lastPositionsStart = firstPos + i;
+                        lastPositionsEnd = firstPos + nrVals;
+                        return nextIndex;
+                    }
+                }
+            }
+            lastPositionsStart = -1;
+            return cardinality;
+        }
+    }
+
     @Override
     public int nextIndexInScope(int[] tupleIndices, int tid, int[] nextSize) {
         int priorTuple = tupleIndices[priorTable];

src/joining/parallel/join/JoinPartitionIndexWrapper.java

@@ -32,10 +32,34 @@ public abstract class JoinPartitionIndexWrapper {
      * Reference to prior column data.
      */
     final ColumnData priorData;
+    /**
+     * Reference to prior column data.
+     */
+    final IntData priorTest;
+    /**
+     * Reference to prior column data.
+     */
+    final IntData nextTest;
     /**
      * Index on join column to use.
      */
     final Index nextIndex;
+    /**
+     * last prior value.
+     */
+    int lastValue = -1;
+    /**
+     * Last first position according to the last value.
+     */
+    int lastFirst = -1;
+    /**
+     * last start index of positions array.
+     */
+    int lastPositionsStart = -1;
+    /**
+     * last end index of positions array.
+     */
+    int lastPositionsEnd = -1;
     /**
      * Initialize index wrapper for
      * given query and join order.
@@ -55,10 +79,17 @@ public JoinPartitionIndexWrapper(ExpressionInfo equiPred, int[] order) {
         nextTable = pos1<pos2?table2:table1;
         // Get column data reference for prior table
         priorData = equiPred.dataMentioned.get(priorTable);
+        priorTest = (IntData) priorData;
+        nextTest = (IntData) equiPred.dataMentioned.get(nextTable);
         // Get index for next table
         nextIndex = equiPred.indexMentioned.get(nextTable);
     }
 
+    /**
+     * Reset temporary variables at the beginning of the join episode.
+     */
+    public abstract void reset(int[] tupleIndices);
+
     /**
      * Extracts index of table in query column reference.
      *
@@ -97,6 +128,9 @@ int tablePos(int[] order, int table) {
      */
     public abstract int nextIndex(int[] tupleIndices, int[] nextSize);
 
+    public abstract int nextIndexFromLast(int[] tupleIndices, int[] nextSize);
+    public int nextIndexFromLast(int[] tupleIndices, int[] nextSize, int tid) {return 0;}
+
     /**
      * Propose next index in next table that
      * satisfies equi-join condition in partitions with
@@ -108,7 +142,9 @@ int tablePos(int[] order, int table) {
      * @return              next interesting tuple index or cardinality
      */
     public abstract int nextIndexInScope(int[] tupleIndices, int tid, int[] nextSize);
+
     public abstract int nextIndexInScope(int[] tupleIndices, int tid, int[] nextSize, IntSet finishedThreads);
+
     /**
      * Propose next index in next table that
      * satisfies equi-join condition with
@@ -129,6 +165,7 @@ int tablePos(int[] order, int table) {
      * @return	next interesting tuple index or cardinality
      */
     public abstract boolean evaluateInScope(int[] tupleIndices, int tid);
+
     public abstract boolean evaluateInScope(int[] tupleIndices, int tid, IntSet finishedThreads);
 
     @Override

src/joining/parallel/join/ModJoin.java

@@ -151,12 +151,12 @@ public double execute(int[] order, int splitTable, int roundCtr) throws Exceptio
         // Lookup or generate left-deep query plan
         JoinOrder joinOrder = new JoinOrder(order);
         int joinHash = joinOrder.splitHashCode(-1);
+//        long timer1 = System.currentTimeMillis();
         LeftDeepPartitionPlan plan = planCache.get(joinHash);
         if (plan == null) {
             plan = new LeftDeepPartitionPlan(query, predToEval, joinOrder);
             planCache.putIfAbsent(joinHash, plan);
         }
-//        long timer1 = System.currentTimeMillis();
         int splitHash = nrThreads == 1 ? 0 : plan.splitStrategies[splitTable];
         // Execute from ing state, save progress, return progress
 
@@ -661,16 +661,17 @@ private void executeWithBudget(LeftDeepPartitionPlan plan, int splitTable, State
         // Number of completed tuples added
         nrResultTuples = 0;
         Arrays.fill(this.nrVisits, 0);
+        deepIndex = -1;
         // Execute join order until budget depleted or all input finished -
         // at each iteration start, tuple indices contain next tuple
         // combination to look at.
         while (remainingBudget > 0 && joinIndex >= 0) {
-
 //            ++statsInstance.nrIterations;
             //log("Offsets:\t" + Arrays.toString(offsets));
             //log("Indices:\t" + Arrays.toString(tupleIndices));
             // Get next table in join order
             int nextTable = plan.joinOrder.order[joinIndex];
+            deepIndex = Math.max(deepIndex, joinIndex);
 //            writeLog("Indices: " + Arrays.toString(tupleIndices));
             // Integrate table offset
             tupleIndices[nextTable] = Math.max(