Author: daijy
Date: Mon Jul 13 18:58:31 2009
New Revision: 793660

URL: http://svn.apache.org/viewvc?rev=793660&view=rev
Log:
PIG-883: udf import list does not send to the backend

Modified:
    hadoop/pig/trunk/CHANGES.txt
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
    
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
    
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/impl/PigContext.java
    hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java

Modified: hadoop/pig/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/CHANGES.txt (original)
+++ hadoop/pig/trunk/CHANGES.txt Mon Jul 13 18:58:31 2009
@@ -40,6 +40,8 @@
 
 BUG FIXES
 
+    PIG-883: udf import list does not send to the backend (daijy)
+
     PIG-881: Pig should ship load udfs to the backend (daijy)
 
     PIG-876: limit changes order of order-by to ascending (daijy)

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
 Mon Jul 13 18:58:31 2009
@@ -321,6 +321,7 @@
             jobConf.set("pig.inputs", ObjectSerializer.serialize(inp));
             jobConf.set("pig.inpTargets", 
ObjectSerializer.serialize(inpTargets));
             jobConf.set("pig.pigContext", 
ObjectSerializer.serialize(pigContext));
+            jobConf.set("udf.import.list", 
ObjectSerializer.serialize(PigContext.getPackageImportList()));
             // this is for unit tests since some don't create PigServer
             if (pigContext.getProperties().getProperty(PigContext.JOB_NAME) != 
null)
                 
jobConf.setJobName(pigContext.getProperties().getProperty(PigContext.JOB_NAME));

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
 Mon Jul 13 18:58:31 2009
@@ -19,6 +19,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
@@ -94,6 +95,7 @@
             super.configure(jConf);
             sJobConf = jConf;
             try {
+                
PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(jConf.get("udf.import.list")));
                 cp = (PhysicalPlan) ObjectSerializer.deserialize(jConf
                         .get("pig.combinePlan"));
                 pack = 
(POPackage)ObjectSerializer.deserialize(jConf.get("pig.combine.package"));

Modified: 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
 (original)
+++ 
hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigInputFormat.java
 Mon Jul 13 18:58:31 2009
@@ -190,6 +190,7 @@
                                .deserialize(job.get("pig.inpTargets"));
                        pigContext = (PigContext) 
ObjectSerializer.deserialize(job
                                .get("pig.pigContext"));
+                       
PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(job.get("udf.import.list")));
                } catch (Exception e) {
                        int errCode = 2094;
                        String msg = "Unable to deserialize object.";

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=793660&r1=793659&r2=793660&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
 Mon Jul 13 18:58:31 2009
@@ -146,6 +146,7 @@
         SpillableMemoryManager.configure(ConfigurationUtil.toProperties(job));
         PigMapReduce.sJobConf = job;
         try {
+            
PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(job.get("udf.import.list")));
             mp = (PhysicalPlan) ObjectSerializer.deserialize(
                 job.get("pig.mapPlan"));
             stores = PlanHelper.getStores(mp);

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=793660&r1=793659&r2=793660&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
 Mon Jul 13 18:58:31 2009
@@ -19,6 +19,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 
@@ -180,6 +181,7 @@
             
SpillableMemoryManager.configure(ConfigurationUtil.toProperties(jConf));
             sJobConf = jConf;
             try {
+                
PigContext.setPackageImportList((ArrayList<String>)ObjectSerializer.deserialize(jConf.get("udf.import.list")));
                 rp = (PhysicalPlan) ObjectSerializer.deserialize(jConf
                         .get("pig.reducePlan"));
                 stores = PlanHelper.getStores(rp);

Modified: hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java (original)
+++ hadoop/pig/trunk/src/org/apache/pig/impl/PigContext.java Mon Jul 13 
18:58:31 2009
@@ -633,4 +633,8 @@
     {
         return packageImportList;
     }
+    public static void setPackageImportList(ArrayList<String> list)
+    {
+        packageImportList = list;
+    }
 }

Modified: hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java
URL: 
http://svn.apache.org/viewvc/hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java?rev=793660&r1=793659&r2=793660&view=diff
==============================================================================
--- hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java (original)
+++ hadoop/pig/trunk/test/org/apache/pig/test/TestPigContext.java Mon Jul 13 
18:58:31 2009
@@ -23,16 +23,22 @@
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStreamWriter;
+import java.io.PrintStream;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Properties;
+import java.util.Random;
 
 import junit.framework.TestCase;
 
 import org.apache.hadoop.mapred.FileAlreadyExistsException;
+import org.apache.pig.EvalFunc;
 import org.apache.pig.ExecType;
 import org.apache.pig.PigServer;
+import org.apache.pig.data.Tuple;
 import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.io.FileLocalizer;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -45,6 +51,7 @@
 
     private File input;
     private PigContext pigContext;
+    MiniCluster cluster = MiniCluster.buildCluster();
     
     @Before
     @Override
@@ -106,8 +113,14 @@
         File udf1Dir = new 
File(tmpDir.getAbsolutePath()+FILE_SEPARATOR+"com"+FILE_SEPARATOR+"xxx"+FILE_SEPARATOR+"udf1");
         udf1Dir.mkdirs();
         File udf1JavaSrc = new 
File(udf1Dir.getAbsolutePath()+FILE_SEPARATOR+"TestUDF.java");
-        String udf1Src = new String("package com.xxx.udf1;\n" +
-                "public class TestUDF {}\n");
+        String udf1Src = new String("package com.xxx.udf1;\n"+
+                "import java.io.IOException;\n"+
+                "import org.apache.pig.EvalFunc;\n"+
+                "import org.apache.pig.data.Tuple;\n"+
+                "public class TestUDF extends EvalFunc<Integer>{\n"+
+                "public Integer exec(Tuple input) throws IOException {\n"+
+                "return 1;}\n"+
+                "}");
         
         // generate java file
         FileOutputStream outStream = 
@@ -119,7 +132,7 @@
         
         // compile
         int status;
-        status = Util.executeShellCommand("javac " + udf1JavaSrc);
+        status = Util.executeShellCommand("javac -cp 
"+System.getProperty("java.class.path") + " " + udf1JavaSrc);
         
         // generate jar file
         String jarName = "TestUDFJar1.jar";
@@ -142,6 +155,30 @@
         Object udf = PigContext.instantiateFuncFromSpec("TestUDF");
         assertTrue(udf.getClass().toString().endsWith("com.xxx.udf1.TestUDF"));
         
+        int LOOP_COUNT = 40;
+        File tmpFile = File.createTempFile("test", "txt");
+        PrintStream ps = new PrintStream(new FileOutputStream(tmpFile));
+        Random r = new Random(1);
+        int rand;
+        for(int i = 0; i < LOOP_COUNT; i++) {
+            rand = r.nextInt(100);
+            ps.println(rand);
+        }
+        ps.close();
+        
+        FileLocalizer.deleteTempFiles();
+        PigServer pigServer = new PigServer(ExecType.MAPREDUCE, 
cluster.getProperties());
+        pigServer.registerQuery("A = LOAD '" + 
Util.generateURI(tmpFile.toString()) + "' AS (num:chararray);");
+        pigServer.registerQuery("B = foreach A generate TestUDF(num);");
+        Iterator<Tuple> iter = pigServer.openIterator("B");
+        if(!iter.hasNext()) fail("No output found");
+        int numIdentity = 0;
+        while(iter.hasNext()){
+            Tuple t = iter.next();
+            assertTrue(t.get(0) instanceof Integer);
+            assertTrue((Integer)t.get(0) == 1);
+        }
+        
         Util.deleteDirectory(tempDir);
     }
 


Reply via email to