diff --git a/CHANGES.txt b/CHANGES.txt index 50fb26876b..e7a2ff0efc 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -22,6 +22,7 @@ dev * change RandomPartitioner min token to -1 to avoid collision w/ tokens on actual nodes (CASSANDRA-1901) * examine the right nibble when validating TimeUUID (CASSANDRA-1910) + * include secondary indexes in cleanup (CASSANDRA-1916) * CFS.scrubDataDirectories should also cleanup invalid secondary indexes (CASSANDRA-1904) diff --git a/src/java/org/apache/cassandra/db/CompactionManager.java b/src/java/org/apache/cassandra/db/CompactionManager.java index d08bf96d60..cf3f2c349f 100644 --- a/src/java/org/apache/cassandra/db/CompactionManager.java +++ b/src/java/org/apache/cassandra/db/CompactionManager.java @@ -18,10 +18,12 @@ package org.apache.cassandra.db; +import java.io.DataOutput; import java.io.File; import java.io.IOException; import java.lang.management.ManagementFactory; -import java.net.InetAddress; +import java.nio.ByteBuffer; +import java.security.MessageDigest; import java.util.*; import java.util.Map.Entry; import java.util.concurrent.*; @@ -31,7 +33,6 @@ import javax.management.MBeanServer; import javax.management.ObjectName; import org.apache.commons.collections.PredicateUtils; -import org.apache.commons.collections.iterators.CollatingIterator; import org.apache.commons.collections.iterators.FilterIterator; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; @@ -47,7 +48,6 @@ import org.apache.cassandra.io.sstable.*; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.service.AntiEntropyService; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; import org.cliffc.high_scale_lib.NonBlockingHashMap; @@ -367,83 +367,6 @@ public class CompactionManager implements CompactionManagerMBean return max; } - /** - * This function is used to do the anti compaction process , it spits out the file which has keys that belong to a given range - * If the target is not specified it spits out the file as a compacted file with the unecessary ranges wiped out. - * - * @param cfs - * @param sstables - * @param ranges - * @param target - * @return - * @throws java.io.IOException - */ - private List doAntiCompaction(ColumnFamilyStore cfs, Collection sstables, Collection ranges, InetAddress target) - throws IOException - { - Table table = cfs.table; - logger.info("AntiCompacting [" + StringUtils.join(sstables, ",") + "]"); - // Calculate the expected compacted filesize - long expectedRangeFileSize = cfs.getExpectedCompactedFileSize(sstables) / 2; - String compactionFileLocation = table.getDataFileLocation(expectedRangeFileSize); - if (compactionFileLocation == null) - { - throw new UnsupportedOperationException("disk full"); - } - - List results = new ArrayList(); - long startTime = System.currentTimeMillis(); - long totalkeysWritten = 0; - - int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), (int)(SSTableReader.getApproximateKeyCount(sstables) / 2)); - if (logger.isDebugEnabled()) - logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); - - SSTableWriter writer = null; - CompactionIterator ci = new AntiCompactionIterator(cfs, sstables, ranges, (int) (System.currentTimeMillis() / 1000) - cfs.metadata.getGcGraceSeconds(), cfs.isCompleteSSTables(sstables)); - Iterator nni = new FilterIterator(ci, PredicateUtils.notNullPredicate()); - executor.beginCompaction(cfs, ci); - - try - { - if (!nni.hasNext()) - { - return results; - } - - while (nni.hasNext()) - { - AbstractCompactedRow row = nni.next(); - if (writer == null) - { - FileUtils.createDirectory(compactionFileLocation); - String newFilename = new File(cfs.getTempSSTablePath(compactionFileLocation)).getAbsolutePath(); - writer = new SSTableWriter(newFilename, expectedBloomFilterSize, cfs.metadata, cfs.partitioner); - } - writer.append(row); - totalkeysWritten++; - } - } - finally - { - ci.close(); - } - - if (writer != null) - { - results.add(writer.closeAndOpenReader(getMaxDataAge(sstables))); - - String format = "AntiCompacted to %s. %,d to %,d (~%d%% of original) bytes for %,d keys. Time: %,dms."; - long dTime = System.currentTimeMillis() - startTime; - long startsize = SSTable.getTotalBytes(sstables); - long endsize = results.get(0).length(); - double ratio = (double)endsize / (double)startsize; - logger.info(String.format(format, writer.getFilename(), startsize, endsize, (int)(ratio*100), totalkeysWritten, dTime)); - } - - return results; - } - /** * This function goes over each file and removes the keys that the node is not responsible for * and only keeps keys that this node is responsible for. @@ -452,14 +375,104 @@ public class CompactionManager implements CompactionManagerMBean */ private void doCleanupCompaction(ColumnFamilyStore cfs) throws IOException { - Collection originalSSTables = cfs.getSSTables(); - List sstables = doAntiCompaction(cfs, originalSSTables, StorageService.instance.getLocalRanges(cfs.table.name), null); - if (!sstables.isEmpty()) + assert !cfs.isIndex(); + Table table = cfs.table; + Collection ranges = StorageService.instance.getLocalRanges(table.name); + + for (SSTableReader sstable : cfs.getSSTables()) { - cfs.replaceCompactedSSTables(originalSSTables, sstables); + logger.info("AntiCompacting " + sstable); + // Calculate the expected compacted filesize + long expectedRangeFileSize = cfs.getExpectedCompactedFileSize(Arrays.asList(sstable)) / 2; + String compactionFileLocation = table.getDataFileLocation(expectedRangeFileSize); + if (compactionFileLocation == null) + throw new UnsupportedOperationException("disk full"); + + long startTime = System.currentTimeMillis(); + long totalkeysWritten = 0; + + int expectedBloomFilterSize = Math.max(DatabaseDescriptor.getIndexInterval(), + (int)(SSTableReader.getApproximateKeyCount(Arrays.asList(sstable)) / 2)); + if (logger.isDebugEnabled()) + logger.debug("Expected bloom filter size : " + expectedBloomFilterSize); + + SSTableWriter writer = null; + SSTableScanner scanner = sstable.getDirectScanner(CompactionIterator.FILE_BUFFER_SIZE); + SortedSet indexedColumns = cfs.getIndexedColumns(); + executor.beginCompaction(cfs, new CleanupInfo(sstable, scanner)); + try + { + while (scanner.hasNext()) + { + SSTableIdentityIterator row = (SSTableIdentityIterator) scanner.next(); + if (Range.isTokenInRanges(row.getKey().token, ranges)) + { + writer = maybeCreateWriter(cfs, compactionFileLocation, expectedBloomFilterSize, writer); + writer.append(new EchoedRow(row)); + totalkeysWritten++; + } + else + { + while (row.hasNext()) + { + IColumn column = row.next(); + if (indexedColumns.contains(column.name())) + Table.cleanupIndexEntry(cfs, row.getKey().key, column); + } + } + } + } + finally + { + scanner.close(); + } + + List results = new ArrayList(); + if (writer != null) + { + SSTableReader newSstable = writer.closeAndOpenReader(sstable.maxDataAge); + results.add(newSstable); + + String format = "AntiCompacted to %s. %,d to %,d (~%d%% of original) bytes for %,d keys. Time: %,dms."; + long dTime = System.currentTimeMillis() - startTime; + long startsize = sstable.length(); + long endsize = newSstable.length(); + double ratio = (double)endsize / (double)startsize; + logger.info(String.format(format, writer.getFilename(), startsize, endsize, (int)(ratio*100), totalkeysWritten, dTime)); + } + + // flush to ensure we don't lose the tombstones on a restart, since they are not commitlog'd + for (ByteBuffer columnName : cfs.getIndexedColumns()) + { + try + { + cfs.getIndexedColumnFamilyStore(columnName).forceBlockingFlush(); + } + catch (ExecutionException e) + { + throw new RuntimeException(e); + } + catch (InterruptedException e) + { + throw new AssertionError(e); + } + } + cfs.replaceCompactedSSTables(Arrays.asList(sstable), results); } } + private SSTableWriter maybeCreateWriter(ColumnFamilyStore cfs, String compactionFileLocation, int expectedBloomFilterSize, SSTableWriter writer) + throws IOException + { + if (writer == null) + { + FileUtils.createDirectory(compactionFileLocation); + String newFilename = new File(cfs.getTempSSTablePath(compactionFileLocation)).getAbsolutePath(); + writer = new SSTableWriter(newFilename, expectedBloomFilterSize, cfs.metadata, cfs.partitioner); + } + return writer; + } + /** * Performs a readonly "compaction" of all sstables in order to validate complete rows, * but without writing the merge result @@ -633,55 +646,6 @@ public class CompactionManager implements CompactionManagerMBean } } - private static class AntiCompactionIterator extends CompactionIterator - { - private Set scanners; - - public AntiCompactionIterator(ColumnFamilyStore cfStore, Collection sstables, Collection ranges, int gcBefore, boolean isMajor) - throws IOException - { - super(cfStore, getCollatedRangeIterator(sstables, ranges), gcBefore, isMajor); - } - - private static Iterator getCollatedRangeIterator(Collection sstables, final Collection ranges) - throws IOException - { - org.apache.commons.collections.Predicate rangesPredicate = new org.apache.commons.collections.Predicate() - { - public boolean evaluate(Object row) - { - return Range.isTokenInRanges(((SSTableIdentityIterator)row).getKey().token, ranges); - } - }; - // TODO CollatingIterator iter = FBUtilities.getCollatingIterator(); - CollatingIterator iter = FBUtilities.getCollatingIterator(); - for (SSTableReader sstable : sstables) - { - SSTableScanner scanner = sstable.getDirectScanner(FILE_BUFFER_SIZE); - iter.addIterator(new FilterIterator(scanner, rangesPredicate)); - } - return iter; - } - - public Iterable getScanners() - { - if (scanners == null) - { - scanners = new HashSet(); - for (Object o : ((CollatingIterator)source).getIterators()) - { - scanners.add((SSTableScanner)((FilterIterator)o).getIterator()); - } - } - return scanners; - } - - public String getTaskType() - { - return "Cleanup"; - } - } - public void checkAllColumnFamilies() throws IOException { // perform estimates @@ -821,4 +785,63 @@ public class CompactionManager implements CompactionManagerMBean throw new IllegalStateException("May not call SimpleFuture.get(long, TimeUnit)"); } } + + private static class EchoedRow extends AbstractCompactedRow + { + private final SSTableIdentityIterator row; + + public EchoedRow(SSTableIdentityIterator row) + { + super(row.getKey()); + this.row = row; + } + + public void write(DataOutput out) throws IOException + { + row.echoData(out); + } + + public void update(MessageDigest digest) + { + // EchoedRow is not used in anti-entropy validation + throw new UnsupportedOperationException(); + } + + public boolean isEmpty() + { + return !row.hasNext(); + } + + public int columnCount() + { + return row.columnCount; + } + } + + private static class CleanupInfo implements ICompactionInfo + { + private final SSTableReader sstable; + private final SSTableScanner scanner; + + public CleanupInfo(SSTableReader sstable, SSTableScanner scanner) + { + this.sstable = sstable; + this.scanner = scanner; + } + + public long getTotalBytes() + { + return scanner.getFileLength(); + } + + public long getBytesRead() + { + return scanner.getFilePointer(); + } + + public String getTaskType() + { + return "Cleanup of " + sstable.getColumnFamilyName(); + } + } } diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 10f7c4db1f..0274446d20 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -571,6 +571,21 @@ public class Table return fullMemtables; } + public static void cleanupIndexEntry(ColumnFamilyStore cfs, ByteBuffer key, IColumn column) + { + if (column.isMarkedForDelete()) + return; + int localDeletionTime = (int) (System.currentTimeMillis() / 1000); + DecoratedKey valueKey = cfs.getIndexKeyFor(column.name(), column.value()); + ColumnFamily cfi = cfs.newIndexedColumnFamily(column.name()); + cfi.addTombstone(key, localDeletionTime, column.timestamp()); + Memtable fullMemtable = cfs.getIndexedColumnFamilyStore(column.name()).apply(valueKey, cfi); + if (logger.isDebugEnabled()) + logger.debug("removed index entry for cleaned-up value {}:{}", valueKey, cfi); + if (fullMemtable != null) + fullMemtable.cfs.maybeSwitchMemtable(fullMemtable, false); + } + public IndexBuilder createIndexBuilder(ColumnFamilyStore cfs, SortedSet columns, ReducingKeyIterator iter) { return new IndexBuilder(cfs, columns, iter); diff --git a/src/java/org/apache/cassandra/io/CompactionIterator.java b/src/java/org/apache/cassandra/io/CompactionIterator.java index 16b10b98c6..b99fe3f271 100644 --- a/src/java/org/apache/cassandra/io/CompactionIterator.java +++ b/src/java/org/apache/cassandra/io/CompactionIterator.java @@ -44,7 +44,7 @@ implements Closeable, ICompactionInfo { private static Logger logger = LoggerFactory.getLogger(CompactionIterator.class); - protected static final int FILE_BUFFER_SIZE = 1024 * 1024; + public static final int FILE_BUFFER_SIZE = 1024 * 1024; protected final List rows = new ArrayList(); private final ColumnFamilyStore cfs; diff --git a/test/unit/org/apache/cassandra/db/CleanupTest.java b/test/unit/org/apache/cassandra/db/CleanupTest.java new file mode 100644 index 0000000000..ddf338dc03 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/CleanupTest.java @@ -0,0 +1,124 @@ +/* +* 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; + +import org.junit.Test; + +import org.apache.cassandra.CleanupHelper; +import org.apache.cassandra.Util; +import org.apache.cassandra.db.columniterator.IdentityQueryFilter; +import org.apache.cassandra.db.filter.IFilter; +import org.apache.cassandra.db.filter.QueryFilter; +import org.apache.cassandra.db.filter.QueryPath; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.dht.Range; +import org.apache.cassandra.locator.TokenMetadata; +import org.apache.cassandra.service.StorageService; +import org.apache.cassandra.thrift.IndexClause; +import org.apache.cassandra.thrift.IndexExpression; +import org.apache.cassandra.thrift.IndexOperator; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; + +public class CleanupTest extends CleanupHelper +{ + public static final int LOOPS = 800; + public static final String TABLE1 = "Keyspace1"; + public static final String CF1 = "Indexed1"; + public static final ByteBuffer COLUMN = ByteBuffer.wrap("birthdate".getBytes()); + public static final ByteBuffer VALUE = ByteBuffer.allocate(8); + static + { + VALUE.putLong(20101229); + VALUE.flip(); + } + + @Test + public void testCleanup() throws IOException, ExecutionException, InterruptedException + { + Table table = Table.open(TABLE1); + + ColumnFamilyStore cfs = table.getColumnFamilyStore(CF1); + fillCF(cfs, LOOPS); + + assertEquals(cfs.getIndexedColumns().iterator().next(), COLUMN); + + ColumnFamilyStore cfi = cfs.getIndexedColumnFamilyStore(COLUMN); + + assertTrue(cfi.isIndexBuilt()); + + IndexExpression expr = new IndexExpression(COLUMN, IndexOperator.EQ, VALUE); + IndexClause clause = new IndexClause(Arrays.asList(expr), FBUtilities.EMPTY_BYTE_BUFFER, Integer.MAX_VALUE); + IFilter filter = new IdentityQueryFilter(); + IPartitioner p = StorageService.getPartitioner(); + Range range = new Range(p.getMinimumToken(), p.getMinimumToken()); + List rows = table.getColumnFamilyStore(CF1).scan(clause, range, filter); + + assertEquals(LOOPS, rows.size()); + + TokenMetadata tmd = StorageService.instance.getTokenMetadata(); + + assertNotNull(tmd); + assertEquals(0, tmd.getTokenToEndpointMap().size()); + + // Since this test has no ring cleanup will remove all + CompactionManager.instance.performCleanup(cfs); + + // row data should be gone + rows = cfs.getRangeSlice(null, Util.range("", ""), 1000, new IdentityQueryFilter()); + assertEquals(0, rows.size()); + + // not only should it be gone but there should be no data on disk, not even tombstones + assert cfs.getSSTables().isEmpty(); + + // 2ary indexes should result in no results, but + rows = cfs.scan(clause, range, filter); + assertEquals(0, rows.size()); + } + + protected void fillCF(ColumnFamilyStore store, int rowsPerSSTable) throws ExecutionException, InterruptedException, IOException + { + CompactionManager.instance.disableAutoCompaction(); + + for (int i = 0; i < rowsPerSSTable; i++) + { + String key = String.valueOf(i); + + // create a row and update the birthdate value, test that the index query fetches the new version + RowMutation rm; + rm = new RowMutation(TABLE1, ByteBufferUtil.bytes(key)); + rm.add(new QueryPath(CF1, null, COLUMN), VALUE, System.currentTimeMillis()); + rm.apply(); + } + + store.forceBlockingFlush(); + store.buildSecondaryIndexes(store.getSSTables(), store.getIndexedColumns()); + } +}