Author: daijy Date: Mon May 3 20:22:14 2010 New Revision: 940601 URL: http://svn.apache.org/viewvc?rev=940601&view=rev Log: PIG-1394: POCombinerPackage hold too much memory for InternalCachedBag
Modified: hadoop/pig/trunk/CHANGES.txt hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCombinerPackage.java Modified: hadoop/pig/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/CHANGES.txt?rev=940601&r1=940600&r2=940601&view=diff ============================================================================== --- hadoop/pig/trunk/CHANGES.txt (original) +++ hadoop/pig/trunk/CHANGES.txt Mon May 3 20:22:14 2010 @@ -249,6 +249,8 @@ OPTIMIZATIONS BUG FIXES +PIG-1394: POCombinerPackage hold too much memory for InternalCachedBag (daijy) + PIG-1374: PushDownForeachFlatten shall not push ForEach below Join if the flattened fields is used in the next statement (daijy) PIG-1336: Optimize POStore serialized into JobConf (daijy) Modified: hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCombinerPackage.java URL: http://svn.apache.org/viewvc/hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCombinerPackage.java?rev=940601&r1=940600&r2=940601&view=diff ============================================================================== --- hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCombinerPackage.java (original) +++ hadoop/pig/trunk/src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POCombinerPackage.java Mon May 3 20:22:14 2010 @@ -66,7 +66,8 @@ public class POCombinerPackage extends P private boolean[] keyPositions; private Map<Integer, Integer> keyLookup; - + + private int numBags; /** * A new POPostCombinePackage will be constructed as a near clone of the @@ -91,6 +92,10 @@ public class POCombinerPackage extends P if (bags != null) { mBags = Arrays.copyOf(bags, bags.length); } + numBags = 0; + for (int i = 0; i < mBags.length; i++) { + if (mBags[i]) numBags++; + } if (keyPos != null) { keyPositions = Arrays.copyOf(keyPos, keyPos.length); } @@ -122,7 +127,7 @@ public class POCombinerPackage extends P keyLookup = lrKeyInfo.second; } - private DataBag createDataBag() { + private DataBag createDataBag(int numBags) { String bagType = null; if (PigMapReduce.sJobConf != null) { bagType = PigMapReduce.sJobConf.get("pig.cachedbag.type"); @@ -131,7 +136,7 @@ public class POCombinerPackage extends P if (bagType != null && bagType.equalsIgnoreCase("default")) { return new NonSpillableDataBag(); } - return new InternalCachedBag(); + return new InternalCachedBag(numBags); } @Override @@ -140,7 +145,7 @@ public class POCombinerPackage extends P //Create numInputs bags Object[] fields = new Object[mBags.length]; for (int i = 0; i < mBags.length; i++) { - if (mBags[i]) fields[i] = createDataBag(); + if (mBags[i]) fields[i] = createDataBag(numBags); } // For each indexed tup in the inp, split them up and place their