Author: daijy
Date: Fri Dec 25 00:19:17 2009
New Revision: 893826

URL: http://svn.apache.org/viewvc?rev=893826&view=rev
Log:
PIG-761: ERROR 2086 on simple JOIN

Modified:
    hadoop/pig/branches/branch-0.6/CHANGES.txt
    
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
    
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceOper.java
    
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java
    
hadoop/pig/branches/branch-0.6/test/org/apache/pig/test/TestEvalPipeline2.java

Modified: hadoop/pig/branches/branch-0.6/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/CHANGES.txt?rev=893826&r1=893825&r2=893826&view=diff
==============================================================================
--- hadoop/pig/branches/branch-0.6/CHANGES.txt (original)
+++ hadoop/pig/branches/branch-0.6/CHANGES.txt Fri Dec 25 00:19:17 2009
@@ -253,6 +253,8 @@
 
 PIG-1165: Signature of loader does not set correctly for order by (daijy)
 
+PIG-761: ERROR 2086 on simple JOIN (daijy)
+
 Release 0.5.0
 
 INCOMPATIBLE CHANGES

Modified: 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java?rev=893826&r1=893825&r2=893826&view=diff
==============================================================================
--- 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
 (original)
+++ 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
 Fri Dec 25 00:19:17 2009
@@ -2419,6 +2419,7 @@
                     throw new MRCompilerException(msg, errCode, 
PigException.BUG);
                 }
                 FileSpec oldSpec = ((POStore)mpLeaf).getSFile();
+                boolean oldIsTmpStore = ((POStore)mpLeaf).isTmpStore();
                 
                 FileSpec fSpec = getTempFileSpec();
                 ((POStore)mpLeaf).setSFile(fSpec);
@@ -2440,9 +2441,10 @@
                 limitAdjustMROp.reducePlan.addAsLeaf(pLimit2);
                 POStore st = getStore();
                 st.setSFile(oldSpec);
-                st.setIsTmpStore(false);
+                st.setIsTmpStore(oldIsTmpStore);
                 limitAdjustMROp.reducePlan.addAsLeaf(st);
                 limitAdjustMROp.requestedParallelism = 1;
+                limitAdjustMROp.setLimitOnly(true);
                 // If the operator we're following has global sort set, we
                 // need to indicate that this is a limit after a sort.
                 // This will assure that we get the right sort comparator

Modified: 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceOper.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceOper.java?rev=893826&r1=893825&r2=893826&view=diff
==============================================================================
--- 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceOper.java
 (original)
+++ 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceOper.java
 Fri Dec 25 00:19:17 2009
@@ -83,6 +83,10 @@
 
     // Indicates if this is a limit after a sort
     boolean limitAfterSort = false;
+    
+    // Indicate if the entire purpose for this map reduce job is doing limit, 
does not change
+    // anything else. This is to help POPackageAnnotator to find the right 
POPackage to annotate
+    boolean limitOnly = false;
 
     // If true, putting an identity combine in this
     // mapreduce job will speed things up.
@@ -284,6 +288,14 @@
     public void setLimitAfterSort(boolean las) {
         limitAfterSort = las;
     }
+    
+    public boolean isLimitOnly() {
+        return limitOnly;
+    }
+    
+    public void setLimitOnly(boolean limitOnly) {
+        this.limitOnly = limitOnly;
+    }
 
     public boolean needsDistinctCombiner() { 
         return needsDistinctCombiner;

Modified: 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java?rev=893826&r1=893825&r2=893826&view=diff
==============================================================================
--- 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java
 (original)
+++ 
hadoop/pig/branches/branch-0.6/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/plans/POPackageAnnotator.java
 Fri Dec 25 00:19:17 2009
@@ -105,6 +105,8 @@
             List<MapReduceOper> preds = this.mPlan.getPredecessors(mr);
             for (Iterator<MapReduceOper> it = preds.iterator(); it.hasNext();) 
{
                 MapReduceOper mrOper = it.next();
+                if (mrOper.isLimitOnly())
+                    mrOper = this.mPlan.getPredecessors(mrOper).get(0);
                 lrFound += patchPackage(mrOper.reducePlan, pkg);
                 if(lrFound == pkg.getNumInps()) {
                     break;

Modified: 
hadoop/pig/branches/branch-0.6/test/org/apache/pig/test/TestEvalPipeline2.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/branches/branch-0.6/test/org/apache/pig/test/TestEvalPipeline2.java?rev=893826&r1=893825&r2=893826&view=diff
==============================================================================
--- 
hadoop/pig/branches/branch-0.6/test/org/apache/pig/test/TestEvalPipeline2.java 
(original)
+++ 
hadoop/pig/branches/branch-0.6/test/org/apache/pig/test/TestEvalPipeline2.java 
Fri Dec 25 00:19:17 2009
@@ -418,4 +418,39 @@
         
         assertTrue(iter.hasNext()==false);
     }
+    
+    // See PIG-761
+    @Test
+    public void testLimitPOPackageAnnotator() throws Exception{
+        File tmpFile1 = File.createTempFile("test1", "txt");
+        PrintStream ps1 = new PrintStream(new FileOutputStream(tmpFile1));
+        ps1.println("1\t2\t3");
+        ps1.println("2\t5\t2");
+        ps1.close();
+        
+        File tmpFile2 = File.createTempFile("test2", "txt");
+        PrintStream ps2 = new PrintStream(new FileOutputStream(tmpFile2));
+        ps2.println("1\t1");
+        ps2.println("2\t2");
+        ps2.close();
+
+        pigServer.registerQuery("A = LOAD '" + 
Util.generateURI(tmpFile1.toString()) + "' AS (a0, a1, a2);");
+        pigServer.registerQuery("B = LOAD '" + 
Util.generateURI(tmpFile2.toString()) + "' AS (b0, b1);");
+        pigServer.registerQuery("C = LIMIT B 100;");
+        pigServer.registerQuery("D = COGROUP C BY b0, A BY a0 PARALLEL 2;");
+        Iterator<Tuple> iter = pigServer.openIterator("D");
+        
+        assertTrue(iter.hasNext());
+        Tuple t = iter.next();
+        
+        assertTrue(t.toString().equals("(1,{(1,1)},{(1,2,3)})"));
+        
+        assertTrue(iter.hasNext());
+        t = iter.next();
+        
+        assertTrue(t.toString().equals("(2,{(2,2)},{(2,5,2)})"));
+        
+        assertFalse(iter.hasNext());
+    }
+
 }


Reply via email to