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

Merge branch 'cassandra-2.1' into trunk 

Conflicts:
	src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
	test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java
This commit is contained in:
Aleksey Yeschenko 2015-01-22 03:22:34 +03:00
parent 39108df7c8 06cd494c14
commit 2b4cd105e8
7 changed files with 238 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

96
src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
View File

@ -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<Cell> batchRemoveIterator()
{
maybeSortCells();
return new BatchRemoveIterator<Cell>()
{
private final Iterator<Cell> 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<Composite> internalComparator()
{
return reversed ? getComparator().reverseComparator() : getComparator();

5
src/java/org/apache/cassandra/db/AtomicBTreeColumns.java
View File

@ -399,6 +399,11 @@ public class AtomicBTreeColumns extends ColumnFamily
return false;
}
public BatchRemoveIterator<Cell> batchRemoveIterator()
{
throw new UnsupportedOperationException();
}
private static final class Holder
{
final DeletionInfo deletionInfo;

6
src/java/org/apache/cassandra/db/ColumnFamily.java
View File

@ -516,6 +516,12 @@ public abstract class ColumnFamily implements Iterable<Cell>, 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<Cell> batchRemoveIterator();
public abstract static class Factory <T extends ColumnFamily>
{
/**

7
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -1247,7 +1247,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
*/
public static ColumnFamily removeDeletedColumnsOnly(ColumnFamily cf, int gcBefore, SecondaryIndexManager.Updater indexer)
{
Iterator<Cell> iter = cf.iterator();
BatchRemoveIterator<Cell> 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<Cell> iter = cf.iterator();
BatchRemoveIterator<Cell> iter = cf.batchRemoveIterator();
while (iter.hasNext())
if (isDroppedColumn(iter.next(), metadata))
iter.remove();
iter.commit();
}
/**

32
src/java/org/apache/cassandra/utils/BatchRemoveIterator.java Normal file
View File

@ -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<T> extends Iterator<T>
{
/**
* 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();
}

103
test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java
View File

@ -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<Cell> 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<Cell> 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<Cell> 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<Integer> toRemove = Sets.newHashSet(3, 12, 13, 15, 58, 103, 112);
for (int value : values)
map.addColumn(new BufferCell(type.makeCellName(value)));
BatchRemoveIterator<Cell> 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);
}
}