hadoop-hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From nzh...@apache.org
Subject svn commit: r903109 - in /hadoop/hive/branches/branch-0.5: CHANGES.txt ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java
Date Tue, 26 Jan 2010 07:36:43 GMT
Author: nzhang
Date: Tue Jan 26 07:36:42 2010
New Revision: 903109

URL: http://svn.apache.org/viewvc?rev=903109&view=rev
Log:
HIVE-1088: RCFile RecordReaders first split will read duplicate rows if the split end is <
the first SYNC mark (Yongqiang He via Ning Zhang)

Modified:
    hadoop/hive/branches/branch-0.5/CHANGES.txt
    hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java
    hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java
    hadoop/hive/branches/branch-0.5/ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java

Modified: hadoop/hive/branches/branch-0.5/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/hive/branches/branch-0.5/CHANGES.txt?rev=903109&r1=903108&r2=903109&view=diff
==============================================================================
--- hadoop/hive/branches/branch-0.5/CHANGES.txt (original)
+++ hadoop/hive/branches/branch-0.5/CHANGES.txt Tue Jan 26 07:36:42 2010
@@ -1452,3 +1452,6 @@
 
     HIVE-283. Do case insensitive comparison of aliases in partition
     pruning. (athusoo)
+
+    HIVE-1088. RCFileRecordReader's first split will read duplicate rows if
+    the split end is < the first sync mark (Yongqiang He via Ning Zhang)

Modified: hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java
URL: http://svn.apache.org/viewvc/hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java?rev=903109&r1=903108&r2=903109&view=diff
==============================================================================
--- hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java (original)
+++ hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java Tue
Jan 26 07:36:42 2010
@@ -909,7 +909,9 @@
     private byte[] sync = new byte[SYNC_HASH_SIZE];
     private byte[] syncCheck = new byte[SYNC_HASH_SIZE];
     private boolean syncSeen;
+    private long lastSeenSyncPos = 0;
 
+    private long headerEnd;
     private long end;
     private int currentKeyLength;
     private int currentRecordLength;
@@ -1074,6 +1076,7 @@
 
       if (version > 1) { // if version > 1
         in.readFully(sync); // read sync bytes
+        headerEnd = in.getPos();
       }
     }
 
@@ -1102,6 +1105,15 @@
         seek(end);
         return;
       }
+   
+     //this is to handle syn(pos) where pos < headerEnd.
+     if (position < headerEnd) {
+       // seek directly to first record
+       in.seek(headerEnd);
+       // note the sync marker "seen" in the header
+       syncSeen = true;
+       return;
+     }
 
       try {
         seek(position + 4); // skip escape
@@ -1159,6 +1171,7 @@
       if (version > 1 && sync != null && length == SYNC_ESCAPE) { // process
         // a
         // sync entry
+        lastSeenSyncPos = in.getPos() - 4; // minus SYNC_ESCAPE's length
         in.readFully(syncCheck); // read syncCheck
         if (!Arrays.equals(sync, syncCheck)) // check it
           throw new IOException("File is corrupt!");
@@ -1403,6 +1416,12 @@
     public boolean syncSeen() {
       return syncSeen;
     }
+    
+   
+    /** Returns the last seen sync position */
+    public long lastSeenSyncPos() {
+      return lastSeenSyncPos;
+    }
 
     /** Returns the name of the file. */
     public String toString() {

Modified: hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java
URL: http://svn.apache.org/viewvc/hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java?rev=903109&r1=903108&r2=903109&view=diff
==============================================================================
--- hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java
(original)
+++ hadoop/hive/branches/branch-0.5/ql/src/java/org/apache/hadoop/hive/ql/io/RCFileRecordReader.java
Tue Jan 26 07:36:42 2010
@@ -71,39 +71,32 @@
         conf);
   }
 
-  private boolean firstSeen = true;
-
   @Override
   public boolean next(LongWritable key, BytesRefArrayWritable value)
       throws IOException {
-    if (!more)
-      return false;
-    long pos = in.getPosition();
-    boolean hasMore = in.next(key);
-    if (hasMore) {
+    
+    more = next(key);
+    
+    if (more) {
       in.getCurrentRow(value);
     }
-    if (pos >= end && in.syncSeen() && !in.hasRecordsInBuffer()) {
-      more = false;
-      if (firstSeen) {
-        firstSeen = false;
-        return true;
-      }
-    } else {
-      more = hasMore;
-    }
     return more;
   }
 
   protected boolean next(LongWritable key) throws IOException {
-    if (!more)
+    if (!more) {
       return false;
-    long pos = in.getPosition();
-    boolean hasMore = in.next(key);
-    if (pos >= end && in.syncSeen()) {
+    }
+    
+    more = in.next(key);
+    if (!more) {
+      return false;
+    }
+    
+    long lastSeenSyncPos = in.lastSeenSyncPos();
+    if(lastSeenSyncPos >= end) {
       more = false;
-    } else {
-      more = hasMore;
+      return more;
     }
     return more;
   }

Modified: hadoop/hive/branches/branch-0.5/ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java
URL: http://svn.apache.org/viewvc/hadoop/hive/branches/branch-0.5/ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java?rev=903109&r1=903108&r2=903109&view=diff
==============================================================================
--- hadoop/hive/branches/branch-0.5/ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java
(original)
+++ hadoop/hive/branches/branch-0.5/ql/src/test/org/apache/hadoop/hive/ql/io/TestRCFile.java
Tue Jan 26 07:36:42 2010
@@ -30,6 +30,9 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.io.RCFile;
+import org.apache.hadoop.hive.ql.io.RCFileInputFormat;
+import org.apache.hadoop.hive.ql.io.RCFileOutputFormat;
 import org.apache.hadoop.hive.serde.Constants;
 import org.apache.hadoop.hive.serde2.SerDeException;
 import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
@@ -47,7 +50,12 @@
 import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.DefaultCodec;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.Reporter;
 
 public class TestRCFile extends TestCase {
 
@@ -364,4 +372,79 @@
     long cost = System.currentTimeMillis() - start;
     LOG.debug("reading fully costs:" + cost + " milliseconds");
   }
+  
+  public void testSynAndSplit() throws IOException {
+    splitBeforeSync();
+    splitRightBeforeSync();
+    splitInMiddleOfSync();
+    splitRightAfterSync();
+    splitAfterSync();
+  }
+
+  private void splitBeforeSync() throws IOException {
+    writeThenReadByRecordReader(600, 1000, 2, 1, null);
+  }
+  
+  private void splitRightBeforeSync() throws IOException {
+    writeThenReadByRecordReader(500, 1000, 2, 17750, null);
+  }
+  
+  private void splitInMiddleOfSync() throws IOException {
+    writeThenReadByRecordReader(500, 1000, 2, 17760, null);
+    
+  }
+  
+  private void splitRightAfterSync() throws IOException {
+    writeThenReadByRecordReader(500, 1000, 2, 17770, null);
+  }
+  
+  private void splitAfterSync() throws IOException {
+    writeThenReadByRecordReader(500, 1000, 2, 19950, null);
+  }
+
+  private void writeThenReadByRecordReader(int intervalRecordCount,
+      int writeCount, int splitNumber, long minSplitSize, CompressionCodec codec)
+      throws IOException {
+    Path testDir = new Path(System.getProperty("test.data.dir", ".") + "/mapred/testsmallfirstsplit");
+    Path testFile = new Path(testDir, "test_rcfile");
+    fs.delete(testFile, true);
+    Configuration cloneConf = new Configuration(conf);
+    RCFileOutputFormat.setColumnNumber(cloneConf, bytesArray.length);
+    cloneConf.setInt(RCFile.RECORD_INTERVAL_CONF_STR, intervalRecordCount);
+
+    RCFile.Writer writer = new RCFile.Writer(fs, cloneConf, testFile, null, codec);
+
+    BytesRefArrayWritable bytes = new BytesRefArrayWritable(bytesArray.length);
+    for (int i = 0; i < bytesArray.length; i++) {
+      BytesRefWritable cu = null;
+      cu = new BytesRefWritable(bytesArray[i], 0, bytesArray[i].length);
+      bytes.set(i, cu);
+    }
+    for (int i = 0; i < writeCount; i++) {
+      if(i == intervalRecordCount)
+        System.out.println("write position:" + writer.getLength());
+      writer.append(bytes);
+    }
+    writer.close();
+    
+    RCFileInputFormat inputFormat = new RCFileInputFormat();
+    JobConf jonconf = new JobConf(cloneConf);
+    jonconf.set("mapred.input.dir", testDir.toString());
+    jonconf.setLong("mapred.min.split.size", minSplitSize);
+    InputSplit[] splits = inputFormat.getSplits(jonconf, splitNumber);
+    assertEquals("splits length should be " + splitNumber, splits.length, splitNumber);
+    int readCount = 0;
+    for (int i = 0; i < splits.length; i++) {
+      int previousReadCount = readCount;
+      RecordReader rr = inputFormat.getRecordReader(splits[i], jonconf, Reporter.NULL);
+      Object key = rr.createKey();
+      Object value = rr.createValue();
+      while(rr.next(key, value)) 
+        readCount ++;
+      System.out.println("The " + i + "th split read "
+          + (readCount - previousReadCount));
+    }
+    assertEquals("readCount should be equal to writeCount", readCount, writeCount);
+  }
+  
 }



Mime
View raw message