changes.txt

Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/Over.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/Over.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/Over.java	(working copy)
@@ -461,8 +461,7 @@
                     // beginning of the bag.
                     if (nextRow < 0) return mTupleFactory.newTuple(1);
 
-                    // Check if the pointer has moved past the end of the window 
-                    if (nextRow >= tuples.size()) {
+                    // Check if the pointer has moved past the end of the window    if (nextRow >= tuples.size()) {
                         return mTupleFactory.newTuple(1);
                     }
 
@@ -838,8 +837,7 @@
                     + iter.begin + " end " + iter.end + " nextRow " +
                     iter.nextRow + " size " + iter.tuples.size());
             System.out.print("{");
-            while (iter.hasNext()) { 
-                Tuple t = iter.next();
+            while (iter.hasNext()) {Tuple t = iter.next();
                 if (t == null) System.out.print("null,");
                 else System.out.print(t.toString() + ",");
             }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COR.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COR.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COR.java	(working copy)
@@ -106,8 +106,7 @@
                         temp.set(0, "var"+i);
                         temp.set(1, "var"+j);
                     }
-                
-                    Tuple tempResult = computeAll((DataBag)input.get(i),(DataBag)input.get(j));
+                   Tuple tempResult = computeAll((DataBag)input.get(i),(DataBag)input.get(j));
                     double size = ((DataBag)input.get(i)).size();
                     double sum_x_y = (Double)tempResult.get(0);
                     double sum_x = (Double)tempResult.get(1);
@@ -231,8 +230,7 @@
             int totalSchemas=2;
             while(totalSchemas*(totalSchemas-1)<combined.size()){
                 totalSchemas++;
-            } 
-                
+            }
             DataBag output = DefaultBagFactory.getInstance().newDefaultBag();
             for(int i=0;i<totalSchemas;i++){
                 for(int j=i+1;j<totalSchemas;j++){
@@ -241,8 +239,7 @@
                         if(flag){
                             result.set(0, schemaName.elementAt(i));
                             result.set(1, schemaName.elementAt(j));
-                        } 
-                        else{
+                        }        else{
                             result.set(0, "var"+i);
                             result.set(1, "var"+j);
                         }
@@ -258,8 +255,7 @@
                     }catch(Exception e){
                         System.err.println("Failed to process input record; error - " + e.getMessage());
                         return null;
-                    } 
-                    output.add(result);
+                    }    output.add(result);
                     count+=2;
                 }
             }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COV.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COV.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/stats/COV.java	(working copy)
@@ -108,8 +108,7 @@
                         temp.set(0, "var"+i);
                         temp.set(1, "var"+j);
                     }
-                
-                    Tuple tempResult = computeAll((DataBag)input.get(i), (DataBag)input.get(j));
+                   Tuple tempResult = computeAll((DataBag)input.get(i), (DataBag)input.get(j));
                     double size = ((DataBag)input.get(i)).size();
                     double sum_x_y = (Double)tempResult.get(0);
                     double sum_x = (Double)tempResult.get(1);
@@ -227,8 +226,7 @@
             //number of schemas would be root of x*x - x - n =0 
             
             try{
-                Tuple combined = combine((DataBag)input.get(0));    
-                int totalSchemas=2;
+                Tuple combined = combine((DataBag)input.get(0));   int totalSchemas=2;
                 while(totalSchemas*(totalSchemas-1)<combined.size()){
                     totalSchemas++;
                 }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/LookupInFiles.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/LookupInFiles.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/LookupInFiles.java	(working copy)
@@ -81,10 +81,8 @@
             for (int i = 0; i < mFiles.size(); ++i) {
                 // Files contain only 1 column with the key. No Schema. All keys
                 // separated by new line.
-    
-                BufferedReader reader = null;
-    
-                InputStream is = null;
+   BufferedReader reader = null;
+   InputStream is = null;
                 try {
                     is = FileLocalizer.openDFSFile(mFiles.get(i), props);
                 } catch (IOException e) {
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/RegexExtractAll.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/RegexExtractAll.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/string/RegexExtractAll.java	(working copy)
@@ -92,8 +92,7 @@
     @Override
     public Schema outputSchema(Schema input) {
         try {
-            return new Schema(new Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input), 
-                    DataType.TUPLE));
+            return new Schema(new Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input),    DataType.TUPLE));
         } catch (Exception e) {
             return null;
         }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/Top.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/Top.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/Top.java	(working copy)
@@ -69,7 +69,7 @@
 @Deprecated 
 
 public class Top extends EvalFunc<DataBag> implements Algebraic{
-    private static final Log log = LogFactory.getLog(Top.class);
+    private static final Log LOG = LogFactory.getLog(Top.class);
     static BagFactory mBagFactory = BagFactory.getInstance();
     static TupleFactory mTupleFactory = TupleFactory.getInstance();
     private Random randomizer = new Random();
@@ -102,8 +102,7 @@
                 }
                 return DataType.compare(field1, field2, datatype, datatype);
             } catch (ExecException e) {
-                throw new RuntimeException("Error while comparing o1:" + o1
-                        + " and o2:" + o2, e);
+                throw new RuntimeException("Error while comparing o1:" + o1+" and o2:" + o2, e);
             }
         }
     }
@@ -124,12 +123,11 @@
             for (Tuple t : store) {
                 outputBag.add(t);
             }
-            if (log.isDebugEnabled()) {
+            if (LOG.isDebugEnabled()) {
                 if (randomizer.nextInt(1000) == 1) {
-                    log.debug("outputting a bag: ");
-                    for (Tuple t : outputBag) 
-                        log.debug("outputting "+t.toDelimitedString("\t"));
-                    log.debug("==================");
+                    LOG.debug("outputting a bag: ");
+                    for (Tuple t : outputBag)        LOG.debug("outputting "+t.toDelimitedString("\t"));
+                    LOG.debug("==================");
                 }
             }
             return outputBag;
@@ -202,7 +200,7 @@
      * <Int, Int, DataBag> -- same schema as expected input.
      */
     static public class Initial extends EvalFunc<Tuple> {
-        //private static final Log log = LogFactory.getLog(Initial.class);
+        //private static final Log LOG = LogFactory.getLog(Initial.class);
         //private final Random randomizer = new Random();
         @Override
         public Tuple exec(Tuple tuple) throws IOException {
@@ -222,8 +220,7 @@
                 }
                 retTuple.set(0, n);
                 retTuple.set(1,fieldNum);
-                retTuple.set(2, outputBag);               
-                return retTuple;
+                retTuple.set(2, outputBag);              return retTuple;
             } catch (Exception e) {
                 throw new RuntimeException("General Exception executing function: " + e);
             }
@@ -231,7 +228,7 @@
     }
 
     static public class Intermed extends EvalFunc<Tuple> {
-        private static final Log log = LogFactory.getLog(Intermed.class);
+        private static final Log LOG = LogFactory.getLog(Intermed.class);
         private final Random randomizer = new Random();
         /* The input is a tuple that contains a single bag.
          * This bag contains outputs of the Initial step --
@@ -278,9 +275,7 @@
                 retTuple.set(0, n);
                 retTuple.set(1,fieldNum);
                 retTuple.set(2, outputBag);
-                if (log.isDebugEnabled()) { 
-                    if (randomizer.nextInt(1000) == 1) log.debug("outputting "+retTuple.toDelimitedString("\t")); 
-                }
+                if (LOG.isDebugEnabled()) {    if (randomizer.nextInt(1000) == 1) LOG.debug("outputting "+retTuple.toDelimitedString("\t"));}
                 return retTuple;
             } catch (ExecException e) {
                 throw new RuntimeException("ExecException executing function: ", e);
@@ -293,7 +288,7 @@
     
     static public class Final extends EvalFunc<DataBag> {
 
-        private static final Log log = LogFactory.getLog(Final.class);
+        private static final Log LOG = LogFactory.getLog(Final.class);
         private final Random randomizer = new Random();
 
 
@@ -340,8 +335,8 @@
                 for (Tuple t : store) {
                     outputBag.add(t);
                 }
-                if (log.isDebugEnabled()) {
-                    if (randomizer.nextInt(1000) == 1) for (Tuple t : outputBag) log.debug("outputting "+t.toDelimitedString("\t"));
+                if (LOG.isDebugEnabled()) {
+                    if (randomizer.nextInt(1000) == 1) for (Tuple t : outputBag) LOG.debug("outputting "+t.toDelimitedString("\t"));
                 }
                 return outputBag;
             } catch (ExecException e) {
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/apachelogparser/DateExtractor.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/apachelogparser/DateExtractor.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/util/apachelogparser/DateExtractor.java	(working copy)
@@ -31,7 +31,7 @@
 /**
  * DateExtractor has four different constructors which each allow for different functionality. The
  * incomingDateFormat ("dd/MMM/yyyy:HH:mm:ss Z" by default) is used to match the date string that gets passed in from the
- * log. The outgoingDateFormat ("yyyy-MM-dd" by default) is used to format the returned string.
+ * LOG. The outgoingDateFormat ("yyyy-MM-dd" by default) is used to format the returned string.
  * 
  * Different constructors exist for each combination; please use the appropriate respective constructor.
  * 
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/xml/XPath.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/xml/XPath.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/evaluation/xml/XPath.java	(working copy)
@@ -88,13 +88,11 @@
             if(!cache || xpath == null || !xml.equals(this.xml))
             {
                 final InputSource source = new InputSource(new StringReader(xml));
-                
-                this.xml = xml; //track the xml for subsequent calls to this udf
+               this.xml = xml; //track the xml for subsequent calls to this udf
 
                 final DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
                 final DocumentBuilder db = dbf.newDocumentBuilder();
-                
-                this.document = db.parse(source);
+               this.document = db.parse(source);
 
                 final XPathFactory xpathFactory = XPathFactory.newInstance();
 
@@ -109,8 +107,7 @@
             return value;
 
         } catch (Exception e) {
-            warn("Error processing input " + input.getType(0), 
-                    PigWarning.UDF_WARNING_1);
+            warn("Error processing input " + input.getType(0),    PigWarning.UDF_WARNING_1);
             
             return null;
         }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/AllLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/AllLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/AllLoader.java	(working copy)
@@ -665,8 +665,7 @@
             FuncSpec funcSpec = loadFuncHelper.determineFunction(fileName);
 
             if (funcSpec == null) {
-                throw new IOException("Cannot determine LoadFunc for "
-                        + fileName);
+                throw new IOException("Cannot determine LoadFunc for "+fileName);
             }
 
             selectedLoadFunc = (LoadFunc) PigContext
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVExcelStorage.java	(working copy)
@@ -319,8 +319,7 @@
             // then the entire field must be enclosed in double quotes:
             embeddedNewlineIndex =  fieldStr.indexOf(LINEFEED);
             
-            if ((fieldStr.indexOf(fieldDelimiter) != -1) || 
-                (fieldStr.indexOf(DOUBLE_QUOTE) != -1) ||
+            if ((fieldStr.indexOf(fieldDelimiter) != -1) ||(fieldStr.indexOf(DOUBLE_QUOTE) != -1) ||
                 (multilineTreatment == Multiline.YES) && (embeddedNewlineIndex != -1))  {
                 fieldStr = "\"" + fieldStr + "\"";
             }
@@ -376,8 +375,7 @@
             } catch (InterruptedException e) {
                 int errCode = 6018;
                 String errMsg = "Error while reading input";
-                throw new ExecException(errMsg, errCode, 
-                        PigException.REMOTE_ENVIRONMENT, e);
+                throw new ExecException(errMsg, errCode,        PigException.REMOTE_ENVIRONMENT, e);
             }
         }
         loadingFirstRecord = false;
@@ -407,54 +405,42 @@
             while (sawEmbeddedRecordDelimiter || getNextFieldID == 0) {
                 Text value = null;
                 if (sawEmbeddedRecordDelimiter) {
-                    
-                    // Deal with pulling more records from the input, because
+                       // Deal with pulling more records from the input, because
                     // a double quoted embedded newline was encountered in a field.
-                    // Save the length of the record so far, plus one byte for the 
-                    // record delimiter (usually newline) that's embedded in the field 
-                    // we were working on before falling into this branch:
+                    // Save the length of the record so far, plus one byte for the    // record delimiter (usually newline) that's embedded in the field    // we were working on before falling into this branch:
                     int prevLineLen = recordLen + 1;
-                    
-                    // Save previous line (the one with the field that has the newline) in a new array.
+                       // Save previous line (the one with the field that has the newline) in a new array.
                     // The last byte will be random; we'll fill in the embedded
                     // record delimiter (usually newline) below:
                     byte[] prevLineSaved = Arrays.copyOf(buf, prevLineLen);
                     prevLineSaved[prevLineLen - 1] = RECORD_DEL;
-                    
-                    // Read the continuation of the record, unless EOF:
+                       // Read the continuation of the record, unless EOF:
                     if (!in.nextKeyValue()) {
                         return null;
-                    }                                                                                           
-                    value = (Text) in.getCurrentValue();
+                    }                                                                                              value = (Text) in.getCurrentValue();
                     recordLen = value.getLength();
                     // Grab the continuation's bytes:
                     buf = value.getBytes();
-                    
-                    // Combine the previous line and the continuation into a new array.
+                       // Combine the previous line and the continuation into a new array.
                     // The following copyOf() does half the job: it allocates all the
                     // space, and also copies the previous line into that space:
                     byte[] prevLineAndContinuation = Arrays.copyOf(prevLineSaved, prevLineLen + recordLen);
-                    
-                    // Now append the continuation. Parms: fromBuf, fromStartPos, toBuf, toStartPos, lengthToCopy:
+                       // Now append the continuation. Parms: fromBuf, fromStartPos, toBuf, toStartPos, lengthToCopy:
                     System.arraycopy(buf, 0, prevLineAndContinuation, prevLineLen, recordLen);
-                    
-                    // We'll work with the combination now:
+                       // We'll work with the combination now:
                     buf = prevLineAndContinuation;
-                    
-                    // Do the whole record over from the start:
+                       // Do the whole record over from the start:
                     mProtoTuple.clear();
                     getNextInQuotedField = false;
                     evenQuotesSeen = true;
                     getNextFieldID = 0;
                     recordLen = prevLineAndContinuation.length;
-                    
-                } else {
+                   } else {
                     // Previous record finished cleanly: start with the next record,
                     // unless EOF:
                     if (!in.nextKeyValue()) {
                         return null;
-                    }                                                                                           
-                    value = (Text) in.getCurrentValue();
+                    }                                                                                              value = (Text) in.getCurrentValue();
 
                     // if the line is a duplicate header and 'SKIP_INPUT_HEADER' is set, ignore it
                     // (this might happen if multiple files each with a header are combined into a single split)
@@ -468,8 +454,7 @@
                     getNextFieldID = 0;
                     recordLen = value.getLength();
                 }
-                
-                nextTupleSkipChar = false;
+               nextTupleSkipChar = false;
 
                 ByteBuffer fieldBuffer = ByteBuffer.allocate(recordLen);
 
@@ -481,15 +466,13 @@
                 // The '!sawEmbeddedRecordDelimiter' handles the case of
                 // embedded newlines; we are amidst a field, not at
                 // the final record:
-                if (!sawEmbeddedRecordDelimiter) 
-                    readField(fieldBuffer, getNextFieldID++);
+                if (!sawEmbeddedRecordDelimiter)    readField(fieldBuffer, getNextFieldID++);
             } // end while
 
         } catch (InterruptedException e) {
             int errCode = 6018;
             String errMsg = "Error while reading input";
-            throw new ExecException(errMsg, errCode, 
-                    PigException.REMOTE_ENVIRONMENT, e);
+            throw new ExecException(errMsg, errCode,    PigException.REMOTE_ENVIRONMENT, e);
         }
 
         Tuple t =  mTupleFactory.newTupleNoCopy(mProtoTuple);
@@ -574,8 +557,7 @@
                     fieldBuffer.put(b);
                 }
             } else if (b == DOUBLE_QUOTE) {
-                // Does a double quote immediately follow?                  
-                if ((i < recordLen-1) && (buf[i+1] == DOUBLE_QUOTE)) {
+                // Does a double quote immediately follow?                 if ((i < recordLen-1) && (buf[i+1] == DOUBLE_QUOTE)) {
                     fieldBuffer.put(b);
                     nextTupleSkipChar = true;
                     continue;
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java	(working copy)
@@ -146,8 +146,7 @@
         } catch (InterruptedException e) {
             int errCode = 6018;
             String errMsg = "Error while reading input";
-            throw new ExecException(errMsg, errCode, 
-                    PigException.REMOTE_ENVIRONMENT, e);
+            throw new ExecException(errMsg, errCode,    PigException.REMOTE_ENVIRONMENT, e);
         }
 
         Tuple t =  mTupleFactory.newTupleNoCopy(mProtoTuple);
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/DBStorage.java	(working copy)
@@ -38,7 +38,7 @@
 import java.sql.*;
 
 public class DBStorage extends StoreFunc {
-  private final Log log = LogFactory.getLog(getClass());
+  private final Log LOG = LogFactory.getLog(getClass());
 
   private PreparedStatement ps;
   private Connection con;
@@ -60,12 +60,12 @@
 
   public DBStorage(String driver, String jdbcURL, String user, String pass,
       String insertQuery, String batchSize) throws RuntimeException {
-    log.debug("DBStorage(" + driver + "," + jdbcURL + "," + user + ",XXXX,"
+    LOG.debug("DBStorage(" + driver + "," + jdbcURL + "," + user + ",XXXX,"
         + insertQuery + ")");
     try {
       Class.forName(driver);
     } catch (ClassNotFoundException e) {
-      log.error("can't load DB driver:" + driver, e);
+      LOG.error("can't load DB driver:" + driver, e);
       throw new RuntimeException("Can't load DB Driver", e);
     }
     this.jdbcURL = jdbcURL;
@@ -219,7 +219,7 @@
               ps = null;
               con = null;
             } catch (SQLException e) {
-              log.error("ps.close", e);
+              LOG.error("ps.close", e);
               throw new IOException("JDBC Error", e);
             }
           }
@@ -294,7 +294,7 @@
       con.setAutoCommit(false);
       ps = con.prepareStatement(insertQuery);
     } catch (SQLException e) {
-      log.error("Unable to connect to JDBC @" + jdbcURL);
+      LOG.error("Unable to connect to JDBC @" + jdbcURL);
       throw new IOException("JDBC Error", e);
     }
     count = 0;
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthLoader.java	(working copy)
@@ -112,7 +112,7 @@
     private String udfContextSignature = null;
     private static final String SCHEMA_SIGNATURE = "pig.fixedwidthloader.schema";
     private static final String REQUIRED_FIELDS_SIGNATURE = "pig.fixedwidthloader.required_fields";
-    private static final Log log = LogFactory.getLog(FixedWidthLoader.class);
+    private static final Log LOG = LogFactory.getLog(FixedWidthLoader.class);
 
     /*
      * Constructors and helper methods
@@ -154,8 +154,7 @@
                 byte fieldType = fields[i].getType();
                 if (fieldType == DataType.MAP || fieldType == DataType.TUPLE || fieldType == DataType.BAG) {
                     throw new IllegalArgumentException(
-                        "Field \"" + fields[i].getName() + "\" is an object type (map, tuple, or bag). " + 
-                        "Object types are not supported by FixedWidthLoader."
+                        "Field \"" + fields[i].getName() + "\" is an object type (map, tuple, or bag). " +        "Object types are not supported by FixedWidthLoader."
                     );
                 }
             }
@@ -188,8 +187,7 @@
                 String offsets[] = range.split("-", 2);
                 offsets[0] = offsets[0].trim();
                 offsets[1] = offsets[1].trim();
-                
-                if (offsets[0].equals(""))
+               if (offsets[0].equals(""))
                     start = 0;
                 else
                     start = Integer.parseInt(offsets[0]) - 1;
@@ -297,8 +295,7 @@
     public Tuple getNext() throws IOException {
         if (loadingFirstRecord && skipHeader && (splitIndex == 0 || splitIndex == -1)) {
             try {
-                if (!reader.nextKeyValue()) 
-                    return null;
+                if (!reader.nextKeyValue())    return null;
                 header = ((Text) reader.getCurrentValue()).toString();
             } catch (Exception e) {
                 throw new IOException(e);
@@ -353,8 +350,7 @@
         return t;
     }
 
-    private Object readField(String line, ResourceFieldSchema field, FixedWidthField column) 
-                             throws IOException, IllegalArgumentException {
+    private Object readField(String line, ResourceFieldSchema field, FixedWidthField column)             throws IOException, IllegalArgumentException {
 
         int start = column.start;
         int end = Math.min(column.end, line.length());
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthStorer.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthStorer.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/FixedWidthStorer.java	(working copy)
@@ -86,7 +86,7 @@
 
     private String udfContextSignature = null;
     private static final String SCHEMA_SIGNATURE = "pig.fixedwidthloader.schema";
-    private static final Log log = LogFactory.getLog(FixedWidthStorer.class);
+    private static final Log LOG = LogFactory.getLog(FixedWidthStorer.class);
 
     /*
      * Constructors and contructor helper methods
@@ -248,8 +248,7 @@
                     int numDigitsRightOfDecimal = width - numDigitsLeftOfDecimal - 1; // should be at least 1
                     String truncated = String.format("%." + numDigitsRightOfDecimal + "f", doubleVal);
 
-                    warn("Cannot fit " + fieldStr + " in field starting at column " + 
-                         column.start + " and ending at column " + (column.end - 1) + ". " +
+                    warn("Cannot fit " + fieldStr + " in field starting at column " +         column.start + " and ending at column " + (column.end - 1) + ". " +
                          "Since the field is a decimal type, truncating it to " + truncated + " " +
                          "to fit in the column.",
                          PigWarning.UDF_WARNING_1);
@@ -256,8 +255,7 @@
                     sb.append(truncated);
                 } else {
                     // Field is float or double but cannot be rounded to fit
-                    warn("Cannot fit " + fieldStr + " in field starting at column " + 
-                         column.start + " and ending at column " + (column.end - 1) + ". " +
+                    warn("Cannot fit " + fieldStr + " in field starting at column " +         column.start + " and ending at column " + (column.end - 1) + ". " +
                          "Writing null (all spaces) instead.",
                          PigWarning.UDF_WARNING_2);
                     for (int i = 0; i < width; i++) {
@@ -265,8 +263,7 @@
                     }
                 }
             } else {
-                warn("Cannot fit " + fieldStr + " in field starting at column " + 
-                      column.start + " and ending at column " + (column.end - 1) + ". " +
+                warn("Cannot fit " + fieldStr + " in field starting at column " +      column.start + " and ending at column " + (column.end - 1) + ". " +
                       "Writing null (all spaces) instead.",
                       PigWarning.UDF_WARNING_2);
                 for (int i = 0; i < width; i++) {
@@ -291,8 +288,7 @@
         return null;
     }
 
-    public void storeStatistics(ResourceStatistics stats, String location, Job job) 
-                                throws IOException {
+    public void storeStatistics(ResourceStatistics stats, String location, Job job)                throws IOException {
         // Not implemented
     }
 
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/HadoopJobHistoryLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/HadoopJobHistoryLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/HadoopJobHistoryLoader.java	(working copy)
@@ -177,39 +177,28 @@
         @Override
         public boolean nextKeyValue() throws IOException, InterruptedException {
             if (location != null) {
-                LOG.info("load: " + location);  
-                Path full = new Path(location);  
-                String[] jobDetails = 
-                    JobInfo.decodeJobHistoryFileName(full.getName()).split("_");
-                String jobId = jobDetails[2] + "_" + jobDetails[3] + "_"
-                        + jobDetails[4];
+                LOG.info("load: " + location); Path full = new Path(location); String[] jobDetails =    JobInfo.decodeJobHistoryFileName(full.getName()).split("_");
+                String jobId = jobDetails[2] + "_" + jobDetails[3] + "_"+jobDetails[4];
                 JobHistory.JobInfo job = new JobHistory.JobInfo(jobId); 
- 
-                value = new MRJobInfo();
-                                            
-                FileSystem fs = full.getFileSystem(conf);
+value = new MRJobInfo();
+                                           FileSystem fs = full.getFileSystem(conf);
                 FileStatus fstat = fs.getFileStatus(full);
-                
-                LOG.info("file size: " + fstat.getLen());
+               LOG.info("file size: " + fstat.getLen());
                 DefaultJobHistoryParser.parseJobTasks(location, job,
-                        full.getFileSystem(conf)); 
-                LOG.info("job history parsed sucessfully");
+                        full.getFileSystem(conf));LOG.info("job history parsed sucessfully");
                 HadoopJobHistoryLoader.parseJobHistory(conf, job, value);
                 LOG.info("get parsed job history");
-                
-                // parse Hadoop job xml file
+               // parse Hadoop job xml file
                 Path parent = full.getParent();
                 String jobXml = jobDetails[0] + "_" + jobDetails[1] + "_" + jobDetails[2] + "_conf.xml";
                 Path p = new Path(parent, jobXml);  
-             
-                FSDataInputStream fileIn = fs.open(p);
+            FSDataInputStream fileIn = fs.open(p);
                 Map<String, String> val = HadoopJobHistoryLoader
                         .parseJobXML(fileIn);
                 for (String key : val.keySet()) {
                     value.job.put(key, val.get(key));
                 }
-                
-                location = null;
+               location = null;
                 return true;
             }          
             value = null;
@@ -285,8 +274,7 @@
                 value.concat(",");
                 parseAndAddJobCounters(job, value);
                 break;
-            default: 
-                LOG.debug("JobHistory.Keys."+ key + " : NOT INCLUDED IN LOADER RETURN VALUE");
+            default:LOG.debug("JobHistory.Keys."+ key + " : NOT INCLUDED IN LOADER RETURN VALUE");
                 break;
             }
         }
@@ -358,18 +346,15 @@
                     JobHistory.Keys key = mtc.getKey();
                     String val = mtc.getValue();
                     switch (key) {
-                    case START_TIME: 
-                        startTime = Long.valueOf(val);
+                    case START_TIME:        startTime = Long.valueOf(val);
                         break;
                     case FINISH_TIME:
                         endTime = Long.valueOf(val);
-                        break;                    
-                    case COUNTERS: {
+                        break;                       case COUNTERS: {
                         try {
                             Counters counters = Counters.fromEscapedCompactString(val);
                             long rows = counters.getGroup(TASK_COUNTER_GROUP)
-                                    .getCounterForName(MAP_INPUT_RECORDS).getCounter(); 
-                            if (rows < minMapRows) minMapRows = rows;
+                                    .getCounterForName(MAP_INPUT_RECORDS).getCounter();            if (rows < minMapRows) minMapRows = rows;
                             if (rows > maxMapRows) maxMapRows = rows;
                         } catch (ParseException e) {
                             LOG.warn("Failed to parse job counters", e);
@@ -376,9 +361,7 @@
                         }
                     }
                     break;
-                    default: 
-                        LOG.warn("JobHistory.Keys." + key 
-                                + " : NOT INCLUDED IN PERFORMANCE ADVISOR MAP COUNTERS");
+                    default:        LOG.warn("JobHistory.Keys." + key                + " : NOT INCLUDED IN PERFORMANCE ADVISOR MAP COUNTERS");
                         break;
                     }
                 }
@@ -407,8 +390,7 @@
                     JobHistory.Keys key = rtc.getKey();
                     String val = rtc.getValue();
                     switch (key) {
-                    case START_TIME: 
-                        startTime = Long.valueOf(val);
+                    case START_TIME:        startTime = Long.valueOf(val);
                         break;
                     case FINISH_TIME:
                         endTime = Long.valueOf(val);
@@ -417,8 +399,7 @@
                         try {
                             Counters counters = Counters.fromEscapedCompactString(val);
                             long rows = counters.getGroup(TASK_COUNTER_GROUP)
-                                    .getCounterForName(REDUCE_INPUT_RECORDS).getCounter(); 
-                            if (rows < minReduceRows) minReduceRows = rows;
+                                    .getCounterForName(REDUCE_INPUT_RECORDS).getCounter();            if (rows < minReduceRows) minReduceRows = rows;
                             if (rows > maxReduceRows) maxReduceRows = rows;
                         } catch (ParseException e) {
                             LOG.warn("Failed to parse job counters", e);
@@ -425,14 +406,11 @@
                         }
                     }
                     break;
-                    default: 
-                        LOG.warn("JobHistory.Keys." + key 
-                                + " : NOT INCLUDED IN PERFORMANCE ADVISOR REDUCE COUNTERS");
+                    default:        LOG.warn("JobHistory.Keys." + key                + " : NOT INCLUDED IN PERFORMANCE ADVISOR REDUCE COUNTERS");
                         break;
                     }
                 }
-                
-                duration = endTime - startTime;
+               duration = endTime - startTime;
                 if (minReduceTime > duration) minReduceTime = duration;
                 if (maxReduceTime < duration) maxReduceTime = duration;
                 totalReduceTime += duration;
@@ -556,8 +534,7 @@
                 String qName) throws SAXException {
             String tag = tags.pop();
             if (tag == null || !tag.equalsIgnoreCase(qName)) {
-                throw new SAXException("Malformatted XML file: " + tag + " : "
-                        + qName);
+                throw new SAXException("Malformatted XML file: " + tag + " : "+qName);
             }
             curTag = null;
         }
@@ -565,8 +542,7 @@
         public void characters(char[] ch, int start, int length)
                 throws SAXException {
             if (tags.size() > 1) {
-                String s = new String(ch, start, length); 
-                if (curTag.equalsIgnoreCase(NAME)) {
+                String s = new String(ch, start, length);if (curTag.equalsIgnoreCase(NAME)) {
                     key = s;
                 }
                 if (curTag.equalsIgnoreCase(VALUE)) {
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/MultiStorage.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/MultiStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/MultiStorage.java	(working copy)
@@ -247,8 +247,7 @@
           CompressionCodec codec = null;
           String extension = "";
           if (isCompressed) {
-             Class<? extends CompressionCodec> codecClass = 
-                getOutputCompressorClass(ctx, GzipCodec.class);
+             Class<? extends CompressionCodec> codecClass =getOutputCompressorClass(ctx, GzipCodec.class);
              codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, ctx.getConfiguration());
              extension = codec.getDefaultExtension();
           }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/RegExLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/RegExLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/RegExLoader.java	(working copy)
@@ -38,7 +38,7 @@
  * There is a single abstract method, getPattern which needs to return a Pattern. Each group will be returned
  * as a different DataAtom.
  * 
- * Look to org.apache.pig.piggybank.storage.apachelog.CommonLogLoader for example usage.
+ * Look to org.apache.pig.piggybank.storage.apacheLOG.CommonLogLoader for example usage.
  */
 
 public abstract class RegExLoader extends LoadFunc {
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/XMLLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/XMLLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/XMLLoader.java	(working copy)
@@ -466,7 +466,7 @@
     /**
      * logger from pig
      */
-    protected final Log mLog = LogFactory.getLog(getClass());
+    protected final Log LOG = LogFactory.getLog(getClass());
 
     private XMLFileRecordReader reader = null;
 
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/allloader/LoadFuncHelper.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/allloader/LoadFuncHelper.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/allloader/LoadFuncHelper.java	(working copy)
@@ -265,8 +265,7 @@
 
             if (loadFuncDefinition == null) {
                 // if still null thrown an error
-                throw new RuntimeException("Cannot find loader for " + path
-                        + " extension mapping " + extensionMapping);
+                throw new RuntimeException("Cannot find loader for " + path+" extension mapping " + extensionMapping);
             }
 
             funcSpec = new FuncSpec(loadFuncDefinition);
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CombinedLogLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CombinedLogLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CombinedLogLoader.java	(working copy)
@@ -24,11 +24,11 @@
  * 
  * The log filename ends up being access_log from a line like
  * 
- * CustomLog logs/combined_log combined
+ * CustoLOG LOGs/combined_log combined
  * 
  * Example:
  * 
- * raw = LOAD 'combined_log' USING org.apache.pig.piggybank.storage.apachelog.CombinedLogLoader AS
+ * raw = LOAD 'combined_log' USING org.apache.pig.piggybank.storage.apacheLOG.CombinedLogLoader AS
  * (remoteAddr, remoteLogname, user, time, method, uri, proto, status, bytes, referer, userAgent);
  * 
  */
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CommonLogLoader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CommonLogLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/apachelog/CommonLogLoader.java	(working copy)
@@ -24,11 +24,11 @@
  * 
  * The log filename ends up being access_log from a line like
  * 
- * CustomLog logs/access_log common
+ * CustoLOG LOGs/access_log common
  * 
  * Example:
  * 
- * raw = LOAD 'access_log' USING org.apache.pig.piggybank.storage.apachelog.CommongLogLoader AS (remoteAddr,
+ * raw = LOAD 'access_log' USING org.apache.pig.piggybank.storage.apacheLOG.CommongLogLoader AS (remoteAddr,
  * remoteLogname, user, time, method, uri, proto, bytes);
  * 
  */
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchema2Pig.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchema2Pig.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchema2Pig.java	(working copy)
@@ -77,7 +77,7 @@
      private static ResourceFieldSchema inconvert(Schema in, String fieldName, Set<Schema> visitedRecords)
              throws IOException {
 
-        AvroStorageLog.details("InConvert avro schema with field name " + fieldName);
+        AvroStorageLOG.details("InConvert avro schema with field name " + fieldName);
 
         Schema.Type avroType = in.getType();
         ResourceFieldSchema fieldSchema = new ResourceFieldSchema();
@@ -85,7 +85,7 @@
 
         if (avroType.equals(Schema.Type.RECORD)) {
 
-            AvroStorageLog.details("convert to a pig tuple");
+            AvroStorageLOG.details("convert to a pig tuple");
 
             if (visitedRecords.contains(in)) {
                 fieldSchema.setType(DataType.BYTEARRAY);
@@ -107,7 +107,7 @@
 
         } else if (avroType.equals(Schema.Type.ARRAY)) {
 
-            AvroStorageLog.details("convert array to a pig bag");
+            AvroStorageLOG.details("convert array to a pig bag");
             fieldSchema.setType(DataType.BAG);
             Schema elemSchema = in.getElementType();
             ResourceFieldSchema subFieldSchema = inconvert(elemSchema, ARRAY_FIELD, visitedRecords);
@@ -115,7 +115,7 @@
 
         } else if (avroType.equals(Schema.Type.MAP)) {
 
-            AvroStorageLog.details("convert map to a pig map");
+            AvroStorageLOG.details("convert map to a pig map");
             fieldSchema.setType(DataType.MAP);
 
         } else if (avroType.equals(Schema.Type.UNION)) {
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchemaManager.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchemaManager.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroSchemaManager.java	(working copy)
@@ -73,13 +73,9 @@
         if (isNamedSchema(schema)) {
             String typeName = schema.getName();
             if (typeName2Schema.containsKey(typeName))
-                AvroStorageLog.warn("Duplicate schemas defined for type:"
-                        + typeName
-                        + ". will ignore the second one:"
-                        + schema);
+                AvroStorageLOG.warn("Duplicate schemas defined for type:"+typeName+". will ignore the second one:"+schema);
             else {
-                AvroStorageLog.details("add " + schema.getName() + "=" + schema
-                        + " to type2Schema");
+                AvroStorageLOG.details("add " + schema.getName() + "=" + schema+" to type2Schema");
                 typeName2Schema.put(schema.getName(), schema);
             }
         }
@@ -95,10 +91,10 @@
 
                 if (!ignoreNameMap) {
                     if (name2Schema.containsKey(name))
-                        AvroStorageLog.warn("Duplicate schemas defined for alias:" + name
+                        AvroStorageLOG.warn("Duplicate schemas defined for alias:" + name
                                           + ". Will ignore the second one:"+ fieldSchema);
                     else {
-                        AvroStorageLog.details("add " + name + "=" + fieldSchema + " to name2Schema");
+                        AvroStorageLOG.details("add " + name + "=" + fieldSchema + " to name2Schema");
                         name2Schema.put(name, fieldSchema);
                     }
                 }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorage.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/AvroStorage.java	(working copy)
@@ -111,7 +111,7 @@
     public AvroStorage() {
         outputAvroSchema = null;
         nullable = true;
-        AvroStorageLog.setDebugLevel(0);
+        AvroStorageLOG.setDebugLevel(0);
         checkSchema = true;
     }
 
@@ -334,7 +334,7 @@
     @SuppressWarnings("rawtypes")
     @Override
     public InputFormat getInputFormat() throws IOException {
-        AvroStorageLog.funcCall("getInputFormat");
+        AvroStorageLOG.funcCall("getInputFormat");
         InputFormat result = null;
         if(inputAvroSchema != null) {
             result = new PigAvroInputFormat(
@@ -349,7 +349,7 @@
     @Override
     public void prepareToRead(RecordReader reader, PigSplit split)
     throws IOException {
-        AvroStorageLog.funcCall("prepareToRead");
+        AvroStorageLOG.funcCall("prepareToRead");
         this.reader = (PigAvroRecordReader) reader;
     }
 
@@ -376,7 +376,7 @@
     public ResourceSchema getSchema(String location, Job job) throws IOException {
 
         /* get avro schema */
-        AvroStorageLog.funcCall("getSchema");
+        AvroStorageLOG.funcCall("getSchema");
         if (inputAvroSchema == null) {
             Configuration conf = job.getConfiguration();
             // If within a script, you store to one location and read from same
@@ -390,11 +390,11 @@
             }
         }
         if(inputAvroSchema != null) {
-            AvroStorageLog.details( "avro input schema:"  + inputAvroSchema);
+            AvroStorageLOG.details( "avro input schema:"  + inputAvroSchema);
 
             /* convert to pig schema */
             ResourceSchema pigSchema = AvroSchema2Pig.convert(inputAvroSchema);
-            AvroStorageLog.details("pig input schema:" + pigSchema);
+            AvroStorageLOG.details("pig input schema:" + pigSchema);
             if (pigSchema.getFields().length == 1){
                 pigSchema = pigSchema.getFields()[0].getSchema();
             }
@@ -516,7 +516,7 @@
 
         /* set debug level */
         if (inputs.containsKey("debug")) {
-            AvroStorageLog.setDebugLevel((Integer) inputs.get("debug"));
+            AvroStorageLOG.setDebugLevel((Integer) inputs.get("debug"));
         }
 
         /* initialize schema manager, if any */
@@ -523,7 +523,7 @@
         AvroSchemaManager schemaManager = null;
         if (inputs.containsKey("data")) {
             Path path = new Path((String) inputs.get("data"));
-            AvroStorageLog.details("data path=" + path.toUri().toString());
+            AvroStorageLOG.details("data path=" + path.toUri().toString());
             FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
             Schema schema = getAvroSchema(path, fs);
             schemaManager = new AvroSchemaManager(schema);
@@ -530,7 +530,7 @@
         }
         else if (inputs.containsKey("schema_file")) {
             Path path = new Path((String) inputs.get("schema_file"));
-            AvroStorageLog.details("schemaFile path=" + path.toUri().toString());
+            AvroStorageLOG.details("schemaFile path=" + path.toUri().toString());
             FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
             Schema schema = getSchemaFromFile(path, fs);
             schemaManager = new AvroSchemaManager(schema);
@@ -547,7 +547,7 @@
             } else if (name.equalsIgnoreCase("same")) {
                 /* use schema in the specified path as output schema */
                 Path path = new Path( ((String) value).trim());
-                AvroStorageLog.details("data path=" + path.toUri().toString());
+                AvroStorageLOG.details("data path=" + path.toUri().toString());
                 FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
                 outputAvroSchema = getAvroSchema(path, fs);
             } else if (name.equalsIgnoreCase("nullable")) {
@@ -558,7 +558,7 @@
             } else if (name.equalsIgnoreCase("schema_uri")) {
                 /* use the contents of the specified path as output schema */
                 Path path = new Path( ((String) value).trim());
-                AvroStorageLog.details("schema_uri path=" + path.toUri().toString());
+                AvroStorageLOG.details("schema_uri path=" + path.toUri().toString());
                 FileSystem fs = FileSystem.get(path.toUri(), new Configuration());
                 outputAvroSchema = getSchemaFromFile(path, fs);
                 userSpecifiedAvroSchema = outputAvroSchema;
@@ -584,7 +584,7 @@
                     /* use pre-defined schema*/
                     field = AvroStorageUtils.createUDField(index, s);
 
-                    AvroStorageLog.details("Use pre-defined schema(" + alias + "): " + s + " for field " + index);
+                    AvroStorageLOG.details("Use pre-defined schema(" + alias + "): " + s + " for field " + index);
                 } else {
                     Schema schema = null;
                     try {
@@ -615,8 +615,7 @@
          * and nullable will be ignored.*/
         if (outputAvroSchema != null) {
             if (!nullable) {
-                AvroStorageLog.warn("Invalid parameter--nullable cannot be false while "
-                        + "output schema is not null. Will ignore nullable.\n\n");
+                AvroStorageLOG.warn("Invalid parameter--nullable cannot be false while "+"output schema is not null. Will ignore nullable.\n\n");
                 nullable = true;
             }
         }
@@ -630,7 +629,7 @@
 
     @Override
     public void setStoreLocation(String location, Job job) throws IOException {
-        AvroStorageLog.details("output location=" + location);
+        AvroStorageLOG.details("output location=" + location);
         FileOutputFormat.setOutputPath(job, new Path(location));
     }
 
@@ -639,10 +638,10 @@
      */
     @Override
     public void checkSchema(ResourceSchema s) throws IOException {
-        AvroStorageLog.funcCall("Check schema");
+        AvroStorageLOG.funcCall("Check schema");
         Properties property = getUDFProperties();
         String prevSchemaStr = property.getProperty(AVRO_OUTPUT_SCHEMA_PROPERTY);
-        AvroStorageLog.details("Previously defined schemas=" + prevSchemaStr);
+        AvroStorageLOG.details("Previously defined schemas=" + prevSchemaStr);
 
         String key = getSchemaKey();
         Map<String, String> schemaMap = (prevSchemaStr != null)
@@ -650,7 +649,7 @@
                                                                 : null;
 
         if (schemaMap != null && schemaMap.containsKey(key)) {
-            AvroStorageLog.warn("Duplicate value for key-" + key + ". Will ignore the new schema.");
+            AvroStorageLOG.warn("Duplicate value for key-" + key + ". Will ignore the new schema.");
             return;
         }
 
@@ -661,7 +660,7 @@
                                           : outputAvroSchema)
                                   : PigSchema2Avro.convert(s, nullable);
 
-        AvroStorageLog.info("key=" + key + " outputSchema=" + schema);
+        AvroStorageLOG.info("key=" + key + " outputSchema=" + schema);
 
         String schemaStr = schema.toString();
         String append = key + SCHEMA_KEYVALUE_DELIM + schemaStr;
@@ -670,7 +669,7 @@
                                                 ? prevSchemaStr + SCHEMA_DELIM + append
                                                 : append;
         property.setProperty(AVRO_OUTPUT_SCHEMA_PROPERTY, newSchemaStr);
-        AvroStorageLog.details("New schemas=" + newSchemaStr);
+        AvroStorageLOG.details("New schemas=" + newSchemaStr);
     }
 
     /**
@@ -686,12 +685,12 @@
     }
 
     private Map<String, String> parseSchemaMap(String input) throws IOException {
-        AvroStorageLog.details("Parse schema map from " + input);
+        AvroStorageLOG.details("Parse schema map from " + input);
         String[] entries = input.split(SCHEMA_DELIM);
         Map<String, String> map = new HashMap<String, String>();
         for (String entry : entries) {
 
-            AvroStorageLog.details("Entry = " + entry);
+            AvroStorageLOG.details("Entry = " + entry);
             if (entry.length() == 0)
                 continue;
 
@@ -706,7 +705,7 @@
     @SuppressWarnings("rawtypes")
     @Override
     public OutputFormat getOutputFormat() throws IOException {
-        AvroStorageLog.funcCall("getOutputFormat");
+        AvroStorageLOG.funcCall("getOutputFormat");
 
         Properties property = getUDFProperties();
         String allSchemaStr = property.getProperty(AVRO_OUTPUT_SCHEMA_PROPERTY);
@@ -717,7 +716,7 @@
 
         if (schema == null)
             throw new IOException("Output schema is null!");
-        AvroStorageLog.details("Output schema=" + schema);
+        AvroStorageLOG.details("Output schema=" + schema);
 
         return new PigAvroOutputFormat(schema);
     }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigAvroRecordReader.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigAvroRecordReader.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigAvroRecordReader.java	(working copy)
@@ -82,7 +82,7 @@
         this.in = new AvroStorageInputStream(path, context);
         this.useMultipleSchemas = useMultipleSchemas;
         if(readerSchema == null) {
-            AvroStorageLog.details("No avro schema given; assuming the schema is embedded");
+            AvroStorageLOG.details("No avro schema given; assuming the schema is embedded");
         }
 
         Schema writerSchema;
@@ -90,7 +90,7 @@
             FileSystem fs = FileSystem.get(path.toUri(), context.getConfiguration());
             writerSchema = AvroStorageUtils.getSchema(path, fs);
         } catch (IOException e) {
-            AvroStorageLog.details("No avro writer schema found in '"+path+"'; assuming writer schema matches reader schema");
+            AvroStorageLOG.details("No avro writer schema found in '"+path+"'; assuming writer schema matches reader schema");
             writerSchema = null;
         }
 
@@ -119,7 +119,7 @@
                 }
             }
             int tupleSize = maxPos + 1;
-            AvroStorageLog.details("Creating proto tuple of fixed size: " + tupleSize);
+            AvroStorageLOG.details("Creating proto tuple of fixed size: " + tupleSize);
             mProtoTuple = new ArrayList<Object>(tupleSize);
             for (int i = 0; i < tupleSize; i++) {
                 // Get the list of fields from the passed schema
@@ -197,14 +197,14 @@
         Object obj = reader.next();
         Tuple result = null;
         if (obj instanceof Tuple) {
-            AvroStorageLog.details("Class =" + obj.getClass());
+            AvroStorageLOG.details("Class =" + obj.getClass());
             result = (Tuple) obj;
         } else {
             if (obj != null) {
-                AvroStorageLog.details("Wrap class " + obj.getClass() + " as a tuple.");
+                AvroStorageLOG.details("Wrap class " + obj.getClass() + " as a tuple.");
             }
             else {
-                AvroStorageLog.details("Wrap null as a tuple.");
+                AvroStorageLOG.details("Wrap null as a tuple.");
             }
             result = wrapAsTuple(obj);
         }
Index: contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigSchema2Avro.java
===================================================================
--- contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigSchema2Avro.java	(revision 1579421)
+++ contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/avro/PigSchema2Avro.java	(working copy)
@@ -58,7 +58,7 @@
         /* remove the pig tuple wrapper */
         if (pigFields.length == 1) {
 
-            AvroStorageLog.details("Ignore the pig tuple wrapper.");
+            AvroStorageLOG.details("Ignore the pig tuple wrapper.");
             return convert(pigFields[0], nullable);
         } else
             return convertRecord(pigFields, nullable);
@@ -70,12 +70,12 @@
      */
     protected static Schema convert(ResourceFieldSchema pigSchema, boolean nullable) throws IOException {
 
-        AvroStorageLog.details("Convert pig field schema:" + pigSchema);
+        AvroStorageLOG.details("Convert pig field schema:" + pigSchema);
 
         final byte pigType = pigSchema.getType();
 
         if (pigType == DataType.TUPLE) {
-            AvroStorageLog.details("Convert a pig field tuple: " + pigSchema);
+            AvroStorageLOG.details("Convert a pig field tuple: " + pigSchema);
 
             ResourceFieldSchema[] listSchemas = pigSchema.getSchema()
                                             .getFields();
@@ -83,7 +83,7 @@
 
             if (AvroStorageUtils.isTupleWrapper(pigSchema)) {
                 /* remove Pig tuple wrapper */
-                AvroStorageLog.details("Ignore the pig tuple wrapper.");
+                AvroStorageLOG.details("Ignore the pig tuple wrapper.");
                 if (listSchemas.length != 1)
                     throw new IOException("Expect one subfield from "
                                                     + pigSchema);
@@ -96,7 +96,7 @@
 
         } else if (pigType == DataType.BAG) {
 
-            AvroStorageLog.details("Convert a pig field bag:" + pigSchema);
+            AvroStorageLOG.details("Convert a pig field bag:" + pigSchema);
 
             /* Bag elements have to be Tuples */
             ResourceFieldSchema[] fs = pigSchema.getSchema().getFields();
@@ -125,7 +125,7 @@
                                         || pigType == DataType.INTEGER
                                         || pigType == DataType.LONG) {
 
-            AvroStorageLog.details("Convert a pig field primitive:" + pigSchema);
+            AvroStorageLOG.details("Convert a pig field primitive:" + pigSchema);
             Schema outSchema = convertPrimitiveType(pigType);
             return AvroStorageUtils.wrapAsUnion(outSchema, nullable);
 
@@ -140,7 +140,7 @@
      */
     protected static Schema convertRecord(ResourceFieldSchema[] pigFields, boolean nullable) throws IOException {
 
-        AvroStorageLog.funcCall("convertRecord");
+        AvroStorageLOG.funcCall("convertRecord");
 
         // Type name is required for Avro record
         String typeName = getRecordName();
@@ -229,7 +229,7 @@
      */
     protected static Schema validateAndConvert(Schema avroSchema, ResourceFieldSchema pigSchema) throws IOException {
 
-        AvroStorageLog.details("Validate pig field schema:" + pigSchema);
+        AvroStorageLOG.details("Validate pig field schema:" + pigSchema);
 
         /* compatibility check based on data types */
         if (!isCompatible(avroSchema, pigSchema))
@@ -237,7 +237,7 @@
 
         final byte pigType = pigSchema.getType();
         if (avroSchema.getType().equals(Schema.Type.UNION)) {
-            AvroStorageLog.details("Validate Pig schema with Avro union:" + avroSchema);
+            AvroStorageLOG.details("Validate Pig schema with Avro union:" + avroSchema);
 
             List<Schema> unionSchemas = avroSchema.getTypes();
             for (Schema schema : unionSchemas) {
@@ -251,13 +251,13 @@
             }
             throw new IOException("pig schema " + pigSchema  + " is not compatible with avro " + avroSchema);
         } else if (pigType == DataType.TUPLE) {
-            AvroStorageLog.details("Validate a pig tuple: " + pigSchema);
+            AvroStorageLOG.details("Validate a pig tuple: " + pigSchema);
             ResourceFieldSchema[] pigFields = pigSchema.getSchema().getFields();
             Schema outSchema = validateAndConvertRecord(avroSchema, pigFields);
             return outSchema;
 
         } else if (pigType == DataType.BAG) {
-            AvroStorageLog.details("Validate a pig bag:" + pigSchema);
+            AvroStorageLOG.details("Validate a pig bag:" + pigSchema);
 
             /* get fields of containing tuples */
             ResourceFieldSchema[] fs = pigSchema.getSchema().getFields();
@@ -268,7 +268,7 @@
             Schema outSchema = Schema.createArray(validateAndConvert(inElemSchema, fs[0]));
             return outSchema;
         } else if (pigType == DataType.MAP) {
-            AvroStorageLog.details("Cannot validate a pig map. Will use user defined Avro schema.");
+            AvroStorageLOG.details("Cannot validate a pig map. Will use user defined Avro schema.");
             return avroSchema;
 
         } else if (pigType == DataType.UNKNOWN  || pigType == DataType.CHARARRAY
@@ -281,7 +281,7 @@
                                                 || pigType == DataType.INTEGER
                                                 || pigType == DataType.LONG) {
 
-            AvroStorageLog.details("Validate a pig primitive type:" + pigSchema);
+            AvroStorageLOG.details("Validate a pig primitive type:" + pigSchema);
             return avroSchema;
 
         } else
@@ -309,7 +309,7 @@
 
         /* validate and convert a pig tuple with avro record */
         boolean isPartialSchema = AvroStorageUtils.isUDPartialRecordSchema(avroSchema);
-        AvroStorageLog.details("isPartialSchema=" + isPartialSchema);
+        AvroStorageLOG.details("isPartialSchema=" + isPartialSchema);
 
         String typeName = isPartialSchema ? getRecordName() : avroSchema.getName();
         Schema outSchema = Schema.createRecord(typeName, avroSchema.getDoc(), avroSchema.getNamespace(), false);
@@ -327,14 +327,11 @@
 
             /* get schema */
             Schema fieldSchema = null;
-            if (inputField == null) { 
-                /* convert pig schema (nullable) */
+            if (inputField == null) {/* convert pig schema (nullable) */
                 fieldSchema = convert(pigFields[i], true);
-            } else if (inputField.schema() == null) { 
-                /* convert pig schema (not-null) */
+            } else if (inputField.schema() == null) {/* convert pig schema (not-null) */
                 fieldSchema = convert(pigFields[i], false);
-            } else { 
-                /* validate pigFields[i] with given avro schema */
+            } else {/* validate pigFields[i] with given avro schema */
                 fieldSchema = validateAndConvert(inputField.schema(),
                                                 pigFields[i]);
             }
@@ -376,32 +373,20 @@
         }
         return  (avroType.equals(Schema.Type.ARRAY) && pigType == DataType.BAG)
                       || (avroType.equals(Schema.Type.MAP) && pigType == DataType.MAP)
-                      || (avroType.equals(Schema.Type.STRING) 
-                                                      && pigType == DataType.CHARARRAY 
-                                                      || pigType == DataType.BIGCHARARRAY)
-                      || (avroType.equals(Schema.Type.ENUM) 
-                                                      && pigType == DataType.CHARARRAY)
-                      || (avroType.equals(Schema.Type.BOOLEAN) 
-                                                      && pigType == DataType.BOOLEAN 
-                                                      || pigType == DataType.INTEGER)
-                      || (avroType.equals(Schema.Type.BYTES) 
-                                                      && pigType == DataType.BYTEARRAY)
-                      || (avroType.equals(Schema.Type.DOUBLE) 
-                                                      && pigType == DataType.DOUBLE
+                      || (avroType.equals(Schema.Type.STRING)                                      && pigType == DataType.CHARARRAY                                      || pigType == DataType.BIGCHARARRAY)
+                      || (avroType.equals(Schema.Type.ENUM)                                      && pigType == DataType.CHARARRAY)
+                      || (avroType.equals(Schema.Type.BOOLEAN)                                      && pigType == DataType.BOOLEAN                                      || pigType == DataType.INTEGER)
+                      || (avroType.equals(Schema.Type.BYTES)                                      && pigType == DataType.BYTEARRAY)
+                      || (avroType.equals(Schema.Type.DOUBLE)                                      && pigType == DataType.DOUBLE
                                                       || pigType == DataType.FLOAT
-                                                      || pigType == DataType.INTEGER 
-                                                      || pigType == DataType.LONG)
+                                                      || pigType == DataType.INTEGER                                      || pigType == DataType.LONG)
                       || (avroType.equals(Schema.Type.FLOAT)
                                                       && pigType == DataType.FLOAT
-                                                      || pigType == DataType.INTEGER 
-                                                      || pigType == DataType.LONG)
-                      || (avroType.equals(Schema.Type.FIXED) 
-                                                      && pigType == DataType.BYTEARRAY)
-                      || (avroType.equals(Schema.Type.INT) 
-                                                      && pigType == DataType.INTEGER)
+                                                      || pigType == DataType.INTEGER                                      || pigType == DataType.LONG)
+                      || (avroType.equals(Schema.Type.FIXED)                                      && pigType == DataType.BYTEARRAY)
+                      || (avroType.equals(Schema.Type.INT)                                      && pigType == DataType.INTEGER)
                       || (avroType.equals(Schema.Type.LONG)
-                                                      && pigType == DataType.LONG 
-                                                      || pigType == DataType.INTEGER);
+                                                      && pigType == DataType.LONG                                      || pigType == DataType.INTEGER);
 
     }
 
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/TestMathUDF.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/TestMathUDF.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/TestMathUDF.java	(working copy)
@@ -190,7 +190,7 @@
              LOG  log= new LOG();
             Tuple tup = DefaultTupleFactory.getInstance().newTuple(1);
             try{tup.set(0, 0.5);}catch(Exception e){}
-            Double actual = log.exec(tup);
+            Double actual = LOG.exec(tup);
             double expected = Math.log(0.5);
             assertEquals(actual, expected, delta);
      }
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/string/TestSplit.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/string/TestSplit.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/string/TestSplit.java	(working copy)
@@ -41,8 +41,7 @@
         test2_.set(1, ":");
         Tuple splits = splitter_.exec(test2_);
         assertEquals("no matches should return tuple with original string", 1, splits.size());
-        assertEquals("no matches should return tuple with original string", "foo", 
-                splits.get(0));
+        assertEquals("no matches should return tuple with original string", "foo",splits.get(0));
         
         // test default delimiter
         test1_.set(0, "f ooo bar");
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/xml/XPathTest.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/xml/XPathTest.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/evaluation/xml/XPathTest.java	(working copy)
@@ -155,8 +155,7 @@
                           "<author>Gambardella, Matthew</author>" +
                           "<title>XML Developer's Guide</title>" +
                           "<genre>Computer</genre>" +
-                          expandXml() + 
-                          "<price>44.95</price>" +
+                          expandXml() +          "<price>44.95</price>" +
                           "<publish_date>2000-10-01</publish_date>" +
                           "<description>An in-depth look at creating applications with XML.</description>" +
                      "</book>");
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestAllLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestAllLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestAllLoader.java	(working copy)
@@ -424,7 +424,7 @@
             count++;
         }
 
-        Log.info("Validating expected: " + totalRowCount + " against " + count);
+        LOG.info("Validating expected: " + totalRowCount + " against " + count);
         assertEquals(totalRowCount, count);
     }
 
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVExcelStorage.java	(working copy)
@@ -361,8 +361,7 @@
             "using org.apache.pig.piggybank.storage.CSVExcelStorage(',', 'YES_MULTILINE', 'UNIX', 'SKIP_INPUT_HEADER') " + 
             "AS (" + schema + ");"
         );
-        pig.store("data", dataDir + output, 
-                  "org.apache.pig.piggybank.storage.CSVExcelStorage(',', 'YES_MULTILINE', 'UNIX', 'WRITE_OUTPUT_HEADER')");
+        pig.store("data", dataDir + output,  "org.apache.pig.piggybank.storage.CSVExcelStorage(',', 'YES_MULTILINE', 'UNIX', 'WRITE_OUTPUT_HEADER')");
 
         // Read it back
 
@@ -409,8 +408,7 @@
             "using PigStorage('|')" + 
             "AS (" + schema + ");"
         );
-        pig.store("data", dataDir + output, 
-                  "org.apache.pig.piggybank.storage.CSVExcelStorage(',', 'YES_MULTILINE', 'UNIX', 'SKIP_OUTPUT_HEADER')");
+        pig.store("data", dataDir + output,  "org.apache.pig.piggybank.storage.CSVExcelStorage(',', 'YES_MULTILINE', 'UNIX', 'SKIP_OUTPUT_HEADER')");
 
         pig.registerQuery(
             "data = load '" + dataDir + output + "' " +
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java	(working copy)
@@ -74,8 +74,7 @@
     @Test
     public void testQuotedQuotes() throws IOException {
         String inputFileName = "TestCSVLoader-quotedquotes.txt";
-        Util.createLocalInputFile(inputFileName, 
-                new String[] {"\"foo,\"\"bar\"\",baz\"", "\"\"\"\"\"\"\"\""});
+        Util.createLocalInputFile(inputFileName,new String[] {"\"foo,\"\"bar\"\",baz\"", "\"\"\"\"\"\"\"\""});
         String script = "a = load '" + inputFileName + "' using org.apache.pig.piggybank.storage.CSVLoader() " +
         "   as (a:chararray); ";
         Util.registerMultiLineQuery(pigServer, script);
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCombinedLogLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCombinedLogLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCombinedLogLoader.java	(working copy)
@@ -22,7 +22,7 @@
 import org.apache.pig.PigServer;
 import org.apache.pig.data.DataByteArray;
 import org.apache.pig.data.Tuple;
-import org.apache.pig.piggybank.storage.apachelog.CombinedLogLoader;
+import org.apache.pig.piggybank.storage.apacheLOG.CombinedLogLoader;
 import org.junit.Test;
 
 public class TestCombinedLogLoader extends TestCase {
@@ -82,7 +82,7 @@
         String filename = TestHelper.createTempFile(data, " ");
         PigServer pig = new PigServer(ExecType.LOCAL);
         filename = filename.replace("\\", "\\\\");
-        pig.registerQuery("A = LOAD '" + filename + "' USING org.apache.pig.piggybank.storage.apachelog.CombinedLogLoader();");
+        pig.registerQuery("A = LOAD '" + filename + "' USING org.apache.pig.piggybank.storage.apacheLOG.CombinedLogLoader();");
         Iterator<?> it = pig.openIterator("A");
 
         int tupleCount = 0;
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCommonLogLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCommonLogLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCommonLogLoader.java	(working copy)
@@ -22,7 +22,7 @@
 import org.apache.pig.PigServer;
 import org.apache.pig.data.DataByteArray;
 import org.apache.pig.data.Tuple;
-import org.apache.pig.piggybank.storage.apachelog.CommonLogLoader;
+import org.apache.pig.piggybank.storage.apacheLOG.CommonLogLoader;
 import org.junit.Test;
 
 public class TestCommonLogLoader extends TestCase {
@@ -73,7 +73,7 @@
         String filename = TestHelper.createTempFile(data, " ");
         PigServer pig = new PigServer(ExecType.LOCAL);
         filename = filename.replace("\\", "\\\\");
-        pig.registerQuery("A = LOAD '" + filename + "' USING org.apache.pig.piggybank.storage.apachelog.CommonLogLoader();");
+        pig.registerQuery("A = LOAD '" + filename + "' USING org.apache.pig.piggybank.storage.apacheLOG.CommonLogLoader();");
         Iterator<?> it = pig.openIterator("A");
 
         int tupleCount = 0;
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthLoader.java	(working copy)
@@ -96,9 +96,7 @@
         pig.registerQuery(
             "data = load '" + dataDir + testFile + "' " +
             "using org.apache.pig.piggybank.storage.FixedWidthLoader(" +
-                "'-5, 9-21, 25-29, 33-40, 44, 48-52, 55-74, 78-82, 83-90', " + 
-                "'SKIP_HEADER', " + 
-                "'i: int, l: long, f: float, d: double, bit: int, b: boolean, dt: datetime, c_arr: chararray, b_arr: bytearray'" + 
+                "'-5, 9-21, 25-29, 33-40, 44, 48-52, 55-74, 78-82, 83-90', " +"'SKIP_HEADER', " +"'i: int, l: long, f: float, d: double, bit: int, b: boolean, dt: datetime, c_arr: chararray, b_arr: bytearray'" + 
             ");"
         );
 
@@ -119,9 +117,7 @@
     public void userSchemaFewerFieldsThanColumns() throws IOException, ParseException {
         pig.registerQuery(
             "data = load '" + dataDir + testFile + "' " +
-            "using org.apache.pig.piggybank.storage.FixedWidthLoader(" + 
-                "'-5, 9-21, 25-29, 33-40, 44, 48-52, 55-74, 78-82, 83-90', " + 
-                "'SKIP_HEADER', " +
+            "using org.apache.pig.piggybank.storage.FixedWidthLoader(" +"'-5, 9-21, 25-29, 33-40, 44, 48-52, 55-74, 78-82, 83-90', " +"'SKIP_HEADER', " +
                 "'i: int, l: long, f: float, d: double'" +
             ");"
         );
@@ -191,9 +187,7 @@
         pig.registerQuery(
             "data = load '" + dataDir + testFile + "' " +
             "using org.apache.pig.piggybank.storage.FixedWidthLoader(" +
-                "'-5, 9-21, 25-29 , 33 - 40, 44-44, 48-52, 55-74, 78-82, 83-90', " + 
-                "'SKIP_HEADER', " + 
-                "'i: int, l: long, f: float, d: double, bit: int, b: boolean, dt: datetime, c_arr: chararray, b_arr: bytearray'" + 
+                "'-5, 9-21, 25-29 , 33 - 40, 44-44, 48-52, 55-74, 78-82, 83-90', " +"'SKIP_HEADER', " +"'i: int, l: long, f: float, d: double, bit: int, b: boolean, dt: datetime, c_arr: chararray, b_arr: bytearray'" + 
             ");"
         );
 
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthStorer.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthStorer.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestFixedWidthStorer.java	(working copy)
@@ -79,8 +79,7 @@
             "data = load '" + dataDir + input + "' " +
             "using PigStorage('|') as (" + schema + ");"
         );
-        pig.store("data", dataDir + output, 
-                  "org.apache.pig.piggybank.storage.FixedWidthStorer('-5, 8-12, 15-19, 22-27, 29-33, 35-58, 62-69, 70-81', 'WRITE_HEADER')");
+        pig.store("data", dataDir + output,  "org.apache.pig.piggybank.storage.FixedWidthStorer('-5, 8-12, 15-19, 22-27, 29-33, 35-58, 62-69, 70-81', 'WRITE_HEADER')");
 
         // Load the output and see if it is what it ought to be
         pig.registerQuery(
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHadoopJobHistoryLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHadoopJobHistoryLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHadoopJobHistoryLoader.java	(working copy)
@@ -48,9 +48,7 @@
     @Test
     public void testHadoopJHLoader() throws Exception {
         PigServer pig = new PigServer(ExecType.LOCAL);
-        pig.registerQuery("a = load '" + INPUT_DIR 
-                + "' using org.apache.pig.piggybank.storage.HadoopJobHistoryLoader() " 
-                + "as (j:map[], m:map[], r:map[]);");
+        pig.registerQuery("a = load '" + INPUT_DIR+ "' using org.apache.pig.piggybank.storage.HadoopJobHistoryLoader() "+ "as (j:map[], m:map[], r:map[]);");
         Iterator<Tuple> iter = pig.openIterator("a");
         
         assertTrue(iter.hasNext());
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHelper.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHelper.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHelper.java	(working copy)
@@ -40,10 +40,7 @@
             Matcher matcher = pattern.matcher(string);
             matcher.groupCount();
             matcher.find();
-            DataByteArray[] toAdd = new DataByteArray[] { 
-                new DataByteArray(matcher.group(1)), 
-                new DataByteArray(matcher.group(2)), 
-                new DataByteArray(matcher.group(3)) };
+            DataByteArray[] toAdd = new DataByteArray[] {new DataByteArray(matcher.group(1)),new DataByteArray(matcher.group(2)),new DataByteArray(matcher.group(3)) };
             expected.add(toAdd);
         }
 
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHiveColumnarLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHiveColumnarLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestHiveColumnarLoader.java	(working copy)
@@ -440,8 +440,7 @@
 
             for (int monthIndex = 0; monthIndex < months; monthIndex++) {
 
-                File monthFile = new File(file, "month="
-                        + yearMonthDayHourcalendar.get(Calendar.MONTH));
+                File monthFile = new File(file, "month="+yearMonthDayHourcalendar.get(Calendar.MONTH));
                 monthFile.mkdir();
                 monthFile.deleteOnExit();
 
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestRegExLoader.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestRegExLoader.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestRegExLoader.java	(working copy)
@@ -66,8 +66,7 @@
         String filename = TestHelper.createTempFile(data, "");
         ArrayList<DataByteArray[]> expected = TestHelper.getExpected(data, pattern);
         
-        pigServer.registerQuery("A = LOAD '" + Util.encodeEscape(filename) + 
-                "' USING " + DummyRegExLoader.class.getName() + "() AS (key, val);");
+        pigServer.registerQuery("A = LOAD '" + Util.encodeEscape(filename) +"' USING " + DummyRegExLoader.class.getName() + "() AS (key, val);");
         Iterator<?> it = pigServer.openIterator("A");
         int tupleCount = 0;
         while (it.hasNext()) {
@@ -92,8 +91,7 @@
         dataE.add(new String[] { "3,three;iii" });
        	ArrayList<DataByteArray[]> expected = TestHelper.getExpected(dataE, pattern2);
         
-        pigServer.registerQuery("A = LOAD '" + Util.encodeEscape(filename) + 
-                "' USING " + DummyRegExLoader2.class.getName() + "() AS (key, val);");
+        pigServer.registerQuery("A = LOAD '" + Util.encodeEscape(filename) +"' USING " + DummyRegExLoader2.class.getName() + "() AS (key, val);");
         Iterator<?> it = pigServer.openIterator("A");
         int tupleCount = 0;
         while (it.hasNext()) {
Index: contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java
===================================================================
--- contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java	(revision 1579421)
+++ contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/avro/TestAvroStorage.java	(working copy)
@@ -83,8 +83,7 @@
         else {
             ArrayList<String> pathStrings = new ArrayList<String>();
             for (int index = 0; index < locations.length; index++) {
-                String f = System.getProperty("user.dir") + "/"
-                        + basedir + locations[index].trim();
+                String f = System.getProperty("user.dir") + "/"+basedir + locations[index].trim();
                 pathStrings.add(f);
             }
             return LoadFunc.join(pathStrings, ",");
Index: contrib/zebra/src/java/org/apache/hadoop/zebra/io/BasicTable.java
===================================================================
--- contrib/zebra/src/java/org/apache/hadoop/zebra/io/BasicTable.java	(revision 1579421)
+++ contrib/zebra/src/java/org/apache/hadoop/zebra/io/BasicTable.java	(working copy)
@@ -133,8 +133,7 @@
    *         
    */
   public static void dropColumnGroup(Path path, Configuration conf,
-                                     String cgName) 
-                                     throws IOException {
+                                     String cgName)                     throws IOException {
     
     FileSystem fs = FileSystem.get(conf);
     int triedCount = 0;
@@ -170,8 +169,7 @@
       // Clean up unfinished delete if it exists. so that clean up can 
       // complete if the previous deletion was interrupted for some reason.
       if (fs.exists(cgPath)) {
-        LOG.info(path + " : " + 
-                 " clearing unfinished deletion of column group " +
+        LOG.info(path + " : " + " clearing unfinished deletion of column group " +
                  cgName + ".");
         fs.delete(cgPath, true);
       }
@@ -202,8 +200,7 @@
       // already deleted CG. 
       SchemaFile tempSchema = new SchemaFile(path, null, conf);
       if (tempSchema.isCGDeleted(cgIdx)) {
-        LOG.info(path + " : " + cgName + 
-                 " is deleted by someone else. That is ok.");
+        LOG.info(path + " : " + cgName + " is deleted by someone else. That is ok.");
         return;
       }
       // otherwise, it is some other error.
@@ -248,9 +245,7 @@
           if (partition.isCGNeeded(i)) {
             if (isCGDeleted(i)) {
               // this is a deleted column group. Warn about it.
-              LOG.warn("Trying to read from deleted column group " + 
-                       schemaFile.getName(i) + 
-                       ". NULL is returned for corresponding columns. " +
+              LOG.warn("Trying to read from deleted column group " +       schemaFile.getName(i) +       ". NULL is returned for corresponding columns. " +
                        "Table at " + path);
             } else {
               cgTuples[i] = TypesUtils.createTuple(colGroups[i].getSchema());
@@ -930,8 +925,7 @@
         init(null, split, null, null, closeReader, partition);
       }
       
-      public BTScanner(RowSplit rowSplit,  boolean closeReader, 
-                       Partition partition) throws IOException {
+      public BTScanner(RowSplit rowSplit,  boolean closeReader,       Partition partition) throws IOException {
         init(rowSplit, null, null, null, closeReader, partition);
       }      
 
@@ -957,12 +951,8 @@
       }
     
       // Helper function for initialization.
-      private CGScanner createCGScanner(int cgIndex, CGRowSplit cgRowSplit, 
-                                           RangeSplit rangeSplit,
-                                           BytesWritable beginKey, 
-                                           BytesWritable endKey) 
-                      throws IOException, ParseException, 
-                             ParseException {        
+      private CGScanner createCGScanner(int cgIndex, CGRowSplit cgRowSplit,                           RangeSplit rangeSplit,
+                                           BytesWritable beginKey,                           BytesWritable endKey)      throws IOException, ParseException,             ParseException {        
         if (cgRowSplit != null) {
           return colGroups[cgIndex].getScanner(false, cgRowSplit);
         }      
@@ -970,8 +960,7 @@
           return colGroups[cgIndex].getScanner(beginKey, endKey, false);
         }
         return colGroups[cgIndex].getScanner
-                ((rangeSplit == null ? null : rangeSplit.getCGRangeSplit()), 
-                 false);
+                ((rangeSplit == null ? null : rangeSplit.getCGRangeSplit()), false);
       }
       
       /**
@@ -981,8 +970,7 @@
        * otherwise, these are based on keys.
        */
       private void init(RowSplit rowSplit, RangeSplit rangeSplit,
-                   BytesWritable beginKey, BytesWritable endKey, 
-                   boolean closeReader, Partition partition) throws IOException {
+                   BytesWritable beginKey, BytesWritable endKey,   boolean closeReader, Partition partition) throws IOException {
         this.partition = partition;
         boolean anyScanner = false;
         
@@ -1005,8 +993,7 @@
           }
           if (!anyScanner && firstValidCG >= 0) {
             // if no CG is needed explicitly by projection but the "countRow" still needs to access some column group
-            cgScanners[firstValidCG] = createCGScanner(firstValidCG, cgRowSplit, 
-                                                       rangeSplit,
+            cgScanners[firstValidCG] = createCGScanner(firstValidCG, cgRowSplit,                                       rangeSplit,
                                                        beginKey, endKey);            
           }
           this.closeReader = closeReader;
@@ -1301,11 +1288,9 @@
         sorted = schemaFile.isSorted();
         for (int nx = 0; nx < numCGs; nx++) {
           colGroups[nx] =
-              new ColumnGroup.Writer( 
-                 new Path(path, schemaFile.getName(nx)),
+              new ColumnGroup.Writer( new Path(path, schemaFile.getName(nx)),
             		 schemaFile.getPhysicalSchema(nx), 
-            		 sorted, 
-                 comparator,
+            		 sorted, comparator,
             		 schemaFile.getName(nx),
             		 schemaFile.getSerializer(nx), 
             		 schemaFile.getCompressor(nx), 
@@ -1488,8 +1473,7 @@
               {
                 if (!cgIndex.get(i).name.equals(firstCGIndex.get(i).name))
                   throw new IOException("File["+i+"] in Column Group "+colGroups[nx].path.getName()+
-                      " has a different name: "+cgIndex.get(i).name+" than " + 
-                      firstCGIndex.get(i).name + " in column group " + colGroups[first].path.getName());
+                      " has a different name: "+cgIndex.get(i).name+" than " +      firstCGIndex.get(i).name + " in column group " + colGroups[first].path.getName());
                 if (cgIndex.get(i).rows != firstCGIndex.get(i).rows)
                   throw new IOException("File "+cgIndex.get(i).name+"Column Group "+colGroups[nx].path.getName()+
                       " has a different number of rows, " + cgIndex.get(i).rows + ", than " +
@@ -1529,8 +1513,7 @@
         Path pathToRemove = new Path(actualOutputPath, "_temporary");
         if (fileSys.exists(pathToRemove)) {
             if(!fileSys.delete(pathToRemove, true)) {
-              LOG.error("Failed to delete the temporary output" + 
-                      " directory: " + pathToRemove.toString());            
+              LOG.error("Failed to delete the temporary output" +      " directory: " + pathToRemove.toString());            
             }
         }
     }    
@@ -2145,8 +2128,7 @@
           // print basic info for deleted column groups.
           out.printf("\nColum Group : DELETED");
           out.printf("\nName : %s", reader.schemaFile.getName(nx));
-          out.printf("\nSchema : %s\n", 
-                     reader.schemaFile.cgschemas[nx].getSchema().toString());
+          out.printf("\nSchema : %s\n",     reader.schemaFile.cgschemas[nx].getSchema().toString());
         }
       }
     }
Index: contrib/zebra/src/java/org/apache/hadoop/zebra/io/ColumnGroup.java
===================================================================
--- contrib/zebra/src/java/org/apache/hadoop/zebra/io/ColumnGroup.java	(revision 1579421)
+++ contrib/zebra/src/java/org/apache/hadoop/zebra/io/ColumnGroup.java	(working copy)
@@ -463,8 +463,7 @@
      * @param rowSplit specifies part index, start row, and end row.
      * @return A scanner object.
      */
-    public synchronized CGScanner getScanner(boolean closeReader, 
-                                                CGRowSplit rowSplit)
+    public synchronized CGScanner getScanner(boolean closeReader,                                CGRowSplit rowSplit)
                         throws IOException, ParseException {
       if (closed) {
         throw new EOFException("Reader already closed");
@@ -998,8 +997,7 @@
           namesInSplit[j] = paths[indexFirst+j].getName();
           sizesInSplit[j] = cgindex.get(cgindex.getFileIndex(paths[indexFirst+j])).bytes;
         }
-        cgRowSplit = new CGRowSplit(namesInSplit, sizesInSplit, length, 
-                startFirst, bytesFirst, bytesLast);
+        cgRowSplit = new CGRowSplit(namesInSplit, sizesInSplit, length,startFirst, bytesFirst, bytesLast);
         lst.add(cgRowSplit);
       }
       
@@ -1050,8 +1048,7 @@
       TFile.Reader.Scanner scanner;
       TupleReader tupleReader;
 
-      TFileScanner(FileSystem fs, Path path, CGRowSplit rowRange, 
-                    RawComparable begin, RawComparable end, boolean first, boolean last,
+      TFileScanner(FileSystem fs, Path path, CGRowSplit rowRange,    RawComparable begin, RawComparable end, boolean first, boolean last,
                     CGSchema cgschema, Projection projection,
           Configuration conf) throws IOException, ParseException {
         try {
@@ -1062,8 +1059,7 @@
           reader = new TFile.Reader(ins, fs.getFileStatus(path).getLen(), conf);
           if (rowRange != null && rowRange.startByteFirst != -1) {
             if (first && rowRange.startByteFirst != -1)
-              scanner = reader.createScannerByRecordNum(rowRange.startRowFirst, 
-                                              rowRange.startRowFirst + rowRange.numRowsFirst);
+              scanner = reader.createScannerByRecordNum(rowRange.startRowFirst,                              rowRange.startRowFirst + rowRange.numRowsFirst);
             else if (last && rowRange.numBytesLast != -1)
               scanner = reader.createScannerByRecordNum(0, rowRange.numRowsLast);
             else
@@ -1206,8 +1202,7 @@
         
         TFileScanner getTFileScanner() throws IOException {
           try {
-            return new TFileScanner(fs, path, rowRange, 
-                  begin, end, first, last, cgschema, logicalSchema, conf);
+            return new TFileScanner(fs, path, rowRange,  begin, end, first, last, cgschema, logicalSchema, conf);
           } catch (ParseException e) {
             throw new IOException(e.getMessage());
           }
@@ -1236,8 +1231,7 @@
        * @param rowRange see {@link CGRowSplit}
        * @param closeReader
        */
-      CGScanner(CGRowSplit rowRange, boolean closeReader) 
-                 throws IOException, ParseException {
+      CGScanner(CGRowSplit rowRange, boolean closeReader) throws IOException, ParseException {
         beginIndex = 0;
         endIndex = rowRange.length;
         init(rowRange, null, null, closeReader);
@@ -1259,8 +1253,7 @@
         init(null, beginKey, endKey, closeReader);
       }
 
-      private void init(CGRowSplit rowRange, RawComparable beginKey, 
-                        RawComparable endKey, boolean doClose) 
+      private void init(CGRowSplit rowRange, RawComparable beginKey,        RawComparable endKey, boolean doClose) 
              throws IOException, ParseException {
         this.rowRange = rowRange;
         if (beginIndex > endIndex) {
Index: contrib/zebra/src/java/org/apache/hadoop/zebra/pig/SchemaConverter.java
===================================================================
--- contrib/zebra/src/java/org/apache/hadoop/zebra/pig/SchemaConverter.java	(revision 1579421)
+++ contrib/zebra/src/java/org/apache/hadoop/zebra/pig/SchemaConverter.java	(working copy)
@@ -79,8 +79,7 @@
     throws FrontendException {
         Schema ret = new Schema();
         for (String col : tschema.getColumns()) {
-            org.apache.hadoop.zebra.schema.Schema.ColumnSchema columnSchema = 
-                tschema.getColumn(col);
+            org.apache.hadoop.zebra.schema.Schema.ColumnSchema columnSchema =tschema.getColumn(col);
             if (columnSchema != null) {
                 ColumnType ct = columnSchema.getType();
                 if (ct == org.apache.hadoop.zebra.schema.ColumnType.RECORD ||
@@ -103,12 +102,9 @@
             columnSchema = pschema.getField(i);
             if (columnSchema != null) {
                 if (DataType.isSchemaType(columnSchema.type))
-                    tschema.add(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(columnSchema.alias, 
-                            fromPigSchema(columnSchema.schema), toTableType(columnSchema.type)));
+                    tschema.add(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(columnSchema.alias,            fromPigSchema(columnSchema.schema), toTableType(columnSchema.type)));
                 else if (columnSchema.type == DataType.MAP)
-                    tschema.add(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(columnSchema.alias, 
-                            new org.apache.hadoop.zebra.schema.Schema(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(null, 
-                                    org.apache.hadoop.zebra.schema.ColumnType.BYTES)), toTableType(columnSchema.type)));
+                    tschema.add(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(columnSchema.alias,            new org.apache.hadoop.zebra.schema.Schema(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(null,                    org.apache.hadoop.zebra.schema.ColumnType.BYTES)), toTableType(columnSchema.type)));
                 else
                     tschema.add(new org.apache.hadoop.zebra.schema.Schema.ColumnSchema(columnSchema.alias, toTableType(columnSchema.type)));
             } else {
@@ -133,8 +129,7 @@
                 cSchema = new org.apache.hadoop.zebra.schema.Schema();
                 cSchema.add( new org.apache.hadoop.zebra.schema.Schema.ColumnSchema( "", ColumnType.BYTES ) );
             }
-            org.apache.hadoop.zebra.schema.Schema.ColumnSchema columnSchema = 
-                new org.apache.hadoop.zebra.schema.Schema.ColumnSchema( name, cSchema, type );
+            org.apache.hadoop.zebra.schema.Schema.ColumnSchema columnSchema =new org.apache.hadoop.zebra.schema.Schema.ColumnSchema( name, cSchema, type );
             schema.add( columnSchema );
         }
 
@@ -151,8 +146,7 @@
         ResourceFieldSchema[] rFields = new ResourceFieldSchema[fieldCount];
         for( int i = 0; i < fieldCount; i++ ) {
             org.apache.hadoop.zebra.schema.Schema.ColumnSchema cSchema = tSchema.getColumn( i );
-            if( cSchema != null ) 
-                rFields[i] = convertToResourceFieldSchema( cSchema );
+            if( cSchema != null )rFields[i] = convertToResourceFieldSchema( cSchema );
             else
                 rFields[i] = new ResourceFieldSchema();
         }
Index: contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java
===================================================================
--- contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java	(revision 1579421)
+++ contrib/zebra/src/java/org/apache/hadoop/zebra/pig/TableLoader.java	(working copy)
@@ -217,8 +217,7 @@
         if (requireGlobalOrder && !sorted)
           throw new IOException("Global sorting can be only asked on table loaded as sorted");
         if( sorted ) {
-            SplitMode splitMode = 
-                requireGlobalOrder ? SplitMode.GLOBALLY_SORTED : SplitMode.LOCALLY_SORTED;
+            SplitMode splitMode =requireGlobalOrder ? SplitMode.GLOBALLY_SORTED : SplitMode.LOCALLY_SORTED;
 
             Configuration conf = job.getConfiguration();
             conf.setBoolean(INPUT_SORT, true);
@@ -331,8 +330,7 @@
       */
      @Override
      public void setLocation(String location, Job job) throws IOException {
-         Properties properties = UDFContext.getUDFContext().getUDFProperties( 
-                 this.getClass(), new String[]{ udfContextSignature } );
+         Properties properties = UDFContext.getUDFContext().getUDFProperties( this.getClass(), new String[]{ udfContextSignature } );
 
          // Retrieve paths from UDFContext to avoid name node call in mapper.
          String pathString = properties.getProperty( UDFCONTEXT_PATHS_STRING );
@@ -385,8 +383,7 @@
 
      @Override
      public ResourceSchema getSchema(String location, Job job) throws IOException {
-         Properties properties = UDFContext.getUDFContext().getUDFProperties( 
-                 this.getClass(), new String[]{ udfContextSignature } );
+         Properties properties = UDFContext.getUDFContext().getUDFProperties( this.getClass(), new String[]{ udfContextSignature } );
          
          // Save the paths in UDFContext so that it can be retrieved in setLocation().
          Path[] paths = getPathsFromLocation( location, job );
@@ -432,8 +429,7 @@
          try {
         	 if(projectionString != null)
         		 TableInputFormat.setProjection(job, projectionString);        	         	 
-             Projection projection = new org.apache.hadoop.zebra.types.Projection( tableSchema, 
-                     TableInputFormat.getProjection( job ) );
+             Projection projection = new org.apache.hadoop.zebra.types.Projection( tableSchema,     TableInputFormat.getProjection( job ) );
              projectionSchema = projection.getProjectionSchema();
          } catch (ParseException e) {
              throw new IOException( "Schema parsing failed : "+ e.getMessage() );
@@ -480,51 +476,38 @@
             while( it.hasNext() ) {
                 RequiredField rField = (RequiredField) it.next();
                 ColumnSchema cs = projectionSchema.getColumn(rField.getIndex());
-                
-                if(cs == null) {
+               if(cs == null) {
                     throw new FrontendException
-                    ("Null column schema in projection schema in fieldsToRead at index " + rField.getIndex()); 
-                }
-                
-                if(cs.getType() != ColumnType.MAP && (rField.getSubFields() != null)) {        
-                    throw new FrontendException
-                    ("Zebra cannot have subfields for a non-map column type in fieldsToRead " + 
-                     "ColumnType:" + cs.getType() + " index in zebra projection schema: " + rField.getIndex()
+                    ("Null column schema in projection schema in fieldsToRead at index " + rField.getIndex());}
+               if(cs.getType() != ColumnType.MAP && (rField.getSubFields() != null)) {           throw new FrontendException
+                    ("Zebra cannot have subfields for a non-map column type in fieldsToRead " +     "ColumnType:" + cs.getType() + " index in zebra projection schema: " + rField.getIndex()
                     );
                 }
                 String name = cs.getName();
                 projectionStr = projectionStr + name ;
-                if(cs.getType() == ColumnType.MAP) {        
-                    List<RequiredField> subFields = rField.getSubFields();
-                    
-                    if( subFields != null ) {
-                    
-                        Iterator<RequiredField> its= subFields.iterator();
+                if(cs.getType() == ColumnType.MAP) {           List<RequiredField> subFields = rField.getSubFields();
+                       if( subFields != null ) {
+                           Iterator<RequiredField> its= subFields.iterator();
                         boolean flag = false;
                         if(its.hasNext()) {
                             flag = true;
                             projectionStr += "#" + "{";
-                        }    
-                        String tmp = "";
+                        }           String tmp = "";
                         while(its.hasNext()) {
-                            RequiredField sField = (RequiredField) its.next();    
-                            tmp = tmp + sField.getAlias();
+                            RequiredField sField = (RequiredField) its.next();               tmp = tmp + sField.getAlias();
                             if(its.hasNext()) {
                                 tmp = tmp + "|";
                             }
-                        }  
-                        if( flag ) {
+                        }         if( flag ) {
                             projectionStr = projectionStr + tmp + "}";
                         }
-                    }    
-                }
+                    }   }
                 if(it.hasNext()) {
                     projectionStr = projectionStr + " , ";
                 }
             }
             
-            Properties properties = UDFContext.getUDFContext().getUDFProperties( 
-                    this.getClass(), new String[]{ udfContextSignature } );
+            Properties properties = UDFContext.getUDFContext().getUDFProperties(    this.getClass(), new String[]{ udfContextSignature } );
             if( projectionStr != null && !projectionStr.isEmpty() )
                 properties.setProperty( UDFCONTEXT_PROJ_STRING, projectionStr );
 
Index: contrib/zebra/src/test/org/apache/hadoop/zebra/BaseTestCase.java
===================================================================
--- contrib/zebra/src/test/org/apache/hadoop/zebra/BaseTestCase.java	(revision 1579421)
+++ contrib/zebra/src/test/org/apache/hadoop/zebra/BaseTestCase.java	(working copy)
@@ -61,9 +61,9 @@
       }
     }
     
-    if (System.getenv("hadoop.log.dir") == null) {
+    if (System.getenv("hadoop.LOG.dir") == null) {
       String base = new File(".").getPath(); // getAbsolutePath();
-      System.setProperty("hadoop.log.dir", new Path(base).toString() + "./logs");
+      System.setProperty("hadoop.LOG.dir", new Path(base).toString() + "./logs");
     }
    
     String str = System.getenv("ZebraTestMode");
Index: contrib/zebra/src/test/org/apache/hadoop/zebra/mapred/TestBasicTableIOFormatLocalFS.java
===================================================================
--- contrib/zebra/src/test/org/apache/hadoop/zebra/mapred/TestBasicTableIOFormatLocalFS.java	(revision 1579421)
+++ contrib/zebra/src/test/org/apache/hadoop/zebra/mapred/TestBasicTableIOFormatLocalFS.java	(working copy)
@@ -112,11 +112,11 @@
 
   @Override
   protected void setUp() throws IOException {
-    if (System.getProperty("hadoop.log.dir") == null) {
+    if (System.getProperty("hadoop.LOG.dir") == null) {
       String base = new File(".").getPath(); // getAbsolutePath();
       Path path = new Path(".");
       System
-          .setProperty("hadoop.log.dir", new Path(base).toString() + "./logs");
+          .setProperty("hadoop.LOG.dir", new Path(base).toString() + "./logs");
     }
 
     if (options == null) {
Index: contrib/zebra/src/test/org/apache/hadoop/zebra/mapreduce/TestBasicTableIOFormatLocalFS.java
===================================================================
--- contrib/zebra/src/test/org/apache/hadoop/zebra/mapreduce/TestBasicTableIOFormatLocalFS.java	(revision 1579421)
+++ contrib/zebra/src/test/org/apache/hadoop/zebra/mapreduce/TestBasicTableIOFormatLocalFS.java	(working copy)
@@ -107,10 +107,10 @@
 
 	@Override
 	protected void setUp() throws IOException {
-		if (System.getProperty("hadoop.log.dir") == null) {
+		if (System.getProperty("hadoop.LOG.dir") == null) {
 			String base = new File(".").getPath(); // getAbsolutePath();
 			System
-			.setProperty("hadoop.log.dir", new Path(base).toString() + "./logs");
+			.setProperty("hadoop.LOG.dir", new Path(base).toString() + "./logs");
 		}
 
 		if (options == null) {
Index: contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestBasicUnion.java
===================================================================
--- contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestBasicUnion.java	(revision 1579421)
+++ contrib/zebra/src/test/org/apache/hadoop/zebra/pig/TestBasicUnion.java	(working copy)
@@ -699,8 +699,7 @@
 
     String[] exp1 = new String[] { "0_01", "1_01", "2_01", "3_01", "4_01", "5_01", "6_01", "7_01", "8_01", "9_01", 
     		                       "0_11", "1_11", "2_11", "3_11", "4_11", "5_11", "6_11", "7_11", "8_11", "9_11"};
-    String[] exp2 = new String[] { "0_00", "1_00", "2_00", "3_00", "4_00", "5_00", "6_00", "7_00", "8_00", "9_00", 
-                                   "0_10", "1_10", "2_10", "3_10", "4_10", "5_10", "6_10", "7_10", "8_10", "9_10"};
+    String[] exp2 = new String[] { "0_00", "1_00", "2_00", "3_00", "4_00", "5_00", "6_00", "7_00", "8_00", "9_00",                   "0_10", "1_10", "2_10", "3_10", "4_10", "5_10", "6_10", "7_10", "8_10", "9_10"};
     while (it.hasNext()) {
       count++;
       cur = it.next();
Index: lib-src/bzip2/org/apache/pig/bzip2r/Bzip2TextInputFormat.java
===================================================================
--- lib-src/bzip2/org/apache/pig/bzip2r/Bzip2TextInputFormat.java	(revision 1579421)
+++ lib-src/bzip2/org/apache/pig/bzip2r/Bzip2TextInputFormat.java	(working copy)
@@ -124,8 +124,7 @@
          * LineRecordReader.readLine() is depricated in HAdoop 0.17. So it is added here
          * locally.
          */
-        private long readLine(InputStream in, 
-                OutputStream out) throws IOException {
+        private long readLine(InputStream in,OutputStream out) throws IOException {
             long bytes = 0;
             while (true) {
                 int b = -1;
@@ -241,8 +240,7 @@
     @Override
     public RecordReader createRecordReader(InputSplit split,
             TaskAttemptContext context) throws IOException, InterruptedException {
-        return new BZip2LineRecordReader(context.getConfiguration(), 
-                (FileSplit) split);
+        return new BZip2LineRecordReader(context.getConfiguration(),(FileSplit) split);
     }
 
 }
Index: lib-src/bzip2/org/apache/tools/bzip2r/CBZip2InputStream.java
===================================================================
--- lib-src/bzip2/org/apache/tools/bzip2r/CBZip2InputStream.java	(revision 1579421)
+++ lib-src/bzip2/org/apache/tools/bzip2r/CBZip2InputStream.java	(working copy)
@@ -289,8 +289,7 @@
             magic2 = bsGetUChar();
             magic3 = bsGetUChar();
             magic4 = bsGetUChar();
-            if (magic1 != 'B' || magic2 != 'Z' || 
-                    magic3 != 'h' || magic4 < '1' || magic4 > '9') {
+            if (magic1 != 'B' || magic2 != 'Z' ||    magic3 != 'h' || magic4 < '1' || magic4 > '9') {
                 bsFinishedWithStream();
                 streamEnd = true;
                 return;
@@ -346,8 +345,7 @@
                 magic <<= 1;
                 magic &= mask;
                 magic |= bsR(1);
-                // if we just found the block header, the beginning of the bzip 
-                // header would be 6 bytes before the current stream position
+                // if we just found the block header, the beginning of the bzip// header would be 6 bytes before the current stream position
                 // when we eventually break from this while(), if it is because
                 // we found a block header then pos will have the correct start
                 // of header position
@@ -400,8 +398,7 @@
 
     private void complete() throws IOException {
         storedCombinedCRC = bsGetInt32();
-        if (checkComputedCombinedCRC && 
-                storedCombinedCRC != computedCombinedCRC) {
+        if (checkComputedCombinedCRC &&storedCombinedCRC != computedCombinedCRC) {
             crcError();
         }
         if (innerBsStream.getPos() < endOffsetOfSplit) {
Index: lib-src/bzip2/org/apache/tools/bzip2r/CBZip2OutputStream.java
===================================================================
--- lib-src/bzip2/org/apache/tools/bzip2r/CBZip2OutputStream.java	(revision 1579421)
+++ lib-src/bzip2/org/apache/tools/bzip2r/CBZip2OutputStream.java	(working copy)
@@ -93,7 +93,7 @@
  */
 public class CBZip2OutputStream extends OutputStream implements BZip2Constants {
 
-    private final static Log log = LogFactory.getLog(CBZip2OutputStream.class);
+    private final static Log LOG = LogFactory.getLog(CBZip2OutputStream.class);
     
     protected static final int SETMASK = (1 << 21);
     protected static final int CLEARMASK = (~SETMASK);
@@ -113,7 +113,7 @@
     protected static final int QSORT_STACK_SIZE = 1000;
 
     private static void panic() {
-        log.info("panic");
+        LOG.info("panic");
         //throw new CError();
     }
 
Index: shims/src/hadoop20/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
===================================================================
--- shims/src/hadoop20/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java	(revision 1579421)
+++ shims/src/hadoop20/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java	(working copy)
@@ -76,8 +76,7 @@
                 org.apache.hadoop.mapreduce.Job nwJob = new org.apache.hadoop.mapreduce.Job(job.getJobConf());
                 sortComparator = nwJob.getSortComparator();
                 groupingComparator = nwJob.getGroupingComparator();
-                
-                Collections.sort(input, new Comparator<Pair<PigNullableWritable, Writable>>() {
+               Collections.sort(input, new Comparator<Pair<PigNullableWritable, Writable>>() {
                         @Override
                         public int compare(Pair<PigNullableWritable, Writable> o1,
                                            Pair<PigNullableWritable, Writable> o2) {
Index: shims/src/hadoop20/org/apache/pig/backend/hadoop20/PigJobControl.java
===================================================================
--- shims/src/hadoop20/org/apache/pig/backend/hadoop20/PigJobControl.java	(revision 1579421)
+++ shims/src/hadoop20/org/apache/pig/backend/hadoop20/PigJobControl.java	(working copy)
@@ -35,7 +35,7 @@
  *
  */
 public class PigJobControl extends JobControl {
-  private static final Log log = LogFactory.getLog(PigJobControl.class);
+  private static final Log LOG = LogFactory.getLog(PigJobControl.class);
 
   private static Field runnerState;
 
@@ -59,7 +59,7 @@
       startReadyJobs.setAccessible(true);
       initSuccesful = true;
     } catch (Exception e) {
-      log.warn("falling back to default JobControl (not using hadoop 0.20 ?)", e);
+      LOG.warn("falling back to default JobControl (not using hadoop 0.20 ?)", e);
       initSuccesful = false;
     }
   }
Index: shims/src/hadoop23/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java
===================================================================
--- shims/src/hadoop23/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java	(revision 1579421)
+++ shims/src/hadoop23/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigMapReduce.java	(working copy)
@@ -118,8 +118,7 @@
                 org.apache.hadoop.mapreduce.Job nwJob = new org.apache.hadoop.mapreduce.Job(job.getJobConf());
                 sortComparator = nwJob.getSortComparator();
                 groupingComparator = nwJob.getGroupingComparator();
-                
-                Collections.sort(input, new Comparator<Pair<PigNullableWritable, Writable>>() {
+               Collections.sort(input, new Comparator<Pair<PigNullableWritable, Writable>>() {
                         @Override
                         public int compare(Pair<PigNullableWritable, Writable> o1,
                                            Pair<PigNullableWritable, Writable> o2) {
@@ -148,8 +147,7 @@
             }
             
             public class IllustratorValueIterator implements ReduceContext.ValueIterator<NullableTuple> {
-                
-                private int pos = -1;
+               private int pos = -1;
                 private int mark = -1;
 
                 @Override
Index: shims/src/hadoop23/org/apache/pig/backend/hadoop23/PigJobControl.java
===================================================================
--- shims/src/hadoop23/org/apache/pig/backend/hadoop23/PigJobControl.java	(revision 1579421)
+++ shims/src/hadoop23/org/apache/pig/backend/hadoop23/PigJobControl.java	(working copy)
@@ -36,7 +36,7 @@
  *
  */
 public class PigJobControl extends JobControl {
-  private static final Log log = LogFactory.getLog(PigJobControl.class);
+  private static final Log LOG = LogFactory.getLog(PigJobControl.class);
 
   private static Field runnerState;
   private static Field jobsInProgress;
@@ -72,7 +72,7 @@
 
       initSuccesful = true;
     } catch (Exception e) {
-      log.warn("falling back to default JobControl (not using hadoop 0.23 ?)", e);
+      LOG.warn("falling back to default JobControl (not using hadoop 0.23 ?)", e);
       initSuccesful = false;
     }
   }
@@ -176,7 +176,7 @@
           Iterator<ControlledJob> it = getJobs(jobsInProgress).iterator();
           while(it.hasNext()) {
             ControlledJob j = it.next();
-            log.debug("Checking state of job "+j);
+            LOG.debug("Checking state of job "+j);
             switch(checkState(j)) {
             case SUCCESS:
               getJobs(successfulJobs).add(j);
@@ -214,7 +214,7 @@
         }
       }
     }catch(Throwable t) {
-      log.error("Error while trying to run jobs.",t);
+      LOG.error("Error while trying to run jobs.",t);
       //Mark all jobs as failed because we got something bad.
       failAllJobs(t);
     }
Index: shims/test/hadoop20/org/apache/pig/test/MiniCluster.java
===================================================================
--- shims/test/hadoop20/org/apache/pig/test/MiniCluster.java	(revision 1579421)
+++ shims/test/hadoop20/org/apache/pig/test/MiniCluster.java	(working copy)
@@ -37,7 +37,7 @@
     @Override
     protected void setupMiniDfsAndMrClusters() {
         try {
-            System.setProperty("hadoop.log.dir", "build/test/logs");
+            System.setProperty("hadoop.LOG.dir", "build/test/logs");
             final int dataNodes = 4;     // There will be 4 data nodes
             final int taskTrackers = 4;  // There will be 4 task tracker nodes
 
Index: shims/test/hadoop23/org/apache/pig/test/MiniCluster.java
===================================================================
--- shims/test/hadoop23/org/apache/pig/test/MiniCluster.java	(revision 1579421)
+++ shims/test/hadoop23/org/apache/pig/test/MiniCluster.java	(working copy)
@@ -59,7 +59,7 @@
             final int dataNodes = 4;     // There will be 4 data nodes
             final int taskTrackers = 4;  // There will be 4 task tracker nodes
 
-            System.setProperty("hadoop.log.dir", "build/test/logs");
+            System.setProperty("hadoop.LOG.dir", "build/test/logs");
             // Create the dir that holds hadoop-site.xml file
             // Delete if hadoop-site.xml exists already
             CONF_DIR.mkdirs();
Index: src/org/apache/pig/Main.java
===================================================================
--- src/org/apache/pig/Main.java	(revision 1579421)
+++ src/org/apache/pig/Main.java	(working copy)
@@ -38,6 +38,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Scanner;
 import java.util.jar.Attributes;
 import java.util.jar.JarFile;
 import java.util.jar.Manifest;
@@ -119,7 +120,7 @@
                 final Map<String,Attributes> attrs = manifest.getEntries();
                 attr = attrs.get("org/apache/pig");
             } else {
-                log.info("Unable to read pigs manifest file as we are not running from a jar, version information unavailable");
+                //log.info("Unable to read pigs manifest file as we are not running from a jar, version information unavailable");
             }
         } catch (Exception e) {
             log.warn("Unable to read pigs manifest file, version information unavailable", e);
@@ -214,7 +215,17 @@
             ExecMode mode = ExecMode.UNKNOWN;
             String file = null;
             String engine = null;
-
+            
+            /*Scanner scanner = new Scanner(new File("pivot"));
+    		int [] tall = new int [1];
+    		int idx = 0;
+    		while(scanner.hasNextInt())
+    		{
+    		     tall[idx++] = scanner.nextInt();
+    		}
+        	
+        	System.out.println(tall[0]);*/
+           
             // set up client side system properties in UDF context
             UDFContext.getUDFContext().setClientSystemProps(properties);
 
@@ -380,13 +391,13 @@
             pigContext.getProperties().setProperty("pig.logfile", (logFileName == null? "": logFileName));
 
             // configure logging
-            configureLog4J(properties, pigContext);
+            //configureLog4J(properties, pigContext);
 
-            log.info(getVersionString().replace("\n", ""));
+            ////log.info(getVersionString().replace("\n", ""));
 
-            if(logFileName != null) {
-                log.info("Logging error messages to: " + logFileName);
-            }
+            //if(logFileName != null) {
+              //  //log.info("Logging error messages to: " + logFileName);
+            //}
 
             deleteTempFiles = Boolean.valueOf(properties.getProperty(
                     PigConfiguration.PIG_DELETE_TEMP_FILE, "true"));
@@ -413,6 +424,7 @@
             switch (mode) {
 
             case FILE: {
+            	System.out.println("FILE");
                 String remainders[] = opts.getRemainingArgs();
                 if (remainders != null) {
                     pigContext.getProperties().setProperty(PigContext.PIG_CMD_ARGS_REMAINDERS,
@@ -441,18 +453,18 @@
                 substFile = file + ".substituted";
 
                 pin = runParamPreprocessor(pigContext, in, substFile, debug || dryrun || checkScriptOnly);
-                if (dryrun) {
+               /* if (dryrun) {
                     if (dryrun(substFile, pigContext)) {
-                        log.info("Dry run completed. Substituted pig script is at "
+                        //log.info("Dry run completed. Substituted pig script is at "
                                 + substFile
                                 + ". Expanded pig script is at "
                                 + file + ".expanded");
                     } else {
-                        log.info("Dry run completed. Substituted pig script is at "
+                        //log.info("Dry run completed. Substituted pig script is at "
                                     + substFile);
                     }
                     return ReturnCode.SUCCESS;
-                }
+                }*/
 
 
                 logFileName = validateLogFile(logFileName, file);
@@ -470,7 +482,6 @@
 
                 grunt = new Grunt(pin, pigContext);
                 gruntCalled = true;
-
                 if(checkScriptOnly) {
                     grunt.checkScript(substFile);
                     System.err.println(file + " syntax OK");
@@ -484,6 +495,7 @@
             }
 
             case STRING: {
+            	System.out.println("STRING");
                 if(checkScriptOnly) {
                     System.err.println("ERROR:" +
                             "-c (-check) option is only valid " +
@@ -570,18 +582,18 @@
                 // run parameter substitution preprocessor first
                 substFile = remainders[0] + ".substituted";
                 pin = runParamPreprocessor(pigContext, in, substFile, debug || dryrun || checkScriptOnly);
-                if (dryrun) {
+                /*if (dryrun) {
                     if (dryrun(substFile, pigContext)) {
-                        log.info("Dry run completed. Substituted pig script is at "
+                        //log.info("Dry run completed. Substituted pig script is at "
                                 + substFile
                                 + ". Expanded pig script is at "
                                 + remainders[0] + ".expanded");
                     } else {
-                        log.info("Dry run completed. Substituted pig script is at "
+                        //log.info("Dry run completed. Substituted pig script is at "
                                 + substFile);
                     }
                     return ReturnCode.SUCCESS;
-                }
+                }*/
 
                 logFileName = validateLogFile(logFileName, remainders[0]);
                 pigContext.getProperties().setProperty("pig.logfile", (logFileName == null? "": logFileName));
@@ -928,8 +940,8 @@
                     scriptFileAbsPath = scriptFile.getCanonicalPath();
                     strippedDownScriptName = getFileFromCanonicalPath(scriptFileAbsPath);
                 } catch (IOException ioe) {
-                    log.warn("Could not compute canonical path to the script file " + ioe.getMessage());
-                    strippedDownScriptName = null;
+                    //log.warn("Could not compute canonical path to the script file " + ioe.getMessage());
+                    //strippedDownScriptName = null;
                 }
             }
         }
@@ -947,12 +959,12 @@
                     try {
                         logFileName = logFile.getCanonicalPath() + File.separator + defaultLogFileName;
                     } catch (IOException ioe) {
-                        log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
+                        //log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                         return null;
                     }
                     return logFileName;
                 } else {
-                    log.warn("Need write permission in the directory: " + logFileName + " to create log file.");
+                    //log.warn("Need write permission in the directory: " + logFileName + " to create log file.");
                     return null;
                 }
             } else {
@@ -964,7 +976,7 @@
                         try {
                             logFileName = new File(logFileName).getCanonicalPath();
                         } catch (IOException ioe) {
-                            log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
+                            //log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                             return null;
                         }
                         return logFileName;
@@ -971,7 +983,7 @@
                     } else {
                         //do not have write permissions for the log file
                         //bail out with an error message
-                        log.warn("Cannot write to file: " + logFileName + ". Need write permission.");
+                        //log.warn("Cannot write to file: " + logFileName + ". Need write permission.");
                         return logFileName;
                     }
                 } else {
@@ -983,12 +995,12 @@
                             try {
                                 logFileName = new File(logFileName).getCanonicalPath();
                             } catch (IOException ioe) {
-                                log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
+                                //log.warn("Could not compute canonical path to the log file " + ioe.getMessage());
                                 return null;
                             }
                             return logFileName;
                         } else {
-                            log.warn("Need write permission in the directory: " + logFile + " to create log file.");
+                            //log.warn("Need write permission in the directory: " + logFile + " to create log file.");
                             return logFileName;
                         }
                     }//end if logFile != null else is the default in fall through
@@ -1004,7 +1016,7 @@
         if(logFile.canWrite()) {
             return logFileName;
         }
-        log.warn("Cannot write to log file: " + logFileName);
+        //log.warn("Cannot write to log file: " + logFileName);
         return null;
     }
 
@@ -1019,7 +1031,7 @@
 
     private static int runEmbeddedScript(PigContext pigContext, String file, String engine)
     throws IOException {
-        log.info("Run embedded script: " + engine);
+        ////log.info("Run embedded script: " + engine);
         pigContext.connect();
         ScriptEngine scriptEngine = ScriptEngine.getInstance(engine);
         Map<String, List<PigStats>> statsMap = scriptEngine.run(pigContext, file);
Index: src/org/apache/pig/PigServer.java
===================================================================
--- src/org/apache/pig/PigServer.java	(revision 1579421)
+++ src/org/apache/pig/PigServer.java	(working copy)
@@ -993,10 +993,10 @@
             throw new IOException("Couldn't retrieve job.");
         }
         OutputStats output = stats.getOutputStats().get(0);
-
+        System.out.println(output.getName().toString());
         if(stats.isSuccessful()){
-            return  new HJob(JOB_STATUS.COMPLETED, pigContext, output
-                    .getPOStore(), output.getAlias(), stats);
+        	System.out.println("JOB COMPLETED!!!!!");
+            return  new HJob(JOB_STATUS.COMPLETED, pigContext, output.getPOStore(), output.getAlias(), stats);
         }else{
             HJob job = new HJob(JOB_STATUS.FAILED, pigContext,
                     output.getPOStore(), output.getAlias(), stats);
@@ -1316,8 +1316,8 @@
         }
 
     }
+    private void buildStorePlan(String alias) throws IOException {
 
-    private void buildStorePlan(String alias) throws IOException {
         currDAG.parseQuery();
         currDAG.buildPlan( alias );
 
Index: src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/HExecutionEngine.java	(working copy)
@@ -27,8 +27,8 @@
 import java.util.Map;
 import java.util.Properties;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.mapred.JobConf;
@@ -77,7 +77,7 @@
 
 public abstract class HExecutionEngine implements ExecutionEngine {
 
-    private static final Log LOG = LogFactory.getLog(HExecutionEngine.class);
+    //private static final Log LOG = LogFactory.getLog(HExecutionEngine.class);
 
     public static final String HADOOP_SITE = "hadoop-site.xml";
     public static final String CORE_SITE = "core-site.xml";
@@ -226,14 +226,12 @@
             properties.setProperty(FILE_SYSTEM_LOCATION, nameNode);
         }
 
-        LOG.info("Connecting to hadoop file system at: "
-                + (nameNode == null ? LOCAL : nameNode));
+        //log.info("Connecting to hadoop file system at: "+ (nameNode == null ? LOCAL : nameNode));
         // constructor sets DEFAULT_REPLICATION_FACTOR_KEY
         ds = new HDataStorage(properties);
 
         if (cluster != null && !cluster.equalsIgnoreCase(LOCAL)) {
-            LOG.info("Connecting to map-reduce job tracker at: "
-                    + jc.get(JOB_TRACKER_LOCATION));
+            //log.info("Connecting to map-reduce job tracker at: "+ jc.get(JOB_TRACKER_LOCATION));
         }
     }
 
@@ -256,15 +254,12 @@
         UidResetter uidResetter = new UidResetter(plan);
         uidResetter.visit();
 
-        SchemaResetter schemaResetter = new SchemaResetter(plan,
-                true /* skip duplicate uid check*/);
+        SchemaResetter schemaResetter = new SchemaResetter(plan, true /* skip duplicate uid check*/);
         schemaResetter.visit();
 
         HashSet<String> disabledOptimizerRules;
         try {
-            disabledOptimizerRules = (HashSet<String>) ObjectSerializer
-                    .deserialize(pigContext.getProperties().getProperty(
-                            PigImplConstants.PIG_OPTIMIZER_RULES_KEY));
+            disabledOptimizerRules = (HashSet<String>) ObjectSerializer.deserialize(pigContext.getProperties().getProperty(PigImplConstants.PIG_OPTIMIZER_RULES_KEY));
         } catch (IOException ioe) {
             int errCode = 2110;
             String msg = "Unable to deserialize optimizer rules.";
@@ -274,11 +269,9 @@
             disabledOptimizerRules = new HashSet<String>();
         }
 
-        String pigOptimizerRulesDisabled = this.pigContext.getProperties()
-                .getProperty(PigConstants.PIG_OPTIMIZER_RULES_DISABLED_KEY);
+        String pigOptimizerRulesDisabled = this.pigContext.getProperties().getProperty(PigConstants.PIG_OPTIMIZER_RULES_DISABLED_KEY);
         if (pigOptimizerRulesDisabled != null) {
-            disabledOptimizerRules.addAll(Lists.newArrayList((Splitter.on(",")
-                    .split(pigOptimizerRulesDisabled))));
+            disabledOptimizerRules.addAll(Lists.newArrayList((Splitter.on(",").split(pigOptimizerRulesDisabled))));
         }
 
         if (pigContext.inIllustrator) {
@@ -292,6 +285,7 @@
             disabledOptimizerRules.add("ColumnMapKeyPrune");
             disabledOptimizerRules.add("AddForEach");
             disabledOptimizerRules.add("GroupByConstParallelSetter");
+            disabledOptimizerRules.add("RollupH2IRGOptimizer");
         }
 
         StoreAliasSetter storeAliasSetter = new StoreAliasSetter(plan);
@@ -298,8 +292,7 @@
         storeAliasSetter.visit();
 
         // run optimizer
-        LogicalPlanOptimizer optimizer = new LogicalPlanOptimizer(plan, 100,
-                disabledOptimizerRules);
+        LogicalPlanOptimizer optimizer = new LogicalPlanOptimizer(plan, 100, disabledOptimizerRules);
         optimizer.optimize();
 
         // compute whether output data is sorted or not
@@ -310,8 +303,7 @@
             // Validate input/output file. Currently no validation framework in
             // new logical plan, put this validator here first.
             // We might decide to move it out to a validator framework in future
-            LogicalRelationalNodeValidator validator = new InputOutputFileValidator(
-                    plan, pigContext);
+            LogicalRelationalNodeValidator validator = new InputOutputFileValidator(plan, pigContext);
             validator.validate();
 
             // Check for blacklist and whitelist properties and disable
@@ -369,7 +361,7 @@
             //skipped; a SimpleFetchPigStats will be returned through which the result
             //can be directly fetched from the underlying storage
             if (FetchOptimizer.isPlanFetchable(pc, pp)) {
-                return new FetchLauncher(pc).launchPig(pp);
+            	return new FetchLauncher(pc).launchPig(pp);
             }
             return launcher.launchPig(pp, grpName, pigContext);
         } catch (ExecException e) {
Index: src/org/apache/pig/backend/hadoop/executionengine/HJob.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/HJob.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/HJob.java	(working copy)
@@ -22,8 +22,8 @@
 import java.util.Iterator;
 import java.util.Properties;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.pig.LoadFunc;
 import org.apache.pig.PigException;
 import org.apache.pig.backend.executionengine.ExecException;
@@ -39,7 +39,7 @@
 
 public class HJob implements ExecJob {
 
-    private final Log log = LogFactory.getLog(getClass());
+    //private final Log log = LogFactory.getLog(getClass());
     
     protected JOB_STATUS status;
     protected PigContext pigContext;
@@ -112,7 +112,7 @@
                     if (t == null)
                         atEnd = true;
                 } catch (Exception e) {
-                    log.error(e);
+                    //log.error(e);
                     t = null;
                     atEnd = true;
                     throw new Error(e);
@@ -129,7 +129,7 @@
                 try {
                     next = p.getNext();
                 } catch (Exception e) {
-                    log.error(e);
+                    //log.error(e);
                 }
                 if (next == null)
                     atEnd = true;
Index: src/org/apache/pig/backend/hadoop/executionengine/Launcher.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/Launcher.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/Launcher.java	(working copy)
@@ -28,8 +28,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobClient;
 import org.apache.hadoop.mapred.JobID;
@@ -63,7 +63,7 @@
  *
  */
 public abstract class Launcher {
-    private static final Log log = LogFactory.getLog(Launcher.class);
+    //private static final Log log = LogFactory.getLog(Launcher.class);
     private static final String OOM_ERR = "OutOfMemoryError";
     private boolean pigException = false;
     private boolean outOfMemory = false;
@@ -86,7 +86,7 @@
             try {
                 kill();
             } catch (Exception e) {
-                log.warn("Error in killing Execution Engine: " + e);
+                //log.warn("Error in killing Execution Engine: " + e);
             }
         }
     }
@@ -190,10 +190,7 @@
                             // errNotDbg is used only for failed jobs
                             // keep track of all the unique exceptions
                             try {
-                                LogUtils.writeLog("Backend error message",
-                                        msgs[j], pigContext.getProperties()
-                                                .getProperty("pig.logfile"),
-                                        log);
+                                //LogUtils.writeLog("Backend error message",msgs[j], pigContext.getProperties().getProperty("pig.logfile"),log);
                                 Exception e = getExceptionFromString(msgs[j]);
                                 exceptions.add(e);
                             } catch (Exception e1) {
@@ -201,8 +198,7 @@
 
                             }
                         } else {
-                            log.debug("Error message from task (" + type + ") "
-                                    + reports[i].getTaskID() + msgs[j]);
+                            //log.debug("Error message from task (" + type + ") "+ reports[i].getTaskID() + msgs[j]);
                         }
                     }
                 }
@@ -227,9 +223,7 @@
                     for (int j = 0; j < exceptions.size(); ++j) {
                         String headerMessage = "Error message from task ("
                                 + type + ") " + reports[i].getTaskID();
-                        LogUtils.writeLog(exceptions.get(j), pigContext
-                                .getProperties().getProperty("pig.logfile"),
-                                log, false, headerMessage, false, false);
+                        //LogUtils.writeLog(exceptions.get(j), pigContext.getProperties().getProperty("pig.logfile"),log, false, headerMessage, false, false);
                     }
                     throw exceptions.get(0);
                 } else if (exceptions.size() == 1) {
Index: src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchOptimizer.java	(working copy)
@@ -23,8 +23,8 @@
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.pig.PigConfiguration;
 import org.apache.pig.backend.datastorage.DataStorageException;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PhyPlanSetter;
@@ -72,7 +72,7 @@
  *
  */
 public class FetchOptimizer {
-    private static final Log LOG = LogFactory.getLog(FetchOptimizer.class);
+    //private static final Log LOG = LogFactory.getLog(FetchOptimizer.class);
 
     /**
      * Checks whether the fetch is enabled
@@ -132,7 +132,7 @@
             if (!(po instanceof POLoad)) {
                 String msg = "Expected physical operator at root is POLoad. Found : "
                         + po.getClass().getCanonicalName() + ". Fetch optimizer will be disabled.";
-                LOG.debug(msg);
+                //LOG.debug(msg);
                 return false;
             }
         }
@@ -140,7 +140,7 @@
         //consider single leaf jobs only
         int leafSize = pp.getLeaves().size();
         if (pp.getLeaves().size() != 1) {
-            LOG.debug("Expected physical plan should have one leaf. Found " + leafSize);
+            //LOG.debug("Expected physical plan should have one leaf. Found " + leafSize);
             return false;
         }
 
Index: src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchProgressableReporter.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchProgressableReporter.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/fetch/FetchProgressableReporter.java	(working copy)
@@ -27,7 +27,7 @@
  */
 public class FetchProgressableReporter implements PigProgressable {
 
-    private static final Log LOG = LogFactory.getLog(FetchProgressableReporter.class);
+    //private static final Log LOG = LogFactory.getLog(FetchProgressableReporter.class);
 
     public void progress() {
 
@@ -34,7 +34,7 @@
     }
 
     public void progress(String msg) {
-        LOG.info(msg);
+        //LOG.info(msg);
     }
 
 }
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/CombinerOptimizer.java	(working copy)
@@ -25,7 +25,6 @@
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
-
 import org.apache.pig.PigException;
 import org.apache.pig.FuncSpec;
 import org.apache.pig.PigWarning;
@@ -42,6 +41,7 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PODistinct;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORollupH2IRGForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
@@ -114,7 +114,7 @@
 
     @Override
     public void visitMROp(MapReduceOper mr) throws VisitorException {
-        log.trace("Entering CombinerOptimizer.visitMROp");
+        //LOG.trace("Entering CombinerOptimizer.visitMROp");
         if (mr.reducePlan.isEmpty()) return;
 
         // part one - check if this MR job represents a group-by + foreach
@@ -182,7 +182,7 @@
                 return;
             }
 
-            log.info("Choosing to move algebraic foreach to combiner");
+            //LOG.info("Choosing to move algebraic foreach to combiner");
 
             try {
 
@@ -324,6 +324,162 @@
                 throw new OptimizerException(msg, errCode, PigException.BUG, e);
             }
         }
+        
+        if (successor instanceof PORollupH2IRGForEach) {
+            PORollupH2IRGForEach hforeach = (PORollupH2IRGForEach) successor;
+            List<PhysicalPlan> feInners = hforeach.getInputPlans();
+
+            // find algebraic operators and also check if the foreach statement
+            // is suitable for combiner use
+            List<Pair<PhysicalOperator, PhysicalPlan>> algebraicOps = findAlgebraicOps(feInners);
+            if (algebraicOps == null || algebraicOps.size() == 0) {
+                // the plan is not combinable or there is nothing to combine
+                // we're done
+                return;
+            }
+            if (mr.combinePlan.getRoots().size() != 0) {
+                messageCollector.collect("Wasn't expecting to find anything already " + "in the combiner!",
+                        MessageType.Warning, PigWarning.NON_EMPTY_COMBINE_PLAN);
+                return;
+            }
+
+            log.info("Choosing to move algebraic foreach to combiner");
+
+            try {
+
+                // replace PODistinct->Project[*] with distinct udf (which is Algebriac)
+                for (Pair<PhysicalOperator, PhysicalPlan> op2plan : algebraicOps) {
+                    if (!(op2plan.first instanceof PODistinct))
+                        continue;
+                    DistinctPatcher distinctPatcher = new DistinctPatcher(op2plan.second);
+                    distinctPatcher.visit();
+                    if (distinctPatcher.getDistinct() == null) {
+                        int errCode = 2073;
+                        String msg = "Problem with replacing distinct operator with distinct built-in function.";
+                        throw new PlanException(msg, errCode, PigException.BUG);
+                    }
+                    op2plan.first = distinctPatcher.getDistinct();
+                }
+
+                // create new map foreach
+                POForEach mfe = createForEachWithGrpProj(hforeach, rearrange.getKeyType());
+                Map<PhysicalOperator, Integer> op2newpos = new HashMap<PhysicalOperator, Integer>();
+                Integer pos = 1;
+                // create plan for each algebraic udf and add as inner plan in map-foreach
+                for (Pair<PhysicalOperator, PhysicalPlan> op2plan : algebraicOps) {
+                    PhysicalPlan udfPlan = createPlanWithPredecessors(op2plan.first, op2plan.second);
+                    mfe.addInputPlan(udfPlan, false);
+                    op2newpos.put(op2plan.first, pos++);
+                }
+                changeFunc(mfe, POUserFunc.INITIAL);
+
+                // since we will only be creating SingleTupleBag as input to
+                // the map foreach, we should flag the POProjects in the map
+                // foreach inner plans to also use SingleTupleBag
+                for (PhysicalPlan mpl : mfe.getInputPlans()) {
+                    try {
+                        new fixMapProjects(mpl).visit();
+                    } catch (VisitorException e) {
+                        int errCode = 2089;
+                        String msg = "Unable to flag project operator to use single tuple bag.";
+                        throw new PlanException(msg, errCode, PigException.BUG, e);
+                    }
+                }
+
+                // create new combine foreach
+                POForEach cfe = createForEachWithGrpProj(hforeach, rearrange.getKeyType());
+                // add algebraic functions with appropriate projection
+                addAlgebraicFuncToCombineFE(cfe, op2newpos);
+                changeFunc(cfe, POUserFunc.INTERMEDIATE);
+
+                // fix projection and function time for algebraic functions in reduce foreach
+                for (Pair<PhysicalOperator, PhysicalPlan> op2plan : algebraicOps) {
+                    setProjectInput(op2plan.first, op2plan.second, op2newpos.get(op2plan.first));
+                    ((POUserFunc) op2plan.first).setAlgebraicFunction(POUserFunc.FINAL);
+                }
+
+                // we have modified the foreach inner plans - so set them
+                // again for the foreach so that foreach can do any re-initialization
+                // around them.
+                // FIXME - this is a necessary evil right now because the leaves are explicitly
+                // stored in the POForeach as a list rather than computed each time at
+                // run time from the plans for optimization. Do we want to have the Foreach
+                // compute the leaves each time and have Java optimize it (will Java optimize?)?
+                mfe.setInputPlans(mfe.getInputPlans());
+                cfe.setInputPlans(cfe.getInputPlans());
+                hforeach.setInputPlans(hforeach.getInputPlans());
+
+                // tell POCombinerPackage which fields need projected and
+                // which placed in bags. First field is simple project
+                // rest need to go into bags
+                int numFields = algebraicOps.size() + 1; // algebraic funcs + group key
+                boolean[] bags = new boolean[numFields];
+                bags[0] = false;
+                for (int i = 1; i < numFields; i++) {
+                    bags[i] = true;
+                }
+
+                // Use the POCombiner package in the combine plan
+                // as it needs to act differently than the regular
+                // package operator.
+                mr.combinePlan = new PhysicalPlan();
+                POCombinerPackage combinePack = new POCombinerPackage(pack, bags);
+                mr.combinePlan.add(combinePack);
+                mr.combinePlan.add(cfe);
+                mr.combinePlan.connect(combinePack, cfe);
+
+                // No need to connect projections in cfe to cp, because
+                // PigCombiner directly attaches output from package to
+                // root of remaining plan.
+
+                POLocalRearrange mlr = getNewRearrange(rearrange);
+
+                POPartialAgg mapAgg = null;
+                if (doMapAgg) {
+                    mapAgg = createPartialAgg(cfe);
+                }
+
+                // A specialized local rearrange operator will replace
+                // the normal local rearrange in the map plan. This behaves
+                // like the regular local rearrange in the getNext()
+                // as far as getting its input and constructing the
+                // "key" out of the input. It then returns a tuple with
+                // two fields - the key in the first position and the
+                // "value" inside a bag in the second position. This output
+                // format resembles the format out of a Package. This output
+                // will feed to the map foreach which expects this format.
+                // If the key field isn't in the project of the combiner or map foreach,
+                // it is added to the end (This is required so that we can
+                // set up the inner plan of the new Local Rearrange leaf in the map
+                // and combine plan to contain just the project of the key).
+                patchUpMap(mr.mapPlan, getPreCombinerLR(rearrange), mfe, mapAgg, mlr);
+                POLocalRearrange clr = getNewRearrange(rearrange);
+
+                mr.combinePlan.add(clr);
+                mr.combinePlan.connect(cfe, clr);
+
+                // Change the package operator in the reduce plan to
+                // be the POCombiner package, as it needs to act
+                // differently than the regular package operator.
+                POCombinerPackage newReducePack = new POCombinerPackage(pack, bags);
+                mr.reducePlan.replace(pack, newReducePack);
+
+                // the replace() above only changes
+                // the plan and does not change "inputs" to
+                // operators
+                // set up "inputs" for the operator after
+                // package correctly
+                List<PhysicalOperator> packList = new ArrayList<PhysicalOperator>();
+                packList.add(newReducePack);
+                List<PhysicalOperator> sucs = mr.reducePlan.getSuccessors(newReducePack);
+                // there should be only one successor to package
+                sucs.get(0).setInputs(packList);
+            } catch (Exception e) {
+                int errCode = 2018;
+                String msg = "Internal error. Unable to introduce the combiner for optimization.";
+                throw new OptimizerException(msg, errCode, PigException.BUG, e);
+            }
+        }
     }
 
 
@@ -498,6 +654,32 @@
     }
     
     /**
+     * Create a new hforeach with same scope,alias as given foreach add an inner plan that projects
+     * the group column, which is going to be the first input
+     * 
+     * @param hforeach
+     *            source hforeach
+     * @param keyType
+     *            type for group-by key
+     * @return new POForeach
+     */
+    private POForEach createForEachWithGrpProj(PORollupH2IRGForEach hforeach, byte keyType) {
+        String scope = hforeach.getOperatorKey().scope;
+        POForEach newFE = new POForEach(createOperatorKey(scope), new ArrayList<PhysicalPlan>());
+        newFE.addOriginalLocation(hforeach.getAlias(), hforeach.getOriginalLocations());
+        newFE.setResultType(hforeach.getResultType());
+        // create plan that projects the group column
+        PhysicalPlan grpProjPlan = new PhysicalPlan();
+        // group by column is the first column
+        POProject proj = new POProject(createOperatorKey(scope), 1, 0);
+        proj.setResultType(keyType);
+        grpProjPlan.add(proj);
+
+        newFE.addInputPlan(grpProjPlan, false);
+        return newFE;
+    }
+    
+    /**
      * Create new plan and  add to it the clones of operator algeOp  and its 
      * predecessors from the physical plan pplan .
      * @param algeOp algebraic operator 
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/InputSizeReducerEstimator.java	(working copy)
@@ -79,8 +79,7 @@
         List<POLoad> poLoads = PlanHelper.getPhysicalOperators(mapReduceOper.mapPlan, POLoad.class);
         long totalInputFileSize = getTotalInputFileSize(conf, poLoads, job);
 
-        log.info("BytesPerReducer=" + bytesPerReducer + " maxReducers="
-            + maxReducers + " totalInputFileSize=" + totalInputFileSize);
+        //log.info("BytesPerReducer=" + bytesPerReducer + " maxReducers="+ maxReducers + " totalInputFileSize=" + totalInputFileSize);
 
         // if totalInputFileSize == -1, we couldn't get the input size so we can't estimate.
         if (totalInputFileSize == -1) { return -1; }
@@ -149,7 +148,7 @@
             statistics = ((LoadMetadata) ld.getLoadFunc())
                         .getStatistics(ld.getLFile().getFileName(), job);
         } catch (Exception e) {
-            log.warn("Couldn't get statistics from LoadFunc: " + ld.getLoadFunc(), e);
+            //log.warn("Couldn't get statistics from LoadFunc: " + ld.getLoadFunc(), e);
             return -1;
         }
 
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/JobControlCompiler.java	(working copy)
@@ -82,6 +82,8 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeCogroup;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POMergeJoin;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POPackage;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORollupH2IRGForEach;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POStore;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.PlanHelper;
 import org.apache.pig.backend.hadoop.executionengine.shims.HadoopShims;
@@ -264,12 +266,12 @@
         for (FileStatus fstat: fs.listStatus(p)) {
             Path src = fstat.getPath();
             if (fstat.isDir()) {
-                log.info("mkdir: "+src);
+                //log.info("mkdir: "+src);
                 fs.mkdirs(removePart(src, rem));
                 moveResults(fstat.getPath(), rem, fs);
             } else {
                 Path dst = removePart(src, rem);
-                log.info("mv: "+src+" "+dst);
+                //log.info("mv: "+src+" "+dst);
                 fs.rename(src,dst);
             }
         }
@@ -298,7 +300,7 @@
         String defaultPigJobControlSleep = pigContext.getExecType().isLocal() ? "100" : "5000";
         String pigJobControlSleep = conf.get("pig.jobcontrol.sleep", defaultPigJobControlSleep);
         if (!pigJobControlSleep.equals(defaultPigJobControlSleep)) {
-            log.info("overriding default JobControl sleep (" + defaultPigJobControlSleep + ") to " + pigJobControlSleep);
+            //log.info("overriding default JobControl sleep (" + defaultPigJobControlSleep + ") to " + pigJobControlSleep);
         }
 
         try {
@@ -445,13 +447,13 @@
 
         long totalInputFileSize = InputSizeReducerEstimator.getTotalInputFileSize(conf, lds, job);
         long inputByteMax = conf.getLong(PigConfiguration.PIG_AUTO_LOCAL_INPUT_MAXBYTES, 100*1000*1000l);
-        log.info("Size of input: " + totalInputFileSize +" bytes. Small job threshold: " + inputByteMax );
+        //log.info("Size of input: " + totalInputFileSize +" bytes. Small job threshold: " + inputByteMax );
         if (totalInputFileSize < 0 || totalInputFileSize > inputByteMax) {
             return false;
         }
 
         int reducers = conf.getInt("mapred.reduce.tasks", 1);
-        log.info("No of reducers: " + reducers);
+        //log.info("No of reducers: " + reducers);
         if (reducers > 1) {
             return false;
         }
@@ -512,10 +514,10 @@
 
         String buffPercent = conf.get("mapred.job.reduce.markreset.buffer.percent");
         if (buffPercent == null || Double.parseDouble(buffPercent) <= 0) {
-            log.info("mapred.job.reduce.markreset.buffer.percent is not set, set to default 0.3");
+            //log.info("mapred.job.reduce.markreset.buffer.percent is not set, set to default 0.3");
             conf.set("mapred.job.reduce.markreset.buffer.percent", "0.3");
         }else{
-            log.info("mapred.job.reduce.markreset.buffer.percent is set to " + conf.get("mapred.job.reduce.markreset.buffer.percent"));
+            //log.info("mapred.job.reduce.markreset.buffer.percent is set to " + conf.get("mapred.job.reduce.markreset.buffer.percent"));
         }
 
         // Convert mapred.output.* to output.compression.*, See PIG-1791
@@ -530,6 +532,15 @@
         }
 
         try{
+        	//check and setPivot for RollupH2IRGPartitioner
+        	if(mro.customPartitioner!=null && mro.customPartitioner.equals("org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.RollupH2IRGPartitioner")) {
+        		LinkedList<POUserFunc> mapUF = PlanHelper.getPhysicalOperators(mro.mapPlan, POUserFunc.class);
+        		for(POUserFunc uf : mapUF) {
+        			if(uf.getFuncSpec().getClassName().equals("org.apache.pig.builtin.RollupDimensions"))
+        				conf.set("rollup.h2irg.pivot", uf.getPivot().toString());
+        		}
+        	}
+        	
             //Process the POLoads
             List<POLoad> lds = PlanHelper.getPhysicalOperators(mro.mapPlan, POLoad.class);
 
@@ -544,7 +555,7 @@
             }
 
             if(!mro.reducePlan.isEmpty()){
-                log.info("Reduce phase detected, estimating # of required reducers.");
+                //log.info("Reduce phase detected, estimating # of required reducers.");
                 adjustNumReducers(plan, mro, nwJob);
             } else {
                 nwJob.setNumReduceTasks(0);
@@ -573,7 +584,7 @@
             if (!pigContext.inIllustrator && ! pigContext.getExecType().isLocal())
             {
                 if (okToRunLocal(nwJob, mro, lds)) {
-                    log.info(SMALL_JOB_LOG_MSG);
+                    //log.info(SMALL_JOB_LOG_MSG);
                     // override with the default conf to run in local mode
                     for (Entry<String, String> entry : defaultConf) {
                         String key = entry.getKey();
@@ -597,26 +608,26 @@
 
                     conf.setBoolean(PigImplConstants.CONVERTED_TO_LOCAL, true);
                 } else {
-                    log.info(BIG_JOB_LOG_MSG);
+                    //log.info(BIG_JOB_LOG_MSG);
                     // Setup the DistributedCache for this job
                     for (URL extraJar : pigContext.extraJars) {
-                        log.debug("Adding jar to DistributedCache: " + extraJar.toString());
+                        //log.debug("Adding jar to DistributedCache: " + extraJar.toString());
                         putJarOnClassPathThroughDistributedCache(pigContext, conf, extraJar);
                     }
 
                     for (String scriptJar : pigContext.scriptJars) {
-                        log.debug("Adding jar to DistributedCache: " + scriptJar.toString());
+                        //log.debug("Adding jar to DistributedCache: " + scriptJar.toString());
                         putJarOnClassPathThroughDistributedCache(pigContext, conf, new File(scriptJar).toURI().toURL());
                     }
 
                     //Create the jar of all functions and classes required
                     File submitJarFile = File.createTempFile("Job", ".jar");
-                    log.info("creating jar file "+submitJarFile.getName());
+                    //log.info("creating jar file "+submitJarFile.getName());
                     // ensure the job jar is deleted on exit
                     submitJarFile.deleteOnExit();
                     FileOutputStream fos = new FileOutputStream(submitJarFile);
                     JarManager.createJar(fos, mro.UDFs, pigContext);
-                    log.info("jar file "+submitJarFile.getName()+" created");
+                    //log.info("jar file "+submitJarFile.getName()+" created");
                     //Start setting the JobConf properties
                     conf.set("mapred.jar", submitJarFile.getPath());
                 }
@@ -711,8 +722,7 @@
                 }
                 catch (Exception e) {
                     nwJob.setOutputFormatClass(PigOutputFormat.class);
-                    log.warn(PigConfiguration.PIG_OUTPUT_LAZY
-                            + " is set but LazyOutputFormat couldn't be loaded. Default PigOutputFormat will be used");
+                    //log.warn(PigConfiguration.PIG_OUTPUT_LAZY+ " is set but LazyOutputFormat couldn't be loaded. Default PigOutputFormat will be used");
                 }
             }
             else {
@@ -720,7 +730,7 @@
             }
 
             if (mapStores.size() + reduceStores.size() == 1) { // single store case
-                log.info("Setting up single store job");
+                //log.info("Setting up single store job");
 
                 POStore st;
                 if (reduceStores.isEmpty()) {
@@ -737,12 +747,12 @@
                 MapRedUtil.setupStreamingDirsConfSingle(st, pigContext, conf);
             }
             else if (mapStores.size() + reduceStores.size() > 0) { // multi store case
-                log.info("Setting up multi store job");
+                //log.info("Setting up multi store job");
                 MapRedUtil.setupStreamingDirsConfMulti(pigContext, conf);
 
                 boolean disableCounter = conf.getBoolean("pig.disable.counter", false);
                 if (disableCounter) {
-                    log.info("Disable Pig custom output counters");
+                    //log.info("Disable Pig custom output counters");
                 }
                 int idx = 0;
                 for (POStore sto: storeLocations) {
@@ -797,7 +807,7 @@
                     conf.set("pig.combine.package", ObjectSerializer.serialize(combPack));
                 } else if (mro.needsDistinctCombiner()) {
                     nwJob.setCombinerClass(DistinctCombiner.Combine.class);
-                    log.info("Setting identity combiner class.");
+                    //log.info("Setting identity combiner class.");
                 }
                 pack = (POPackage)mro.reducePlan.getRoots().get(0);
                 if(!pigContext.inIllustrator)
@@ -931,7 +941,7 @@
                 try {
                     maxCombinedSplitSize = Long.parseLong(tmp);
                 } catch (NumberFormatException e) {
-                    log.warn("Invalid numeric format for pig.maxCombinedSplitSize; use the default maximum combined split size");
+                    //log.warn("Invalid numeric format for pig.maxCombinedSplitSize; use the default maximum combined split size");
                 }
             }
             if (maxCombinedSplitSize > 0)
@@ -989,7 +999,7 @@
                     new ParallelConstantVisitor(mro.reducePlan, nPartitions);
             visitor.visit();
         }
-        log.info("Setting Parallelism to " + jobParallelism);
+        //log.info("Setting Parallelism to " + jobParallelism);
 
         Configuration conf = nwJob.getConfiguration();
 
@@ -1030,8 +1040,7 @@
                 jobParallelism = mro.estimatedParallelism;
             } else {
                 // reducer estimation could return -1 if it couldn't estimate
-                log.info("Could not estimate number of reducers and no requested or default " +
-                        "parallelism set. Defaulting to 1 reducer.");
+                //log.info("Could not estimate number of reducers and no requested or default " +"parallelism set. Defaulting to 1 reducer.");
                 jobParallelism = 1;
             }
         }
@@ -1057,7 +1066,7 @@
                     PigContext.instantiateObjectFromParams(conf,
                             REDUCER_ESTIMATOR_KEY, REDUCER_ESTIMATOR_ARG_KEY, PigReducerEstimator.class);
 
-                log.info("Using reducer estimator: " + estimator.getClass().getName());
+                //log.info("Using reducer estimator: " + estimator.getClass().getName());
                 int numberOfReducers = estimator.estimateNumberOfReducers(job, mapReducerOper);
                 return numberOfReducers;
     }
@@ -1634,12 +1643,12 @@
             Path cacheDir = new Path(stagingDir, checksum);
             Path cacheFile = new Path(cacheDir, filename);
             if (fs.exists(cacheFile)) {
-               log.info("Found " + url + " in jar cache at "+ stagingDir);
+               //log.info("Found " + url + " in jar cache at "+ stagingDir);
                long curTime = System.currentTimeMillis();
                fs.setTimes(cacheFile, -1, curTime);
                return cacheFile;
             }
-            log.info("Url "+ url + " was not found in jarcache at "+ stagingDir);
+            //log.info("Url "+ url + " was not found in jarcache at "+ stagingDir);
             // attempt to copy to cache else return null
             fs.mkdirs(cacheDir, FileLocalizer.OWNER_ONLY_PERMS);
             OutputStream os = null;
@@ -1656,7 +1665,7 @@
             return cacheFile;
 
         } catch (IOException ioe) {
-            log.info("Unable to retrieve jar from jar cache ", ioe);
+            //log.info("Unable to retrieve jar from jar cache ", ioe);
             return null;
         }
     }
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MRCompiler.java	(working copy)
@@ -65,6 +65,7 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POGlobalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORollupH2IRGForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POJoinPackage;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
@@ -219,8 +220,7 @@
                 .getProperty(FILE_CONCATENATION_THRESHOLD, "100"));
         optimisticFileConcatenation = pigContext.getProperties().getProperty(
                 OPTIMISTIC_FILE_CONCATENATION, "false").equals("true");
-        LOG.info("File concatenation threshold: " + fileConcatenationThreshold
-                + " optimistic? " + optimisticFileConcatenation);
+        //LOG.info("File concatenation threshold: " + fileConcatenationThreshold+ " optimistic? " + optimisticFileConcatenation);
     }
     
     public void aggregateScalarsFiles() throws PlanException, IOException {
@@ -1087,6 +1087,23 @@
     }
     
     @Override
+    public void visitPORollupH2IRGForEach(PORollupH2IRGForEach op) throws VisitorException {
+        try {
+            nonBlocking(op);
+            List<PhysicalPlan> plans = op.getInputPlans();
+            if (plans != null)
+                for (PhysicalPlan plan : plans) {
+                    processUDFs(plan);
+                }
+            phyToMROpMap.put(op, curMROp);
+        } catch (Exception e) {
+            int errCode = 2034;
+            String msg = "Error compiling operator " + op.getClass().getSimpleName();
+            throw new MRCompilerException(msg, errCode, PigException.BUG, e);
+        }
+    }
+    
+    @Override
     public void visitGlobalRearrange(POGlobalRearrange op) throws VisitorException{
         try{
             blocking(op);
@@ -1290,12 +1307,12 @@
                 }
             }
         } catch (IOException e) {
-            LOG.warn("failed to get number of input files", e); 
+            //LOG.warn("failed to get number of input files", e); 
         } catch (InterruptedException e) {
-            LOG.warn("failed to get number of input files", e); 
+            //LOG.warn("failed to get number of input files", e); 
         }
                 
-        LOG.info("number of input files: " + numFiles);
+        //LOG.info("number of input files: " + numFiles);
         return ret ? true : (numFiles >= fileConcatenationThreshold);
     }
     
@@ -1309,7 +1326,7 @@
         mro.mapPlan.addAsLeaf(str);
         mro.setMapDone(true);
         
-        LOG.info("Insert a file-concatenation job");
+        //LOG.info("Insert a file-concatenation job");
                 
         return mro;
     }    
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MapReduceLauncher.java	(working copy)
@@ -29,8 +29,8 @@
 import javax.xml.parsers.ParserConfigurationException;
 import javax.xml.transform.TransformerException;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
@@ -91,23 +91,23 @@
     public static final String SUCCESSFUL_JOB_OUTPUT_DIR_MARKER =
             "mapreduce.fileoutputcommitter.marksuccessfuljobs";
 
-    private static final Log log = LogFactory.getLog(MapReduceLauncher.class);
+    //private static final Log log = LogFactory.getLog(MapReduceLauncher.class);
 
     private boolean aggregateWarning = false;
 
     public void kill() {
         try {
-            log.debug("Receive kill signal");
+            //log.debug("Receive kill signal");
             if (jc!=null) {
                 for (Job job : jc.getRunningJobs()) {
                     RunningJob runningJob = job.getJobClient().getJob(job.getAssignedJobID());
                     if (runningJob!=null)
                         runningJob.killJob();
-                    log.info("Job " + job.getAssignedJobID() + " killed");
+                    //log.info("Job " + job.getAssignedJobID() + " killed");
                 }
             }
         } catch (Exception e) {
-            log.warn("Encounter exception on cleanup:" + e);
+            //log.warn("Encounter exception on cleanup:" + e);
         }
     }
 
@@ -123,7 +123,7 @@
                 else
                 {
                     job.killJob();
-                    log.info("Kill " + id + " submitted.");
+                    //log.info("Kill " + id + " submitted.");
                 }
             }
         } catch (IOException e) {
@@ -216,12 +216,8 @@
                                     " operator job :" + natOp.getJobId() + e.getMessage();
 
                             String stackTrace = Utils.getStackStraceStr(e);
-                            LogUtils.writeLog(msg,
-                                    stackTrace,
-                                    pc.getProperties().getProperty("pig.logfile"),
-                                    log
-                                    );
-                            log.info(msg);
+                            //LogUtils.writeLog(msg,stackTrace,pc.getProperties().getProperty("pig.logfile"),log);
+                            //log.info(msg);
 
                             if (stop_on_failure) {
                                 int errCode = 6017;
@@ -241,7 +237,7 @@
             }
             // Initially, all jobs are in wait state.
             List<Job> jobsWithoutIds = jc.getWaitingJobs();
-            log.info(jobsWithoutIds.size() +" map-reduce job(s) waiting for submission.");
+            //log.info(jobsWithoutIds.size() +" map-reduce job(s) waiting for submission.");
             //notify listeners about jobs submitted
             MRScriptState.get().emitJobsSubmittedNotification(jobsWithoutIds.size());
 
@@ -263,7 +259,7 @@
                 // If it is the case, we don't print out job tracker location,
                 // because it is meaningless for local mode.
                 jobTrackerLoc = null;
-                log.debug("Failed to get job tracker location.");
+                //log.debug("Failed to get job tracker location.");
             }
 
             completeFailedJobsInThisRun.clear();
@@ -311,21 +307,20 @@
                         if (job.getAssignedJobID() != null){
 
                             jobsAssignedIdInThisRun.add(job);
-                            log.info("HadoopJobId: "+job.getAssignedJobID());
+                            //log.info("HadoopJobId: "+job.getAssignedJobID());
 
                             // display the aliases being processed
                             MapReduceOper mro = jcc.getJobMroMap().get(job);
                             if (mro != null) {
                                 String alias = MRScriptState.get().getAlias(mro);
-                                log.info("Processing aliases " + alias);
+                                //log.info("Processing aliases " + alias);
                                 String aliasLocation = MRScriptState.get().getAliasLocation(mro);
-                                log.info("detailed locations: " + aliasLocation);
+                                //log.info("detailed locations: " + aliasLocation);
                             }
 
 
                             if(jobTrackerLoc != null){
-                                log.info("More information at: http://"+ jobTrackerLoc+
-                                        "/jobdetails.jsp?jobid="+job.getAssignedJobID());
+                                //log.info("More information at: http://"+ jobTrackerLoc+"/jobdetails.jsp?jobid="+job.getAssignedJobID());
                             }
 
                             // update statistics for this job so jobId is set
@@ -352,7 +347,7 @@
                             }
                             if (msg.length() > 0) {
                                 msg.setCharAt(msg.length() - 1, ']');
-                                log.info("Running jobs are [" + msg);
+                                //log.info("Running jobs are [" + msg);
                             }
                         }
                         lastProg = prog;
@@ -367,8 +362,7 @@
                     if (warn_failure && !jc.getFailedJobs().isEmpty()) {
                         // we don't warn again for this group of jobs
                         warn_failure = false;
-                        log.warn("Ooops! Some job has failed! Specify -stop_on_failure if you "
-                                + "want Pig to stop immediately on failure.");
+                        //log.warn("Ooops! Some job has failed! Specify -stop_on_failure if you "+ "want Pig to stop immediately on failure.");
                     }
                 }
 
@@ -378,10 +372,7 @@
                 if (jobControlException != null) {
                     if (jobControlException instanceof PigException) {
                         if (jobControlExceptionStackTrace != null) {
-                            LogUtils.writeLog("Error message from job controller",
-                                    jobControlExceptionStackTrace, pc
-                                    .getProperties().getProperty(
-                                            "pig.logfile"), log);
+                            //LogUtils.writeLog("Error message from job controller",jobControlExceptionStackTrace, pc.getProperties().getProperty("pig.logfile"), log);
                         }
                         throw jobControlException;
                     } else {
@@ -400,7 +391,7 @@
                     // that the job completely fail, and we shall stop dependent jobs
                     for (Job job : jc.getFailedJobs()) {
                         completeFailedJobsInThisRun.add(job);
-                        log.info("job " + job.getAssignedJobID() + " has failed! Stop running all dependent jobs");
+                        //log.info("job " + job.getAssignedJobID() + " has failed! Stop running all dependent jobs");
                     }
                     failedJobs.addAll(jc.getFailedJobs());
                 }
@@ -427,7 +418,7 @@
 
         MRScriptState.get().emitProgressUpdatedNotification(100);
 
-        log.info( "100% complete");
+        //log.info( "100% complete");
 
         boolean failed = false;
 
@@ -486,8 +477,7 @@
                         // output location is a filesystem dir
                         createSuccessFile(job, st);
                     } else {
-                        log.debug("Successfully stored result in: \""
-                                + st.getSFile().getFileName() + "\"");
+                        //log.debug("Successfully stored result in: \""+ st.getSFile().getFileName() + "\"");
                     }
                 }
 
@@ -500,18 +490,18 @@
         }
 
         if(aggregateWarning) {
-            CompilationMessageCollector.logAggregate(warningAggMap, MessageType.Warning, log) ;
+            //CompilationMessageCollector.logAggregate(warningAggMap, MessageType.Warning, log) ;
         }
 
-        if (!failed) {
-            log.info("Success!");
+        /*if (!failed) {
+            //log.info("Success!");
         } else {
             if (succJobs != null && succJobs.size() > 0) {
-                log.info("Some jobs have failed! Stop running all dependent jobs");
+                //log.info("Some jobs have failed! Stop running all dependent jobs");
             } else {
-                log.info("Failed!");
+                //log.info("Failed!");
             }
-        }
+        }*/
         jcc.reset();
 
         int ret = failed ? ((succJobs != null && succJobs.size() > 0)
@@ -579,7 +569,7 @@
         if (prog >= (lastProg + 0.04)) {
             int perCom = (int)(prog * 100);
             if(perCom!=100) {
-                log.info( perCom + "% complete");
+                //log.info( perCom + "% complete");
                 MRScriptState.get().emitProgressUpdatedNotification(perCom);
             }
             return true;
@@ -595,7 +585,7 @@
             String format,
             boolean verbose) throws PlanException, VisitorException,
             IOException {
-        log.trace("Entering MapReduceLauncher.explain");
+        //log.trace("Entering MapReduceLauncher.explain");
         MROperPlan mrp = compile(php, pc);
 
         if (format.equals("text")) {
@@ -633,7 +623,7 @@
         MROperPlan plan = comp.getMRPlan();
 
         //display the warning message(s) from the MRCompiler
-        comp.getMessageCollector().logMessages(MessageType.Warning, aggregateWarning, log);
+        //comp.getMessageCollector().logMessages(MessageType.Warning, aggregateWarning, log);
 
         String lastInputChunkSize =
                 pc.getProperties().getProperty(
@@ -646,7 +636,7 @@
             CombinerOptimizer co = new CombinerOptimizer(plan, doMapAgg);
             co.visit();
             //display the warning message(s) from the CombinerOptimizer
-            co.getMessageCollector().logMessages(MessageType.Warning, aggregateWarning, log);
+            //co.getMessageCollector().logMessages(MessageType.Warning, aggregateWarning, log);
         }
 
         // Optimize the jobs that have a load/store only first MR job followed
@@ -737,7 +727,7 @@
                     }
                 }
             } else {
-                log.warn("No FileSystem for scheme: " + scheme + ". Not creating success file");
+                //log.warn("No FileSystem for scheme: " + scheme + ". Not creating success file");
             }
         }
     }
@@ -774,12 +764,12 @@
                         }
                     }
                 } catch (Exception e) {
-                    log.warn("Exception getting counters.", e);
+                    //log.warn("Exception getting counters.", e);
                 }
             }
         } catch (IOException ioe) {
             String msg = "Unable to retrieve job to compute warning aggregation.";
-            log.warn(msg);
+            //log.warn(msg);
         }
     }
 
@@ -790,11 +780,7 @@
         Exception backendException = null;
         if (MRJobID == null) {
             try {
-                LogUtils.writeLog(
-                        "Backend error message during job submission",
-                        jobMessage,
-                        pigContext.getProperties().getProperty("pig.logfile"),
-                        log);
+                //LogUtils.writeLog("Backend error message during job submission",jobMessage,pigContext.getProperties().getProperty("pig.logfile"),log);
                 backendException = getExceptionFromString(jobMessage);
             } catch (Exception e) {
                 int errCode = 2997;
@@ -817,7 +803,7 @@
             if (job.getState() == Job.SUCCESS) {
                 // if the job succeeded, let the user know that
                 // we were unable to get statistics
-                log.warn("Unable to get job related diagnostics");
+                //log.warn("Unable to get job related diagnostics");
             } else {
                 throw new ExecException(e);
             }
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/MultiQueryOptimizer.java	(working copy)
@@ -86,7 +86,7 @@
         List<MapReduceOper> roots = plan.getRoots();
         scope = roots.get(0).getOperatorKey().getScope();
         this.inIllustrator = inIllustrator;
-        log.info("MR plan size before optimization: " + plan.size());
+        //log.info("MR plan size before optimization: " + plan.size());
     }
 
     @Override
@@ -93,7 +93,7 @@
     public void visit() throws VisitorException {
         super.visit();
         
-        log.info("MR plan size after optimization: " + mPlan.size());
+        //log.info("MR plan size after optimization: " + mPlan.size());
     }
     
     @Override
@@ -111,13 +111,11 @@
         List<MapReduceOper> successors = getPlan().getSuccessors(mr);
         for (MapReduceOper successor : successors) {
             if (successor.getUseSecondaryKey()) {
-                log.debug("Splittee " + successor.getOperatorKey().getId()
-                        + " uses secondary key, do not merge it");
+                //log.debug("Splittee " + successor.getOperatorKey().getId()+ " uses secondary key, do not merge it");
                 continue;
             }
             if (successor.getCustomPartitioner() != null) {
-                log.debug("Splittee " + successor.getOperatorKey().getId()
-                        + " uses customPartitioner, do not merge it");
+                //log.debug("Splittee " + successor.getOperatorKey().getId()+ " uses customPartitioner, do not merge it");
                 continue;
             }
             if (isMapOnly(successor)) {
@@ -143,7 +141,7 @@
         if (mappers.size() == 1 && numSplittees == 1) {    
             mergeOnlyMapperSplittee(mappers.get(0), mr);
             
-            log.info("Merged the only map-only splittee.");
+            //log.info("Merged the only map-only splittee.");
             
             return;              
         }
@@ -152,7 +150,7 @@
         if (isMapOnly(mr) && mapReducers.size() == 1 && numSplittees == 1) {            
             mergeOnlyMapReduceSplittee(mapReducers.get(0), mr);
             
-            log.info("Merged the only map-reduce splittee.");
+            //log.info("Merged the only map-reduce splittee.");
             
             return;
         } 
@@ -169,7 +167,7 @@
             splitOp = getSplit();  
             int n = mergeAllMapOnlySplittees(mappers, mr, splitOp);            
             
-            log.info("Merged " + n + " map-only splittees.");
+            //log.info("Merged " + n + " map-only splittees.");
             
             numMerges += n;   
         }            
@@ -198,7 +196,7 @@
                 
             }
             
-            log.info("Merged " + merged + " map-reduce splittees.");
+            //log.info("Merged " + merged + " map-reduce splittees.");
             
             numMerges += merged;      
         }
@@ -222,12 +220,11 @@
         // case 6: special diamond case with trivial MR operator at the head
         if (numMerges == 0 && isDiamondMROper(mr)) {
             int merged = mergeDiamondMROper(mr, getPlan().getSuccessors(mr));
-            log.info("Merged " + merged + " diamond splitter.");
+            //log.info("Merged " + merged + " diamond splitter.");
             numMerges += merged;    
         }
         
-        log.info("Merged " + numMerges + " out of total " 
-                + (numSplittees +1) + " MR operators.");
+        //log.info("Merged " + numMerges + " out of total "+ (numSplittees +1) + " MR operators.");
     }                
     
     private boolean isDiamondMROper(MapReduceOper mr) {
@@ -458,8 +455,7 @@
         // cannot be global sort or limit after sort, they are 
         // using a different partitioner
         if (splittee.isGlobalSort() || splittee.isLimitAfterSort()) {
-            log.info("Cannot merge this splittee: " +
-            		"it is global sort or limit after sort");
+            //log.info("Cannot merge this splittee: " +"it is global sort or limit after sort");
             return false;
         }
         
@@ -467,16 +463,14 @@
         PhysicalOperator leaf = splittee.mapPlan.getLeaves().get(0);
         if (!(leaf instanceof POLocalRearrange) && 
                 ! (leaf instanceof POSplit)) {
-            log.info("Cannot merge this splittee: " +
-            		"its map plan doesn't end with LR or Split operator: " 
-                    + leaf.getClass().getName());
+            /*log.info("Cannot merge this splittee: " +"its map plan doesn't end with LR or Split operator: "+ 
+        	leaf.getClass().getName());*/
             return false;
         }
            
         // cannot have distinct combiner, it uses a different combiner
         if (splittee.needsDistinctCombiner()) {
-            log.info("Cannot merge this splittee: " +
-            		"it has distinct combiner.");
+            //log.info("Cannot merge this splittee: " +"it has distinct combiner.");
             return false;           
         }
         
@@ -955,7 +949,7 @@
        
         boolean sameKeyType = hasSameMapKeyType(mergeList);
         
-        log.debug("Splittees have the same key type: " + sameKeyType);
+        //log.debug("Splittees have the same key type: " + sameKeyType);
         
         // create a new reduce plan that will be the container
         // for the multiple reducer plans of the MROpers in the mergeList
@@ -966,7 +960,7 @@
         PhysicalPlan comPl = needCombiner(mergeList) ? 
                 createDemuxPlan(sameKeyType, true) : null;
 
-        log.debug("Splittees have combiner: " + (comPl != null));
+        //log.debug("Splittees have combiner: " + (comPl != null));
                 
         int index = 0;             
         
@@ -998,8 +992,7 @@
            
             index = incIndex;
             
-            log.info("Merged MR job " + mrOp.getOperatorKey().getId() 
-                    + " into MR job " + splitter.getOperatorKey().getId());
+            //log.info("Merged MR job " + mrOp.getOperatorKey().getId()+ " into MR job " + splitter.getOperatorKey().getId());
         }
 
         PhysicalPlan splitterPl = splitter.mapPlan;
@@ -1034,8 +1027,7 @@
         splitter.mapKeyType = sameKeyType ?
                 mergeList.get(0).mapKeyType : DataType.TUPLE;         
                 
-        log.info("Requested parallelism of splitter: " 
-                + splitter.getRequestedParallelism());               
+        //log.info("Requested parallelism of splitter: "+ splitter.getRequestedParallelism());               
     }
     
     private void mergeSingleMapReduceSplittee(MapReduceOper mapReduce, 
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigCombiner.java	(working copy)
@@ -21,8 +21,8 @@
 import java.io.IOException;
 import java.util.ArrayList;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -51,7 +51,7 @@
     public static class Combine
             extends Reducer<PigNullableWritable, NullableTuple, PigNullableWritable, Writable> {
 
-        private static final Log log = LogFactory.getLog(Combine.class);
+        //private static final Log log = LogFactory.getLog(Combine.class);
 
         //HADOOP-3226 Combiners can be called multiple times in both map and reduce
         private static boolean firstTime = true;
@@ -97,11 +97,12 @@
                 pack = (POPackage)ObjectSerializer.deserialize(jConf.get("pig.combine.package"));
                 // To be removed
                 if(cp.isEmpty())
-                    log.debug("Combine Plan empty!");
+                {}
+                    //log.debug("Combine Plan empty!");
                 else{
                     ByteArrayOutputStream baos = new ByteArrayOutputStream();
                     cp.explain(baos);
-                    log.debug(baos.toString());
+                    //log.debug(baos.toString());
                 }
 
                 keyType = ((byte[])ObjectSerializer.deserialize(jConf.get("pig.map.keytype")))[0];
@@ -119,7 +120,7 @@
 
             // Avoid log spamming
             if (firstTime) {
-                log.info("Aliases being processed per job phase (AliasName[line,offset]): " + jConf.get("pig.alias.location"));
+                //log.info("Aliases being processed per job phase (AliasName[line,offset]): " + jConf.get("pig.alias.location"));
                 firstTime = false;
             }
         }
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapBase.java	(working copy)
@@ -23,7 +23,6 @@
 import java.util.List;
 
 import org.joda.time.DateTimeZone;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -31,10 +30,13 @@
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Partitioner;
+import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
 import org.apache.log4j.PropertyConfigurator;
 import org.apache.pig.PigException;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.datastorage.ConfigurationUtil;
+import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.partitioners.RollupH2IRGPartitioner;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.POStatus;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.PhysicalOperator;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.Result;
@@ -210,8 +212,7 @@
         }
         
         PigStatusReporter.setContext(context);
- 
-        log.info("Aliases being processed per job phase (AliasName[line,offset]): " + job.get("pig.alias.location"));
+        //log.info("Aliases being processed per job phase (AliasName[line,offset]): " + job.get("pig.alias.location"));
         
         String dtzStr = PigMapReduce.sJobConfInternal.get().get("pig.datetime.default.tz");
         if (dtzStr != null && dtzStr.length() > 0) {
Index: src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/mapReduceLayer/PigGenericMapReduce.java	(working copy)
@@ -24,7 +24,6 @@
 import java.util.List;
 
 import org.joda.time.DateTimeZone;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -118,7 +117,7 @@
             // assign the tuple to its slot in the projection.
             key.setIndex(index);
             val.setIndex(index);
-
+            System.out.println(tuple.get(1) + " " + tuple.get(2));
             oc.write(key, val);
         }
     }
@@ -542,6 +541,29 @@
                 throw new IOException("Error trying to finish UDFs",e);
             }
             
+            //Calling PORollupH2IRGForEach.finish()
+            PORollupH2IRGForEachFinishVisitor finisher2 = new PORollupH2IRGForEachFinishVisitor(rp,
+                    new DependencyOrderWalker<PhysicalOperator, PhysicalPlan>(rp));
+            try {
+                finisher2.visit();
+                if (finisher2.returnRes != null) {
+                    for (int i = finisher2.returnRes.length - 1; i >= 0; i--) {
+                        if (finisher2.returnRes[i] != null) {
+                            if (finisher2.returnRes[i].returnStatus == POStatus.STATUS_OK) {
+                                try {
+                                    outputCollector.write(null, (Tuple) finisher2.returnRes[i].result);
+                                } catch (Exception e) {
+                                    throw new IOException(e);
+                                }
+                            }
+                            finisher2.returnRes[i] = null;
+                        }
+                    }
+                }
+            } catch (VisitorException e) {
+                throw new IOException("Error trying to finish PORollupH2IRGForEach", e);
+            }
+            
             PhysicalOperator.setReporter(null);
             initialized = false;
         }
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POProject.java	(working copy)
@@ -401,8 +401,10 @@
         if(!processingBagOfTuples){
             Tuple inpValue = null;
             res = processInput();
-            if(res.returnStatus!=POStatus.STATUS_OK)
-                return res;
+            
+            if(res.returnStatus!=POStatus.STATUS_OK){
+            	return res;	
+            }
             if(isStar())
                 return res;
 
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/expressionOperators/POUserFunc.java	(working copy)
@@ -43,6 +43,7 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhyPlanVisitor;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.util.MonitoredUDFExecutor;
 import org.apache.pig.builtin.MonitoredUDF;
+import org.apache.pig.builtin.RollupDimensions;
 import org.apache.pig.data.DataType;
 import org.apache.pig.data.SchemaTupleClassGenerator.GenContext;
 import org.apache.pig.data.SchemaTupleFactory;
@@ -88,6 +89,26 @@
     private long numInvocations = 0L;
     private boolean doTiming = false;
 
+    private Boolean rollupH2IRGoptimizable = null;
+    private Integer pivot = null;
+    
+    public void setPivot(Integer pvt) {
+    	this.pivot = pvt;
+    	System.out.println("POUserFunc NEW PIVOT " + this.pivot);
+    }
+    
+    public Integer getPivot() {
+    	return this.pivot;
+    }
+    
+    public void setRollupH2IRGOptimizable(boolean check) {
+    	this.rollupH2IRGoptimizable = check;
+    }
+    
+    public boolean getRollupH2IRGOptimizable() {
+    	return this.rollupH2IRGoptimizable;
+    }
+    
     public PhysicalOperator getReferencedOperator() {
         return referencedOperator;
     }
@@ -188,6 +209,7 @@
         }
 
         Result res = new Result();
+        //System.out.println(res.result.toString());
         if (input == null && (inputs == null || inputs.size()==0)) {
             res.returnStatus = POStatus.STATUS_EOP;
             return res;
@@ -350,8 +372,13 @@
                     if (executor != null) {
                         result.result = executor.monitorExec((Tuple) result.result);
                     } else {
-                        result.result = func.exec((Tuple) result.result);
-                    }
+                    	if(funcSpec.getClassName().toString().equals("org.apache.pig.builtin.RollupDimensions") && this.rollupH2IRGoptimizable!=null) {//&& isOptimized (disable rule)
+                    		if(this.pivot!=null)
+                    			((RollupDimensions)func).setPivot(this.pivot);
+                    			((RollupDimensions)func).setRollupH2IRGOptimizable(this.rollupH2IRGoptimizable);
+                    	}
+                    	result.result = func.exec((Tuple) result.result);
+                    } 
                 }
             }
             if (timeThis) {
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/DotPOPrinter.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/DotPOPrinter.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/DotPOPrinter.java	(working copy)
@@ -21,7 +21,9 @@
 import java.util.List;
 import java.util.LinkedList;
 import java.util.Collection;
+
 import org.apache.pig.impl.util.MultiMap;
+
 import java.util.HashSet;
 import java.util.Set;
 
@@ -114,6 +116,9 @@
         else if(op instanceof POForEach){
             plans.addAll(((POForEach)op).getInputPlans());
         }
+    	else if (op instanceof PORollupH2IRGForEach) {
+    		plans.addAll(((PORollupH2IRGForEach) op).getInputPlans());
+    	}
         else if(op instanceof POSort){
             plans.addAll(((POSort)op).getSortPlans());
         }
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PhyPlanVisitor.java	(working copy)
@@ -106,6 +106,15 @@
         }
     }
     
+    public void visitPORollupH2IRGForEach(PORollupH2IRGForEach nhfe) throws VisitorException {
+        List<PhysicalPlan> inpPlans = nhfe.getInputPlans();
+        for (PhysicalPlan plan : inpPlans) {
+            pushWalker(mCurrentWalker.spawnChildWalker(plan));
+            visit();
+            popWalker();
+        }
+    }
+    
     public void visitUnion(POUnion un) throws VisitorException{
         //do nothing
     }
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PlanPrinter.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PlanPrinter.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/plans/PlanPrinter.java	(working copy)
@@ -175,6 +175,9 @@
           else if(node instanceof POForEach){
             sb.append(planString(((POForEach)node).getInputPlans()));
           }
+          else if(node instanceof PORollupH2IRGForEach){
+              sb.append(planString(((PORollupH2IRGForEach)node).getInputPlans()));
+          }
           else if (node instanceof POMultiQueryPackage) {
               List<POPackage> pkgs = ((POMultiQueryPackage)node).getPackages();
               for (POPackage pkg : pkgs) {
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POForEach.java	(working copy)
@@ -203,7 +203,7 @@
      * Calls getNext on the generate operator inside the nested
      * physical plan and returns it maintaining an additional state
      * to denote the begin and end of the nested plan processing.
-     */
+     **/
     @Override
     public Result getNextTuple() throws ExecException {
         try {
@@ -230,7 +230,7 @@
                         return res;
                     }
                     if(res.returnStatus==POStatus.STATUS_NULL) {
-                        continue;
+                    	continue;
                     }
                 }
             }
@@ -239,7 +239,7 @@
             //nested plan processing on the input tuple
             //read
             while (true) {
-                inp = processInput();
+            	inp = processInput();
                 if (inp.returnStatus == POStatus.STATUS_EOP ||
                         inp.returnStatus == POStatus.STATUS_ERR) {
                     return inp;
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POLocalRearrange.java	(working copy)
@@ -150,7 +150,7 @@
         secondaryLeafOps = new ArrayList<ExpressionOperator>();
         mProjectedColsMap = new HashMap<Integer, Integer>();
         mSecondaryProjectedColsMap = new HashMap<Integer, Integer>();
-    }
+	}
 
     @Override
     public void visit(PhyPlanVisitor v) throws VisitorException {
Index: src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPreCombinerLocalRearrange.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPreCombinerLocalRearrange.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/physicalLayer/relationalOperators/POPreCombinerLocalRearrange.java	(working copy)
@@ -123,7 +123,7 @@
         Result inp = null;
         Result res = ERR_RESULT;
         while (true) {
-            inp = processInput();
+        	inp = processInput();
             if (inp.returnStatus == POStatus.STATUS_EOP || inp.returnStatus == POStatus.STATUS_ERR) {
                 break;
             }
Index: src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java
===================================================================
--- src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java	(revision 1579421)
+++ src/org/apache/pig/backend/hadoop/executionengine/util/MapRedUtil.java	(working copy)
@@ -29,8 +29,8 @@
 import java.util.List;
 import java.util.Map;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
+//import org.apache.commons.logging.Log;
+//import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -66,7 +66,7 @@
  */
 public class MapRedUtil {
 
-    private static Log log = LogFactory.getLog(MapRedUtil.class);
+    //private static Log log = LogFactory.getLog(MapRedUtil.class);
          
     public static final String FILE_SYSTEM_NAME = "fs.default.name";
 
@@ -106,7 +106,7 @@
         Tuple t = loader.getNext();
         if (t == null) {
             // this could happen if the input directory for sampling is empty
-            log.warn("Empty dist file: " + keyDistFile);
+            //log.warn("Empty dist file: " + keyDistFile);
             return reducerMap;
         }
         // The keydist file is structured as (key, min, max)
@@ -267,7 +267,7 @@
                 result.add(file);
             }
         }
-        log.info("Total input paths to process : " + result.size()); 
+        //log.info("Total input paths to process : " + result.size()); 
         return result;
     }
     
@@ -660,7 +660,7 @@
         if (totalLen != origTotalLen)
           throw new AssertionError("The total length ["+totalLen+"] does not match the original ["+origTotalLen+"]");
         */ 
-        log.info("Total input paths (combined) to process : " + result.size());
+        //log.info("Total input paths (combined) to process : " + result.size());
         return result;
     }
     
Index: src/org/apache/pig/builtin/RollupDimensions.java
===================================================================
--- src/org/apache/pig/builtin/RollupDimensions.java	(revision 1579421)
+++ src/org/apache/pig/builtin/RollupDimensions.java	(working copy)
@@ -17,8 +17,11 @@
  */
 package org.apache.pig.builtin;
 
+import java.io.File;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.util.List;
+import java.util.Scanner;
 
 import org.apache.pig.EvalFunc;
 import org.apache.pig.backend.executionengine.ExecException;
@@ -47,7 +50,9 @@
     private static BagFactory bf = BagFactory.getInstance();
     private static TupleFactory tf = TupleFactory.getInstance();
     private final String allMarker;
-
+    private Integer pivot = null;
+    private Boolean rollupH2IRGoptimizable = null;
+    
     public RollupDimensions() {
 	this(null);
     }
@@ -56,7 +61,19 @@
 	super();
 	this.allMarker = allMarker;
     }
+    
+    public void setRollupH2IRGOptimizable(Boolean check) {
+    	this.rollupH2IRGoptimizable = check;
+    }
+    
+    public Boolean getRollupH2IRGOptimizable() {
+    	return this.rollupH2IRGoptimizable;
+    }
 
+    public void setPivot(int pvt) throws IOException {
+    	this.pivot = pvt;
+    }
+    
     @Override
     public DataBag exec(Tuple tuple) throws IOException {
 	List<Tuple> result = Lists.newArrayListWithCapacity(tuple.size() + 1);
@@ -66,14 +83,33 @@
 	return bf.newDefaultBag(result);
     }
 
-    private void iterativelyRollup(List<Tuple> result, Tuple input) throws ExecException {
-	Tuple tempTup = tf.newTuple(input.getAll());
-	for (int i = input.size() - 1; i >= 0; i--) {
-	    tempTup.set(i, allMarker);
-	    result.add(tf.newTuple(tempTup.getAll()));
-	}
+    private void iterativelyRollup(List<Tuple> result, Tuple input) throws IOException {
+    	
+    	Tuple tempTup = tf.newTuple(input.getAll());
+    	
+    	if(this.rollupH2IRGoptimizable !=null) { // rule is enabled
+    		if(this.rollupH2IRGoptimizable == true) {
+	    		if(this.pivot == null) //user did not specify the pivot position --> IRG approach
+	    			return;
+	    		else { //user did specify the pivot position --> IRG + IRG
+	    			if (this.pivot == 0) // we use the IRG approach
+	    				return;
+	    			else { // we use IRG+IRG approach
+	    				for (int i = this.pivot - 1; i < input.size(); i++)
+	    	    			tempTup.set(i, allMarker);
+	    	    		result.add(tf.newTuple(tempTup.getAll()));
+	    			}
+	    		}
+    		}
+    	}
+    	else { //we can not optimize --> Vanilla approach
+    		for (int i = input.size() - 1; i >= 0; i--) {
+    		    tempTup.set(i, allMarker);
+    		    result.add(tf.newTuple(tempTup.getAll()));
+    		}
+    	}
     }
-
+ 
     @Override
     public Schema outputSchema(Schema input) {
 	// "dimensions" string is the default namespace assigned to the output
Index: src/org/apache/pig/data/SchemaTupleFrontend.java
===================================================================
--- src/org/apache/pig/data/SchemaTupleFrontend.java	(revision 1579421)
+++ src/org/apache/pig/data/SchemaTupleFrontend.java	(working copy)
@@ -71,7 +71,7 @@
             return pr.getFirst();
         }
         if (!SchemaTupleFactory.isGeneratable(udfSchema)) {
-            LOG.debug("Given Schema is not generatable: " + udfSchema);
+            //LOG.debug("Given Schema is not generatable: " + udfSchema);
             return -1;
         }
         int id = SchemaTupleClassGenerator.getNextGlobalClassIdentifier();
@@ -79,8 +79,7 @@
         contexts.add(GenContext.FORCE_LOAD);
         contexts.add(type);
         schemasToGenerate.put(key, Pair.make(Integer.valueOf(id), contexts));
-        LOG.debug("Registering "+(isAppendable ? "Appendable" : "")+"Schema for generation ["
-                + udfSchema + "] with id [" + id + "] and context: " + type);
+        //LOG.debug("Registering "+(isAppendable ? "Appendable" : "")+"Schema for generation ["+ udfSchema + "] with id [" + id + "] and context: " + type);
         return id;
     }
 
@@ -96,8 +95,7 @@
         public SchemaTupleFrontendGenHelper(PigContext pigContext, Configuration conf) {
             codeDir = Files.createTempDir();
             codeDir.deleteOnExit();
-            LOG.debug("Temporary directory for generated code created: "
-                    + codeDir.getAbsolutePath());
+            //LOG.debug("Temporary directory for generated code created: "+ codeDir.getAbsolutePath());
             this.pigContext = pigContext;
             this.conf = conf;
         }
@@ -110,12 +108,11 @@
          * @param conf
          */
         private void internalCopyAllGeneratedToDistributedCache() {
-            LOG.info("Starting process to move generated code to distributed cacche");
+            //LOG.info("Starting process to move generated code to distributed cacche");
             if (pigContext.getExecType().isLocal()) {
                 String codePath = codeDir.getAbsolutePath();
-                LOG.info("Distributed cache not supported or needed in local mode. Setting key ["
-                        + LOCAL_CODE_DIR + "] with code temp directory: " + codePath);
-                    conf.set(LOCAL_CODE_DIR, codePath);
+                //LOG.info("Distributed cache not supported or needed in local mode. Setting key ["+ LOCAL_CODE_DIR + "] with code temp directory: " + codePath);
+                conf.set(LOCAL_CODE_DIR, codePath);
                 return;
             } else {
                 // This let's us avoid NPE in some of the non-traditional pipelines
@@ -161,10 +158,10 @@
                 } catch (URISyntaxException e) {
                     throw new RuntimeException("Unable to add file to distributed cache: " + destination, e);
                 }
-                LOG.info("File successfully added to the distributed cache: " + symlink);
+                //LOG.info("File successfully added to the distributed cache: " + symlink);
             }
             String toSer = serialized.toString();
-            LOG.info("Setting key [" + GENERATED_CLASSES_KEY + "] with classes to deserialize [" + toSer + "]");
+            //LOG.info("Setting key [" + GENERATED_CLASSES_KEY + "] with classes to deserialize [" + toSer + "]");
             // we must set a key in the job conf so individual jobs know to resolve the shipped classes
             conf.set(GENERATED_CLASSES_KEY, toSer);
         }
@@ -177,10 +174,10 @@
         private boolean generateAll(Map<Pair<SchemaKey, Boolean>, Pair<Integer, Set<GenContext>>> schemasToGenerate) {
             boolean filesToShip = false;
             if (!conf.getBoolean(SHOULD_USE_SCHEMA_TUPLE, SCHEMA_TUPLE_ON_BY_DEFAULT)) {
-                LOG.info("Key ["+SHOULD_USE_SCHEMA_TUPLE+"] is false, will not generate code.");
+                //LOG.info("Key ["+SHOULD_USE_SCHEMA_TUPLE+"] is false, will not generate code.");
                 return false;
             }
-            LOG.info("Generating all registered Schemas.");
+            //LOG.info("Generating all registered Schemas.");
             for (Map.Entry<Pair<SchemaKey, Boolean>, Pair<Integer, Set<GenContext>>> entry : schemasToGenerate.entrySet()) {
                 Pair<SchemaKey, Boolean> keyPair = entry.getKey();
                 Schema s = keyPair.getFirst().get();
@@ -189,7 +186,7 @@
                 boolean isShipping = false;
                 for (GenContext context : valuePair.getSecond()) {
                     if (!context.shouldGenerate(conf)) {
-                        LOG.info("Skipping generation of Schema [" + s + "], as key value [" + context.key() + "] was false.");
+                        //LOG.info("Skipping generation of Schema [" + s + "], as key value [" + context.key() + "] was false.");
                     } else {
                         isShipping = true;
                         contextsToInclude.add(context);
@@ -271,7 +268,7 @@
      */
     public static void copyAllGeneratedToDistributedCache(PigContext pigContext, Configuration conf) {
         if (stf == null) {
-            LOG.debug("Nothing registered to generate.");
+            //LOG.debug("Nothing registered to generate.");
             return;
         }
         SchemaTupleFrontendGenHelper stfgh = new SchemaTupleFrontendGenHelper(pigContext, conf);
Index: src/org/apache/pig/impl/PigContext.java
===================================================================
--- src/org/apache/pig/impl/PigContext.java	(revision 1579421)
+++ src/org/apache/pig/impl/PigContext.java	(working copy)
@@ -239,13 +239,13 @@
         this(ExecType.MAPREDUCE, new Properties());
     }
 
-        public PigContext(Configuration conf) throws PigException {
-            this(ConfigurationUtil.toProperties(conf));
-        }
+    public PigContext(Configuration conf) throws PigException {
+    	this(ConfigurationUtil.toProperties(conf));
+    }
         
-        public PigContext(Properties properties) throws PigException {
-            this(ExecTypeProvider.selectExecType(properties), properties);
-        }
+    public PigContext(Properties properties) throws PigException {
+    	this(ExecTypeProvider.selectExecType(properties), properties);
+    }
     
     public PigContext(ExecType execType, Configuration conf) {
         this(execType, ConfigurationUtil.toProperties(conf));
Index: src/org/apache/pig/impl/util/Utils.java
===================================================================
--- src/org/apache/pig/impl/util/Utils.java	(revision 1579421)
+++ src/org/apache/pig/impl/util/Utils.java	(working copy)
@@ -392,7 +392,7 @@
             conf.set(PigConfiguration.PIG_TEMP_FILE_COMPRESSION_STORAGE, "seqfile");
             if("".equals(codec)) {
                 // codec is not specified, ensure  is set
-                log.warn("Temporary file compression codec is not specified. Using mapred.output.compression.codec property.");
+                ////log.warn("Temporary file compression codec is not specified. Using mapred.output.compression.codec property.");
                 if(conf.get("mapred.output.compression.codec") == null) {
                     throw new IOException("mapred.output.compression.codec is not set");
                 }
@@ -480,7 +480,7 @@
             final InputStream inputSteam = new FileInputStream(new File(bootupFile));
             return new SequenceInputStream(inputSteam, in);
         } catch(FileNotFoundException fe) {
-            log.info("Default bootup file " +bootupFile+ " not found");
+            //log.info("Default bootup file " +bootupFile+ " not found");
             return in;
         }
     }
Index: src/org/apache/pig/newplan/logical/DotLOPrinter.java
===================================================================
--- src/org/apache/pig/newplan/logical/DotLOPrinter.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/DotLOPrinter.java	(working copy)
@@ -35,6 +35,7 @@
 import org.apache.pig.newplan.logical.relational.LOFilter;
 import org.apache.pig.newplan.logical.relational.LOForEach;
 import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LORollupH2IRGForEach;
 import org.apache.pig.newplan.logical.relational.LOJoin;
 import org.apache.pig.newplan.logical.relational.LOLimit;
 import org.apache.pig.newplan.logical.relational.LOLoad;
@@ -140,6 +141,9 @@
         else if(op instanceof LOForEach){
             plans.add(((LOForEach)op).getInnerPlan());
         }
+    	else if (op instanceof LORollupH2IRGForEach) {
+    		plans.add(((LORollupH2IRGForEach) op).getInnerPlan());
+    	}
         else if(op instanceof LOGenerate){
             plans.addAll(((LOGenerate)op).getOutputPlans());
         }
Index: src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java
===================================================================
--- src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/expression/ExpToPhyTranslationVisitor.java	(working copy)
@@ -506,6 +506,16 @@
                     .getNextNodeId(DEFAULT_SCOPE)), -1,
                     null, op.getFuncSpec(), (EvalFunc) f);
             ((POUserFunc)p).setSignature(op.getSignature());
+            
+            if( op.getFuncSpec().toString().equals("org.apache.pig.builtin.RollupDimensions"))
+            {
+            	if(op.getPivot()!=null)
+            		((POUserFunc)p).setPivot(op.getPivot());
+            	if(op.getRollupH2IRGOptimizable()!=null)
+            		((POUserFunc)p).setRollupH2IRGOptimizable(op.getRollupH2IRGOptimizable());
+            }
+            
+            
             //reinitialize input schema from signature
             if (((POUserFunc)p).getFunc().getInputSchema() == null) {
                 ((POUserFunc)p).setFuncInputSchema(op.getSignature());
Index: src/org/apache/pig/newplan/logical/expression/UserFuncExpression.java
===================================================================
--- src/org/apache/pig/newplan/logical/expression/UserFuncExpression.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/expression/UserFuncExpression.java	(working copy)
@@ -57,6 +57,26 @@
     private static int sigSeq=0;
     private boolean viaDefine=false; //this represents whether the function was instantiate via a DEFINE statement or not
 
+    private Boolean rollupH2IRGoptimizable = null;
+    private Integer pivot = null;
+    
+    public void setPivot(Integer pvt) {
+    	this.pivot = pvt;
+    	System.out.println("UF NEW PIVOT " + this.pivot);
+    }
+    
+    public Integer getPivot() {
+    	return this.pivot;
+    }
+    
+    public void setRollupH2IRGOptimizable(boolean check) {
+    	this.rollupH2IRGoptimizable = check;
+    }
+    
+    public Boolean getRollupH2IRGOptimizable() {
+    	return this.rollupH2IRGoptimizable;
+    }
+    
     public UserFuncExpression(OperatorPlan plan, FuncSpec funcSpec) {
         super("UserFunc", plan);
         mFuncSpec = funcSpec;
Index: src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java
===================================================================
--- src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/optimizer/LogicalPlanOptimizer.java	(working copy)
@@ -43,6 +43,7 @@
 import org.apache.pig.newplan.logical.rules.PushUpFilter;
 import org.apache.pig.newplan.logical.rules.SplitFilter;
 import org.apache.pig.newplan.logical.rules.StreamTypeCastInserter;
+import org.apache.pig.newplan.logical.rules.RollupH2IRGOptimizer;
 import org.apache.pig.newplan.optimizer.PlanOptimizer;
 import org.apache.pig.newplan.optimizer.Rule;
 
@@ -66,9 +67,8 @@
         if (mRulesOff.contains("all")) {
             allRulesDisabled = true;
         }
-
         ruleSets = buildRuleSets();
-        LOG.info(rulesReport);
+        //LOG.info(rulesReport);
         addListeners();
     }
 
@@ -75,6 +75,7 @@
     protected List<Set<Rule>> buildRuleSets() {
         List<Set<Rule>> ls = new ArrayList<Set<Rule>>();
 
+        
         // Logical expression simplifier
         Set <Rule> s = new HashSet<Rule>();
         // add logical expression simplification rule
@@ -180,7 +181,15 @@
         checkAndAddRule(s, r);
         if (!s.isEmpty())
             ls.add(s);
-
+        
+        // This set of rules for rollup
+        s = new HashSet<Rule>();
+        // Optimize limit
+        r = new RollupH2IRGOptimizer("RollupH2IRGOptimizer");
+        checkAndAddRule(s, r);
+        if (!s.isEmpty())
+            ls.add(s);
+        
         return ls;
     }
 
@@ -192,7 +201,7 @@
     private void checkAndAddRule(Set<Rule> ruleSet, Rule rule) {
         Preconditions.checkArgument(ruleSet != null);
         Preconditions.checkArgument(rule != null && rule.getName() != null);
-
+        
         if (rule.isMandatory()) {
             ruleSet.add(rule);
             rulesReport.put(RulesReportKey.RULES_ENABLED, rule.getName());
Index: src/org/apache/pig/newplan/logical/optimizer/LogicalPlanPrinter.java
===================================================================
--- src/org/apache/pig/newplan/logical/optimizer/LogicalPlanPrinter.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/optimizer/LogicalPlanPrinter.java	(working copy)
@@ -33,6 +33,7 @@
 import org.apache.pig.newplan.logical.relational.LOFilter;
 import org.apache.pig.newplan.logical.relational.LOForEach;
 import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LORollupH2IRGForEach;
 import org.apache.pig.newplan.logical.relational.LOJoin;
 import org.apache.pig.newplan.logical.relational.LOLimit;
 import org.apache.pig.newplan.logical.relational.LORank;
@@ -155,6 +156,10 @@
         else if(node instanceof LOForEach){
             printPlan(((LOForEach)node).getInnerPlan());        
         }
+        else if(node instanceof LORollupH2IRGForEach){
+            printPlan(((LORollupH2IRGForEach)node).getInnerPlan());        
+        }
+        
         else if(node instanceof LOCogroup){
             MultiMap<Integer, LogicalExpressionPlan> plans = ((LOCogroup)node).getExpressionPlans();
             for (int i : plans.keySet()) {
Index: src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java
===================================================================
--- src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/optimizer/SchemaResetter.java	(working copy)
@@ -39,6 +39,7 @@
 import org.apache.pig.newplan.logical.relational.LODistinct;
 import org.apache.pig.newplan.logical.relational.LOFilter;
 import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LORollupH2IRGForEach;
 import org.apache.pig.newplan.logical.relational.LOGenerate;
 import org.apache.pig.newplan.logical.relational.LOInnerLoad;
 import org.apache.pig.newplan.logical.relational.LOJoin;
@@ -117,6 +118,17 @@
     }
     
     @Override
+    public void visit(LORollupH2IRGForEach hforeach) throws FrontendException {
+        hforeach.resetSchema();
+        OperatorPlan innerPlan = hforeach.getInnerPlan();
+        PlanWalker newWalker = currentWalker.spawnChildWalker(innerPlan);
+        pushWalker(newWalker);
+        currentWalker.walk(this);
+        popWalker();
+        validate(hforeach.getSchema());
+    }
+    
+    @Override
     public void visit(LOGenerate gen) throws FrontendException {
         gen.resetSchema();
         visitAll(gen.getOutputPlans());
Index: src/org/apache/pig/newplan/logical/optimizer/UidResetter.java
===================================================================
--- src/org/apache/pig/newplan/logical/optimizer/UidResetter.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/optimizer/UidResetter.java	(working copy)
@@ -35,6 +35,7 @@
 import org.apache.pig.newplan.logical.relational.LODistinct;
 import org.apache.pig.newplan.logical.relational.LOFilter;
 import org.apache.pig.newplan.logical.relational.LOForEach;
+import org.apache.pig.newplan.logical.relational.LORollupH2IRGForEach;
 import org.apache.pig.newplan.logical.relational.LOGenerate;
 import org.apache.pig.newplan.logical.relational.LOInnerLoad;
 import org.apache.pig.newplan.logical.relational.LOJoin;
@@ -83,6 +84,16 @@
     }
     
     @Override
+    public void visit(LORollupH2IRGForEach hforeach) throws FrontendException {
+        hforeach.resetUid();
+        OperatorPlan innerPlan = hforeach.getInnerPlan();
+        PlanWalker newWalker = currentWalker.spawnChildWalker(innerPlan);
+        pushWalker(newWalker);
+        currentWalker.walk(this);
+        popWalker();
+    }
+    
+    @Override
     public void visit(LOForEach foreach) throws FrontendException {
         foreach.resetUid();
         OperatorPlan innerPlan = foreach.getInnerPlan();
Index: src/org/apache/pig/newplan/logical/relational/LOCogroup.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LOCogroup.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LOCogroup.java	(working copy)
@@ -71,6 +71,17 @@
      */
     public final static Integer OPTION_GROUPTYPE = 1;
     
+    private Integer pivot = null;
+    
+    public void setPivot(Integer pvt) {
+    	this.pivot = pvt;
+    	System.out.println("COG NEW PIVOT " + this.pivot);
+    }
+    
+    public Integer getPivot() {
+    	return this.pivot;
+    }
+    
     /**
      * Constructor for use in defining rule patterns
      * @param plan
Index: src/org/apache/pig/newplan/logical/relational/LOCube.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LOCube.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LOCube.java	(working copy)
@@ -84,6 +84,7 @@
 public class LOCube extends LogicalRelationalOperator {
     private MultiMap<Integer, LogicalExpressionPlan> mExpressionPlans;
     private List<String> operations;
+    private Integer pivot = null;
 
     public LOCube(LogicalPlan plan) {
 	super("LOCube", plan);
@@ -158,4 +159,13 @@
     public void setOperations(List<String> operations) {
 	this.operations = operations;
     }
+    
+    public void setPivot(Integer pvt) {
+    	this.pivot = pvt;
+    	System.out.println("NEW PIVOT " + this.pivot);
+    }
+    
+    public Integer getPivot() {
+    	return this.pivot;
+    }
 }
Index: src/org/apache/pig/newplan/logical/relational/LOForEach.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LOForEach.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LOForEach.java	(working copy)
@@ -34,13 +34,44 @@
 public class LOForEach extends LogicalRelationalOperator {
 
     private static final long serialVersionUID = 2L;
+    
+    private LogicalPlan innerPlan;
+    
+    /*private Integer pivot = null;
 
-    private LogicalPlan innerPlan;
+    private Integer rollup_position = 0;
+    
+    private boolean isOnlyIRG = false;*/
       
     public LOForEach(OperatorPlan plan) {
         super("LOForEach", plan);		
     }
 
+    /*public boolean getOnlyIRG() {
+    	return isOnlyIRG;
+    }
+    
+    public void setOnlyIRG() {
+    	isOnlyIRG = true;
+	}
+    
+    public void setRollupPosition(Integer ru_pos) {
+    	rollup_position = ru_pos;
+    }
+    
+    public Integer getRollupPosition() {
+    	return this.rollup_position;
+    }
+    
+    public void setPivot(Integer pvt) {
+    	this.pivot = pvt;
+    	System.out.println("LOFOREACH GOT PIVOT " + this.pivot);
+    }
+    
+    public Integer getPivot() {
+    	return this.pivot;
+    }*/
+    
     public LogicalPlan getInnerPlan() {
         return innerPlan;
     }
@@ -117,7 +148,7 @@
         NestedRelationalOperatorFinder opFinder = new NestedRelationalOperatorFinder(innerPlan, nestedAlias);
         opFinder.visit();
         
-        if (opFinder.getMatchedOperator()!=null) {
+        if (opFinder.getMatchedOperator() != null) {
             LogicalSchema nestedSc = opFinder.getMatchedOperator().getSchema();
             return nestedSc;
         }
@@ -127,16 +158,19 @@
     private static class NestedRelationalOperatorFinder extends AllSameRalationalNodesVisitor {
         String aliasOfOperator;
         LogicalRelationalOperator opFound = null;
+        
         public NestedRelationalOperatorFinder(LogicalPlan plan, String alias) throws FrontendException {
             super(plan, new ReverseDependencyOrderWalker(plan));
             aliasOfOperator = alias;
         }
+        
         public LogicalRelationalOperator getMatchedOperator() {
             return opFound;
         }
+        
         @Override
         public void execute(LogicalRelationalOperator op) throws FrontendException {
-            if (op.getAlias()!=null && op.getAlias().equals(aliasOfOperator))
+            if (op.getAlias() != null && op.getAlias().equals(aliasOfOperator))
                 opFound = op;
         }
     }
Index: src/org/apache/pig/newplan/logical/relational/LOGenerate.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LOGenerate.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LOGenerate.java	(working copy)
@@ -40,7 +40,7 @@
      // If LOGenerate generate new uid, cache it here.
      // This happens when expression plan does not have complete schema, however,
      // user give complete schema in ForEach statement in script
-     private List<LogicalSchema> uidOnlySchemas = null;
+     private List<LogicalSchema> uidOnlySchemas = null;     
 
     public LOGenerate(OperatorPlan plan, List<LogicalExpressionPlan> ps, boolean[] flatten) {
         this( plan );
Index: src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LogToPhyTranslationVisitor.java	(working copy)
@@ -49,6 +49,7 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POGlobalRearrange;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORollupH2IRGForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLimit;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
@@ -133,8 +134,7 @@
         String scope = DEFAULT_SCOPE;
         // The last parameter here is set to true as we assume all files are
         // splittable due to LoadStore Refactor
-        POLoad load = new POLoad(new OperatorKey(scope, nodeGen
-                .getNextNodeId(scope)), loLoad.getLoadFunc());
+        POLoad load = new POLoad(new OperatorKey(scope, nodeGen.getNextNodeId(scope)), loLoad.getLoadFunc());
         load.addOriginalLocation(loLoad.getAlias(), loLoad.getLocation());
         load.setLFile(loLoad.getFileSpec());
         load.setPc(pc);
@@ -776,16 +776,18 @@
 
     @Override
     public void visit(LOForEach foreach) throws FrontendException {
-        String scope = DEFAULT_SCOPE;
+        
+    	String scope = DEFAULT_SCOPE;
 
         List<PhysicalPlan> innerPlans = new ArrayList<PhysicalPlan>();
 
-        org.apache.pig.newplan.logical.relational.LogicalPlan inner = foreach.getInnerPlan();
+        LogicalPlan inner = foreach.getInnerPlan();
         LOGenerate gen = (LOGenerate)inner.getSinks().get(0);
-
+        
         List<LogicalExpressionPlan> exps = gen.getOutputPlans();
+        
+        
         List<Operator> preds = inner.getPredecessors(gen);
-
         currentPlans.push(currentPlan);
 
         // we need to translate each predecessor of LOGenerate into a physical plan.
@@ -893,6 +895,131 @@
         translateSoftLinks(foreach);
     }
 
+    @Override
+    public void visit(LORollupH2IRGForEach hforeach) throws FrontendException {
+        String scope = DEFAULT_SCOPE;
+
+        List<PhysicalPlan> innerPlans = new ArrayList<PhysicalPlan>();
+
+        org.apache.pig.newplan.logical.relational.LogicalPlan inner = hforeach.getInnerPlan();
+        LOGenerate gen = (LOGenerate) inner.getSinks().get(0);
+
+        List<LogicalExpressionPlan> exps = gen.getOutputPlans();
+        List<Operator> preds = inner.getPredecessors(gen);
+
+        currentPlans.push(currentPlan);
+
+        // we need to translate each predecessor of LOGenerate into a physical plan.
+        // The physical plan should contain the expression plan for this predecessor plus
+        // the subtree starting with this predecessor
+        for (int i = 0; i < exps.size(); i++) {
+            currentPlan = new PhysicalPlan();
+            // translate the expression plan
+            PlanWalker childWalker = new ReverseDependencyOrderWalkerWOSeenChk(exps.get(i));
+            pushWalker(childWalker);
+            childWalker.walk(new ExpToPhyTranslationVisitor(exps.get(i), childWalker, gen,
+                    currentPlan, logToPhyMap));
+            popWalker();
+
+            List<Operator> leaves = exps.get(i).getSinks();
+            for (Operator l : leaves) {
+                PhysicalOperator op = logToPhyMap.get(l);
+                if (l instanceof ProjectExpression) {
+                    int input = ((ProjectExpression) l).getInputNum();
+
+                    // for each sink projection, get its input logical plan and translate it
+                    Operator pred = preds.get(input);
+                    childWalker = new SubtreeDependencyOrderWalker(inner, pred);
+                    pushWalker(childWalker);
+                    childWalker.walk(this);
+                    popWalker();
+
+                    // get the physical operator of the leaf of input logical plan
+                    PhysicalOperator leaf = logToPhyMap.get(pred);
+
+                    if (pred instanceof LOInnerLoad) {
+                        // if predecessor is only an LOInnerLoad, remove the project that
+                        // comes from LOInnerLoad and change the column of project that
+                        // comes from expression plan
+                        currentPlan.remove(leaf);
+                        logToPhyMap.remove(pred);
+
+                        POProject leafProj = (POProject) leaf;
+                        try {
+                            if (leafProj.isStar()) {
+                                ((POProject) op).setStar(true);
+                            } else if (leafProj.isProjectToEnd()) {
+                                ((POProject) op).setProjectToEnd(leafProj.getStartCol());
+                            } else {
+                                ((POProject) op).setColumn(leafProj.getColumn());
+                            }
+
+                        } catch (ExecException e) {
+                            throw new FrontendException(hforeach, "Cannot get column from " + leaf,
+                                    2230, e);
+                        }
+
+                    } else {
+                        currentPlan.connect(leaf, op);
+                    }
+                }
+            }
+            innerPlans.add(currentPlan);
+        }
+
+        currentPlan = currentPlans.pop();
+
+        // PhysicalOperator poGen = new POGenerate(new OperatorKey("",
+        // r.nextLong()), inputs, toBeFlattened);
+        boolean[] flatten = gen.getFlattenFlags();
+        List<Boolean> flattenList = new ArrayList<Boolean>();
+        for (boolean fl : flatten) {
+            flattenList.add(fl);
+        }
+        PORollupH2IRGForEach poHFE = new PORollupH2IRGForEach(new OperatorKey(scope, nodeGen.getNextNodeId(scope)),
+                hforeach.getRequestedParallelism(), innerPlans, flattenList);
+        if(hforeach.getPivot()!=0 || hforeach.getPivot() == null)
+        	poHFE.setPivot(hforeach.getPivot()-1);
+        
+        if(hforeach.getPivot()!=null)
+        	if(hforeach.getPivot() == 0)
+        		poHFE.setPivot(0);
+        	else
+        		poHFE.setPivot(hforeach.getPivot() - 1);
+        else
+        	poHFE.setPivot(0);
+        
+        if(hforeach.getOnlyIRG())
+        	poHFE.setOnlyIRG();
+        
+        poHFE.setRollupPosition(hforeach.getRollupPosition());
+        poHFE.setRollupStartPosition(hforeach.getRollupStartPosition());
+        poHFE.setRollupSize(hforeach.getRollupSize());
+        
+        poHFE.addOriginalLocation(hforeach.getAlias(), hforeach.getLocation());
+        poHFE.setResultType(DataType.BAG);
+        logToPhyMap.put(hforeach, poHFE);
+        currentPlan.add(poHFE);
+
+        // generate cannot have multiple inputs
+        List<Operator> op = hforeach.getPlan().getPredecessors(hforeach);
+
+        // generate may not have any predecessors
+        if (op == null)
+            return;
+
+        PhysicalOperator from = logToPhyMap.get(op.get(0));
+        try {
+            currentPlan.connect(from, poHFE);
+        } catch (Exception e) {
+            int errCode = 2015;
+            String msg = "Invalid physical operators in the physical plan";
+            throw new LogicalToPhysicalTranslatorException(msg, errCode, PigException.BUG, e);
+        }
+
+        translateSoftLinks(hforeach);
+    }
+    
     /**
      * This function takes in a List of LogicalExpressionPlan and converts them to
      * a list of PhysicalPlans
Index: src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LogicalRelationalNodesVisitor.java	(working copy)
@@ -57,6 +57,9 @@
     public void visit(LOForEach foreach) throws FrontendException {
     }
 
+    public void visit(LORollupH2IRGForEach horeach) throws FrontendException {
+    }
+    
     public void visit(LOGenerate gen) throws FrontendException {
     }
 
Index: src/org/apache/pig/newplan/logical/relational/LogicalRelationalOperator.java
===================================================================
--- src/org/apache/pig/newplan/logical/relational/LogicalRelationalOperator.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/relational/LogicalRelationalOperator.java	(working copy)
@@ -132,7 +132,7 @@
     public void setRequestedParallelism(int parallel) {
         this.requestedParallelism = parallel;
     }
-
+    
     /**
      * Get the line number in the submitted Pig Latin script where this operator
      * occurred.
Index: src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java
===================================================================
--- src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java	(revision 1579421)
+++ src/org/apache/pig/newplan/logical/rules/OptimizerUtils.java	(working copy)
@@ -27,6 +27,7 @@
 import org.apache.pig.newplan.logical.expression.UserFuncExpression;
 import org.apache.pig.newplan.logical.relational.LOForEach;
 import org.apache.pig.newplan.logical.relational.LOGenerate;
+import org.apache.pig.newplan.logical.relational.LORollupH2IRGForEach;
 import org.apache.pig.newplan.logical.relational.LogicalPlan;
 
 public class OptimizerUtils {
@@ -40,6 +41,11 @@
         return (LOGenerate)inner.getSinks().get(0);
     }
 
+    public static LOGenerate findGenerate(LORollupH2IRGForEach hfe) {
+        LogicalPlan inner = hfe.getInnerPlan();
+        return (LOGenerate) inner.getSinks().get(0);
+    }
+    
     /**
      * Check if a given LOGenerate operator has any flatten fields.
      * @param gen the given LOGenerate instance
Index: src/org/apache/pig/parser/AliasMasker.g
===================================================================
--- src/org/apache/pig/parser/AliasMasker.g	(revision 1579421)
+++ src/org/apache/pig/parser/AliasMasker.g	(working copy)
@@ -247,6 +247,10 @@
     : ^( CUBE cube_item )
 ;
 
+pivot_clause
+	: ^( PIVOT INTEGER )
+;
+
 cube_item
     : rel ( cube_by_clause )
 ;
@@ -260,9 +264,21 @@
 ;
 
 cube_rollup_list
-    : ^( ( CUBE | ROLLUP ) cube_by_expr_list )
+    : ^( CUBE cube_by_expr_list ) | ^( ROLLUP cube_by_expr_list pivot_clause? )
 ;
 
+//cube_rollup_list
+//    : cube_list | ( rollup_list pivot_clause? ) 
+//;
+
+//cube_list
+//	: ^( CUBE cube_by_expr_list )
+//;
+
+//rollup_list
+//	: ^( ROLLUP cube_by_expr_list )
+//;
+
 cube_by_expr_list
     : cube_by_expr+
 ;
@@ -642,6 +658,7 @@
     | FOREACH
     | CUBE
     | ROLLUP
+    | PIVOT
     | MATCHES
     | ORDER
     | RANK
Index: src/org/apache/pig/parser/AstPrinter.g
===================================================================
--- src/org/apache/pig/parser/AstPrinter.g	(revision 1579421)
+++ src/org/apache/pig/parser/AstPrinter.g	(working copy)
@@ -106,6 +106,10 @@
     : ^( PARALLEL INTEGER ) { sb.append(" ").append($PARALLEL.text).append(" ").append($INTEGER.text); }
 ;
 
+pivot_clause
+	: ^( PIVOT INTEGER ) { sb.append(" ").append($PIVOT.text).append(" ").append($INTEGER.text); }
+;
+
 alias
     : IDENTIFIER { sb.append($IDENTIFIER.text); }
 ;
@@ -262,15 +266,27 @@
 ;
 
 cube_rollup_list
-    : ^( ( CUBE { sb.append($CUBE.text).append("("); } | ROLLUP { sb.append($ROLLUP.text).append("("); } ) cube_by_expr_list { sb.append(")"); })
+    : ^( CUBE { sb.append($CUBE.text).append("("); } cube_by_expr_list { sb.append(")"); } ) | ^( ROLLUP { sb.append($ROLLUP.text).append("("); } cube_by_expr_list { sb.append(")"); } )
 ;
 
+//cube_rollup_list
+//    : cube_list | ( rollup_list pivot_clause? )
+//;
+
+//cube_list
+//	: ^( CUBE { sb.append($CUBE.text).append("("); } cube_by_expr_list { sb.append(")"); })
+//;
+
+//rollup_list
+//	: ^( ROLLUP { sb.append($ROLLUP.text).append("("); } cube_by_expr_list { sb.append(")"); } )
+//;
+	
 cube_by_expr_list
     : ( cube_by_expr ( { sb.append(", "); } cube_by_expr )* )
 ;
 
 cube_by_expr
-    : col_range | expr | STAR { sb.append($STAR.text); }
+    : col_range | expr | STAR { sb.append($STAR.text); } { sb.append(" "); }
 ;
 
 group_clause
@@ -672,6 +688,7 @@
     | FOREACH   { sb.append($FOREACH.text); }
     | CUBE      { sb.append($CUBE.text); }
     | ROLLUP    { sb.append($ROLLUP.text); }
+    | PIVOT		{ sb.append($PIVOT.text); }
     | MATCHES   { sb.append($MATCHES.text); }
     | ORDER     { sb.append($ORDER.text); }
     | RANK      { sb.append($RANK.text); }
Index: src/org/apache/pig/parser/AstValidator.g
===================================================================
--- src/org/apache/pig/parser/AstValidator.g	(revision 1579421)
+++ src/org/apache/pig/parser/AstValidator.g	(working copy)
@@ -296,6 +296,10 @@
  : ^( CUBE cube_item )
 ;
 
+pivot_clause
+	: ^( PIVOT INTEGER )
+;
+
 cube_item
  : rel ( cube_by_clause )
 ;
@@ -309,9 +313,21 @@
 ;
 
 cube_rollup_list
- : ^( ( CUBE | ROLLUP ) cube_by_expr_list )
+ : ^( CUBE cube_by_expr_list ) | ^( ROLLUP cube_by_expr_list pivot_clause? )
 ;
 
+//cube_rollup_list
+//    : cube_list | ( rollup_list pivot_clause? )
+//;
+
+//cube_list
+//	: ^( CUBE cube_by_expr_list )
+//;
+
+//rollup_list
+//	: ^( ROLLUP cube_by_expr_list )
+//;
+
 cube_by_expr_list
  : cube_by_expr+
 ;
@@ -663,6 +679,7 @@
     | FOREACH
     | CUBE
     | ROLLUP
+    | PIVOT
     | MATCHES
     | ORDER
     | RANK
Index: src/org/apache/pig/parser/LogicalPlanBuilder.java
===================================================================
--- src/org/apache/pig/parser/LogicalPlanBuilder.java	(revision 1579421)
+++ src/org/apache/pig/parser/LogicalPlanBuilder.java	(working copy)
@@ -436,10 +436,28 @@
 	return new LOCube(plan);
     }
 
+    void setPivotRollupCubeOp(LOCube op, Integer pivot) throws ParserValidationException {
+    	if(pivot!=null)
+    		op.setPivot(pivot);
+    }
+    
+    
     String buildCubeOp(SourceLocation loc, LOCube op, String alias, String inputAlias,
 	    List<String> operations, MultiMap<Integer, LogicalExpressionPlan> expressionPlans)
 	    throws ParserValidationException {
 
+    // check value of pivot if it is valid or not
+    try {
+    	if(op.getPivot()!=null) {
+	    	if (op.getPivot() < 0 || op.getPivot() > expressionPlans.get(0).size()) {
+	    		FrontendException fe = new FrontendException("PIVOT is out of bound");
+	    		throw fe;
+	    	}
+    	}
+    } catch (FrontendException e) {
+    	throw new ParserValidationException(intStream, loc, e);
+    }
+    	
 	// check if continuously occurring cube operations be combined
 	combineCubeOperations((ArrayList<String>) operations, expressionPlans);
 
@@ -446,6 +464,7 @@
 	// set the expression plans for cube operator and build cube operator
 	op.setExpressionPlans(expressionPlans);
 	op.setOperations(operations);
+	//op.setPivot(pivot);
 	buildOp(loc, op, alias, inputAlias, null);
         try {
             (new ProjectStarExpander(op.getPlan())).visit(op);
@@ -644,6 +663,7 @@
 	    } else {
 		new UserFuncExpression(uexpPlan, new FuncSpec(RollupDimensions.class.getName()),
 		        lexpList);
+			
 	    }
 
 	    for (LogicalExpressionPlan lexp : lexpPlanList) {
@@ -676,6 +696,8 @@
 	    buildGenerateOp(loc, (LOForEach) foreach, (LOGenerate) gen, allExprPlan,
 		    flattenFlags, getUserDefinedSchema(allExprPlan));
 	    falias = buildForeachOp(loc, (LOForEach) foreach, "cube", inputAlias, innerPlan);
+		if(op.getPivot()!=null)
+			foreach.setPivot(op.getPivot());
 	} catch (ParserValidationException pve) {
 	    throw new FrontendException(pve);
 	}
@@ -698,6 +720,8 @@
 
 	// build group by operator
 	try {
+		if(op.getPivot()!=null)
+			groupby.setPivot(op.getPivot());
 	    return buildGroupOp(loc, (LOCogroup) groupby, op.getAlias(), inpAliases, exprPlansCopy,
 		    GROUPTYPE.REGULAR, innerFlags, null);
 	} catch (ParserValidationException pve) {
@@ -1272,7 +1296,7 @@
             op.setRequestedParallelism( pigContext.getExecType() == ExecType.LOCAL ? 1 : parallel );
         }
     }
-
+    
     static void setPartitioner(LogicalRelationalOperator op, String partitioner) {
         if( partitioner != null )
             op.setCustomPartitioner( partitioner );
Index: src/org/apache/pig/parser/LogicalPlanGenerator.g
===================================================================
--- src/org/apache/pig/parser/LogicalPlanGenerator.g	(revision 1579421)
+++ src/org/apache/pig/parser/LogicalPlanGenerator.g	(working copy)
@@ -498,6 +498,13 @@
 // It also outputs the order of operations i.e in this case CUBE operation followed by ROLLUP operation
 // These inputs are passed to buildCubeOp methods which then builds the logical plan for CUBE operator.
 // If user specifies STAR or RANGE expression for dimensions then it will be expanded inside buildCubeOp.
+pivot_clause returns[int pivot]
+ : ^( PIVOT INTEGER )
+   {
+       $pivot = Integer.parseInt( $INTEGER.text );
+   }
+;
+
 cube_clause returns[String alias]
 scope {
     LOCube cubeOp;
@@ -504,6 +511,7 @@
     MultiMap<Integer, LogicalExpressionPlan> cubePlans;
     List<String> operations;
     int inputIndex;
+    int pivot;
 }
 scope GScope;
 @init {
@@ -511,15 +519,17 @@
     $GScope::currentOp = $cube_clause::cubeOp;
     $cube_clause::cubePlans = new MultiMap<Integer, LogicalExpressionPlan>();
     $cube_clause::operations = new ArrayList<String>();
+    $cube_clause::pivot = -1;
 }
  : ^( CUBE cube_item )
    {
        SourceLocation loc = new SourceLocation( (PigParserNode)$cube_clause.start );
        $alias = builder.buildCubeOp( loc, $cube_clause::cubeOp, $statement::alias,
-       $statement::inputAlias, $cube_clause::operations, $cube_clause::cubePlans );
+       $statement::inputAlias, $cube_clause::operations, $cube_clause::cubePlans);
    }
 ;
 
+
 cube_item
  : rel ( cube_by_clause
    {
@@ -553,7 +563,7 @@
 @init {
     $plans = new ArrayList<LogicalExpressionPlan>();
 }
- : ^( ( CUBE { $operation = "CUBE"; } | ROLLUP { $operation = "ROLLUP"; } ) cube_by_expr_list { $plans = $cube_by_expr_list.plans; } )
+ : ^( CUBE { $operation = "CUBE"; } cube_by_expr_list { $plans = $cube_by_expr_list.plans; } ) | ^( ROLLUP { $operation = "ROLLUP"; } cube_by_expr_list { $plans = $cube_by_expr_list.plans; } pivot_clause? { if ($pivot_clause.tree!=null) builder.setPivotRollupCubeOp($cube_clause::cubeOp, $pivot_clause.pivot); } )
 ;
 
 cube_by_expr_list returns[List<LogicalExpressionPlan> plans]
@@ -1946,6 +1956,7 @@
     | COGROUP { $id = $COGROUP.text; }
     | CUBE { $id = $CUBE.text; }
     | ROLLUP { $id = $ROLLUP.text; }
+    | PIVOT { $id = $PIVOT.text; }
     | JOIN { $id = $JOIN.text; }
     | CROSS { $id = $CROSS.text; }
     | UNION { $id = $UNION.text; }
Index: src/org/apache/pig/parser/QueryLexer.g
===================================================================
--- src/org/apache/pig/parser/QueryLexer.g	(revision 1579421)
+++ src/org/apache/pig/parser/QueryLexer.g	(working copy)
@@ -153,6 +153,9 @@
 PARALLEL : 'PARALLEL'
 ;
 
+PIVOT : 'PIVOT'
+;
+
 PARTITION : 'PARTITION'
 ;
 
Index: src/org/apache/pig/parser/QueryParser.g
===================================================================
--- src/org/apache/pig/parser/QueryParser.g	(revision 1579421)
+++ src/org/apache/pig/parser/QueryParser.g	(working copy)
@@ -595,9 +595,12 @@
 cube_clause : CUBE rel BY cube_rollup_list ( COMMA cube_rollup_list )* -> ^( CUBE rel ^( BY cube_rollup_list+ ) )
 ;
 
-cube_rollup_list : ( CUBE | ROLLUP )^ LEFT_PAREN! real_arg ( COMMA! real_arg )* RIGHT_PAREN!
+cube_rollup_list : ( CUBE^ LEFT_PAREN! real_arg ( COMMA! real_arg )* RIGHT_PAREN! ) | ( ROLLUP^ LEFT_PAREN! real_arg ( COMMA! real_arg )* RIGHT_PAREN! pivot_clause? ) 
 ;
 
+pivot_clause : PIVOT^ INTEGER
+;
+
 flatten_clause : FLATTEN^ LEFT_PAREN! expr RIGHT_PAREN!
 ;
 
@@ -982,6 +985,7 @@
     | FILTER
     | FOREACH
     | ROLLUP
+    | PIVOT
     | ORDER
     | DISTINCT
     | COGROUP
Index: src/org/apache/pig/tools/pigstats/OutputStats.java
===================================================================
--- src/org/apache/pig/tools/pigstats/OutputStats.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/OutputStats.java	(working copy)
@@ -165,7 +165,7 @@
                     if (t == null) t = p.getNext();
                     if (t == null) atEnd = true;
                 } catch (Exception e) {
-                    LOG.error(e);
+                    //LOG.error(e);
                     t = null;
                     atEnd = true;
                     throw new Error(e);
@@ -182,7 +182,7 @@
                 try {
                     next = p.getNext();
                 } catch (Exception e) {
-                    LOG.error(e);
+                    //LOG.error(e);
                 }
                 if (next == null)
                     atEnd = true;
Index: src/org/apache/pig/tools/pigstats/PigStats.java
===================================================================
--- src/org/apache/pig/tools/pigstats/PigStats.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/PigStats.java	(working copy)
@@ -306,7 +306,7 @@
                 for (OutputStats os : iter.next().getOutputs()) {
                     String a = os.getAlias();
                     if (a == null || a.length() == 0) {
-                        LOG.warn("Output alias isn't avalable for " + os.getLocation());
+                        //LOG.warn("Output alias isn't avalable for " + os.getLocation());
                         continue;
                     }
                     aliasOuputMap.put(a, os);
@@ -385,7 +385,7 @@
             try {
                 jp.visit();
             } catch (FrontendException e) {
-                LOG.warn("unable to print job plan", e);
+                //LOG.warn("unable to print job plan", e);
             }
             return jp.toString();
         }
@@ -467,14 +467,13 @@
     @Private
     public void setBackendException(String jobId, Exception e) {
         if (e instanceof PigException) {
-            LOG.error("ERROR " + ((PigException)e).getErrorCode() + ": "
-                    + e.getLocalizedMessage());
+            //LOG.error("ERROR " + ((PigException)e).getErrorCode() + ": " + e.getLocalizedMessage());
         } else if (e != null) {
-            LOG.error("ERROR: " + e.getLocalizedMessage());
+            //LOG.error("ERROR: " + e.getLocalizedMessage());
         }
 
         if (jobId == null || e == null) {
-            LOG.debug("unable to set backend exception");
+            //LOG.debug("unable to set backend exception");
             return;
         }
         Iterator<JobStats> iter = jobPlan.iterator();
Index: src/org/apache/pig/tools/pigstats/ScriptState.java
===================================================================
--- src/org/apache/pig/tools/pigstats/ScriptState.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/ScriptState.java	(working copy)
@@ -185,7 +185,7 @@
         try {
             setScript(new BufferedReader(new FileReader(file)));
         } catch (FileNotFoundException e) {
-            LOG.warn("unable to find the file", e);
+            //LOG.warn("unable to find the file", e);
         }
     }
 
@@ -215,12 +215,11 @@
         try {
             new LogicalPlanFeatureVisitor(plan, bs).visit();
         } catch (FrontendException e) {
-            LOG.warn("unable to get script feature", e);
+            //LOG.warn("unable to get script feature", e);
         }
         scriptFeatures = bitSetToLong(bs);
 
-        LOG.info("Pig features used in the script: "
-                + featureLongToString(scriptFeatures));
+        //LOG.info("Pig features used in the script: "+ featureLongToString(scriptFeatures));
     }
 
     public String getHadoopVersion() {
@@ -242,10 +241,10 @@
                     Attributes attr = attrs.get("org/apache/pig");
                     pigVersion = attr.getValue("Implementation-Version");
                 } catch (Exception e) {
-                    LOG.warn("unable to read pigs manifest file");
+                    //LOG.warn("unable to read pigs manifest file");
                 }
             } else {
-                LOG.warn("unable to read pigs manifest file. Not running from the Pig jar");
+                //LOG.warn("unable to read pigs manifest file. Not running from the Pig jar");
             }
         }
         return (pigVersion == null) ? "" : pigVersion;
@@ -285,7 +284,7 @@
                 line = reader.readLine();
             }
         } catch (IOException e) {
-            LOG.warn("unable to parse the script", e);
+            //LOG.warn("unable to parse the script", e);
         }
         setScript(sb.toString());
     }
Index: src/org/apache/pig/tools/pigstats/mapreduce/MRPigStatsUtil.java
===================================================================
--- src/org/apache/pig/tools/pigstats/mapreduce/MRPigStatsUtil.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/mapreduce/MRPigStatsUtil.java	(working copy)
@@ -101,7 +101,7 @@
                 value = counter.getValue();
             }
         } catch (IOException e) {
-            LOG.warn("Failed to get the counter for " + counterName, e);
+            //LOG.warn("Failed to get the counter for " + counterName, e);
         }
         return value;
     }
@@ -182,10 +182,10 @@
         SimplePigStats ps = (SimplePigStats)PigStats.get();
         ps.finish();
         if (!ps.isSuccessful()) {
-            LOG.error(ps.getNumberFailedJobs() + " map reduce job(s) failed!");
+            //LOG.error(ps.getNumberFailedJobs() + " map reduce job(s) failed!");
             String errMsg = ps.getErrorMessage();
             if (errMsg != null) {
-                LOG.error("Error message: " + errMsg);
+                //LOG.error("Error message: " + errMsg);
             }
         }
         MRScriptState.get().emitLaunchCompletedNotification(
@@ -266,7 +266,7 @@
 
         MRJobStats js = ps.addMRJobStats(job);
         if (js == null) {
-            LOG.warn("unable to add failed job stats");
+            //LOG.warn("unable to add failed job stats");
         } else {
             js.setSuccessful(false);
             js.addOutputStatistics();
@@ -287,7 +287,7 @@
         }
         MRJobStats js = ((SimplePigStats)ps).addMRJobStatsForNative(mr);
         if(js == null) {
-            LOG.warn("unable to add native job stats");
+            //LOG.warn("unable to add native job stats");
         } else {
             js.setSuccessful(success);
             if(e != null)
@@ -301,7 +301,7 @@
 
         MRJobStats js = ps.addMRJobStats(job);
         if (js == null) {
-            LOG.warn("unable to add job stats");
+            //LOG.warn("unable to add job stats");
         } else {
             js.setSuccessful(true);
 
@@ -312,14 +312,14 @@
             try {
                 rjob = client.getJob(job.getAssignedJobID());
             } catch (IOException e) {
-                LOG.warn("Failed to get running job", e);
+                //LOG.warn("Failed to get running job", e);
             }
-            if (rjob == null) {
-                LOG.warn("Failed to get RunningJob for job "
+            /*if (rjob == null) {
+                //LOG.warn("Failed to get RunningJob for job "
                         + job.getAssignedJobID());
             } else {
                 js.addCounters(rjob);
-            }
+            }*/
 
             js.addOutputStatistics();
 
Index: src/org/apache/pig/tools/pigstats/mapreduce/MRScriptState.java
===================================================================
--- src/org/apache/pig/tools/pigstats/mapreduce/MRScriptState.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/mapreduce/MRScriptState.java	(working copy)
@@ -107,9 +107,7 @@
             try {
                 listener.initialPlanNotification(id, plan);
             } catch (NoSuchMethodError e) {
-                LOG.warn("PigProgressNotificationListener implementation doesn't "
-                       + "implement initialPlanNotification(..) method: "
-                       + listener.getClass().getName(), e);
+                //LOG.warn("PigProgressNotificationListener implementation doesn't "+ "implement initialPlanNotification(..) method: "+ listener.getClass().getName(), e);
             }
         }
     }
@@ -164,7 +162,7 @@
     }
     
     public void addSettingsToConf(MapReduceOper mro, Configuration conf) {
-        LOG.info("Pig script settings are added to the job");
+        //LOG.info("Pig script settings are added to the job");
         conf.set(PIG_PROPERTY.HADOOP_VERSION.toString(), getHadoopVersion());
         conf.set(PIG_PROPERTY.VERSION.toString(), getPigVersion());
         conf.set(PIG_PROPERTY.SCRIPT_ID.toString(), id);
@@ -179,7 +177,7 @@
             }                 
             conf.set(PIG_PROPERTY.MAP_OUTPUT_DIRS.toString(), LoadFunc.join(outputDirs, ","));
         } catch (VisitorException e) {
-            LOG.warn("unable to get the map stores", e);
+            //LOG.warn("unable to get the map stores", e);
         }
         if (!mro.reducePlan.isEmpty()) {
             try {
@@ -190,7 +188,7 @@
                 }                      
                 conf.set(PIG_PROPERTY.REDUCE_OUTPUT_DIRS.toString(), LoadFunc.join(outputDirs, ","));
             } catch (VisitorException e) {
-                LOG.warn("unable to get the reduce stores", e);
+                //LOG.warn("unable to get the reduce stores", e);
             }
         }        
         try {
@@ -203,7 +201,7 @@
                 conf.set(PIG_PROPERTY.INPUT_DIRS.toString(), LoadFunc.join(inputDirs, ","));       
             }
         } catch (VisitorException e) {
-            LOG.warn("unable to get the map loads", e);
+            //LOG.warn("unable to get the map loads", e);
         }
 
         setPigFeature(mro, conf);
@@ -292,7 +290,7 @@
                 Collections.sort(alias);
             }              
         } catch (VisitorException e) {
-            LOG.warn("unable to get alias", e);
+            //LOG.warn("unable to get alias", e);
         }
         aliasMap.put(mro, LoadFunc.join(alias, ","));
         aliasLocationMap.put(mro, aliasLocationStr);
@@ -354,7 +352,7 @@
                         new FeatureVisitor(mro.reducePlan, feature).visit();
                     }
                 } catch (VisitorException e) {
-                    LOG.warn("Feature visitor failed", e);
+                    //LOG.warn("Feature visitor failed", e);
                 }
             }
             StringBuilder sb = new StringBuilder();
Index: src/org/apache/pig/tools/pigstats/mapreduce/SimplePigStats.java
===================================================================
--- src/org/apache/pig/tools/pigstats/mapreduce/SimplePigStats.java	(revision 1579421)
+++ src/org/apache/pig/tools/pigstats/mapreduce/SimplePigStats.java	(working copy)
@@ -148,7 +148,7 @@
         super.start();
 
         if (pigContext == null || jobClient == null || jcc == null) {
-            LOG.warn("invalid params: " + pigContext + jobClient + jcc);
+            //LOG.warn("invalid params: " + pigContext + jobClient + jcc);
             return;
         }
 
@@ -160,7 +160,7 @@
         try {
             new JobGraphBuilder(mrPlan).visit();
         } catch (VisitorException e) {
-            LOG.warn("unable to build job plan", e);
+            //LOG.warn("unable to build job plan", e);
         }
     }
 
@@ -185,7 +185,7 @@
         MapReduceOper mro = jobMroMap.get(job);
 
         if (mro == null) {
-            LOG.warn("unable to get MR oper for job: " + job.toString());
+            //LOG.warn("unable to get MR oper for job: " + job.toString());
             return null;
         }
         MRJobStats js = mroJobMap.get(mro);
@@ -207,11 +207,11 @@
 
     void display() {
         if (returnCode == ReturnCode.UNKNOWN) {
-            LOG.warn("unknown return code, can't display the results");
+            //LOG.warn("unknown return code, can't display the results");
             return;
         }
         if (pigContext == null) {
-            LOG.warn("unknown exec type, don't display the results");
+            //LOG.warn("unknown exec type, don't display the results");
             return;
         }
 
@@ -218,7 +218,7 @@
         // currently counters are not working in local mode - see PIG-1286
         ExecType execType = pigContext.getExecType();
         if (execType.isLocal()) {
-            LOG.info("Detected Local mode. Stats reported below may be incomplete");
+            //LOG.info("Detected Local mode. Stats reported below may be incomplete");
         }
 
         SimpleDateFormat sdf = new SimpleDateFormat(DATE_FORMAT);
@@ -287,16 +287,16 @@
 
         sb.append("\nJob DAG:\n").append(jobPlan.toString());
 
-        LOG.info("Script Statistics: \n" + sb.toString());
+        //LOG.info("Script Statistics: \n" + sb.toString());
     }
 
     void mapMROperToJob(MapReduceOper mro, Job job) {
         if (mro == null) {
-            LOG.warn("null MR operator");
+            //LOG.warn("null MR operator");
         } else {
             MRJobStats js = mroJobMap.get(mro);
             if (js == null) {
-                LOG.warn("null job stats for mro: " + mro.getOperatorKey());
+                //LOG.warn("null job stats for mro: " + mro.getOperatorKey());
             } else {
                 jobMroMap.put(job, mro);
             }
Index: src/pig-default.properties
===================================================================
--- src/pig-default.properties	(revision 1579421)
+++ src/pig-default.properties	(working copy)
@@ -31,7 +31,7 @@
 exectype=mapreduce
 
 #Enable insertion of information about script into hadoop job conf 
-pig.script.info.enabled=true
+pig.script.info.enabled=false
 
 #Do not spill temp files smaller than this size (bytes)
 pig.spill.size.threshold=5000000
@@ -50,7 +50,7 @@
 pig.files.concatenation.threshold=100
 pig.optimistic.files.concatenation=false;
 
-pig.disable.counter=false
+pig.disable.counter=true
 
 pig.sql.type=hcat
 
Index: test/org/apache/pig/parser/TestQueryParser.java
===================================================================
--- test/org/apache/pig/parser/TestQueryParser.java	(revision 1579421)
+++ test/org/apache/pig/parser/TestQueryParser.java	(working copy)
@@ -242,8 +242,7 @@
     				   "store z into 'cube_output';";
     	shouldPass( query );
     }
-    
-    
+        
     @Test
     public void testCubePositive3() throws IOException, RecognitionException {
 	// range projection
Index: test/org/apache/pig/test/TestCubeOperator.java
===================================================================
--- test/org/apache/pig/test/TestCubeOperator.java	(revision 1579421)
+++ test/org/apache/pig/test/TestCubeOperator.java	(working copy)
@@ -142,6 +142,34 @@
     }
 
     @Test
+    public void testRollupH2IRGBasic() throws IOException {
+	// basic correctness test
+	String query = "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);"
+	        + "b = cube a by rollup(x,y) pivot 1;"
+	        + "c = foreach b generate flatten(group) as (type,location), COUNT_STAR(cube) as count, SUM(cube.z) as total;"
+	        + "store c into 'output' using mock.Storage();";
+	Util.registerMultiLineQuery(pigServer, query);
+
+	Set<Tuple> expected = ImmutableSet.of(
+	        tf.newTuple(Lists.newArrayList("cat", "miami", (long) 1, (long) 18)),
+	        tf.newTuple(Lists.newArrayList("cat", "naples", (long) 1, (long) 9)),
+	        tf.newTuple(Lists.newArrayList("cat", null, (long) 2, (long) 27)),
+	        tf.newTuple(Lists.newArrayList("dog", "miami", (long) 1, (long) 12)),
+	        tf.newTuple(Lists.newArrayList("dog", "tampa", (long) 1, (long) 14)),
+	        tf.newTuple(Lists.newArrayList("dog", "naples", (long) 1, (long) 5)),
+	        tf.newTuple(Lists.newArrayList("dog", null, (long) 3, (long) 31)),
+	        tf.newTuple(Lists.newArrayList("turtle", "tampa", (long) 1, (long) 4)),
+	        tf.newTuple(Lists.newArrayList("turtle", "naples", (long) 1, (long) 1)),
+	        tf.newTuple(Lists.newArrayList("turtle", null, (long) 2, (long) 5)),
+	        tf.newTuple(Lists.newArrayList(null, null, (long) 7, (long) 63)));
+
+	List<Tuple> out = data.get("output");
+	for (Tuple tup : out) {
+	    assertTrue(expected + " contains " + tup, expected.contains(tup));
+	}
+    }
+    
+    @Test
     public void testCubeAndRollup() throws IOException {
 	// basic correctness test
 	String query = "a = load 'input2' USING mock.Storage() as (v:chararray,w:chararray,x:chararray,y:chararray,z:long);"
@@ -174,6 +202,38 @@
     }
 
     @Test
+    public void testCubeAndRollupH2IRG() throws IOException {
+	// basic correctness test
+	String query = "a = load 'input2' USING mock.Storage() as (v:chararray,w:chararray,x:chararray,y:chararray,z:long);"
+	        + "b = cube a by cube(v,w), rollup(x,y) pivot 1;"
+	        + "c = foreach b generate flatten(group) as (type,location,color,category), COUNT_STAR(cube) as count, SUM(cube.z) as total;"
+	        + "store c into 'output' using mock.Storage();";
+	Util.registerMultiLineQuery(pigServer, query);
+
+	Set<Tuple> expected = ImmutableSet
+	        .of(tf.newTuple(Lists.newArrayList("dog", "miami", "white", "pet", (long) 1,
+	                (long) 5)), tf.newTuple(Lists.newArrayList("dog", null, "white", "pet",
+	                (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList(null, "miami",
+	                "white", "pet", (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList(null,
+	                null, "white", "pet", (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList(
+	                "dog", "miami", "white", null, (long) 1, (long) 5)), tf.newTuple(Lists
+	                .newArrayList("dog", null, "white", null, (long) 1, (long) 5)), tf
+	                .newTuple(Lists.newArrayList(null, "miami", "white", null, (long) 1,
+	                        (long) 5)), tf.newTuple(Lists.newArrayList(null, null, "white",
+	                null, (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList("dog", "miami",
+	                null, null, (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList("dog",
+	                null, null, null, (long) 1, (long) 5)), tf.newTuple(Lists.newArrayList(
+	                null, "miami", null, null, (long) 1, (long) 5)), tf.newTuple(Lists
+	                .newArrayList(null, null, null, null, (long) 1, (long) 5)));
+
+	List<Tuple> out = data.get("output");
+	for (Tuple tup : out) {
+	    assertTrue(expected + " contains " + tup, expected.contains(tup));
+	}
+
+    }
+    
+    @Test
     public void testCubeMultipleIAliases() throws IOException {
 	// test for input alias to cube being assigned multiple times
 	String query = "a = load 'input' USING mock.Storage() as (x:chararray,y:chararray,z:long);"
@@ -560,7 +620,37 @@
 	    assertTrue(expected + " contains " + tup, expected.contains(tup));
 	}
     }
+    
+    @Test
+    public void testRollupH2IRGAfterCogroup() throws IOException {
+	// test for cubing on co-grouped relation
+	String query = "a = load 'input1' USING mock.Storage() as (a1:chararray,b1,c1,d1); "
+	        + "b = load 'input' USING mock.Storage() as (a2,b2,c2:long,d2:chararray);"
+	        + "c = cogroup a by a1, b by d2;"
+	        + "d = foreach c generate flatten(a), flatten(b);"
+	        + "e = cube d by rollup(a2,b2) pivot 1;"
+	        + "f = foreach e generate flatten(group), COUNT(cube) as count, SUM(cube.c2) as total;"
+	        + "store f into 'output' using mock.Storage();";
 
+	Util.registerMultiLineQuery(pigServer, query);
+
+	Set<Tuple> expected = ImmutableSet.of(
+	        tf.newTuple(Lists.newArrayList("cat", "miami", (long) 1, (long) 18)),
+	        tf.newTuple(Lists.newArrayList("cat", null, (long) 1, (long) 18)),
+	        tf.newTuple(Lists.newArrayList("dog", "miami", (long) 1, (long) 12)),
+	        tf.newTuple(Lists.newArrayList("dog", "tampa", (long) 1, (long) 14)),
+	        tf.newTuple(Lists.newArrayList("dog", null, (long) 2, (long) 26)),
+	        tf.newTuple(Lists.newArrayList("turtle", "tampa", (long) 1, (long) 4)),
+	        tf.newTuple(Lists.newArrayList("turtle", "naples", (long) 1, (long) 1)),
+	        tf.newTuple(Lists.newArrayList("turtle", null, (long) 2, (long) 5)),
+	        tf.newTuple(Lists.newArrayList(null, null, (long) 5, (long) 49)));
+
+	List<Tuple> out = data.get("output");
+	for (Tuple tup : out) {
+	    assertTrue(expected + " contains " + tup, expected.contains(tup));
+	}
+    }
+
     @Test
     public void testIllustrate() throws IOException {
 	// test for illustrate
@@ -599,6 +689,19 @@
     }
 
     @Test
+    public void testExplainRollupH2IRG() throws IOException {
+	// test for explain
+	String query = "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); "
+	        + "b = cube a by rollup(a1,b1) pivot 1;";
+
+	Util.registerMultiLineQuery(pigServer, query);
+	ByteArrayOutputStream baos = new ByteArrayOutputStream();
+	PrintStream ps = new PrintStream(baos);
+	pigServer.explain("b", ps);
+	assertTrue(baos.toString().contains("RollupDimensions"));
+    }
+    
+    @Test
     public void testDescribe() throws IOException {
 	// test for describe
 	String query = "a = load 'input' USING mock.Storage() as (a1:chararray,b1:chararray,c1:long); "
Index: test/org/apache/pig/test/TestNewPlanLogToPhyTranslationVisitor.java
===================================================================
--- test/org/apache/pig/test/TestNewPlanLogToPhyTranslationVisitor.java	(revision 1579421)
+++ test/org/apache/pig/test/TestNewPlanLogToPhyTranslationVisitor.java	(working copy)
@@ -46,6 +46,7 @@
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.plans.PhysicalPlan;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POFilter;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POForEach;
+import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.PORollupH2IRGForEach;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POGlobalRearrange;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLoad;
 import org.apache.pig.backend.hadoop.executionengine.physicalLayer.relationalOperators.POLocalRearrange;
Index: test/org/apache/pig/test/utils/GenPhyOp.java
===================================================================
--- test/org/apache/pig/test/utils/GenPhyOp.java	(revision 1579421)
+++ test/org/apache/pig/test/utils/GenPhyOp.java	(working copy)
@@ -181,6 +181,12 @@
         return ret;
     }
 
+    public static PORollupH2IRGForEach topForEachOp() {
+        PORollupH2IRGForEach ret = new PORollupH2IRGForEach(new OperatorKey("", r
+                .nextLong()));
+        return ret;
+    }
+    
     public static POUnion topUnionOp() {
         POUnion ret = new POUnion(new OperatorKey("", r.nextLong()));
         return ret;
@@ -515,6 +521,16 @@
         return ret;
     }
 
+    public static PORollupH2IRGForEach topRollupH2IRGForEachOPWithPlan(int field, Tuple sample) throws ExecException, PlanException{
+        PlansAndFlattens pf = topGenerateOpWithExPlanForFe(field, sample);
+        
+        PORollupH2IRGForEach ret = topRollupH2IRGForEachOp();
+        ret.setInputPlans(pf.plans);
+        ret.setToBeFlattened(pf.flattens);
+        ret.setResultType(DataType.TUPLE);
+        return ret;
+    }
+    
     /**
      * creates the POForEach operator for
      * foreach A generate field[0] field[1]
@@ -533,6 +549,16 @@
         return ret;
     }
     
+    public static PORollupH2IRGForEach topRollupH2IRGForEachOPWithPlan(int[] fields, Tuple sample) throws ExecException, PlanException{
+        PlansAndFlattens pf = topGenerateOpWithExPlanForFe(fields, sample);
+        
+        PORollupH2IRGForEach ret = topRollupH2IRGForEachOp();
+        ret.setInputPlans(pf.plans);
+        ret.setToBeFlattened(pf.flattens);
+        ret.setResultType(DataType.TUPLE);
+        return ret;
+    }
+    
     /**
      * creates the POForEach operator for
      * foreach A generate field[0] field[1]
@@ -555,6 +581,20 @@
         return ret;
     }
     
+    public static PORollupH2IRGForEach topRollupH2IRGForEachOPWithPlan(
+            int[] fields,
+            Tuple sample,
+            List<Boolean> toBeFlattened) throws ExecException, PlanException{
+        PlansAndFlattens pf =
+            topGenerateOpWithExPlanForFe(fields, sample, toBeFlattened);
+        
+        PORollupH2IRGForEach ret = topRollupH2IRGForEachOp();
+        ret.setInputPlans(pf.plans);
+        ret.setToBeFlattened(toBeFlattened);
+        ret.setResultType(DataType.TUPLE);
+        return ret;
+    }
+    
     /**
      * creates the POForEach operator for
      * foreach A generate flatten(field)
@@ -572,6 +612,19 @@
         ret.setResultType(DataType.TUPLE);
         return ret;
     }
+    
+    public static PORollupH2IRGForEach topRollupH2IRGForEachOPWithPlan(int field) throws ExecException, PlanException{
+        PlansAndFlattens pf = topGenerateOpWithExPlanForFeFlat(field);
+        
+        PORollupH2IRGForEach ret = topRollupH2IRGForEachOp();
+        ret.setInputPlans(pf.plans);
+        ret.setToBeFlattened(pf.flattens);
+        ret.setResultType(DataType.TUPLE);
+        return ret;
+    }
+    
+    //ADD
+    
 
     
     public static POLoad topLoadOp() {
Index: tutorial/src/org/apache/pig/tutorial/NGramGenerator.java
===================================================================
--- tutorial/src/org/apache/pig/tutorial/NGramGenerator.java	(revision 1579421)
+++ tutorial/src/org/apache/pig/tutorial/NGramGenerator.java	(working copy)
@@ -76,8 +76,7 @@
          Schema bagSchema = new Schema();
          bagSchema.add(new Schema.FieldSchema("ngram", DataType.CHARARRAY));
          try{
-            return new Schema(new Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input), 
-                                                    bagSchema, DataType.BAG));
+            return new Schema(new Schema.FieldSchema(getSchemaName(this.getClass().getName().toLowerCase(), input),                                    bagSchema, DataType.BAG));
          }catch (FrontendException e){
             return null;
          }