svn commit: r1529011 - in /pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java test/org/apache/pig/test/TestPigServer.java

[xu...@apache.org]

Author: xuefu
Date: Thu Oct  3 21:32:43 2013
New Revision: 1529011

URL: http://svn.apache.org/r1529011
Log:
PIG-3469: Skewed join can cause unrecoverable NullPointerException when one of its inputs is missing (Jarek Jarcec Cecho  via xuefuz)

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
    pig/trunk/test/org/apache/pig/test/TestPigServer.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1529011&r1=1529010&r2=1529011&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Thu Oct  3 21:32:43 2013
@@ -32,6 +32,7 @@ OPTIMIZATIONS
  
 BUG FIXES
 
+PIG-3469: Skewed join can cause unrecoverable NullPointerException when one of its inputs is missing (Jarek Jarcec Cecho  via xuefuz)
 
 Release 0.12.0 (unreleased changes)
 

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java?rev=1529011&r1=1529010&r2=1529011&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java Thu Oct  3 21:32:43 2013
@@ -864,7 +864,7 @@ public class JobControlCompiler{
             org.apache.hadoop.mapreduce.Job nwJob) throws IOException {
         int jobParallelism = calculateRuntimeReducers(mro, nwJob);
 
-        if (mro.isSampler()) {
+        if (mro.isSampler() && plan.getSuccessors(mro) != null) {
             // We need to calculate the final number of reducers of the next job (order-by or skew-join)
             // to generate the quantfile.
             MapReduceOper nextMro = plan.getSuccessors(mro).get(0);

Modified: pig/trunk/test/org/apache/pig/test/TestPigServer.java
URL: http://svn.apache.org/viewvc/pig/trunk/test/org/apache/pig/test/TestPigServer.java?rev=1529011&r1=1529010&r2=1529011&view=diff
==============================================================================
--- pig/trunk/test/org/apache/pig/test/TestPigServer.java (original)
+++ pig/trunk/test/org/apache/pig/test/TestPigServer.java Thu Oct  3 21:32:43 2013
@@ -601,6 +601,25 @@ public class TestPigServer {
         assertFalse(iter.hasNext());
     }
 
+    // PIG-3469
+    @Test
+    public void testNonExistingSecondDirectoryInSkewJoin() throws Exception {
+        String script =
+          "exists = LOAD 'test/org/apache/pig/test/data/InputFiles/jsTst1.txt' AS (x:chararray, a:long);" +
+          "missing = LOAD '/non/existing/directory' AS (a:long);" +
+          "missing = FOREACH ( GROUP missing BY a ) GENERATE $0 AS a, COUNT_STAR($1);" +
+          "joined = JOIN exists BY a, missing BY a USING 'skewed';" +
+          "STORE joined INTO '/tmp/test_out.tsv';";
+
+        PigServer pig = new PigServer(ExecType.LOCAL);
+        // Execution of the script should fail, but without throwing any exceptions (such as NPE)
+        try {
+            pig.registerScript(new ByteArrayInputStream(script.getBytes("UTF-8")));
+        } catch(Exception ex) {
+            fail("Unexpected exception: " + ex);
+        }
+    }
+
     @Test
     public void testParamSubstitution() throws Exception{
         // using params map