Merge lp:~mmcm/akiban-server/pipeline-select-bloom-filter into lp:~akiban-technologies/akiban-server/trunk

[Mike McMahon]

Pipeline support for Select_BloomFilter.

The pipeline looks like Map_NestedLoops. In fact, the implementation inherits from those cursor classes.

Input rows are hashed against the Bloom filter taken from the given binding position. If this indicates that they might match, a query bindings is output with the row in that position (a different bindings object, so using the same position is now somewhat safer). This stream of bindings feeds the check plan. At the other end, if any bindings at the given depth has a row (from the check plan), the original input row from those bindings is output.

[lp:~mmcm/akiban-server/pipeline-select-bloom-filter]

2717. By Mike McMahon on 2013-07-18

Accidental.

[Nathan Williams (nwilliams)]

Looks good.

review: Approve

=== modified file 'src/main/java/com/akiban/qp/operator/API.java'
--- src/main/java/com/akiban/qp/operator/API.java	2013-07-18 22:48:23 +0000
+++ src/main/java/com/akiban/qp/operator/API.java	2013-07-18 22:48:23 +0000
@@ -710,18 +710,22 @@
     public static Operator select_BloomFilter(Operator input,
                                               Operator onPositive,
                                               List<? extends ExpressionGenerator> filterFields,
-                                              int bindingPosition)
+                                              int bindingPosition,
+                                              boolean pipeline,
+                                              int depth)
     {
-        return select_BloomFilter(input, onPositive, generateOld(filterFields), generateNew(filterFields), null, bindingPosition);
+        return select_BloomFilter(input, onPositive, generateOld(filterFields), generateNew(filterFields), null, bindingPosition, pipeline, depth);
     }
 
     public static Operator select_BloomFilter(Operator input,
                                               Operator onPositive,
                                               List<? extends Expression> filterFields,
                                               List<? extends TPreparedExpression> tFilterFields,
-                                              int bindingPosition)
+                                              int bindingPosition,
+                                              boolean pipeline,
+                                              int depth)
     {
-        return select_BloomFilter(input, onPositive, filterFields, tFilterFields, null, bindingPosition);
+        return select_BloomFilter(input, onPositive, filterFields, tFilterFields, null, bindingPosition, pipeline, depth);
     }
 
     public static Operator select_BloomFilter(Operator input,
@@ -729,14 +733,18 @@
                                               List<? extends Expression> filterFields,
                                               List<? extends TPreparedExpression> tFilterFields,
                                               List<AkCollator> collators,
-                                              int bindingPosition)
+                                              int bindingPosition,
+                                              boolean pipeline,
+                                              int depth)
     {
         return new Select_BloomFilter(input,
                                       onPositive,
                                       filterFields,
                                       tFilterFields,
                                       collators,
-                                      bindingPosition);
+                                      bindingPosition,
+                                      pipeline,
+                                      depth);
     }
 
     public static Operator select_BloomFilter(Operator input,
@@ -744,6 +752,8 @@
                                               List<? extends ExpressionGenerator> filterFields,
                                               List<AkCollator> collators,
                                               int bindingPosition,
+                                              boolean pipeline,
+                                              int depth,
                                               ExpressionGenerator.ErasureMaker marker)
     {
         return new Select_BloomFilter(input,
@@ -751,7 +761,9 @@
                 generateOld(filterFields),
                 generateNew(filterFields),
                 collators,
-                bindingPosition);
+                bindingPosition,
+                pipeline,
+                depth);
     }
 
     // EmitBoundRow_Nested

=== modified file 'src/main/java/com/akiban/qp/operator/Map_NestedLoops.java'
--- src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-18 14:20:06 +0000
+++ src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-18 22:48:23 +0000
@@ -174,9 +174,9 @@
     // Pipeline execution: turn outer loop row stream into binding stream for inner loop.
     protected static class RowToBindingsCursor implements QueryBindingsCursor
     {
-        private final Cursor input;
-        private final int depth, bindingPosition;
-        private QueryBindings baseBindings;
+        protected final Cursor input;
+        protected final int depth, bindingPosition;
+        protected QueryBindings baseBindings;
 
         public RowToBindingsCursor(Cursor input, int bindingPosition, int depth) {
             this.input = input;
@@ -193,7 +193,7 @@
         @Override
         public QueryBindings nextBindings() {
             if (baseBindings != null) {
-                Row row = input.next();
+                Row row = nextInputRow();
                 if (row != null) {
                     QueryBindings bindings = baseBindings.createBindings();
                     assert (bindings.getDepth() == depth);
@@ -231,6 +231,10 @@
                 input.cancelBindings(bindings);
             }
         }
+
+        protected Row nextInputRow() {
+            return input.next();
+        }
     }
 
     // Other end of pipeline: remove the extra binding levels that we
@@ -238,9 +242,9 @@
     // entire outer rowset.
     protected static class CollapseBindingsCursor extends OperatorCursor
     {
-        private final Cursor input;
-        private final int depth;
-        private QueryBindings currentBindings, pendingBindings, openBindings, inputOpenBindings;
+        protected final Cursor input;
+        protected final int depth;
+        protected QueryBindings currentBindings, pendingBindings, openBindings, inputOpenBindings;
         
         public CollapseBindingsCursor(QueryContext context, Cursor input, int depth) {
             super(context);

=== modified file 'src/main/java/com/akiban/qp/operator/Select_BloomFilter.java'
--- src/main/java/com/akiban/qp/operator/Select_BloomFilter.java	2013-07-15 22:01:49 +0000
+++ src/main/java/com/akiban/qp/operator/Select_BloomFilter.java	2013-07-18 22:48:23 +0000
@@ -53,6 +53,10 @@
  * loaded by the Using_BloomFilter operator. This bindingPosition is also used to hold a row from
  * the input stream that passes the filter and needs to be tested for existence using the onPositive
  * operator.
+ <li><b>boolean pipeline:</b> Whether to use bracketing cursors instead of rebinding.
+
+ <li><b>int depth:</b> Number of nested Maps, including this one.
+
  * <p/>
  * <h1>Behavior</h1>
  * <p/>
@@ -103,10 +107,19 @@
     @Override
     protected Cursor cursor(QueryContext context, QueryBindingsCursor bindingsCursor)
     {
-        if (tFields == null)
-            return new Execution<>(context, bindingsCursor, fields, oldExpressionsAdapater);
-        else
-            return new Execution<>(context, bindingsCursor, tFields, newExpressionsAdapter);
+        if (!pipeline) {
+            if (tFields == null)
+                return new Execution<>(context, bindingsCursor, fields, oldExpressionsAdapater);
+            else
+                return new Execution<>(context, bindingsCursor, tFields, newExpressionsAdapter);
+        }
+        else {
+            assert (tFields != null);
+            Cursor inputCursor = input.cursor(context, bindingsCursor);
+            QueryBindingsCursor toBindings = new FilterBindingsCursor(context, inputCursor, bindingPosition, depth, tFields, newExpressionsAdapter);
+            Cursor checkCursor = onPositive.cursor(context, toBindings);
+            return new RecoverRowsCursor(context, checkCursor, bindingPosition, depth);
+        }
     }
 
     @Override
@@ -128,7 +141,9 @@
                               List<? extends Expression> fields,
                               List<? extends TPreparedExpression> tFields,
                               List<AkCollator> collators,
-                              int bindingPosition)
+                              int bindingPosition,
+                              boolean pipeline,
+                              int depth)
     {
         ArgumentValidation.notNull("input", input);
         ArgumentValidation.notNull("onPositive", onPositive);
@@ -142,9 +157,12 @@
         }
         ArgumentValidation.isGT("fields.size()", size, 0);
         ArgumentValidation.isGTE("bindingPosition", bindingPosition, 0);
+        ArgumentValidation.isGT("depth", depth, 0);
         this.input = input;
         this.onPositive = onPositive;
         this.bindingPosition = bindingPosition;
+        this.pipeline = pipeline;
+        this.depth = depth;
         this.fields = fields;
         this.tFields = tFields;
         this.collators = collators;
@@ -168,7 +186,8 @@
 
     private final Operator input;
     private final Operator onPositive;
-    private final int bindingPosition;
+    private final int bindingPosition, depth;
+    private final boolean pipeline;
     private final List<? extends Expression> fields;
     private final List<? extends TPreparedExpression> tFields;
     private final List<AkCollator> collators;
@@ -380,6 +399,9 @@
             // It is safe to reuse the binding position in this way because the filter is extracted and stored
             // in a field during open(), while the use of the binding position for use in the onPositive lookup
             // occurs during next().
+            if (LOG_EXECUTION) {
+                LOG.debug("Select_BloomFilter: candidate {}", row);
+            }
             TAP_CHECK.in();
             try {
                 bindings.setRow(bindingPosition, row);
@@ -408,4 +430,108 @@
         private boolean idle = true;
         private boolean destroyed = false;
     }
+
+    // Turn input rows that match the filter into bindings for the onPositive plan.
+    private class FilterBindingsCursor extends Map_NestedLoops.RowToBindingsCursor
+    {
+        private final StoreAdapter storeAdapter;
+        private final List<TEvaluatableExpression> fieldEvals = new ArrayList<>();
+        private final ExpressionAdapter<TPreparedExpression, TEvaluatableExpression> expressionAdapter;
+
+        public FilterBindingsCursor(QueryContext context, Cursor input, 
+                                    int bindingPosition, int depth,
+                                    List<? extends TPreparedExpression> expressions, ExpressionAdapter<TPreparedExpression, TEvaluatableExpression> expressionAdapter) {
+            super(input, bindingPosition, depth);
+            this.storeAdapter = context.getStore();
+            this.expressionAdapter = expressionAdapter;
+            for (TPreparedExpression field : expressions) {
+                TEvaluatableExpression eval = expressionAdapter.evaluate(field, context);
+                fieldEvals.add(eval);
+            }
+        }
+
+        @Override
+        protected Row nextInputRow() {
+            BloomFilter filter = baseBindings.getBloomFilter(bindingPosition);
+            while (true) {
+                Row row = input.next();
+                if (row == null) {
+                    return row;
+                }
+                if (filter.maybePresent(hashProjectedRow(row))) {
+                    if (ExecutionBase.LOG_EXECUTION) {
+                        LOG.debug("Select_BloomFilter: candidate {}", row);
+                    }
+                    return row;
+                }
+            }
+        }
+
+        private int hashProjectedRow(Row row)
+        {
+            int hash = 0;
+            for (int f = 0; f < fieldEvals.size(); f++) {
+                TEvaluatableExpression fieldEval = fieldEvals.get(f);
+                hash = hash ^ expressionAdapter.hash(storeAdapter, fieldEval, row, collator(f));
+            }
+            return hash;
+        }
+    }
+
+    // If any context at our depth has a non-empty rowset from
+    // onPositive, it passed, so let it through.
+    private static class RecoverRowsCursor extends Map_NestedLoops.CollapseBindingsCursor
+    {
+        private final int bindingPosition;
+
+        public RecoverRowsCursor(QueryContext context, Cursor input, int bindingPosition, int depth) {
+            super(context, input, depth);
+            this.bindingPosition = bindingPosition;
+        }
+
+        @Override
+        public Row next() {
+            if (TAP_NEXT_ENABLED) {
+                TAP_NEXT.in();
+            }
+            try {
+                if (CURSOR_LIFECYCLE_ENABLED) {
+                    CursorLifecycle.checkIdleOrActive(this);
+                }
+                checkQueryCancelation();
+                Row row = null;
+                while (true) {
+                    QueryBindings bindings = input.nextBindings();
+                    if (bindings == null) {
+                        openBindings = null;
+                        break;
+                    }
+                    if (bindings.getDepth() < depth) {
+                        pendingBindings = bindings;
+                        openBindings = null;
+                        break;
+                    }
+                    assert (bindings.getDepth() == depth);
+                    input.open();
+                    inputOpenBindings = bindings;
+                    row = input.next();
+                    input.close();
+                    inputOpenBindings = null;
+                    if (row != null) {
+                        row = bindings.getRow(bindingPosition);
+                        break;
+                    }
+                }
+                if (LOG_EXECUTION) {
+                    LOG.debug("Select_BloomFilter: yield {}", row);
+                }
+                return row;
+            } 
+            finally {
+                if (TAP_NEXT_ENABLED) {
+                    TAP_NEXT.out();
+                }
+            }
+        }
+    }
 }

=== modified file 'src/main/java/com/akiban/qp/operator/Using_BloomFilter.java'
--- src/main/java/com/akiban/qp/operator/Using_BloomFilter.java	2013-07-09 17:26:33 +0000
+++ src/main/java/com/akiban/qp/operator/Using_BloomFilter.java	2013-07-18 22:48:23 +0000
@@ -215,7 +215,9 @@
         {
             close();
             input.destroy();
-            bindings.setBloomFilter(filterBindingPosition, null);
+            if (bindings != null) {
+                bindings.setBloomFilter(filterBindingPosition, null);
+            }
         }
 
         // Execution interface

=== modified file 'src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java'
--- src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-18 22:48:23 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-18 22:48:23 +0000
@@ -1827,6 +1827,7 @@
             int pos = getHashTablePosition(bloomFilter);
             RowStream stream = assembleStream(bloomFilterFilter.getInput());
             boundRows.set(pos, stream.fieldOffsets);
+            nestedBindingsDepth++;
             RowStream cstream = assembleStream(bloomFilterFilter.getCheck());
             boundRows.set(pos, null);
             List<Expression> fields = oldPartialAssembler.assembleExpressions(bloomFilterFilter.getLookupExpressions(),
@@ -1842,7 +1843,10 @@
                                                      fields,
                                                      tFields,
                                                      collators,
-                                                     pos + loopBindingsOffset);
+                                                     pos + loopBindingsOffset,
+                                                     rulesContext.getPipelineConfiguration().isSelectBloomFilterEnabled(),
+                                                     nestedBindingsDepth);
+            nestedBindingsDepth--;
             return stream;
         }
 

=== modified file 'src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java'
--- src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java	2013-07-13 20:04:24 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java	2013-07-18 22:48:23 +0000
@@ -25,6 +25,7 @@
     private int indexScanLookaheadQuantum = 1;
     private int groupLookupLookaheadQuantum = 1;
     private boolean unionAllOpenBoth = false;
+    private boolean selectBloomFilterEnabled = false;
 
     public PipelineConfiguration() {
     }
@@ -49,6 +50,10 @@
         return unionAllOpenBoth;
     }
 
+    public boolean isSelectBloomFilterEnabled() {
+        return selectBloomFilterEnabled;
+    }
+
     public void load(Properties properties) {
         for (String prop : properties.stringPropertyNames()) {
             String val = properties.getProperty(prop);
@@ -60,6 +65,8 @@
                 groupLookupLookaheadQuantum = Integer.parseInt(val);
             else if ("unionAll.openBoth".equals(prop))
                 unionAllOpenBoth = Boolean.parseBoolean(val);
+            else if ("selectBloomFilter.enabled".equals(prop))
+                selectBloomFilterEnabled = Boolean.parseBoolean(val);
             else
                 throw new IllegalArgumentException("Unknown property " + prop);
         }

=== modified file 'src/test/java/com/akiban/server/test/costmodel/Select_BloomFilterCT.java'
--- src/test/java/com/akiban/server/test/costmodel/Select_BloomFilterCT.java	2013-07-09 21:10:22 +0000
+++ src/test/java/com/akiban/server/test/costmodel/Select_BloomFilterCT.java	2013-07-18 22:48:23 +0000
@@ -160,8 +160,8 @@
                         new Ordering()),
                     // filterFields
                     Arrays.asList(ExpressionGenerators.field(dIndexRowType, 0)),
-                    // filterBindingPosition
-                    0));
+                    // filterBindingPosition, pipeline, depth
+                    0, false, 1));
         return plan;
     }
 
@@ -213,8 +213,8 @@
                         new Ordering()),
                     // filterFields
                     Arrays.asList(ExpressionGenerators.field(dIndexRowType, 0)),
-                    // filterBindingPosition
-                    0));
+                    // filterBindingPosition, pipeline, depth
+                    0, false, 1));
         return plan;
     }
 

=== modified file 'src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsIT.java'
--- src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsIT.java	2013-07-18 02:32:32 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsIT.java	2013-07-18 22:48:23 +0000
@@ -262,7 +262,7 @@
             public boolean reopenTopLevel() {
                 // You cannot just re-open() a pipelined Map_NestedLoops, but you can
                 // openTopLevel() it again.
-                return true;
+                return pipelineMap();
             }
         };
         testCursorLifecycle(plan, testCase);

=== modified file 'src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterIT.java'
--- src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterIT.java	2013-07-08 20:48:05 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterIT.java	2013-07-18 22:48:23 +0000
@@ -107,6 +107,10 @@
         use(db);
     }
 
+    protected boolean pipelineSelectBloomFilter() {
+        return false;
+    }
+
     // Test argument validation
 
     @Test
@@ -239,6 +243,11 @@
                     row(outputRowType, 6L, 62L, 602L),
                 };
             }
+
+            @Override
+            public boolean reopenTopLevel() {
+                return pipelineSelectBloomFilter();
+            }
         };
         testCursorLifecycle(plan, testCase);
     }
@@ -301,7 +310,9 @@
                             ExpressionGenerators.field(dIndexRowType, 1),
                             ExpressionGenerators.field(dIndexRowType, 2)),
                         // filterBindingPosition
-                        0)),
+                        0,
+                        pipelineSelectBloomFilter(),
+                        1)),
                 dIndexRowType,
                 Arrays.asList(
                     ExpressionGenerators.field(dIndexRowType, 0),   // test_id

=== added file 'src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterPipelineIT.java'
--- src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterPipelineIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Select_BloomFilterPipelineIT.java	2013-07-18 22:48:23 +0000
@@ -0,0 +1,26 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.akiban.server.test.it.qp;
+
+public class Select_BloomFilterPipelineIT extends Select_BloomFilterIT
+{
+    @Override
+    protected boolean pipelineSelectBloomFilter() {
+        return true;
+    }
+}

=== modified file 'src/test/java/com/akiban/server/test/it/qp/Select_BloomFilter_CaseInsensitive_IT.java'
--- src/test/java/com/akiban/server/test/it/qp/Select_BloomFilter_CaseInsensitive_IT.java	2013-07-08 20:48:05 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Select_BloomFilter_CaseInsensitive_IT.java	2013-07-18 22:48:23 +0000
@@ -317,7 +317,7 @@
                         // collators
                         collators,
                         // filterBindingPosition
-                        0,
+                        0, false, 1,
                         ExpressionGenerator.ErasureMaker.MARK),
                     // collators
                     collators