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

Stop reading from sstables once we know we have the most recent columns 

patch by Daniel Lundin and jbellis for CASSANDRA-2498

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1159942 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2011-08-21 04:50:55 +00:00
parent 604c0e87dc
commit 791a78f92c
11 changed files with 345 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -35,6 +35,9 @@
* fix missing logging for some exceptions (CASSANDRA-2061)
* refactor and optimize ColumnFamilyStore.files(...) and Descriptor.fromFilename(String)
and few other places responsible for work with SSTable files (CASSANDRA-3040)
* Stop reading from sstables once we know we have the most recent columns,
for query-by-name requests (CASSANDRA-2498)
0.8.5
* fix NPE when encryption_options is unspecified (CASSANDRA-3007)

231
src/java/org/apache/cassandra/db/CollationController.java Normal file
View File

@ -0,0 +1,231 @@
/**
* 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 org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.columniterator.IColumnIterator;
import org.apache.cassandra.db.columniterator.SimpleAbstractColumnIterator;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.marshal.CounterColumnType;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.CloseableIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.Iterables;
public class CollationController
{
private static Logger logger = LoggerFactory.getLogger(CollationController.class);
private final DataTracker.View dataview;
private final ISortedColumns.Factory factory;
private final QueryFilter filter;
private final int gcBefore;
private final CFMetaData metadata;
private int sstablesIterated = 0;
public CollationController(DataTracker.View dataview, ISortedColumns.Factory factory, QueryFilter filter, CFMetaData metadata, int gcBefore)
{
this.dataview = dataview;
this.factory = factory;
this.filter = filter;
this.gcBefore = gcBefore;
this.metadata = metadata;
}
public ColumnFamily getTopLevelColumns()
{
return filter.filter instanceof NamesQueryFilter && metadata.getDefaultValidator() != CounterColumnType.instance
? collectTimeOrderedData()
: collectAllData();
}
/**
* Collects data in order of recency, using the sstable maxtimestamp data.
* Once we have data for all requests columns that is newer than the newest remaining maxtimestamp,
* we stop.
*/
private ColumnFamily collectTimeOrderedData()
{
logger.debug("collectTimeOrderedData");
List<IColumnIterator> iterators = new ArrayList<IColumnIterator>();
final ColumnFamily container = ColumnFamily.create(metadata, factory, filter.filter.isReversed());
try
{
for (Memtable memtable : Iterables.concat(dataview.memtablesPendingFlush, Collections.singleton(dataview.memtable)))
{
IColumnIterator iter = filter.getMemtableColumnIterator(memtable, metadata.comparator);
if (iter != null)
{
iterators.add(iter);
container.delete(iter.getColumnFamily());
while (iter.hasNext())
container.addColumn(iter.next());
}
}
// avoid changing the filter columns of the original filter
// (reduceNameFilter removes columns that are known to be irrelevant)
TreeSet<ByteBuffer> filterColumns = new TreeSet<ByteBuffer>(metadata.comparator);
filterColumns.addAll(((NamesQueryFilter) filter.filter).columns);
QueryFilter reducedFilter = new QueryFilter(filter.key, filter.path, new NamesQueryFilter(filterColumns));
// read sorted sstables
for (SSTableReader sstable : dataview.sstables)
{
long currentMaxTs = sstable.getMaxTimestamp();
reduceNameFilter(reducedFilter, container, currentMaxTs);
if (((NamesQueryFilter) reducedFilter.filter).columns.isEmpty())
break;
IColumnIterator iter = reducedFilter.getSSTableColumnIterator(sstable);
iterators.add(iter);
if (iter.getColumnFamily() != null)
{
container.delete(iter.getColumnFamily());
sstablesIterated++;
while (iter.hasNext())
container.addColumn(iter.next());
}
}
}
finally
{
for (IColumnIterator iter : iterators)
FileUtils.closeQuietly(iter);
}
// we need to distinguish between "there is no data at all for this row" (BF will let us rebuild that efficiently)
// and "there used to be data, but it's gone now" (we should cache the empty CF so we don't need to rebuild that slower)
if (iterators.isEmpty())
return null;
// do a final collate. toCollate is boilerplate required to provide a CloseableIterator
CloseableIterator<IColumn> toCollate = new SimpleAbstractColumnIterator()
{
final Iterator<IColumn> iter = container.iterator();
protected IColumn computeNext()
{
return iter.hasNext() ? iter.next() : endOfData();
}
public ColumnFamily getColumnFamily()
{
return container;
}
public DecoratedKey getKey()
{
return filter.key;
}
};
ColumnFamily returnCF = container.cloneMeShallow();
filter.collateColumns(returnCF, Collections.singletonList(toCollate), metadata.comparator, gcBefore);
// Caller is responsible for final removeDeletedCF. This is important for cacheRow to work correctly:
return returnCF;
}
/**
* remove columns from @param filter where we already have data in @param returnCF newer than @param sstableTimestamp
*/
private void reduceNameFilter(QueryFilter filter, ColumnFamily returnCF, long sstableTimestamp)
{
AbstractColumnContainer container = filter.path.superColumnName != null
? (SuperColumn) returnCF.getColumn(filter.path.superColumnName)
: returnCF;
if (container == null)
return;
for (Iterator<ByteBuffer> iterator = ((NamesQueryFilter) filter.filter).columns.iterator(); iterator.hasNext(); )
{
ByteBuffer filterColumn = iterator.next();
IColumn column = container.getColumn(filterColumn);
if (column != null && column.minTimestamp() > sstableTimestamp)
iterator.remove();
}
}
/**
* Collects data the brute-force way: gets an iterator for the filter in question
* from every memtable and sstable, then merges them together.
*/
private ColumnFamily collectAllData()
{
logger.debug("collectAllData");
List<IColumnIterator> iterators = new ArrayList<IColumnIterator>();
ColumnFamily returnCF = ColumnFamily.create(metadata, factory, filter.filter.isReversed());
try
{
for (Memtable memtable : Iterables.concat(dataview.memtablesPendingFlush, Collections.singleton(dataview.memtable)))
{
IColumnIterator iter = filter.getMemtableColumnIterator(memtable, metadata.comparator);
if (iter != null)
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
}
/* add the SSTables on disk */
for (SSTableReader sstable : dataview.sstables)
{
IColumnIterator iter = filter.getSSTableColumnIterator(sstable);
iterators.add(iter);
if (iter.getColumnFamily() != null)
{
returnCF.delete(iter.getColumnFamily());
sstablesIterated++;
}
}
}
finally
{
for (IColumnIterator iter : iterators)
FileUtils.closeQuietly(iter);
}
// we need to distinguish between "there is no data at all for this row" (BF will let us rebuild that efficiently)
// and "there used to be data, but it's gone now" (we should cache the empty CF so we don't need to rebuild that slower)
if (iterators.isEmpty())
return null;
filter.collateColumns(returnCF, iterators, metadata.comparator, gcBefore);
// Caller is responsible for final removeDeletedCF. This is important for cacheRow to work correctly:
return returnCF;
}
public int getSstablesIterated()
{
return sstablesIterated;
}
}

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

@ -104,6 +104,11 @@ public class Column implements IColumn
return timestamp;
}
public long minTimestamp()
{
return timestamp;
}
public boolean isMarkedForDelete()
{
return false;

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

@ -1278,73 +1278,17 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean
private ColumnFamily getTopLevelColumns(QueryFilter filter, int gcBefore, ISortedColumns.Factory factory)
{
// we are querying top-level columns, do a merging fetch with indexes.
List<IColumnIterator> iterators = new ArrayList<IColumnIterator>();
final ColumnFamily returnCF = ColumnFamily.create(metadata, factory, filter.filter.isReversed());
DataTracker.View currentView = markCurrentViewReferenced();
try
{
IColumnIterator iter;
int sstablesToIterate = 0;
/* add the current memtable */
iter = filter.getMemtableColumnIterator(currentView.memtable, getComparator());
if (iter != null)
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
/* add the memtables being flushed */
for (Memtable memtable : currentView.memtablesPendingFlush)
{
iter = filter.getMemtableColumnIterator(memtable, getComparator());
if (iter != null)
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
}
/* add the SSTables on disk */
for (SSTableReader sstable : currentView.sstables)
{
iter = filter.getSSTableColumnIterator(sstable);
if (iter.getColumnFamily() != null)
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
sstablesToIterate++;
}
}
recentSSTablesPerRead.add(sstablesToIterate);
sstablesPerRead.add(sstablesToIterate);
// we need to distinguish between "there is no data at all for this row" (BF will let us rebuild that efficiently)
// and "there used to be data, but it's gone now" (we should cache the empty CF so we don't need to rebuild that slower)
if (iterators.size() == 0)
return null;
filter.collateColumns(returnCF, iterators, getComparator(), gcBefore);
// Caller is responsible for final removeDeletedCF. This is important for cacheRow to work correctly:
return returnCF;
CollationController controller = new CollationController(currentView, factory, filter, metadata, gcBefore);
ColumnFamily columns = controller.getTopLevelColumns();
recentSSTablesPerRead.add(controller.getSstablesIterated());
sstablesPerRead.add(controller.getSstablesIterated());
return columns;
}
finally
{
/* close all cursors */
for (IColumnIterator ci : iterators)
{
try
{
ci.close();
}
catch (Throwable th)
{
logger.error("error closing " + ci, th);
}
}
SSTableReader.releaseReferences(currentView.sstables);
}
}

47
src/java/org/apache/cassandra/db/DataTracker.java
View File

@ -27,6 +27,8 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.ImmutableSortedSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -34,6 +36,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.cache.AutoSavingCache;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.sstable.SSTable;
import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.utils.Pair;
@ -66,7 +69,7 @@ public class DataTracker
return view.get().memtablesPendingFlush;
}
public Set<SSTableReader> getSSTables()
public List<SSTableReader> getSSTables()
{
return view.get().sstables;
}
@ -242,7 +245,7 @@ public class DataTracker
{
view.set(new View(new Memtable(cfstore),
Collections.<Memtable>emptySet(),
Collections.<SSTableReader>emptySet(),
Collections.<SSTableReader>emptyList(),
Collections.<SSTableReader>emptySet()));
}
@ -461,10 +464,15 @@ public class DataTracker
{
public final Memtable memtable;
public final Set<Memtable> memtablesPendingFlush;
public final Set<SSTableReader> sstables;
public final Set<SSTableReader> compacting;
// We can't use a SortedSet here because "the ordering maintained by a sorted set (whether or not an
// explicit comparator is provided) must be <i>consistent with equals</i>." In particular,
// ImmutableSortedSet will ignore any objects that compare equally with an existing Set member.
// Obviously, dropping sstables whose max column timestamp happens to be equal to another's
// is not acceptable for us. So, we use a List instead.
public final List<SSTableReader> sstables;
View(Memtable memtable, Set<Memtable> pendingFlush, Set<SSTableReader> sstables, Set<SSTableReader> compacting)
View(Memtable memtable, Set<Memtable> pendingFlush, List<SSTableReader> sstables, Set<SSTableReader> compacting)
{
this.memtable = memtable;
this.memtablesPendingFlush = pendingFlush;
@ -486,15 +494,14 @@ public class DataTracker
public View replaceFlushed(Memtable flushedMemtable, SSTableReader newSSTable)
{
Set<Memtable> newPending = ImmutableSet.copyOf(Sets.difference(memtablesPendingFlush, Collections.singleton(flushedMemtable)));
Set<SSTableReader> newSSTables = ImmutableSet.<SSTableReader>builder().addAll(sstables).add(newSSTable).build();
return new View(memtable, newPending, newSSTables, compacting);
List<SSTableReader> newSSTables = newSSTables(newSSTable);
return new View(memtable, newPending, Collections.unmodifiableList(newSSTables), compacting);
}
public View replace(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements)
{
Sets.SetView<SSTableReader> remaining = Sets.difference(sstables, ImmutableSet.copyOf(oldSSTables));
Set<SSTableReader> newSSTables = ImmutableSet.<SSTableReader>builder().addAll(remaining).addAll(replacements).build();
return new View(memtable, memtablesPendingFlush, newSSTables, compacting);
List<SSTableReader> newSSTables = newSSTables(oldSSTables, replacements);
return new View(memtable, memtablesPendingFlush, Collections.unmodifiableList(newSSTables), compacting);
}
public View markCompacting(Collection<SSTableReader> tomark)
@ -508,5 +515,27 @@ public class DataTracker
Set<SSTableReader> compactingNew = ImmutableSet.copyOf(Sets.difference(compacting, ImmutableSet.copyOf(tounmark)));
return new View(memtable, memtablesPendingFlush, sstables, compactingNew);
}
private List<SSTableReader> newSSTables(SSTableReader newSSTable)
{
// not performance-sensitive, don't obsess over doing a selection merge here
return newSSTables(Collections.<SSTableReader>emptyList(), Collections.singletonList(newSSTable));
}
private List<SSTableReader> newSSTables(Collection<SSTableReader> oldSSTables, Iterable<SSTableReader> replacements)
{
ImmutableSet<SSTableReader> oldSet = ImmutableSet.copyOf(oldSSTables);
int newSSTablesSize = sstables.size() - oldSSTables.size() + Iterables.size(replacements);
List<SSTableReader> newSSTables = new ArrayList<SSTableReader>(newSSTablesSize);
for (SSTableReader sstable : sstables)
{
if (!oldSet.contains(sstable))
newSSTables.add(sstable);
}
Iterables.addAll(newSSTables, replacements);
assert newSSTables.size() == newSSTablesSize;
Collections.sort(newSSTables, SSTable.maxTimestampComparator);
return newSSTables;
}
}
}

8
src/java/org/apache/cassandra/db/IColumn.java
View File

@ -73,7 +73,13 @@ public interface IColumn
/**
* For a standard column, this is the same as timestamp().
* For a super column, this is max the column value timestamp of the sub columns.
* For a super column, this is the max column timestamp of the sub columns.
*/
public long maxTimestamp();
/**
* For a standard column, this is the same as timestamp().
* For a super column, this is the min column timestamp of the sub columns.
*/
public long minTimestamp();
}

11
src/java/org/apache/cassandra/db/SuperColumn.java
View File

@ -132,11 +132,18 @@ public class SuperColumn extends AbstractColumnContainer implements IColumn
public long maxTimestamp()
{
long maxTimestamp = Long.MIN_VALUE;
long maxTimestamp = getMarkedForDeleteAt();
for (IColumn subColumn : getSubColumns())
maxTimestamp = Math.max(maxTimestamp, subColumn.maxTimestamp());
return maxTimestamp;
}
return Math.max(maxTimestamp, getMarkedForDeleteAt());
public long minTimestamp()
{
long minTimestamp = getMarkedForDeleteAt();
for (IColumn subColumn : getSubColumns())
minTimestamp = Math.min(minTimestamp, subColumn.maxTimestamp());
return minTimestamp;
}
public long mostRecentLiveChangeAt()

2
src/java/org/apache/cassandra/db/columniterator/SSTableSliceIterator.java
View File

@ -104,7 +104,7 @@ public class SSTableSliceIterator implements IColumnIterator
public boolean hasNext()
{
return reader.hasNext();
return reader != null && reader.hasNext();
}
public IColumn next()

14
src/java/org/apache/cassandra/io/sstable/SSTable.java
View File

@ -22,9 +22,7 @@ package org.apache.cassandra.io.sstable;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import java.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -60,6 +58,16 @@ public abstract class SSTable
public static final String TEMPFILE_MARKER = "tmp";
public static final Comparator<SSTableReader> maxTimestampComparator = new Comparator<SSTableReader>()
{
public int compare(SSTableReader o1, SSTableReader o2)
{
long ts1 = o1.getMaxTimestamp();
long ts2 = o2.getMaxTimestamp();
return (ts1 > ts2 ? -1 : (ts1 == ts2 ? 0 : 1));
}
};
public final Descriptor descriptor;
protected final Set<Component> components;
public final CFMetaData metadata;

34
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -53,6 +53,7 @@ import static org.apache.cassandra.Util.column;
import static org.apache.cassandra.Util.getBytes;
import static org.junit.Assert.assertNull;
import org.junit.Assert;
import org.junit.Test;
public class ColumnFamilyStoreTest extends CleanupHelper
@ -68,22 +69,49 @@ public class ColumnFamilyStoreTest extends CleanupHelper
random.nextBytes(bytes2);
}
@Test
// create two sstables, and verify that we only deserialize data from the most recent one
public void testTimeSortedQuery() throws IOException, ExecutionException, InterruptedException
{
Table table = Table.open("Keyspace1");
ColumnFamilyStore cfs = table.getColumnFamilyStore("Standard1");
cfs.truncate().get();
RowMutation rm;
rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("key1"));
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column1")), ByteBufferUtil.bytes("asdf"), 0);
rm.apply();
cfs.forceBlockingFlush();
rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("key1"));
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column1")), ByteBufferUtil.bytes("asdf"), 1);
rm.apply();
cfs.forceBlockingFlush();
cfs.getRecentSSTablesPerReadHistogram(); // resets counts
cfs.getColumnFamily(QueryFilter.getNamesFilter(Util.dk("key1"), new QueryPath("Standard1", null), ByteBufferUtil.bytes("Column1")));
assertEquals(1, cfs.getRecentSSTablesPerReadHistogram()[0]);
}
@Test
public void testGetColumnWithWrongBF() throws IOException, ExecutionException, InterruptedException
{
Table table = Table.open("Keyspace1");
ColumnFamilyStore cfs = table.getColumnFamilyStore("Standard1");
cfs.truncate().get();
List<IMutation> rms = new LinkedList<IMutation>();
RowMutation rm;
rm = new RowMutation("Keyspace1", ByteBufferUtil.bytes("key1"));
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column1")), ByteBufferUtil.bytes("asdf"), 0);
rm.add(new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column2")), ByteBufferUtil.bytes("asdf"), 0);
rms.add(rm);
ColumnFamilyStore store = Util.writeColumnFamily(rms);
Util.writeColumnFamily(rms);
Table table = Table.open("Keyspace1");
List<SSTableReader> ssTables = table.getAllSSTables();
assertEquals(1, ssTables.size());
ssTables.get(0).forceFilterFailures();
ColumnFamily cf = store.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk("key2"), new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column1"))));
ColumnFamily cf = cfs.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk("key2"), new QueryPath("Standard1", null, ByteBufferUtil.bytes("Column1"))));
assertNull(cf);
}

4
test/unit/org/apache/cassandra/db/compaction/CompactionsTest.java
View File

@ -185,6 +185,7 @@ public class CompactionsTest extends CleanupHelper
store.forceBlockingFlush();
}
Collection<SSTableReader> toCompact = store.getSSTables();
assert toCompact.size() == 2;
// Reinserting the same keys. We will compact only the previous sstable, but we need those new ones
// to make sure we use EchoedRow, otherwise it won't be used because purge can be done.
@ -200,12 +201,15 @@ public class CompactionsTest extends CleanupHelper
for (SSTableReader sstable : store.getSSTables())
if (!toCompact.contains(sstable))
tmpSSTable = sstable;
assert tmpSSTable != null;
// Force compaction on first sstables. Since each row is in only one sstable, we will be using EchoedRow.
Util.compact(store, toCompact, false);
assertEquals(2, store.getSSTables().size());
// Now, we remove the sstable that was just created to force the use of EchoedRow (so that it doesn't hide the problem)
store.markCompacted(Collections.singleton(tmpSSTable));
assertEquals(1, store.getSSTables().size());
// Now assert we do have the 4 keys
assertEquals(4, Util.getRangeSlice(store).size());