diff --git a/CHANGES.txt b/CHANGES.txt index 306f46f23c..677ea114d1 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -25,6 +25,7 @@ Merged from 2.2: header is received (CASSANDRA-11464) * Validate that num_tokens and initial_token are consistent with one another (CASSANDRA-10120) Merged from 2.2: + * Fix commit log replay after out-of-order flush completion (CASSANDRA-9669) * cqlsh: correctly handle non-ascii chars in error messages (CASSANDRA-11626) * Exit JVM if JMX server fails to startup (CASSANDRA-11540) * Produce a heap dump when exiting on OOM (CASSANDRA-9861) diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index 5b8de8ff15..e9a2938eb0 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -173,6 +173,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + @VisibleForTesting + public static volatile ColumnFamilyStore discardFlushResults; + public final Keyspace keyspace; public final String name; public final CFMetaData metadata; @@ -777,14 +780,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * * @param memtable */ - public Future switchMemtableIfCurrent(Memtable memtable) + public ListenableFuture switchMemtableIfCurrent(Memtable memtable) { synchronized (data) { if (data.getView().getCurrentMemtable() == memtable) return switchMemtable(); } - return Futures.immediateFuture(null); + return waitForFlushes(); } /* @@ -794,14 +797,14 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * not complete until the Memtable (and all prior Memtables) have been successfully flushed, and the CL * marked clean up to the position owned by the Memtable. */ - public ListenableFuture switchMemtable() + public ListenableFuture switchMemtable() { synchronized (data) { logFlush(); Flush flush = new Flush(false); flushExecutor.execute(flush); - ListenableFutureTask task = ListenableFutureTask.create(flush.postFlush, null); + ListenableFutureTask task = ListenableFutureTask.create(flush.postFlush); postFlushExecutor.submit(task); return task; } @@ -833,77 +836,93 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } - public ListenableFuture forceFlush() + /** + * Flush if there is unflushed data in the memtables + * + * @return a Future yielding the commit log position that can be guaranteed to have been successfully written + * to sstables for this table once the future completes + */ + public ListenableFuture forceFlush() { - return forceFlush(null); + Memtable current = data.getView().getCurrentMemtable(); + for (ColumnFamilyStore cfs : concatWithIndexes()) + if (!cfs.data.getView().getCurrentMemtable().isClean()) + return switchMemtableIfCurrent(current); + return waitForFlushes(); } /** * Flush if there is unflushed data that was written to the CommitLog before @param flushIfDirtyBefore - * (inclusive). If @param flushIfDirtyBefore is null, flush if there is any unflushed data. + * (inclusive). * - * @return a Future such that when the future completes, all data inserted before forceFlush was called, - * will be flushed. + * @return a Future yielding the commit log position that can be guaranteed to have been successfully written + * to sstables for this table once the future completes */ - public ListenableFuture forceFlush(ReplayPosition flushIfDirtyBefore) + public ListenableFuture forceFlush(ReplayPosition flushIfDirtyBefore) { - // we synchronize on the data tracker to ensure we don't race against other calls to switchMemtable(), - // unnecessarily queueing memtables that are about to be made clean - synchronized (data) - { - // during index build, 2ary index memtables can be dirty even if parent is not. if so, - // we want to flush the 2ary index ones too. - boolean clean = true; - for (ColumnFamilyStore cfs : concatWithIndexes()) - clean &= cfs.data.getView().getCurrentMemtable().isCleanAfter(flushIfDirtyBefore); - - if (clean) - { - // We could have a memtable for this column family that is being - // flushed. Make sure the future returned wait for that so callers can - // assume that any data inserted prior to the call are fully flushed - // when the future returns (see #5241). - ListenableFutureTask task = ListenableFutureTask.create(new Runnable() - { - public void run() - { - logger.trace("forceFlush requested but everything is clean in {}", name); - } - }, null); - postFlushExecutor.execute(task); - return task; - } - - return switchMemtable(); - } + // we don't loop through the remaining memtables since here we only care about commit log dirtiness + // and this does not vary between a table and its table-backed indexes + Memtable current = data.getView().getCurrentMemtable(); + if (current.mayContainDataBefore(flushIfDirtyBefore)) + return switchMemtableIfCurrent(current); + return waitForFlushes(); } - public void forceBlockingFlush() + /** + * @return a Future yielding the commit log position that can be guaranteed to have been successfully written + * to sstables for this table once the future completes + */ + private ListenableFuture waitForFlushes() { - FBUtilities.waitOnFuture(forceFlush()); + // we grab the current memtable; once any preceding memtables have flushed, we know its + // commitLogLowerBound has been set (as this it is set with the upper bound of the preceding memtable) + final Memtable current = data.getView().getCurrentMemtable(); + ListenableFutureTask task = ListenableFutureTask.create(new Callable() + { + public ReplayPosition call() + { + logger.debug("forceFlush requested but everything is clean in {}", name); + return current.getCommitLogLowerBound(); + } + }); + postFlushExecutor.execute(task); + return task; + } + + public ReplayPosition forceBlockingFlush() + { + return FBUtilities.waitOnFuture(forceFlush()); } /** * Both synchronises custom secondary indexes and provides ordering guarantees for futures on switchMemtable/flush * etc, which expect to be able to wait until the flush (and all prior flushes) requested have completed. */ - private final class PostFlush implements Runnable + private final class PostFlush implements Callable { final boolean flushSecondaryIndexes; final OpOrder.Barrier writeBarrier; final CountDownLatch latch = new CountDownLatch(1); - final ReplayPosition lastReplayPosition; volatile FSWriteError flushFailure = null; + final ReplayPosition commitLogUpperBound; + final List memtables; + final List> readers; - private PostFlush(boolean flushSecondaryIndexes, OpOrder.Barrier writeBarrier, ReplayPosition lastReplayPosition) + private PostFlush(boolean flushSecondaryIndexes, OpOrder.Barrier writeBarrier, ReplayPosition commitLogUpperBound, + List memtables, List> readers) { this.writeBarrier = writeBarrier; this.flushSecondaryIndexes = flushSecondaryIndexes; - this.lastReplayPosition = lastReplayPosition; + this.commitLogUpperBound = commitLogUpperBound; + this.memtables = memtables; + this.readers = readers; } - public void run() + public ReplayPosition call() { + if (discardFlushResults == ColumnFamilyStore.this) + return commitLogUpperBound; + writeBarrier.await(); /** @@ -918,7 +937,7 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean try { - // we wait on the latch for the lastReplayPosition to be set, and so that waiters + // we wait on the latch for the commitLogUpperBound to be set, and so that waiters // on this task can rely on all prior flushes being complete latch.await(); } @@ -927,18 +946,25 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean throw new IllegalStateException(); } - // must check lastReplayPosition != null because Flush may find that all memtables are clean - // and so not set a lastReplayPosition // If a flush errored out but the error was ignored, make sure we don't discard the commit log. - if (lastReplayPosition != null && flushFailure == null) + if (flushFailure == null) { - CommitLog.instance.discardCompletedSegments(metadata.cfId, lastReplayPosition); + CommitLog.instance.discardCompletedSegments(metadata.cfId, commitLogUpperBound); + for (int i = 0 ; i < memtables.size() ; i++) + { + Memtable memtable = memtables.get(i); + Collection reader = readers.get(i); + memtable.cfs.data.permitCompactionOfFlushed(reader); + memtable.cfs.compactionStrategyManager.replaceFlushed(memtable, reader); + } } metric.pendingFlushes.dec(); if (flushFailure != null) throw flushFailure; + + return commitLogUpperBound; } } @@ -953,7 +979,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean private final class Flush implements Runnable { final OpOrder.Barrier writeBarrier; - final List memtables; + final List memtables = new ArrayList<>(); + final List> readers = new ArrayList<>(); final PostFlush postFlush; final boolean truncate; @@ -969,43 +996,33 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * that all write operations register themselves with, and assigning this barrier to the memtables, * after which we *.issue()* the barrier. This barrier is used to direct write operations started prior * to the barrier.issue() into the memtable we have switched out, and any started after to its replacement. - * In doing so it also tells the write operations to update the lastReplayPosition of the memtable, so + * In doing so it also tells the write operations to update the commitLogUpperBound of the memtable, so * that we know the CL position we are dirty to, which can be marked clean when we complete. */ writeBarrier = keyspace.writeOrder.newBarrier(); - memtables = new ArrayList<>(); // submit flushes for the memtable for any indexed sub-cfses, and our own - AtomicReference lastReplayPositionHolder = new AtomicReference<>(); + AtomicReference commitLogUpperBound = new AtomicReference<>(); for (ColumnFamilyStore cfs : concatWithIndexes()) { // switch all memtables, regardless of their dirty status, setting the barrier // so that we can reach a coordinated decision about cleanliness once they // are no longer possible to be modified - Memtable mt = cfs.data.switchMemtable(truncate); - mt.setDiscarding(writeBarrier, lastReplayPositionHolder); - memtables.add(mt); + Memtable newMemtable = new Memtable(commitLogUpperBound, cfs); + Memtable oldMemtable = cfs.data.switchMemtable(truncate, newMemtable); + oldMemtable.setDiscarding(writeBarrier, commitLogUpperBound); + memtables.add(oldMemtable); } - // we now attempt to define the lastReplayPosition; we do this by grabbing the current limit from the CL - // and attempting to set the holder to this value. at the same time all writes to the memtables are - // also maintaining this value, so if somebody sneaks ahead of us somehow (should be rare) we simply retry, - // so that we know all operations prior to the position have not reached it yet - ReplayPosition lastReplayPosition; - while (true) - { - lastReplayPosition = new Memtable.LastReplayPosition(CommitLog.instance.getContext()); - ReplayPosition currentLast = lastReplayPositionHolder.get(); - if ((currentLast == null || currentLast.compareTo(lastReplayPosition) <= 0) - && lastReplayPositionHolder.compareAndSet(currentLast, lastReplayPosition)) - break; - } + // we then ensure an atomic decision is made about the upper bound of the continuous range of commit log + // records owned by this memtable + setCommitLogUpperBound(commitLogUpperBound); // we then issue the barrier; this lets us wait for all operations started prior to the barrier to complete; - // since this happens after wiring up the lastReplayPosition, we also know all operations with earlier + // since this happens after wiring up the commitLogUpperBound, we also know all operations with earlier // replay positions have also completed, i.e. the memtables are done and ready to flush writeBarrier.issue(); - postFlush = new PostFlush(!truncate, writeBarrier, lastReplayPosition); + postFlush = new PostFlush(!truncate, writeBarrier, commitLogUpperBound.get(), memtables, readers); } public void run() @@ -1024,27 +1041,23 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean memtable.cfs.data.markFlushing(memtable); if (memtable.isClean() || truncate) { - memtable.cfs.replaceFlushed(memtable, null); + memtable.cfs.data.replaceFlushed(memtable, Collections.emptyList()); + memtable.cfs.compactionStrategyManager.replaceFlushed(memtable, Collections.emptyList()); reclaim(memtable); iter.remove(); } } - if (memtables.isEmpty()) - { - postFlush.latch.countDown(); - return; - } - metric.memtableSwitchCount.inc(); try { for (Memtable memtable : memtables) { - // flush the memtable - MoreExecutors.sameThreadExecutor().execute(memtable.flushRunnable()); + Collection readers = memtable.flush(); + memtable.cfs.data.replaceFlushed(memtable, readers); reclaim(memtable); + this.readers.add(readers); } } catch (FSWriteError e) @@ -1074,6 +1087,38 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean } } + // atomically set the upper bound for the commit log + private static void setCommitLogUpperBound(AtomicReference commitLogUpperBound) + { + // we attempt to set the holder to the current commit log context. at the same time all writes to the memtables are + // also maintaining this value, so if somebody sneaks ahead of us somehow (should be rare) we simply retry, + // so that we know all operations prior to the position have not reached it yet + ReplayPosition lastReplayPosition; + while (true) + { + lastReplayPosition = new Memtable.LastReplayPosition(CommitLog.instance.getContext()); + ReplayPosition currentLast = commitLogUpperBound.get(); + if ((currentLast == null || currentLast.compareTo(lastReplayPosition) <= 0) + && commitLogUpperBound.compareAndSet(currentLast, lastReplayPosition)) + break; + } + } + + @VisibleForTesting + // this method should ONLY be used for testing commit log behaviour; it discards the current memtable + // contents without marking the commit log clean, and prevents any proceeding flushes from marking + // the commit log as done, however they *will* terminate (unlike under typical failures) to ensure progress is made + public void simulateFailedFlush() + { + discardFlushResults = this; + data.markFlushing(data.switchMemtable(false, new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), this))); + } + + public void resumeFlushing() + { + discardFlushResults = null; + } + /** * Finds the largest memtable, as a percentage of *either* on- or off-heap memory limits, and immediately * queues it for flushing. If the memtable selected is flushed before this completes, no work is done. @@ -1406,6 +1451,16 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean return data; } + public Collection getSSTables() + { + return data.getSSTables(); + } + + public Iterable getPermittedToCompactSSTables() + { + return data.getPermittedToCompact(); + } + public Set getLiveSSTables() { return data.getView().liveSSTables(); @@ -1910,40 +1965,52 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean // position in the System keyspace. logger.trace("truncating {}", name); - if (keyspace.getMetadata().params.durableWrites || DatabaseDescriptor.isAutoSnapshot()) - { - // flush the CF being truncated before forcing the new segment - forceBlockingFlush(); + final long truncatedAt; + final ReplayPosition replayAfter; - viewManager.forceBlockingFlush(); - - // sleep a little to make sure that our truncatedAt comes after any sstable - // that was part of the flushed we forced; otherwise on a tie, it won't get deleted. - Uninterruptibles.sleepUninterruptibly(1, TimeUnit.MILLISECONDS); - } - else + synchronized (data) { - dumpMemtable(); - viewManager.dumpMemtables(); + if (keyspace.getMetadata().params.durableWrites || DatabaseDescriptor.isAutoSnapshot()) + { + replayAfter = forceBlockingFlush(); + viewManager.forceBlockingFlush(); + } + else + { + // just nuke the memtable data w/o writing to disk first + viewManager.dumpMemtables(); + try + { + replayAfter = dumpMemtable().get(); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + } + + long now = System.currentTimeMillis(); + // make sure none of our sstables are somehow in the future (clock drift, perhaps) + for (ColumnFamilyStore cfs : concatWithIndexes()) + for (SSTableReader sstable : cfs.data.getSSTables()) + now = Math.max(now, sstable.maxDataAge); + truncatedAt = now; } Runnable truncateRunnable = new Runnable() { public void run() { - logger.trace("Discarding sstable data for truncated CF + indexes"); - - final long truncatedAt = System.currentTimeMillis(); + logger.debug("Discarding sstable data for truncated CF + indexes"); data.notifyTruncated(truncatedAt); if (DatabaseDescriptor.isAutoSnapshot()) snapshot(Keyspace.getTimestampedSnapshotName(name)); - ReplayPosition replayAfter = discardSSTables(truncatedAt); + discardSSTables(truncatedAt); indexManager.truncateAllIndexesBlocking(truncatedAt); - - viewManager.truncateBlocking(truncatedAt); + viewManager.truncateBlocking(replayAfter, truncatedAt); SystemKeyspace.saveTruncationRecord(ColumnFamilyStore.this, truncatedAt, replayAfter); logger.trace("cleaning out row cache"); @@ -1958,13 +2025,13 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean /** * Drops current memtable without flushing to disk. This should only be called when truncating a column family which is not durable. */ - public void dumpMemtable() + public Future dumpMemtable() { synchronized (data) { final Flush flush = new Flush(true); flushExecutor.execute(flush); - postFlushExecutor.submit(flush.postFlush); + return postFlushExecutor.submit(flush.postFlush); } } @@ -2024,7 +2091,9 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean public LifecycleTransaction call() throws Exception { assert data.getCompacting().isEmpty() : data.getCompacting(); - Collection sstables = Lists.newArrayList(AbstractCompactionStrategy.filterSuspectSSTables(getSSTables(SSTableSet.LIVE))); + Iterable sstables = getPermittedToCompactSSTables(); + sstables = AbstractCompactionStrategy.filterSuspectSSTables(sstables); + sstables = ImmutableList.copyOf(sstables); LifecycleTransaction modifier = data.tryModify(sstables, operationType); assert modifier != null: "something marked things compacting while compactions are disabled"; return modifier; @@ -2291,10 +2360,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean * * @param truncatedAt The timestamp of the truncation * (all SSTables before that timestamp are going be marked as compacted) - * - * @return the most recent replay position of the truncated data */ - public ReplayPosition discardSSTables(long truncatedAt) + public void discardSSTables(long truncatedAt) { assert data.getCompacting().isEmpty() : data.getCompacting(); @@ -2306,11 +2373,8 @@ public class ColumnFamilyStore implements ColumnFamilyStoreMBean truncatedSSTables.add(sstable); } - if (truncatedSSTables.isEmpty()) - return ReplayPosition.NONE; - - markObsolete(truncatedSSTables, OperationType.UNKNOWN); - return ReplayPosition.getReplayPosition(truncatedSSTables); + if (!truncatedSSTables.isEmpty()) + markObsolete(truncatedSSTables, OperationType.UNKNOWN); } public double getDroppableTombstoneRatio() diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java index 5f23baa922..e00c8b92c2 100644 --- a/src/java/org/apache/cassandra/db/Directories.java +++ b/src/java/org/apache/cassandra/db/Directories.java @@ -385,7 +385,7 @@ public class Directories if (candidates.isEmpty()) if (tooBig) - return null; + throw new RuntimeException("Insufficient disk space to write " + writeSize + " bytes"); else throw new FSWriteError(new IOException("All configured data directories have been blacklisted as unwritable for erroring out"), ""); diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 5d5f7bf009..93dc5afc23 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -70,14 +70,17 @@ public class Memtable implements Comparable // the write barrier for directing writes to this memtable during a switch private volatile OpOrder.Barrier writeBarrier; - // the last ReplayPosition owned by this Memtable; all ReplayPositions lower are owned by this or an earlier Memtable - private volatile AtomicReference lastReplayPosition; - // the "first" ReplayPosition owned by this Memtable; this is inaccurate, and only used as a convenience to prevent CLSM flushing wantonly - private final ReplayPosition minReplayPosition = CommitLog.instance.getContext(); + // the precise upper bound of ReplayPosition owned by this memtable + private volatile AtomicReference commitLogUpperBound; + // the precise lower bound of ReplayPosition owned by this memtable; equal to its predecessor's commitLogUpperBound + private AtomicReference commitLogLowerBound; + // the approximate lower bound by this memtable; must be <= commitLogLowerBound once our predecessor + // has been finalised, and this is enforced in the ColumnFamilyStore.setCommitLogUpperBound + private final ReplayPosition approximateCommitLogLowerBound = CommitLog.instance.getContext(); public int compareTo(Memtable that) { - return this.minReplayPosition.compareTo(that.minReplayPosition); + return this.approximateCommitLogLowerBound.compareTo(that.approximateCommitLogLowerBound); } public static final class LastReplayPosition extends ReplayPosition @@ -92,7 +95,6 @@ public class Memtable implements Comparable // actually only store DecoratedKey. private final ConcurrentNavigableMap partitions = new ConcurrentSkipListMap<>(); public final ColumnFamilyStore cfs; - private final long creationTime = System.currentTimeMillis(); private final long creationNano = System.nanoTime(); // The smallest timestamp for all partitions stored in this memtable @@ -106,9 +108,11 @@ public class Memtable implements Comparable private final ColumnsCollector columnsCollector; private final StatsCollector statsCollector = new StatsCollector(); - public Memtable(ColumnFamilyStore cfs) + // only to be used by init(), to setup the very first memtable for the cfs + public Memtable(AtomicReference commitLogLowerBound, ColumnFamilyStore cfs) { this.cfs = cfs; + this.commitLogLowerBound = commitLogLowerBound; this.allocator = MEMORY_POOL.newAllocator(); this.initialComparator = cfs.metadata.comparator; this.cfs.scheduleFlush(); @@ -144,7 +148,7 @@ public class Memtable implements Comparable public void setDiscarding(OpOrder.Barrier writeBarrier, AtomicReference lastReplayPosition) { assert this.writeBarrier == null; - this.lastReplayPosition = lastReplayPosition; + this.commitLogUpperBound = lastReplayPosition; this.writeBarrier = writeBarrier; allocator.setDiscarding(); } @@ -174,16 +178,21 @@ public class Memtable implements Comparable // its current value and ours; if it HAS been finalised, we simply accept its judgement // this permits us to coordinate a safe boundary, as the boundary choice is made // atomically wrt our max() maintenance, so an operation cannot sneak into the past - ReplayPosition currentLast = lastReplayPosition.get(); + ReplayPosition currentLast = commitLogUpperBound.get(); if (currentLast instanceof LastReplayPosition) return currentLast.compareTo(replayPosition) >= 0; if (currentLast != null && currentLast.compareTo(replayPosition) >= 0) return true; - if (lastReplayPosition.compareAndSet(currentLast, replayPosition)) + if (commitLogUpperBound.compareAndSet(currentLast, replayPosition)) return true; } } + public ReplayPosition getCommitLogLowerBound() + { + return commitLogLowerBound.get(); + } + public boolean isLive() { return allocator.isLive(); @@ -194,9 +203,9 @@ public class Memtable implements Comparable return partitions.isEmpty(); } - public boolean isCleanAfter(ReplayPosition position) + public boolean mayContainDataBefore(ReplayPosition position) { - return isClean() || (position != null && minReplayPosition.compareTo(position) >= 0); + return approximateCommitLogLowerBound.compareTo(position) < 0; } /** @@ -254,11 +263,6 @@ public class Memtable implements Comparable return partitions.size(); } - public FlushRunnable flushRunnable() - { - return new FlushRunnable(lastReplayPosition.get()); - } - public String toString() { return String.format("Memtable-%s@%s(%s serialized bytes, %s ops, %.0f%%/%.0f%% of on/off-heap limit)", @@ -311,9 +315,15 @@ public class Memtable implements Comparable return partitions.get(key); } - public long creationTime() + public Collection flush() { - return creationTime; + long estimatedSize = estimatedSize(); + Directories.DataDirectory dataDirectory = cfs.getDirectories().getWriteableLocation(estimatedSize); + if (dataDirectory == null) + throw new RuntimeException("Insufficient disk space to write " + estimatedSize + " bytes"); + File sstableDirectory = cfs.getDirectories().getLocationForDisk(dataDirectory); + assert sstableDirectory != null : "Flush task is not bound to any disk"; + return writeSortedContents(sstableDirectory); } public long getMinTimestamp() @@ -321,136 +331,109 @@ public class Memtable implements Comparable return minTimestamp; } - class FlushRunnable extends DiskAwareRunnable + private long estimatedSize() { - private final ReplayPosition context; - private final long estimatedSize; - - private final boolean isBatchLogTable; - - FlushRunnable(ReplayPosition context) + long keySize = 0; + for (PartitionPosition key : partitions.keySet()) { - this.context = context; + // make sure we don't write non-sensical keys + assert key instanceof DecoratedKey; + keySize += ((DecoratedKey)key).getKey().remaining(); + } + return (long) ((keySize // index entries + + keySize // keys in data file + + liveDataSize.get()) // data + * 1.2); // bloom filter and row index overhead + } - long keySize = 0; - for (PartitionPosition key : partitions.keySet()) + private Collection writeSortedContents(File sstableDirectory) + { + boolean isBatchLogTable = cfs.name.equals(SystemKeyspace.BATCHES) && cfs.keyspace.getName().equals(SystemKeyspace.NAME); + + logger.debug("Writing {}", Memtable.this.toString()); + + Collection ssTables; + try (SSTableTxnWriter writer = createFlushWriter(cfs.getSSTablePath(sstableDirectory), columnsCollector.get(), statsCollector.get())) + { + boolean trackContention = logger.isTraceEnabled(); + int heavilyContendedRowCount = 0; + // (we can't clear out the map as-we-go to free up memory, + // since the memtable is being used for queries in the "pending flush" category) + for (AtomicBTreePartition partition : partitions.values()) { - // make sure we don't write non-sensical keys - assert key instanceof DecoratedKey; - keySize += ((DecoratedKey)key).getKey().remaining(); - } - estimatedSize = (long) ((keySize // index entries - + keySize // keys in data file - + liveDataSize.get()) // data - * 1.2); // bloom filter and row index overhead + // Each batchlog partition is a separate entry in the log. And for an entry, we only do 2 + // operations: 1) we insert the entry and 2) we delete it. Further, BL data is strictly local, + // we don't need to preserve tombstones for repair. So if both operation are in this + // memtable (which will almost always be the case if there is no ongoing failure), we can + // just skip the entry (CASSANDRA-4667). + if (isBatchLogTable && !partition.partitionLevelDeletion().isLive() && partition.hasRows()) + continue; - this.isBatchLogTable = cfs.name.equals(SystemKeyspace.BATCHES) && cfs.keyspace.getName().equals(SystemKeyspace.NAME); - } + if (trackContention && partition.usePessimisticLocking()) + heavilyContendedRowCount++; - public long getExpectedWriteSize() - { - return estimatedSize; - } - - protected void runMayThrow() throws Exception - { - long writeSize = getExpectedWriteSize(); - Directories.DataDirectory dataDirectory = getWriteDirectory(writeSize); - File sstableDirectory = cfs.getDirectories().getLocationForDisk(dataDirectory); - assert sstableDirectory != null : "Flush task is not bound to any disk"; - Collection sstables = writeSortedContents(context, sstableDirectory); - cfs.replaceFlushed(Memtable.this, sstables); - } - - protected Directories getDirectories() - { - return cfs.getDirectories(); - } - - private Collection writeSortedContents(ReplayPosition context, File sstableDirectory) - { - logger.debug("Writing {}", Memtable.this.toString()); - - Collection ssTables; - try (SSTableTxnWriter writer = createFlushWriter(cfs.getSSTablePath(sstableDirectory), columnsCollector.get(), statsCollector.get())) - { - boolean trackContention = logger.isTraceEnabled(); - int heavilyContendedRowCount = 0; - // (we can't clear out the map as-we-go to free up memory, - // since the memtable is being used for queries in the "pending flush" category) - for (AtomicBTreePartition partition : partitions.values()) + if (!partition.isEmpty()) { - // Each batchlog partition is a separate entry in the log. And for an entry, we only do 2 - // operations: 1) we insert the entry and 2) we delete it. Further, BL data is strictly local, - // we don't need to preserve tombstones for repair. So if both operation are in this - // memtable (which will almost always be the case if there is no ongoing failure), we can - // just skip the entry (CASSANDRA-4667). - if (isBatchLogTable && !partition.partitionLevelDeletion().isLive() && partition.hasRows()) - continue; - - if (trackContention && partition.usePessimisticLocking()) - heavilyContendedRowCount++; - - if (!partition.isEmpty()) + try (UnfilteredRowIterator iter = partition.unfilteredIterator()) { - try (UnfilteredRowIterator iter = partition.unfilteredIterator()) - { - writer.append(iter); - } + writer.append(iter); } } - - if (writer.getFilePointer() > 0) - { - logger.debug(String.format("Completed flushing %s (%s) for commitlog position %s", - writer.getFilename(), - FBUtilities.prettyPrintMemory(writer.getFilePointer()), - context)); - - // sstables should contain non-repaired data. - ssTables = writer.finish(true); - } - else - { - logger.debug("Completed flushing {}; nothing needed to be retained. Commitlog position was {}", - writer.getFilename(), context); - writer.abort(); - ssTables = null; - } - - if (heavilyContendedRowCount > 0) - logger.trace(String.format("High update contention in %d/%d partitions of %s ", heavilyContendedRowCount, partitions.size(), Memtable.this.toString())); - - return ssTables; } + + if (writer.getFilePointer() > 0) + { + logger.debug(String.format("Completed flushing %s (%s) for commitlog position %s", + writer.getFilename(), + FBUtilities.prettyPrintMemory(writer.getFilePointer()), + commitLogUpperBound)); + + // sstables should contain non-repaired data. + ssTables = writer.finish(true); + } + else + { + logger.debug("Completed flushing {}; nothing needed to be retained. Commitlog position was {}", + writer.getFilename(), commitLogUpperBound); + writer.abort(); + ssTables = Collections.emptyList(); + } + + if (heavilyContendedRowCount > 0) + logger.trace(String.format("High update contention in %d/%d partitions of %s ", heavilyContendedRowCount, partitions.size(), Memtable.this.toString())); + + return ssTables; } + } - @SuppressWarnings("resource") // log and writer closed by SSTableTxnWriter - public SSTableTxnWriter createFlushWriter(String filename, - PartitionColumns columns, - EncodingStats stats) + @SuppressWarnings("resource") // log and writer closed by SSTableTxnWriter + public SSTableTxnWriter createFlushWriter(String filename, + PartitionColumns columns, + EncodingStats stats) + { + // we operate "offline" here, as we expose the resulting reader consciously when done + // (although we may want to modify this behaviour in future, to encapsulate full flush behaviour in LifecycleTransaction) + LifecycleTransaction txn = null; + try { - // we operate "offline" here, as we expose the resulting reader consciously when done - // (although we may want to modify this behaviour in future, to encapsulate full flush behaviour in LifecycleTransaction) - LifecycleTransaction txn = null; - try - { - txn = LifecycleTransaction.offline(OperationType.FLUSH); - MetadataCollector sstableMetadataCollector = new MetadataCollector(cfs.metadata.comparator).replayPosition(context); - return new SSTableTxnWriter(txn, - cfs.createSSTableMultiWriter(Descriptor.fromFilename(filename), - (long) partitions.size(), - ActiveRepairService.UNREPAIRED_SSTABLE, - sstableMetadataCollector, - new SerializationHeader(true, cfs.metadata, columns, stats), - txn)); - } - catch (Throwable t) - { - if (txn != null) - txn.close(); - throw t; - } + txn = LifecycleTransaction.offline(OperationType.FLUSH); + MetadataCollector sstableMetadataCollector = new MetadataCollector(cfs.metadata.comparator) + .commitLogLowerBound(commitLogLowerBound.get()) + .commitLogUpperBound(commitLogUpperBound.get()); + + return new SSTableTxnWriter(txn, + cfs.createSSTableMultiWriter(Descriptor.fromFilename(filename), + (long) partitions.size(), + ActiveRepairService.UNREPAIRED_SSTABLE, + sstableMetadataCollector, + new SerializationHeader(true, cfs.metadata, columns, stats), + txn)); + } + catch (Throwable t) + { + if (txn != null) + txn.close(); + throw t; } } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java index 7169b2f6db..f45a47aa43 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReplayer.java @@ -35,7 +35,6 @@ import com.google.common.base.Throwables; import com.google.common.collect.HashMultimap; import com.google.common.collect.Iterables; import com.google.common.collect.Multimap; -import com.google.common.collect.Ordering; import com.google.common.util.concurrent.Uninterruptibles; import org.apache.commons.lang3.StringUtils; @@ -48,7 +47,6 @@ import org.apache.cassandra.config.Schema; import org.apache.cassandra.db.*; import org.apache.cassandra.db.rows.SerializationHelper; import org.apache.cassandra.db.partitions.PartitionUpdate; -import org.apache.cassandra.db.lifecycle.SSTableSet; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.util.FileSegmentInputStream; import org.apache.cassandra.io.util.RebufferingInputStream; @@ -76,7 +74,7 @@ public class CommitLogReplayer private final List> futures; private final Map invalidMutations; private final AtomicInteger replayedCount; - private final Map cfPositions; + private final Map cfPersisted; private final ReplayPosition globalPosition; private final CRC32 checksum; private byte[] buffer; @@ -85,7 +83,7 @@ public class CommitLogReplayer private final ReplayFilter replayFilter; private final CommitLogArchiver archiver; - CommitLogReplayer(CommitLog commitLog, ReplayPosition globalPosition, Map cfPositions, ReplayFilter replayFilter) + CommitLogReplayer(CommitLog commitLog, ReplayPosition globalPosition, Map cfPersisted, ReplayFilter replayFilter) { this.keyspacesRecovered = new NonBlockingHashSet(); this.futures = new ArrayList>(); @@ -95,7 +93,7 @@ public class CommitLogReplayer // count the number of replayed mutation. We don't really care about atomicity, but we need it to be a reference. this.replayedCount = new AtomicInteger(); this.checksum = new CRC32(); - this.cfPositions = cfPositions; + this.cfPersisted = cfPersisted; this.globalPosition = globalPosition; this.replayFilter = replayFilter; this.archiver = commitLog.archiver; @@ -104,16 +102,11 @@ public class CommitLogReplayer public static CommitLogReplayer construct(CommitLog commitLog) { // compute per-CF and global replay positions - Map cfPositions = new HashMap(); - Ordering replayPositionOrdering = Ordering.from(ReplayPosition.comparator); + Map cfPersisted = new HashMap<>(); ReplayFilter replayFilter = ReplayFilter.create(); + ReplayPosition globalPosition = null; for (ColumnFamilyStore cfs : ColumnFamilyStore.all()) { - // it's important to call RP.gRP per-cf, before aggregating all the positions w/ the Ordering.min call - // below: gRP will return NONE if there are no flushed sstables, which is important to have in the - // list (otherwise we'll just start replay from the first flush position that we do have, which is not correct). - ReplayPosition rp = ReplayPosition.getReplayPosition(cfs.getSSTables(SSTableSet.CANONICAL)); - // but, if we've truncated the cf in question, then we need to need to start replay after the truncation ReplayPosition truncatedAt = SystemKeyspace.getTruncatedPosition(cfs.metadata.cfId); if (truncatedAt != null) @@ -131,19 +124,21 @@ public class CommitLogReplayer cfs.metadata.ksName, cfs.metadata.cfName); SystemKeyspace.removeTruncationRecord(cfs.metadata.cfId); + truncatedAt = null; } } - else - { - rp = replayPositionOrdering.max(Arrays.asList(rp, truncatedAt)); - } } - cfPositions.put(cfs.metadata.cfId, rp); + ReplayPosition.ReplayFilter filter = new ReplayPosition.ReplayFilter(cfs.getSSTables(), truncatedAt); + if (!filter.isEmpty()) + cfPersisted.put(cfs.metadata.cfId, filter); + else + globalPosition = ReplayPosition.NONE; // if we have no ranges for this CF, we must replay everything and filter } - ReplayPosition globalPosition = replayPositionOrdering.min(cfPositions.values()); - logger.trace("Global replay position is {} from columnfamilies {}", globalPosition, FBUtilities.toString(cfPositions)); - return new CommitLogReplayer(commitLog, globalPosition, cfPositions, replayFilter); + if (globalPosition == null) + globalPosition = ReplayPosition.firstNotCovered(cfPersisted.values()); + logger.debug("Global replay position is {} from columnfamilies {}", globalPosition, FBUtilities.toString(cfPersisted)); + return new CommitLogReplayer(commitLog, globalPosition, cfPersisted, replayFilter); } public void recover(File[] clogs) throws IOException @@ -290,6 +285,19 @@ public class CommitLogReplayer } } + /** + * consult the known-persisted ranges for our sstables; + * if the position is covered by one of them it does not need to be replayed + * + * @return true iff replay is necessary + */ + private boolean shouldReplay(UUID cfId, ReplayPosition position) + { + ReplayPosition.ReplayFilter filter = cfPersisted.get(cfId); + return filter == null || filter.shouldReplay(position); + } + + @SuppressWarnings("resource") public void recover(File file, boolean tolerateTruncation) throws IOException { CommitLogDescriptor desc = CommitLogDescriptor.fromFileName(file.getName()); @@ -508,7 +516,7 @@ public class CommitLogReplayer mutationStart, errorContext); continue; } - replayMutation(buffer, serializedSize, reader.getFilePointer(), desc); + replayMutation(buffer, serializedSize, (int) reader.getFilePointer(), desc); } return true; } @@ -517,7 +525,7 @@ public class CommitLogReplayer * Deserializes and replays a commit log entry. */ void replayMutation(byte[] inputBuffer, int size, - final long entryLocation, final CommitLogDescriptor desc) throws IOException + final int entryLocation, final CommitLogDescriptor desc) throws IOException { final Mutation mutation; @@ -589,11 +597,9 @@ public class CommitLogReplayer if (Schema.instance.getCF(update.metadata().cfId) == null) continue; // dropped - ReplayPosition rp = cfPositions.get(update.metadata().cfId); - // replay if current segment is newer than last flushed one or, // if it is the last known segment, if we are after the replay position - if (desc.id > rp.segment || (desc.id == rp.segment && entryLocation > rp.position)) + if (shouldReplay(update.metadata().cfId, new ReplayPosition(desc.id, entryLocation))) { if (newMutation == null) newMutation = new Mutation(mutation.getKeyspaceName(), mutation.key()); diff --git a/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java b/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java index 28416f3dd2..0b21763af8 100644 --- a/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java +++ b/src/java/org/apache/cassandra/db/commitlog/ReplayPosition.java @@ -18,10 +18,10 @@ package org.apache.cassandra.db.commitlog; import java.io.IOException; -import java.util.Comparator; +import java.util.Map; +import java.util.NavigableMap; +import java.util.TreeMap; -import com.google.common.base.Function; -import com.google.common.collect.Iterables; import com.google.common.collect.Ordering; import org.apache.cassandra.db.TypeSizes; @@ -35,46 +35,78 @@ public class ReplayPosition implements Comparable public static final ReplayPositionSerializer serializer = new ReplayPositionSerializer(); // NONE is used for SSTables that are streamed from other nodes and thus have no relationship - // with our local commitlog. The values satisfy the critera that + // with our local commitlog. The values satisfy the criteria that // - no real commitlog segment will have the given id // - it will sort before any real replayposition, so it will be effectively ignored by getReplayPosition public static final ReplayPosition NONE = new ReplayPosition(-1, 0); - /** - * Convenience method to compute the replay position for a group of SSTables. - * @param sstables - * @return the most recent (highest) replay position - */ - public static ReplayPosition getReplayPosition(Iterable sstables) - { - if (Iterables.isEmpty(sstables)) - return NONE; - - Function f = new Function() - { - public ReplayPosition apply(SSTableReader sstable) - { - return sstable.getReplayPosition(); - } - }; - Ordering ordering = Ordering.from(ReplayPosition.comparator); - return ordering.max(Iterables.transform(sstables, f)); - } - - public final long segment; public final int position; - public static final Comparator comparator = new Comparator() + /** + * A filter of known safe-to-discard commit log replay positions, based on + * the range covered by on disk sstables and those prior to the most recent truncation record + */ + public static class ReplayFilter { - public int compare(ReplayPosition o1, ReplayPosition o2) + final NavigableMap persisted = new TreeMap<>(); + public ReplayFilter(Iterable onDisk, ReplayPosition truncatedAt) { - if (o1.segment != o2.segment) - return Long.compare(o1.segment, o2.segment); - - return Integer.compare(o1.position, o2.position); + for (SSTableReader reader : onDisk) + { + ReplayPosition start = reader.getSSTableMetadata().commitLogLowerBound; + ReplayPosition end = reader.getSSTableMetadata().commitLogUpperBound; + add(persisted, start, end); + } + if (truncatedAt != null) + add(persisted, ReplayPosition.NONE, truncatedAt); } - }; + + private static void add(NavigableMap ranges, ReplayPosition start, ReplayPosition end) + { + // extend ourselves to cover any ranges we overlap + // record directly preceding our end may extend past us, so take the max of our end and its + Map.Entry extend = ranges.floorEntry(end); + if (extend != null && extend.getValue().compareTo(end) > 0) + end = extend.getValue(); + + // record directly preceding our start may extend into us; if it does, we take it as our start + extend = ranges.lowerEntry(start); + if (extend != null && extend.getValue().compareTo(start) >= 0) + start = extend.getKey(); + + ranges.subMap(start, end).clear(); + ranges.put(start, end); + } + + public boolean shouldReplay(ReplayPosition position) + { + // replay ranges are start exclusive, end inclusive + Map.Entry range = persisted.lowerEntry(position); + return range == null || position.compareTo(range.getValue()) > 0; + } + + public boolean isEmpty() + { + return persisted.isEmpty(); + } + } + + public static ReplayPosition firstNotCovered(Iterable ranges) + { + ReplayPosition min = null; + for (ReplayFilter map : ranges) + { + ReplayPosition first = map.persisted.firstEntry().getValue(); + if (min == null) + min = first; + else + min = Ordering.natural().min(min, first); + } + if (min == null) + return NONE; + return min; + } public ReplayPosition(long segment, int position) { @@ -83,9 +115,12 @@ public class ReplayPosition implements Comparable this.position = position; } - public int compareTo(ReplayPosition other) + public int compareTo(ReplayPosition that) { - return comparator.compare(this, other); + if (this.segment != that.segment) + return Long.compare(this.segment, that.segment); + + return Integer.compare(this.position, that.position); } @Override diff --git a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java index c205d5c2ce..8928db5df3 100644 --- a/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java +++ b/src/java/org/apache/cassandra/db/compaction/AbstractCompactionStrategy.java @@ -241,9 +241,6 @@ public abstract class AbstractCompactionStrategy */ public void replaceFlushed(Memtable memtable, Collection sstables) { - cfs.getTracker().replaceFlushed(memtable, sstables); - if (sstables != null && !sstables.isEmpty()) - CompactionManager.instance.submitBackground(cfs); } /** diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java index 82fd872349..f1127c9bf1 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionStrategyManager.java @@ -189,9 +189,6 @@ public class CompactionStrategyManager implements INotificationConsumer public void replaceFlushed(Memtable memtable, Collection sstables) { - cfs.getTracker().replaceFlushed(memtable, sstables); - if (sstables != null && !sstables.isEmpty()) - CompactionManager.instance.submitBackground(cfs); } public int getUnleveledSSTables() diff --git a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java index c09d49ccd5..16090a116d 100644 --- a/src/java/org/apache/cassandra/db/lifecycle/Tracker.java +++ b/src/java/org/apache/cassandra/db/lifecycle/Tracker.java @@ -31,11 +31,13 @@ import com.google.common.collect.*; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.Memtable; +import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.commitlog.ReplayPosition; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.FileUtils; @@ -46,6 +48,8 @@ import org.apache.cassandra.utils.Throwables; import org.apache.cassandra.utils.concurrent.OpOrder; import static com.google.common.base.Predicates.and; +import static com.google.common.base.Predicates.in; +import static com.google.common.base.Predicates.not; import static com.google.common.collect.ImmutableSet.copyOf; import static com.google.common.collect.Iterables.filter; import static java.util.Collections.singleton; @@ -195,10 +199,12 @@ public class Tracker public void reset() { view.set(new View( - !isDummy() ? ImmutableList.of(new Memtable(cfstore)) : Collections.emptyList(), + !isDummy() ? ImmutableList.of(new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), cfstore)) + : ImmutableList.of(), ImmutableList.of(), Collections.emptyMap(), Collections.emptyMap(), + Collections.emptySet(), SSTableIntervalTree.empty())); } @@ -312,9 +318,8 @@ public class Tracker * * @return the previously active memtable */ - public Memtable switchMemtable(boolean truncating) + public Memtable switchMemtable(boolean truncating, Memtable newMemtable) { - Memtable newMemtable = new Memtable(cfstore); Pair result = apply(View.switchMemtable(newMemtable)); if (truncating) notifyRenewed(newMemtable); @@ -330,7 +335,7 @@ public class Tracker public void replaceFlushed(Memtable memtable, Collection sstables) { assert !isDummy(); - if (sstables == null || sstables.isEmpty()) + if (sstables.isEmpty()) { // sstable may be null if we flushed batchlog and nothing needed to be retained // if it's null, we don't care what state the cfstore is in, we just replace it and continue @@ -346,19 +351,49 @@ public class Tracker Throwable fail; fail = updateSizeTracking(emptySet(), sstables, null); - // TODO: if we're invalidated, should we notifyadded AND removed, or just skip both? - fail = notifyAdded(sstables, fail); - - if (!isDummy() && !cfstore.isValid()) - dropSSTables(); maybeFail(fail); } + /** + * permit compaction of the provided sstable; this translates to notifying compaction + * strategies of its existence, and potentially submitting a background task + */ + public void permitCompactionOfFlushed(Collection sstables) + { + if (sstables.isEmpty()) + return; + + apply(View.permitCompactionOfFlushed(sstables)); + + if (isDummy()) + return; + + if (cfstore.isValid()) + { + notifyAdded(sstables); + CompactionManager.instance.submitBackground(cfstore); + } + else + { + dropSSTables(); + } + } // MISCELLANEOUS public utility calls + public Set getSSTables() + { + return view.get().sstables; + } + + public Iterable getPermittedToCompact() + { + View view = this.view.get(); + return filter(view.sstables, not(in(view.premature))); + } + public Set getCompacting() { return view.get().compacting; diff --git a/src/java/org/apache/cassandra/db/lifecycle/View.java b/src/java/org/apache/cassandra/db/lifecycle/View.java index b62c7e3b61..17062b418d 100644 --- a/src/java/org/apache/cassandra/db/lifecycle/View.java +++ b/src/java/org/apache/cassandra/db/lifecycle/View.java @@ -33,6 +33,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.utils.Interval; import static com.google.common.base.Predicates.equalTo; +import static com.google.common.base.Predicates.in; import static com.google.common.base.Predicates.not; import static com.google.common.collect.ImmutableList.copyOf; import static com.google.common.collect.ImmutableList.of; @@ -40,7 +41,6 @@ import static com.google.common.collect.Iterables.all; import static com.google.common.collect.Iterables.concat; import static com.google.common.collect.Iterables.filter; import static com.google.common.collect.Iterables.transform; -import static java.util.Collections.singleton; import static org.apache.cassandra.db.lifecycle.Helpers.emptySet; import static org.apache.cassandra.db.lifecycle.Helpers.filterOut; import static org.apache.cassandra.db.lifecycle.Helpers.replace; @@ -70,6 +70,7 @@ public class View public final List flushingMemtables; final Set compacting; final Set sstables; + final Set premature; // we use a Map here so that we can easily perform identity checks as well as equality checks. // When marking compacting, we now indicate if we expect the sstables to be present (by default we do), // and we then check that not only are they all present in the live set, but that the exact instance present is @@ -79,7 +80,7 @@ public class View final SSTableIntervalTree intervalTree; - View(List liveMemtables, List flushingMemtables, Map sstables, Map compacting, SSTableIntervalTree intervalTree) + View(List liveMemtables, List flushingMemtables, Map sstables, Map compacting, Set premature, SSTableIntervalTree intervalTree) { assert liveMemtables != null; assert flushingMemtables != null; @@ -94,6 +95,7 @@ public class View this.sstables = sstablesMap.keySet(); this.compactingMap = compacting; this.compacting = compactingMap.keySet(); + this.premature = premature; this.intervalTree = intervalTree; } @@ -239,7 +241,7 @@ public class View assert all(mark, Helpers.idIn(view.sstablesMap)); return new View(view.liveMemtables, view.flushingMemtables, view.sstablesMap, replace(view.compactingMap, unmark, mark), - view.intervalTree); + view.premature, view.intervalTree); } }; } @@ -253,7 +255,7 @@ public class View public boolean apply(View view) { for (SSTableReader reader : readers) - if (view.compacting.contains(reader) || view.sstablesMap.get(reader) != reader || reader.isMarkedCompacted()) + if (view.compacting.contains(reader) || view.sstablesMap.get(reader) != reader || reader.isMarkedCompacted() || view.premature.contains(reader)) return false; return true; } @@ -270,7 +272,7 @@ public class View public View apply(View view) { Map sstableMap = replace(view.sstablesMap, remove, add); - return new View(view.liveMemtables, view.flushingMemtables, sstableMap, view.compactingMap, + return new View(view.liveMemtables, view.flushingMemtables, sstableMap, view.compactingMap, view.premature, SSTableIntervalTree.build(sstableMap.keySet())); } }; @@ -285,7 +287,7 @@ public class View { List newLive = ImmutableList.builder().addAll(view.liveMemtables).add(newMemtable).build(); assert newLive.size() == view.liveMemtables.size() + 1; - return new View(newLive, view.flushingMemtables, view.sstablesMap, view.compactingMap, view.intervalTree); + return new View(newLive, view.flushingMemtables, view.sstablesMap, view.compactingMap, view.premature, view.intervalTree); } }; } @@ -304,7 +306,7 @@ public class View filter(flushing, not(lessThan(toFlush))))); assert newLive.size() == live.size() - 1; assert newFlushing.size() == flushing.size() + 1; - return new View(newLive, newFlushing, view.sstablesMap, view.compactingMap, view.intervalTree); + return new View(newLive, newFlushing, view.sstablesMap, view.compactingMap, view.premature, view.intervalTree); } }; } @@ -319,17 +321,34 @@ public class View List flushingMemtables = copyOf(filter(view.flushingMemtables, not(equalTo(memtable)))); assert flushingMemtables.size() == view.flushingMemtables.size() - 1; - if (flushed == null) + if (flushed == null || flushed.isEmpty()) return new View(view.liveMemtables, flushingMemtables, view.sstablesMap, - view.compactingMap, view.intervalTree); + view.compactingMap, view.premature, view.intervalTree); Map sstableMap = replace(view.sstablesMap, emptySet(), flushed); - return new View(view.liveMemtables, flushingMemtables, sstableMap, view.compactingMap, + Map compactingMap = replace(view.compactingMap, emptySet(), flushed); + Set premature = replace(view.premature, emptySet(), flushed); + return new View(view.liveMemtables, flushingMemtables, sstableMap, compactingMap, premature, SSTableIntervalTree.build(sstableMap.keySet())); } }; } + static Function permitCompactionOfFlushed(final Collection readers) + { + Set expectAndRemove = ImmutableSet.copyOf(readers); + return new Function() + { + public View apply(View view) + { + Set premature = replace(view.premature, expectAndRemove, emptySet()); + Map compactingMap = replace(view.compactingMap, expectAndRemove, emptySet()); + return new View(view.liveMemtables, view.flushingMemtables, view.sstablesMap, compactingMap, premature, view.intervalTree); + } + }; + } + + private static > Predicate lessThan(final T lessThan) { return new Predicate() diff --git a/src/java/org/apache/cassandra/db/view/TableViews.java b/src/java/org/apache/cassandra/db/view/TableViews.java index 893bdd5d34..7feb67c43d 100644 --- a/src/java/org/apache/cassandra/db/view/TableViews.java +++ b/src/java/org/apache/cassandra/db/view/TableViews.java @@ -94,11 +94,11 @@ public class TableViews extends AbstractCollection viewCfs.dumpMemtable(); } - public void truncateBlocking(long truncatedAt) + public void truncateBlocking(ReplayPosition replayAfter, long truncatedAt) { for (ColumnFamilyStore viewCfs : allViewsCfs()) { - ReplayPosition replayAfter = viewCfs.discardSSTables(truncatedAt); + viewCfs.discardSSTables(truncatedAt); SystemKeyspace.saveTruncationRecord(viewCfs, truncatedAt, replayAfter); } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java index b9561ec631..9bb1767bf8 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java @@ -1889,11 +1889,6 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted= 0 && version.compareTo("ma") < 0) + || version.compareTo("mb") >= 0; } @Override @@ -243,6 +248,11 @@ public class BigFormat implements SSTableFormat return newFileName; } + public boolean hasCommitLogLowerBound() + { + return hasCommitLogLowerBound; + } + @Override public boolean storeRows() { diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/CompactionMetadata.java b/src/java/org/apache/cassandra/io/sstable/metadata/CompactionMetadata.java index 4f4c04b0a1..ef3453a9d7 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/CompactionMetadata.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/CompactionMetadata.java @@ -74,12 +74,21 @@ public class CompactionMetadata extends MetadataComponent { public int serializedSize(Version version, CompactionMetadata component) throws IOException { + int sz = 0; + if (version.hasCompactionAncestors()) + { // write empty ancestor marker + sz = 4; + } byte[] serializedCardinality = component.cardinalityEstimator.getBytes(); - return TypeSizes.sizeof(serializedCardinality.length) + serializedCardinality.length; + return TypeSizes.sizeof(serializedCardinality.length) + serializedCardinality.length + sz; } public void serialize(Version version, CompactionMetadata component, DataOutputPlus out) throws IOException { + if (version.hasCompactionAncestors()) + { // write empty ancestor marker + out.writeInt(0); + } ByteBufferUtil.writeWithLength(component.cardinalityEstimator.getBytes(), out); } diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/LegacyMetadataSerializer.java b/src/java/org/apache/cassandra/io/sstable/metadata/LegacyMetadataSerializer.java index 2ed845cdf6..456152098c 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/LegacyMetadataSerializer.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/LegacyMetadataSerializer.java @@ -55,7 +55,7 @@ public class LegacyMetadataSerializer extends MetadataSerializer EstimatedHistogram.serializer.serialize(stats.estimatedPartitionSize, out); EstimatedHistogram.serializer.serialize(stats.estimatedColumnCount, out); - ReplayPosition.serializer.serialize(stats.replayPosition, out); + ReplayPosition.serializer.serialize(stats.commitLogUpperBound, out); out.writeLong(stats.minTimestamp); out.writeLong(stats.maxTimestamp); out.writeInt(stats.maxLocalDeletionTime); @@ -71,6 +71,8 @@ public class LegacyMetadataSerializer extends MetadataSerializer out.writeInt(stats.maxClusteringValues.size()); for (ByteBuffer value : stats.maxClusteringValues) ByteBufferUtil.writeWithShortLength(value, out); + if (version.hasCommitLogLowerBound()) + ReplayPosition.serializer.serialize(stats.commitLogLowerBound, out); } /** @@ -92,7 +94,8 @@ public class LegacyMetadataSerializer extends MetadataSerializer { EstimatedHistogram partitionSizes = EstimatedHistogram.serializer.deserialize(in); EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in); - ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in); + ReplayPosition commitLogLowerBound = ReplayPosition.NONE; + ReplayPosition commitLogUpperBound = ReplayPosition.serializer.deserialize(in); long minTimestamp = in.readLong(); long maxTimestamp = in.readLong(); int maxLocalDeletionTime = in.readInt(); @@ -116,6 +119,9 @@ public class LegacyMetadataSerializer extends MetadataSerializer for (int i = 0; i < colCount; i++) maxColumnNames.add(ByteBufferUtil.readWithShortLength(in)); + if (descriptor.version.hasCommitLogLowerBound()) + commitLogLowerBound = ReplayPosition.serializer.deserialize(in); + if (types.contains(MetadataType.VALIDATION)) components.put(MetadataType.VALIDATION, new ValidationMetadata(partitioner, bloomFilterFPChance)); @@ -123,7 +129,8 @@ public class LegacyMetadataSerializer extends MetadataSerializer components.put(MetadataType.STATS, new StatsMetadata(partitionSizes, columnCounts, - replayPosition, + commitLogLowerBound, + commitLogUpperBound, minTimestamp, maxTimestamp, Integer.MAX_VALUE, diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java index c2b0cafe39..53cf0b0b88 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java @@ -27,6 +27,7 @@ import java.util.Map; import java.util.Set; import com.google.common.collect.Maps; +import com.google.common.collect.Ordering; import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus; import com.clearspring.analytics.stream.cardinality.ICardinality; @@ -35,7 +36,6 @@ import org.apache.cassandra.db.commitlog.ReplayPosition; import org.apache.cassandra.db.marshal.AbstractType; import org.apache.cassandra.db.partitions.PartitionStatisticsCollector; import org.apache.cassandra.db.rows.Cell; -import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.SSTable; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.service.ActiveRepairService; @@ -70,6 +70,7 @@ public class MetadataCollector implements PartitionStatisticsCollector return new StatsMetadata(defaultPartitionSizeHistogram(), defaultCellPerPartitionCountHistogram(), ReplayPosition.NONE, + ReplayPosition.NONE, Long.MIN_VALUE, Long.MAX_VALUE, Integer.MAX_VALUE, @@ -90,7 +91,8 @@ public class MetadataCollector implements PartitionStatisticsCollector protected EstimatedHistogram estimatedPartitionSize = defaultPartitionSizeHistogram(); // TODO: cound the number of row per partition (either with the number of cells, or instead) protected EstimatedHistogram estimatedCellPerPartitionCount = defaultCellPerPartitionCountHistogram(); - protected ReplayPosition replayPosition = ReplayPosition.NONE; + protected ReplayPosition commitLogLowerBound = ReplayPosition.NONE; + protected ReplayPosition commitLogUpperBound = ReplayPosition.NONE; protected final MinMaxLongTracker timestampTracker = new MinMaxLongTracker(); protected final MinMaxIntTracker localDeletionTimeTracker = new MinMaxIntTracker(Cell.NO_DELETION_TIME, Cell.NO_DELETION_TIME); protected final MinMaxIntTracker ttlTracker = new MinMaxIntTracker(Cell.NO_TTL, Cell.NO_TTL); @@ -124,7 +126,23 @@ public class MetadataCollector implements PartitionStatisticsCollector { this(comparator); - replayPosition(ReplayPosition.getReplayPosition(sstables)); + ReplayPosition min = null, max = null; + for (SSTableReader sstable : sstables) + { + if (min == null) + { + min = sstable.getSSTableMetadata().commitLogLowerBound; + max = sstable.getSSTableMetadata().commitLogUpperBound; + } + else + { + min = Ordering.natural().min(min, sstable.getSSTableMetadata().commitLogLowerBound); + max = Ordering.natural().max(max, sstable.getSSTableMetadata().commitLogUpperBound); + } + } + + commitLogLowerBound(min); + commitLogUpperBound(max); sstableLevel(level); } @@ -211,9 +229,15 @@ public class MetadataCollector implements PartitionStatisticsCollector ttlTracker.update(newTTL); } - public MetadataCollector replayPosition(ReplayPosition replayPosition) + public MetadataCollector commitLogLowerBound(ReplayPosition commitLogLowerBound) { - this.replayPosition = replayPosition; + this.commitLogLowerBound = commitLogLowerBound; + return this; + } + + public MetadataCollector commitLogUpperBound(ReplayPosition commitLogUpperBound) + { + this.commitLogUpperBound = commitLogUpperBound; return this; } @@ -278,7 +302,8 @@ public class MetadataCollector implements PartitionStatisticsCollector components.put(MetadataType.VALIDATION, new ValidationMetadata(partitioner, bloomFilterFPChance)); components.put(MetadataType.STATS, new StatsMetadata(estimatedPartitionSize, estimatedCellPerPartitionCount, - replayPosition, + commitLogLowerBound, + commitLogUpperBound, timestampTracker.min(), timestampTracker.max(), localDeletionTimeTracker.min(), diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java index 824bf8b5bf..07e35bb72b 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java @@ -42,7 +42,8 @@ public class StatsMetadata extends MetadataComponent public final EstimatedHistogram estimatedPartitionSize; public final EstimatedHistogram estimatedColumnCount; - public final ReplayPosition replayPosition; + public final ReplayPosition commitLogLowerBound; + public final ReplayPosition commitLogUpperBound; public final long minTimestamp; public final long maxTimestamp; public final int minLocalDeletionTime; @@ -61,7 +62,8 @@ public class StatsMetadata extends MetadataComponent public StatsMetadata(EstimatedHistogram estimatedPartitionSize, EstimatedHistogram estimatedColumnCount, - ReplayPosition replayPosition, + ReplayPosition commitLogLowerBound, + ReplayPosition commitLogUpperBound, long minTimestamp, long maxTimestamp, int minLocalDeletionTime, @@ -80,7 +82,8 @@ public class StatsMetadata extends MetadataComponent { this.estimatedPartitionSize = estimatedPartitionSize; this.estimatedColumnCount = estimatedColumnCount; - this.replayPosition = replayPosition; + this.commitLogLowerBound = commitLogLowerBound; + this.commitLogUpperBound = commitLogUpperBound; this.minTimestamp = minTimestamp; this.maxTimestamp = maxTimestamp; this.minLocalDeletionTime = minLocalDeletionTime; @@ -131,7 +134,8 @@ public class StatsMetadata extends MetadataComponent { return new StatsMetadata(estimatedPartitionSize, estimatedColumnCount, - replayPosition, + commitLogLowerBound, + commitLogUpperBound, minTimestamp, maxTimestamp, minLocalDeletionTime, @@ -153,7 +157,8 @@ public class StatsMetadata extends MetadataComponent { return new StatsMetadata(estimatedPartitionSize, estimatedColumnCount, - replayPosition, + commitLogLowerBound, + commitLogUpperBound, minTimestamp, maxTimestamp, minLocalDeletionTime, @@ -181,7 +186,8 @@ public class StatsMetadata extends MetadataComponent return new EqualsBuilder() .append(estimatedPartitionSize, that.estimatedPartitionSize) .append(estimatedColumnCount, that.estimatedColumnCount) - .append(replayPosition, that.replayPosition) + .append(commitLogLowerBound, that.commitLogLowerBound) + .append(commitLogUpperBound, that.commitLogUpperBound) .append(minTimestamp, that.minTimestamp) .append(maxTimestamp, that.maxTimestamp) .append(minLocalDeletionTime, that.minLocalDeletionTime) @@ -206,7 +212,8 @@ public class StatsMetadata extends MetadataComponent return new HashCodeBuilder() .append(estimatedPartitionSize) .append(estimatedColumnCount) - .append(replayPosition) + .append(commitLogLowerBound) + .append(commitLogUpperBound) .append(minTimestamp) .append(maxTimestamp) .append(minLocalDeletionTime) @@ -232,7 +239,7 @@ public class StatsMetadata extends MetadataComponent int size = 0; size += EstimatedHistogram.serializer.serializedSize(component.estimatedPartitionSize); size += EstimatedHistogram.serializer.serializedSize(component.estimatedColumnCount); - size += ReplayPosition.serializer.serializedSize(component.replayPosition); + size += ReplayPosition.serializer.serializedSize(component.commitLogUpperBound); if (version.storeRows()) size += 8 + 8 + 4 + 4 + 4 + 4 + 8 + 8; // mix/max timestamp(long), min/maxLocalDeletionTime(int), min/max TTL, compressionRatio(double), repairedAt (long) else @@ -250,6 +257,8 @@ public class StatsMetadata extends MetadataComponent size += TypeSizes.sizeof(component.hasLegacyCounterShards); if (version.storeRows()) size += 8 + 8; // totalColumnsSet, totalRows + if (version.hasCommitLogLowerBound()) + size += ReplayPosition.serializer.serializedSize(component.commitLogLowerBound); return size; } @@ -257,7 +266,7 @@ public class StatsMetadata extends MetadataComponent { EstimatedHistogram.serializer.serialize(component.estimatedPartitionSize, out); EstimatedHistogram.serializer.serialize(component.estimatedColumnCount, out); - ReplayPosition.serializer.serialize(component.replayPosition, out); + ReplayPosition.serializer.serialize(component.commitLogUpperBound, out); out.writeLong(component.minTimestamp); out.writeLong(component.maxTimestamp); if (version.storeRows()) @@ -285,13 +294,17 @@ public class StatsMetadata extends MetadataComponent out.writeLong(component.totalColumnsSet); out.writeLong(component.totalRows); } + + if (version.hasCommitLogLowerBound()) + ReplayPosition.serializer.serialize(component.commitLogLowerBound, out); } public StatsMetadata deserialize(Version version, DataInputPlus in) throws IOException { EstimatedHistogram partitionSizes = EstimatedHistogram.serializer.deserialize(in); EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in); - ReplayPosition replayPosition = ReplayPosition.serializer.deserialize(in); + ReplayPosition commitLogLowerBound = ReplayPosition.NONE, commitLogUpperBound; + commitLogUpperBound = ReplayPosition.serializer.deserialize(in); long minTimestamp = in.readLong(); long maxTimestamp = in.readLong(); // We use MAX_VALUE as that's the default value for "no deletion time" @@ -323,9 +336,13 @@ public class StatsMetadata extends MetadataComponent long totalColumnsSet = version.storeRows() ? in.readLong() : -1L; long totalRows = version.storeRows() ? in.readLong() : -1L; + if (version.hasCommitLogLowerBound()) + commitLogLowerBound = ReplayPosition.serializer.deserialize(in); + return new StatsMetadata(partitionSizes, columnCounts, - replayPosition, + commitLogLowerBound, + commitLogUpperBound, minTimestamp, maxTimestamp, minLocalDeletionTime, diff --git a/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java b/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java index 3728d34808..420b802591 100644 --- a/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java +++ b/src/java/org/apache/cassandra/tools/SSTableMetadataViewer.java @@ -70,7 +70,8 @@ public class SSTableMetadataViewer out.printf("Estimated droppable tombstones: %s%n", stats.getEstimatedDroppableTombstoneRatio((int) (System.currentTimeMillis() / 1000))); out.printf("SSTable Level: %d%n", stats.sstableLevel); out.printf("Repaired at: %d%n", stats.repairedAt); - out.println(stats.replayPosition); + out.printf("Minimum replay position: %s\n", stats.commitLogLowerBound); + out.printf("Maximum replay position: %s\n", stats.commitLogUpperBound); out.println("Estimated tombstone drop times:"); for (Map.Entry entry : stats.estimatedTombstoneDropTime.getAsMap().entrySet()) { diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..e32d12f780 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Data.db new file mode 100644 index 0000000000..790cbe16f8 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..66d1d34594 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +163939259 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Index.db new file mode 100644 index 0000000000..38d05374a1 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Statistics.db new file mode 100644 index 0000000000..a53b7c1149 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..e3d437799a Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Data.db new file mode 100644 index 0000000000..c5aa7ce248 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..4e3bf8955a --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +3387396134 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Index.db new file mode 100644 index 0000000000..54afd0cd9d Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Statistics.db new file mode 100644 index 0000000000..9d9b085252 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_compact/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..5a34d6e727 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Data.db new file mode 100644 index 0000000000..55ad88d858 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..86240b529a --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +346435755 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Index.db new file mode 100644 index 0000000000..d1a401b309 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Statistics.db new file mode 100644 index 0000000000..4f3f45a95c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..32fa731a8b Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Data.db new file mode 100644 index 0000000000..ca10874ffb Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..98f578404a --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +3740016396 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Index.db new file mode 100644 index 0000000000..3661d77233 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Statistics.db new file mode 100644 index 0000000000..bb1e3fb721 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_clust_counter_compact/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..288663f2fb Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Data.db new file mode 100644 index 0000000000..6a2f28e9d8 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..fb255bdf97 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +655951031 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Index.db new file mode 100644 index 0000000000..b3094bffba Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Statistics.db new file mode 100644 index 0000000000..55dcdce665 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..adb7fc4579 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Data.db new file mode 100644 index 0000000000..5cd0481040 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..7d2bdbea85 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +3883019031 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Index.db new file mode 100644 index 0000000000..56f29df1cb Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Statistics.db new file mode 100644 index 0000000000..dff24fe7c6 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_compact/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..19d25c950f Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Data.db new file mode 100644 index 0000000000..77a13948df Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..172b695a88 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +2876949266 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Index.db new file mode 100644 index 0000000000..59e65cab85 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Statistics.db new file mode 100644 index 0000000000..4c6771550c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-CompressionInfo.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-CompressionInfo.db new file mode 100644 index 0000000000..56c95a8a36 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-CompressionInfo.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Data.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Data.db new file mode 100644 index 0000000000..00a7a3968e Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Data.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Digest.crc32 b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Digest.crc32 new file mode 100644 index 0000000000..f9f2fffbcb --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Digest.crc32 @@ -0,0 +1 @@ +1214766167 \ No newline at end of file diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Filter.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Filter.db new file mode 100644 index 0000000000..2e1d5d29ca Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Filter.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Index.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Index.db new file mode 100644 index 0000000000..d094f73e21 Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Index.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Statistics.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Statistics.db new file mode 100644 index 0000000000..33c33f201e Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Statistics.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Summary.db b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Summary.db new file mode 100644 index 0000000000..9b24e0450c Binary files /dev/null and b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-Summary.db differ diff --git a/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-TOC.txt b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-TOC.txt new file mode 100644 index 0000000000..fe9581f576 --- /dev/null +++ b/test/data/legacy-sstables/mb/legacy_tables/legacy_mb_simple_counter_compact/mb-1-big-TOC.txt @@ -0,0 +1,8 @@ +CompressionInfo.db +Summary.db +TOC.txt +Digest.crc32 +Filter.db +Statistics.db +Data.db +Index.db diff --git a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java index be3abb494d..31dea3e67f 100644 --- a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java +++ b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java @@ -452,7 +452,7 @@ public class CommitLogStressTest int cells = 0; @Override - void replayMutation(byte[] inputBuffer, int size, final long entryLocation, final CommitLogDescriptor desc) + void replayMutation(byte[] inputBuffer, int size, final int entryLocation, final CommitLogDescriptor desc) { if (desc.id < discardedPos.segment) { diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java index fea83c1ce5..e690785682 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java @@ -59,7 +59,7 @@ public class CommitLogTestReplayer extends CommitLogReplayer } @Override - void replayMutation(byte[] inputBuffer, int size, final long entryLocation, final CommitLogDescriptor desc) + void replayMutation(byte[] inputBuffer, int size, final int entryLocation, final CommitLogDescriptor desc) { RebufferingInputStream bufIn = new DataInputBuffer(inputBuffer, 0, size); Mutation mutation; diff --git a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java index 7b9b19c1f1..b8de711367 100644 --- a/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java +++ b/test/unit/org/apache/cassandra/db/lifecycle/TrackerTest.java @@ -38,6 +38,7 @@ import org.apache.cassandra.MockSchema; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Memtable; +import org.apache.cassandra.db.commitlog.CommitLog; import org.apache.cassandra.db.commitlog.ReplayPosition; import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.io.sstable.format.SSTableReader; @@ -265,15 +266,15 @@ public class TrackerTest Tracker tracker = cfs.getTracker(); tracker.subscribe(listener); - Memtable prev1 = tracker.switchMemtable(true); + Memtable prev1 = tracker.switchMemtable(true, new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), cfs)); OpOrder.Group write1 = cfs.keyspace.writeOrder.getCurrent(); OpOrder.Barrier barrier1 = cfs.keyspace.writeOrder.newBarrier(); - prev1.setDiscarding(barrier1, new AtomicReference()); + prev1.setDiscarding(barrier1, new AtomicReference<>(CommitLog.instance.getContext())); barrier1.issue(); - Memtable prev2 = tracker.switchMemtable(false); + Memtable prev2 = tracker.switchMemtable(false, new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), cfs)); OpOrder.Group write2 = cfs.keyspace.writeOrder.getCurrent(); OpOrder.Barrier barrier2 = cfs.keyspace.writeOrder.newBarrier(); - prev2.setDiscarding(barrier2, new AtomicReference()); + prev2.setDiscarding(barrier2, new AtomicReference<>(CommitLog.instance.getContext())); barrier2.issue(); Memtable cur = tracker.getView().getCurrentMemtable(); OpOrder.Group writecur = cfs.keyspace.writeOrder.getCurrent(); @@ -292,13 +293,16 @@ public class TrackerTest Assert.assertTrue(tracker.getView().flushingMemtables.contains(prev1)); Assert.assertEquals(2, tracker.getView().flushingMemtables.size()); - tracker.replaceFlushed(prev1, null); + tracker.replaceFlushed(prev1, Collections.emptyList()); Assert.assertEquals(1, tracker.getView().flushingMemtables.size()); Assert.assertTrue(tracker.getView().flushingMemtables.contains(prev2)); SSTableReader reader = MockSchema.sstable(0, 10, false, cfs); tracker.replaceFlushed(prev2, Collections.singleton(reader)); Assert.assertEquals(1, tracker.getView().sstables.size()); + Assert.assertEquals(1, tracker.getView().premature.size()); + tracker.permitCompactionOfFlushed(singleton(reader)); + Assert.assertEquals(0, tracker.getView().premature.size()); Assert.assertEquals(1, listener.received.size()); Assert.assertEquals(singleton(reader), ((SSTableAddedNotification) listener.received.get(0)).added); listener.received.clear(); @@ -310,18 +314,17 @@ public class TrackerTest tracker = cfs.getTracker(); listener = new MockListener(false); tracker.subscribe(listener); - prev1 = tracker.switchMemtable(false); + prev1 = tracker.switchMemtable(false, new Memtable(new AtomicReference<>(CommitLog.instance.getContext()), cfs)); tracker.markFlushing(prev1); reader = MockSchema.sstable(0, 10, true, cfs); cfs.invalidate(false); tracker.replaceFlushed(prev1, Collections.singleton(reader)); + tracker.permitCompactionOfFlushed(Collections.singleton(reader)); Assert.assertEquals(0, tracker.getView().sstables.size()); Assert.assertEquals(0, tracker.getView().flushingMemtables.size()); Assert.assertEquals(0, cfs.metric.liveDiskSpaceUsed.getCount()); - Assert.assertEquals(3, listener.received.size()); - Assert.assertEquals(singleton(reader), ((SSTableAddedNotification) listener.received.get(0)).added); - Assert.assertTrue(listener.received.get(1) instanceof SSTableDeletingNotification); - Assert.assertEquals(1, ((SSTableListChangedNotification) listener.received.get(2)).removed.size()); + Assert.assertEquals(reader, (((SSTableDeletingNotification) listener.received.get(0)).deleting)); + Assert.assertEquals(1, ((SSTableListChangedNotification) listener.received.get(1)).removed.size()); DatabaseDescriptor.setIncrementalBackupsEnabled(backups); } diff --git a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java index 523c203c91..8a5e00e1bc 100644 --- a/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java +++ b/test/unit/org/apache/cassandra/db/lifecycle/ViewTest.java @@ -212,6 +212,6 @@ public class ViewTest for (int i = 0 ; i < sstableCount ; i++) sstables.add(MockSchema.sstable(i, cfs)); return new View(ImmutableList.copyOf(memtables), Collections.emptyList(), Helpers.identityMap(sstables), - Collections.emptyMap(), SSTableIntervalTree.build(sstables)); + Collections.emptyMap(), Collections.emptySet(), SSTableIntervalTree.build(sstables)); } } diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java index 74b5c7474b..802d9c8b2b 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java @@ -120,7 +120,7 @@ public class CompressedRandomAccessReaderTest final String filename = f.getAbsolutePath(); try(ChannelProxy channel = new ChannelProxy(f)) { - MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)).replayPosition(null); + MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)); try(SequentialWriter writer = compressed ? new CompressedSequentialWriter(f, filename + ".metadata", CompressionParams.snappy(), sstableMetadataCollector) : SequentialWriter.open(f)) @@ -190,7 +190,7 @@ public class CompressedRandomAccessReaderTest assertTrue(file.createNewFile()); assertTrue(metadata.createNewFile()); - MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)).replayPosition(null); + MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)); try (SequentialWriter writer = new CompressedSequentialWriter(file, metadata.getPath(), CompressionParams.snappy(), sstableMetadataCollector)) { writer.write(CONTENT.getBytes()); diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java index 9b09f0b488..e045aadf2f 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java @@ -88,7 +88,7 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest try { - MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(Arrays.>asList(BytesType.instance))).replayPosition(null); + MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(Arrays.>asList(BytesType.instance))); byte[] dataPre = new byte[bytesToTest]; byte[] rawPost = new byte[bytesToTest]; diff --git a/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java b/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java index 4b9a7692f6..62228e3ed7 100644 --- a/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java @@ -71,7 +71,7 @@ public class LegacySSTableTest * See {@link #testGenerateSstables()} to generate sstables. * Take care on commit as you need to add the sstable files using {@code git add -f} */ - public static final String[] legacyVersions = {"ma", "la", "ka", "jb"}; + public static final String[] legacyVersions = {"mb", "ma", "la", "ka", "jb"}; // 1200 chars static final String longString = "0123456789012345678901234567890123456789012345678901234567890123456789012345678901234567890123456789" + diff --git a/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java b/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java index 4c108824ce..93365ef0c4 100644 --- a/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java @@ -19,6 +19,7 @@ package org.apache.cassandra.io.sstable.metadata; import java.io.File; +import java.io.FileNotFoundException; import java.io.FileOutputStream; import java.io.IOException; import java.util.Collections; @@ -30,10 +31,12 @@ import org.junit.Test; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.CFMetaData; import org.apache.cassandra.db.SerializationHeader; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.commitlog.ReplayPosition; import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.format.Version; import org.apache.cassandra.io.sstable.format.big.BigFormat; import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus; import org.apache.cassandra.io.util.DataOutputStreamPlus; @@ -46,24 +49,10 @@ public class MetadataSerializerTest @Test public void testSerialization() throws IOException { - - CFMetaData cfm = SchemaLoader.standardCFMD("ks1", "cf1"); - - ReplayPosition rp = new ReplayPosition(11L, 12); - - MetadataCollector collector = new MetadataCollector(cfm.comparator).replayPosition(rp); - - String partitioner = RandomPartitioner.class.getCanonicalName(); - double bfFpChance = 0.1; - Map originalMetadata = collector.finalizeMetadata(partitioner, bfFpChance, 0, SerializationHeader.make(cfm, Collections.EMPTY_LIST)); + Map originalMetadata = constructMetadata(); MetadataSerializer serializer = new MetadataSerializer(); - // Serialize to tmp file - File statsFile = File.createTempFile(Component.STATS.name, null); - try (DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(statsFile))) - { - serializer.serialize(originalMetadata, out, BigFormat.latestVersion); - } + File statsFile = serialize(originalMetadata, serializer, BigFormat.latestVersion); Descriptor desc = new Descriptor( statsFile.getParentFile(), "", "", 0); try (RandomAccessReader in = RandomAccessReader.open(statsFile)) @@ -76,4 +65,72 @@ public class MetadataSerializerTest } } } + + public File serialize(Map metadata, MetadataSerializer serializer, Version version) + throws IOException, FileNotFoundException + { + // Serialize to tmp file + File statsFile = File.createTempFile(Component.STATS.name, null); + try (DataOutputStreamPlus out = new BufferedDataOutputStreamPlus(new FileOutputStream(statsFile))) + { + serializer.serialize(metadata, out, version); + } + return statsFile; + } + + public Map constructMetadata() + { + ReplayPosition club = new ReplayPosition(11L, 12); + ReplayPosition cllb = new ReplayPosition(9L, 12); + + CFMetaData cfm = SchemaLoader.standardCFMD("ks1", "cf1"); + MetadataCollector collector = new MetadataCollector(cfm.comparator) + .commitLogLowerBound(cllb) + .commitLogUpperBound(club); + + String partitioner = RandomPartitioner.class.getCanonicalName(); + double bfFpChance = 0.1; + Map originalMetadata = collector.finalizeMetadata(partitioner, bfFpChance, 0, SerializationHeader.make(cfm, Collections.emptyList())); + return originalMetadata; + } + + @Test + public void testLaReadLb() throws IOException + { + testOldReadsNew("la", "lb"); + } + + @Test + public void testMaReadMb() throws IOException + { + testOldReadsNew("ma", "mb"); + } + + public void testOldReadsNew(String oldV, String newV) throws IOException + { + Map originalMetadata = constructMetadata(); + + MetadataSerializer serializer = new MetadataSerializer(); + // Write metadata in two minor formats. + File statsFileLb = serialize(originalMetadata, serializer, BigFormat.instance.getVersion(newV)); + File statsFileLa = serialize(originalMetadata, serializer, BigFormat.instance.getVersion(oldV)); + // Reading both as earlier version should yield identical results. + Descriptor desc = new Descriptor(oldV, statsFileLb.getParentFile(), "", "", 0, DatabaseDescriptor.getSSTableFormat()); + try (RandomAccessReader inLb = RandomAccessReader.open(statsFileLb); + RandomAccessReader inLa = RandomAccessReader.open(statsFileLa)) + { + Map deserializedLb = serializer.deserialize(desc, inLb, EnumSet.allOf(MetadataType.class)); + Map deserializedLa = serializer.deserialize(desc, inLa, EnumSet.allOf(MetadataType.class)); + + for (MetadataType type : MetadataType.values()) + { + assertEquals(deserializedLa.get(type), deserializedLb.get(type)); + if (!originalMetadata.get(type).equals(deserializedLb.get(type))) + { + // Currently only STATS can be different. Change if no longer the case + assertEquals(MetadataType.STATS, type); + } + } + } + } } diff --git a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java index 9df3fed468..7cf7bd3283 100644 --- a/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java +++ b/test/unit/org/apache/cassandra/io/util/MmappedRegionsTest.java @@ -297,8 +297,7 @@ public class MmappedRegionsTest File cf = File.createTempFile(f.getName() + ".metadata", "1"); cf.deleteOnExit(); - MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)) - .replayPosition(null); + MetadataCollector sstableMetadataCollector = new MetadataCollector(new ClusteringComparator(BytesType.instance)); try(SequentialWriter writer = new CompressedSequentialWriter(f, cf.getAbsolutePath(), CompressionParams.snappy(),