| /* |
| * 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.util.*; |
| import java.util.concurrent.ExecutionException; |
| import java.util.concurrent.Future; |
| |
| import com.google.common.collect.ImmutableMap; |
| import com.google.common.collect.Iterators; |
| import org.junit.BeforeClass; |
| import org.junit.Test; |
| |
| import org.apache.cassandra.SchemaLoader; |
| import org.apache.cassandra.UpdateBuilder; |
| import org.apache.cassandra.Util; |
| import org.apache.cassandra.config.ColumnDefinition; |
| import org.apache.cassandra.cql3.statements.IndexTarget; |
| import org.apache.cassandra.db.compaction.CompactionManager; |
| import org.apache.cassandra.db.filter.ColumnFilter; |
| import org.apache.cassandra.db.marshal.Int32Type; |
| import org.apache.cassandra.db.marshal.UTF8Type; |
| import org.apache.cassandra.db.partitions.*; |
| import org.apache.cassandra.db.rows.*; |
| import org.apache.cassandra.exceptions.ConfigurationException; |
| import org.apache.cassandra.index.StubIndex; |
| import org.apache.cassandra.io.sstable.format.SSTableReader; |
| import org.apache.cassandra.io.sstable.metadata.StatsMetadata; |
| import org.apache.cassandra.schema.IndexMetadata; |
| import org.apache.cassandra.schema.KeyspaceParams; |
| import org.apache.cassandra.utils.ByteBufferUtil; |
| import org.apache.cassandra.utils.FBUtilities; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertFalse; |
| import static org.junit.Assert.assertTrue; |
| |
| public class RangeTombstoneTest |
| { |
| private static final String KSNAME = "RangeTombstoneTest"; |
| private static final String CFNAME = "StandardInteger1"; |
| |
| @BeforeClass |
| public static void defineSchema() throws ConfigurationException |
| { |
| SchemaLoader.prepareServer(); |
| SchemaLoader.createKeyspace(KSNAME, |
| KeyspaceParams.simple(1), |
| SchemaLoader.standardCFMD(KSNAME, |
| CFNAME, |
| 1, |
| UTF8Type.instance, |
| Int32Type.instance, |
| Int32Type.instance)); |
| } |
| |
| @Test |
| public void simpleQueryWithRangeTombstoneTest() throws Exception |
| { |
| Keyspace keyspace = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CFNAME); |
| boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness(); |
| |
| // Inserting data |
| String key = "k1"; |
| |
| UpdateBuilder builder; |
| |
| builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 40; i += 2) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(10, 22).build().applyUnsafe(); |
| |
| builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(2); |
| for (int i = 1; i < 40; i += 2) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| |
| new RowUpdateBuilder(cfs.metadata, 3, key).addRangeTombstone(19, 27).build().applyUnsafe(); |
| // We don't flush to test with both a range tomsbtone in memtable and in sstable |
| |
| // Queries by name |
| int[] live = new int[]{ 4, 9, 11, 17, 28 }; |
| int[] dead = new int[]{ 12, 19, 21, 24, 27 }; |
| |
| AbstractReadCommandBuilder.SinglePartitionBuilder cmdBuilder = Util.cmd(cfs, key); |
| for (int i : live) |
| cmdBuilder.includeRow(i); |
| for (int i : dead) |
| cmdBuilder.includeRow(i); |
| |
| Partition partition = Util.getOnlyPartitionUnfiltered(cmdBuilder.build()); |
| int nowInSec = FBUtilities.nowInSeconds(); |
| |
| for (int i : live) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| for (int i : dead) |
| assertFalse("Row " + i + " shouldn't be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| |
| // Queries by slices |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(7).toIncl(30).build()); |
| |
| for (int i : new int[]{ 7, 8, 9, 11, 13, 15, 17, 28, 29, 30 }) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| for (int i : new int[]{ 10, 12, 14, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27 }) |
| assertFalse("Row " + i + " shouldn't be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| } |
| |
| @Test |
| public void rangeTombstoneFilteringTest() throws Exception |
| { |
| CompactionManager.instance.disableAutoCompaction(); |
| Keyspace keyspace = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CFNAME); |
| |
| // Inserting data |
| String key = "k111"; |
| |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 40; i += 2) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(5, 10).build().applyUnsafe(); |
| |
| new RowUpdateBuilder(cfs.metadata, 2, key).addRangeTombstone(15, 20).build().applyUnsafe(); |
| |
| ImmutableBTreePartition partition; |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(11).toIncl(14).build()); |
| Collection<RangeTombstone> rt = rangeTombstones(partition); |
| assertEquals(0, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(11).toIncl(15).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(20).toIncl(25).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(12).toIncl(25).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(25).toIncl(35).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(0, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(1).toIncl(40).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(2, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(7).toIncl(17).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(2, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(5).toIncl(20).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(2, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(5).toIncl(20).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(2, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(1).toIncl(2).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(0, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(1).toIncl(5).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(1).toIncl(10).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(5).toIncl(6).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(17).toIncl(20).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).fromIncl(17).toIncl(18).build()); |
| rt = rangeTombstones(partition); |
| assertEquals(1, rt.size()); |
| |
| Slices.Builder sb = new Slices.Builder(cfs.getComparator()); |
| sb.add(ClusteringBound.create(cfs.getComparator(), true, true, 1), ClusteringBound.create(cfs.getComparator(), false, true, 10)); |
| sb.add(ClusteringBound.create(cfs.getComparator(), true, true, 16), ClusteringBound.create(cfs.getComparator(), false, true, 20)); |
| |
| partition = Util.getOnlyPartitionUnfiltered(SinglePartitionReadCommand.create(cfs.metadata, FBUtilities.nowInSeconds(), Util.dk(key), sb.build())); |
| rt = rangeTombstones(partition); |
| assertEquals(2, rt.size()); |
| } |
| |
| private Collection<RangeTombstone> rangeTombstones(ImmutableBTreePartition partition) |
| { |
| List<RangeTombstone> tombstones = new ArrayList<RangeTombstone>(); |
| Iterators.addAll(tombstones, partition.deletionInfo().rangeIterator(false)); |
| return tombstones; |
| } |
| |
| @Test |
| public void testTrackTimesPartitionTombstone() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.truncateBlocking(); |
| String key = "rt_times"; |
| |
| int nowInSec = FBUtilities.nowInSeconds(); |
| new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 1000, nowInSec)).apply(); |
| cfs.forceBlockingFlush(); |
| |
| SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec); |
| cfs.forceMajorCompaction(); |
| sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec); |
| } |
| |
| @Test |
| public void testTrackTimesPartitionTombstoneWithData() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.truncateBlocking(); |
| String key = "rt_times"; |
| |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(999).newRow(5).add("val", 5).apply(); |
| |
| key = "rt_times2"; |
| int nowInSec = FBUtilities.nowInSeconds(); |
| new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 1000, nowInSec)).apply(); |
| cfs.forceBlockingFlush(); |
| |
| SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE); |
| cfs.forceMajorCompaction(); |
| sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE); |
| } |
| |
| @Test |
| public void testTrackTimesRangeTombstone() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.truncateBlocking(); |
| String key = "rt_times"; |
| |
| int nowInSec = FBUtilities.nowInSeconds(); |
| new RowUpdateBuilder(cfs.metadata, nowInSec, 1000L, key).addRangeTombstone(1, 2).build().apply(); |
| cfs.forceBlockingFlush(); |
| |
| SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec); |
| cfs.forceMajorCompaction(); |
| sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 1000, 1000, nowInSec); |
| } |
| |
| @Test |
| public void testTrackTimesRangeTombstoneWithData() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.truncateBlocking(); |
| String key = "rt_times"; |
| |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(999).newRow(5).add("val", 5).apply(); |
| |
| key = "rt_times2"; |
| int nowInSec = FBUtilities.nowInSeconds(); |
| new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 1000, nowInSec)).apply(); |
| cfs.forceBlockingFlush(); |
| |
| cfs.forceBlockingFlush(); |
| SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE); |
| cfs.forceMajorCompaction(); |
| sstable = cfs.getLiveSSTables().iterator().next(); |
| assertTimes(sstable.getSSTableMetadata(), 999, 1000, Integer.MAX_VALUE); |
| } |
| |
| private void assertTimes(StatsMetadata metadata, long min, long max, int localDeletionTime) |
| { |
| assertEquals(min, metadata.minTimestamp); |
| assertEquals(max, metadata.maxTimestamp); |
| assertEquals(localDeletionTime, metadata.maxLocalDeletionTime); |
| } |
| |
| @Test |
| public void test7810() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.metadata.gcGraceSeconds(2); |
| |
| String key = "7810"; |
| |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 10; i < 20; i ++) |
| builder.newRow(i).add("val", i); |
| builder.apply(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(10, 11).build().apply(); |
| cfs.forceBlockingFlush(); |
| |
| Thread.sleep(5); |
| cfs.forceMajorCompaction(); |
| assertEquals(8, Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()).rowCount()); |
| } |
| |
| @Test |
| public void test7808_1() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.metadata.gcGraceSeconds(2); |
| |
| String key = "7808_1"; |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 40; i += 2) |
| builder.newRow(i).add("val", i); |
| builder.apply(); |
| cfs.forceBlockingFlush(); |
| |
| new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 1, 1)).apply(); |
| cfs.forceBlockingFlush(); |
| Thread.sleep(5); |
| cfs.forceMajorCompaction(); |
| } |
| |
| @Test |
| public void test7808_2() throws ExecutionException, InterruptedException |
| { |
| Keyspace ks = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = ks.getColumnFamilyStore(CFNAME); |
| cfs.metadata.gcGraceSeconds(2); |
| |
| String key = "7808_2"; |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 10; i < 20; i ++) |
| builder.newRow(i).add("val", i); |
| builder.apply(); |
| cfs.forceBlockingFlush(); |
| |
| new Mutation(PartitionUpdate.fullPartitionDelete(cfs.metadata, Util.dk(key), 0, 0)).apply(); |
| |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(1).newRow(5).add("val", 5).apply(); |
| |
| cfs.forceBlockingFlush(); |
| Thread.sleep(5); |
| cfs.forceMajorCompaction(); |
| assertEquals(1, Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()).rowCount()); |
| } |
| |
| @Test |
| public void overlappingRangeTest() throws Exception |
| { |
| CompactionManager.instance.disableAutoCompaction(); |
| Keyspace keyspace = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CFNAME); |
| boolean enforceStrictLiveness = cfs.metadata.enforceStrictLiveness(); |
| // Inserting data |
| String key = "k2"; |
| |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 20; i++) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(5, 15).build().applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(5, 10).build().applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 2, key).addRangeTombstone(5, 8).build().applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| Partition partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()); |
| int nowInSec = FBUtilities.nowInSeconds(); |
| |
| for (int i = 0; i < 5; i++) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| for (int i = 16; i < 20; i++) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| for (int i = 5; i <= 15; i++) |
| assertFalse("Row " + i + " shouldn't be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| |
| // Compact everything and re-test |
| CompactionManager.instance.performMaximal(cfs, false); |
| partition = Util.getOnlyPartitionUnfiltered(Util.cmd(cfs, key).build()); |
| |
| for (int i = 0; i < 5; i++) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds(), |
| enforceStrictLiveness)); |
| for (int i = 16; i < 20; i++) |
| assertTrue("Row " + i + " should be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(FBUtilities.nowInSeconds(), |
| enforceStrictLiveness)); |
| for (int i = 5; i <= 15; i++) |
| assertFalse("Row " + i + " shouldn't be live", |
| partition.getRow(Clustering.make(bb(i))).hasLiveData(nowInSec, enforceStrictLiveness)); |
| } |
| |
| @Test |
| public void reverseQueryTest() throws Exception |
| { |
| Keyspace table = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME); |
| |
| // Inserting data |
| String key = "k3"; |
| |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(0).newRow(2).add("val", 2).applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(0, 10).build().applyUnsafe(); |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(2).newRow(1).add("val", 1).applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| // Get the last value of the row |
| FilteredPartition partition = Util.getOnlyPartition(Util.cmd(cfs, key).build()); |
| assertTrue(partition.rowCount() > 0); |
| |
| int last = i(partition.unfilteredIterator(ColumnFilter.all(cfs.metadata), Slices.ALL, true).next().clustering().get(0)); |
| assertEquals("Last column should be column 1 since column 2 has been deleted", 1, last); |
| } |
| |
| @Test |
| public void testRowWithRangeTombstonesUpdatesSecondaryIndex() throws Exception |
| { |
| Keyspace table = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME); |
| ByteBuffer key = ByteBufferUtil.bytes("k5"); |
| ByteBuffer indexedColumnName = ByteBufferUtil.bytes("val"); |
| |
| cfs.truncateBlocking(); |
| cfs.disableAutoCompaction(); |
| |
| ColumnDefinition cd = cfs.metadata.getColumnDefinition(indexedColumnName).copy(); |
| IndexMetadata indexDef = |
| IndexMetadata.fromIndexTargets(cfs.metadata, |
| Collections.singletonList(new IndexTarget(cd.name, IndexTarget.Type.VALUES)), |
| "test_index", |
| IndexMetadata.Kind.CUSTOM, |
| ImmutableMap.of(IndexTarget.CUSTOM_INDEX_OPTION_NAME, |
| StubIndex.class.getName())); |
| |
| if (!cfs.metadata.getIndexes().get("test_index").isPresent()) |
| cfs.metadata.indexes(cfs.metadata.getIndexes().with(indexDef)); |
| |
| Future<?> rebuild = cfs.indexManager.addIndex(indexDef); |
| // If rebuild there is, wait for the rebuild to finish so it doesn't race with the following insertions |
| if (rebuild != null) |
| rebuild.get(); |
| |
| StubIndex index = (StubIndex)cfs.indexManager.listIndexes() |
| .stream() |
| .filter(i -> "test_index".equals(i.getIndexMetadata().name)) |
| .findFirst() |
| .orElseThrow(() -> new RuntimeException(new AssertionError("Index not found"))); |
| index.reset(); |
| |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 10; i++) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 0, key).addRangeTombstone(0, 7).build().applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| assertEquals(10, index.rowsInserted.size()); |
| |
| CompactionManager.instance.performMaximal(cfs, false); |
| |
| // compacted down to single sstable |
| assertEquals(1, cfs.getLiveSSTables().size()); |
| |
| assertEquals(8, index.rowsDeleted.size()); |
| } |
| |
| @Test |
| public void testRangeTombstoneCompaction() throws Exception |
| { |
| Keyspace table = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME); |
| ByteBuffer key = ByteBufferUtil.bytes("k4"); |
| |
| // remove any existing sstables before starting |
| cfs.truncateBlocking(); |
| cfs.disableAutoCompaction(); |
| |
| UpdateBuilder builder = UpdateBuilder.create(cfs.metadata, key).withTimestamp(0); |
| for (int i = 0; i < 10; i += 2) |
| builder.newRow(i).add("val", i); |
| builder.applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| new RowUpdateBuilder(cfs.metadata, 0, key).addRangeTombstone(0, 7).build().applyUnsafe(); |
| cfs.forceBlockingFlush(); |
| |
| // there should be 2 sstables |
| assertEquals(2, cfs.getLiveSSTables().size()); |
| |
| // compact down to single sstable |
| CompactionManager.instance.performMaximal(cfs, false); |
| assertEquals(1, cfs.getLiveSSTables().size()); |
| |
| // test the physical structure of the sstable i.e. rt & columns on disk |
| SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); |
| try (UnfilteredPartitionIterator scanner = sstable.getScanner()) |
| { |
| try (UnfilteredRowIterator iter = scanner.next()) |
| { |
| // after compaction, we should have a single RT with a single row (the row 8) |
| Unfiltered u1 = iter.next(); |
| assertTrue("Expecting open marker, got " + u1.toString(cfs.metadata), u1 instanceof RangeTombstoneMarker); |
| Unfiltered u2 = iter.next(); |
| assertTrue("Expecting close marker, got " + u2.toString(cfs.metadata), u2 instanceof RangeTombstoneMarker); |
| Unfiltered u3 = iter.next(); |
| assertTrue("Expecting row, got " + u3.toString(cfs.metadata), u3 instanceof Row); |
| } |
| } |
| } |
| |
| @Test |
| public void testOverwritesToDeletedColumns() throws Exception |
| { |
| Keyspace table = Keyspace.open(KSNAME); |
| ColumnFamilyStore cfs = table.getColumnFamilyStore(CFNAME); |
| ByteBuffer key = ByteBufferUtil.bytes("k6"); |
| ByteBuffer indexedColumnName = ByteBufferUtil.bytes("val"); |
| |
| cfs.truncateBlocking(); |
| cfs.disableAutoCompaction(); |
| |
| ColumnDefinition cd = cfs.metadata.getColumnDefinition(indexedColumnName).copy(); |
| IndexMetadata indexDef = |
| IndexMetadata.fromIndexTargets(cfs.metadata, |
| Collections.singletonList(new IndexTarget(cd.name, IndexTarget.Type.VALUES)), |
| "test_index", |
| IndexMetadata.Kind.CUSTOM, |
| ImmutableMap.of(IndexTarget.CUSTOM_INDEX_OPTION_NAME, |
| StubIndex.class.getName())); |
| |
| if (!cfs.metadata.getIndexes().get("test_index").isPresent()) |
| cfs.metadata.indexes(cfs.metadata.getIndexes().with(indexDef)); |
| |
| Future<?> rebuild = cfs.indexManager.addIndex(indexDef); |
| // If rebuild there is, wait for the rebuild to finish so it doesn't race with the following insertions |
| if (rebuild != null) |
| rebuild.get(); |
| |
| StubIndex index = (StubIndex)cfs.indexManager.getIndexByName("test_index"); |
| index.reset(); |
| |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(0).newRow(1).add("val", 1).applyUnsafe(); |
| |
| // add a RT which hides the column we just inserted |
| new RowUpdateBuilder(cfs.metadata, 1, key).addRangeTombstone(0, 1).build().applyUnsafe(); |
| |
| // now re-insert that column |
| UpdateBuilder.create(cfs.metadata, key).withTimestamp(2).newRow(1).add("val", 1).applyUnsafe(); |
| |
| cfs.forceBlockingFlush(); |
| |
| // We should have 1 insert and 1 update to the indexed "1" column |
| // CASSANDRA-6640 changed index update to just update, not insert then delete |
| assertEquals(1, index.rowsInserted.size()); |
| assertEquals(1, index.rowsUpdated.size()); |
| } |
| |
| private static ByteBuffer bb(int i) |
| { |
| return ByteBufferUtil.bytes(i); |
| } |
| |
| private static int i(ByteBuffer bb) |
| { |
| return ByteBufferUtil.toInt(bb); |
| } |
| } |
