natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix New SASI view creation during Index Redistribution 

patch by Jordan West; reviewed by jasobrown for CASSANDRA-14055
This commit is contained in:
Jordan West 2018-02-12 21:18:21 -08:00 committed by Jason Brown
parent 704f9b093a
commit ab8348c578
3 changed files with 82 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.11.3
* Fix New SASI view creation during Index Redistribution (CASSANDRA-14055)
* Remove string formatting lines from BufferPool hot path (CASSANDRA-14416)
* Update metrics to 3.1.5 (CASSANDRA-12924)
* Detect OpenJDK jvm type and architecture (CASSANDRA-12793)

10
src/java/org/apache/cassandra/index/sasi/conf/view/View.java
View File

@ -19,6 +19,7 @@ package org.apache.cassandra.index.sasi.conf.view;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.stream.Collectors;
import org.apache.cassandra.index.sasi.SSTableIndex;
import org.apache.cassandra.index.sasi.conf.ColumnIndex;
@ -59,10 +60,15 @@ public class View implements Iterable<SSTableIndex>
: new RangeTermTree.Builder(index.getMode().mode, validator);
List<Interval<Key, SSTableIndex>> keyIntervals = new ArrayList<>();
for (SSTableIndex sstableIndex : Iterables.concat(currentView, newIndexes))
// Ensure oldSSTables and newIndexes are disjoint (in index redistribution case the intersection can be non-empty).
// also favor newIndexes over currentView in case an SSTable has been re-opened (also occurs during redistribution)
// See CASSANDRA-14055
Collection<SSTableReader> toRemove = new HashSet<>(oldSSTables);
toRemove.removeAll(newIndexes.stream().map(SSTableIndex::getSSTable).collect(Collectors.toSet()));
for (SSTableIndex sstableIndex : Iterables.concat(newIndexes, currentView))
{
SSTableReader sstable = sstableIndex.getSSTable();
if (oldSSTables.contains(sstable) || sstable.isMarkedCompacted() || newView.containsKey(sstable.descriptor))
if (toRemove.contains(sstable) || sstable.isMarkedCompacted() || newView.containsKey(sstable.descriptor))
{
sstableIndex.release();
continue;

73
test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.index.sasi;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.nio.ByteBuffer;
import java.nio.file.FileSystems;
@ -60,6 +61,7 @@ import org.apache.cassandra.index.sasi.exceptions.TimeQuotaExceededException;
import org.apache.cassandra.index.sasi.memory.IndexMemtable;
import org.apache.cassandra.index.sasi.plan.QueryController;
import org.apache.cassandra.index.sasi.plan.QueryPlan;
import org.apache.cassandra.io.sstable.IndexSummaryManager;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.KeyspaceMetadata;
@ -897,6 +899,77 @@ public class SASIIndexTest
Assert.assertTrue(rows.toString(), Arrays.equals(new String[]{ "key9" }, rows.toArray(new String[rows.size()])));
}
@Test
public void testIndexRedistribution() throws IOException
{
Map<String, Pair<String, Integer>> part1 = new HashMap<String, Pair<String, Integer>>()
{{
put("key01", Pair.create("a", 33));
put("key02", Pair.create("a", 41));
}};
Map<String, Pair<String, Integer>> part2 = new HashMap<String, Pair<String, Integer>>()
{{
put("key03", Pair.create("a", 22));
put("key04", Pair.create("a", 45));
}};
Map<String, Pair<String, Integer>> part3 = new HashMap<String, Pair<String, Integer>>()
{{
put("key05", Pair.create("a", 32));
put("key06", Pair.create("a", 38));
}};
Map<String, Pair<String, Integer>> part4 = new HashMap<String, Pair<String, Integer>>()
{{
put("key07", Pair.create("a", 36));
put("key08", Pair.create("a", 36));
}};
Map<String, Pair<String, Integer>> part5 = new HashMap<String, Pair<String, Integer>>()
{{
put("key09", Pair.create("a", 21));
put("key10", Pair.create("a", 35));
}};
ColumnFamilyStore store = loadData(part1, 1000, true);
loadData(part2, true);
loadData(part3, true);
final ByteBuffer firstName = UTF8Type.instance.decompose("first_name");
Set<String> rows = getIndexed(store, 100, buildExpression(firstName, Operator.LIKE_CONTAINS, UTF8Type.instance.decompose("a")));
Assert.assertEquals(rows.toString(), 6, rows.size());
loadData(part4, true);
rows = getIndexed(store, 100, buildExpression(firstName, Operator.LIKE_CONTAINS, UTF8Type.instance.decompose("a")));
Assert.assertEquals(rows.toString(), 8, rows.size());
loadData(part5, true);
int minIndexInterval = store.metadata.params.minIndexInterval;
try
{
redistributeSummaries(10, store, firstName, minIndexInterval * 2);
redistributeSummaries(10, store, firstName, minIndexInterval * 4);
redistributeSummaries(10, store, firstName, minIndexInterval * 8);
redistributeSummaries(10, store, firstName, minIndexInterval * 16);
} finally
{
store.metadata.minIndexInterval(minIndexInterval);
}
}
private void redistributeSummaries(int expected, ColumnFamilyStore store, ByteBuffer firstName, int minIndexInterval) throws IOException
{
store.metadata.minIndexInterval(minIndexInterval);
IndexSummaryManager.instance.redistributeSummaries();
store.forceBlockingFlush();
Set<String> rows = getIndexed(store, 100, buildExpression(firstName, Operator.LIKE_CONTAINS, UTF8Type.instance.decompose("a")));
Assert.assertEquals(rows.toString(), expected, rows.size());
}
@Test
public void testTruncate()
{