From f327b63db09a907206749a3c88aba38a4554e548 Mon Sep 17 00:00:00 2001 From: Branimir Lambov Date: Tue, 19 Nov 2024 12:41:41 +0200 Subject: [PATCH 1/3] Introduce SSTableSimpleScanner for compaction This removes the usage of index files during compaction and simplifies and improves the performance of compaction. patch by Branimir Lambov; reviewed by Sylvain Lebresne for CASSANDRA-20092 --- CHANGES.txt | 2 + .../apache/cassandra/cache/KeyCacheKey.java | 12 +- .../db/compaction/CompactionIterator.java | 2 +- .../io/compress/CompressionMetadata.java | 8 + .../io/sstable/SSTableIdentityIterator.java | 62 ++++++ .../io/sstable/format/SSTableReader.java | 148 ++++++++++--- .../io/sstable/format/SSTableScanner.java | 16 -- .../sstable/format/SSTableSimpleScanner.java | 195 +++++++++++++++++ .../io/sstable/format/big/BigTableReader.java | 37 ---- .../sstable/format/big/BigTableScanner.java | 21 -- .../io/sstable/format/bti/BtiTableReader.java | 23 -- .../sstable/format/bti/BtiTableScanner.java | 21 -- .../cassandra/tools/SSTablePartitions.java | 4 +- .../SSTableCorruptionDetectionTest.java | 33 +++ .../io/sstable/SSTableReaderTest.java | 198 ++++++++++++++++++ .../io/sstable/SSTableScannerTest.java | 133 ++++++++++-- 16 files changed, 748 insertions(+), 167 deletions(-) create mode 100644 src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java diff --git a/CHANGES.txt b/CHANGES.txt index 4af9fb7b63..2e5aeed0b5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,7 @@ 5.0.5 * Grant permission on keyspaces system_views and system_virtual_schema not possible (CASSANDRA-20171) + * Fix marking an SSTable as suspected and BufferPool leakage in case of a corrupted SSTable read during a compaction (CASSANDRA-20396) + * Introduce SSTableSimpleScanner for compaction (CASSANDRA-20092) * Avoid purging deletions in RowFilter when reconciliation is required (CASSANDRA-20541) * Fixed multiple single-node SAI query bugs relating to static columns (CASSANDRA-20338) * Upgrade com.datastax.cassandra:cassandra-driver-core:3.11.5 to org.apache.cassandra:cassandra-driver-core:3.12.1 (CASSANDRA-17231) diff --git a/src/java/org/apache/cassandra/cache/KeyCacheKey.java b/src/java/org/apache/cassandra/cache/KeyCacheKey.java index ac6f1f9693..1a722b05e6 100644 --- a/src/java/org/apache/cassandra/cache/KeyCacheKey.java +++ b/src/java/org/apache/cassandra/cache/KeyCacheKey.java @@ -21,7 +21,6 @@ import java.nio.ByteBuffer; import java.util.Arrays; import java.util.Objects; -import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.io.sstable.Descriptor; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; @@ -31,9 +30,7 @@ public class KeyCacheKey extends CacheKey { public final Descriptor desc; - private static final long EMPTY_SIZE = ObjectSizes.measure(new KeyCacheKey(TableMetadata.builder("ks", "tab") - .addPartitionKeyColumn("pk", UTF8Type.instance) - .build(), null, ByteBufferUtil.EMPTY_BYTE_BUFFER)); + private static final long EMPTY_SIZE = ObjectSizes.measure(new KeyCacheKey()); // keeping an array instead of a ByteBuffer lowers the overhead of the key cache working set, // without extra copies on lookup since client-provided key ByteBuffers will be array-backed already @@ -47,6 +44,13 @@ public class KeyCacheKey extends CacheKey assert this.key != null; } + private KeyCacheKey() // Only for EMPTY_SIZE + { + super(null, null); + this.desc = null; + this.key = null; + } + public String toString() { return String.format("KeyCacheKey(%s, %s)", desc, ByteBufferUtil.bytesToHex(ByteBuffer.wrap(key))); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 589cf39c77..00e3dee5af 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -290,7 +290,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte { long n = 0; for (ISSTableScanner scanner : scanners) - n += scanner.getCurrentPosition(); + n += scanner.getBytesScanned(); bytesRead = n; } diff --git a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java index 96b4ce8418..d5f5f05655 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java +++ b/src/java/org/apache/cassandra/io/compress/CompressionMetadata.java @@ -539,6 +539,14 @@ public class CompressionMetadata extends WrappedSharedCloseable { return String.format("Chunk", offset, length); } + + /** + * @return the end of the chunk in the file, including the checksum + */ + public long chunkEnd() + { + return offset + length + 4; + } } static class ChunkSerializer implements IVersionedSerializer diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index cc201b4125..072a364af3 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -46,6 +46,7 @@ public class SSTableIdentityIterator implements Comparable getPositionsForRanges(Collection> ranges) { @@ -753,29 +753,112 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, for (Range range : Range.normalize(ranges)) { assert !range.isWrapAround() || range.right.isMinimum(); - // truncate the range so it at most covers the sstable AbstractBounds bounds = Range.makeRowRange(range); - PartitionPosition leftBound = bounds.left.compareTo(first) > 0 ? bounds.left : first.getToken().minKeyBound(); - PartitionPosition rightBound = bounds.right.isMinimum() ? last.getToken().maxKeyBound() : bounds.right; - - if (leftBound.compareTo(last) > 0 || rightBound.compareTo(first) < 0) - continue; - - long left = getPosition(leftBound, Operator.GT); - long right = (rightBound.compareTo(last) > 0) - ? uncompressedLength() - : getPosition(rightBound, Operator.GT); - - if (left == right) - // empty range - continue; - - assert left < right : String.format("Range=%s openReason=%s first=%s last=%s left=%d right=%d", range, openReason, first, last, left, right); - positions.add(new PartitionPositionBounds(left, right)); + PartitionPositionBounds pb = getPositionsForBounds(bounds); + if (pb != null) + positions.add(pb); } return positions; } + /** + * Get a list of data positions in this SSTable that correspond to the given list of bounds. This method will remove + * non-covered intervals, but will not correct order or overlap in the supplied list, e.g. if bounds overlap, the + * result will be sections of the data file that repeat the same positions. + * + * @return A sorted list of [offset,end) pairs corresponding to the given boundsList in the datafile for this + * SSTable. + */ + public List getPositionsForBoundsIterator(Iterator> boundsList) + { + // use the index to determine a minimal section for each range + List positions = new ArrayList<>(); + while (boundsList.hasNext()) + { + AbstractBounds bounds = boundsList.next(); + PartitionPositionBounds pb = getPositionsForBounds(bounds); + if (pb != null) + positions.add(pb); + } + return positions; + } + + /** + * Determine the data positions in this SSTable that cover the given bounds. + * + * @return An [offset,end) pair that cover the given bounds in the datafile for this SSTable, or null if the range + * is not covered by the sstable or is empty. + */ + public PartitionPositionBounds getPositionsForBounds(AbstractBounds bounds) + { + long left = getPosition(bounds.left, bounds.inclusiveLeft() ? Operator.GE : Operator.GT); + // Note: getPosition will apply a moved start if the sstable is in MOVED_START state. + if (left < 0) // empty range + return null; + + long right = bounds.right.isMinimum() ? -1 + : getPosition(bounds.right, bounds.inclusiveRight() ? Operator.GT + : Operator.GE); + if (right < 0) // right is beyond end + right = uncompressedLength(); // this should also be correct for EARLY readers + + if (left >= right) // empty range + return null; + + return new PartitionPositionBounds(left, right); + } + + /** + * Return an [offset,end) pair that covers the whole file. This could be null if the sstable's moved start has + * made the sstable effectively empty. + */ + public PartitionPositionBounds getPositionsForFullRange() + { + if (openReason != OpenReason.MOVED_START) + return new PartitionPositionBounds(0, uncompressedLength()); + else + { + // query a full range, so that the required adjustments can be applied + PartitionPosition minToken = getPartitioner().getMinimumToken().minKeyBound(); + return getPositionsForBounds(new Range<>(minToken, minToken)); + } + } + + /** + * Calculate a total on-disk (compressed) size for the given partition positions. For uncompressed files this is + * equal to the sum of the size of the covered ranges. For compressed files this is the sum of the size of the + * chunks that contain the requested ranges and may be significantly bigger than the size of the requested ranges. + * + * @param positionBounds a list of [offset,end) pairs that specify the relevant sections of the data file; this must + * be non-overlapping and in ascending order. + */ + public long onDiskSizeForPartitionPositions(Collection positionBounds) + { + long total = 0; + if (!compression) + { + for (PartitionPositionBounds position : positionBounds) + total += position.upperPosition - position.lowerPosition; + } + else + { + final CompressionMetadata compressionMetadata = getCompressionMetadata(); + long lastEnd = 0; + for (PartitionPositionBounds position : positionBounds) + { + // The end of the chunk that contains the last required byte from the range. + long upperChunkEnd = compressionMetadata.chunkFor(position.upperPosition - 1).chunkEnd(); + // The start of the chunk that contains the first required byte from the range. + long lowerChunkStart = compressionMetadata.chunkFor(position.lowerPosition).offset; + if (lowerChunkStart < lastEnd) // if regions include the same chunk, count it only once + lowerChunkStart = lastEnd; + total += upperChunkEnd - lowerChunkStart; + lastEnd = upperChunkEnd; + } + } + return total; + } + /** * Retrieves the position while updating the key cache and the stats. * @@ -965,11 +1048,18 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, } /** - * Direct I/O SSTableScanner over the entirety of the sstable.. + * Direct I/O SSTableScanner over the entirety of the sstable. * * @return A Scanner over the full content of the SSTable. */ - public abstract ISSTableScanner getScanner(); + public ISSTableScanner getScanner() + { + PartitionPositionBounds fullRange = getPositionsForFullRange(); + if (fullRange != null) + return new SSTableSimpleScanner(this, Collections.singletonList(fullRange)); + else + return new SSTableSimpleScanner(this, Collections.emptyList()); + } /** * Direct I/O SSTableScanner over a defined collection of ranges of tokens. @@ -977,15 +1067,25 @@ public abstract class SSTableReader extends SSTable implements UnfilteredSource, * @param ranges the range of keys to cover * @return A Scanner for seeking over the rows of the SSTable. */ - public abstract ISSTableScanner getScanner(Collection> ranges); + public ISSTableScanner getScanner(Collection> ranges) + { + if (ranges != null) + return new SSTableSimpleScanner(this, getPositionsForRanges(ranges)); + else + return getScanner(); + } /** * Direct I/O SSTableScanner over an iterator of bounds. * - * @param rangeIterator the keys to cover + * @param boundsIterator the keys to cover * @return A Scanner for seeking over the rows of the SSTable. */ - public abstract ISSTableScanner getScanner(Iterator> rangeIterator); + public ISSTableScanner getScanner(Iterator> boundsIterator) + { + return new SSTableSimpleScanner(this, getPositionsForBoundsIterator(boundsIterator)); + } + /** * Create a {@link FileDataInput} for the data file of the sstable represented by this reader. This method returns diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java index 5136a06bca..28035a85da 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableScanner.java @@ -19,7 +19,6 @@ package org.apache.cassandra.io.sstable.format; import java.io.IOException; import java.util.ArrayList; -import java.util.Collection; import java.util.Iterator; import java.util.List; import java.util.Set; @@ -35,9 +34,7 @@ import org.apache.cassandra.db.rows.LazilyInitializedUnfilteredRowIterator; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.AbstractBounds.Boundary; -import org.apache.cassandra.dht.Bounds; import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.AbstractRowIndexEntry; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.ISSTableScanner; @@ -86,14 +83,6 @@ implements ISSTableScanner this.listener = listener; } - protected static List> makeBounds(SSTableReader sstable, Collection> tokenRanges) - { - List> boundsList = new ArrayList<>(tokenRanges.size()); - for (Range range : Range.normalize(tokenRanges)) - addRange(sstable, Range.makeRowRange(range), boundsList); - return boundsList; - } - protected static List> makeBounds(SSTableReader sstable, DataRange dataRange) { List> boundsList = new ArrayList<>(2); @@ -101,11 +90,6 @@ implements ISSTableScanner return boundsList; } - protected static AbstractBounds fullRange(SSTableReader sstable) - { - return new Bounds<>(sstable.getFirst(), sstable.getLast()); - } - private static void addRange(SSTableReader sstable, AbstractBounds requested, List> boundsList) { if (requested instanceof Range && ((Range) requested).isWrapAround()) diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java new file mode 100644 index 0000000000..1789aff72c --- /dev/null +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableSimpleScanner.java @@ -0,0 +1,195 @@ +/* + * 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.io.sstable.format; + +import java.util.Collection; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; + +import com.google.common.collect.ImmutableSet; + +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.io.sstable.CorruptSSTableException; +import org.apache.cassandra.io.sstable.ISSTableScanner; +import org.apache.cassandra.io.sstable.SSTableIdentityIterator; +import org.apache.cassandra.io.util.RandomAccessReader; +import org.apache.cassandra.schema.TableMetadata; + +import static org.apache.cassandra.io.sstable.format.SSTableReader.PartitionPositionBounds; + +/// Simple SSTable scanner that reads sequentially through an SSTable without using the index. +/// +/// This is a significant improvement for the performance of compaction over using the full-blown DataRange-capable +/// [SSTableScanner] and enables correct calculation of data sizes to process. +public class SSTableSimpleScanner +implements ISSTableScanner +{ + private final AtomicBoolean isClosed = new AtomicBoolean(false); + private final RandomAccessReader dfile; + private final SSTableReader sstable; + + private final Iterator rangeIterator; + + private long bytesScannedInPreviousRanges; + + private final long sizeInBytes; + private final long compressedSizeInBytes; + + private long currentEndPosition; + private long currentStartPosition; + + private SSTableIdentityIterator currentIterator; + private DecoratedKey lastKey; + + /// Create a new simple scanner over the given sstables and the given ranges of uncompressed positions. + /// Each range must start and end on a partition boundary, and, to satisfy the contract of [ISSTableScanner], the + /// ranges must be non-overlapping and in ascending order. This scanner will throw an [IllegalArgumentException] if + /// the latter is not true. + /// + /// The ranges can be constructed by [SSTableReader#getPositionsForRanges] and similar methods as done by the + /// various [SSTableReader#getScanner] variations. + public SSTableSimpleScanner(SSTableReader sstable, + Collection boundsList) + { + assert sstable != null; + + this.dfile = sstable.openDataReaderForScan(); + this.sstable = sstable; + this.sizeInBytes = boundsList.stream().mapToLong(ppb -> ppb.upperPosition - ppb.lowerPosition).sum(); + this.compressedSizeInBytes = sstable.compression ? sstable.onDiskSizeForPartitionPositions(boundsList) : sizeInBytes; + this.rangeIterator = boundsList.iterator(); + this.currentEndPosition = 0; + this.currentStartPosition = 0; + this.bytesScannedInPreviousRanges = 0; + this.currentIterator = null; + this.lastKey = null; + } + + public void close() + { + if (isClosed.compareAndSet(false, true)) + { + // ensure we report what we have actually processed + bytesScannedInPreviousRanges += dfile.getFilePointer() - currentStartPosition; + dfile.close(); + // close() may change the file pointer, update so that the difference is 0 when reported by getBytesScanned() + currentStartPosition = dfile.getFilePointer(); + } + } + + @Override + public long getLengthInBytes() + { + return sizeInBytes; + } + + + public long getCompressedLengthInBytes() + { + return compressedSizeInBytes; + } + + @Override + public long getCurrentPosition() + { + return dfile.getFilePointer(); + } + + public long getBytesScanned() + { + return bytesScannedInPreviousRanges + dfile.getFilePointer() - currentStartPosition; + } + + @Override + public Set getBackingSSTables() + { + return ImmutableSet.of(sstable); + } + + public TableMetadata metadata() + { + return sstable.metadata(); + } + + public boolean hasNext() + { + if (currentIterator != null) + { + currentIterator.close(); // Ensure that the iterator cannot be used further. No op if already closed. + + // Row iterator must be exhausted to advance to next partition + currentIterator.exhaust(); + currentIterator = null; + } + + if (dfile.getFilePointer() < currentEndPosition) + return true; + + return advanceRange(); + } + + boolean advanceRange() + { + if (!rangeIterator.hasNext()) + return false; + + bytesScannedInPreviousRanges += currentEndPosition - currentStartPosition; + + PartitionPositionBounds nextRange = rangeIterator.next(); + if (currentEndPosition > nextRange.lowerPosition) + throw new IllegalArgumentException("Ranges supplied to SSTableSimpleScanner must be non-overlapping and in ascending order."); + + currentEndPosition = nextRange.upperPosition; + currentStartPosition = nextRange.lowerPosition; + dfile.seek(currentStartPosition); + return true; + } + + public UnfilteredRowIterator next() + { + if (!hasNext()) + throw new NoSuchElementException(); + + currentIterator = SSTableIdentityIterator.create(sstable, dfile, false); + DecoratedKey currentKey = currentIterator.partitionKey(); + if (lastKey != null && lastKey.compareTo(currentKey) >= 0) + { + sstable.markSuspect(); + throw new CorruptSSTableException(new IllegalStateException(String.format("Invalid key order: current %s <= previous %s", + currentKey, + lastKey)), + sstable.getFilename()); + } + lastKey = currentKey; + return currentIterator; + } + + public void remove() + { + throw new UnsupportedOperationException(); + } + + @Override + public String toString() + { + return String.format("%s(sstable=%s)", getClass().getSimpleName(), sstable); + } +} diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java index b58dbc532e..692cadf34d 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableReader.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; -import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -46,7 +45,6 @@ import org.apache.cassandra.db.rows.Rows; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.db.rows.UnfilteredRowIteratorWithLowerBound; import org.apache.cassandra.db.rows.UnfilteredRowIterators; -import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.AbstractRowIndexEntry; @@ -157,41 +155,6 @@ public class BigTableReader extends SSTableReaderWithFilter implements IndexSumm return BigTableKeyReader.create(ifile, rowIndexEntrySerializer); } - /** - * Direct I/O SSTableScanner over an iterator of bounds. - * - * @param boundsIterator the keys to cover - * @return A Scanner for seeking over the rows of the SSTable. - */ - public ISSTableScanner getScanner(Iterator> boundsIterator) - { - return BigTableScanner.getScanner(this, boundsIterator); - } - - /** - * Direct I/O SSTableScanner over the full sstable. - * - * @return A Scanner for reading the full SSTable. - */ - public ISSTableScanner getScanner() - { - return BigTableScanner.getScanner(this); - } - - /** - * Direct I/O SSTableScanner over a defined collection of ranges of tokens. - * - * @param ranges the range of keys to cover - * @return A Scanner for seeking over the rows of the SSTable. - */ - public ISSTableScanner getScanner(Collection> ranges) - { - if (ranges != null) - return BigTableScanner.getScanner(this, ranges); - else - return getScanner(); - } - /** * Finds and returns the first key beyond a given token in this SSTable or null if no such key exists. */ diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java index 887d997846..83243529c4 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableScanner.java @@ -18,11 +18,8 @@ package org.apache.cassandra.io.sstable.format.big; import java.io.IOException; -import java.util.Collection; import java.util.Iterator; -import com.google.common.collect.Iterators; - import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.PartitionPosition; @@ -30,8 +27,6 @@ import org.apache.cassandra.db.filter.ClusteringIndexFilter; import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.dht.AbstractBounds; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.SSTable; @@ -50,12 +45,6 @@ public class BigTableScanner extends SSTableScanner> tokenRanges) - { - return getScanner(sstable, makeBounds(sstable, tokenRanges).iterator()); - } - - public static ISSTableScanner getScanner(BigTableReader sstable, Iterator> rangeIterator) - { - return new BigTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, rangeIterator, SSTableReadsListener.NOOP_LISTENER); - } - private BigTableScanner(BigTableReader sstable, ColumnFilter columns, DataRange dataRange, diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReader.java index c5571e7fbb..9a65be1137 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableReader.java @@ -22,7 +22,6 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; -import java.util.Iterator; import java.util.List; import com.google.common.annotations.VisibleForTesting; @@ -45,7 +44,6 @@ import org.apache.cassandra.dht.Range; import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.CorruptSSTableException; import org.apache.cassandra.io.sstable.Descriptor; -import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.IVerifier; import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.SSTableReadsListener; @@ -382,27 +380,6 @@ public class BtiTableReader extends SSTableReaderWithFilter return new SSTableIterator(this, dataFileInput, key, indexEntry, slices, selectedColumns, rowIndexFile); } - @Override - public ISSTableScanner getScanner() - { - return BtiTableScanner.getScanner(this); - } - - @Override - public ISSTableScanner getScanner(Collection> ranges) - { - if (ranges != null) - return BtiTableScanner.getScanner(this, ranges); - else - return getScanner(); - } - - @Override - public ISSTableScanner getScanner(Iterator> rangeIterator) - { - return BtiTableScanner.getScanner(this, rangeIterator); - } - @VisibleForTesting @Override public BtiTableReader cloneAndReplace(IFilter filter) diff --git a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableScanner.java b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableScanner.java index a9f862c68b..4507ccf7f5 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableScanner.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiTableScanner.java @@ -19,11 +19,8 @@ package org.apache.cassandra.io.sstable.format.bti; import java.io.Closeable; import java.io.IOException; -import java.util.Collection; import java.util.Iterator; -import com.google.common.collect.Iterators; - import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.PartitionPosition; @@ -31,20 +28,12 @@ import org.apache.cassandra.db.filter.ClusteringIndexFilter; import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.dht.AbstractBounds; -import org.apache.cassandra.dht.Range; -import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.sstable.SSTableReadsListener; import org.apache.cassandra.io.sstable.format.SSTableScanner; import org.apache.cassandra.io.util.FileUtils; public class BtiTableScanner extends SSTableScanner { - // Full scan of the sstables - public static BtiTableScanner getScanner(BtiTableReader sstable) - { - return getScanner(sstable, Iterators.singletonIterator(fullRange(sstable))); - } - public static BtiTableScanner getScanner(BtiTableReader sstable, ColumnFilter columns, DataRange dataRange, @@ -53,16 +42,6 @@ public class BtiTableScanner extends SSTableScanner> tokenRanges) - { - return getScanner(sstable, makeBounds(sstable, tokenRanges).iterator()); - } - - public static BtiTableScanner getScanner(BtiTableReader sstable, Iterator> rangeIterator) - { - return new BtiTableScanner(sstable, ColumnFilter.all(sstable.metadata()), null, rangeIterator, SSTableReadsListener.NOOP_LISTENER); - } - private BtiTableScanner(BtiTableReader sstable, ColumnFilter columns, DataRange dataRange, diff --git a/src/java/org/apache/cassandra/tools/SSTablePartitions.java b/src/java/org/apache/cassandra/tools/SSTablePartitions.java index 2181346271..b435994bba 100644 --- a/src/java/org/apache/cassandra/tools/SSTablePartitions.java +++ b/src/java/org/apache/cassandra/tools/SSTablePartitions.java @@ -369,13 +369,15 @@ public class SSTablePartitions { while (scanner.hasNext()) { + // hasNext() positions us on the next partition, next() has to advance to read its header. + long startOfPartition = scanner.getCurrentPosition(); try (UnfilteredRowIterator partition = scanner.next()) { ByteBuffer key = partition.partitionKey().getKey(); boolean isExcluded = excludedKeys.contains(metadata.partitionKeyType.getString(key)); PartitionStats partitionStats = new PartitionStats(key, - scanner.getCurrentPosition(), + startOfPartition, partition.partitionLevelDeletion().isLive()); // Consume the partition to populate the stats. diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java index 21ac51ee86..c631d5d099 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableCorruptionDetectionTest.java @@ -37,6 +37,7 @@ import org.apache.cassandra.cache.ChunkCache; import org.apache.cassandra.config.Config; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DataRange; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.Slices; @@ -156,6 +157,12 @@ public class SSTableCorruptionDetectionTest extends SSTableWriterTestBase bruteForceCorruptionTest(ssTableReader, sstableScanner()); } + @Test + public void testSSTableSimpleScanner() throws Throwable + { + bruteForceCorruptionTest(ssTableReader, sstableSimpleScanner()); + } + private void bruteForceCorruptionTest(SSTableReader ssTableReader, Consumer walker) throws Throwable { FileChannel fc = new File(ssTableReader.getFilename()).newReadWriteChannel(); @@ -193,6 +200,32 @@ public class SSTableCorruptionDetectionTest extends SSTableWriterTestBase } private Consumer sstableScanner() + { + return (SSTableReader sstable) -> { + try (var scanner = sstable.partitionIterator(ColumnFilter.NONE, DataRange.allData(sstable.getPartitioner()), SSTableReadsListener.NOOP_LISTENER)) + { + while (scanner.hasNext()) + { + try (UnfilteredRowIterator rowIter = scanner.next()) + { + if (rowIter.hasNext()) + { + Unfiltered unfiltered = rowIter.next(); + if (unfiltered.isRow()) + { + Row row = (Row) unfiltered; + assertEquals(2, row.clustering().size()); + // no-op read + } + } + } + + } + } + }; + } + + private Consumer sstableSimpleScanner() { return (SSTableReader sstable) -> { try (ISSTableScanner scanner = sstable.getScanner()) diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java index 9cb7ca3d14..f17301e064 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java @@ -30,8 +30,11 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import java.util.stream.Stream; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Sets; import org.junit.Assume; import org.junit.BeforeClass; @@ -89,6 +92,7 @@ import org.mockito.Mockito; import static java.lang.String.format; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; +import static org.apache.cassandra.db.ColumnFamilyStore.FlushReason.UNIT_TESTS; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -178,6 +182,200 @@ public class SSTableReaderTest } } + @Test + public void testOnDiskSizeForRanges() + { + ColumnFamilyStore store = discardSSTables(KEYSPACE1, CF_STANDARD2); + partitioner = store.getPartitioner(); + int count = 1000; + + // insert data and compact to a single sstable + for (int j = 0; j < count; j++) + { + new RowUpdateBuilder(store.metadata(), 15000, k0(j)) + .clustering("0") + .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER) + .build() + .applyUnsafe(); + } + store.forceBlockingFlush(UNIT_TESTS); + store.forceMajorCompaction(); + + SSTableReader sstable = store.getLiveSSTables().iterator().next(); + + // Non-compression-dependent checks + // Check several ways of going through the whole file + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(0), 1)), t0(count - 1))))); + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(sstable.getPartitioner().getMinimumToken(), + sstable.getPartitioner().getMinimumToken())))); + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(sstable.getPartitioner().getMinimumToken(), + sstable.getLast().getToken())))); + + // Split at exact match + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t0(347)), + new Range<>(t0(347), t0(count - 1))))); + + // Split at different prefixes pointing to the same position + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t(cut(k0(600), 2))), + new Range<>(t(cut(k0(600), 1)), t0(count - 1))))); + + // Size one row + double oneRowSize = sstable.uncompressedLength() * 1.0 / count; + System.out.println("One row size: " + oneRowSize); + + if (!sstable.compression) + { + double delta = 0.9; + + // Ranges are end-inclusive, indexes are adjusted by one here to account for that. + assertEquals((52 - 38), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t0(37), t0(51)))) / oneRowSize, + delta); + + // Try non-matching positions (inexact indexes are not adjusted for the count). + assertEquals((34 - 30), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(30), 1)), + t0(33)))) / oneRowSize, + delta); + + assertEquals((700 - 554), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t0(553), + t(cut(k0(700), 2))))) / oneRowSize, + delta); + + assertEquals((500 - 30), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(30), 1)), + t(cut(k0(500), 2))))) / oneRowSize, + delta); + + // Try a list + List> ranges = ImmutableList.of(new Range<>(t0(37), t0(51)), + new Range<>(t0(71), t(cut(k0(100), 2))), + new Range<>(t(cut(k0(230), 1)), t0(243)), + new Range<>(t(cut(k0(260), 1)), t(cut(k0(300), 2))), + new Range<>(t0(373), t0(382)), + new Range<>(t0(382), t0(385)), + new Range<>(t(cut(k0(400), 2)), t(cut(k0(400), 1))), // empty range + new Range<>(t0(563), t(cut(k0(600), 2))), // touching ranges + new Range<>(t(cut(k0(600), 1)), t0(621)) + ); + assertEquals((52 - 38 + 100 - 72 + 244 - 230 + 300 - 260 + 383 - 374 + 386 - 383 + 400 - 400 + 622 - 564), + onDiskSizeForRanges(sstable, ranges) / oneRowSize, + delta); + + // Check going through the whole file + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(0), 1)), t0(count - 1))))); + + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t0(347)), + new Range<>(t0(347), t0(count - 1))))); + + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t(cut(k0(600), 2))), + new Range<>(t(cut(k0(600), 1)), t0(count - 1))))); + } + else + { + // It's much harder to test with compression. + + // Check first three rows have the same size (they must be in the same chunk) + final long row0size = onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(0), 1)), t0(0)))); + assertEquals(row0size, onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t0(0), t0(1))))); + assertEquals(row0size, onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t0(1), t0(2))))); + + // As well as the first three rows together + assertEquals(row0size, onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(t(cut(k0(0), 1)), t0(2))))); + + // And also when we query for them in separate ranges + assertEquals(row0size, onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t0(0)), + new Range<>(t0(0), t0(1))))); + assertEquals(row0size, onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t0(0)), + new Range<>(t0(1), t0(2))))); + assertEquals(row0size, onDiskSizeForRanges(sstable, ImmutableList.of(new Range<>(t(cut(k0(0), 1)), t0(0)), + new Range<>(t0(0), t0(1)), + new Range<>(t0(1), t0(2))))); + + // Finally, check that if we query for every second row we get the total size of the file. + assertEquals(sstable.onDiskLength(), + onDiskSizeForRanges(sstable, IntStream.range(0, count) + .filter(i -> i % 2 != 0) + .mapToObj(i -> new Range<>(t0(i), t0(i + 1))) + .collect(Collectors.toList()))); + } + } + + + @Test + public void testOnDiskSizeCompressedBoundaries() + { + ColumnFamilyStore store = discardSSTables(KEYSPACE1, CF_COMPRESSED); + partitioner = store.getPartitioner(); + int count = 100000; + // Use a longish string to let a key align with a chunk boundary + ByteBuffer dataBuf = ByteBufferUtil.bytes(String.format("%43d", 123)); + + // insert data and compact to a single sstable + for (int j = 0; j < count; j++) + { + new RowUpdateBuilder(store.metadata(), 15000, k0(j)) + .clustering("0") + .add("val", dataBuf) + .build() + .applyUnsafe(); + } + store.forceBlockingFlush(UNIT_TESTS); + store.forceMajorCompaction(); + + SSTableReader sstable = store.getLiveSSTables().iterator().next(); + + int chunkLength = sstable.getCompressionMetadata().chunkLength(); + System.out.println("Chunk length: " + chunkLength); + int[] alignedKeys = IntStream.range(0, count).filter(i -> (sstable.getPosition(dk0(i), SSTableReader.Operator.EQ) & (chunkLength - 1)) == 0).toArray(); + assertTrue("Test needs an aligned key, try changing the length of dataBuf", alignedKeys.length > 1); + for (int k : alignedKeys) + assertEquals("Coverage must not include chunk starting at end position", + sstable.getCompressionMetadata().chunkFor(sstable.getPosition(dk0(k), SSTableReader.Operator.EQ)).offset, + onDiskSizeForRanges(sstable, Collections.singleton(new Range<>(partitioner.getMinimumToken(), t0(k - 1))))); // inclusive end + } + + + long onDiskSizeForRanges(SSTableReader sstable, Collection> ranges) + { + return sstable.onDiskSizeForPartitionPositions(sstable.getPositionsForRanges(ranges)); + } + + private Token t(String key) + { + return partitioner.getToken(ByteBufferUtil.bytes(key)); + } + + private String k0(int k) + { + return String.format("%08d", k); + } + + private Token t0(int k) + { + return t(k0(k)); + } + + private DecoratedKey dk0(int k) + { + return partitioner.decorateKey(ByteBufferUtil.bytes(k0(k))); + } + + private String cut(String s, int n) + { + return s.substring(0, s.length() - n); + } + + @Test public void testSpannedIndexPositions() throws IOException { diff --git a/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java b/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java index 17b8a6cbb2..73195b0617 100644 --- a/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/SSTableScannerTest.java @@ -21,10 +21,13 @@ import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import java.util.function.Consumer; +import java.util.function.Function; import com.google.common.collect.Iterables; +import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; @@ -40,6 +43,7 @@ import org.apache.cassandra.db.Slices; import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter; import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.dht.AbstractBounds; import org.apache.cassandra.dht.ByteOrderedPartitioner; @@ -49,6 +53,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; +import org.hamcrest.Matchers; import static org.apache.cassandra.dht.AbstractBounds.isEmpty; import static org.junit.Assert.assertEquals; @@ -180,6 +185,12 @@ public class SSTableScannerTest } private static void assertScanMatches(SSTableReader sstable, int scanStart, int scanEnd, int ... boundaries) + { + assertScanMatchesUsingScanner(sstable, scanStart, scanEnd, boundaries); + assertScanMatchesUsingSimple(sstable, scanStart, scanEnd, boundaries); + } + + private static void assertScanMatchesUsingScanner(SSTableReader sstable, int scanStart, int scanEnd, int ... boundaries) { assert boundaries.length % 2 == 0; for (DataRange range : dataRanges(sstable.metadata(), scanStart, scanEnd)) @@ -200,6 +211,28 @@ public class SSTableScannerTest } } + private static void assertScanMatchesUsingSimple(SSTableReader sstable, int scanStart, int scanEnd, int ... boundaries) + { + assert boundaries.length % 2 == 0; + for (DataRange range : dataRanges(sstable.metadata(), scanStart, scanEnd)) + { + if (range.isWrapAround() && !range.keyRange().right.isMinimum()) // getScanner on AbstractBounds does not handle wraparounds + continue; + + try(UnfilteredPartitionIterator scanner = sstable.getScanner(Collections.singleton(range.keyRange()).iterator())) + { + for (int b = 0; b < boundaries.length; b += 2) + for (int i = boundaries[b]; i <= boundaries[b + 1]; i++) + assertEquals(toKey(i), new String(scanner.next().partitionKey().getKey().array())); + assertFalse(scanner.hasNext()); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + } + private static void assertScanEmpty(SSTableReader sstable, int scanStart, int scanEnd) { assertScanMatches(sstable, scanStart, scanEnd); @@ -547,7 +580,30 @@ public class SSTableScannerTest assertScanContainsRanges(scanner, 205, 205); } - private static void testRequestNextRowIteratorWithoutConsumingPrevious(Consumer consumer) + private static void testRequestNextRowIteratorWithoutConsumingPrevious(Function makeScanner, + Consumer requestNext, + String messagePattern) + { + final SSTableReader sstable = prepareSmallSSTable(); + + try (UnfilteredPartitionIterator scanner = makeScanner.apply(sstable); + UnfilteredRowIterator currentRowIterator = scanner.next()) + { + assertTrue(currentRowIterator.hasNext()); + try + { + requestNext.accept(scanner); + currentRowIterator.next(); + fail("Should have thrown IllegalStateException"); + } + catch (IllegalStateException e) + { + Assert.assertThat(e.getMessage(), Matchers.matchesPattern(messagePattern)); + } + } + } + + private static SSTableReader prepareSmallSSTable() { Keyspace keyspace = Keyspace.open(KEYSPACE); ColumnFamilyStore store = keyspace.getColumnFamilyStore(TABLE); @@ -557,38 +613,77 @@ public class SSTableScannerTest store.disableAutoCompaction(); insertRowWithKey(store.metadata(), 0); + insertRowWithKey(store.metadata(), 3); store.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS); assertEquals(1, store.getLiveSSTables().size()); SSTableReader sstable = store.getLiveSSTables().iterator().next(); + return sstable; + } - try (ISSTableScanner scanner = sstable.getScanner(); - UnfilteredRowIterator currentRowIterator = scanner.next()) - { - assertTrue(currentRowIterator.hasNext()); - try - { - consumer.accept(scanner); - fail("Should have thrown IllegalStateException"); - } - catch (IllegalStateException e) - { - assertEquals("The UnfilteredRowIterator returned by the last call to next() was initialized: " + - "it must be closed before calling hasNext() or next() again.", - e.getMessage()); - } - } + @Test + public void testSimpleHasNextRowIteratorWithoutConsumingPrevious() + { + testRequestNextRowIteratorWithoutConsumingPrevious(SSTableReader::getScanner, + UnfilteredPartitionIterator::hasNext, + "Iterator used after closing."); + } + + @Test + public void testSimpleNextRowIteratorWithoutConsumingPrevious() + { + testRequestNextRowIteratorWithoutConsumingPrevious(SSTableReader::getScanner, + UnfilteredPartitionIterator::next, + "Iterator used after closing."); } @Test public void testHasNextRowIteratorWithoutConsumingPrevious() { - testRequestNextRowIteratorWithoutConsumingPrevious(ISSTableScanner::hasNext); + testRequestNextRowIteratorWithoutConsumingPrevious(r -> r.partitionIterator(ColumnFilter.NONE, DataRange.allData(r.getPartitioner()), SSTableReadsListener.NOOP_LISTENER), + UnfilteredPartitionIterator::hasNext, + ".*UnfilteredRowIterator.*must be closed.*"); } @Test public void testNextRowIteratorWithoutConsumingPrevious() { - testRequestNextRowIteratorWithoutConsumingPrevious(ISSTableScanner::next); + testRequestNextRowIteratorWithoutConsumingPrevious(r -> r.partitionIterator(ColumnFilter.NONE, DataRange.allData(r.getPartitioner()), SSTableReadsListener.NOOP_LISTENER), + UnfilteredPartitionIterator::next, + ".*UnfilteredRowIterator.*must be closed.*"); + } + + private static void testRequestNextRowIteratorAfterClosingPrevious(Function makeScanner) + { + final SSTableReader sstable = prepareSmallSSTable(); + + try (UnfilteredPartitionIterator scanner = makeScanner.apply(sstable)) + { + try (UnfilteredRowIterator p = scanner.next()) + { + assertEquals(toKey(0), new String(p.partitionKey().getKey().array())); + // do not read it, but close it + } + + try (UnfilteredRowIterator p = scanner.next()) + { + assertEquals(toKey(3), new String(p.partitionKey().getKey().array())); + assertTrue(p.hasNext()); + assertTrue(p.next() instanceof Row); + } + } + } + + + @Test + public void testSimpleRequestNextRowIteratorAfterClosingPreviouss() + { + testRequestNextRowIteratorAfterClosingPrevious(SSTableReader::getScanner); + } + + @Test + public void testRequestNextRowIteratorAfterClosingPrevious() + { + testRequestNextRowIteratorAfterClosingPrevious(r -> r.partitionIterator(ColumnFilter.NONE, DataRange.allData(r.getPartitioner()), SSTableReadsListener.NOOP_LISTENER)); } } From 17cb89208c804680ffd4445d6a826171a67edb79 Mon Sep 17 00:00:00 2001 From: Dmitry Konstantinov Date: Fri, 7 Mar 2025 23:24:22 +0300 Subject: [PATCH 2/3] Fix marking an SSTable as suspected and BufferPool leakage in case of a corrupted SSTable read during a compaction Patch by Dmitry Konstantinov; reviewed by Branimir Lambov for CASSANDRA-20396 --- .../apache/cassandra/cache/ChunkCache.java | 12 +++++- .../io/sstable/SSTableIdentityIterator.java | 30 +++++++++++++ .../sstable/format/SSTableSimpleScanner.java | 42 ++++++++++++++----- 3 files changed, 72 insertions(+), 12 deletions(-) diff --git a/src/java/org/apache/cassandra/cache/ChunkCache.java b/src/java/org/apache/cassandra/cache/ChunkCache.java index e7e50296ac..5fb0bc6a2f 100644 --- a/src/java/org/apache/cassandra/cache/ChunkCache.java +++ b/src/java/org/apache/cassandra/cache/ChunkCache.java @@ -162,8 +162,16 @@ public class ChunkCache { ByteBuffer buffer = bufferPool.get(key.file.chunkSize(), key.file.preferredBufferType()); assert buffer != null; - key.file.readChunk(key.position, buffer); - return new Buffer(buffer, key.position); + try + { + key.file.readChunk(key.position, buffer); + return new Buffer(buffer, key.position); + } + catch (Throwable t) + { + bufferPool.put(buffer); + throw t; + } } @Override diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index 072a364af3..d5a1ae8bcc 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -76,6 +76,11 @@ public class SSTableIdentityIterator implements Comparable nextRange.lowerPosition) - throw new IllegalArgumentException("Ranges supplied to SSTableSimpleScanner must be non-overlapping and in ascending order."); + PartitionPositionBounds nextRange = rangeIterator.next(); + if (currentEndPosition > nextRange.lowerPosition) + throw new IllegalArgumentException("Ranges supplied to SSTableSimpleScanner must be non-overlapping and in ascending order."); - currentEndPosition = nextRange.upperPosition; - currentStartPosition = nextRange.lowerPosition; - dfile.seek(currentStartPosition); - return true; + currentEndPosition = nextRange.upperPosition; + currentStartPosition = nextRange.lowerPosition; + dfile.seek(currentStartPosition); + return true; + } + catch (CorruptSSTableException e) + { + sstable.markSuspect(); + throw e; + } + catch (IOError e) + { + if (e.getCause() instanceof IOException) + { + sstable.markSuspect(); + throw new CorruptSSTableException((Exception)e.getCause(), sstable.getFilename()); + } + else + { + throw e; + } + } } public UnfilteredRowIterator next() From d4d858d3822c85e6b4b71b1004d8ba6c63fa5134 Mon Sep 17 00:00:00 2001 From: Jon Meredith Date: Tue, 22 Apr 2025 14:31:22 -0600 Subject: [PATCH 3/3] Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. patch by Jon Meredith; reviewed by Marcus Eriksson for CASSANDRA-20561 --- CHANGES.txt | 1 + .../org/apache/cassandra/db/LivenessInfo.java | 11 ++- .../org/apache/cassandra/db/rows/Cells.java | 10 +++ .../org/apache/cassandra/db/CellTest.java | 85 ++++++++++++++++++- .../apache/cassandra/db/LivenessInfoTest.java | 5 ++ .../apache/cassandra/db/rows/RowsTest.java | 54 ++++++++++++ 6 files changed, 163 insertions(+), 3 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 4332b75555..ddd87100ad 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0.18 + * Updating a column with a new TTL but same expiration time is non-deterministic and causes repair mismatches. (CASSANDRA-20561) * Grant permission on keyspaces system_views and system_virtual_schema not possible (CASSANDRA-20171) * Avoid computing prepared statement size for unprepared batches (CASSANDRA-20556) * Fix Dropwizard Meter causes timeouts when infrequently used (CASSANDRA-19332) diff --git a/src/java/org/apache/cassandra/db/LivenessInfo.java b/src/java/org/apache/cassandra/db/LivenessInfo.java index f3e6daabad..2857b0b381 100644 --- a/src/java/org/apache/cassandra/db/LivenessInfo.java +++ b/src/java/org/apache/cassandra/db/LivenessInfo.java @@ -193,7 +193,10 @@ public class LivenessInfo implements IMeasurableMemory * supersedes, ie. tombstone supersedes. * * If timestamps are the same and both of them are expired livenessInfo(Ideally it shouldn't happen), - * greater localDeletionTime wins. + * greater localDeletionTime wins. If the localDeletion times are the same, prefer the + * lower TTL to make the merge deterministic (it is likely that the row has been rewritten with + * USING TTL/TIMESTAMP with an updated TTL that computes to the same local deletion time -- perhaps + * from rerunning a process to migrate user data between clusters or tables). * * @param other * the {@code LivenessInfo} to compare this info to. @@ -207,7 +210,11 @@ public class LivenessInfo implements IMeasurableMemory if (isExpired() ^ other.isExpired()) return isExpired(); if (isExpiring() == other.isExpiring()) - return localExpirationTime() > other.localExpirationTime(); + { + return localExpirationTime() > other.localExpirationTime() || + (localExpirationTime() == other.localExpirationTime() && ttl() < other.ttl()); + } + return isExpiring(); } diff --git a/src/java/org/apache/cassandra/db/rows/Cells.java b/src/java/org/apache/cassandra/db/rows/Cells.java index 59f1d3f7fd..ce774080ee 100644 --- a/src/java/org/apache/cassandra/db/rows/Cells.java +++ b/src/java/org/apache/cassandra/db/rows/Cells.java @@ -113,6 +113,16 @@ public abstract class Cells // would otherwise always win (unless it had an empty value), until it expired and was translated to a tombstone if (leftLocalDeletionTime != rightLocalDeletionTime) return leftLocalDeletionTime > rightLocalDeletionTime ? left : right; + + // Both cells are either tombstones or expiring at the same timestamp. If expiring and the + // TTLs differ, write the lower one -- the write is probably from a more recent + // UPDATE USING TTL AND TIMESTAMP, so select the most recent one to be deterministic and be + // closest to client intent. + if (!leftIsTombstone && left.ttl() != right.ttl()) + { + assert !rightIsTombstone; + return left.ttl() < right.ttl() ? left : right; + } } return compareValues(left, right) >= 0 ? left : right; diff --git a/test/unit/org/apache/cassandra/db/CellTest.java b/test/unit/org/apache/cassandra/db/CellTest.java index d4dec05e3b..a7c334bb79 100644 --- a/test/unit/org/apache/cassandra/db/CellTest.java +++ b/test/unit/org/apache/cassandra/db/CellTest.java @@ -41,7 +41,8 @@ import org.apache.cassandra.serializers.MarshalException; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; -import static java.util.Arrays.*; +import static java.util.Arrays.asList; +import static org.junit.Assert.assertEquals; public class CellTest { @@ -62,6 +63,8 @@ public class CellTest .addRegularColumn("v", IntegerType.instance) .addRegularColumn("m", MapType.getInstance(IntegerType.instance, IntegerType.instance, true)) .build(); + public static final ByteBuffer TEST_VALUE = ByteBufferUtil.bytes("a"); + @BeforeClass public static void defineSchema() throws ConfigurationException @@ -253,6 +256,86 @@ public class CellTest Assert.assertEquals(-1, testExpiring("val", "b", 2, 1, null, "a", null, 1)); } + + public static void assertCellsEqual(Cell cellA, Cell cellB) + { + assertEquals(cellA.timestamp(), cellB.timestamp()); + assertEquals(cellA.ttl(), cellB.ttl()); + assertEquals(cellA.localDeletionTime(), cellB.localDeletionTime()); + assertEquals(cellA.buffer(), cellB.buffer()); + } + + static void checkCommutes(ColumnMetadata cmd, long timestamp, long tsDiff, int ttl, int ttlDiff, int nowInSeconds, int nowDiff) + { + long timestampA = timestamp; + long timestampB = timestampA + tsDiff; + int ttlA = ttl; + int ttlB = ttl + ttlDiff; + int nowInSecsA = nowInSeconds; + int nowInSecsB = nowInSecsA + nowDiff; + if (nowInSecsA < 0 || nowInSecsB < 0) + return; + + Cell cellA = ttlA == 0 ? BufferCell.tombstone(cmd, timestampA, nowInSecsA) : + ttlA < 0 ? BufferCell.live(cmd, timestampA, TEST_VALUE) : + BufferCell.expiring(cmd, timestampA, ttlA, nowInSecsA, TEST_VALUE); + Cell cellB = ttlB == 0 ? BufferCell.tombstone(cmd, timestampB, nowInSecsB) : + ttlB < 0 ? BufferCell.live(cmd, timestampB, TEST_VALUE) : + BufferCell.expiring(cmd, timestampB, ttlB, nowInSecsB, TEST_VALUE); + + Cell cellAB = Cells.reconcile(cellA, cellB); + Cell cellBA = Cells.reconcile(cellB, cellA); + + assertCellsEqual(cellAB, cellBA); + } + + @Test + public void checkSameValueDifferentLivenessCommutes() + { + ColumnMetadata cmd = fakeColumn("c", UTF8Type.instance); + long[] tsDiffs = new long[] {0L, + 1L, // microsecond + 1000L, // millisecond + 1000000L, // second + 60000000L}; // minute + int[] ttls = new int[] { -1, 0, 1, 3600, 24 * 3600, 7 * 24 * 3600, 60 * 24 * 3600, 366 * 24 * 3600 }; + int[] ttlDiffs = new int[] { 0, 1, 60, 3600, 24 * 3600, 7 * 24 * 3600, 60 * 24 * 3600, 366 * 24 * 3600 }; + + int nowInSeconds = FBUtilities.nowInSeconds(); + long timestamp = FBUtilities.timestampMicros(); + + for (long tsDiff: tsDiffs) + { + for (int ttl: ttls) + { + for (int ttlDiff : ttlDiffs) + { + for (Integer nowDiff : ttlDiffs) + checkCommutes(cmd, timestamp, tsDiff, ttl, ttlDiff, nowInSeconds, nowDiff); + } + } + } + } + + // Checks that reconciling a cell with a smaller TTL reconcile commutatively + // Similar to rewriting data retrieved with SELECT v, TTL(v), WRITETIMESTAMP(v) with + // INSERT SET v=? USING TTL ? AND TIMESTAMP ? + @Test + public void rewriteCellWithSmallerTTL() + { + ColumnMetadata cmd = fakeColumn("c", UTF8Type.instance); + int[] nowDiffs = new int[] { 0, 1, 60, 3600, 24 * 3600, 7 * 24 * 3600, 60 * 24 * 3600, 366 * 24 * 3600 }; + long timestamp = FBUtilities.timestampMicros(); + int nowInSeconds = FBUtilities.nowInSeconds(); + int ttl = 3600; + + for (Integer nowDiff : nowDiffs) + { + checkCommutes(cmd, timestamp, 0L, ttl, -nowDiff, nowInSeconds, nowDiff); + } + } + + class SimplePurger implements DeletionPurger { private final int gcBefore; diff --git a/test/unit/org/apache/cassandra/db/LivenessInfoTest.java b/test/unit/org/apache/cassandra/db/LivenessInfoTest.java index 557870672e..0482423f9d 100644 --- a/test/unit/org/apache/cassandra/db/LivenessInfoTest.java +++ b/test/unit/org/apache/cassandra/db/LivenessInfoTest.java @@ -75,6 +75,11 @@ public class LivenessInfoTest first = LivenessInfo.withExpirationTime(100, LivenessInfo.EXPIRED_LIVENESS_TTL, nowInSeconds + 1); second = LivenessInfo.withExpirationTime(100, LivenessInfo.EXPIRED_LIVENESS_TTL, nowInSeconds); assertSupersedes(first, second); + + // rewritten expiring with the same expiration time and a lower TTL, take the lower TTL as likely to be more recent + first = LivenessInfo.withExpirationTime(100, 4, nowInSeconds); + second = LivenessInfo.withExpirationTime(100, 5, nowInSeconds); + assertSupersedes(first, second); } @Test diff --git a/test/unit/org/apache/cassandra/db/rows/RowsTest.java b/test/unit/org/apache/cassandra/db/rows/RowsTest.java index cfeebfd2b1..47e42b2542 100644 --- a/test/unit/org/apache/cassandra/db/rows/RowsTest.java +++ b/test/unit/org/apache/cassandra/db/rows/RowsTest.java @@ -45,6 +45,8 @@ import org.apache.cassandra.db.partitions.PartitionStatisticsCollector; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; +import static org.apache.cassandra.db.CellTest.assertCellsEqual; + public class RowsTest { private static final String KEYSPACE = "rows_test"; @@ -520,6 +522,58 @@ public class RowsTest Assert.assertEquals(0, merged.columns().size()); } + + public static BufferCell expiringWithExpirationTime(ColumnMetadata column, long timestamp, int ttl, int localDeletionTime, ByteBuffer value) + { + return expiringWithExpirationTime(column, timestamp, ttl, localDeletionTime, value, null); + } + + public static BufferCell expiringWithExpirationTime(ColumnMetadata column, long timestamp, int ttl, int localDeletionTime, ByteBuffer value, CellPath path) + { + assert ttl != Cell.NO_TTL; + return new BufferCell(column, timestamp, ttl, localDeletionTime, value, path); + } + + @Test + public void mergeRowsWithSameExpiryDifferentTTLCommutesLiveness() + { + int now1 = FBUtilities.nowInSeconds(); + long ts1 = secondToTs(now1); + int ldt = now1 + 1000; + + Row.Builder r1Builder = BTreeRow.unsortedBuilder(); + r1Builder.newRow(c1); + LivenessInfo originalLiveness = LivenessInfo.withExpirationTime(ts1, 100, ldt); + r1Builder.addPrimaryKeyLivenessInfo(originalLiveness); + + Row.Builder r2Builder = BTreeRow.unsortedBuilder(); + r2Builder.newRow(c1); + LivenessInfo loweredTTL = LivenessInfo.withExpirationTime(ts1, 50, ldt); + r2Builder.addPrimaryKeyLivenessInfo(loweredTTL); + + Cell r2v = expiringWithExpirationTime(v, ts1, 75, ldt, BB1); + Cell r2m2 = expiringWithExpirationTime(m, ts1, 50, ldt, BB1, CellPath.create(BB2)); + Cell r2m3 = expiringWithExpirationTime(m, ts1, 75, ldt, BB2, CellPath.create(BB3)); + Cell r2m4 = expiringWithExpirationTime(m, ts1, 100, ldt, BB3, CellPath.create(BB4)); + List> expectedCells = Lists.newArrayList(r2v, r2m2, r2m3, r2m4); + + expectedCells.forEach(r1Builder::addCell); + expectedCells.forEach(r2Builder::addCell); + + Row r1 = r1Builder.build(); + Row r2 = r2Builder.build(); + + Row r1r2 = Rows.merge(r1, r2); + Row r2r1 = Rows.merge(r2, r1); + + DiffListener mergedListener = new DiffListener(); + Rows.diff(mergedListener, r1r2, r2r1); + + mergedListener.liveness.forEach(pair -> Assert.assertEquals(pair.merged, pair.original)); + mergedListener.cells.forEach(pair -> assertCellsEqual(pair.merged, pair.original)); + } + + // Creates a dummy cell for a (regular) column for the provided name and without a cellPath. private static Cell liveCell(ColumnMetadata name) {