merge from 0.7

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

CHANGES.txt

@@ -10,11 +10,18 @@
  * check for null encryption in MessagingService (CASSANDRA-2152)
 
 
+0.7.3
+ * Keep endpoint state until aVeryLongTime (CASSANDRA-2115)
+ * lower-latency read repair (CASSANDRA-2069)
+ * add hinted_handoff_throttle_delay_in_ms option (CASSANDRA-2161)
+ * don't save empty caches (CASSANDRA-2172)
+
+
 0.7.2
  * copy DecoratedKey.key when inserting into caches to avoid retaining
    a reference to the underlying buffer (CASSANDRA-2102)
  * format subcolumn names with subcomparator (CASSANDRA-2136)
- * lower-latency read repair (CASSANDRA-2069)
+ * fix column bloom filter deserialization (CASSANDRA-2165)
 
 
 0.7.1

conf/cassandra.yaml

@@ -34,6 +34,8 @@ hinted_handoff_enabled: true
 # this defines the maximum amount of time a dead host will have hints
 # generated.  After it has been dead this long, hints will be dropped.
 max_hint_window_in_ms: 3600000 # one hour
+# Sleep this long after delivering each row or row fragment
+hinted_handoff_throttle_delay_in_ms: 50
 
 # authentication backend, implementing IAuthenticator; used to identify users
 authenticator: org.apache.cassandra.auth.AllowAllAuthenticator

src/java/org/apache/cassandra/config/Config.java

@@ -109,6 +109,7 @@ public class Config
     public Double flush_largest_memtables_at = 1.0;
     public Double reduce_cache_sizes_at = 1.0;
     public double reduce_cache_capacity_to = 0.6;
+    public int hinted_handoff_throttle_delay_in_ms = 0;
 
     public static enum CommitLogSync {
         periodic,

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1191,4 +1191,9 @@ public class DatabaseDescriptor
     {
         return conf.reduce_cache_capacity_to;
     }
+
+    public static int getHintedHandoffThrottleDelay()
+    {
+        return conf.hinted_handoff_throttle_delay_in_ms;
+    }
 }

src/java/org/apache/cassandra/db/HintedHandOffManager.java

@@ -157,6 +157,15 @@ public class HintedHandOffManager implements HintedHandOffManagerMBean
             {
                 return false;
             }
+
+            try
+            {
+                Thread.sleep(DatabaseDescriptor.getHintedHandoffThrottleDelay());
+            }
+            catch (InterruptedException e)
+            {
+                throw new AssertionError(e);
+            }
         }
         return true;
     }

src/java/org/apache/cassandra/io/sstable/CacheWriter.java

@@ -64,6 +64,14 @@ public class CacheWriter<K, V> implements ICompactionInfo
     public void saveCache() throws IOException
     {
         long start = System.currentTimeMillis();
+
+        if (keys.size() == 0 || estimatedTotalBytes == 0)
+        {
+            logger.debug("Deleting {} (cache is empty)");
+            path.delete();
+            return;
+        }
+
         logger.debug("Saving {}", path);
         File tmpFile = File.createTempFile(path.getName(), null, path.getParentFile());
 

src/java/org/apache/cassandra/io/util/MappedFileDataInput.java

@@ -118,7 +118,7 @@ public class MappedFileDataInput extends AbstractDataInput implements FileDataIn
         assert length <= remaining
                 : String.format("mmap segment underflow; remaining is %d but %d requested", remaining, length);
 
-        ByteBuffer bytes = buffer.slice();
+        ByteBuffer bytes = buffer.duplicate();
         bytes.position(buffer.position() + position).limit(buffer.position() + position + length);
         position += length;
 

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

@@ -398,9 +398,10 @@ public class ByteBufferUtil
                 if (!copy.hasRemaining())
                     return -1;
 
-                return copy.get();
+                return copy.get() & 0xFF;
             }
 
+            @Override
             public int read(byte[] bytes, int off, int len) throws IOException
             {
                 len = Math.min(len, copy.remaining());
@@ -408,6 +409,12 @@ public class ByteBufferUtil
 
                 return len;
             }
+
+            @Override
+            public int available() throws IOException
+            {
+                return copy.remaining();
+            }
         };
     }
 

test/conf/cassandra.yaml

@@ -38,43 +38,57 @@ keyspaces:
       replication_factor: 1
       column_families:
         - name: Standard1
-          rows_cached: 10
+          rows_cached: 0
           keys_cached: 0
 
         - name: Standard2
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Standard3
-          keys_cached: 0.5
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Standard4
-          keys_cached: 1.0
+          rows_cached: 0
+          keys_cached: 0
 
         - name: StandardLong1
-          compare_with: LongType
+          rows_cached: 0
+          keys_cached: 0
 
         - name: StandardLong2
-          compare_with: LongType
+          rows_cached: 0
+          keys_cached: 0
 
         - name: StandardInteger1
+          rows_cached: 0
+          keys_cached: 0
           compare_with: IntegerType
 
         - name: Super1
           column_type: Super
           compare_subcolumns_with: LongType
-          rows_cached: 1000
+          keys_cached: 0
           keys_cached: 0
 
         - name: Super2
           column_type: Super
           compare_subcolumns_with: LongType
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super3
           column_type: Super
           compare_subcolumns_with: LongType
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super4
           column_type: Super
           compare_subcolumns_with: UTF8Type
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Counter1
           column_type: Standard
@@ -89,67 +103,107 @@ keyspaces:
             - name: birthdate
               validator_class: LongType
               index_type: KEYS
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Indexed2 
           column_metadata:
             - name: birthdate
               validator_class: LongType
               # index will be added dynamically
+          rows_cached: 0
+          keys_cached: 0
 
     - name: Keyspace2
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 1
       column_families:
         - name: Standard1
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Standard3
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super3
           column_type: Super
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super4
           column_type: Super
           compare_subcolumns_with: TimeUUIDType
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Indexed1
           column_metadata:
             - name: birthdate
               validator_class: LongType
               index_type: KEYS
+          rows_cached: 0
+          keys_cached: 0
 
     - name: Keyspace3
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 5
       column_families:
         - name: Standard1
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Indexed1
           column_metadata:
             - name: birthdate
               validator_class: LongType
               index_type: KEYS
+          rows_cached: 0
+          keys_cached: 0
 
     - name: Keyspace4
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 3
       column_families:
         - name: Standard1
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Standard3
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super3
           column_type: Super
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Super4
           column_type: Super
           compare_subcolumns_with: TimeUUIDType
+          rows_cached: 0
+          keys_cached: 0
 
     - name: Keyspace5
       replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
       replication_factor: 2
       column_families:
         - name: Standard1
+          rows_cached: 0
+          keys_cached: 0
 
         - name: Counter1
           column_type: Standard
           default_validation_class: CounterColumnType
+          rows_cached: 0
+          keys_cached: 0
+
+    - name: KeyCacheSpace
+      replica_placement_strategy: org.apache.cassandra.locator.SimpleStrategy
+      replication_factor: 1
+      column_families:
+        - name: Standard1
+          keys_cached: 0.5
+
+        - name: Standard2
+          keys_cached: 1.0

test/long/org/apache/cassandra/db/LongTableTest.java

@@ -0,0 +1,85 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.db;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.text.DecimalFormat;
+import java.text.NumberFormat;
+import java.util.*;
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+
+import org.apache.commons.lang.StringUtils;
+import org.junit.Test;
+
+import static junit.framework.Assert.*;
+import org.apache.cassandra.CleanupHelper;
+import org.apache.cassandra.db.filter.QueryFilter;
+import org.apache.cassandra.utils.WrappedRunnable;
+import static org.apache.cassandra.Util.column;
+import static org.apache.cassandra.Util.getBytes;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.db.marshal.LongType;
+import org.apache.cassandra.io.sstable.IndexHelper;
+import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.io.util.BufferedRandomAccessFile;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+
+public class LongTableTest extends CleanupHelper
+{
+    @Test
+    public void testGetRowMultiColumn() throws Throwable
+    {
+        final Table table = Table.open("Keyspace1");
+        final ColumnFamilyStore cfStore = table.getColumnFamilyStore("Standard1");
+
+        for (int i = 1; i < 5000; i += 100)
+        {
+            RowMutation rm = new RowMutation("Keyspace1", Util.dk("key" + i).key);
+            ColumnFamily cf = ColumnFamily.create("Keyspace1", "Standard1");
+            for (int j = 0; j < i; j++)
+                cf.addColumn(column("c" + j, "v" + j, 1L));
+            rm.add(cf);
+            rm.applyUnsafe();
+        }
+
+        Runnable verify = new WrappedRunnable()
+        {
+            public void runMayThrow() throws Exception
+            {
+                ColumnFamily cf;
+                for (int i = 1; i < 5000; i += 100)
+                {
+                    for (int j = 0; j < i; j++)
+                    {
+                        cf = cfStore.getColumnFamily(QueryFilter.getNamesFilter(Util.dk("key" + i), new QueryPath("Standard1"), ByteBufferUtil.bytes("c" + j)));
+                        TableTest.assertColumns(cf, "c" + j);
+                    }
+                }
+
+            }
+        };
+        TableTest.reTest(table.getColumnFamilyStore("Standard1"), verify);
+    }
+}

test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java

@@ -178,59 +178,4 @@ public class CompactionsPurgeTest extends CleanupHelper
         ColumnFamily cf = table.getColumnFamilyStore(cfName).getColumnFamily(QueryFilter.getIdentityFilter(key, new QueryPath(cfName)));
         assert cf == null : cf;
     }
-
-    @Test
-    public void testKeyCache50() throws IOException, ExecutionException, InterruptedException
-    {
-        testKeyCache("Standard3", 64);
-    }
-
-    @Test
-    public void testKeyCache100() throws IOException, ExecutionException, InterruptedException
-    {
-        testKeyCache("Standard4", 128);
-    }
-
-    public void testKeyCache(String cfname, int expectedCacheSize) throws IOException, ExecutionException, InterruptedException
-    {
-        CompactionManager.instance.disableAutoCompaction();
-
-        Table table = Table.open(TABLE1);
-        String cfName = cfname;
-        ColumnFamilyStore store = table.getColumnFamilyStore(cfName);
-
-        // KeyCache should start at size 1 if we're caching X% of zero data.
-        int keyCacheSize = store.getKeyCacheCapacity();
-        assert keyCacheSize == 1 : keyCacheSize;
-
-        DecoratedKey key1 = Util.dk("key1");
-        DecoratedKey key2 = Util.dk("key2");
-        RowMutation rm;
-
-        // inserts
-        rm = new RowMutation(TABLE1, key1.key);
-        rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
-        rm.apply();
-        rm = new RowMutation(TABLE1, key2.key);
-        rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
-        rm.apply();
-
-        // deletes
-        rm = new RowMutation(TABLE1, key1.key);
-        rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), 1);
-        rm.apply();
-        rm = new RowMutation(TABLE1, key2.key);
-        rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), 1);
-        rm.apply();
-
-        // After a flush, the cache should expand to be X% of indices * INDEX_INTERVAL.
-        store.forceBlockingFlush();
-        keyCacheSize = store.getKeyCacheCapacity();
-        assert keyCacheSize == expectedCacheSize : keyCacheSize;
-
-        // After a compaction, the cache should expand to be X% of zero data.
-        CompactionManager.instance.submitMajor(store, 0, Integer.MAX_VALUE).get();
-        keyCacheSize = store.getKeyCacheCapacity();
-        assert keyCacheSize == 1 : keyCacheSize;
-    }
 }

test/unit/org/apache/cassandra/db/KeyCacheTest.java

@@ -0,0 +1,69 @@
+package org.apache.cassandra.db;
+
+import java.io.IOException;
+import java.util.concurrent.ExecutionException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.CleanupHelper;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.db.filter.QueryPath;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class KeyCacheTest extends CleanupHelper
+{
+    private static final String TABLE1 = "KeyCacheSpace";
+
+    @Test
+    public void testKeyCache50() throws IOException, ExecutionException, InterruptedException
+    {
+        testKeyCache("Standard1", 64);
+    }
+
+    @Test
+    public void testKeyCache100() throws IOException, ExecutionException, InterruptedException
+    {
+        testKeyCache("Standard2", 128);
+    }
+
+    public void testKeyCache(String cfName, int expectedCacheSize) throws IOException, ExecutionException, InterruptedException
+    {
+        CompactionManager.instance.disableAutoCompaction();
+
+        Table table = Table.open(TABLE1);
+        ColumnFamilyStore store = table.getColumnFamilyStore(cfName);
+
+        // KeyCache should start at size 1 if we're caching X% of zero data.
+        int keyCacheSize = store.getKeyCacheCapacity();
+        assert keyCacheSize == 1 : keyCacheSize;
+
+        DecoratedKey key1 = Util.dk("key1");
+        DecoratedKey key2 = Util.dk("key2");
+        RowMutation rm;
+
+        // inserts
+        rm = new RowMutation(TABLE1, key1.key);
+        rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
+        rm.apply();
+        rm = new RowMutation(TABLE1, key2.key);
+        rm.add(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), ByteBufferUtil.EMPTY_BYTE_BUFFER, 0);
+        rm.apply();
+
+        // deletes
+        rm = new RowMutation(TABLE1, key1.key);
+        rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("1")), 1);
+        rm.apply();
+        rm = new RowMutation(TABLE1, key2.key);
+        rm.delete(new QueryPath(cfName, null, ByteBufferUtil.bytes("2")), 1);
+        rm.apply();
+
+        // After a flush, the cache should expand to be X% of indices * INDEX_INTERVAL.
+        store.forceBlockingFlush();
+        keyCacheSize = store.getKeyCacheCapacity();
+        assert keyCacheSize == expectedCacheSize : keyCacheSize;
+
+        // After a compaction, the cache should expand to be X% of zero data.
+        CompactionManager.instance.submitMajor(store, 0, Integer.MAX_VALUE).get();
+        keyCacheSize = store.getKeyCacheCapacity();
+        assert keyCacheSize == 1 : keyCacheSize;
+    }}

test/unit/org/apache/cassandra/db/TableTest.java

@@ -48,7 +48,6 @@ import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class TableTest extends CleanupHelper
 {
-    private static final DecoratedKey KEY2 = Util.dk("key2");
     private static final DecoratedKey TEST_KEY = Util.dk("key1");
     private static final DecoratedKey TEST_SLICE_KEY = Util.dk("key1-slicerange");
 
@@ -396,11 +395,13 @@ public class TableTest extends CleanupHelper
         cfStore.forceBlockingFlush();
 
         validateSliceLarge(cfStore);
+
         // compact so we have a big row with more than the minimum index count
         if (cfStore.getSSTables().size() > 1)
         {
             CompactionManager.instance.performMajor(cfStore);
         }
+        // verify that we do indeed have multiple index entries
         SSTableReader sstable = cfStore.getSSTables().iterator().next();
         long position = sstable.getPosition(key, SSTableReader.Operator.EQ);
         BufferedRandomAccessFile file = new BufferedRandomAccessFile(sstable.getFilename(), "r");
@@ -410,6 +411,7 @@ public class TableTest extends CleanupHelper
         IndexHelper.skipBloomFilter(file);
         ArrayList<IndexHelper.IndexInfo> indexes = IndexHelper.deserializeIndex(file);
         assert indexes.size() > 2;
+
         validateSliceLarge(cfStore);
     }