initialize column iteration lazily so filter will be ready when it's first used patch by slebresne and Piotr KoÅaczkowski; reviewed by jbellis for CASSANDRA-4816
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/6f31aba0 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/6f31aba0 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/6f31aba0 Branch: refs/heads/cassandra-1.1 Commit: 6f31aba0e1183a9ad3977bd8e0ed97b42bc7d1d5 Parents: 5d5207b Author: Jonathan Ellis <jbel...@apache.org> Authored: Wed Oct 24 14:41:22 2012 -0500 Committer: Jonathan Ellis <jbel...@apache.org> Committed: Wed Oct 24 14:41:29 2012 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../apache/cassandra/db/RowIteratorFactory.java | 38 +++++++--- .../db/columniterator/IColumnIteratorFactory.java | 6 ++ .../db/columniterator/LazyColumnIterator.java | 62 +++++++++++++++ .../cassandra/io/sstable/SSTableScanner.java | 16 +++- 5 files changed, 109 insertions(+), 14 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f31aba0/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 95feb9b..0857259 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 1.1.7 + * fix get_paged_slice to wrap to next row correctly (CASSANDRA-4816) * fix indexing empty column values (CASSANDRA-4832) * allow JdbcDate to compose null Date objects (CASSANDRA-4830) * fix possible stackoverflow when compacting 1000s of sstables http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f31aba0/src/java/org/apache/cassandra/db/RowIteratorFactory.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/RowIteratorFactory.java b/src/java/org/apache/cassandra/db/RowIteratorFactory.java index b5c57ff..5a53c4a 100644 --- a/src/java/org/apache/cassandra/db/RowIteratorFactory.java +++ b/src/java/org/apache/cassandra/db/RowIteratorFactory.java @@ -17,12 +17,11 @@ */ package org.apache.cassandra.db; -import java.io.Closeable; import java.util.*; -import com.google.common.collect.AbstractIterator; - import org.apache.cassandra.db.columniterator.IColumnIterator; +import org.apache.cassandra.db.columniterator.IColumnIteratorFactory; +import org.apache.cassandra.db.columniterator.LazyColumnIterator; import org.apache.cassandra.db.filter.QueryFilter; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.sstable.SSTableScanner; @@ -123,7 +122,7 @@ public class RowIteratorFactory /** * Get a ColumnIterator for a specific key in the memtable. */ - private static class ConvertToColumnIterator extends AbstractIterator<IColumnIterator> implements CloseableIterator<IColumnIterator> + private static class ConvertToColumnIterator implements CloseableIterator<IColumnIterator> { private final QueryFilter filter; private final Iterator<Map.Entry<DecoratedKey, ColumnFamily>> iter; @@ -134,14 +133,33 @@ public class RowIteratorFactory this.iter = iter; } - public IColumnIterator computeNext() + public boolean hasNext() { - if (iter.hasNext()) + return iter.hasNext(); + } + + /* + * Note that when doing get_paged_slice, we reset the start of the queryFilter after we've fetched the + * first row. This means that this iterator should not use in any way the filter to fetch a row before + * we call next(). Which prevents us for using guava AbstractIterator. + * This is obviously rather fragile and we should consider refactoring that code, but such refactor will go + * deep into the storage engine code so this will have to do until then. + */ + public IColumnIterator next() + { + final Map.Entry<DecoratedKey, ColumnFamily> entry = iter.next(); + return new LazyColumnIterator(entry.getKey(), new IColumnIteratorFactory() { - Map.Entry<DecoratedKey, ColumnFamily> entry = iter.next(); - return filter.getMemtableColumnIterator(entry.getValue(), entry.getKey()); - } - return endOfData(); + public IColumnIterator create() + { + return filter.getMemtableColumnIterator(entry.getValue(), entry.getKey()); + } + }); + } + + public void remove() + { + throw new UnsupportedOperationException(); } public void close() http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f31aba0/src/java/org/apache/cassandra/db/columniterator/IColumnIteratorFactory.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/columniterator/IColumnIteratorFactory.java b/src/java/org/apache/cassandra/db/columniterator/IColumnIteratorFactory.java new file mode 100644 index 0000000..c9ce857 --- /dev/null +++ b/src/java/org/apache/cassandra/db/columniterator/IColumnIteratorFactory.java @@ -0,0 +1,6 @@ +package org.apache.cassandra.db.columniterator; + +public interface IColumnIteratorFactory +{ + IColumnIterator create(); +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f31aba0/src/java/org/apache/cassandra/db/columniterator/LazyColumnIterator.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/columniterator/LazyColumnIterator.java b/src/java/org/apache/cassandra/db/columniterator/LazyColumnIterator.java new file mode 100644 index 0000000..486836d --- /dev/null +++ b/src/java/org/apache/cassandra/db/columniterator/LazyColumnIterator.java @@ -0,0 +1,62 @@ +package org.apache.cassandra.db.columniterator; + +import com.google.common.collect.AbstractIterator; +import org.apache.cassandra.db.ColumnFamily; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.IColumn; + +import java.io.IOException; + + +/* + * The goal of this encapsulating IColumnIterator is to delay the use of + * the filter until columns are actually queried. + * The reason for that is get_paged_slice because it change the start of + * the filter after having seen the first row, and so we must not use the + * filter before the row data is actually queried. However, mergeIterator + * needs to "fetch" a row in advance. But all it needs is the key and so + * this IColumnIterator make sure getKey() can be called without triggering + * the use of the filter itself. + */ +public class LazyColumnIterator extends AbstractIterator<IColumn> implements IColumnIterator +{ + private final DecoratedKey key; + private final IColumnIteratorFactory subIteratorFactory; + + private IColumnIterator subIterator; + + public LazyColumnIterator(DecoratedKey key, IColumnIteratorFactory subIteratorFactory) + { + this.key = key; + this.subIteratorFactory = subIteratorFactory; + } + + private IColumnIterator getSubIterator() + { + if (subIterator == null) + subIterator = subIteratorFactory.create(); + return subIterator; + } + + protected IColumn computeNext() + { + getSubIterator(); + return subIterator.hasNext() ? subIterator.next() : endOfData(); + } + + public ColumnFamily getColumnFamily() + { + return getSubIterator().getColumnFamily(); + } + + public DecoratedKey getKey() + { + return key; + } + + public void close() throws IOException + { + if (subIterator != null) + subIterator.close(); + } +} http://git-wip-us.apache.org/repos/asf/cassandra/blob/6f31aba0/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java index 26ed908..d3bc0b3 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableScanner.java @@ -24,6 +24,8 @@ import java.io.IOException; import java.util.Arrays; import java.util.Iterator; +import org.apache.cassandra.db.columniterator.IColumnIteratorFactory; +import org.apache.cassandra.db.columniterator.LazyColumnIterator; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -175,7 +177,7 @@ public class SSTableScanner implements ICompactionScanner file.seek(finishedAt); assert !file.isEOF(); - DecoratedKey<?> key = SSTableReader.decodeKey(sstable.partitioner, + final DecoratedKey<?> key = SSTableReader.decodeKey(sstable.partitioner, sstable.descriptor, ByteBufferUtil.readWithShortLength(file)); long dataSize = SSTableReader.readRowSize(file, sstable.descriptor); @@ -189,7 +191,13 @@ public class SSTableScanner implements ICompactionScanner } else { - return row = filter.getSSTableColumnIterator(sstable, file, key); + return row = new LazyColumnIterator(key, new IColumnIteratorFactory() + { + public IColumnIterator create() + { + return filter.getSSTableColumnIterator(sstable, file, key); + } + }); } } catch (IOException e) @@ -210,7 +218,7 @@ public class SSTableScanner implements ICompactionScanner "finishedAt:" + finishedAt + ")"; } -} + } @Override public String toString() { @@ -220,4 +228,4 @@ public class SSTableScanner implements ICompactionScanner " exhausted=" + exhausted + ")"; } -} +} \ No newline at end of file