svn commit: r1654120 - in /pig/trunk: ./ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/ src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/

[rohini]

Author: rohini
Date: Fri Jan 23 05:39:49 2015
New Revision: 1654120

URL: http://svn.apache.org/r1654120
Log:
PIG-4375: ObjectCache should use ProcessorContext.getObjectRegistry() (rohini)

Modified:
    pig/trunk/CHANGES.txt
    
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
    
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ObjectCache.java
    
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java

Modified: pig/trunk/CHANGES.txt
URL: 
http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1654120&r1=1654119&r2=1654120&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Fri Jan 23 05:39:49 2015
@@ -121,6 +121,8 @@ IMPROVEMENTS
 
 BUG FIXES
 
+PIG-4375: ObjectCache should use ProcessorContext.getObjectRegistry() (rohini)
+
 PIG-4334: PigProcessor does not set pig.datetime.default.tz (rohini)
 
 PIG-4342: Pig 0.14 cannot identify the uppercase of DECLARE and DEFAULT (daijy)

Modified: 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
URL: 
http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java?rev=1654120&r1=1654119&r2=1654120&view=diff
==============================================================================
--- 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
 (original)
+++ 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
 Fri Jan 23 05:39:49 2015
@@ -268,6 +268,8 @@ public class PigCombiner {
             leaf = null;
             pack = null;
             pigReporter = null;
+            // Avoid OOM in Tez.
+            PhysicalOperator.setReporter(null);
             pigContext = null;
             roots = null;
             cp = null;

Modified: 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ObjectCache.java
URL: 
http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ObjectCache.java?rev=1654120&r1=1654119&r2=1654120&view=diff
==============================================================================
--- 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ObjectCache.java
 (original)
+++ 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/ObjectCache.java
 Fri Jan 23 05:39:49 2015
@@ -21,15 +21,15 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.tez.runtime.api.ObjectRegistry;
-import org.apache.tez.runtime.common.objectregistry.ObjectRegistryImpl;
 
-@InterfaceAudience.Private
+@InterfaceAudience.Public
 public class ObjectCache {
 
     private static final Log LOG = LogFactory.getLog(ObjectCache.class);
-    private final ObjectRegistry registry = new ObjectRegistryImpl();
     private static ObjectCache cache = new ObjectCache();
 
+    private ObjectRegistry registry;
+
     private ObjectCache() {
     }
 
@@ -37,11 +37,34 @@ public class ObjectCache {
         return cache;
     }
 
+    /**
+     * Returns the tez ObjectRegistry which allows caching of objects at the
+     * Session, DAG and Vertex level on container reuse for better performance
+     * and savings
+     */
+    public ObjectRegistry getObjectRegistry() {
+        return registry;
+    }
+
+    /**
+     * For internal use only. This method to be called only by PigProcessor
+     */
+    @InterfaceAudience.Private
+    void setObjectRegistry(ObjectRegistry registry) {
+        this.registry = registry;
+    }
+
+    /**
+     * Convenience method to cache objects in ObjectRegistry for a vertex
+     */
     public void cache(String key, Object value) {
       LOG.info("Adding " + key + " to cache");
       registry.cacheForVertex(key, value);
     }
 
+    /**
+     * Convenience method to retrieve objects cached for the vertex from 
ObjectRegistry
+     */
     public Object retrieve(String key) {
       Object o = registry.get(key);
       if (o != null) {

Modified: 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
URL: 
http://svn.apache.org/viewvc/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java?rev=1654120&r1=1654119&r2=1654120&view=diff
==============================================================================
--- 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
 (original)
+++ 
pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/tez/runtime/PigProcessor.java
 Fri Jan 23 05:39:49 2015
@@ -103,6 +103,7 @@ public class PigProcessor extends Abstra
 
     public PigProcessor(ProcessorContext context) {
         super(context);
+        
ObjectCache.getInstance().setObjectRegistry(context.getObjectRegistry());
     }
 
     @SuppressWarnings("unchecked")