Merge lp:~mmcm/akiban-server/operator-pipeline-options into lp:~akiban-technologies/akiban-server/trunk

[Mike McMahon]

Change how operators get pipeline-related options.

Instead of having each operator look into config properties, set them from the optimizer, which in turn gets them all at once from there.

Add an option for Map_NestedLoops to select the previous rebinding implementation. Make this the default, since there are issues with query cancelation with pipelining.

[Nathan Williams (nwilliams)]

Looks good to me.

review: Approve

=== modified file 'src/main/java/com/akiban/qp/operator/API.java'
--- src/main/java/com/akiban/qp/operator/API.java	2013-07-10 18:42:14 +0000
+++ src/main/java/com/akiban/qp/operator/API.java	2013-07-13 21:55:29 +0000
@@ -497,9 +497,11 @@
     public static Operator map_NestedLoops(Operator outerInput,
                                            Operator innerInput,
                                            int inputBindingPosition,
+                                           boolean pipeline,
                                            int depth)
     {
-        return new Map_NestedLoops(outerInput, innerInput, inputBindingPosition, depth);
+        return new Map_NestedLoops(outerInput, innerInput, inputBindingPosition, 
+                                   pipeline, depth);
     }
 
     // IfEmpty

=== modified file 'src/main/java/com/akiban/qp/operator/Map_NestedLoops.java'
--- src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-10 21:30:34 +0000
+++ src/main/java/com/akiban/qp/operator/Map_NestedLoops.java	2013-07-13 21:55:29 +0000
@@ -51,6 +51,8 @@
 
  <li><b>int inputBindingPosition:</b> Position of inner loop row in query context.
 
+ <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.
 
  </ul>
@@ -90,7 +92,7 @@
     @Override
     protected Cursor cursor(QueryContext context, QueryBindingsCursor bindingsCursor)
     {
-        if (false)
+        if (!pipeline)
             return new Execution(context, bindingsCursor); // Old-style
         else {
             Cursor outerCursor = outerInputOperator.cursor(context, bindingsCursor);
@@ -127,6 +129,7 @@
     public Map_NestedLoops(Operator outerInputOperator,
                            Operator innerInputOperator,
                            int inputBindingPosition,
+                           boolean pipeline,
                            int depth)
     {
         ArgumentValidation.notNull("outerInputOperator", outerInputOperator);
@@ -136,6 +139,7 @@
         this.outerInputOperator = outerInputOperator;
         this.innerInputOperator = innerInputOperator;
         this.inputBindingPosition = inputBindingPosition;
+        this.pipeline = pipeline;
         this.depth = depth;
     }
 
@@ -150,12 +154,14 @@
     private final Operator outerInputOperator;
     private final Operator innerInputOperator;
     private final int inputBindingPosition, depth;
+    private final boolean pipeline;
 
     @Override
     public CompoundExplainer getExplainer(ExplainContext context)
     {
         CompoundExplainer ex = new NestedLoopsExplainer(getName(), innerInputOperator, outerInputOperator, null, null, context);
         ex.addAttribute(Label.BINDING_POSITION, PrimitiveExplainer.getInstance(inputBindingPosition));
+        ex.addAttribute(Label.PIPELINE, PrimitiveExplainer.getInstance(pipeline));
         ex.addAttribute(Label.DEPTH, PrimitiveExplainer.getInstance(depth));
         if (context.hasExtraInfo(this))
             ex.get().putAll(context.getExtraInfo(this).get());
@@ -346,9 +352,13 @@
                 pendingBindings = null;
                 return bindings;
             }
-            bindings = input.nextBindings();
-            assert ((bindings == null) || (bindings.getDepth() < depth));
-            return bindings;
+            while (true) {
+                // Skip over any that we would elide.
+                bindings = input.nextBindings();
+                if ((bindings == null) || (bindings.getDepth() < depth))
+                    return bindings;
+                assert (bindings.getDepth() == depth);
+            }
         }
 
         @Override

=== modified file 'src/main/java/com/akiban/server/explain/Label.java'
--- src/main/java/com/akiban/server/explain/Label.java	2013-07-10 18:42:14 +0000
+++ src/main/java/com/akiban/server/explain/Label.java	2013-07-13 21:55:29 +0000
@@ -48,6 +48,7 @@
     INFIX_REPRESENTATION(Category.DESCRIPTION),
     ASSOCIATIVE(Category.DESCRIPTION),
     INDEX(Category.DESCRIPTION),
+    PIPELINE(Category.DESCRIPTION),
     DEPTH(Category.DESCRIPTION),
     
     // IDENTIFIER

=== modified file 'src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java'
--- src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-10 18:42:14 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-13 21:55:29 +0000
@@ -1529,6 +1529,7 @@
             stream.operator = API.map_NestedLoops(ostream.operator, 
                                                   stream.operator,
                                                   currentBindingPosition(),
+                                                  rulesContext.getPipelineConfiguration().isMapEnabled(),
                                                   nestedBindingsDepth);
             nestedBindingsDepth--;
             popBoundRow();

=== added file 'src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java'
--- src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/PipelineConfiguration.java	2013-07-13 21:55:29 +0000
@@ -0,0 +1,67 @@
+/**
+ * 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.sql.optimizer.rule;
+
+import java.util.Properties;
+
+public class PipelineConfiguration
+{
+    private boolean mapEnabled = false;
+    private int indexScanLookaheadQuantum = 1;
+    private int groupLookupLookaheadQuantum = 1;
+    private boolean unionAllOpenBoth = false;
+
+    public PipelineConfiguration() {
+    }
+
+    public PipelineConfiguration(Properties properties) {
+        load(properties);
+    }
+
+    public boolean isMapEnabled() {
+        return mapEnabled;
+    }
+
+    public int getIndexScanLookaheadQuantum() {
+        return indexScanLookaheadQuantum;
+    }
+
+    public int getGroupLookupLookaheadQuantum() {
+        return groupLookupLookaheadQuantum;
+    }
+
+    public boolean isUnionAllOpenBoth() {
+        return unionAllOpenBoth;
+    }
+
+    public void load(Properties properties) {
+        for (String prop : properties.stringPropertyNames()) {
+            String val = properties.getProperty(prop);
+            if ("map.enabled".equals(prop))
+                mapEnabled = Boolean.parseBoolean(val);
+            else if ("indexScan.lookaheadQuantum".equals(prop))
+                indexScanLookaheadQuantum = Integer.parseInt(val);
+            else if ("groupLookup.lookaheadQuantum".equals(prop))
+                groupLookupLookaheadQuantum = Integer.parseInt(val);
+            else if ("unionAll.openBoth".equals(prop))
+                unionAllOpenBoth = Boolean.parseBoolean(val);
+            else
+                throw new IllegalArgumentException("Unknown property " + prop);
+        }
+    }
+}

=== modified file 'src/main/java/com/akiban/sql/optimizer/rule/SchemaRulesContext.java'
--- src/main/java/com/akiban/sql/optimizer/rule/SchemaRulesContext.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/SchemaRulesContext.java	2013-07-13 21:55:29 +0000
@@ -36,6 +36,8 @@
     private FunctionsRegistry functionsRegistry;
     private CostEstimator costEstimator;
     private T3RegistryService t3Registry;
+    private PipelineConfiguration pipelineConfiguration;
+
     protected SchemaRulesContext() {
     }
 
@@ -55,12 +57,17 @@
         this.costEstimator = costEstimator;
     }
 
+    protected void initPipelineConfiguration(PipelineConfiguration pipelineConfiguration) {
+        this.pipelineConfiguration = pipelineConfiguration;
+    }
+
     @Override
     protected void initDone() {
         super.initDone();
         assert (schema != null) : "initSchema() not called";
         assert (functionsRegistry != null) : "initFunctionsRegistry() not called";
         assert (costEstimator != null) : "initCostEstimator() not called";
+        assert (pipelineConfiguration != null) : "initPipelineConfiguration() not called";
     }
 
     public Schema getSchema() {
@@ -89,4 +96,8 @@
 
     public abstract String getDefaultSchemaName();
 
+    public PipelineConfiguration getPipelineConfiguration() {
+        return pipelineConfiguration;
+    }
+
 }

=== modified file 'src/main/java/com/akiban/sql/server/ServerOperatorCompiler.java'
--- src/main/java/com/akiban/sql/server/ServerOperatorCompiler.java	2013-06-11 16:30:17 +0000
+++ src/main/java/com/akiban/sql/server/ServerOperatorCompiler.java	2013-07-13 21:55:29 +0000
@@ -36,6 +36,7 @@
         initParser(server.getParser());
         initFunctionsRegistry(server.functionsRegistry());
         initCostEstimator(server.costEstimator(this, keyCreator), usePValues);
+        initPipelineConfiguration(server.getPipelineConfiguration());
         if (usePValues)
             initT3Registry(server.t3RegistryService());
         

=== modified file 'src/main/java/com/akiban/sql/server/ServerSession.java'
--- src/main/java/com/akiban/sql/server/ServerSession.java	2013-06-11 16:32:59 +0000
+++ src/main/java/com/akiban/sql/server/ServerSession.java	2013-07-13 21:55:29 +0000
@@ -23,6 +23,7 @@
 import com.akiban.sql.parser.SQLParser;
 
 import com.akiban.sql.optimizer.AISBinderContext;
+import com.akiban.sql.optimizer.rule.PipelineConfiguration;
 import com.akiban.sql.optimizer.rule.cost.CostEstimator;
 
 import com.akiban.ais.model.AkibanInformationSchema;
@@ -165,4 +166,7 @@
 
     /** Check access to given schema */
     public boolean isSchemaAccessible(String schemaName);
+
+    /** Get the pipeline configuration. */
+    public PipelineConfiguration getPipelineConfiguration();
 }

=== modified file 'src/main/java/com/akiban/sql/server/ServerSessionBase.java'
--- src/main/java/com/akiban/sql/server/ServerSessionBase.java	2013-06-12 21:51:17 +0000
+++ src/main/java/com/akiban/sql/server/ServerSessionBase.java	2013-07-13 21:55:29 +0000
@@ -40,6 +40,7 @@
 import com.akiban.server.service.tree.KeyCreator;
 import com.akiban.server.t3expressions.T3RegistryService;
 import com.akiban.sql.optimizer.AISBinderContext;
+import com.akiban.sql.optimizer.rule.PipelineConfiguration;
 import com.akiban.sql.optimizer.rule.cost.CostEstimator;
 
 import java.util.*;
@@ -47,10 +48,12 @@
 public abstract class ServerSessionBase extends AISBinderContext implements ServerSession
 {
     public static final String COMPILER_PROPERTIES_PREFIX = "optimizer.";
+    public static final String PIPELINE_PROPERTIES_PREFIX = "akserver.pipeline.";
 
     protected final ServerServiceRequirements reqs;
     protected Properties compilerProperties;
     protected Map<String,Object> attributes = new HashMap<>();
+    protected PipelineConfiguration pipelineConfiguration;
     
     protected Session session;
     protected Map<StoreAdapter.AdapterType, StoreAdapter> adapters = 
@@ -401,4 +404,11 @@
         return reqs.securityService().isAccessible(session, schemaName);
     }
 
+    @Override
+    public PipelineConfiguration getPipelineConfiguration() {
+        if (pipelineConfiguration == null)
+            pipelineConfiguration = new PipelineConfiguration(reqs.config().deriveProperties(PIPELINE_PROPERTIES_PREFIX));
+        return pipelineConfiguration;
+    }
+
 }

=== modified file 'src/test/java/com/akiban/server/test/ApiTestBase.java'
--- src/test/java/com/akiban/server/test/ApiTestBase.java	2013-07-03 15:32:40 +0000
+++ src/test/java/com/akiban/server/test/ApiTestBase.java	2013-07-13 21:55:29 +0000
@@ -1353,6 +1353,10 @@
         return true;
     }
 
+    protected boolean pipelineMap() {
+        return false;
+    }
+
     protected DDLFunctions ddlForAlter() {
         return ddl();
     }

=== modified file 'src/test/java/com/akiban/server/test/costmodel/MapCT.java'
--- src/test/java/com/akiban/server/test/costmodel/MapCT.java	2013-07-10 18:42:14 +0000
+++ src/test/java/com/akiban/server/test/costmodel/MapCT.java	2013-07-13 21:55:29 +0000
@@ -83,7 +83,8 @@
         TimeOperator timeSetupOuter = new TimeOperator(setupOuter);
         Operator setupInner = limit_Default(groupScan_Default(group), innerRows);
         TimeOperator timeSetupInner = new TimeOperator(setupInner);
-        Operator plan = map_NestedLoops(timeSetupOuter, timeSetupInner, 0, 1);
+        Operator plan = map_NestedLoops(timeSetupOuter, timeSetupInner, 
+                                        0, pipelineMap(), 1);
         long start = System.nanoTime();
         for (int r = 0; r < runs; r++) {
             Cursor cursor = cursor(plan, queryContext, queryBindings);

=== modified file 'src/test/java/com/akiban/server/test/it/qp/AncestorLookup_NestedIT.java'
--- src/test/java/com/akiban/server/test/it/qp/AncestorLookup_NestedIT.java	2013-07-10 21:30:34 +0000
+++ src/test/java/com/akiban/server/test/it/qp/AncestorLookup_NestedIT.java	2013-07-13 21:55:29 +0000
@@ -141,7 +141,7 @@
             map_NestedLoops(
                 indexScan_Default(aValueIndexRowType),
                 ancestorLookup_Nested(rabc, aValueIndexRowType, Collections.singleton(aRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(aRowType, 13L, 1L, "a13"),
@@ -159,7 +159,7 @@
             map_NestedLoops(
                 indexScan_Default(aValueIndexRowType),
                 ancestorLookup_Nested(rabc, aValueIndexRowType, Arrays.asList(aRowType, rRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(rRowType, 1L, "r1"),
@@ -181,7 +181,7 @@
             map_NestedLoops(
                 indexScan_Default(aValueIndexRowType),
                 ancestorLookup_Nested(rabc, aValueIndexRowType, Arrays.asList(rRowType, aRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(rRowType, 1L, "r1"),
@@ -204,9 +204,9 @@
                 map_NestedLoops(
                     indexScan_Default(aValueIndexRowType),
                     ancestorLookup_Nested(rabc, aValueIndexRowType, Arrays.asList(aRowType), 0),
-                    0, 1),
+                    0, pipelineMap(), 1),
                 ancestorLookup_Nested(rabc, aRowType, Arrays.asList(rRowType), 1),
-                1, 1);
+                1, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(rRowType, 1L, "r1"),
@@ -236,7 +236,7 @@
             map_NestedLoops(
                 abIndexScan,
                 ancestorLookup_Nested(rabc, abIndexScan.rowType(), Collections.singleton(rRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(rRowType, 1L, "r1"),
@@ -272,7 +272,7 @@
             map_NestedLoops(
                 abcIndexScan,
                 ancestorLookup_Nested(rabc, abcIndexScan.rowType(), Collections.singleton(rRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(rRowType, 1L, "r1"),
@@ -287,7 +287,7 @@
             map_NestedLoops(
                 indexScan_Default(aValueIndexRowType),
                 ancestorLookup_Nested(rabc, aValueIndexRowType, Collections.singleton(aRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         CursorLifecycleTestCase testCase = new CursorLifecycleTestCase()
         {
             @Override

=== modified file 'src/test/java/com/akiban/server/test/it/qp/BranchLookup_NestedIT.java'
--- src/test/java/com/akiban/server/test/it/qp/BranchLookup_NestedIT.java	2013-07-10 21:30:34 +0000
+++ src/test/java/com/akiban/server/test/it/qp/BranchLookup_NestedIT.java	2013-07-13 21:55:29 +0000
@@ -154,7 +154,7 @@
             map_NestedLoops(
                 indexScan_Default(aValueIndexRowType),
                 branchLookup_Nested(rabc, aValueIndexRowType, rRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             // Each r row, and everything below it, is duplicated, because the A index refers to each r value twice.
@@ -202,7 +202,7 @@
                     Collections.singleton(aRowType),
                     InputPreservationOption.DISCARD_INPUT),
                 branchLookup_Nested(rabc, aRowType, rRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             // Each r row, and everything below it, is duplicated, because the A index refers to each r value twice.
@@ -247,7 +247,7 @@
                     groupScan_Default(rabc),
                     Collections.singleton(aRowType)),
                 branchLookup_Nested(rabc, aRowType, bRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(bRowType, 15L, 1L, "b15"),
@@ -272,9 +272,9 @@
                         groupScan_Default(rabc),
                         Collections.singleton(aRowType)),
                     branchLookup_Nested(rabc, aRowType, bRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                    0, 1),
+                    0, pipelineMap(), 1),
                 branchLookup_Nested(rabc, bRowType, cRowType, InputPreservationOption.KEEP_INPUT, 1),
-                1, 1);
+                1, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(bRowType, 15L, 1L, "b15"),
@@ -324,7 +324,7 @@
             map_NestedLoops(
                 abIndexScan,
                 branchLookup_Nested(rabc, abIndexScan.rowType(), cRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                0, 1);
+                0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
             row(cRowType, 17L, 1L, "c17"),
@@ -342,7 +342,7 @@
                                 groupScan_Default(rabc),
                                 Collections.singleton(aRowType)),
                         branchLookup_Nested(rabc, aRowType, rRowType, aRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                        0, 1);
+                        0, pipelineMap(), 1);
         Cursor cursor = cursor(plan, queryContext, queryBindings);
         RowBase[] expected = new RowBase[]{
                 row(aRowType, 13L, 1L, "a13"),
@@ -366,7 +366,7 @@
                     groupScan_Default(rabc),
                     Collections.singleton(aRowType)),
                 branchLookup_Nested(rabc, aRowType, rRowType, aRowType, InputPreservationOption.DISCARD_INPUT, 0),
-                0, 1);
+                0, pipelineMap(), 1);
         CursorLifecycleTestCase testCase = new CursorLifecycleTestCase()
         {
             @Override

=== 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-10 21:30:34 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsIT.java	2013-07-13 21:55:29 +0000
@@ -102,25 +102,25 @@
     @Test(expected = IllegalArgumentException.class)
     public void testLeftInputNull()
     {
-        map_NestedLoops(null, groupScan_Default(coi), 0, 1);
+        map_NestedLoops(null, groupScan_Default(coi), 0, pipelineMap(), 1);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testRightInputNull()
     {
-        map_NestedLoops(groupScan_Default(coi), null, 0, 1);
+        map_NestedLoops(groupScan_Default(coi), null, 0, pipelineMap(), 1);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNegativeInputBindingPosition()
     {
-        map_NestedLoops(groupScan_Default(coi), groupScan_Default(coi), -1, 1);
+        map_NestedLoops(groupScan_Default(coi), groupScan_Default(coi), -1, pipelineMap(), 1);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNonPositiveDepth()
     {
-        map_NestedLoops(groupScan_Default(coi), groupScan_Default(coi), 0, 0);
+        map_NestedLoops(groupScan_Default(coi), groupScan_Default(coi), 0, pipelineMap(),0);
     }
 
     // Test operator execution
@@ -132,7 +132,7 @@
             map_NestedLoops(
                 indexScan_Default(itemOidIndexRowType, false),
                 ancestorLookup_Nested(coi, itemOidIndexRowType, Collections.singleton(itemRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         RowBase[] expected = new RowBase[]{
             row(itemRowType, 1000L, 100L),
             row(itemRowType, 1001L, 100L),
@@ -175,7 +175,7 @@
                     groupScan_Default(coi),
                     Collections.singleton(customerRowType)),
                 project,
-                0, 1);
+                0, pipelineMap(), 1);
         RowType projectRowType = project.rowType();
         RowBase[] expected = new RowBase[]{
             row(projectRowType, 1L, 100L),
@@ -212,7 +212,7 @@
                     groupScan_Default(coi),
                     Collections.singleton(customerRowType)),
                 ifEmpty_Default(project, projectRowType, Arrays.asList(boundField(customerRowType, 0, 0), literal(null)), InputPreservationOption.KEEP_INPUT),
-                0, 1);
+                0, pipelineMap(), 1);
         RowBase[] expected = new RowBase[]{
             row(projectRowType, 1L, 100L),
             row(projectRowType, 1L, 101L),
@@ -234,7 +234,7 @@
             map_NestedLoops(
                 indexScan_Default(itemOidIndexRowType, false),
                 ancestorLookup_Nested(coi, itemOidIndexRowType, Collections.singleton(itemRowType), 0),
-                0, 1);
+                0, pipelineMap(), 1);
         CursorLifecycleTestCase testCase = new CursorLifecycleTestCase()
         {
             @Override
@@ -291,8 +291,8 @@
                 map_NestedLoops(
                     indexScan_Default(customerCidIndexRowType, false, cidRange),
                     ancestorLookup_Nested(coi, customerCidIndexRowType, Collections.singleton(customerRowType), 0),
-                    0, 2),
-                1, 1);
+                    0, pipelineMap(), 2),
+                1, pipelineMap(), 1);
         RowBase[] expected = new RowBase[]{
             row(customerRowType, 1L, "northbridge"),
             row(customerRowType, 2L, "foundation"),

=== added file 'src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsPipelineIT.java'
--- src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsPipelineIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Map_NestedLoopsPipelineIT.java	2013-07-13 21:55:29 +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 Map_NestedLoopsPipelineIT extends Map_NestedLoopsIT 
+{
+    @Override
+    protected boolean pipelineMap() {
+        return true;
+    }
+}

=== modified file 'src/test/java/com/akiban/server/test/it/qp/Sort_InsertionLimitedIT.java'
--- src/test/java/com/akiban/server/test/it/qp/Sort_InsertionLimitedIT.java	2013-07-10 18:42:14 +0000
+++ src/test/java/com/akiban/server/test/it/qp/Sort_InsertionLimitedIT.java	2013-07-13 21:55:29 +0000
@@ -540,7 +540,7 @@
                 map_NestedLoops(
                     filter_Default(groupScan_Default(coi),
                                    Collections.singleton(customerRowType)),
-                    project, 0, 1),
+                    project, 0, pipelineMap(), 1),
                 projectType,
                 ordering(field(projectType, 0), true),
                 SortOption.PRESERVE_DUPLICATES,

=== modified file 'src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultIT.java'
--- src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultIT.java	2013-07-10 18:42:14 +0000
+++ src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultIT.java	2013-07-13 21:55:29 +0000
@@ -270,7 +270,7 @@
                                       xEQ9Range,
                                       ordering),
                     txIndexRowType),
-                0, 1);
+                0, pipelineMap(), 1);
         String[] expected = new String[]{
             hKey(1000),
             hKey(1002),

=== modified file 'src/test/java/com/akiban/sql/optimizer/OperatorCompilerTest.java'
--- src/test/java/com/akiban/sql/optimizer/OperatorCompilerTest.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/sql/optimizer/OperatorCompilerTest.java	2013-07-13 21:55:29 +0000
@@ -33,6 +33,7 @@
 import com.akiban.sql.optimizer.plan.ResultSet.ResultField;
 import com.akiban.sql.optimizer.rule.ExplainPlanContext;
 import com.akiban.sql.optimizer.rule.RulesTestHelper;
+import com.akiban.sql.optimizer.rule.PipelineConfiguration;
 import com.akiban.sql.optimizer.rule.cost.TestCostEstimator;
 
 import com.akiban.junit.NamedParameterizedRunner;
@@ -122,6 +123,7 @@
                 compiler.initT3Registry(t3Registry);
             }
             compiler.initCostEstimator(new TestCostEstimator(ais, compiler.getSchema(), statsFile, false, properties), usePValues);
+            compiler.initPipelineConfiguration(new PipelineConfiguration());
             compiler.initDone();
             return compiler;
         }

=== modified file 'src/test/java/com/akiban/sql/optimizer/rule/RulesTestContext.java'
--- src/test/java/com/akiban/sql/optimizer/rule/RulesTestContext.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/sql/optimizer/rule/RulesTestContext.java	2013-07-13 21:55:29 +0000
@@ -50,6 +50,7 @@
         context.initCostEstimator(new TestCostEstimator(ais, context.getSchema(), 
                                                         statsFile, statsIgnoreMissingIndexes,
                                                         properties), false);
+        context.initPipelineConfiguration(new PipelineConfiguration());
         context.initDone();
         return context;
     }