pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From o...@apache.org
Subject svn commit: r741763 - in /hadoop/pig/trunk: ./ src/org/apache/pig/ src/org/apache/pig/impl/logicalLayer/optimizer/ test/org/apache/pig/test/ test/org/apache/pig/test/data/DotFiles/
Date Fri, 06 Feb 2009 22:35:00 GMT
Author: olga
Date: Fri Feb  6 22:34:59 2009
New Revision: 741763

URL: http://svn.apache.org/viewvc?rev=741763&view=rev
Log:
PIG-637
: limit after order by is broken in the local mode

Added:
    hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan8.dot
    hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan9.dot
Modified:
    hadoop/pig/trunk/CHANGES.txt
    hadoop/pig/trunk/src/org/apache/pig/PigServer.java
    hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
    hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
    hadoop/pig/trunk/test/org/apache/pig/test/TestLogicalOptimizer.java

Modified: hadoop/pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=741763&r1=741762&r2=741763&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Fri Feb  6 22:34:59 2009
@@ -406,3 +406,6 @@
     storing strings > 65536 bytes (in UTF8 form) using BinStorage() (sms)
 
     PIG-642: Limit after FRJ causes problems (daijy)
+
+    PIG-637: Limit broken after order by in the local mode (shubhamc via
+    olgan)

Modified: hadoop/pig/trunk/src/org/apache/pig/PigServer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/PigServer.java?rev=741763&r1=741762&r2=741763&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/PigServer.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/PigServer.java Fri Feb  6 22:34:59 2009
@@ -758,7 +758,8 @@
 
         // optimize
         if (optimize) {
-            LogicalOptimizer optimizer = new LogicalOptimizer(lpClone);
+            //LogicalOptimizer optimizer = new LogicalOptimizer(lpClone);
+            LogicalOptimizer optimizer = new LogicalOptimizer(lpClone, pigContext.getExecType());
             optimizer.optimize();
         }
 

Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java?rev=741763&r1=741762&r2=741763&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
(original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/LogicalOptimizer.java
Fri Feb  6 22:34:59 2009
@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.apache.pig.ExecType;
 import org.apache.pig.impl.logicalLayer.LogicalOperator;
 import org.apache.pig.impl.logicalLayer.LogicalPlan;
 import org.apache.pig.impl.plan.optimizer.*;
@@ -29,21 +30,29 @@
 /**
  * An optimizer for logical plans.
  */
-public class LogicalOptimizer extends PlanOptimizer<LogicalOperator, LogicalPlan> {
-    
+public class LogicalOptimizer extends
+        PlanOptimizer<LogicalOperator, LogicalPlan> {
+
     public static final String LOLOAD_CLASSNAME = "org.apache.pig.impl.logicalLayer.LOLoad";
     public static final String LOSTREAM_CLASSNAME = "org.apache.pig.impl.logicalLayer.LOStream";
 
     public LogicalOptimizer(LogicalPlan plan) {
+        this(plan, ExecType.MAPREDUCE);
+    }
+
+    public LogicalOptimizer(LogicalPlan plan, ExecType mode) {
         super(plan);
+        runOptimizations(plan, mode);
+    }
 
+    private void runOptimizations(LogicalPlan plan, ExecType mode) {
         // List of rules for the logical optimizer
-        
+
         // This one has to be first, as the type cast inserter expects the
         // load to only have one output.
         // Find any places in the plan that have an implicit split and make
-        // it explicit.  Since the RuleMatcher doesn't handle trees properly,
-        // we cheat and say that we match any node.  Then we'll do the actual
+        // it explicit. Since the RuleMatcher doesn't handle trees properly,
+        // we cheat and say that we match any node. Then we'll do the actual
         // test in the transformers check method.
         List<String> nodes = new ArrayList<String>(1);
         Map<Integer, Integer> edges = new HashMap<Integer, Integer>();
@@ -51,8 +60,8 @@
         nodes.add("any");
         required.add(true);
         mRules.add(new Rule<LogicalOperator, LogicalPlan>(nodes, edges,
-            required, new ImplicitSplitInserter(plan)));
-        
+                required, new ImplicitSplitInserter(plan)));
+
         // Add type casting to plans where the schema has been declared (by
         // user, data, or data catalog).
         nodes = new ArrayList<String>(1);
@@ -60,9 +69,9 @@
         edges = new HashMap<Integer, Integer>();
         required = new ArrayList<Boolean>(1);
         required.add(true);
-        mRules.add(new Rule<LogicalOperator, LogicalPlan>(nodes, edges, required,
-            new TypeCastInserter(plan, LOLOAD_CLASSNAME)));
-        
+        mRules.add(new Rule<LogicalOperator, LogicalPlan>(nodes, edges,
+                required, new TypeCastInserter(plan, LOLOAD_CLASSNAME)));
+
         // Add type casting to plans where the schema has been declared by
         // user in a statement with stream operator.
         nodes = new ArrayList<String>(1);
@@ -70,28 +79,24 @@
         edges = new HashMap<Integer, Integer>();
         required = new ArrayList<Boolean>(1);
         required.add(true);
-        mRules.add(new Rule(nodes, edges, required,
-            new TypeCastInserter(plan, LOSTREAM_CLASSNAME)));
-        
+        mRules.add(new Rule(nodes, edges, required, new TypeCastInserter(plan,
+                LOSTREAM_CLASSNAME)));
+
         // Optimize when LOAD precedes STREAM and the loader class
         // is the same as the serializer for the STREAM.
         // Similarly optimize when STREAM is followed by store and the
         // deserializer class is same as the Storage class.
-        mRules.add(new Rule(nodes, edges, required,
-            new StreamOptimizer(plan, LOSTREAM_CLASSNAME)));
-        
+        mRules.add(new Rule(nodes, edges, required, new StreamOptimizer(plan,
+                LOSTREAM_CLASSNAME)));
+
         // Push up limit where ever possible.
         nodes = new ArrayList<String>(1);
         edges = new HashMap<Integer, Integer>();
         required = new ArrayList<Boolean>(1);
         nodes.add("org.apache.pig.impl.logicalLayer.LOLimit");
         required.add(true);
-        mRules.add(new Rule<LogicalOperator, LogicalPlan>(nodes, edges, required,
-            new OpLimitOptimizer(plan)));
-        
-        
+        mRules.add(new Rule<LogicalOperator, LogicalPlan>(nodes, edges,
+                required, new OpLimitOptimizer(plan, mode)));
     }
 
 }
-
-

Modified: hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java?rev=741763&r1=741762&r2=741763&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
(original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/logicalLayer/optimizer/OpLimitOptimizer.java
Fri Feb  6 22:34:59 2009
@@ -23,6 +23,7 @@
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.pig.ExecType;
 import org.apache.pig.PigException;
 import org.apache.pig.impl.logicalLayer.LOCogroup;
 import org.apache.pig.impl.logicalLayer.LOCross;
@@ -53,11 +54,17 @@
 public class OpLimitOptimizer extends LogicalTransformer {
 
     private static final Log log = LogFactory.getLog(OpLimitOptimizer.class);
+    private ExecType mode = ExecType.MAPREDUCE;
 
     public OpLimitOptimizer(LogicalPlan plan) {
         super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
     }
 
+    public OpLimitOptimizer(LogicalPlan plan, ExecType mode) {
+        super(plan, new DepthFirstWalker<LogicalOperator, LogicalPlan>(plan));
+        this.mode = mode;
+    }
+
     @Override
     public boolean check(List<LogicalOperator> nodes) throws OptimizerException {
         if((nodes == null) || (nodes.size() <= 0)) {
@@ -193,18 +200,23 @@
             // Limit can be merged into LOSort, result a "limited sort"
             else if (predecessor instanceof LOSort)
             {
-            	LOSort sort = (LOSort)predecessor;
-            	if (sort.getLimit()==-1)
-            		sort.setLimit(limit.getLimit());
-            	else
-            	    sort.setLimit(sort.getLimit()<limit.getLimit()?sort.getLimit():limit.getLimit());
-            	try {
-            		removeFromChain(limit, null);
-            	} catch (Exception e) {
-            	    int errCode = 2012;
-            	    String msg = "Can not remove LOLimit after LOSort";
-            		throw new OptimizerException(msg, errCode, PigException.BUG, e);
-            	}
+                if(mode == ExecType.LOCAL) {
+                    //We don't need this optimisation to happen in the local mode.
+                    //so we do nothing here.
+                } else {
+                    LOSort sort = (LOSort)predecessor;
+                    if (sort.getLimit()==-1)
+                        sort.setLimit(limit.getLimit());
+                    else
+                        sort.setLimit(sort.getLimit()<limit.getLimit()?sort.getLimit():limit.getLimit());
+                    try {
+                        removeFromChain(limit, null);
+                    } catch (Exception e) {
+                        int errCode = 2012;
+                        String msg = "Can not remove LOLimit after LOSort";
+                        throw new OptimizerException(msg, errCode, PigException.BUG, e);
+                    }
+                }
             }
             // Limit is merged into another LOLimit
             else if (predecessor instanceof LOLimit)

Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestLogicalOptimizer.java
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestLogicalOptimizer.java?rev=741763&r1=741762&r2=741763&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestLogicalOptimizer.java (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestLogicalOptimizer.java Fri Feb  6 22:34:59
2009
@@ -17,8 +17,11 @@
  */
 package org.apache.pig.test;
 
+import java.io.File;
 import java.io.FileInputStream;
+import java.io.PrintWriter;
 
+import org.apache.pig.ExecType;
 import org.apache.pig.impl.logicalLayer.*;
 import org.apache.pig.impl.logicalLayer.optimizer.*;
 import org.apache.pig.test.utils.LogicalPlanTester;
@@ -58,6 +61,11 @@
         optimizer.optimize();
     }
     
+    public static void optimizePlan(LogicalPlan plan, ExecType mode) throws OptimizerException
{
+        LogicalOptimizer optimizer = new LogicalOptimizer(plan, mode);
+        optimizer.optimize();
+    }
+    
     void compareWithGoldenFile(LogicalPlan plan, String filename) throws Exception
     {
         FileInputStream fis = new FileInputStream(filename);
@@ -153,6 +161,28 @@
         optimizePlan(plan);
         compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan7.dot");
     }
+    
+    @Test
+    //Limit in the local mode, need to make sure limit stays after a sort
+    public void testOPLimit8Optimizer() throws Exception {
+        planTester.buildPlan("A = load 'myfile';");
+        planTester.buildPlan("B = order A by $0;");
+        LogicalPlan plan = planTester.buildPlan("C = limit B 10;");
+        optimizePlan(plan, ExecType.LOCAL);
+        compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan8.dot");
+        
+    }
+    
+    @Test
+    //Limit in the local mode, need to make sure limit stays after a sort
+    public void testOPLimit9Optimizer() throws Exception {
+        planTester.buildPlan("A = load 'myfile';");
+        planTester.buildPlan("B = order A by $0;");
+        LogicalPlan plan = planTester.buildPlan("C = limit B 10;");
+        optimizePlan(plan);
+        compareWithGoldenFile(plan, FILE_BASE_LOCATION + "optlimitplan9.dot");
+        
+    }
 
     @Test
     //Test to ensure that the right exception is thrown

Added: hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan8.dot
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan8.dot?rev=741763&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan8.dot (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan8.dot Fri Feb  6 22:34:59
2009
@@ -0,0 +1,6 @@
+digraph graph1 {
+    LOLoad55 -> LOSort57;
+    LOSort57 [limit="-1"];
+    LOSort57 -> LOLimit58;
+    LOLimit58 [limit="10"];
+}

Added: hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan9.dot
URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan9.dot?rev=741763&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan9.dot (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/data/DotFiles/optlimitplan9.dot Fri Feb  6 22:34:59
2009
@@ -0,0 +1,4 @@
+digraph graph1 {
+    LOLoad59 -> LOSort61;
+    LOSort61 [limit="10"];
+}



Mime
View raw message