Fix New SASI view creation during Index Redistribution patch by Jordan West; reviewed by jasobrown for CASSANDRA-14055
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ab8348c5 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ab8348c5 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ab8348c5 Branch: refs/heads/trunk Commit: ab8348c578c0bb2d3baefaf387b4d9bc67f4c861 Parents: 704f9b0 Author: Jordan West <jorda...@gmail.com> Authored: Mon Feb 12 21:18:21 2018 -0800 Committer: Jason Brown <jasedbr...@gmail.com> Committed: Mon May 14 12:43:27 2018 -0700 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../cassandra/index/sasi/conf/view/View.java | 10 ++- .../cassandra/index/sasi/SASIIndexTest.java | 73 ++++++++++++++++++++ 3 files changed, 82 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8348c5/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 0e4346b..d70b381 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8348c5/src/java/org/apache/cassandra/index/sasi/conf/view/View.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/index/sasi/conf/view/View.java b/src/java/org/apache/cassandra/index/sasi/conf/view/View.java index 25f32d9..b0afc5b 100644 --- a/src/java/org/apache/cassandra/index/sasi/conf/view/View.java +++ b/src/java/org/apache/cassandra/index/sasi/conf/view/View.java @@ -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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/ab8348c5/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java index 1b2f97d..a57af61 100644 --- a/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java +++ b/test/unit/org/apache/cassandra/index/sasi/SASIIndexTest.java @@ -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; @@ -898,6 +900,77 @@ public class SASIIndexTest } @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() { Map<String, Pair<String, Integer>> part1 = new HashMap<String, Pair<String, Integer>>() --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org For additional commands, e-mail: commits-h...@cassandra.apache.org