preparation for Merkle tree; patched by Stu Hood, reviewed by junrao for CASSANDRA-193

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@887572 13f79535-47bb-0310-9956-ffa450edef68

src/java/org/apache/cassandra/dht/Range.java

@@ -72,15 +72,9 @@ public class Range implements Comparable<Range>, Serializable
         return right_;
     }
 
-    /**
-     * Helps determine if a given point on the DHT ring is contained
-     * in the range in question.
-     * @param bi point in question
-     * @return true if the point contains within the range else false.
-     */
-    public boolean contains(Token bi)
+    public static boolean contains(Token left, Token right, Token bi)
     {
-        if ( isWrapAround(this) )
+        if ( isWrapAround(left, right) )
         {
             /* 
              * We are wrapping around, so the interval is (a,b] where a >= b,
@@ -89,24 +83,66 @@ public class Range implements Comparable<Range>, Serializable
              * (2) k <= b -- return true
              * (3) b < k <= a -- return false
              */
-            if ( bi.compareTo(left_) > 0 )
+            if ( bi.compareTo(left) > 0 )
                 return true;
             else
-                return right_.compareTo(bi) >= 0;
+                return right.compareTo(bi) >= 0;
         }
         else
         {
             /*
              * This is the range (a, b] where a < b. 
              */
-            return ( bi.compareTo(left_) > 0 && right_.compareTo(bi) >= 0 );
+            return ( bi.compareTo(left) > 0 && right.compareTo(bi) >= 0 );
         }        
     }
 
-    public boolean contains(Range range)
+    public boolean contains(Range that)
     {
-        return (contains(range.left_) || range.left_.equals(left_))
-               && contains(range.right_);
+        boolean thiswraps = isWrapAround(this.left(), this.right());
+        boolean thatwraps = isWrapAround(that.left(), that.right());
+        if (thiswraps == thatwraps)
+            return this.left().compareTo(that.left()) <= 0 &&
+                that.right().compareTo(this.right()) <= 0;
+        else if (thiswraps)
+            // wrapping might contain non-wrapping
+            return this.left().compareTo(that.left()) <= 0 ||
+                that.right().compareTo(this.right()) <= 0;
+        else // (thatwraps)
+            // non-wrapping cannot contain wrapping
+            return false;
+    }
+
+    /**
+     * Helps determine if a given point on the DHT ring is contained
+     * in the range in question.
+     * @param bi point in question
+     * @return true if the point contains within the range else false.
+     */
+    public boolean contains(Token bi)
+    {
+        return contains(left_, right_, bi);
+    }
+
+    /**
+     * @param range range to check for intersection
+     * @return true if the given range intersects with this range.
+     */
+    public boolean intersects(Range that)
+    {
+        boolean thiswraps = isWrapAround(this.left(), this.right());
+        boolean thatwraps = isWrapAround(that.left(), that.right());
+        if (thiswraps && thatwraps)
+            // both (must contain the minimum token)
+            return true;
+        else if (!thiswraps && !thatwraps)
+            // neither
+            return this.left().compareTo(that.right()) < 0 &&
+                that.left().compareTo(this.right()) < 0;
+        else
+            // either
+            return this.left().compareTo(that.right()) < 0 ||
+                that.left().compareTo(this.right()) < 0;
     }
 
     /**
@@ -114,9 +150,9 @@ public class Range implements Comparable<Range>, Serializable
      * @param range
      * @return
      */
-    private static boolean isWrapAround(Range range)
+    public static boolean isWrapAround(Token left, Token right)
     {
-        return range.left_.compareTo(range.right_) >= 0;
+        return left.compareTo(right) >= 0;
     }
     
     public int compareTo(Range rhs)
@@ -125,10 +161,10 @@ public class Range implements Comparable<Range>, Serializable
          * If the range represented by the "this" pointer
          * is a wrap around then it is the smaller one.
          */
-        if ( isWrapAround(this) )
+        if ( isWrapAround(left(), right()) )
             return -1;
         
-        if ( isWrapAround(rhs) )
+        if ( isWrapAround(rhs.left(), rhs.right()) )
             return 1;
         
         return right_.compareTo(rhs.right_);
@@ -141,7 +177,7 @@ public class Range implements Comparable<Range>, Serializable
 
         for (Range range : ranges)
         {
-            if(range.contains(token))
+            if (range.contains(token))
             {
                 return true;
             }
@@ -157,6 +193,7 @@ public class Range implements Comparable<Range>, Serializable
         return left_.equals(rhs.left_) && right_.equals(rhs.right_);
     }
     
+    @Override
     public int hashCode()
     {
         return toString().hashCode();

src/java/org/apache/cassandra/io/SSTable.java

@@ -142,6 +142,11 @@ public abstract class SSTable
         return columnFamilyName;
     }
 
+    public String getTableName()
+    {
+        return parseTableName(path);
+    }
+
     public static String parseTableName(String filename)
     {
         return new File(filename).getParentFile().getName();        

src/java/org/apache/cassandra/io/SSTableReader.java

@@ -35,7 +35,10 @@ import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.marshal.AbstractType;
+
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
+import com.google.common.base.Predicate;
+import com.google.common.base.Predicates;
 import com.reardencommerce.kernel.collections.shared.evictable.ConcurrentLinkedHashMap;
 
 /**
@@ -109,19 +112,21 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
     }
 
     /**
-     * Get all indexed keys in any SSTable for our primary range
-     * TODO add option to include keys from one or more other ranges
+     * Get all indexed keys defined by the two predicates.
+     * @param cfpred A Predicate defining matching column families.
+     * @param dkpred A Predicate defining matching DecoratedKeys.
      */
-    public static List<DecoratedKey> getIndexedDecoratedKeys()
+    public static List<DecoratedKey> getIndexedDecoratedKeysFor(Predicate<SSTable> cfpred, Predicate<DecoratedKey> dkpred)
     {
-        Range range = StorageService.instance().getLocalPrimaryRange();
         List<DecoratedKey> indexedKeys = new ArrayList<DecoratedKey>();
         
         for (SSTableReader sstable : openedFiles.values())
         {
+            if (!cfpred.apply(sstable))
+                continue;
             for (KeyPosition kp : sstable.getIndexPositions())
             {
-                if (range.contains(kp.key.token))
+                if (dkpred.apply(kp.key))
                 {
                     indexedKeys.add(kp.key);
                 }
@@ -132,6 +137,23 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         return indexedKeys;
     }
 
+    /**
+     * Get all indexed keys in any SSTable for our primary range.
+     */
+    public static List<DecoratedKey> getIndexedDecoratedKeys()
+    {
+        final Range range = StorageService.instance().getLocalPrimaryRange();
+
+        Predicate<SSTable> cfpred = Predicates.alwaysTrue();
+        return getIndexedDecoratedKeysFor(cfpred,
+                                          new Predicate<DecoratedKey>(){
+            public boolean apply(DecoratedKey dk)
+            {
+               return range.contains(dk.token);
+            }
+        });
+    }
+
     public static SSTableReader open(String dataFileName) throws IOException
     {
         return open(dataFileName, StorageService.getPartitioner(), DatabaseDescriptor.getKeysCachedFraction(parseTableName(dataFileName)));
@@ -386,11 +408,6 @@ public class SSTableReader extends SSTable implements Comparable<SSTableReader>
         return new SSTableScanner(this);
     }
 
-    public String getTableName()
-    {
-        return parseTableName(path);
-    }
-
     public AbstractType getColumnComparator()
     {
         return DatabaseDescriptor.getComparator(getTableName(), getColumnFamilyName());

src/java/org/apache/cassandra/net/MessagingService.java

@@ -88,7 +88,7 @@ public class MessagingService
     private final static ReentrantLock lock_ = new ReentrantLock();
     private static Map<String, TcpConnectionManager> poolTable_ = new Hashtable<String, TcpConnectionManager>();
     
-    private static boolean bShutdown_ = false;
+    private static volatile boolean bShutdown_ = false;
     
     private static Logger logger_ = Logger.getLogger(MessagingService.class);
     

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -129,13 +129,15 @@ public class FBUtilities
         return hash.abs();        
     }
 
-    public static byte[] hash(String type, byte[] data)
+    public static byte[] hash(String type, byte[]... data)
     {
     	byte[] result = null;
     	try
         {
-    		MessageDigest messageDigest = MessageDigest.getInstance(type);
-    		result = messageDigest.digest(data);
+            MessageDigest messageDigest = MessageDigest.getInstance(type);
+            for(byte[] block : data)
+                messageDigest.update(block);
+            result = messageDigest.digest();
     	}
     	catch (Exception e)
         {

test/unit/org/apache/cassandra/dht/RangeTest.java

@@ -22,7 +22,7 @@ import org.junit.Test;
 
 public class RangeTest {
     @Test
-    public void testRange() {
+    public void testContains() {
         Range left = new Range(new BigIntegerToken("0"), new BigIntegerToken("100"));
         assert !left.contains(new BigIntegerToken("0"));
         assert left.contains(new BigIntegerToken("10"));
@@ -31,7 +31,7 @@ public class RangeTest {
     }
 
     @Test
-    public void testWrappingRange() {
+    public void testContainsWrapping() {
         Range range = new Range(new BigIntegerToken("0"), new BigIntegerToken("0"));
         assert range.contains(new BigIntegerToken("0"));
         assert range.contains(new BigIntegerToken("10"));
@@ -44,4 +44,84 @@ public class RangeTest {
         assert !range.contains(new BigIntegerToken("100"));
         assert range.contains(new BigIntegerToken("200"));
     }
+
+    @Test
+    public void testContainsRange() {
+        Range one = new Range(new BigIntegerToken("2"), new BigIntegerToken("10"));
+        Range two = new Range(new BigIntegerToken("2"), new BigIntegerToken("5"));
+        Range thr = new Range(new BigIntegerToken("5"), new BigIntegerToken("10"));
+        Range fou = new Range(new BigIntegerToken("10"), new BigIntegerToken("12"));
+
+        assert one.contains(two);
+        assert one.contains(thr);
+        assert !one.contains(fou);
+
+        assert !two.contains(one);
+        assert !two.contains(thr);
+        assert !two.contains(fou);
+
+        assert !thr.contains(one);
+        assert !thr.contains(two);
+        assert !thr.contains(fou);
+
+        assert !fou.contains(one);
+        assert !fou.contains(two);
+        assert !fou.contains(thr);
+    }
+
+    @Test
+    public void testContainsRangeWrapping() {
+        Range one = new Range(new BigIntegerToken("10"), new BigIntegerToken("2"));
+        Range two = new Range(new BigIntegerToken("5"), new BigIntegerToken("3"));
+        Range thr = new Range(new BigIntegerToken("10"), new BigIntegerToken("12"));
+        Range fou = new Range(new BigIntegerToken("2"), new BigIntegerToken("6"));
+
+        assert !one.contains(two);
+        assert one.contains(thr);
+        assert !one.contains(fou);
+
+        assert two.contains(one);
+        assert two.contains(thr);
+        assert !two.contains(fou);
+
+        assert !thr.contains(one);
+        assert !thr.contains(two);
+        assert !thr.contains(fou);
+
+        assert !fou.contains(one);
+        assert !fou.contains(two);
+        assert !fou.contains(thr);
+    }
+
+    @Test
+    public void testIntersects() {
+        Range one = new Range(new BigIntegerToken("2"), new BigIntegerToken("10"));
+        Range two = new Range(new BigIntegerToken("0"), new BigIntegerToken("8"));
+        Range not = new Range(new BigIntegerToken("10"), new BigIntegerToken("12"));
+
+        assert one.intersects(two);
+        assert two.intersects(one);
+
+        assert !one.intersects(not);
+        assert !not.intersects(one);
+
+        assert !two.intersects(not);
+        assert !not.intersects(two);
+    }
+
+    @Test
+    public void testIntersectsWrapping() {
+        Range onewrap = new Range(new BigIntegerToken("10"), new BigIntegerToken("2"));
+        Range twowrap = new Range(new BigIntegerToken("5"), new BigIntegerToken("3"));
+        Range not = new Range(new BigIntegerToken("2"), new BigIntegerToken("6"));
+
+        assert onewrap.intersects(twowrap);
+        assert twowrap.intersects(onewrap);
+
+        assert !onewrap.intersects(not);
+        assert !not.intersects(onewrap);
+
+        assert twowrap.intersects(not);
+        assert not.intersects(twowrap);
+    }
 }