diff --git a/CHANGES.txt b/CHANGES.txt index 99a5d51f84..2642c23860 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -118,6 +118,7 @@ * Log failed host when preparing incremental repair (CASSANDRA-8228) * Force config client mode in CQLSSTableWriter (CASSANDRA-8281) Merged from 2.0: + * Add batch remove iterator to ABSC (CASSANDRA-8414) * Round up time deltas lower than 1ms in BulkLoader (CASSANDRA-8645) * Use more efficient slice size for querying internal secondary index tables (CASSANDRA-8550) diff --git a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java index e5a731c7ac..9574d0928e 100644 --- a/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java +++ b/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java @@ -17,12 +17,7 @@ */ package org.apache.cassandra.db; -import java.util.AbstractCollection; -import java.util.Arrays; -import java.util.Collection; -import java.util.Comparator; -import java.util.Iterator; -import java.util.NoSuchElementException; +import java.util.*; import com.google.common.base.Function; import com.google.common.collect.AbstractIterator; @@ -32,6 +27,7 @@ import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.composites.CellName; import org.apache.cassandra.db.composites.Composite; import org.apache.cassandra.db.filter.ColumnSlice; +import org.apache.cassandra.utils.BatchRemoveIterator; import org.apache.cassandra.utils.memory.AbstractAllocator; import org.apache.cassandra.utils.SearchIterator; @@ -115,6 +111,94 @@ public class ArrayBackedSortedColumns extends ColumnFamily return reversed; } + public BatchRemoveIterator batchRemoveIterator() + { + maybeSortCells(); + + return new BatchRemoveIterator() + { + private final Iterator iter = iterator(); + private BitSet removedIndexes = new BitSet(size); + private int idx = -1; + private boolean shouldCallNext = false; + private boolean isCommitted = false; + private boolean removedAnything = false; + + public void commit() + { + if (isCommitted) + throw new IllegalStateException(); + isCommitted = true; + + if (!removedAnything) + return; + + // the lowest index both not visited and known to be not removed + int keepIdx = removedIndexes.nextClearBit(0); + // the running total of kept items + int resultLength = 0; + // start from the first not-removed cell, and shift left. + int removeIdx = removedIndexes.nextSetBit(keepIdx + 1); + while (removeIdx >= 0) + { + int length = removeIdx - keepIdx; + if (length > 0) + { + copy(keepIdx, resultLength, length); + resultLength += length; + } + keepIdx = removedIndexes.nextClearBit(removeIdx + 1); + if (keepIdx < 0) + keepIdx = size; + removeIdx = removedIndexes.nextSetBit(keepIdx + 1); + } + // Copy everything after the last deleted column + int length = size - keepIdx; + if (length > 0) + { + copy(keepIdx, resultLength, length); + resultLength += length; + } + + for (int i = resultLength; i < size; i++) + cells[i] = null; + + size = sortedSize = resultLength; + } + + private void copy(int src, int dst, int len) + { + // [src, src+len) and [dst, dst+len) might overlap but it's okay because we're going from left to right + assert dst <= src : "dst must not be greater than src"; + + if (dst < src) + System.arraycopy(cells, src, cells, dst, len); + } + + public boolean hasNext() + { + return iter.hasNext(); + } + + public Cell next() + { + idx++; + shouldCallNext = false; + return iter.next(); + } + + public void remove() + { + if (shouldCallNext) + throw new IllegalStateException(); + + removedIndexes.set(reversed ? size - idx - 1 : idx); + removedAnything = true; + shouldCallNext = true; + } + }; + } + private Comparator internalComparator() { return reversed ? getComparator().reverseComparator() : getComparator(); diff --git a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java index e766f65498..0f083e3746 100644 --- a/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java +++ b/src/java/org/apache/cassandra/db/AtomicBTreeColumns.java @@ -399,6 +399,11 @@ public class AtomicBTreeColumns extends ColumnFamily return false; } + public BatchRemoveIterator batchRemoveIterator() + { + throw new UnsupportedOperationException(); + } + private static final class Holder { final DeletionInfo deletionInfo; diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 25904aed4c..a8ea3863c8 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -516,6 +516,12 @@ public abstract class ColumnFamily implements Iterable, IRowCacheEntry return ByteBuffer.wrap(out.getData(), 0, out.getLength()); } + + /** + * @return an iterator where the removes are carried out once everything has been iterated + */ + public abstract BatchRemoveIterator batchRemoveIterator(); + public abstract static class Factory { /** diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index c2ee0aca4d..78537fad11 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -1247,7 +1247,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean */ public static ColumnFamily removeDeletedColumnsOnly(ColumnFamily cf, int gcBefore, SecondaryIndexManager.Updater indexer) { - Iterator iter = cf.iterator(); + BatchRemoveIterator iter = cf.batchRemoveIterator(); DeletionInfo.InOrderTester tester = cf.inOrderDeletionTester(); boolean hasDroppedColumns = !cf.metadata.getDroppedColumns().isEmpty(); while (iter.hasNext()) @@ -1263,7 +1263,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean indexer.remove(c); } } - + iter.commit(); return cf; } @@ -1281,10 +1281,11 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean if (cf == null || cf.metadata.getDroppedColumns().isEmpty()) return; - Iterator iter = cf.iterator(); + BatchRemoveIterator iter = cf.batchRemoveIterator(); while (iter.hasNext()) if (isDroppedColumn(iter.next(), metadata)) iter.remove(); + iter.commit(); } /** diff --git a/src/java/org/apache/cassandra/utils/BatchRemoveIterator.java b/src/java/org/apache/cassandra/utils/BatchRemoveIterator.java new file mode 100644 index 0000000000..437742604c --- /dev/null +++ b/src/java/org/apache/cassandra/utils/BatchRemoveIterator.java @@ -0,0 +1,32 @@ +/* + * 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.utils; + +import java.util.Iterator; + +/** + * Iterator that allows us to more efficiently remove many items + */ +public interface BatchRemoveIterator extends Iterator +{ + /** + * Commits the remove operations in this batch iterator. After this no more + * deletes can be made. Any further calls to remove() or commit() will throw IllegalStateException. + */ + void commit(); +} diff --git a/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java b/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java index 46fe812871..0fdabe9d73 100644 --- a/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java +++ b/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java @@ -27,17 +27,20 @@ import org.junit.Test; import static org.junit.Assert.*; +import com.google.common.collect.Sets; + import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.config.KSMetaData; import org.apache.cassandra.config.Schema; -import org.apache.cassandra.exceptions.ConfigurationException; -import org.apache.cassandra.locator.SimpleStrategy; -import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.db.composites.*; import org.apache.cassandra.db.filter.ColumnSlice; import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.exceptions.ConfigurationException; +import org.apache.cassandra.locator.SimpleStrategy; +import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.SearchIterator; +import org.apache.cassandra.utils.BatchRemoveIterator; public class ArrayBackedSortedColumnsTest { @@ -326,4 +329,98 @@ public class ArrayBackedSortedColumnsTest iter.remove(); assertTrue(!iter.hasNext()); } + + @Test(expected = IllegalStateException.class) + public void testBatchRemoveTwice() + { + CellNameType type = new SimpleDenseCellNameType(Int32Type.instance); + ColumnFamily map = ArrayBackedSortedColumns.factory.create(metadata(), false); + map.addColumn(new BufferCell(type.makeCellName(1))); + map.addColumn(new BufferCell(type.makeCellName(2))); + + BatchRemoveIterator batchIter = map.batchRemoveIterator(); + batchIter.next(); + batchIter.remove(); + batchIter.remove(); + } + + @Test(expected = IllegalStateException.class) + public void testBatchCommitTwice() + { + CellNameType type = new SimpleDenseCellNameType(Int32Type.instance); + ColumnFamily map = ArrayBackedSortedColumns.factory.create(metadata(), false); + map.addColumn(new BufferCell(type.makeCellName(1))); + map.addColumn(new BufferCell(type.makeCellName(2))); + + BatchRemoveIterator batchIter = map.batchRemoveIterator(); + batchIter.next(); + batchIter.remove(); + batchIter.commit(); + batchIter.commit(); + } + + @Test + public void testBatchRemove() + { + testBatchRemoveInternal(false); + testBatchRemoveInternal(true); + } + + public void testBatchRemoveInternal(boolean reversed) + { + CellNameType type = new SimpleDenseCellNameType(Int32Type.instance); + ColumnFamily map = ArrayBackedSortedColumns.factory.create(metadata(), reversed); + int[] values = new int[]{ 1, 2, 3, 5 }; + + for (int i = 0; i < values.length; ++i) + map.addColumn(new BufferCell(type.makeCellName(values[reversed ? values.length - 1 - i : i]))); + + BatchRemoveIterator batchIter = map.batchRemoveIterator(); + batchIter.next(); + batchIter.remove(); + batchIter.next(); + batchIter.remove(); + + assertEquals("1st column before commit", 1, map.iterator().next().name().toByteBuffer().getInt(0)); + + batchIter.commit(); + + assertEquals("1st column after commit", 3, map.iterator().next().name().toByteBuffer().getInt(0)); + } + + @Test + public void testBatchRemoveCopy() + { + // Test delete some random columns and check the result + CellNameType type = new SimpleDenseCellNameType(Int32Type.instance); + ColumnFamily map = ArrayBackedSortedColumns.factory.create(metadata(), false); + int n = 127; + int[] values = new int[n]; + for (int i = 0; i < n; i++) + values[i] = i; + Set toRemove = Sets.newHashSet(3, 12, 13, 15, 58, 103, 112); + + for (int value : values) + map.addColumn(new BufferCell(type.makeCellName(value))); + + BatchRemoveIterator batchIter = map.batchRemoveIterator(); + while (batchIter.hasNext()) + if (toRemove.contains(batchIter.next().name().toByteBuffer().getInt(0))) + batchIter.remove(); + + batchIter.commit(); + + int expected = 0; + while (toRemove.contains(expected)) + expected++; + + for (Cell column : map) + { + assertEquals(expected, column.name().toByteBuffer().getInt(0)); + expected++; + while (toRemove.contains(expected)) + expected++; + } + assertEquals(expected, n); + } }