pig-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cheol...@apache.org
Subject svn commit: r1404269 - in /pig/branches/branch-0.11: CHANGES.txt src/org/apache/pig/impl/streaming/ExecutableManager.java test/org/apache/pig/test/TestStreaming.java
Date Wed, 31 Oct 2012 17:48:59 GMT
Author: cheolsoo
Date: Wed Oct 31 17:48:59 2012
New Revision: 1404269

URL: http://svn.apache.org/viewvc?rev=1404269&view=rev
Log:
PIG-2973: TestStreaming test times out (cheolsoo)

Modified:
    pig/branches/branch-0.11/CHANGES.txt
    pig/branches/branch-0.11/src/org/apache/pig/impl/streaming/ExecutableManager.java
    pig/branches/branch-0.11/test/org/apache/pig/test/TestStreaming.java

Modified: pig/branches/branch-0.11/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.11/CHANGES.txt?rev=1404269&r1=1404268&r2=1404269&view=diff
==============================================================================
--- pig/branches/branch-0.11/CHANGES.txt (original)
+++ pig/branches/branch-0.11/CHANGES.txt Wed Oct 31 17:48:59 2012
@@ -310,6 +310,8 @@ OPTIMIZATIONS
 
 BUG FIXES
 
+PIG-2973: TestStreaming test times out (cheolsoo)
+
 PIG-3001: TestExecutableManager.testAddJobConfToEnv fails randomly (cheolsoo)
 
 PIG-3017: Pig's object serialization should use compression (jcoveney)

Modified: pig/branches/branch-0.11/src/org/apache/pig/impl/streaming/ExecutableManager.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.11/src/org/apache/pig/impl/streaming/ExecutableManager.java?rev=1404269&r1=1404268&r2=1404269&view=diff
==============================================================================
--- pig/branches/branch-0.11/src/org/apache/pig/impl/streaming/ExecutableManager.java (original)
+++ pig/branches/branch-0.11/src/org/apache/pig/impl/streaming/ExecutableManager.java Wed
Oct 31 17:48:59 2012
@@ -319,8 +319,12 @@ public class ExecutableManager {
     public void run() throws IOException {
         // Check if we need to exec the process NOW ...
         if (inputHandler.getInputType() == InputType.ASYNCHRONOUS) {
-            // start the thread to handle input
-            fileInputThread = new ProcessInputThread(inputHandler, poStream);
+            // start the thread to handle input. we pass the UDFContext to the
+            // fileInputThread because when input type is asynchronous, the
+            // exec() is called by fileInputThread, and it needs to access to
+            // the UDFContext.
+            fileInputThread = new ProcessInputThread(
+                    inputHandler, poStream, UDFContext.getUDFContext());
             fileInputThread.start();
 
             // If Input type is ASYNCHRONOUS that means input to the
@@ -338,7 +342,7 @@ public class ExecutableManager {
         inputHandler.bindTo(stdin);
 
         // Start the thread to send input to the executable's stdin
-        stdinThread = new ProcessInputThread(inputHandler, poStream);
+        stdinThread = new ProcessInputThread(inputHandler, poStream, null);
         stdinThread.start();
     }
 
@@ -350,12 +354,15 @@ public class ExecutableManager {
 
         InputHandler inputHandler;
         private POStream poStream;
+        private UDFContext udfContext;
         private BlockingQueue<Result> binaryInputQueue;
 
-        ProcessInputThread(InputHandler inputHandler, POStream poStream) {
+        ProcessInputThread(InputHandler inputHandler, POStream poStream, UDFContext udfContext)
{
             setDaemon(true);
             this.inputHandler = inputHandler;
             this.poStream = poStream;
+            // a copy of UDFContext passed from the ExecutableManager thread
+            this.udfContext = udfContext;
             // the input queue from where this thread will read
             // input tuples
             this.binaryInputQueue = poStream.getBinaryInputQueue();
@@ -363,6 +370,13 @@ public class ExecutableManager {
 
         @Override
         public void run() {
+            // If input type is asynchronous, set the udfContext of the current
+            // thread to the copy of ExecutableManager thread's udfContext. This
+            // is necessary because the exec() method is called by the current
+            // thread (fileInputThread) instead of the ExecutableManager thread.
+            if (inputHandler.getInputType() == InputType.ASYNCHRONOUS && udfContext
!= null) {
+                UDFContext.setUdfContext(udfContext);
+            }
             try {
                 // Read tuples from the previous operator in the pipeline
                 // and pass it to the executable

Modified: pig/branches/branch-0.11/test/org/apache/pig/test/TestStreaming.java
URL: http://svn.apache.org/viewvc/pig/branches/branch-0.11/test/org/apache/pig/test/TestStreaming.java?rev=1404269&r1=1404268&r2=1404269&view=diff
==============================================================================
--- pig/branches/branch-0.11/test/org/apache/pig/test/TestStreaming.java (original)
+++ pig/branches/branch-0.11/test/org/apache/pig/test/TestStreaming.java Wed Oct 31 17:48:59
2012
@@ -22,6 +22,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Random;
 
 import org.apache.pig.ExecType;
 import org.apache.pig.PigServer;
@@ -40,6 +41,8 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
+import static org.apache.pig.PigConfiguration.PIG_STREAMING_ENVIRONMENT;
+
 public class TestStreaming {
 
     private static final MiniCluster cluster = MiniCluster.buildCluster();
@@ -803,6 +806,77 @@ public class TestStreaming {
         Assert.assertFalse(Util.exists(pig.getPigContext(), "output_dir_001/_logs/mycmd"));
 
     }
+
+    /**
+     * PIG-2973: Verify that JobConf is added to environment even when input to
+     * the streaming binary is asynchronous (i.e. it is from a file).
+     */
+    @Test
+    public void testAddJobConfToEnvironmentWithASynchInput() throws Exception {
+        File input = Util.createInputFile("tmp", "", new String[] {"A"});
+
+        // Generate a random number that will be passed via an environment
+        // variable to the streaming process
+        Random rand = new Random();
+        final int ENV_VAR_VALUE = rand.nextInt();
+        final String ENV_VAR_NAME = "MY_RANDOM_NUMBER";
+
+        // Perl script
+        String[] script =
+            new String[] {
+                          "#!/usr/bin/perl",
+                          "open(INFILE, $ARGV[0]) or die \"Can't open \".$ARGV[0].\"!: $!\";",
+                          "while (<INFILE>) {",
+                          "  chomp $_;",
+                          // Append the value of the environment variable to the line
+                          "  print STDOUT \"$_,$ENV{'" + ENV_VAR_NAME + "'}\n\";",
+                          "  print STDERR \"STDERR: $_\n\";",
+                          "}",
+                         };
+        File command = Util.createInputFile("script", "pl", script);
+
+        // Expected results
+        String[] expectedFirstFields = new String[] {"A"};
+        Integer[] expectedSecondFields = new Integer[] {ENV_VAR_VALUE};
+        Tuple[] expectedResults =
+                setupExpectedResults(Util.toDataByteArrays(expectedFirstFields),
+                                     Util.toDataByteArrays(expectedSecondFields));
+
+        // Set a property and pass it via environment variable to the streaming process
+        pigServer.getPigContext().getProperties()
+                 .setProperty(PIG_STREAMING_ENVIRONMENT, ENV_VAR_NAME);
+        pigServer.getPigContext().getProperties()
+                 .setProperty(ENV_VAR_NAME, Integer.toString(ENV_VAR_VALUE));
+
+        // Pig query to run
+        pigServer.registerQuery(
+                "define CMD `" + command.getName() + " foo` " +
+                "ship ('" + Util.encodeEscape(command.toString()) + "') " +
+                "input('foo' using " + PigStreaming.class.getName() + "()) " +
+                "output(stdout using " + PigStreaming.class.getName() + "(',')) " +
+                "stderr();");
+        pigServer.registerQuery("IP = load '"
+                + Util.generateURI(Util.encodeEscape(input.toString()),
+                        pigServer.getPigContext())
+                + "' using PigStorage();");
+        pigServer.registerQuery("STREAMED_DATA = stream IP through CMD;");
+
+        String output = "/pig/out";
+        pigServer.deleteFile(output);
+        pigServer.store("STREAMED_DATA", output, PigStorage.class.getName() + "(',')");
+
+        pigServer.registerQuery("A = load '" + output + "' using PigStorage(',');");
+        Iterator<Tuple> iter = pigServer.openIterator("A");
+
+        List<Tuple> outputs = new ArrayList<Tuple>();
+        while (iter.hasNext()) {
+            outputs.add(iter.next());
+        }
+
+        // Run the query and check the results
+        Util.checkQueryOutputs(outputs.iterator(), expectedResults);
+    }
+
     public static class PigStreamDump implements PigToStream {
 
         public static final String recordDelimiter = "\n";



Mime
View raw message