svn commit: r1625347 - in /pig/trunk: CHANGES.txt src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java src/org/apache/pig/impl/PigImplConstants.java src/org/apache/pig/scripting/ScriptingOutputCapturer.java

[ch...@apache.org]

Author: cheolsoo
Date: Tue Sep 16 18:02:06 2014
New Revision: 1625347

URL: http://svn.apache.org/r1625347
Log:
PIG-4171: Streaming UDF fails when direct fetch optimization is enabled (cheolsoo)

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java
    pig/trunk/src/org/apache/pig/impl/PigImplConstants.java
    pig/trunk/src/org/apache/pig/scripting/ScriptingOutputCapturer.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1625347&r1=1625346&r2=1625347&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Tue Sep 16 18:02:06 2014
@@ -70,6 +70,8 @@ OPTIMIZATIONS
  
 BUG FIXES
 
+PIG-4171: Streaming UDF fails when direct fetch optimization is enabled (cheolsoo)
+
 PIG-4170: Multiquery with different type of key gives wrong result (daijy)
 
 PIG-4104: Accumulator UDF throws OOM in Tez (rohini)

Modified: pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java?rev=1625347&r1=1625346&r2=1625347&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java (original)
+++ pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java Tue Sep 16 18:02:06 2014
@@ -57,6 +57,7 @@ import org.apache.pig.backend.hadoop.exe
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStream;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
 import org.apache.pig.impl.PigContext;
+import org.apache.pig.impl.PigImplConstants;
 import org.apache.pig.impl.builtin.SampleLoader;
 import org.apache.pig.impl.plan.DepthFirstWalker;
 import org.apache.pig.impl.plan.VisitorException;
@@ -102,8 +103,10 @@ public class FetchOptimizer {
             // entire input to the client. That can be dangerous.
             boolean isFetchable = fpv.isPlanFetchable() && 
                     PlanHelper.containsPhysicalOperator(pp, POLimit.class);
-            if (isFetchable)
+            if (isFetchable) {
+                pc.getProperties().setProperty(PigImplConstants.CONVERTED_TO_FETCH, "true");
                 init(pp);
+            }
             return isFetchable;
         }
         return false;

Modified: pig/trunk/src/org/apache/pig/impl/PigImplConstants.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/impl/PigImplConstants.java?rev=1625347&r1=1625346&r2=1625347&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/impl/PigImplConstants.java (original)
+++ pig/trunk/src/org/apache/pig/impl/PigImplConstants.java Tue Sep 16 18:02:06 2014
@@ -43,6 +43,11 @@ public class PigImplConstants {
     public static final String CONVERTED_TO_LOCAL = "pig.job.converted.local";
 
     /**
+     * Used by pig to indicate that current job has been converted to run in fetch mode
+     */
+    public static final String CONVERTED_TO_FETCH = "pig.job.converted.fetch";
+
+    /**
      * Indicate the split index of the task. Used by merge cogroup
      */
     public static final String PIG_SPLIT_INDEX = "pig.split.index";

Modified: pig/trunk/src/org/apache/pig/scripting/ScriptingOutputCapturer.java
URL: http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/scripting/ScriptingOutputCapturer.java?rev=1625347&r1=1625346&r2=1625347&view=diff
==============================================================================
--- pig/trunk/src/org/apache/pig/scripting/ScriptingOutputCapturer.java (original)
+++ pig/trunk/src/org/apache/pig/scripting/ScriptingOutputCapturer.java Tue Sep 16 18:02:06 2014
@@ -32,6 +32,7 @@ import org.apache.commons.logging.LogFac
 import org.apache.hadoop.conf.Configuration;
 import org.apache.pig.ExecType;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
+import org.apache.pig.impl.PigImplConstants;
 import org.apache.pig.impl.util.UDFContext;
 
 import com.google.common.base.Charsets;
@@ -100,7 +101,7 @@ public class ScriptingOutputCapturer {
         log.debug("TaskId: " + taskId);
         log.debug("hadoopLogDir: " + hadoopLogDir);
 
-        if (execType.isLocal()) {
+        if (execType.isLocal() || conf.getBoolean(PigImplConstants.CONVERTED_TO_FETCH, false)) {
             String logDir = System.getProperty("pig.udf.scripting.log.dir");
             if (logDir == null)
                 logDir = ".";