HBASE-3472 MapFile.Reader getClosest() function returns incorrect results when before is true

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.17@663311 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -7,6 +7,9 @@ Release 0.17.1
 
     HADOOP-2159 Namenode stuck in safemode. The counter blockSafe should
     not be decremented for invalid blocks. (hairong)
+
+    HADOOP-3472 MapFile.Reader getClosest() function returns incorrect results
+    when before is true (Todd Lipcon via Stack)
  
 Release 0.17.0 - 2008-05-18
 

src/java/org/apache/hadoop/io/MapFile.java

@@ -453,20 +453,37 @@ public class MapFile {
       
       if (nextKey == null)
         nextKey = comparator.newKey();
-      long oldPosition = -1;
-      if (before) {
-        oldPosition = data.getPosition();
-      }
+     
+      // If we're looking for the key before, we need to keep track
+      // of the position we got the current key as well as the position
+      // of the key before it.
+      long prevPosition = -1;
+      long curPosition = seekPosition;
+
       while (data.next(nextKey)) {
         int c = comparator.compare(key, nextKey);
         if (c <= 0) {                             // at or beyond desired
           if (before && c != 0) {
-            // Need to back up to previous record.
-            data.seek(oldPosition);
-            data.next(nextKey);
+            if (prevPosition == -1) {
+              // We're on the first record of this index block
+              // and we've already passed the search key. Therefore
+              // we must be at the beginning of the file, so seek
+              // to the beginning of this block and return c
+              data.seek(curPosition);
+            } else {
+              // We have a previous record to back up to
+              data.seek(prevPosition);
+              data.next(nextKey);
+              // now that we've rewound, the search key must be greater than this key
+              return 1;
+            }
           }
           return c;
         }
+        if (before) {
+          prevPosition = curPosition;
+          curPosition = data.getPosition();
+        }
       }
 
       return 1;
@@ -537,10 +554,17 @@ public class MapFile {
     public synchronized WritableComparable getClosest(WritableComparable key,
         Writable val, final boolean before)
       throws IOException {
-      
-      if (seekInternal(key, before) > 0) {
+     
+      int c = seekInternal(key, before);
+
+      // If we didn't get an exact match, and we ended up in the wrong
+      // direction relative to the query key, return null since we
+      // must be at the beginning or end of the file.
+      if ((!before && c > 0) ||
+          (before && c < 0)) {
         return null;
       }
+
       data.getCurrentValue(val);
       return nextKey;
     }

src/test/org/apache/hadoop/io/TestMapFile.java

@@ -38,12 +38,12 @@ public class TestMapFile extends TestCase {
       getName() + ".mapfile"); 
     FileSystem fs = FileSystem.getLocal(conf);
     Path qualifiedDirName = fs.makeQualified(dirName);
-    // Make an index entry for each insertion.
-    MapFile.Writer.setIndexInterval(conf, 1);
+    // Make an index entry for every third insertion.
+    MapFile.Writer.setIndexInterval(conf, 3);
     MapFile.Writer writer = new MapFile.Writer(conf, fs,
       qualifiedDirName.toString(), Text.class, Text.class);
     // Assert that the index interval is 1
-    assertEquals(1, writer.getIndexInterval());
+    assertEquals(3, writer.getIndexInterval());
     // Add entries up to 100 in intervals of ten.
     final int FIRST_KEY = 10;
     for (int i = FIRST_KEY; i < 100; i += 10) {
@@ -59,28 +59,34 @@ public class TestMapFile extends TestCase {
     Text value = new Text();
     Text closest = (Text)reader.getClosest(key, value);
     // Assert that closest after 55 is 60
-    assertTrue(closest.equals(new Text("60")));
+    assertEquals(new Text("60"), closest);
     // Get closest that falls before the passed key: 50
     closest = (Text)reader.getClosest(key, value, true);
-    assertTrue(closest.equals(new Text("50")));
+    assertEquals(new Text("50"), closest);
     // Test get closest when we pass explicit key
     final Text TWENTY = new Text("20");
     closest = (Text)reader.getClosest(TWENTY, value);
-    assertTrue(closest.equals(new Text(TWENTY)));
+    assertEquals(TWENTY, closest);
     closest = (Text)reader.getClosest(TWENTY, value, true);
-    assertTrue(closest.equals(new Text(TWENTY)));
+    assertEquals(TWENTY, closest);
     // Test what happens at boundaries.  Assert if searching a key that is
     // less than first key in the mapfile, that the first key is returned.
     key = new Text("00");
     closest = (Text)reader.getClosest(key, value);
-    assertEquals(Integer.parseInt(closest.toString()), FIRST_KEY);
+    assertEquals(FIRST_KEY, Integer.parseInt(closest.toString()));
+    
+    // If we're looking for the first key before, and we pass in a key before 
+    // the first key in the file, we should get null
     closest = (Text)reader.getClosest(key, value, true);
-    assertEquals(Integer.parseInt(closest.toString()), FIRST_KEY);
+    assertNull(closest);
+    
     // Assert that null is returned if key is > last entry in mapfile.
     key = new Text("99");
     closest = (Text)reader.getClosest(key, value);
     assertNull(closest);
+
+    // If we were looking for the key before, we should get the last key
     closest = (Text)reader.getClosest(key, value, true);
-    assertNull(closest);
+    assertEquals(new Text("90"), closest);
   }
 }