Merge lp:~mmcm/akiban-server/query-bindings-cancel into lp:~akiban-technologies/akiban-server/trunk

[Mike McMahon]

Enhance query bindings stream to address early close.

When Map output is closed before end-of-stream, it must advance query bindings until the next one at a higher level than the depth it is collapsing. It can do this by actually calling nextBindings(), but that requires upstream operators to continue to process the depth that is actually being skipped. For instance, in the new test added to the Map IT, the 20 and 30 rows would still flow through even after Limit has closed the Map. Also query cancel of, say, a 10-way cross-product would not stop because the level that happened to detect the cancel, which is not likely to be near the leaf, will try to advance.

The new cancelBindings methods advances past the given query bindings. As with open() and close(), trivial operators can assume that their neighbors are taking care of binding complexity and just do like close(), except that they pass along cancel to their input. Operators that do lookahead for pipelining should discard any lookahead state that is under the given bindings.

[Nathan Williams (nwilliams)]

Looks as expected.

review: Approve

=== modified file 'src/main/java/com/akiban/qp/operator/Aggregate_Partial.java'
--- src/main/java/com/akiban/qp/operator/Aggregate_Partial.java	2013-07-09 17:26:33 +0000
+++ src/main/java/com/akiban/qp/operator/Aggregate_Partial.java	2013-07-17 08:37:29 +0000
@@ -503,6 +503,12 @@
             inputCursor.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            inputCursor.cancelBindings(bindings);
+            cursorState = CursorState.CLOSED;
+        }
+
         // for use in this class
 
         private void aggregate(Row input) {

=== modified file 'src/main/java/com/akiban/qp/operator/AncestorLookup_Default.java'
--- src/main/java/com/akiban/qp/operator/AncestorLookup_Default.java	2013-07-13 21:17:31 +0000
+++ src/main/java/com/akiban/qp/operator/AncestorLookup_Default.java	2013-07-17 08:37:29 +0000
@@ -578,6 +578,35 @@
             clearBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            while (true) {
+                QueryBindings pending = pendingBindings.peek();
+                if (pending == null) break;
+                if (!pending.isAncestor(bindings)) break;
+                pendingBindings.remove();
+            }
+            int nancestors = ancestors.size();
+            while ((inputRowBindings[currentIndex] != null) &&
+                   inputRowBindings[currentIndex].isAncestor(bindings)) {
+                inputRows[currentIndex].release();
+                inputRowBindings[currentIndex] = null;
+                for (int i = 0; i < nancestors; i++) {
+                    int index = currentIndex * nancestors + i;
+                    ancestorCursors[index].close();
+                    ancestorHKeys[index] = null;
+                }
+                currentIndex = (currentIndex + 1) % quantum;
+            }
+            currentBindings = null;
+            newBindings = false;
+            input.cancelBindings(bindings);
+            if ((nextBindings != null) && nextBindings.isAncestor(bindings)) {
+                nextBindings = null;
+            }
+            closed = true;
+        }
+
         // LookaheadExecution interface
 
         LookaheadExecution(QueryContext context, Cursor input, int quantum) {

=== modified file 'src/main/java/com/akiban/qp/operator/ChainedCursor.java'
--- src/main/java/com/akiban/qp/operator/ChainedCursor.java	2013-07-09 17:26:33 +0000
+++ src/main/java/com/akiban/qp/operator/ChainedCursor.java	2013-07-17 08:37:29 +0000
@@ -91,4 +91,10 @@
     public void closeBindings() {
         input.closeBindings();
     }
+
+    @Override
+    public void cancelBindings(QueryBindings ancestor) {
+        input.cancelBindings(ancestor);
+        close();                // In case override maintains some additional state.
+    }
 }

=== modified file 'src/main/java/com/akiban/qp/operator/HKeyUnion_Ordered.java'
--- src/main/java/com/akiban/qp/operator/HKeyUnion_Ordered.java	2013-07-09 19:53:07 +0000
+++ src/main/java/com/akiban/qp/operator/HKeyUnion_Ordered.java	2013-07-17 08:37:29 +0000
@@ -332,6 +332,13 @@
             rightInput.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            leftInput.cancelBindings(bindings);
+            rightInput.cancelBindings(bindings);
+            bindingsCursor.cancelBindings(bindings);
+        }
+
         // Execution interface
 
         Execution(QueryContext context, QueryBindingsCursor bindingsCursor)

=== modified file 'src/main/java/com/akiban/qp/operator/IndexScan_Default.java'
--- src/main/java/com/akiban/qp/operator/IndexScan_Default.java	2013-07-13 21:03:15 +0000
+++ src/main/java/com/akiban/qp/operator/IndexScan_Default.java	2013-07-17 08:37:29 +0000
@@ -537,6 +537,34 @@
             recyclePending();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            if ((currentBindings != null) && currentBindings.isAncestor(bindings)) {
+                if (currentCursor != null) {
+                    currentCursor.close();
+                    cursorPool.add(currentCursor);
+                    currentCursor = null;
+                }
+                if (pendingCursor != null) {
+                    pendingCursor.close();
+                    cursorPool.add(pendingCursor);
+                    pendingCursor = null;
+                }
+                currentBindings = null;
+            }
+            while (true) {
+                BindingsAndCursor bandc = pendingBindings.peek();
+                if (bandc == null) break;
+                if (!bandc.bindings.isAncestor(bindings)) break;
+                bandc = pendingBindings.remove();
+                if (bandc.cursor != null) {
+                    bandc.cursor.close();
+                    cursorPool.add(bandc.cursor);
+                }
+            }
+            bindingsCursor.cancelBindings(bindings);
+        }
+
         // LookaheadExecution interface
 
         LookaheadExecution(QueryContext context, QueryBindingsCursor bindingsCursor, 

=== modified file 'src/main/java/com/akiban/qp/operator/Intersect_Ordered.java'
--- src/main/java/com/akiban/qp/operator/Intersect_Ordered.java	2013-07-09 19:53:07 +0000
+++ src/main/java/com/akiban/qp/operator/Intersect_Ordered.java	2013-07-17 08:37:29 +0000
@@ -431,6 +431,13 @@
             rightInput.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            leftInput.cancelBindings(bindings);
+            rightInput.cancelBindings(bindings);
+            bindingsCursor.cancelBindings(bindings);
+        }
+
         // Execution interface
 
         Execution(QueryContext context, QueryBindingsCursor bindingsCursor)

=== modified file 'src/main/java/com/akiban/qp/operator/LeafCursor.java'
--- src/main/java/com/akiban/qp/operator/LeafCursor.java	2013-07-09 17:26:33 +0000
+++ src/main/java/com/akiban/qp/operator/LeafCursor.java	2013-07-17 08:37:29 +0000
@@ -42,4 +42,10 @@
     public void closeBindings() {
         bindingsCursor.closeBindings();
     }
+
+    @Override
+    public void cancelBindings(QueryBindings bindings) {
+        close();
+        bindingsCursor.cancelBindings(bindings);
+    }
 }

=== modified file 'src/main/java/com/akiban/qp/operator/Map_NestedLoops.java'
--- src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-13 20:10:49 +0000
+++ src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-17 08:37:29 +0000
@@ -219,6 +219,17 @@
             }
             input.closeBindings();
         }
+
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            if ((baseBindings != null) && baseBindings.isAncestor(bindings)) {
+                baseBindings = null;
+                input.close();
+            }
+            if (bindings.getDepth() < depth) {
+                input.cancelBindings(bindings);
+            }
+        }
     }
 
     // Other end of pipeline: remove the extra binding levels that we
@@ -228,8 +239,7 @@
     {
         private final Cursor input;
         private final int depth;
-        private QueryBindings pendingBindings;
-        private boolean open, inputOpen;
+        private QueryBindings currentBindings, pendingBindings, openBindings, inputOpenBindings;
         
         public CollapseBindingsCursor(QueryContext context, Cursor input, int depth) {
             super(context);
@@ -240,8 +250,8 @@
         @Override
         public void open() {
             CursorLifecycle.checkIdle(this);
-            open = true;
-            inputOpen = false;
+            openBindings = currentBindings;
+            inputOpenBindings = null;
         }
 
         @Override
@@ -256,25 +266,25 @@
                 checkQueryCancelation();
                 Row row = null;
                 while (true) {
-                    if (inputOpen) {
+                    if (inputOpenBindings != null) {
                         row = input.next();
                         if (row != null) break;
                         input.close();
-                        inputOpen = false;
+                        inputOpenBindings = null;
                     }
                     QueryBindings bindings = input.nextBindings();
                     if (bindings == null) {
-                        open = false;
+                        openBindings = null;
                         break;
                     }
                     if (bindings.getDepth() == depth) {
                         input.open();
-                        inputOpen = true;
+                        inputOpenBindings = bindings;
                     }
                     else if (bindings.getDepth() < depth) {
                         // End of this binding's rowset. Arrange for this to be next one.
                         pendingBindings = bindings;
-                        open = false;
+                        openBindings = null;
                         break;
                     }
                     else {
@@ -301,20 +311,10 @@
         @Override
         public void close() {
             CursorLifecycle.checkIdleOrActive(this);
-            if (open) {
-                if (inputOpen) {
-                    input.close();
-                    inputOpen = false;
-                }
-                // Advance bindings to where stream would have ended.
-                while (pendingBindings == null) {
-                    QueryBindings bindings = input.nextBindings();
-                    if (bindings == null) break;
-                    if (bindings.getDepth() < depth) {
-                        pendingBindings = bindings;
-                    }
-                }
-                open = false;
+            if (openBindings != null) {
+                cancelBindings(openBindings);
+                assert (inputOpenBindings == null);
+                assert (openBindings == null);
             }
         }
 
@@ -326,12 +326,12 @@
 
         @Override
         public boolean isIdle() {
-            return !input.isDestroyed() && !open;
+            return !input.isDestroyed() && (openBindings == null);
         }
 
         @Override
         public boolean isActive() {
-            return open;
+            return (openBindings != null);
         }
 
         @Override
@@ -341,23 +341,23 @@
 
         @Override
         public void openBindings() {
-            pendingBindings = null;
+            pendingBindings = currentBindings = null;
             input.openBindings();
         }
 
         @Override
         public QueryBindings nextBindings() {
-            QueryBindings bindings = pendingBindings;
-            if (bindings != null) {
+            currentBindings = pendingBindings;
+            if (currentBindings != null) {
                 pendingBindings = null;
-                return bindings;
+                return currentBindings;
             }
             while (true) {
                 // Skip over any that we would elide.
-                bindings = input.nextBindings();
-                if ((bindings == null) || (bindings.getDepth() < depth))
-                    return bindings;
-                assert (bindings.getDepth() == depth);
+                currentBindings = input.nextBindings();
+                if ((currentBindings == null) || (currentBindings.getDepth() < depth))
+                    return currentBindings;
+                assert (currentBindings.getDepth() == depth);
             }
         }
 
@@ -365,6 +365,20 @@
         public void closeBindings() {
             input.closeBindings();
         }
+
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            input.cancelBindings(bindings);
+            if ((inputOpenBindings != null) && inputOpenBindings.isAncestor(bindings)) {
+                inputOpenBindings = null;
+            }
+            if ((openBindings != null) && openBindings.isAncestor(bindings)) {
+                openBindings = null;
+            }
+            if ((pendingBindings != null) && pendingBindings.isAncestor(bindings)) {
+                pendingBindings = null;
+            }
+        }
     }
 
     // Old-style execution: bind outer row into existing context and
@@ -477,6 +491,13 @@
             outerInput.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            innerInput.close();
+            outerInput.cancelBindings(bindings);
+            closed = true;
+        }
+
         // Execution interface
 
         Execution(QueryContext context, QueryBindingsCursor bindingsCursor)

=== modified file 'src/main/java/com/akiban/qp/operator/MultipleQueryBindingsCursor.java'
--- src/main/java/com/akiban/qp/operator/MultipleQueryBindingsCursor.java	2013-07-09 21:47:07 +0000
+++ src/main/java/com/akiban/qp/operator/MultipleQueryBindingsCursor.java	2013-07-17 08:37:29 +0000
@@ -58,6 +58,11 @@
         cursors.get(0).closeBindings();
     }
 
+    @Override
+    public void cancelBindings(QueryBindings bindings) {
+        input.cancelBindings(bindings);
+    }
+
     public QueryBindingsCursor newCursor() {
         assert (offset == 0);
         SubCursor cursor = new SubCursor();
@@ -117,5 +122,19 @@
         public void closeBindings() {
             open = false;
         }
+
+        @Override
+        public void cancelBindings(QueryBindings ancestor) {
+            while (index - offset < buffer.size()) {
+                QueryBindings bindings = buffer.get(index - offset);
+                if (bindings.isAncestor(ancestor)) {
+                    index++;
+                }
+                else {
+                    break;
+                }
+            }
+            shrink();
+        }
     }
 }

=== modified file 'src/main/java/com/akiban/qp/operator/Product_NestedLoops.java'
--- src/main/java/com/akiban/qp/operator/Product_NestedLoops.java	2013-07-09 18:29:49 +0000
+++ src/main/java/com/akiban/qp/operator/Product_NestedLoops.java	2013-07-17 08:37:29 +0000
@@ -317,6 +317,13 @@
             outerInput.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            closeInner();
+            outerInput.cancelBindings(bindings);
+            closed = true;
+        }
+
         // Execution interface
 
         Execution(QueryContext context, QueryBindingsCursor bindingsCursor)

=== modified file 'src/main/java/com/akiban/qp/operator/QueryBindings.java'
--- src/main/java/com/akiban/qp/operator/QueryBindings.java	2013-07-10 17:31:42 +0000
+++ src/main/java/com/akiban/qp/operator/QueryBindings.java	2013-07-17 08:37:29 +0000
@@ -114,6 +114,11 @@
     public QueryBindings getParent();
 
     /**
+     * Is this bindings an descendant of the given bindings?
+     */
+    public boolean isAncestor(QueryBindings ancestor);
+
+    /**
      * Get the inheritance depth.
      */
     public int getDepth();

=== modified file 'src/main/java/com/akiban/qp/operator/QueryBindingsCursor.java'
--- src/main/java/com/akiban/qp/operator/QueryBindingsCursor.java	2013-07-09 21:47:07 +0000
+++ src/main/java/com/akiban/qp/operator/QueryBindingsCursor.java	2013-07-17 08:37:29 +0000
@@ -30,4 +30,7 @@
 
     /** Close stream of bindings. */
     public void closeBindings();
+
+    /** Advance stream past given bindings. */
+    public void cancelBindings(QueryBindings bindings);
 }

=== modified file 'src/main/java/com/akiban/qp/operator/Select_BloomFilter.java'
--- src/main/java/com/akiban/qp/operator/Select_BloomFilter.java	2013-07-09 18:29:49 +0000
+++ src/main/java/com/akiban/qp/operator/Select_BloomFilter.java	2013-07-17 08:37:29 +0000
@@ -338,6 +338,12 @@
             inputCursor.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            inputCursor.cancelBindings(bindings);
+            idle = true;
+        }
+
         // Execution interface
 
         <EXPR> Execution(QueryContext context, QueryBindingsCursor bindingsCursor,

=== modified file 'src/main/java/com/akiban/qp/operator/SingletonQueryBindingsCursor.java'
--- src/main/java/com/akiban/qp/operator/SingletonQueryBindingsCursor.java	2013-07-09 00:11:30 +0000
+++ src/main/java/com/akiban/qp/operator/SingletonQueryBindingsCursor.java	2013-07-17 08:37:29 +0000
@@ -54,6 +54,13 @@
         state = State.CLOSED;
     }
 
+    @Override
+    public void cancelBindings(QueryBindings ancestor) {
+        if ((state == State.PENDING) && bindings.isAncestor(ancestor)) {
+            state = State.EXHAUSTED;
+        }
+    }
+
     public void reset(QueryBindings bindings) {
         this.bindings = bindings;
         this.state = State.CLOSED;

=== modified file 'src/main/java/com/akiban/qp/operator/SparseArrayQueryBindings.java'
--- src/main/java/com/akiban/qp/operator/SparseArrayQueryBindings.java	2013-07-12 20:53:41 +0000
+++ src/main/java/com/akiban/qp/operator/SparseArrayQueryBindings.java	2013-07-17 08:37:29 +0000
@@ -246,6 +246,16 @@
     }
 
     @Override
+    public boolean isAncestor(QueryBindings ancestor) {
+        for (QueryBindings descendant = this; descendant != null; descendant = descendant.getParent()) {
+            if (descendant == ancestor) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Override
     public int getDepth() {
         return depth;
     }

=== modified file 'src/main/java/com/akiban/qp/operator/UnionAll_Default.java'
--- src/main/java/com/akiban/qp/operator/UnionAll_Default.java	2013-07-09 19:53:07 +0000
+++ src/main/java/com/akiban/qp/operator/UnionAll_Default.java	2013-07-17 08:37:29 +0000
@@ -327,6 +327,14 @@
             }
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            for (int i = 0; i < cursors.length; i++) {
+                cursors[i].cancelBindings(bindings);
+            }
+            bindingsCursor.cancelBindings(bindings);
+        }
+
         private Execution(QueryContext context, QueryBindingsCursor bindingsCursor)
         {
             super(context);

=== modified file 'src/main/java/com/akiban/qp/operator/Union_Ordered.java'
--- src/main/java/com/akiban/qp/operator/Union_Ordered.java	2013-07-09 19:53:07 +0000
+++ src/main/java/com/akiban/qp/operator/Union_Ordered.java	2013-07-17 08:37:29 +0000
@@ -333,6 +333,13 @@
             rightInput.closeBindings();
         }
 
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            leftInput.cancelBindings(bindings);
+            rightInput.cancelBindings(bindings);
+            bindingsCursor.cancelBindings(bindings);
+        }
+
         // Execution interface
 
         Execution(QueryContext context, QueryBindingsCursor bindingsCursor)

=== modified file 'src/test/java/com/akiban/qp/operator/QueryBindingsTest.java'
--- src/test/java/com/akiban/qp/operator/QueryBindingsTest.java	2013-07-10 18:06:04 +0000
+++ src/test/java/com/akiban/qp/operator/QueryBindingsTest.java	2013-07-17 08:37:29 +0000
@@ -77,6 +77,9 @@
         parent.setPValue(0, value);
         QueryBindings child = parent.createBindings();
         assertEquals(1, child.getDepth());
+        assertTrue(parent.isAncestor(parent));
+        assertTrue(child.isAncestor(parent));
+        assertFalse(parent.isAncestor(child));
         Deque<Row> rows = new RowsBuilder(MNumeric.INT.instance(false))
             .row(100)
             .row(101)

=== 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-13 19:25:25 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsIT.java	2013-07-17 08:37:29 +0000
@@ -303,6 +303,49 @@
         compareRows(expected, cursor(plan, queryContext, queryBindings));
     }
 
+    @Test
+    public void testDeepMapLimit()
+    {
+        RowType intRowType = schema.newValuesType(AkType.INT);
+        List<ExpressionGenerator> expressions = Arrays.asList(boundField(intRowType, 0, 0), boundField(intRowType, 1, 0), field(intRowType, 0));
+        Operator inside = 
+            project_Default(
+                valuesScan_Default(
+                    bindableExpressions(intRow(intRowType, 1),
+                                        intRow(intRowType, 2),
+                                        intRow(intRowType, 3)),
+                    intRowType),
+                intRowType, expressions);
+        RowType insideRowType = inside.rowType();
+        Operator plan =
+            map_NestedLoops(
+                valuesScan_Default(
+                    bindableExpressions(intRow(intRowType, 100),
+                                        intRow(intRowType, 200),
+                                        intRow(intRowType, 300)),
+                    intRowType),
+                limit_Default(
+                    map_NestedLoops(
+                        valuesScan_Default(
+                            bindableExpressions(intRow(intRowType, 10),
+                                                intRow(intRowType, 20),
+                                                intRow(intRowType, 30)),
+                            intRowType),
+                        inside,
+                        1, pipelineMap(), 2),
+                    2),
+                0, pipelineMap(), 1);
+        RowBase[] expected = new RowBase[]{
+            row(insideRowType, 100L, 10L, 1L),
+            row(insideRowType, 100L, 10L, 2L),
+            row(insideRowType, 200L, 10L, 1L),
+            row(insideRowType, 200L, 10L, 2L),
+            row(insideRowType, 300L, 10L, 1L),
+            row(insideRowType, 300L, 10L, 2L),
+        };
+        compareRows(expected, cursor(plan, queryContext, queryBindings));
+    }
+
     private Row intRow(RowType rowType, int x)
     {
         List<Expression> expressions;

=== modified file 'src/test/java/com/akiban/server/test/pt/AggregatePT.java'
--- src/test/java/com/akiban/server/test/pt/AggregatePT.java	2013-07-13 17:24:25 +0000
+++ src/test/java/com/akiban/server/test/pt/AggregatePT.java	2013-07-17 08:37:29 +0000
@@ -390,6 +390,11 @@
         public void closeBindings() {
             inputCursor.closeBindings();
         }
+
+        @Override
+        public void cancelBindings(QueryBindings bindings) {
+            inputCursor.cancelBindings(bindings);
+        }
     }
 
     static final AkType[] TYPES = {