Author: sms
Date: Sat Feb  7 06:43:42 2009
New Revision: 741847

URL: http://svn.apache.org/viewvc?rev=741847&view=rev
Log:
PIG-553: EvalFunc.finish() not getting called (shravanmn via sms)

Added:
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/UDFFinishVisitor.java
    hadoop/pig/trunk/test/org/apache/pig/test/TestFinish.java
Modified:
    hadoop/pig/trunk/CHANGES.txt
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
    
hadoop/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPigLauncher.java

Modified: hadoop/pig/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=741847&r1=741846&r2=741847&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Sat Feb  7 06:43:42 2009
@@ -409,3 +409,5 @@
 
     PIG-637: Limit broken after order by in the local mode (shubhamc via
     olgan)
+
+    PIG-553: EvalFunc.finish() not getting called (shravanmn via sms)

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java?rev=741847&r1=741846&r2=741847&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapBase.java
 Sat Feb  7 06:43:42 2009
@@ -35,7 +35,9 @@
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.io.PigNullableWritable;
 import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.plan.DependencyOrderWalker;
 import org.apache.pig.impl.plan.OperatorKey;
+import org.apache.pig.impl.plan.VisitorException;
 import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
 import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
@@ -99,6 +101,15 @@
                  throw ioe;
             }
         }
+        
+        //Calling EvalFunc.finish()
+        UDFFinishVisitor finisher = new UDFFinishVisitor(mp, new 
DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(mp));
+        try {
+            finisher.visit();
+        } catch (VisitorException e) {
+            throw new IOException("Error trying to finish UDFs",e);
+        }
+        
         mp = null;
         
     }

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java?rev=741847&r1=741846&r2=741847&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
 Sat Feb  7 06:43:42 2009
@@ -47,6 +47,8 @@
 import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.io.PigNullableWritable;
 import org.apache.pig.impl.io.NullableTuple;
+import org.apache.pig.impl.plan.DependencyOrderWalker;
+import org.apache.pig.impl.plan.VisitorException;
 import org.apache.pig.impl.util.ObjectSerializer;
 import org.apache.pig.impl.util.SpillableMemoryManager;
 import org.apache.pig.impl.util.WrappedIOException;
@@ -347,6 +349,14 @@
                      throw ioe;
                 }
             }
+            
+            //Calling EvalFunc.finish()
+            UDFFinishVisitor finisher = new UDFFinishVisitor(rp, new 
DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(rp));
+            try {
+                finisher.visit();
+            } catch (VisitorException e) {
+                throw new IOException("Error trying to finish UDFs",e);
+            }
         }
     }
     

Added: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/UDFFinishVisitor.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/UDFFinishVisitor.java?rev=741847&view=auto
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/UDFFinishVisitor.java
 (added)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/UDFFinishVisitor.java
 Sat Feb  7 06:43:42 2009
@@ -0,0 +1,39 @@
+/*
+ * 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.backend.hadoop.executionengine.mapReduceLayer;
+
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
+import org.apache.pig.impl.plan.PlanWalker;
+import org.apache.pig.impl.plan.VisitorException;
+
+public class UDFFinishVisitor extends PhyPlanVisitor {
+
+    public UDFFinishVisitor(PhysicalPlan plan, PlanWalker<PhysicalOperator, 
PhysicalPlan> walker) {
+        super(plan, walker);
+    }
+
+    @Override
+    public void visitUserFunc(POUserFunc userFunc) throws VisitorException {
+        super.visitUserFunc(userFunc);
+        userFunc.finish();
+    }
+
+}
\ No newline at end of file

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPigLauncher.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPigLauncher.java?rev=741847&r1=741846&r2=741847&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPigLauncher.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/local/executionengine/LocalPigLauncher.java
 Sat Feb  7 06:43:42 2009
@@ -27,13 +27,18 @@
 import org.apache.pig.backend.executionengine.ExecException;
 import 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.JobCreationException;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.Launcher;
+import 
org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.UDFFinishVisitor;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
 import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.expressionOperators.POUserFunc;
+import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
 import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
 import 
org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
 import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.plan.DependencyOrderWalker;
 import org.apache.pig.impl.plan.PlanException;
+import org.apache.pig.impl.plan.PlanWalker;
 import org.apache.pig.impl.plan.VisitorException;
 
 public class LocalPigLauncher extends Launcher {
@@ -63,7 +68,10 @@
             if (res.returnStatus != POStatus.STATUS_EOP)
                 failedJobs++;
         }
-
+        
+        UDFFinishVisitor finisher = new UDFFinishVisitor(php, new 
DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(php));
+        finisher.visit();
+        
         if (failedJobs == 0) {
             log.info("100% complete!");
             log.info("Success!!");
@@ -75,5 +83,7 @@
         return false;
 
     }
+    
+    
 
 }

Added: hadoop/pig/trunk/test/org/apache/pig/test/TestFinish.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestFinish.java?rev=741847&view=auto
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestFinish.java (added)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestFinish.java Sat Feb  7 
06:43:42 2009
@@ -0,0 +1,138 @@
+/*
+ * 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.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.util.Iterator;
+import java.util.Random;
+
+import junit.framework.TestCase;
+
+import org.apache.pig.EvalFunc;
+import org.apache.pig.ExecType;
+import org.apache.pig.PigServer;
+import org.apache.pig.builtin.PigStorage;
+import org.apache.pig.data.BagFactory;
+import org.apache.pig.data.DataBag;
+import org.apache.pig.data.Tuple;
+import org.apache.pig.data.TupleFactory;
+import org.apache.pig.impl.io.FileLocalizer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestFinish extends TestCase {
+
+    private PigServer pigServer;
+
+    TupleFactory mTf = TupleFactory.getInstance();
+    BagFactory mBf = BagFactory.getInstance();
+    File f1;
+    
+    public static int gCount = 0;
+    
+    static public class MyEvalFunction extends EvalFunc<Tuple>{
+        int count = 0;
+        
+        @Override
+        public Tuple exec(Tuple input) throws IOException {
+            ++count;
+            return input;
+        }
+
+        @Override
+        public void finish() {
+            gCount = count;
+        }
+    }
+    
+    @Before
+    @Override
+    public void setUp() throws Exception{
+        pigServer = new PigServer(ExecType.MAPREDUCE);
+        f1 = File.createTempFile("test", "txt");
+        PrintStream ps = new PrintStream(new FileOutputStream(f1));
+        for(int i = 0; i < 3; i++) {
+            ps.println('a'+i + ":1");
+        }
+        ps.close();
+    }
+    
+    @Test
+    public void testFinishInMapMR() throws Exception{
+        gCount = 0;
+        pigServer.registerQuery("a = load '" + Util.generateURI(f1.toString()) 
+ "' using " + PigStorage.class.getName() + "(':');");
+        pigServer.registerQuery("b = foreach a generate " + 
MyEvalFunction.class.getName() + "(*);");
+        Iterator<Tuple> iter = pigServer.openIterator("b");
+        int count = 0;
+        while(iter.hasNext()){
+            ++count;
+            iter.next();
+        }
+        
+        System.out.println(count + ", " + gCount);
+        assertEquals(true, gCount==3);
+    }
+    
+    @Test
+    public void testFinishInReduceMR() throws Exception{
+        gCount = 0;
+        pigServer.registerQuery("a = load '" + Util.generateURI(f1.toString()) 
+ "' using " + PigStorage.class.getName() + "(':');");
+        pigServer.registerQuery("a1 = group a by $1;");
+        pigServer.registerQuery("b = foreach a1 generate " + 
MyEvalFunction.class.getName() + "(*);");
+        Iterator<Tuple> iter = pigServer.openIterator("b");
+        int count = 0;
+        while(iter.hasNext()){
+            ++count;
+            iter.next();
+        }
+        
+        System.out.println(count + ", " + gCount);
+        assertEquals(true, gCount==1);
+    }
+    
+    @Test
+    public void testFinishInMapLoc() throws Exception{
+        pigServer = new PigServer(ExecType.LOCAL);
+        gCount = 0;
+        pigServer.registerQuery("a = load '" + Util.generateURI(f1.toString()) 
+ "' using " + PigStorage.class.getName() + "(':');");
+        pigServer.registerQuery("b = foreach a generate " + 
MyEvalFunction.class.getName() + "(*);");
+        pigServer.openIterator("b");
+        assertEquals(true, gCount==3);
+    }
+    
+    @Test
+    public void testFinishInReduceLoc() throws Exception{
+        pigServer = new PigServer(ExecType.LOCAL);
+        gCount = 0;
+        pigServer.registerQuery("a = load '" + Util.generateURI(f1.toString()) 
+ "' using " + PigStorage.class.getName() + "(':');");
+        pigServer.registerQuery("a1 = group a by $1;");
+        pigServer.registerQuery("b = foreach a1 generate " + 
MyEvalFunction.class.getName() + "(*);");
+        pigServer.openIterator("b");
+        assertEquals(true, gCount==1);
+    }
+
+    @After
+    public void tearDown() throws Exception {
+    }
+
+}


Reply via email to