Merge lp:~mmcm/akiban-server/pipeline-union-all into lp:~akiban-technologies/akiban-server/trunk

[Mike McMahon]

Optionally open both sides of UnionAll_Default.

This is not semantically necessary, unlike Union_Ordered, since the semantics can be satisfied by concatenating the two streams.

While here, rename the API entry to be consistent with everything else.

[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 20:38:26 +0000
+++ src/main/java/com/akiban/qp/operator/API.java	2013-07-18 20:38:26 +0000
@@ -588,9 +588,9 @@
 
     // Union
 
-    public static Operator unionAll(Operator input1, RowType input1RowType, Operator input2, RowType input2RowType)
+    public static Operator unionAll_Default(Operator input1, RowType input1RowType, Operator input2, RowType input2RowType, boolean openBoth)
     {
-        return new UnionAll_Default(input1, input1RowType, input2, input2RowType, USE_PVALUES);
+        return new UnionAll_Default(input1, input1RowType, input2, input2RowType, USE_PVALUES, openBoth);
     }
     
     // Intersect

=== modified file 'src/main/java/com/akiban/qp/operator/UnionAll_Default.java'
--- src/main/java/com/akiban/qp/operator/UnionAll_Default.java	2013-07-15 22:01:49 +0000
+++ src/main/java/com/akiban/qp/operator/UnionAll_Default.java	2013-07-18 20:38:26 +0000
@@ -53,6 +53,7 @@
  <li><b>RowType input1Type:</b> Type of rows in first input stream. 
  <li><b>Operator input2:</b> Source of second input stream. 
  <li><b>RowType input2Type:</b> Type of rows in second input stream. 
+ <li><b>boolean openBoth:</b> Whether to open both input cursors at the same time rather than as needed. 
 
  <h1>Behavior</h1>
 
@@ -105,7 +106,7 @@
         return new Execution(context, bindingsCursor);
     }
 
-    UnionAll_Default(Operator input1, RowType input1Type, Operator input2, RowType input2Type, boolean usePValues) {
+    UnionAll_Default(Operator input1, RowType input1Type, Operator input2, RowType input2Type, boolean usePValues, boolean openBoth) {
         ArgumentValidation.notNull("first input", input1);
         ArgumentValidation.notNull("first input type", input1Type);
         ArgumentValidation.notNull("second input", input2);
@@ -114,6 +115,7 @@
         this.inputs = Arrays.asList(input1, input2);
         this.inputTypes = Arrays.asList(input1Type, input2Type);
         ArgumentValidation.isEQ("inputs.size", inputs.size(), "inputTypes.size", inputTypes.size());
+        this.openBoth = openBoth;
     }
 
     // for use in this package (in ctor and unit tests)
@@ -186,6 +188,7 @@
     private final List<? extends Operator> inputs;
     private final List<? extends RowType> inputTypes;
     private final RowType outputRowType;
+    private final boolean openBoth;
 
     @Override
     public CompoundExplainer getExplainer(ExplainContext context)
@@ -202,6 +205,8 @@
        
         att.put(Label.OUTPUT_TYPE, outputRowType.getExplainer(context));
         
+        att.put(Label.PIPELINE, PrimitiveExplainer.getInstance(openBoth));
+
         return new CompoundExplainer(Type.UNION, att);
     }
 
@@ -213,8 +218,11 @@
             try {
                 CursorLifecycle.checkIdle(this);
                 idle = false;
-                // TODO: Have a mode where all the cursors get opened
-                // so that they can start in parallel.
+                if (openBoth) {
+                    for (int i = 0; i < cursors.length; i++) {
+                        cursors[i].open();
+                    }
+                }
             } finally {
                 TAP_OPEN.out();
             }
@@ -261,11 +269,16 @@
         @Override
         public void close() {
             CursorLifecycle.checkIdleOrActive(this);
-            inputOperatorsIndex = -1;
             if (currentCursor != null) {
                 currentCursor.close();
                 currentCursor = null;
             }
+            if (openBoth) {
+                while (++inputOperatorsIndex < inputs.size()) {
+                    cursors[inputOperatorsIndex].close();
+                }
+            }
+            inputOperatorsIndex = -1;
             currentInputRowType = null;
             rowHolder.release();
             idle = true;
@@ -355,7 +368,9 @@
         private Row nextCursorFirstRow() {
             while (++inputOperatorsIndex < inputs.size()) {
                 Cursor nextCursor = cursors[inputOperatorsIndex];
-                nextCursor.open();
+                if (!openBoth) {
+                    nextCursor.open();
+                }
                 Row nextRow = nextCursor.next();
                 if (nextRow == null) {
                     nextCursor.close();

=== modified file 'src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java'
--- src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-17 20:48:07 +0000
+++ src/main/java/com/akiban/sql/optimizer/rule/OperatorAssembler.java	2013-07-18 20:38:26 +0000
@@ -1243,7 +1243,7 @@
                                                             ascending, unionOrderedAll);
                     }
                     else {
-                        stream.operator = API.unionAll(stream.operator, stream.rowType, scan, indexRowType);
+                        stream.operator = API.unionAll_Default(stream.operator, stream.rowType, scan, indexRowType, rulesContext.getPipelineConfiguration().isUnionAllOpenBoth());
                         stream.rowType = stream.operator.rowType();
                     }
                 }

=== added file 'src/test/java/com/akiban/qp/operator/UnionAll_DefaultOpenBothTest.java'
--- src/test/java/com/akiban/qp/operator/UnionAll_DefaultOpenBothTest.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/qp/operator/UnionAll_DefaultOpenBothTest.java	2013-07-18 20:38:26 +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.qp.operator;
+
+public class UnionAll_DefaultOpenBothTest extends UnionAll_DefaultTest 
+{
+    @Override
+    protected boolean openBoth() {
+        return true;
+    }
+}

=== modified file 'src/test/java/com/akiban/qp/operator/UnionAll_DefaultTest.java'
--- src/test/java/com/akiban/qp/operator/UnionAll_DefaultTest.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/qp/operator/UnionAll_DefaultTest.java	2013-07-18 20:38:26 +0000
@@ -27,8 +27,12 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 
-public final class UnionAll_DefaultTest {
+public class UnionAll_DefaultTest {
     
+    protected boolean openBoth() {
+        return false;
+    }
+
     @Test
     public void unionTwoNormal() {
         DerivedTypesSchema schema = new DerivedTypesSchema();
@@ -167,7 +171,7 @@
         DerivedTypesSchema schema = new DerivedTypesSchema();
         RowsBuilder first = new RowsBuilder(schema, AkType.LONG, AkType.VARCHAR);
         RowsBuilder second = new RowsBuilder(schema, AkType.LONG, AkType.NULL);
-        new UnionAll_Default(null, first.rowType(), new TestOperator(second), second.rowType(), Types3Switch.ON);
+        new UnionAll_Default(null, first.rowType(), new TestOperator(second), second.rowType(), Types3Switch.ON, openBoth());
     }
 
     @Test(expected = IllegalArgumentException.class)
@@ -175,7 +179,7 @@
         DerivedTypesSchema schema = new DerivedTypesSchema();
         RowsBuilder first = new RowsBuilder(schema, AkType.LONG, AkType.VARCHAR);
         RowsBuilder second = new RowsBuilder(schema, AkType.LONG, AkType.NULL);
-        new UnionAll_Default(new TestOperator(first), null, new TestOperator(second), second.rowType(), Types3Switch.ON);
+        new UnionAll_Default(new TestOperator(first), null, new TestOperator(second), second.rowType(), Types3Switch.ON, openBoth());
     }
 
     @Test(expected = IllegalArgumentException.class)
@@ -183,7 +187,7 @@
         DerivedTypesSchema schema = new DerivedTypesSchema();
         RowsBuilder first = new RowsBuilder(schema, AkType.LONG, AkType.VARCHAR);
         RowsBuilder second = new RowsBuilder(schema, AkType.LONG, AkType.NULL);
-        new UnionAll_Default(new TestOperator(first), first.rowType(), null, second.rowType(), Types3Switch.ON);
+        new UnionAll_Default(new TestOperator(first), first.rowType(), null, second.rowType(), Types3Switch.ON, openBoth());
     }
 
     @Test(expected = IllegalArgumentException.class)
@@ -191,7 +195,7 @@
         DerivedTypesSchema schema = new DerivedTypesSchema();
         RowsBuilder first = new RowsBuilder(schema, AkType.LONG, AkType.VARCHAR);
         RowsBuilder second = new RowsBuilder(schema, AkType.LONG, AkType.NULL);
-        new UnionAll_Default(new TestOperator(first), first.rowType(), new TestOperator(second), null, Types3Switch.ON);
+        new UnionAll_Default(new TestOperator(first), first.rowType(), new TestOperator(second), null, Types3Switch.ON, openBoth());
     }
 
     /**
@@ -213,7 +217,7 @@
         OperatorTestHelper.execute(union);
     }
 
-    private static void check(Operator union, RowsBuilder expected) {
+    private void check(Operator union, RowsBuilder expected) {
         final RowType outputRowType = union.rowType();
         checkRowTypes(expected.rowType(), outputRowType);
 
@@ -225,7 +229,7 @@
         });
     }
 
-    private static void check(RowsBuilder rb1, RowsBuilder rb2, RowsBuilder expected) {
+    private void check(RowsBuilder rb1, RowsBuilder rb2, RowsBuilder expected) {
         check(union(rb1, rb2), expected);
     }
 
@@ -239,13 +243,14 @@
         }
     }
 
-    private static Operator union(RowsBuilder rb1, RowsBuilder rb2) {
+    private Operator union(RowsBuilder rb1, RowsBuilder rb2) {
         return new UnionAll_Default(
                     new TestOperator(rb1),
                     rb1.rowType(),
                     new TestOperator(rb2),
                     rb2.rowType(),
-                    Types3Switch.ON
+                    Types3Switch.ON,
+                    openBoth()
             );
     }
 }

=== 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-13 19:25:25 +0000
+++ src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultIT.java	2013-07-18 20:38:26 +0000
@@ -70,30 +70,34 @@
         use(db);
     }
 
+    protected boolean openBoth() {
+        return false;
+    }
+
     // Test argument validation
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullInput1()
     {
-        unionAll(null, tRowType, groupScan_Default(groupTable), tRowType);
+        unionAll_Default(null, tRowType, groupScan_Default(groupTable), tRowType, false);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullInput1Type()
     {
-        unionAll(groupScan_Default(groupTable), null, groupScan_Default(groupTable), tRowType);
+        unionAll_Default(groupScan_Default(groupTable), null, groupScan_Default(groupTable), tRowType, false);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullInput2()
     {
-        unionAll(groupScan_Default(groupTable), tRowType, null, tRowType);
+        unionAll_Default(groupScan_Default(groupTable), tRowType, null, tRowType, false);
     }
 
     @Test(expected = IllegalArgumentException.class)
     public void testNullInput2Type()
     {
-        unionAll(groupScan_Default(groupTable), tRowType, groupScan_Default(groupTable), null);
+        unionAll_Default(groupScan_Default(groupTable), tRowType, groupScan_Default(groupTable), null, false);
     }
 
     // Test operator execution
@@ -102,17 +106,18 @@
     public void testBothEmpty()
     {
         Operator plan =
-            unionAll(
-                select_HKeyOrdered(
-                    groupScan_Default(groupTable),
-                    tRowType,
-                    ExpressionGenerators.literal(false)),
-                tRowType,
-                select_HKeyOrdered(
-                    groupScan_Default(groupTable),
-                    tRowType,
-                    ExpressionGenerators.literal(false)),
-                tRowType);
+            unionAll_Default(
+                select_HKeyOrdered(
+                    groupScan_Default(groupTable),
+                    tRowType,
+                    ExpressionGenerators.literal(false)),
+                tRowType,
+                select_HKeyOrdered(
+                    groupScan_Default(groupTable),
+                    tRowType,
+                    ExpressionGenerators.literal(false)),
+                tRowType,
+                openBoth());
         RowBase[] expected = new RowBase[]{};
         compareRows(expected, cursor(plan, queryContext, queryBindings));
     }
@@ -121,7 +126,7 @@
     public void testLeftEmpty()
     {
         Operator plan =
-            unionAll(
+            unionAll_Default(
                 select_HKeyOrdered(
                     groupScan_Default(groupTable),
                     tRowType,
@@ -134,7 +139,8 @@
                         ExpressionGenerators.field(tRowType, 1),
                         Comparison.EQ,
                         ExpressionGenerators.literal(9), castResolver())),
-                    tRowType);
+                tRowType,
+                openBoth());
         RowBase[] expected = new RowBase[]{
             row(tRowType, 1001L, 9L),
             row(tRowType, 1003L, 9L),
@@ -148,7 +154,7 @@
     public void testRightEmpty()
     {
         Operator plan =
-            unionAll(
+            unionAll_Default(
                 select_HKeyOrdered(
                     groupScan_Default(groupTable),
                     tRowType,
@@ -161,7 +167,8 @@
                     groupScan_Default(groupTable),
                     tRowType,
                     ExpressionGenerators.literal(false)),
-                    tRowType);
+                tRowType, 
+                openBoth());
         RowBase[] expected = new RowBase[]{
             row(tRowType, 1000L, 8L),
             row(tRowType, 1002L, 8L),
@@ -175,7 +182,7 @@
     public void testBothNonEmpty()
     {
         Operator plan =
-            unionAll(
+            unionAll_Default(
                 select_HKeyOrdered(
                     groupScan_Default(groupTable),
                     tRowType,
@@ -191,7 +198,8 @@
                         ExpressionGenerators.field(tRowType, 1),
                         Comparison.EQ,
                         ExpressionGenerators.literal(9), castResolver())),
-                    tRowType);
+                tRowType, 
+                openBoth());
         RowBase[] expected = new RowBase[]{
             row(tRowType, 1000L, 8L),
             row(tRowType, 1002L, 8L),
@@ -209,7 +217,7 @@
     public void testCursor()
     {
         Operator plan =
-            unionAll(
+            unionAll_Default(
                 select_HKeyOrdered(
                     groupScan_Default(groupTable),
                     tRowType,
@@ -225,7 +233,8 @@
                         ExpressionGenerators.field(tRowType, 1),
                         Comparison.EQ,
                         ExpressionGenerators.literal(9), castResolver())),
-                tRowType);
+                tRowType, 
+                openBoth());
         CursorLifecycleTestCase testCase = new CursorLifecycleTestCase()
         {
             @Override
@@ -261,7 +270,7 @@
                 limit_Default(
                     groupScan_Default(groupTable),
                     2),
-                unionAll(
+                unionAll_Default(
                     indexScan_Default(txIndexRowType,
                                       xEQ8Range,
                                       ordering),
@@ -269,7 +278,7 @@
                     indexScan_Default(txIndexRowType,
                                       xEQ9Range,
                                       ordering),
-                    txIndexRowType),
+                    txIndexRowType, openBoth()),
                 0, pipelineMap(), 1);
         String[] expected = new String[]{
             hKey(1000),

=== added file 'src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultPipelineIT.java'
--- src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultPipelineIT.java	1970-01-01 00:00:00 +0000
+++ src/test/java/com/akiban/server/test/it/qp/UnionAll_DefaultPipelineIT.java	2013-07-18 20:38:26 +0000
@@ -0,0 +1,31 @@
+/**
+ * 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 UnionAll_DefaultPipelineIT extends UnionAll_DefaultIT
+{
+    @Override
+    protected boolean pipelineMap() {
+        return true;
+    }
+
+    @Override
+    protected boolean openBoth() {
+        return true;
+    }
+}