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

Notify indexers of partition deletion during cleanup 

Patch by Sam Tunnicliffe; reviewed by Sergio Bossa for CASSANDRA-10685
This commit is contained in:
Sam Tunnicliffe 2015-11-13 17:15:11 +00:00
parent 3223927292
commit 5730e7b9f8
3 changed files with 40 additions and 1 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.0.1
* Notify indexers of partition delete during cleanup (CASSANDRA-10685)
* Keep the file open in trySkipCache (CASSANDRA-10669)
* Updated trigger example (CASSANDRA-10257)
Merged from 2.2:

7
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -530,7 +530,7 @@ public class SecondaryIndexManager implements IndexRegistry
partition.columns(),
nowInSec);
indexTransaction.start();
indexTransaction.onPartitionDeletion(partition.partitionLevelDeletion());
indexTransaction.onPartitionDeletion(new DeletionTime(FBUtilities.timestampMicros(), nowInSec));
indexTransaction.commit();
while (partition.hasNext())
@ -978,8 +978,13 @@ public class SecondaryIndexManager implements IndexRegistry
{
Index.Indexer indexer = index.indexerFor(key, nowInSec, opGroup, Type.CLEANUP);
indexer.begin();
if (partitionDelete != null)
indexer.partitionDelete(partitionDelete);
if (row != null)
indexer.removeRow(row);
indexer.finish();
}
}

33
test/unit/org/apache/cassandra/index/CustomIndexTest.java
View File

@ -9,6 +9,7 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.junit.Test;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ColumnIdentifier;
@ -18,12 +19,15 @@ import org.apache.cassandra.cql3.statements.ModificationStatement;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadOrderGroup;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.Indexes;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.Util.throwAssert;
import static org.apache.cassandra.cql3.statements.IndexTarget.CUSTOM_INDEX_OPTION_NAME;
@ -471,6 +475,35 @@ public class CustomIndexTest extends CQLTester
assertEquals(1, index.reloads.get());
}
@Test
public void notifyIndexersOfPartitionAndRowRemovalDuringCleanup() throws Throwable
{
createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k,c))");
createIndex(String.format("CREATE CUSTOM INDEX cleanup_index ON %%s() USING '%s'", StubIndex.class.getName()));
ColumnFamilyStore cfs = getCurrentColumnFamilyStore();
StubIndex index = (StubIndex)cfs.indexManager.getIndexByName("cleanup_index");
execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 0, 0);
execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 1, 1);
execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 0, 2, 2);
execute("INSERT INTO %s (k, c, v) VALUES (?, ?, ?)", 3, 3, 3);
assertEquals(4, index.rowsInserted.size());
assertEquals(0, index.partitionDeletions.size());
ReadCommand cmd = Util.cmd(cfs, 0).build();
try (ReadOrderGroup orderGroup = cmd.startOrderGroup();
UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
{
assertTrue(iterator.hasNext());
cfs.indexManager.deletePartition(iterator.next(), FBUtilities.nowInSeconds());
}
assertEquals(1, index.partitionDeletions.size());
assertEquals(3, index.rowsDeleted.size());
for (int i = 0; i < 3; i++)
assertEquals(index.rowsDeleted.get(i).clustering(), index.rowsInserted.get(i).clustering());
}
private void testCreateIndex(String indexName, String... targetColumnNames) throws Throwable
{
createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(%s) USING '%s'",