You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2010/08/28 00:27:58 UTC
svn commit: r990289 - in /hadoop/pig/trunk: ./
src/org/apache/pig/newplan/logical/optimizer/
src/org/apache/pig/newplan/logical/relational/
src/org/apache/pig/newplan/logical/rules/ test/org/apache/pig/test/
Author: daijy
Date: Fri Aug 27 22:27:58 2010
New Revision: 990289
URL: http://svn.apache.org/viewvc?rev=990289&view=rev
Log:
PIG-1515: Migrate logical optimization rule: PushDownForeachFlatten
Added:
hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java
hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/PushDownForEachFlatten.java
hadoop/pig/trunk/test/org/apache/pig/test/TestNewPlanPushDownForeachFlatten.java
Modified:
hadoop/pig/trunk/CHANGES.txt
hadoop/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java
hadoop/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOGenerate.java
Modified: hadoop/pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=990289&r1=990288&r2=990289&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Fri Aug 27 22:27:58 2010
@@ -26,6 +26,8 @@ PIG-1249: Safe-guards against misconfigu
IMPROVEMENTS
+PIG-1515: Migrate logical optimization rule: PushDownForeachFlatten (xuefuz via daijy)
+
PIG-1321: Logical Optimizer: Merge cascading foreach (xuefuz via daijy)
PIG-1483: [piggybank] Add HadoopJobHistoryLoader to the piggybank (rding)
Modified: hadoop/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java?rev=990289&r1=990288&r2=990289&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java Fri Aug 27 22:27:58 2010
@@ -38,6 +38,7 @@ import org.apache.pig.newplan.optimizer.
import org.apache.pig.newplan.logical.rules.LimitOptimizer;
import org.apache.pig.newplan.logical.rules.PartitionFilterOptimizer;
+import org.apache.pig.newplan.logical.rules.PushDownForEachFlatten;
public class LogicalPlanOptimizer extends PlanOptimizer {
private Set<String> mRulesOff = null;
@@ -121,6 +122,14 @@ public class LogicalPlanOptimizer extend
if (!s.isEmpty())
ls.add(s);
+ // PushDownForEachFlatten set
+ s = new HashSet<Rule>();
+ // Add the PushDownForEachFlatten
+ r = new PushDownForEachFlatten("PushDownForEachFlatten");
+ checkAndAddRule(s, r);
+ if (!s.isEmpty())
+ ls.add(s);
+
// Prune Set
// This set is used for pruning columns and maps
s = new HashSet<Rule>();
Modified: hadoop/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOGenerate.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOGenerate.java?rev=990289&r1=990288&r2=990289&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOGenerate.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/newplan/logical/relational/LOGenerate.java Fri Aug 27 22:27:58 2010
@@ -19,6 +19,7 @@ package org.apache.pig.newplan.logical.r
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import org.apache.pig.data.DataType;
import org.apache.pig.impl.logicalLayer.FrontendException;
@@ -147,4 +148,31 @@ public class LOGenerate extends LogicalR
}
((LogicalRelationalNodesVisitor)v).visit(this);
}
+
+ @Override
+ public String toString() {
+ StringBuilder msg = new StringBuilder();
+
+ if (alias!=null) {
+ msg.append(alias + ": ");
+ }
+ msg.append("(Name: " + name + "[");
+ for (int i=0;i<flattenFlags.length;i++) {
+ msg.append(flattenFlags[i]);
+ if (i!=flattenFlags.length-1)
+ msg.append(",");
+ }
+ msg.append("] Schema: ");
+ if (schema!=null)
+ msg.append(schema);
+ else
+ msg.append("null");
+ msg.append(")");
+ if (annotations!=null) {
+ for (Map.Entry<String, Object> entry : annotations.entrySet()) {
+ msg.append(entry);
+ }
+ }
+ return msg.toString();
+ }
}
Added: hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java?rev=990289&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java Fri Aug 27 22:27:58 2010
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.newplan.logical.rules;
+
+import java.util.Iterator;
+
+import org.apache.pig.newplan.Operator;
+import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LogicalPlan;
+
+public class OptimizerUtils {
+ /**
+ * Find generate op from the foreach operator.
+ * @param foreach the LOForEach instance
+ * @return LOGenerate instance
+ */
+ public static LOGenerate findGenerate(LOForEach foreach) {
+ LogicalPlan inner = foreach.getInnerPlan();
+ return (LOGenerate)inner.getSinks().get(0);
+ }
+
+ /**
+ * Check if a given LOGenerate operator has any flatten fields.
+ * @param gen the given LOGenerate instance
+ * @return true if LOGenerate instance contains flatten fields, false otherwise
+ */
+ public static boolean hasFlatten(LOGenerate gen) {
+ boolean hasFlatten = false;
+ boolean[] flattenFlags = gen.getFlattenFlags();
+ if( flattenFlags != null ) {
+ for( boolean flatten : flattenFlags ) {
+ if( flatten ) {
+ hasFlatten = true;
+ break;
+ }
+ }
+ }
+ return hasFlatten;
+ }
+
+ /**
+ * Helper method to find if a given LOForEach instance contains any flatten fields.
+ * @param foreach foreach the LOForEach instance
+ * @return true if LOForEach instance contains flatten fields, false otherwise
+ */
+ public static boolean hasFlatten(LOForEach foreach) {
+ LOGenerate gen = findGenerate( foreach );
+ return hasFlatten( gen );
+ }
+}
Added: hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/PushDownForEachFlatten.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/PushDownForEachFlatten.java?rev=990289&view=auto
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/PushDownForEachFlatten.java (added)
+++ hadoop/pig/trunk/src/org/apache/pig/newplan/logical/rules/PushDownForEachFlatten.java Fri Aug 27 22:27:58 2010
@@ -0,0 +1,294 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.newplan.logical.rules;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.pig.newplan.logical.expression.LogicalExpression;
+import org.apache.pig.newplan.logical.expression.LogicalExpressionPlan;
+import org.apache.pig.newplan.logical.expression.ProjectExpression;
+import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LOInnerLoad;
+import org.apache.pig.newplan.logical.relational.LOSort;
+import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LOJoin;
+import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.newplan.logical.relational.LogicalRelationalOperator;
+import org.apache.pig.newplan.Operator;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.OperatorSubPlan;
+import org.apache.pig.newplan.optimizer.Rule;
+import org.apache.pig.newplan.optimizer.Transformer;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.util.Pair;
+
+public class PushDownForEachFlatten extends Rule {
+
+ public PushDownForEachFlatten(String name) {
+ super( name, false );
+ }
+
+ @Override
+ protected OperatorPlan buildPattern() {
+ LogicalPlan plan = new LogicalPlan();
+ LogicalRelationalOperator foreach = new LOForEach(plan);
+ plan.add( foreach );
+ return plan;
+ }
+
+ @Override
+ public Transformer getNewTransformer() {
+ return new PushDownForEachFlattenTransformer();
+ }
+
+ class PushDownForEachFlattenTransformer extends Transformer {
+ private OperatorSubPlan subPlan;
+
+ @Override
+ public boolean check(OperatorPlan matched) throws FrontendException {
+ // the foreach with flatten can be swapped with an order by
+ // as the order by will have lesser number of records to sort
+ // also the sort does not alter the records that are processed
+
+ // the foreach with flatten can be pushed down a cross or a join
+ // for the same reason. In this case the foreach has to be first
+ // unflattened and then a new foreach has to be inserted after
+ // the cross or join. In both cross and foreach the actual columns
+ // from the foreach are not altered but positions might be changed
+
+ // in the case of union the column is transformed and as a result
+ // the foreach flatten cannot be pushed down
+
+ // for distinct the output before flattening and the output
+ // after flattening might be different. For example, consider
+ // {(1), (1)}. Distinct of this bag is still {(1), (1)}.
+ // distinct(flatten({(1), (1)})) is (1). However,
+ // flatten(distinct({(1), (1)})) is (1), (1)
+
+ // in both cases correctness is not affected
+
+ LOForEach foreach = (LOForEach)matched.getSources().get(0);
+ LOGenerate gen = OptimizerUtils.findGenerate( foreach );
+ if( !OptimizerUtils.hasFlatten( gen ) )
+ return false;
+
+ List<Operator> succs = currentPlan.getSuccessors( foreach );
+ if( succs == null || succs.size() != 1 )
+ return false;
+
+ List<Long> uids = getNonFlattenFieldUids( gen );
+
+ Operator succ = succs.get( 0 );
+ if( !( succ instanceof LOSort || succ instanceof LOJoin || succ instanceof LOCross ) )
+ return false;
+
+ if( succ instanceof LOSort ) {
+ // Check if the expressions for the foreach generate are purely projection including flatten fields.
+ List<LogicalExpressionPlan> exprs = gen.getOutputPlans();
+ for( LogicalExpressionPlan expr : exprs ) {
+ if( !isPureProjection( expr ) )
+ return false;
+ }
+
+ // Check if flatten fields are required by the successor.
+ LOSort sort = (LOSort)succ;
+ List<LogicalExpressionPlan> exps = sort.getSortColPlans();
+ for( int i = 0; i < exps.size(); i++ ) {
+ LogicalExpressionPlan exp = exps.get( i );
+ ProjectExpression proj = (ProjectExpression)exp.getOperators().next();
+ if( !uids.contains( proj.getFieldSchema().uid ) )
+ return false;
+ }
+
+ return true;
+ } else {
+ List<Operator> preds = currentPlan.getPredecessors( succ );
+
+ // We do not optimize if peer is ForEach with flatten. This is
+ // a simplification, may change in the future.
+ for( Operator op : preds ) {
+ if( op == foreach )
+ continue;
+ else if( op instanceof LOForEach &&
+ OptimizerUtils.hasFlatten( OptimizerUtils.findGenerate( (LOForEach)op ) ) )
+ return false;
+ }
+
+ if( ( (LogicalRelationalOperator)succ ).getSchema() == null )
+ return false;
+
+ if( succ instanceof LOCross ) {
+ return true;
+ } else {
+ LOJoin join = (LOJoin)succ;
+ for( int i = 0; i < preds.size(); i++ ) {
+ Operator op = preds.get( i );
+ if( op == foreach ) {
+ Collection<LogicalExpressionPlan> exprs = join.getJoinPlan( i );
+ for( LogicalExpressionPlan expr : exprs ) {
+ List<ProjectExpression> projs = getProjectExpressions( expr );
+ for( ProjectExpression proj : projs ) {
+ if( !uids.contains( proj.getFieldSchema().uid ) ) {
+ return false;
+ }
+ }
+ }
+ break;
+ }
+ }
+ return true;
+ }
+ }
+ }
+
+ private List<ProjectExpression> getProjectExpressions(LogicalExpressionPlan expr) {
+ List<Operator> ops = expr.getSources();
+ List<ProjectExpression> projs = new ArrayList<ProjectExpression>( ops.size() );
+ for( Operator op : ops ) {
+ if( op instanceof ProjectExpression ) {
+ projs.add( (ProjectExpression)op );
+ }
+ }
+ return projs;
+ }
+
+ private List<Long> getNonFlattenFieldUids(LOGenerate gen) throws FrontendException {
+ List<Long> uids = new ArrayList<Long>();
+
+ List<LogicalExpressionPlan> exprs = gen.getOutputPlans();
+ for( int i = 0; i < exprs.size(); i++ ) {
+ LogicalExpressionPlan expr = exprs.get( i );
+ if( gen.getFlattenFlags()[i] )
+ continue;
+ LogicalExpression e = (LogicalExpression)expr.getSources().get( 0 );
+ uids.add( e.getFieldSchema().uid );
+ }
+
+ return uids;
+ }
+
+ /**
+ * Check if the given expression contains only a pure projection.
+ * For instance $0 is legal, f1 is legal, but 5 + $2 is not legal.
+ * (int)f1 is not legal either.
+ */
+ private boolean isPureProjection(LogicalExpressionPlan expr) {
+ if (expr.size()!=1)
+ return false;
+ if (!(expr.getSinks().get(0) instanceof ProjectExpression))
+ return false;
+ return true;
+ }
+
+ @Override
+ public OperatorPlan reportChanges() {
+ return subPlan;
+ }
+
+ @Override
+ public void transform(OperatorPlan matched) throws FrontendException {
+ subPlan = new OperatorSubPlan( currentPlan );
+
+ LOForEach foreach = (LOForEach)matched.getSources().get(0);
+ Operator next = currentPlan.getSuccessors( foreach ).get(0);
+ if( next instanceof LOSort ) {
+ Operator pred = currentPlan.getPredecessors( foreach ).get( 0 );
+ List<Operator> succs = currentPlan.getSuccessors( next );
+ currentPlan.disconnect( pred, foreach );
+ currentPlan.disconnect( foreach, next );
+ currentPlan.connect( pred, next );
+ currentPlan.connect( next, foreach );
+ if( succs != null ) {
+ for( Operator succ : succs ) {
+ currentPlan.disconnect( next, succ );
+ currentPlan.connect( foreach, succ );
+ }
+ }
+ } else if( next instanceof LOCross || next instanceof LOJoin ) {
+ List<Operator> preds = currentPlan.getPredecessors( next );
+ List<Integer> fieldsToBeFlattaned = new ArrayList<Integer>();
+ boolean[] flags = null;
+ int fieldCount = 0;
+ for( Operator op : preds ) {
+ if( op == foreach ) {
+ LOGenerate gen = OptimizerUtils.findGenerate( foreach );
+ flags = gen.getFlattenFlags();
+ for( int i = 0; i < flags.length; i++ ) {
+ if( flags[i] ) {
+ fieldsToBeFlattaned.add( fieldCount++ );
+ } else {
+ fieldCount++;
+ }
+ }
+ } else {
+ fieldCount += ( (LogicalRelationalOperator)op ).getSchema().size();
+ }
+ }
+
+ boolean[] flattenFlags = new boolean[fieldCount];
+ for( Integer i : fieldsToBeFlattaned ) {
+ flattenFlags[i] = true;
+ }
+
+ // Now create a new foreach after cross/join and insert it into the plan.
+ LOForEach newForeach = new LOForEach( currentPlan );
+ LogicalPlan innerPlan = new LogicalPlan();
+ List<LogicalExpressionPlan> exprs = new ArrayList<LogicalExpressionPlan>( fieldCount );
+ LOGenerate gen = new LOGenerate( innerPlan, exprs, flattenFlags );
+ innerPlan.add( gen );
+ newForeach.setInnerPlan( innerPlan );
+ for( int i = 0; i < fieldCount; i++ ) {
+ LogicalExpressionPlan expr = new LogicalExpressionPlan();
+ expr.add( new ProjectExpression( expr, i, -1, gen ) );
+ exprs.add( expr );
+
+ LOInnerLoad innerLoad = new LOInnerLoad(innerPlan, newForeach, i);
+ innerPlan.add(innerLoad);
+ innerPlan.connect(innerLoad, gen);
+ }
+
+ newForeach.setAlias(((LogicalRelationalOperator)next).getAlias());
+
+ currentPlan.add( newForeach );
+ Operator opAfterX = null;
+ List<Operator> succs = currentPlan.getSuccessors( next );
+ if( succs == null || succs.size() == 0 ) {
+ currentPlan.connect( next, newForeach );
+ } else {
+ opAfterX = succs.get( 0 );
+ Pair<Integer, Integer> pos = currentPlan.disconnect( next, opAfterX );
+ currentPlan.connect( next, pos.first, newForeach, pos.second );
+ currentPlan.connect( newForeach, opAfterX );
+ }
+
+ // Finally remove flatten flags from the original foreach and regenerate schemas for those impacted.
+ for( int i = 0; i < flags.length; i++ ) {
+ flags[i] = false;
+ }
+
+ subPlan.add(foreach);
+ subPlan.add(next);
+ subPlan.add(newForeach);
+ }
+ }
+ }
+}
Added: hadoop/pig/trunk/test/org/apache/pig/test/TestNewPlanPushDownForeachFlatten.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestNewPlanPushDownForeachFlatten.java?rev=990289&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestNewPlanPushDownForeachFlatten.java (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestNewPlanPushDownForeachFlatten.java Fri Aug 27 22:27:58 2010
@@ -0,0 +1,1065 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.pig.test;
+
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+
+import org.apache.pig.ExecType;
+import org.apache.pig.FilterFunc;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.impl.PigContext;
+import org.apache.pig.test.utils.Identity;
+import org.apache.pig.test.utils.LogicalPlanTester;
+import org.apache.pig.impl.plan.VisitorException;
+import org.apache.pig.newplan.Operator;
+import org.apache.pig.newplan.OperatorPlan;
+import org.apache.pig.newplan.logical.LogicalPlanMigrationVistor;
+import org.apache.pig.newplan.logical.relational.LOCross;
+import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LOJoin;
+import org.apache.pig.newplan.logical.relational.LOLimit;
+import org.apache.pig.newplan.logical.relational.LOLoad;
+import org.apache.pig.newplan.logical.relational.LOSort;
+import org.apache.pig.newplan.logical.relational.LogicalPlan;
+import org.apache.pig.newplan.logical.optimizer.LogicalPlanOptimizer;
+import org.apache.pig.newplan.optimizer.PlanOptimizer;
+import org.apache.pig.newplan.optimizer.Rule;
+import org.apache.pig.newplan.logical.rules.LoadTypeCastInserter;
+import org.apache.pig.newplan.logical.rules.OptimizerUtils;
+import org.apache.pig.newplan.logical.rules.PushDownForEachFlatten;
+import org.apache.pig.newplan.logical.rules.TypeCastInserter;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.Before;
+
+/**
+ * Test the logical optimizer.
+ */
+public class TestNewPlanPushDownForeachFlatten {
+ PigContext pc = new PigContext(ExecType.LOCAL, new Properties());
+ LogicalPlanTester planTester = new LogicalPlanTester(pc) ;
+
+ @Before
+ public void tearDown() {
+ planTester.reset();
+ }
+
+ /**
+ *
+ * A simple filter UDF for testing
+ *
+ */
+ static public class MyFilterFunc extends FilterFunc {
+
+ @Override
+ public Boolean exec(Tuple input) {
+ return false;
+ }
+ }
+
+ /**
+ * Old plan is empty, so is the optimized new plan.
+ */
+ @Test
+ public void testErrorEmptyInput() throws Exception {
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp =
+ new org.apache.pig.impl.logicalLayer.LogicalPlan();
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Assert.assertTrue( newLogicalPlan.getOperators().hasNext() == false );
+ }
+
+ /**
+ * No foreach in the plan, no effect.
+ */
+ @Test
+ public void testErrorNonForeachInput() throws Exception {
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ List<Operator> nexts = newLogicalPlan.getSuccessors( load );
+ Assert.assertTrue( nexts == null || nexts.size() == 0 );
+}
+
+ @Test
+ public void testForeachNoFlatten() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, $2;");
+ planTester.buildPlan("C = order B by $0, $1;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan( "D = store C into 'dummy';" );
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ @Test
+ public void testForeachNoSuccessors() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("B = foreach A generate flatten($1);");
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ }
+
+ @Test
+ public void testForeachStreaming() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate flatten($1);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = stream B through `" + "pc -l" + "`;");
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ }
+
+ @Test
+ public void testForeachDistinct() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate flatten($1);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = distinct B;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ }
+
+ @Test
+ public void testForeachForeach() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten(1);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = foreach B generate $0;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ foreach = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+
+ @Test
+ public void testForeachFilter() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = filter B by $1 < 18;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachSplitOutput() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("split B into C if $1 < 18, D if $1 >= 18;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachLimit() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("B = limit B 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachUnion() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("D = union B, C;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator load = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ load = loads.get( 0 );
+ else
+ load = loads.get( 1 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachCogroup() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("D = cogroup B by $0, C by $0;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator load = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ load = loads.get( 0 );
+ else
+ load = loads.get( 1 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachGroupBy() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = group B by $0;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)foreach ) );
+ }
+
+ @Test
+ public void testForeachSort() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = order B by $0, $1;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan( "D = store C into 'dummy';" );
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator sort = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ Operator foreach = newLogicalPlan.getSuccessors( sort ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ }
+
+ /**
+ * Non-pure-projection, not optimizable.
+ */
+ @Test
+ public void testForeachSortNegative1() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0 + 5, $1, flatten($2);");
+ planTester.buildPlan("C = order B by $0, $1;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan( "D = store C into 'dummy';" );
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+
+ /**
+ * If the flattened field is referenced in the sort condition, then no optimization can be done.
+ */
+ @Test
+ public void testForeachSortNegative2() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:tuple(x,y));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = order B by $0, $3;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan( "D = store C into 'dummy';" );
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator foreach1 = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( foreach1 instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach1 ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ @Test
+ public void testForeachFlattenAddedColumnSort() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten(1);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = order B by $0, $1;");
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ @Test
+ public void testForeachUDFSort() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate $0, $1, " + Identity.class.getName() + "($2) ;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = order B by $0, $1;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ @Test
+ public void testForeachCastSort() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa);");
+ planTester.buildPlan("B = foreach A generate (chararray)$0, $1, flatten($2);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = order B by $0, $1;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ @Test
+ public void testForeachCross() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference);");
+ planTester.buildPlan("D = cross B, C;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOCross );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ @Test
+ public void testForeachCross1() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("C = foreach B generate $0, $1, flatten($2);");
+ planTester.buildPlan("D = cross A, C;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOCross );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ // TODO
+ // The following test case testForeachCross2 has multiple foreach flatten
+ // A new rule should optimize this case
+ @Test
+ public void testForeachCross2() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = foreach C generate $0, $1, flatten($2);");
+ planTester.buildPlan("E = cross B, D;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("F = limit E 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ // No optimization about foreach flatten.
+ Assert.assertTrue( newLogicalPlan.getPredecessors( newLogicalPlan.getSinks().get( 0 ) ).get( 0 ) instanceof LOCross );
+ }
+
+ /**
+ * This actually is a valid case, even though the optimization may not provide any performance benefit. However, detecting
+ * such a case requires more coding. Thus, we allow optimization to go thru in this case.
+ */
+ @Test
+ public void testForeachFlattenAddedColumnCross() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten(1);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = cross B, C;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOCross );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * This is a valid, positive test case. Optimization should go thru.
+ */
+ @Test
+ public void testForeachUDFCross() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, flatten($1), " + Identity.class.getName() + "($2) ;");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = cross B, C;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOCross );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * Cast should NOT matter to cross. This is a valid positive test case.
+ */
+ @Test
+ public void testForeachCastCross() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, (int)$1, flatten( $2 );");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = cross B, C;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOCross );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ @Test
+ public void testForeachFRJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference);");
+ planTester.buildPlan("D = join B by $0, C by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ @Test
+ public void testForeachFRJoin1() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("C = foreach B generate $0, $1, flatten($2);");
+ planTester.buildPlan("D = join A by $0, C by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ // TODO
+ // The following test case testForeachFRJoin2 has multiple foreach flatten
+ // A new rule should optimize this case
+ @Test
+ public void testForeachFRJoin2() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = foreach C generate $0, $1, flatten($2);");
+ planTester.buildPlan("E = join B by $0, D by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("F = limit E 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ // No optimization about foreach flatten.
+ Assert.assertTrue( newLogicalPlan.getPredecessors( newLogicalPlan.getSinks().get( 0 ) ).get( 0 ) instanceof LOJoin );
+ }
+
+ /**
+ * Valid positive test case, even though the benefit from the optimization is questionable. However, putting in additinal check for
+ * this condition requires extra coding.
+ */
+ @Test
+ public void testForeachFlattenAddedColumnFRJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten(1);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * This is actually a valid, positive test case. UDF doesn't prevent optimization.
+ * @throws Exception
+ */
+ @Test
+ public void testForeachUDFFRJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, flatten($1), " + Identity.class.getName() + "($2) ;");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * This is actually a valid, positive test case. Cast doesn't prevent optimization.
+ * @throws Exception
+ */
+ @Test
+ public void testForeachCastFRJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, (int)$1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0 using \"replicated\";");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ @Test
+ public void testForeachInnerJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ @Test
+ public void testForeachInnerJoin1() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("C = foreach B generate $0, $1, flatten($2);");
+ planTester.buildPlan("D = join A by $0, C by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "B" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ // TODO
+ // The following test case testForeachInnerJoin2 has multiple foreach flatten
+ // A new rule should optimize this case
+ @Test
+ public void testForeachInnerJoin2() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = foreach C generate $0, $1, flatten($2);");
+ planTester.buildPlan("E = join B by $0, D by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("F = limit E 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ // No optimization about foreach flatten.
+ Assert.assertTrue( newLogicalPlan.getPredecessors( newLogicalPlan.getSinks().get( 0 ) ).get( 0 ) instanceof LOJoin );
+ }
+
+ /**
+ * This is actually a valid positive test case, even though the benefit of such optimization is questionable. However,
+ * checking for such condition requires additional coding effort.
+ */
+ @Test
+ public void testForeachFlattenAddedColumnInnerJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, $1, flatten(1);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ planTester.setPlan(lp);
+ planTester.setProjectionMap(lp);
+ planTester.rebuildSchema(lp);
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * UDF doesn't prevent optimization.
+ */
+ @Test
+ public void testForeachUDFInnerJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, flatten($1), " + Identity.class.getName() + "($2) ;");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ /**
+ * Cast doesn't prevent optimization.
+ */
+ @Test
+ public void testForeachCastInnerJoin() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (name, age, gpa:(letter_grade, point_score));");
+ planTester.buildPlan("B = foreach A generate $0, (int)$1, flatten($2);");
+ planTester.buildPlan("C = load 'anotherfile' as (name, age, preference:(course_name, instructor));");
+ planTester.buildPlan("D = join B by $0, C by $0;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ List<Operator> loads = newLogicalPlan.getSources();
+ Assert.assertTrue( loads.size() == 2 );
+ Assert.assertTrue( loads.get( 0 ) instanceof LOLoad );
+ Assert.assertTrue( loads.get( 1 ) instanceof LOLoad );
+ Operator op = null;
+ if( ((LOLoad)loads.get( 0 )).getAlias().equals( "A" ) )
+ op = loads.get( 0 );
+ else
+ op = loads.get( 1 );
+
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( !OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOJoin );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOForEach );
+ Assert.assertTrue( OptimizerUtils.hasFlatten( (LOForEach)op ) );
+ op = newLogicalPlan.getSuccessors( op ).get( 0 );
+ Assert.assertTrue( op instanceof LOLimit );
+ }
+
+ // See PIG-1172
+ @Test
+ public void testForeachJoinRequiredField() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (bg:bag{t:tuple(a0,a1)});");
+ planTester.buildPlan("B = FOREACH A generate flatten($0);");
+ planTester.buildPlan("C = load '3.txt' AS (c0, c1);");
+ planTester.buildPlan("D = JOIN B by a1, C by c1;");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("E = limit D 10;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ // No optimization about foreach flatten.
+ Assert.assertTrue( newLogicalPlan.getPredecessors( newLogicalPlan.getSinks().get( 0 ) ).get( 0 ) instanceof LOJoin );
+ }
+
+ // See PIG-1374
+ @Test
+ public void testForeachRequiredField() throws Exception {
+ planTester.buildPlan("A = load 'myfile' as (b{t(a0:chararray,a1:int)});");
+ planTester.buildPlan("B = foreach A generate flatten($0);");
+ org.apache.pig.impl.logicalLayer.LogicalPlan lp = planTester.buildPlan("C = order B by $1 desc;");
+
+ LogicalPlan newLogicalPlan = migrateAndOptimizePlan( lp );
+
+ Operator load = newLogicalPlan.getSources().get( 0 );
+ Assert.assertTrue( load instanceof LOLoad );
+ Operator foreach = newLogicalPlan.getSuccessors( load ).get( 0 );
+ Assert.assertTrue( foreach instanceof LOForEach );
+ Operator foreach1 = newLogicalPlan.getSuccessors( foreach ).get( 0 );
+ Assert.assertTrue( foreach1 instanceof LOForEach );
+ Operator sort = newLogicalPlan.getSuccessors( foreach1 ).get( 0 );
+ Assert.assertTrue( sort instanceof LOSort );
+ }
+
+ public class MyPlanOptimizer extends LogicalPlanOptimizer {
+ protected MyPlanOptimizer(OperatorPlan p, int iterations) {
+ super(p, iterations, new HashSet<String>());
+ }
+
+ protected List<Set<Rule>> buildRuleSets() {
+ List<Set<Rule>> ls = new ArrayList<Set<Rule>>();
+
+ Set<Rule> s = new HashSet<Rule>();
+ // add split filter rule
+ Rule r = new LoadTypeCastInserter( "TypeCastInserter" );
+ s.add(r);
+ ls.add(s);
+
+ s = new HashSet<Rule>();
+ r = new PushDownForEachFlatten( "PushDownForEachFlatten" );
+ s.add(r);
+ ls.add(s);
+
+ return ls;
+ }
+ }
+
+ private LogicalPlan migrateAndOptimizePlan(org.apache.pig.impl.logicalLayer.LogicalPlan plan) throws IOException {
+ LogicalPlan newLogicalPlan = migratePlan( plan );
+ PlanOptimizer optimizer = new MyPlanOptimizer( newLogicalPlan, 3 );
+ optimizer.optimize();
+ return newLogicalPlan;
+ }
+
+ private LogicalPlan migratePlan(org.apache.pig.impl.logicalLayer.LogicalPlan lp) throws VisitorException{
+ LogicalPlanMigrationVistor visitor = new LogicalPlanMigrationVistor(lp);
+ visitor.visit();
+ org.apache.pig.newplan.logical.relational.LogicalPlan newPlan = visitor.getNewLogicalPlan();
+ return newPlan;
+ }
+}
+