From 4a5620dea228470397cea696a75a849050efe03b Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Fri, 1 May 2009 02:24:10 +0000
Subject: [PATCH] avoid raising an exception in SequenceFile.next when a key
 does not exist (e.g. when bloom filter gives a false positive).  patch by Jun
 Rao; reviewed by jbellis for CASSANDRA-126

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@770517 13f79535-47bb-0310-9956-ffa450edef68
---
 .../org/apache/cassandra/io/SequenceFile.java | 16 +++++-------
 .../cassandra/db/ColumnFamilyStoreTest.java   | 25 +++++++++++++++++++
 2 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/src/java/org/apache/cassandra/io/SequenceFile.java b/src/java/org/apache/cassandra/io/SequenceFile.java
index 5584dd41fb..afb450c088 100644
--- a/src/java/org/apache/cassandra/io/SequenceFile.java
+++ b/src/java/org/apache/cassandra/io/SequenceFile.java
@@ -705,14 +705,13 @@ public class SequenceFile
          * @param section indicates the location of the block index.
          * @throws IOException
          */
-        protected void seekTo(String key, Coordinate section) throws IOException
+        private long seekTo(String key, Coordinate section) throws IOException
         {
-            /* Goto the Block Index */
             seek(section.end_);
             long position = getPositionFromBlockIndex(key);
-            if (position == -1)
-                throw new IOException("This key " + key + " does not exist in this file.");
-            seek(position);
+            if (position >= 0)
+                seek(position);
+            return position;
         }
 
         /**
@@ -809,9 +808,8 @@ public class SequenceFile
             assert timeRange == null || columnNames == null; // at most one may be non-null
 
             long bytesRead = -1L;
-            if (isEOF())
+            if (isEOF() || seekTo(key, section) < 0)
                 return bytesRead;
-            seekTo(key, section);
             /* note the position where the key starts */
             long startPosition = file_.getFilePointer();
             String keyInDisk = file_.readUTF();
@@ -1065,10 +1063,8 @@ public class SequenceFile
         public long next(String key, DataOutputBuffer bufOut, Coordinate section) throws IOException
         {
             long bytesRead = -1L;
-            if (isEOF())
+            if (isEOF() || seekTo(key, section) < 0)
                 return bytesRead;
-
-            seekTo(key, section);
             /* note the position where the key starts */
             long startPosition = file_.getFilePointer();
             String keyInDisk = file_.readUTF();
diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
index e96c2056cf..b12891a0f7 100644
--- a/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
+++ b/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
@@ -18,6 +18,9 @@ import org.apache.commons.lang.ArrayUtils;
 import org.apache.commons.lang.StringUtils;
 
 import org.apache.cassandra.ServerTest;
+import org.apache.cassandra.io.DataInputBuffer;
+import org.apache.cassandra.io.SSTable;
+import org.apache.cassandra.service.StorageService;
 import org.testng.annotations.Test;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertNull;
@@ -407,4 +410,26 @@ public class ColumnFamilyStoreTest extends ServerTest
         Future ft = MinorCompactionManager.instance().submit(store);
         ft.get();
     }
+    
+    @Test
+    public void testGetColumnWithWrongBF() throws IOException, ExecutionException, InterruptedException
+    {
+        Table table = Table.open("Table1");
+        ColumnFamilyStore store = table.getColumnFamilyStore("Standard1");
+        RowMutation rm;
+
+        // add data
+        rm = new RowMutation("Table1", "key1");
+        rm.add("Standard1:Column1", "asdf".getBytes(), 0);
+        rm.add("Standard1:Column2", "asdf".getBytes(), 0);
+        rm.apply();
+        store.forceBlockingFlush();
+
+        List<String> ssTables = table.getAllSSTablesOnDisk();
+        /* the following call can happen if BF is wrong. Should return an empty buffer. */
+        IFilter filter = new IdentityFilter(); 
+        SSTable ssTable = new SSTable(ssTables.get(0), StorageService.getPartitioner());
+        DataInputBuffer bufIn = filter.next("key2", "Standard1:Column1", ssTable);
+        assertEquals(bufIn.getLength(), 0);
+    }
 }