diff --git a/CHANGES.txt b/CHANGES.txt index 8040a769fc..f84b26b832 100644 --- a/CHANGES.txt +++ b/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 diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 21649a93be..f4eaad11f8 100644 --- a/conf/cassandra.yaml +++ b/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 diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 117c0bdcc0..ede3cd3690 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/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, diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index ec0d5c1cc4..4f1111cbc4 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/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; + } } diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index b44c63d24b..b8cbd44c18 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/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; } diff --git a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java index d9202571dd..5e84e4b4aa 100644 --- a/src/java/org/apache/cassandra/io/sstable/CacheWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/CacheWriter.java @@ -64,6 +64,14 @@ public class CacheWriter 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()); diff --git a/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java b/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java index b6bda7a7f0..cd8d3ba864 100644 --- a/src/java/org/apache/cassandra/io/util/MappedFileDataInput.java +++ b/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; diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java index 55dd507d9e..4d54c41c72 100644 --- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java +++ b/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(); + } }; } diff --git a/test/conf/cassandra.yaml b/test/conf/cassandra.yaml index 9b475d5586..9625d7364c 100644 --- a/test/conf/cassandra.yaml +++ b/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 diff --git a/test/long/org/apache/cassandra/db/LongTableTest.java b/test/long/org/apache/cassandra/db/LongTableTest.java new file mode 100644 index 0000000000..fc093ff520 --- /dev/null +++ b/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); + } +} diff --git a/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java b/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java index 49d961003e..27ca9ad1d9 100644 --- a/test/unit/org/apache/cassandra/db/CompactionsPurgeTest.java +++ b/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; - } } diff --git a/test/unit/org/apache/cassandra/db/KeyCacheTest.java b/test/unit/org/apache/cassandra/db/KeyCacheTest.java new file mode 100644 index 0000000000..c3579d008d --- /dev/null +++ b/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; + }} diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index aa68ecb833..227a1b6dbd 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/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 indexes = IndexHelper.deserializeIndex(file); assert indexes.size() > 2; + validateSliceLarge(cfStore); }