Updated Branches: refs/heads/cassandra-1.2 c5f4cdd2b -> 7746225dc refs/heads/trunk b59ca1fc8 -> 3a3f612f8
Always record row-level tombstones in index component patch by jbellis; reviewed by jasobrown for CASSANDRA-5487 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/798470e0 Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/798470e0 Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/798470e0 Branch: refs/heads/trunk Commit: 798470e051af794b605cce28031b33b589cfc6d8 Parents: 813a937 Author: Jonathan Ellis <jbel...@apache.org> Authored: Fri Apr 19 13:20:46 2013 -0500 Committer: Jonathan Ellis <jbel...@apache.org> Committed: Fri Apr 19 13:20:46 2013 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + build.xml | 2 +- src/java/org/apache/cassandra/db/ColumnIndex.java | 5 +- .../org/apache/cassandra/db/RowIndexEntry.java | 85 +++++----- .../db/columniterator/IndexedSliceReader.java | 93 +++++----- .../db/columniterator/SSTableNamesIterator.java | 130 +++++---------- .../db/columniterator/SimpleSliceReader.java | 15 +- .../apache/cassandra/io/sstable/Descriptor.java | 7 +- .../apache/cassandra/io/sstable/SSTableReader.java | 2 +- .../cassandra/utils/AlwaysPresentFilter.java | 4 + .../org/apache/cassandra/utils/FilterFactory.java | 2 +- .../apache/cassandra/db/RangeTombstoneTest.java | 4 - 12 files changed, 158 insertions(+), 192 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 7fc93f4..7dbb62a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 1.2.5 + * Fix promoted row-level tombstone writing (CASSANDRA-5486) * Include fatal errors in trace events (CASSANDRA-5447) * Ensure that PerRowSecondaryIndex is notified of row-level deletes (CASSANDRA-5445) http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/build.xml ---------------------------------------------------------------------- diff --git a/build.xml b/build.xml index 3491431..2ddd43d 100644 --- a/build.xml +++ b/build.xml @@ -519,7 +519,7 @@ </artifact:pom> </target> - <target name="maven-ant-tasks-retrieve-build" depends="maven-declare-dependencies"> + <target name="maven-ant-tasks-retrieve-build" depends="maven-declare-dependencies" unless="without.maven"> <artifact:dependencies pomRefId="build-deps-pom" filesetId="build-dependency-jars" sourcesFilesetId="build-dependency-sources" http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/db/ColumnIndex.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnIndex.java b/src/java/org/apache/cassandra/db/ColumnIndex.java index bcd0eef..e2ac3e4 100644 --- a/src/java/org/apache/cassandra/db/ColumnIndex.java +++ b/src/java/org/apache/cassandra/db/ColumnIndex.java @@ -33,7 +33,7 @@ public class ColumnIndex public final List<IndexHelper.IndexInfo> columnsIndex; public final IFilter bloomFilter; - private static final ColumnIndex EMPTY = new ColumnIndex(Collections.<IndexHelper.IndexInfo>emptyList(), new AlwaysPresentFilter()); + private static final ColumnIndex EMPTY = new ColumnIndex(Collections.<IndexHelper.IndexInfo>emptyList(), AlwaysPresentFilter.instance); private ColumnIndex(int estimatedColumnCount) { @@ -42,6 +42,9 @@ public class ColumnIndex private ColumnIndex(List<IndexHelper.IndexInfo> columnsIndex, IFilter bloomFilter) { + assert columnsIndex != null; + assert bloomFilter != null; + this.columnsIndex = columnsIndex; this.bloomFilter = bloomFilter; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/db/RowIndexEntry.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/RowIndexEntry.java b/src/java/org/apache/cassandra/db/RowIndexEntry.java index a831498..a60bf6d 100644 --- a/src/java/org/apache/cassandra/db/RowIndexEntry.java +++ b/src/java/org/apache/cassandra/db/RowIndexEntry.java @@ -28,6 +28,7 @@ import org.apache.cassandra.cache.IMeasurableMemory; import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.IndexHelper; import org.apache.cassandra.io.util.FileUtils; +import org.apache.cassandra.utils.AlwaysPresentFilter; import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.FilterFactory; import org.apache.cassandra.utils.ObjectSizes; @@ -45,35 +46,42 @@ public class RowIndexEntry implements IMeasurableMemory public int serializedSize() { - return TypeSizes.NATIVE.sizeof(position); + return TypeSizes.NATIVE.sizeof(position) + promotedSize(); } - public static RowIndexEntry create(long position, DeletionInfo deletionInfo, ColumnIndex index) + public int promotedSize() { - if (index != null && index.columnsIndex != null && index.columnsIndex.size() > 1) - return new IndexedEntry(position, deletionInfo, index.columnsIndex, index.bloomFilter); - else - return new RowIndexEntry(position); + return 0; } - public boolean isIndexed() + // TODO only store DeletionTime + public static RowIndexEntry create(long position, DeletionInfo deletionInfo, ColumnIndex index) { - return !columnsIndex().isEmpty(); + assert deletionInfo != null; + assert index != null; + + if (index.columnsIndex.size() > 1 || deletionInfo.getTopLevelDeletion() != DeletionTime.LIVE) + return new IndexedEntry(position, + deletionInfo, + index.columnsIndex.isEmpty() ? Collections.<IndexHelper.IndexInfo>emptyList() : index.columnsIndex, + index.columnsIndex.isEmpty() ? AlwaysPresentFilter.instance : index.bloomFilter); + else + return new RowIndexEntry(position); } public DeletionInfo deletionInfo() { - throw new UnsupportedOperationException(); + return DeletionInfo.LIVE; } public List<IndexHelper.IndexInfo> columnsIndex() { - return Collections.<IndexHelper.IndexInfo>emptyList(); + return Collections.emptyList(); } public IFilter bloomFilter() { - throw new UnsupportedOperationException(); + return AlwaysPresentFilter.instance; } public long memorySize() @@ -87,14 +95,15 @@ public class RowIndexEntry implements IMeasurableMemory public void serialize(RowIndexEntry rie, DataOutput dos) throws IOException { dos.writeLong(rie.position); - if (rie.isIndexed()) + if (!rie.columnsIndex().isEmpty() || rie.deletionInfo().getTopLevelDeletion() != DeletionTime.LIVE) { - dos.writeInt(((IndexedEntry)rie).serializedSize()); + dos.writeInt(rie.promotedSize()); DeletionInfo.serializer().serializeForSSTable(rie.deletionInfo(), dos); dos.writeInt(rie.columnsIndex().size()); for (IndexHelper.IndexInfo info : rie.columnsIndex()) info.serialize(dos); - FilterFactory.serialize(rie.bloomFilter(), dos); + if (!rie.columnsIndex().isEmpty()) + FilterFactory.serialize(rie.bloomFilter(), dos); } else { @@ -102,38 +111,24 @@ public class RowIndexEntry implements IMeasurableMemory } } - public RowIndexEntry deserializePositionOnly(DataInput dis, Descriptor.Version version) throws IOException - { - long position = dis.readLong(); - if (version.hasPromotedIndexes) - { - int size = dis.readInt(); - if (size > 0) - FileUtils.skipBytesFully(dis, size); - } - return new RowIndexEntry(position); - } - public RowIndexEntry deserialize(DataInput dis, Descriptor.Version version) throws IOException { long position = dis.readLong(); - if (version.hasPromotedIndexes) + if (!version.hasPromotedIndexes) + return new RowIndexEntry(position); + + int size = dis.readInt(); + if (size > 0) { - int size = dis.readInt(); - if (size > 0) - { - DeletionInfo delInfo = DeletionInfo.serializer().deserializeFromSSTable(dis, version); - int entries = dis.readInt(); - List<IndexHelper.IndexInfo> columnsIndex = new ArrayList<IndexHelper.IndexInfo>(entries); - for (int i = 0; i < entries; i++) - columnsIndex.add(IndexHelper.IndexInfo.deserialize(dis)); - IFilter bf = FilterFactory.deserialize(dis, version.filterType, false); - return new IndexedEntry(position, delInfo, columnsIndex, bf); - } - else - { - return new RowIndexEntry(position); - } + DeletionInfo delInfo = DeletionInfo.serializer().deserializeFromSSTable(dis, version); + int entries = dis.readInt(); + List<IndexHelper.IndexInfo> columnsIndex = new ArrayList<IndexHelper.IndexInfo>(entries); + for (int i = 0; i < entries; i++) + columnsIndex.add(IndexHelper.IndexInfo.deserialize(dis)); + IFilter bf = entries == 0 + ? AlwaysPresentFilter.instance + : FilterFactory.deserialize(dis, version.filterType, false); + return new IndexedEntry(position, delInfo, columnsIndex, bf); } else { @@ -171,7 +166,7 @@ public class RowIndexEntry implements IMeasurableMemory { super(position); assert deletionInfo != null; - assert columnsIndex != null && columnsIndex.size() > 1; + assert columnsIndex != null; this.deletionInfo = deletionInfo; this.columnsIndex = columnsIndex; this.bloomFilter = bloomFilter; @@ -196,7 +191,7 @@ public class RowIndexEntry implements IMeasurableMemory } @Override - public int serializedSize() + public int promotedSize() { TypeSizes typeSizes = TypeSizes.NATIVE; long size = DeletionTime.serializer.serializedSize(deletionInfo.getTopLevelDeletion(), typeSizes); @@ -204,7 +199,7 @@ public class RowIndexEntry implements IMeasurableMemory for (IndexHelper.IndexInfo info : columnsIndex) size += info.serializedSize(typeSizes); - size += FilterFactory.serializedSize(bloomFilter); + size += bloomFilter instanceof AlwaysPresentFilter ? 0 : FilterFactory.serializedSize(bloomFilter); assert size <= Integer.MAX_VALUE; return (int)size; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java index 7289ab0..61ae00e 100644 --- a/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java +++ b/src/java/org/apache/cassandra/db/columniterator/IndexedSliceReader.java @@ -65,7 +65,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA * finish (reverse start) elements. i.e. forward: [a,b],[d,e],[g,h] reverse: [h,g],[e,d],[b,a]. This reader also * assumes that validation has been performed in terms of intervals (no overlapping intervals). */ - public IndexedSliceReader(SSTableReader sstable, RowIndexEntry indexEntry, FileDataInput input, ColumnSlice[] slices, boolean reversed) + public IndexedSliceReader(SSTableReader sstable, RowIndexEntry rowEntry, FileDataInput input, ColumnSlice[] slices, boolean reversed) { this.sstable = sstable; this.originalInput = input; @@ -76,34 +76,53 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA try { Descriptor.Version version = sstable.descriptor.version; + emptyColumnFamily = ColumnFamily.create(sstable.metadata); + + if (version.hasPromotedRowTombstones && !rowEntry.columnsIndex().isEmpty()) + { + // skip the row header entirely + indexes = rowEntry.columnsIndex(); + emptyColumnFamily.delete(rowEntry.deletionInfo()); + fetcher = new IndexedBlockFetcher(rowEntry.position); + return; + } + + // skip up to bloom filter where things get a bit more interesting + if (input == null) + { + file = sstable.getFileDataInput(rowEntry.position); + } + else + { + file = input; + file.seek(rowEntry.position); + } + this.sstable.decodeKey(ByteBufferUtil.readWithShortLength(file)); + SSTableReader.readRowSize(file, this.sstable.descriptor); + + // read the row header up to and including the row-level tombstones if (version.hasPromotedIndexes) { - this.indexes = indexEntry.columnsIndex(); - if (indexes.isEmpty()) - { - setToRowStart(sstable, indexEntry, input); - this.emptyColumnFamily = ColumnFamily.create(sstable.metadata); - emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version)); - fetcher = new SimpleBlockFetcher(); - } - else - { - this.emptyColumnFamily = ColumnFamily.create(sstable.metadata); - emptyColumnFamily.delete(indexEntry.deletionInfo()); - fetcher = new IndexedBlockFetcher(indexEntry.position); - } + indexes = rowEntry.columnsIndex(); + emptyColumnFamily.delete(rowEntry.deletionInfo()); + } + else + { + IndexHelper.skipSSTableBloomFilter(input, version); + indexes = IndexHelper.deserializeIndex(file); + } + emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version)); + + if (indexes.isEmpty()) + { + fetcher = new SimpleBlockFetcher(); } else { - setToRowStart(sstable, indexEntry, input); - IndexHelper.skipSSTableBloomFilter(file, version); - this.indexes = IndexHelper.deserializeIndex(file); - this.emptyColumnFamily = ColumnFamily.create(sstable.metadata); - emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version)); - fetcher = indexes.isEmpty() - ? new SimpleBlockFetcher() - : new IndexedBlockFetcher(file.getFilePointer() + 4); // We still have the column count to - // skip to get the basePosition + // index offsets changed to be based against the row key start in 1.2 + fetcher = version.hasPromotedIndexes + ? new IndexedBlockFetcher(rowEntry.position) + : new IndexedBlockFetcher(file.getFilePointer() + 4); // +4 to skip the int column count } } catch (IOException e) @@ -113,24 +132,6 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA } } - /** - * Sets the seek position to the start of the row for column scanning. - */ - private void setToRowStart(SSTableReader reader, RowIndexEntry indexEntry, FileDataInput input) throws IOException - { - if (input == null) - { - this.file = sstable.getFileDataInput(indexEntry.position); - } - else - { - this.file = input; - input.seek(indexEntry.position); - } - sstable.decodeKey(ByteBufferUtil.readWithShortLength(file)); - SSTableReader.readRowSize(file, sstable.descriptor); - } - public ColumnFamily getColumnFamily() { return emptyColumnFamily; @@ -197,8 +198,6 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA return reversed ? slices[currentSliceIdx].start : slices[currentSliceIdx].finish; } - protected abstract boolean setNextSlice(); - protected abstract boolean fetchMoreData(); protected boolean isColumnBeforeSliceStart(OnDiskAtom column) @@ -248,7 +247,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA setNextSlice(); } - protected boolean setNextSlice() + private boolean setNextSlice() { while (++currentSliceIdx < slices.length) { @@ -350,7 +349,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA /* seek to the correct offset to the data, and calculate the data size */ long positionToSeek = basePosition + currentIndex.offset; - // With new promoted indexes, our first seek in the data file will happen at that point. + // With 1.2 promoted indexes, our first seek in the data file will happen at this point if (file == null) file = originalInput == null ? sstable.getFileDataInput(positionToSeek) : originalInput; @@ -464,7 +463,7 @@ class IndexedSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskA } } - protected boolean setNextSlice() + private boolean setNextSlice() { if (reversed) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java index da4631d..326447f 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java +++ b/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java @@ -19,25 +19,22 @@ package org.apache.cassandra.db.columniterator; import java.io.IOException; import java.nio.ByteBuffer; -import java.util.*; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.SortedSet; import org.apache.cassandra.config.CFMetaData; -import org.apache.cassandra.db.ColumnFamily; -import org.apache.cassandra.db.ColumnFamilySerializer; -import org.apache.cassandra.db.DecoratedKey; -import org.apache.cassandra.db.DeletionInfo; -import org.apache.cassandra.db.IColumn; -import org.apache.cassandra.db.RowIndexEntry; -import org.apache.cassandra.db.OnDiskAtom; +import org.apache.cassandra.db.*; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.io.sstable.CorruptSSTableException; +import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.io.sstable.IndexHelper; import org.apache.cassandra.io.sstable.SSTableReader; import org.apache.cassandra.io.util.FileDataInput; import org.apache.cassandra.io.util.FileMark; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.utils.ByteBufferUtil; -import org.apache.cassandra.utils.IFilter; public class SSTableNamesIterator extends SimpleAbstractColumnIterator implements ISSTableColumnIterator { @@ -55,13 +52,13 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement this.columns = columns; this.key = key; - RowIndexEntry indexEntry = sstable.getPosition(key, SSTableReader.Operator.EQ); - if (indexEntry == null) + RowIndexEntry rowEntry = sstable.getPosition(key, SSTableReader.Operator.EQ); + if (rowEntry == null) return; try { - read(sstable, null, indexEntry); + read(sstable, null, rowEntry); } catch (IOException e) { @@ -75,7 +72,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement } } - public SSTableNamesIterator(SSTableReader sstable, FileDataInput file, DecoratedKey key, SortedSet<ByteBuffer> columns, RowIndexEntry indexEntry) + public SSTableNamesIterator(SSTableReader sstable, FileDataInput file, DecoratedKey key, SortedSet<ByteBuffer> columns, RowIndexEntry rowEntry) { assert columns != null; this.sstable = sstable; @@ -84,7 +81,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement try { - read(sstable, file, indexEntry); + read(sstable, file, rowEntry); } catch (IOException e) { @@ -104,101 +101,66 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement return sstable; } - private void read(SSTableReader sstable, FileDataInput file, RowIndexEntry indexEntry) - throws IOException + private void read(SSTableReader sstable, FileDataInput file, RowIndexEntry rowEntry) + throws IOException { - IFilter bf; List<IndexHelper.IndexInfo> indexList; - // If the entry is not indexed or the index is not promoted, read from the row start - if (!indexEntry.isIndexed()) + Descriptor.Version version = sstable.descriptor.version; + cf = ColumnFamily.create(sstable.metadata); + List<OnDiskAtom> result = new ArrayList<OnDiskAtom>(columns.size()); + + if (version.hasPromotedRowTombstones && !rowEntry.columnsIndex().isEmpty()) { - if (file == null) - file = createFileDataInput(indexEntry.position); - else - file.seek(indexEntry.position); + // skip the row header entirely + cf.delete(rowEntry.deletionInfo()); - DecoratedKey keyInDisk = SSTableReader.decodeKey(sstable.partitioner, - sstable.descriptor, - ByteBufferUtil.readWithShortLength(file)); - assert keyInDisk.equals(key) : String.format("%s != %s in %s", keyInDisk, key, file.getPath()); - SSTableReader.readRowSize(file, sstable.descriptor); + readIndexedColumns(sstable.metadata, file, columns, rowEntry.columnsIndex(), rowEntry.position, result); + iter = result.iterator(); + return; } - if (sstable.descriptor.version.hasPromotedIndexes) - { - bf = indexEntry.isIndexed() ? indexEntry.bloomFilter() : null; - indexList = indexEntry.columnsIndex(); - } + if (file == null) + file = createFileDataInput(rowEntry.position); else - { - assert file != null; - bf = IndexHelper.defreezeBloomFilter(file, sstable.descriptor.version.filterType); - indexList = IndexHelper.deserializeIndex(file); - } + file.seek(rowEntry.position); + + DecoratedKey keyInDisk = SSTableReader.decodeKey(sstable.partitioner, + sstable.descriptor, + ByteBufferUtil.readWithShortLength(file)); + assert keyInDisk.equals(key) : String.format("%s != %s in %s", keyInDisk, key, file.getPath()); + SSTableReader.readRowSize(file, sstable.descriptor); - if (!indexEntry.isIndexed()) + if (sstable.descriptor.version.hasPromotedIndexes) { - // we can stop early if bloom filter says none of the columns actually exist -- but, - // we can't stop before initializing the cf above, in case there's a relevant tombstone - ColumnFamilySerializer serializer = ColumnFamily.serializer; - try - { - cf = ColumnFamily.create(sstable.metadata); - cf.delete(DeletionInfo.serializer().deserializeFromSSTable(file, sstable.descriptor.version)); - } - catch (Exception e) - { - throw new IOException(serializer + " failed to deserialize " + sstable.getColumnFamilyName() + " with " + sstable.metadata + " from " + file, e); - } + indexList = rowEntry.columnsIndex(); + cf.delete(rowEntry.deletionInfo()); } else { - cf = ColumnFamily.create(sstable.metadata); - cf.delete(indexEntry.deletionInfo()); + indexList = IndexHelper.deserializeIndex(file); } - List<OnDiskAtom> result = new ArrayList<OnDiskAtom>(); - List<ByteBuffer> filteredColumnNames = new ArrayList<ByteBuffer>(columns.size()); - for (ByteBuffer name : columns) - { - if (bf == null || bf.isPresent(name)) - { - filteredColumnNames.add(name); - } - } - if (filteredColumnNames.isEmpty()) - return; + cf.delete(DeletionInfo.serializer().deserializeFromSSTable(file, sstable.descriptor.version)); if (indexList.isEmpty()) { - readSimpleColumns(file, columns, filteredColumnNames, result); + readSimpleColumns(file, columns, result); } else { - long basePosition; - if (sstable.descriptor.version.hasPromotedIndexes) - { - basePosition = indexEntry.position; - } - else - { - assert file != null; - file.readInt(); // column count - basePosition = file.getFilePointer(); - } - readIndexedColumns(sstable.metadata, file, columns, filteredColumnNames, indexList, basePosition, result); + long basePosition = version.hasPromotedIndexes ? rowEntry.position : file.getFilePointer() + 4; + readIndexedColumns(sstable.metadata, file, columns, indexList, basePosition, result); } // create an iterator view of the columns we read iter = result.iterator(); } - private void readSimpleColumns(FileDataInput file, SortedSet<ByteBuffer> columnNames, List<ByteBuffer> filteredColumnNames, List<OnDiskAtom> result) throws IOException + private void readSimpleColumns(FileDataInput file, SortedSet<ByteBuffer> columnNames, List<OnDiskAtom> result) throws IOException { OnDiskAtom.Serializer atomSerializer = cf.getOnDiskSerializer(); int columns = file.readInt(); - int n = 0; for (int i = 0; i < columns; i++) { OnDiskAtom column = atomSerializer.deserializeFromSSTable(file, sstable.descriptor.version); @@ -207,7 +169,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement if (columnNames.contains(column.name())) { result.add(column); - if (n++ > filteredColumnNames.size()) + if (result.size() >= columnNames.size()) break; } } @@ -221,17 +183,16 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement private void readIndexedColumns(CFMetaData metadata, FileDataInput file, SortedSet<ByteBuffer> columnNames, - List<ByteBuffer> filteredColumnNames, List<IndexHelper.IndexInfo> indexList, long basePosition, List<OnDiskAtom> result) - throws IOException + throws IOException { /* get the various column ranges we have to read */ AbstractType<?> comparator = metadata.comparator; List<IndexHelper.IndexInfo> ranges = new ArrayList<IndexHelper.IndexInfo>(); int lastIndexIdx = -1; - for (ByteBuffer name : filteredColumnNames) + for (ByteBuffer name : columnNames) { int index = IndexHelper.indexFor(name, indexList, comparator, false, lastIndexIdx); if (index < 0 || index == indexList.size()) @@ -251,7 +212,7 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement { long positionToSeek = basePosition + indexInfo.offset; - // With new promoted indexes, our first seek in the data file will happen at that point. + // With 1.2 promoted indexes, our first seek in the data file will happen at this point if (file == null) file = createFileDataInput(positionToSeek); @@ -262,7 +223,6 @@ public class SSTableNamesIterator extends SimpleAbstractColumnIterator implement while (file.bytesPastMark(mark) < indexInfo.width) { OnDiskAtom column = atomSerializer.deserializeFromSSTable(file, sstable.descriptor.version); - // we check vs the original Set, not the filtered List, for efficiency if (!(column instanceof IColumn) || columnNames.contains(column.name())) result.add(column); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java index b30d360..0cf9af6 100644 --- a/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java +++ b/src/java/org/apache/cassandra/db/columniterator/SimpleSliceReader.java @@ -49,7 +49,7 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt private FileMark mark; private final OnDiskAtom.Serializer atomSerializer; - public SimpleSliceReader(SSTableReader sstable, RowIndexEntry indexEntry, FileDataInput input, ByteBuffer finishColumn) + public SimpleSliceReader(SSTableReader sstable, RowIndexEntry rowEntry, FileDataInput input, ByteBuffer finishColumn) { this.sstable = sstable; this.finishColumn = finishColumn; @@ -58,13 +58,13 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt { if (input == null) { - this.file = sstable.getFileDataInput(indexEntry.position); + this.file = sstable.getFileDataInput(rowEntry.position); this.needsClosing = true; } else { this.file = input; - input.seek(indexEntry.position); + input.seek(rowEntry.position); this.needsClosing = false; } @@ -72,14 +72,19 @@ class SimpleSliceReader extends AbstractIterator<OnDiskAtom> implements OnDiskAt ByteBufferUtil.skipShortLength(file); SSTableReader.readRowSize(file, sstable.descriptor); + emptyColumnFamily = ColumnFamily.create(sstable.metadata); + Descriptor.Version version = sstable.descriptor.version; - if (!version.hasPromotedIndexes) + if (version.hasPromotedIndexes) + { + emptyColumnFamily.delete(rowEntry.deletionInfo()); + } + else { IndexHelper.skipSSTableBloomFilter(file, version); IndexHelper.skipIndex(file); } - emptyColumnFamily = ColumnFamily.create(sstable.metadata); emptyColumnFamily.delete(DeletionInfo.serializer().deserializeFromSSTable(file, version)); atomSerializer = emptyColumnFamily.getOnDiskSerializer(); columns = file.readInt(); http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/io/sstable/Descriptor.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/Descriptor.java b/src/java/org/apache/cassandra/io/sstable/Descriptor.java index f21a0d5..7b916cb 100644 --- a/src/java/org/apache/cassandra/io/sstable/Descriptor.java +++ b/src/java/org/apache/cassandra/io/sstable/Descriptor.java @@ -47,7 +47,7 @@ public class Descriptor public static class Version { // This needs to be at the begining for initialization sake - public static final String current_version = "ib"; + public static final String current_version = "ic"; public static final Version LEGACY = new Version("a"); // "pre-history" // b (0.7.0): added version to sstable filenames @@ -62,10 +62,11 @@ public class Descriptor // hd (1.0.10): includes row tombstones in maxtimestamp // he (1.1.3): includes ancestors generation in metadata component // hf (1.1.6): marker that replay position corresponds to 1.1.5+ millis-based id (see CASSANDRA-4782) - // ia (1.2.0): column indexes are promoted to the index file + // ia (1.2.0): column indexes are promoted to the index file. (this means index offsets are now against the start of the row key, rather than the start of columns data, since the former allows us to skip the row header) // records estimated histogram of deletion times in tombstones // bloom filter (keys and columns) upgraded to Murmur3 // ib (1.2.1): tracks min client timestamp in metadata component + // ic (1.2.6): always promotes row-level tombstones into index file; previously this was unreliable public static final Version CURRENT = new Version(current_version); @@ -83,6 +84,7 @@ public class Descriptor public final boolean hasPartitioner; public final boolean tracksTombstones; public final boolean hasPromotedIndexes; + public final boolean hasPromotedRowTombstones; public final FilterFactory.Type filterType; public final boolean hasAncestors; public final boolean hasBloomFilterSizeInHeader; @@ -102,6 +104,7 @@ public class Descriptor metadataIncludesModernReplayPosition = version.compareTo("hf") >= 0; tracksTombstones = version.compareTo("ia") >= 0; hasPromotedIndexes = version.compareTo("ia") >= 0; + hasPromotedRowTombstones = version.compareTo("ic") >= 0; isLatestVersion = version.compareTo(current_version) == 0; if (version.compareTo("f") < 0) filterType = FilterFactory.Type.SHA; http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/io/sstable/SSTableReader.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java index 21a8673..61f505d 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableReader.java @@ -323,7 +323,7 @@ public class SSTableReader extends SSTable { if (!components.contains(Component.FILTER)) { - bf = new AlwaysPresentFilter(); + bf = AlwaysPresentFilter.instance; return; } http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/utils/AlwaysPresentFilter.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/AlwaysPresentFilter.java b/src/java/org/apache/cassandra/utils/AlwaysPresentFilter.java index 67ac111..39b3d5d 100644 --- a/src/java/org/apache/cassandra/utils/AlwaysPresentFilter.java +++ b/src/java/org/apache/cassandra/utils/AlwaysPresentFilter.java @@ -26,6 +26,10 @@ import java.nio.ByteBuffer; public class AlwaysPresentFilter implements IFilter { + public static final AlwaysPresentFilter instance = new AlwaysPresentFilter(); + + private AlwaysPresentFilter() { } + public boolean isPresent(ByteBuffer key) { return true; http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/src/java/org/apache/cassandra/utils/FilterFactory.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/utils/FilterFactory.java b/src/java/org/apache/cassandra/utils/FilterFactory.java index 88c8973..1b9027d 100644 --- a/src/java/org/apache/cassandra/utils/FilterFactory.java +++ b/src/java/org/apache/cassandra/utils/FilterFactory.java @@ -131,7 +131,7 @@ public class FilterFactory { assert maxFalsePosProbability <= 1.0 : "Invalid probability"; if (maxFalsePosProbability == 1.0) - return new AlwaysPresentFilter(); + return AlwaysPresentFilter.instance; int bucketsPerElement = BloomCalculations.maxBucketsPerElement(numElements); BloomCalculations.BloomSpecification spec = BloomCalculations.computeBloomSpec(bucketsPerElement, maxFalsePosProbability); return createFilter(spec.K, numElements, spec.bucketsPerElement, type, offheap); http://git-wip-us.apache.org/repos/asf/cassandra/blob/798470e0/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java index 1bc846b..c531461 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneTest.java @@ -164,10 +164,6 @@ public class RangeTombstoneTest extends SchemaLoader return ByteBufferUtil.bytes(i); } - private static void insertData(ColumnFamilyStore cfs, String key) throws Exception - { - } - private static void add(RowMutation rm, int value, long timestamp) { rm.add(new QueryPath(CFNAME, null, b(value)), b(value), timestamp);
