Merge pull request #211 from svladykin/batchview2

Batched index lookups for sub-query

h2/src/main/org/h2/command/Command.java

@@ -68,6 +68,11 @@ public abstract class Command implements CommandInterface {
     @Override
     public abstract boolean isQuery();
 
+    /**
+     * Prepare join batching.
+     */
+    public abstract void prepareJoinBatch();
+
     /**
      * Get the list of parameters.
      *

h2/src/main/org/h2/command/CommandContainer.java

@@ -7,6 +7,8 @@ package org.h2.command;
 
 import java.util.ArrayList;
 import org.h2.api.DatabaseEventListener;
+import org.h2.command.dml.Explain;
+import org.h2.command.dml.Query;
 import org.h2.expression.Parameter;
 import org.h2.expression.ParameterInterface;
 import org.h2.result.ResultInterface;
@@ -44,6 +46,23 @@ public class CommandContainer extends Command {
         return prepared.isQuery();
     }
 
+    @Override
+    public void prepareJoinBatch() {
+        if (session.isJoinBatchEnabled()) {
+            prepareJoinBatch(prepared);
+        }
+    }
+
+    private static void prepareJoinBatch(Prepared prepared) {
+        if (prepared.isQuery()) {
+            if (prepared.getType() == CommandInterface.SELECT) {
+                ((Query) prepared).prepareJoinBatch();
+            } else if (prepared.getType() == CommandInterface.EXPLAIN) {
+                prepareJoinBatch(((Explain) prepared).getCommand());
+            }
+        }
+    }
+
     private void recompileIfRequired() {
         if (prepared.needRecompile()) {
             // TODO test with 'always recompile'
@@ -65,6 +84,7 @@ public class CommandContainer extends Command {
             }
             prepared.prepare();
             prepared.setModificationMetaId(mod);
+            prepareJoinBatch();
         }
     }
 

h2/src/main/org/h2/command/CommandList.java

@@ -44,6 +44,11 @@ class CommandList extends Command {
         return updateCount;
     }
 
+    @Override
+    public void prepareJoinBatch() {
+        command.prepareJoinBatch();
+    }
+
     @Override
     public ResultInterface query(int maxrows) {
         ResultInterface result = command.query(maxrows);

h2/src/main/org/h2/command/ddl/AlterTableAddConstraint.java

@@ -294,6 +294,9 @@ public class AlterTableAddConstraint extends SchemaCommand {
     }
 
     private static Index getUniqueIndex(Table t, IndexColumn[] cols) {
+        if (t.getIndexes() == null) {
+            return null;
+        }
         for (Index idx : t.getIndexes()) {
             if (canUseUniqueIndex(idx, t, cols)) {
                 return idx;
@@ -303,6 +306,9 @@ public class AlterTableAddConstraint extends SchemaCommand {
     }
 
     private static Index getIndex(Table t, IndexColumn[] cols, boolean moreColumnOk) {
+        if (t.getIndexes() == null) {
+            return null;
+        }
         for (Index idx : t.getIndexes()) {
             if (canUseIndex(idx, t, cols, moreColumnOk)) {
                 return idx;

h2/src/main/org/h2/command/dml/Explain.java

@@ -40,6 +40,10 @@ public class Explain extends Prepared {
         this.command = command;
     }
 
+    public Prepared getCommand() {
+        return command;
+    }
+
     @Override
     public void prepare() {
         command.prepare();

h2/src/main/org/h2/command/dml/NoOperation.java

@@ -24,11 +24,6 @@ public class NoOperation extends Prepared {
         return 0;
     }
 
-    @Override
-    public boolean isQuery() {
-        return false;
-    }
-
     @Override
     public boolean isTransactional() {
         return true;

h2/src/main/org/h2/command/dml/Query.java

@@ -71,6 +71,18 @@ public abstract class Query extends Prepared {
         super(session);
     }
 
+    /**
+     * Check if this is a UNION query.
+     * 
+     * @return {@code true} if this is a UNION query
+     */
+    public abstract boolean isUnion();
+
+    /**
+     * Prepare join batching.
+     */
+    public abstract void prepareJoinBatch();
+
     /**
      * Execute the query without checking the cache. If a target is specified,
      * the results are written to it, and the method returns null. If no target

h2/src/main/org/h2/command/dml/Select.java

@@ -9,7 +9,6 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
-
 import org.h2.api.ErrorCode;
 import org.h2.api.Trigger;
 import org.h2.command.CommandInterface;
@@ -36,6 +35,7 @@ import org.h2.result.SortOrder;
 import org.h2.table.Column;
 import org.h2.table.ColumnResolver;
 import org.h2.table.IndexColumn;
+import org.h2.table.JoinBatch;
 import org.h2.table.Table;
 import org.h2.table.TableFilter;
 import org.h2.util.New;
@@ -87,6 +87,11 @@ public class Select extends Query {
         super(session);
     }
 
+    @Override
+    public boolean isUnion() {
+        return false;
+    }
+
     /**
      * Add a table to the query.
      *
@@ -945,6 +950,30 @@ public class Select extends Query {
         isPrepared = true;
     }
 
+    @Override
+    public void prepareJoinBatch() {
+        ArrayList<TableFilter> list = New.arrayList();
+        TableFilter f = getTopTableFilter();
+        do {
+            if (f.getNestedJoin() != null) {
+                // we do not support batching with nested joins 
+                return;
+            }
+            list.add(f);
+            f = f.getJoin();
+        } while (f != null);
+        TableFilter[] fs = list.toArray(new TableFilter[list.size()]);
+        // prepare join batch
+        JoinBatch jb = null;
+        for (int i = fs.length - 1; i >= 0; i--) {
+            jb = fs[i].prepareJoinBatch(jb, fs, i);
+        }
+    }
+
+    public JoinBatch getJoinBatch() {
+        return getTopTableFilter().getJoinBatch();
+    }
+
     @Override
     public double getCost() {
         return cost;

h2/src/main/org/h2/command/dml/SelectUnion.java

@@ -7,7 +7,6 @@ package org.h2.command.dml;
 
 import java.util.ArrayList;
 import java.util.HashSet;
-
 import org.h2.api.ErrorCode;
 import org.h2.command.CommandInterface;
 import org.h2.engine.Session;
@@ -72,6 +71,17 @@ public class SelectUnion extends Query {
         this.left = query;
     }
 
+    @Override
+    public boolean isUnion() {
+        return true;
+    }
+
+    @Override
+    public void prepareJoinBatch() {
+        left.prepareJoinBatch();
+        right.prepareJoinBatch();
+    }
+
     public void setUnionType(int type) {
         this.unionType = type;
     }

h2/src/main/org/h2/command/dml/Set.java

@@ -497,6 +497,15 @@ public class Set extends Prepared {
             database.setRowFactory(rowFactory);
             break;
         }
+        case SetTypes.BATCH_JOINS: {
+            int value = getIntValue();
+            if (value != 0 && value != 1) {
+                throw DbException.getInvalidValueException("BATCH_JOINS",
+                        getIntValue());
+            }
+            session.setJoinBatchEnabled(value == 1);
+            break;
+        }
         default:
             DbException.throwInternalError("type="+type);
         }

h2/src/main/org/h2/command/dml/SetTypes.java

@@ -228,6 +228,11 @@ public class SetTypes {
      */
     public static final int ROW_FACTORY = 43;
 
+    /**
+     * The type of SET BATCH_JOINS statement.
+     */
+    public static final int BATCH_JOINS = 44;
+
     private static final ArrayList<String> TYPES = New.arrayList();
 
     private SetTypes() {
@@ -280,6 +285,7 @@ public class SetTypes {
         list.add(QUERY_STATISTICS, "QUERY_STATISTICS");
         list.add(QUERY_STATISTICS_MAX_ENTRIES, "QUERY_STATISTICS_MAX_ENTRIES");
         list.add(ROW_FACTORY, "ROW_FACTORY");
+        list.add(BATCH_JOINS, "BATCH_JOINS");
     }
 
     /**

h2/src/main/org/h2/engine/Session.java

@@ -30,12 +30,14 @@ import org.h2.mvstore.db.TransactionStore.Change;
 import org.h2.mvstore.db.TransactionStore.Transaction;
 import org.h2.result.LocalResult;
 import org.h2.result.Row;
+import org.h2.result.SortOrder;
 import org.h2.schema.Schema;
 import org.h2.store.DataHandler;
 import org.h2.store.InDoubtTransaction;
 import org.h2.store.LobStorageFrontend;
 import org.h2.table.SubQueryInfo;
 import org.h2.table.Table;
+import org.h2.table.TableFilter;
 import org.h2.util.New;
 import org.h2.util.SmallLRUCache;
 import org.h2.value.Value;
@@ -115,6 +117,7 @@ public class Session extends SessionWithState {
     private int parsingView;
     private volatile SmallLRUCache<Object, ViewIndex> viewIndexCache;
     private HashMap<Object, ViewIndex> subQueryIndexCache;
+    private boolean joinBatchEnabled;
 
     /**
      * Temporary LOBs from result sets. Those are kept for some time. The
@@ -148,12 +151,25 @@ public class Session extends SessionWithState {
         this.currentSchemaName = Constants.SCHEMA_MAIN;
     }
 
+    public void setJoinBatchEnabled(boolean joinBatchEnabled) {
+        this.joinBatchEnabled = joinBatchEnabled;
+    }
+
+    public boolean isJoinBatchEnabled() {
+        return joinBatchEnabled;
+    }
+
     public Row createRow(Value[] data, int memory) {
         return database.createRow(data, memory);
     }
 
-    public void setSubQueryInfo(SubQueryInfo subQueryInfo) {
-        this.subQueryInfo = subQueryInfo;
+    public void pushSubQueryInfo(int[] masks, TableFilter[] filters, int filter,
+            SortOrder sortOrder) {
+        subQueryInfo = new SubQueryInfo(subQueryInfo, masks, filters, filter, sortOrder);
+    }
+
+    public void popSubQueryInfo() {
+        subQueryInfo = subQueryInfo.getUpper();
     }
 
     public SubQueryInfo getSubQueryInfo() {
@@ -492,6 +508,7 @@ public class Session extends SessionWithState {
             // we can't reuse sub-query indexes, so just drop the whole cache
             subQueryIndexCache = null;
         }
+        command.prepareJoinBatch();
         if (queryCache != null) {
             if (command.isCacheable()) {
                 queryCache.put(sql, command);

h2/src/main/org/h2/index/IndexLookupBatch.java

@@ -15,7 +15,8 @@ import org.h2.result.SearchRow;
  * method {@link #isBatchFull()}} will return {@code true} or there are no more
  * search rows to add. Then method {@link #find()} will be called to execute batched lookup.
  * Note that a single instance of {@link IndexLookupBatch} can be reused for multiple 
- * sequential batched lookups.
+ * sequential batched lookups, moreover it can be reused for multiple queries for
+ * the same prepared statement.
  * 
  * @see Index#createLookupBatch(org.h2.table.TableFilter)
  * @author Sergi Vladykin
@@ -26,9 +27,11 @@ public interface IndexLookupBatch {
      * 
      * @param first the first row, or null for no limit
      * @param last the last row, or null for no limit
+     * @return {@code false} if this search row pair is known to produce no results
+     *          and thus the given row pair was not added
      * @see Index#find(TableFilter, SearchRow, SearchRow)
      */
-    void addSearchRows(SearchRow first, SearchRow last);
+    boolean addSearchRows(SearchRow first, SearchRow last);
 
     /**
      * Check if this batch is full.
@@ -47,4 +50,16 @@ public interface IndexLookupBatch {
      * @return List of future cursors for collected search rows.
      */
     List<Future<Cursor>> find();
+
+    /**
+     * Get plan for EXPLAIN.
+     *
+     * @return plan
+     */
+    String getPlanSQL();
+
+    /**
+     * Reset this batch to clear state. This method will be called before each query execution.
+     */
+    void reset();
 }

h2/src/main/org/h2/index/MultiVersionIndex.java

@@ -6,8 +6,6 @@
 package org.h2.index;
 
 import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.Future;
 import org.h2.api.ErrorCode;
 import org.h2.engine.Database;
 import org.h2.engine.DbObject;

h2/src/main/org/h2/index/ViewCursor.java

@@ -24,7 +24,7 @@ public class ViewCursor implements Cursor {
     private final SearchRow first, last;
     private Row current;
 
-    ViewCursor(ViewIndex index, LocalResult result, SearchRow first,
+    public ViewCursor(ViewIndex index, LocalResult result, SearchRow first,
             SearchRow last) {
         this.table = index.getTable();
         this.index = index;

h2/src/main/org/h2/index/ViewIndex.java

@@ -21,7 +21,7 @@ import org.h2.result.SearchRow;
 import org.h2.result.SortOrder;
 import org.h2.table.Column;
 import org.h2.table.IndexColumn;
-import org.h2.table.SubQueryInfo;
+import org.h2.table.JoinBatch;
 import org.h2.table.TableFilter;
 import org.h2.table.TableView;
 import org.h2.util.IntArray;
@@ -93,6 +93,15 @@ public class ViewIndex extends BaseIndex implements SpatialIndex {
         }
     }
 
+    @Override
+    public IndexLookupBatch createLookupBatch(TableFilter filter) {
+        if (recursive) {
+            // we do not support batching for recursive queries
+            return null;
+        }
+        return JoinBatch.createViewIndexLookupBatch(this);
+    }
+
     public Session getSession() {
         return createSession;
     }
@@ -199,69 +208,69 @@ public class ViewIndex extends BaseIndex implements SpatialIndex {
             TableFilter[] filters, int filter, SortOrder sortOrder, boolean preliminary) {
         assert filters != null;
         Prepared p;
-        SubQueryInfo upper = session.getSubQueryInfo();
-        SubQueryInfo info = new SubQueryInfo(upper,
-                masks, filters, filter, sortOrder, preliminary);
-        session.setSubQueryInfo(info);
+        session.pushSubQueryInfo(masks, filters, filter, sortOrder);
         try {
             p = session.prepare(sql, true);
         } finally {
-            session.setSubQueryInfo(upper);
+            session.popSubQueryInfo();
         }
         return (Query) p;
     }
 
-    private Cursor find(Session session, SearchRow first, SearchRow last,
+    private Cursor findRecursive(Session session, SearchRow first, SearchRow last,
             SearchRow intersection) {
-        if (recursive) {
-            LocalResult recResult = view.getRecursiveResult();
-            if (recResult != null) {
-                recResult.reset();
-                return new ViewCursor(this, recResult, first, last);
-            }
-            if (query == null) {
-                query = (Query) createSession.prepare(querySQL, true);
-            }
-            if (!(query instanceof SelectUnion)) {
-                throw DbException.get(ErrorCode.SYNTAX_ERROR_2,
-                        "recursive queries without UNION ALL");
-            }
-            SelectUnion union = (SelectUnion) query;
-            if (union.getUnionType() != SelectUnion.UNION_ALL) {
-                throw DbException.get(ErrorCode.SYNTAX_ERROR_2,
-                        "recursive queries without UNION ALL");
+        assert recursive;
+        LocalResult recResult = view.getRecursiveResult();
+        if (recResult != null) {
+            recResult.reset();
+            return new ViewCursor(this, recResult, first, last);
+        }
+        if (query == null) {
+            query = (Query) createSession.prepare(querySQL, true);
+        }
+        if (!query.isUnion()) {
+            throw DbException.get(ErrorCode.SYNTAX_ERROR_2,
+                    "recursive queries without UNION ALL");
+        }
+        SelectUnion union = (SelectUnion) query;
+        if (union.getUnionType() != SelectUnion.UNION_ALL) {
+            throw DbException.get(ErrorCode.SYNTAX_ERROR_2,
+                    "recursive queries without UNION ALL");
+        }
+        Query left = union.getLeft();
+        // to ensure the last result is not closed
+        left.disableCache();
+        LocalResult r = left.query(0);
+        LocalResult result = union.getEmptyResult();
+        // ensure it is not written to disk,
+        // because it is not closed normally
+        result.setMaxMemoryRows(Integer.MAX_VALUE);
+        while (r.next()) {
+            result.addRow(r.currentRow());
+        }
+        Query right = union.getRight();
+        r.reset();
+        view.setRecursiveResult(r);
+        // to ensure the last result is not closed
+        right.disableCache();
+        while (true) {
+            r = right.query(0);
+            if (r.getRowCount() == 0) {
+                break;
             }
-            Query left = union.getLeft();
-            // to ensure the last result is not closed
-            left.disableCache();
-            LocalResult r = left.query(0);
-            LocalResult result = union.getEmptyResult();
-            // ensure it is not written to disk,
-            // because it is not closed normally
-            result.setMaxMemoryRows(Integer.MAX_VALUE);
             while (r.next()) {
                 result.addRow(r.currentRow());
             }
-            Query right = union.getRight();
             r.reset();
             view.setRecursiveResult(r);
-            // to ensure the last result is not closed
-            right.disableCache();
-            while (true) {
-                r = right.query(0);
-                if (r.getRowCount() == 0) {
-                    break;
-                }
-                while (r.next()) {
-                    result.addRow(r.currentRow());
-                }
-                r.reset();
-                view.setRecursiveResult(r);
-            }
-            view.setRecursiveResult(null);
-            result.done();
-            return new ViewCursor(this, result, first, last);
         }
+        view.setRecursiveResult(null);
+        result.done();
+        return new ViewCursor(this, result, first, last);
+    }
+
+    public void setupQueryParameters(Session session, SearchRow first, SearchRow last,
+            SearchRow intersection) {
         ArrayList<Parameter> paramList = query.getParameters();
         if (originalParameters != null) {
             for (int i = 0, size = originalParameters.size(); i < size; i++) {
@@ -298,6 +307,14 @@ public class ViewIndex extends BaseIndex implements SpatialIndex {
                 setParameter(paramList, idx++, intersection.getValue(i));
             }
         }
+    }
+
+    private Cursor find(Session session, SearchRow first, SearchRow last,
+            SearchRow intersection) {
+        if (recursive) {
+            return findRecursive(session, first, last, intersection);
+        }
+        setupQueryParameters(session, first, last, intersection);
         LocalResult result = query.query(0);
         return new ViewCursor(this, result, first, last);
     }
@@ -313,6 +330,10 @@ public class ViewIndex extends BaseIndex implements SpatialIndex {
         param.setValue(v);
     }
 
+    public Query getQuery() {
+        return query;
+    }
+
     private Query getQuery(Session session, int[] masks,
             TableFilter[] filters, int filter, SortOrder sortOrder) {
         Query q = prepareSubQuery(querySQL, session, masks, filters, filter, sortOrder, true);
@@ -454,5 +475,4 @@ public class ViewIndex extends BaseIndex implements SpatialIndex {
     public boolean isRecursive() {
         return recursive;
     }
-
 }

h2/src/main/org/h2/table/SubQueryInfo.java

@@ -19,7 +19,6 @@ public class SubQueryInfo {
     private TableFilter[] filters;
     private int filter;
     private SortOrder sortOrder;
-    private boolean preliminary;
     private SubQueryInfo upper;
 
     /**
@@ -28,17 +27,14 @@ public class SubQueryInfo {
      * @param filters table filters
      * @param filter current filter
      * @param sortOrder sort order
-     * @param preliminary if this is a preliminary query optimization 
-     *          without global conditions
      */
     public SubQueryInfo(SubQueryInfo upper, int[] masks, TableFilter[] filters, int filter,
-            SortOrder sortOrder, boolean preliminary) {
+            SortOrder sortOrder) {
         this.upper = upper;
         this.masks = masks;
         this.filters = filters;
         this.filter = filter;
         this.sortOrder = sortOrder;
-        this.preliminary = preliminary;
     }
 
     public SubQueryInfo getUpper() {
@@ -60,8 +56,4 @@ public class SubQueryInfo {
     public SortOrder getSortOrder() {
         return sortOrder;
     }
-
-    public boolean isPreliminary() {
-        return preliminary;
-    }
 }

h2/src/main/org/h2/table/Table.java

@@ -127,6 +127,10 @@ public abstract class Table extends SchemaObjectBase {
         }
     }
 
+    public boolean isView() {
+        return false;
+    }
+
     /**
      * Lock the table for the given session.
      * This method waits until the lock is granted.

h2/src/main/org/h2/table/TableView.java

@@ -218,6 +218,11 @@ public class TableView extends Table {
         }
     }
 
+    @Override
+    public boolean isView() {
+        return true;
+    }
+
     /**
      * Check if this view is currently invalid.
      *

h2/src/main/org/h2/util/LazyFuture.java

+/*
+ * Copyright 2004-2014 H2 Group. Multiple-Licensed under the MPL 2.0,
+ * and the EPL 1.0 (http://h2database.com/html/license.html).
+ * Initial Developer: H2 Group
+ */
+package org.h2.util;
+
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.h2.message.DbException;
+
+/**
+ * Single threaded lazy future. 
+ * 
+ * @param <T>
+ * @author Sergi Vladykin
+ */
+public abstract class LazyFuture<T> implements Future<T> {
+
+    private static final int S_READY = 0;
+    private static final int S_DONE = 1;
+    private static final int S_ERROR = 2;
+    private static final int S_CANCELED = 3;
+    
+    private int state = S_READY;
+    private T result;
+    private Exception error;
+
+    /**
+     * Reset this future to the initial state.
+     * 
+     * @return {@code false} if it was already in initial state 
+     */
+    public boolean reset() {
+        if (state == S_READY) {
+            return false;
+        }
+        state = S_READY;
+        result = null;
+        error = null;
+        return true;
+    }
+
+    /**
+     * Run computation and produce the result.
+     * 
+     * @return the result of computation
+     */
+    protected abstract T run() throws Exception;
+    
+    @Override
+    public boolean cancel(boolean mayInterruptIfRunning) {
+        if (state != S_READY) {
+            return false;
+        }
+        state = S_CANCELED;
+        return true;
+    }
+
+    @Override
+    public T get() throws InterruptedException, ExecutionException {
+        switch (state) {
+        case S_READY:
+            try {
+                result = run();
+                state = S_DONE;
+            } catch (Exception e) {
+                error = e;
+                if (e instanceof InterruptedException) {
+                    throw (InterruptedException) e;
+                }
+                throw new ExecutionException(e);
+            } finally {
+                if (state != S_DONE) {
+                    state = S_ERROR;
+                }
+            }
+            return result;
+        case S_DONE:
+            return result;
+        case S_ERROR:
+            throw new ExecutionException(error);
+        case S_CANCELED:
+            throw new CancellationException();
+        default:
+            throw DbException.throwInternalError();
+        }
+    }
+
+    @Override
+    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException {
+        return get();
+    }
+
+    @Override
+    public boolean isCancelled() {
+        return state == S_CANCELED;
+    }
+
+    @Override
+    public boolean isDone() {
+        return state != S_READY;
+    }
+}