You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pig.apache.org by da...@apache.org on 2015/08/05 23:04:43 UTC

svn commit: r1694329 - in /pig/trunk: CHANGES.txt contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java

Author: daijy
Date: Wed Aug  5 21:04:43 2015
New Revision: 1694329

URL: http://svn.apache.org/r1694329
Log:
PIG-4623: Fixed the 'new line' character inside double-quote causing the csv parsing failure

Modified:
    pig/trunk/CHANGES.txt
    pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java
    pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java

Modified: pig/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/pig/trunk/CHANGES.txt?rev=1694329&r1=1694328&r2=1694329&view=diff
==============================================================================
--- pig/trunk/CHANGES.txt (original)
+++ pig/trunk/CHANGES.txt Wed Aug  5 21:04:43 2015
@@ -38,6 +38,8 @@ PIG-4639: Add better parser for Apache H
 
 BUG FIXES
 
+PIG-4623: Fixed the 'new line' character inside double-quote causing the csv parsing failure (ken11223 via daijy)
+
 PIG-4649: [Pig on Tez] Union followed by HCatStorer misses some data (rohini)
 
 PIG-4636: Occurred spelled incorrectly in error message for Launcher and POMergeCogroup (stevenmz via daijy)

Modified: pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java?rev=1694329&r1=1694328&r2=1694329&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java (original)
+++ pig/trunk/contrib/piggybank/java/src/main/java/org/apache/pig/piggybank/storage/CSVLoader.java Wed Aug  5 21:04:43 2015
@@ -87,6 +87,25 @@ public class CSVLoader extends FileInput
     public CSVLoader() {
     }
 
+    /*
+     * Merge two ByteBuffer
+     * Precondition is both input buffer cannot be null nor empty
+     */
+    private ByteBuffer enlargeBuffer(ByteBuffer a, int extraLen, boolean putSpaceInBetween) {
+    	int totalLen = a.capacity() + extraLen;
+    	if (putSpaceInBetween) {
+    		totalLen++;
+    	}
+    	ByteBuffer c = ByteBuffer.allocate(totalLen);
+    	for(int i=0; i<a.position(); i++) {
+    		c.put(a.get(i));
+    	}
+    	if (putSpaceInBetween) {
+    		c.put((byte) 32);	//Put a space afterwards
+    	}
+    	return c;
+    }
+    
     @Override
     public Tuple getNext() throws IOException {
         mProtoTuple = new ArrayList<Object>();
@@ -103,46 +122,64 @@ public class CSVLoader extends FileInput
             mRequiredColumnsInitialized = true;
         }
         try {
-            if (!in.nextKeyValue()) {
-                return null;
-            }                                                                                           
-            Text value = (Text) in.getCurrentValue();
-            byte[] buf = value.getBytes();
-            int len = value.getLength();
-            int fieldID = 0;
-
-            ByteBuffer fieldBuffer = ByteBuffer.allocate(len);
-
-            for (int i = 0; i < len; i++) {
-                byte b = buf[i];
-                inField = true;
-                if (inQuotedField) {
-                    if (b == DOUBLE_QUOTE) {
-                        evenQuotesSeen = !evenQuotesSeen;
-                        if (evenQuotesSeen) {
-                            fieldBuffer.put(DOUBLE_QUOTE);
-                        }
-                    } else
-                        if (!evenQuotesSeen &&
-                                (b == FIELD_DEL || b == RECORD_DEL)) {
-                            inQuotedField = false;
-                            inField = false;
-                            readField(fieldBuffer, fieldID++);
-                        } else {
-                            fieldBuffer.put(b);
-                        }
-                } else if (b == DOUBLE_QUOTE) {
-                    inQuotedField = true;
-                    evenQuotesSeen = true;
-                } else if (b == FIELD_DEL) {
-                    inField = false;
-                    readField(fieldBuffer, fieldID++); // end of the field
-                } else {
-                    evenQuotesSeen = true;
-                    fieldBuffer.put(b);
-                }
-            }
-            if (inField) readField(fieldBuffer, fieldID++);
+        	boolean doneThisLineLogically = false;
+        	ByteBuffer previousBufferToBeRead = null;
+        	int fieldID = 0;
+        	while(!doneThisLineLogically)	{
+	            if (!in.nextKeyValue()) {
+	                return null;
+	            }
+	            Text value = (Text) in.getCurrentValue();
+	            byte[] buf = value.getBytes();
+	            int len = value.getLength();
+	            ByteBuffer fieldBuffer = null;
+	            if (previousBufferToBeRead != null) {
+	            	fieldBuffer = enlargeBuffer(previousBufferToBeRead, len, true);
+	            } else {
+	            	fieldBuffer = ByteBuffer.allocate(len);	
+	            }
+	
+	            for (int i = 0; i < len; i++) {
+	                byte b = buf[i];
+	                inField = true;
+	                if (inQuotedField) {
+	                    if (b == DOUBLE_QUOTE) {
+	                        evenQuotesSeen = !evenQuotesSeen;
+	                        if (evenQuotesSeen) {
+	                            fieldBuffer.put(DOUBLE_QUOTE);
+	                        }
+	                    } else
+	                        if (!evenQuotesSeen &&
+	                                (b == FIELD_DEL || b == RECORD_DEL)) {
+	                            inQuotedField = false;
+	                            inField = false;
+	                            readField(fieldBuffer, fieldID++);
+	                        } else {
+	                            fieldBuffer.put(b);
+	                        }
+	                } else if (b == DOUBLE_QUOTE) {
+	                    inQuotedField = true;
+	                    evenQuotesSeen = true;
+	                } else if (b == FIELD_DEL) {
+	                    inField = false;
+	                    readField(fieldBuffer, fieldID++); // end of the field
+	                } else {
+	                    evenQuotesSeen = true;
+	                    fieldBuffer.put(b);
+	                }
+	            }
+	            doneThisLineLogically = true;
+	            
+	            if (inField)  {
+	            	if (inQuotedField && evenQuotesSeen) {
+	            		doneThisLineLogically = false;
+	            		previousBufferToBeRead = fieldBuffer;
+	            	} else {
+	            		readField(fieldBuffer, fieldID++);
+	            	}
+	            }
+	            
+        	}	//End of while loo\
         } catch (InterruptedException e) {
             int errCode = 6018;
             String errMsg = "Error while reading input";
@@ -153,7 +190,7 @@ public class CSVLoader extends FileInput
         Tuple t =  mTupleFactory.newTupleNoCopy(mProtoTuple);
         return t;
     }
-
+    
     private void readField(ByteBuffer buf, int fieldID) {
         if (mRequiredColumns==null || (mRequiredColumns.length>fieldID && mRequiredColumns[fieldID])) {
             byte[] bytes = new byte[buf.position()];

Modified: pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java
URL: http://svn.apache.org/viewvc/pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java?rev=1694329&r1=1694328&r2=1694329&view=diff
==============================================================================
--- pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java (original)
+++ pig/trunk/contrib/piggybank/java/src/test/java/org/apache/pig/piggybank/test/storage/TestCSVStorage.java Wed Aug  5 21:04:43 2015
@@ -30,7 +30,6 @@ import org.apache.pig.PigServer;
 import org.apache.pig.backend.executionengine.ExecException;
 import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.MRConfiguration;
 import org.apache.pig.data.Tuple;
-import org.apache.pig.test.MiniCluster;
 import org.apache.pig.test.Util;
 import org.junit.Test;
 
@@ -38,10 +37,8 @@ public class TestCSVStorage {
     protected static final Log LOG = LogFactory.getLog(TestCSVStorage.class);
     
     private PigServer pigServer;
-    private MiniCluster cluster;
     
     public TestCSVStorage() throws ExecException, IOException {
-        cluster = MiniCluster.buildCluster();
         pigServer = new PigServer(ExecType.LOCAL, new Properties());
         pigServer.getPigContext().getProperties()
                 .setProperty(MRConfiguration.MAP_MAX_ATTEMPTS, "1");
@@ -85,4 +82,20 @@ public class TestCSVStorage {
         assertEquals(Util.createTuple(new String[] {"\"\"\""}), it.next());
     }
     
+    /*
+     * See PIG-4623
+     */
+    @Test
+    public void testQuotedQuotesWithNewLines() throws IOException {
+        String inputFileName = "TestCSVLoader-quotedquotes.txt";
+        Util.createLocalInputFile(inputFileName, 
+                new String[] {"\"foo \n ,\"\"bar\"\",baz\"", "\"\"\"\"\"\"\"\""});
+        String script = "a = load '" + inputFileName + "' using org.apache.pig.piggybank.storage.CSVLoader() " +
+        "   as (a:chararray); ";
+        Util.registerMultiLineQuery(pigServer, script);
+        Iterator<Tuple> it = pigServer.openIterator("a");
+        assertEquals(Util.createTuple(new String[] {"foo   ,\"bar\",baz"}), it.next());
+        assertEquals(Util.createTuple(new String[] {"\"\"\""}), it.next());
+    }
+    
 }