This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch cassandra-5.0 in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 0701c98edc7db955970913967d1585580fa5c525 Merge: 284c9573f6 30f434a613 Author: Caleb Rackliffe <[email protected]> AuthorDate: Tue Nov 3 14:58:52 2026 -0600 Merge branch 'cassandra-4.1' into cassandra-5.0 * cassandra-4.1: Delete SSTable components oldest-first so a partially deleted SSTable's transaction log is not reported as corrupt CHANGES.txt | 1 + .../io/sstable/format/AbstractSSTableFormat.java | 51 +++++++ .../io/sstable/format/SortedTableScrubber.java | 21 --- .../cassandra/io/sstable/format/big/BigFormat.java | 34 +---- .../cassandra/io/sstable/format/bti/BtiFormat.java | 29 +--- .../cassandra/db/lifecycle/LogTransactionTest.java | 169 ++++++++++++++++++++- 6 files changed, 222 insertions(+), 83 deletions(-) diff --cc CHANGES.txt index 61e4b7caa6,ecf92b38fc..aa5c5c7d37 --- a/CHANGES.txt +++ b/CHANGES.txt @@@ -1,30 -1,7 +1,31 @@@ -4.1.13 +5.0.10 + * Avoid NPE in GossipShutdownVerbHandler when endpoint state is null (CASSANDRA-20603) + * Fix ShardManager#coveringRange to include the first partition position (CASSANDRA-20927) + * Respect cassandra.skip_sync in FileOutputStreamPlus.sync and ChecksumWriter (CASSANDRA-21800) + * Fail sstableloader when no replicas are found for the target keyspace (CASSANDRA-20096) + * Fix ignored global auto repair settings repair_max_retries and repair_retry_backoff (CASSANDRA-21748) + * Replica filtering protection resurrects rows deleted by a range tombstone that spans pages (CASSANDRA-21781) + * Write an sstable with the compression its components were chosen for when ALTER TABLE changes it (CASSANDRA-21780) + * Clarify the unset, true and false behaviours of cassandra.reset_bootstrap_progress in the abort message and docs (CASSANDRA-21595) + * Isolate SAI index rebuilds from ZCS (CASSANDRA-21520) + * Bound and chunk the BTI partition index preload (CASSANDRA-21632) + * Restore volatile on CommitLogSegment.cdcState dropped in merge (CASSANDRA-21665) + * Fix compressed scan read-ahead buffer Block leak and chunk-cache pollution by scans (CASSANDRA-21671) + * A read on a replica that does not know the index yet fails with a NullPointerException (CASSANDRA-21658) + * Release hints write buffer when the write executor shuts down (CASSANDRA-21657) + * Fix concurrent SAI vector inserts failing once jvector's per-graph pool limit is exceeded (CASSANDRA-21644) + * Let the commit log allocator observe the shutdown state (CASSANDRA-21616) + * Unwrap LongType properly when calculating min/max terms in V1SSTableIndex (CASSANDRA-21635) + * Force repair should ignore min_repair_interval (CASSANDRA-21552) + * Render SubnetGroups as JSON in system_views.settings (CASSANDRA-21579) + * Avoid rebuilding per-SSTable SAI components unless missing or corrupted (CASSANDRA-21515) + * Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487) + * Allow DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522) + * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519) +Merged from 4.1: * PreV5Handlers.toFlushItem may leak memory on disconnected channel (CASSANDRA-21766) Merged from 4.0: + * Delete SSTable files in order of last-updated timestamp (CASSANDRA-21649) * Prevent negative BufferPool overflow memory accounting (CASSANDRA-21815) * Respect cassandra.skip_sync in LogReplica, SSTable.appendTOC and ChecksumWriter (CASSANDRA-21835) * Fix cassandra-env.sh treating Xmx/Xms/Xmn inside paths as heap flags (CASSANDRA-21804) diff --cc src/java/org/apache/cassandra/io/sstable/format/AbstractSSTableFormat.java index 05073ea12f,0000000000..7d9fb35e66 mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/io/sstable/format/AbstractSSTableFormat.java +++ b/src/java/org/apache/cassandra/io/sstable/format/AbstractSSTableFormat.java @@@ -1,61 -1,0 +1,112 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.sstable.format; + ++import java.util.ArrayList; ++import java.util.Comparator; ++import java.util.HashMap; ++import java.util.List; +import java.util.Map; +import java.util.Objects; ++import java.util.Set; ++ ++import org.slf4j.Logger; ++import org.slf4j.LoggerFactory; ++ ++import org.apache.cassandra.io.sstable.Component; ++import org.apache.cassandra.io.sstable.Descriptor; ++import org.apache.cassandra.io.util.File; + +public abstract class AbstractSSTableFormat<R extends SSTableReader, W extends SSTableWriter> implements SSTableFormat<R, W> +{ ++ private final Logger logger = LoggerFactory.getLogger(getClass()); ++ + public final String name; + protected final Map<String, String> options; + + protected AbstractSSTableFormat(String name, Map<String, String> options) + { + this.name = Objects.requireNonNull(name); + this.options = options; + } + + @Override + public final String name() + { + return name; + } + ++ @Override ++ public final void deleteOrphanedComponents(Descriptor descriptor, Set<Component> components) ++ { ++ if (components.contains(Components.DATA) && descriptor.fileFor(Components.DATA).length() > 0) ++ // everything appears to be in order... moving on. ++ return; ++ ++ // missing the DATA file! all components are orphaned ++ logger.warn("[{}] Removing orphans for {}: {}", getClass().getSimpleName(), descriptor, components); ++ for (Component component : components) ++ descriptor.fileFor(component).deleteIfExists(); ++ } ++ ++ protected final void deleteComponentsOldestFirst(Descriptor desc, Set<Component> components) ++ { ++ logger.info("[{}] Deleting sstable: {}", getClass().getSimpleName(), desc); ++ ++ // delete older files first so the overall SSTable timestamp stays the same on partial deletes ++ Map<Component, File> files = new HashMap<>(); ++ // snapshot each file's last-modified time once, so the sort's repeated comparisons see stable values and don't hit the filesystem ++ Map<Component, Long> lastModified = new HashMap<>(); ++ for (Component c : components) ++ { ++ File file = desc.fileFor(c); ++ files.put(c, file); ++ lastModified.put(c, file.lastModified()); ++ } ++ List<Component> oldestFirst = new ArrayList<>(components); ++ oldestFirst.sort(Comparator.comparingLong(lastModified::get)); ++ ++ for (Component component : oldestFirst) ++ { ++ logger.trace("[{}] Deleting component {} of {}", getClass().getSimpleName(), component, desc); ++ files.get(component).deleteIfExists(); ++ } ++ } ++ + @Override + public final boolean equals(Object o) + { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + AbstractSSTableFormat<?, ?> that = (AbstractSSTableFormat<?, ?>) o; + return Objects.equals(name, that.name); + } + + @Override + public final int hashCode() + { + return Objects.hash(name); + } + + @Override + public String toString() + { + return name + ":" + options; + } +} diff --cc src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java index 66a7a43108,0000000000..acce33b503 mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableScrubber.java @@@ -1,717 -1,0 +1,696 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.sstable.format; + +import java.io.IOError; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; - import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import javax.annotation.concurrent.NotThreadSafe; + +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableSet; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.cassandra.db.ClusteringComparator; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.LivenessInfo; +import org.apache.cassandra.db.compaction.CompactionInfo; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.compaction.OperationType; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.db.partitions.ImmutableBTreePartition; +import org.apache.cassandra.db.partitions.Partition; +import org.apache.cassandra.db.rows.AbstractCell; +import org.apache.cassandra.db.rows.Cell; +import org.apache.cassandra.db.rows.ColumnData; +import org.apache.cassandra.db.rows.ComplexColumnData; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.db.rows.Rows; +import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIterators; +import org.apache.cassandra.db.rows.WrappingUnfilteredRowIterator; +import org.apache.cassandra.exceptions.ConfigurationException; - import org.apache.cassandra.io.sstable.Component; - import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.IScrubber; +import org.apache.cassandra.io.sstable.SSTableIdentityIterator; +import org.apache.cassandra.io.sstable.SSTableRewriter; - import org.apache.cassandra.io.sstable.format.SSTableFormat.Components; +import org.apache.cassandra.io.sstable.metadata.StatsMetadata; +import org.apache.cassandra.io.util.File; +import org.apache.cassandra.io.util.RandomAccessReader; +import org.apache.cassandra.service.ActiveRepairService; +import org.apache.cassandra.utils.AbstractIterator; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.FBUtilities; +import org.apache.cassandra.utils.OutputHandler; +import org.apache.cassandra.utils.TimeUUID; +import org.apache.cassandra.utils.concurrent.Refs; +import org.apache.cassandra.utils.memory.HeapCloner; + +import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; + +@NotThreadSafe +public abstract class SortedTableScrubber<R extends SSTableReaderWithFilter> implements IScrubber +{ + private final static Logger logger = LoggerFactory.getLogger(SortedTableScrubber.class); + + protected final ColumnFamilyStore cfs; + protected final LifecycleTransaction transaction; + protected final File destination; + protected final IScrubber.Options options; + protected final R sstable; + protected final OutputHandler outputHandler; + protected final boolean isCommutative; + protected final long expectedBloomFilterSize; + protected final ReadWriteLock fileAccessLock = new ReentrantReadWriteLock(); + protected final RandomAccessReader dataFile; + protected final ScrubInfo scrubInfo; + + protected final NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics = new NegativeLocalDeletionInfoMetrics(); + + private static final Comparator<Partition> partitionComparator = Comparator.comparing(Partition::partitionKey); + protected final SortedSet<Partition> outOfOrder = new TreeSet<>(partitionComparator); + + + protected int goodPartitions; + protected int badPartitions; + protected int emptyPartitions; + + + protected SortedTableScrubber(ColumnFamilyStore cfs, + LifecycleTransaction transaction, + OutputHandler outputHandler, + Options options) + { + this.sstable = (R) transaction.onlyOne(); + Preconditions.checkNotNull(sstable.metadata()); + assert sstable.metadata().keyspace.equals(cfs.getKeyspaceName()); + if (!sstable.descriptor.cfname.equals(cfs.metadata().name)) + { + logger.warn("Descriptor points to a different table {} than metadata {}", sstable.descriptor.cfname, cfs.metadata().name); + } + try + { + sstable.metadata().validateCompatibility(cfs.metadata()); + } + catch (ConfigurationException ex) + { + logger.warn("Descriptor points to a different table {} than metadata {}", sstable.descriptor.cfname, cfs.metadata().name); + } + + this.cfs = cfs; + this.transaction = transaction; + this.outputHandler = outputHandler; + this.options = options; + this.destination = cfs.getDirectories().getLocationForDisk(cfs.getDiskBoundaries().getCorrectDiskForSSTable(sstable)); + this.isCommutative = cfs.metadata().isCounter(); + + List<SSTableReader> toScrub = Collections.singletonList(sstable); + + long approximateKeyCount; + try + { + approximateKeyCount = SSTableReader.getApproximateKeyCount(toScrub); + } + catch (RuntimeException ex) + { + approximateKeyCount = 0; + } + this.expectedBloomFilterSize = Math.max(cfs.metadata().params.minIndexInterval, approximateKeyCount); + + // loop through each partition, deserializing to check for damage. + // We'll also loop through the index at the same time, using the position from the index to recover if the + // partition header (key or data size) is corrupt. (This means our position in the index file will be one + // partition "ahead" of the data file.) + this.dataFile = transaction.isOffline() + ? sstable.openDataReaderForScan() + : sstable.openDataReaderForScan(CompactionManager.instance.getRateLimiter()); + + this.scrubInfo = new ScrubInfo(dataFile, sstable, fileAccessLock.readLock()); + + if (options.reinsertOverflowedTTLRows) + outputHandler.output("Starting scrub with reinsert overflowed TTL option"); + } + - public static void deleteOrphanedComponents(Descriptor descriptor, Set<Component> components) - { - File dataFile = descriptor.fileFor(Components.DATA); - if (components.contains(Components.DATA) && dataFile.length() > 0) - // everything appears to be in order... moving on. - return; - - // missing the DATA file! all components are orphaned - logger.warn("Removing orphans for {}: {}", descriptor, components); - for (Component component : components) - { - File file = descriptor.fileFor(component); - if (file.exists()) - descriptor.fileFor(component).delete(); - } - } - + @Override + public void scrub() + { + List<SSTableReader> finished = new ArrayList<>(); + outputHandler.output("Scrubbing %s (%s)", sstable, FBUtilities.prettyPrintMemory(dataFile.length())); + try (SSTableRewriter writer = SSTableRewriter.construct(cfs, transaction, false, sstable.maxDataAge); + Refs<SSTableReader> refs = Refs.ref(Collections.singleton(sstable))) + { + StatsMetadata metadata = sstable.getSSTableMetadata(); + writer.switchWriter(CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, metadata.repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction)); + + scrubInternal(writer); + + if (!outOfOrder.isEmpty()) + finished.add(writeOutOfOrderPartitions(metadata)); + + // finish obsoletes the old sstable + transaction.obsoleteOriginals(); + finished.addAll(writer.setRepairedAt(badPartitions > 0 ? ActiveRepairService.UNREPAIRED_SSTABLE : sstable.getSSTableMetadata().repairedAt).finish()); + } + catch (IOException ex) + { + throw new RuntimeException(ex); + } + finally + { + if (transaction.isOffline()) + finished.forEach(sstable -> sstable.selfRef().release()); + } + + outputSummary(finished); + } + + protected abstract void scrubInternal(SSTableRewriter writer) throws IOException; + + private void outputSummary(List<SSTableReader> finished) + { + if (!finished.isEmpty()) + { + outputHandler.output("Scrub of %s complete: %d partitions in new sstable and %d empty (tombstoned) partitions dropped", sstable, goodPartitions, emptyPartitions); + if (negativeLocalDeletionInfoMetrics.fixedRows > 0) + outputHandler.output("Fixed %d rows with overflowed local deletion time.", negativeLocalDeletionInfoMetrics.fixedRows); + if (badPartitions > 0) + outputHandler.warn("Unable to recover %d partitions that were skipped. You can attempt manual recovery from the pre-scrub snapshot. You can also run nodetool repair to transfer the data from a healthy replica, if any", badPartitions); + } + else + { + if (badPartitions > 0) + outputHandler.warn("No valid partitions found while scrubbing %s; it is marked for deletion now. If you want to attempt manual recovery, you can find a copy in the pre-scrub snapshot", sstable); + else + outputHandler.output("Scrub of %s complete; looks like all %d partitions were tombstoned", sstable, emptyPartitions); + } + } + + private SSTableReader writeOutOfOrderPartitions(StatsMetadata metadata) + { + // out of order partitions/rows, but no bad partition found - we can keep our repairedAt time + long repairedAt = badPartitions > 0 ? ActiveRepairService.UNREPAIRED_SSTABLE : sstable.getSSTableMetadata().repairedAt; + SSTableReader newInOrderSstable; + try (SSTableWriter inOrderWriter = CompactionManager.createWriter(cfs, destination, expectedBloomFilterSize, repairedAt, metadata.pendingRepair, metadata.isTransient, sstable, transaction)) + { + for (Partition partition : outOfOrder) + inOrderWriter.append(partition.unfilteredIterator()); + inOrderWriter.setRepairedAt(-1); + inOrderWriter.setMaxDataAge(sstable.maxDataAge); + newInOrderSstable = inOrderWriter.finish(true); + } + transaction.update(newInOrderSstable, false); + outputHandler.warn("%d out of order partition (or partitions without of order rows) found while scrubbing %s; " + + "Those have been written (in order) to a new sstable (%s)", outOfOrder.size(), sstable, newInOrderSstable); + return newInOrderSstable; + } + + protected abstract UnfilteredRowIterator withValidation(UnfilteredRowIterator iter, String filename); + + @Override + @VisibleForTesting + public ScrubResult scrubWithResult() + { + scrub(); + return new ScrubResult(goodPartitions, badPartitions, emptyPartitions); + } + + @Override + public CompactionInfo.Holder getScrubInfo() + { + return scrubInfo; + } + + protected String keyString(DecoratedKey key) + { + if (key == null) + return "(unknown)"; + + try + { + return cfs.metadata().partitionKeyType.getString(key.getKey()); + } + catch (Exception e) + { + return String.format("(corrupted; hex value: %s)", ByteBufferUtil.bytesToHex(key.getKey())); + } + } + + protected boolean tryAppend(DecoratedKey prevKey, DecoratedKey key, SSTableRewriter writer) + { + // OrderCheckerIterator will check, at iteration time, that the rows are in the proper order. If it detects + // that one row is out of order, it will stop returning them. The remaining rows will be sorted and added + // to the outOfOrder set that will be later written to a new SSTable. + try (OrderCheckerIterator sstableIterator = new OrderCheckerIterator(getIterator(key), cfs.metadata().comparator); + UnfilteredRowIterator iterator = withValidation(sstableIterator, dataFile.getPath())) + { + if (prevKey != null && prevKey.compareTo(key) > 0) + { + saveOutOfOrderPartition(prevKey, key, iterator); + return false; + } + + if (writer.tryAppend(iterator) == null) + emptyPartitions++; + else + goodPartitions++; + + if (sstableIterator.hasRowsOutOfOrder()) + { + outputHandler.warn("Out of order rows found in partition: %s", keyString(key)); + outOfOrder.add(sstableIterator.getRowsOutOfOrder()); + } + } + + return true; + } + + /** + * Only wrap with {@link FixNegativeLocalDeletionTimeIterator} if {@link IScrubber.Options#reinsertOverflowedTTLRows} option + * is specified + */ + private UnfilteredRowIterator getIterator(DecoratedKey key) + { + RowMergingSSTableIterator rowMergingIterator = new RowMergingSSTableIterator(SSTableIdentityIterator.create(sstable, + dataFile, + key), + outputHandler, + sstable.descriptor.version, + options.reinsertOverflowedTTLRows); + if (options.reinsertOverflowedTTLRows) + return new FixNegativeLocalDeletionTimeIterator(rowMergingIterator, outputHandler, negativeLocalDeletionInfoMetrics); + else + return rowMergingIterator; + } + + private void saveOutOfOrderPartition(DecoratedKey prevKey, DecoratedKey key, UnfilteredRowIterator iterator) + { + // TODO bitch if the row is too large? if it is there's not much we can do ... + outputHandler.warn("Out of order partition detected (%s found after %s)", keyString(key), keyString(prevKey)); + outOfOrder.add(ImmutableBTreePartition.create(iterator)); + } + + protected static void throwIfFatal(Throwable th) + { + if (th instanceof Error && !(th instanceof AssertionError || th instanceof IOError)) + throw (Error) th; + } + + protected void throwIfCannotContinue(DecoratedKey key, Throwable th) + { + if (isCommutative && !options.skipCorrupted) + { + outputHandler.warn("An error occurred while scrubbing the partition with key '%s'. Skipping corrupt " + + "data in counter tables will result in undercounts for the affected " + + "counters (see CASSANDRA-2759 for more details), so by default the scrub will " + + "stop at this point. If you would like to skip the row anyway and continue " + + "scrubbing, re-run the scrub with the --skip-corrupted option.", + keyString(key)); + throw new IOError(th); + } + } + + + public static class ScrubInfo extends CompactionInfo.Holder + { + private final RandomAccessReader dataFile; + private final SSTableReader sstable; + private final TimeUUID scrubCompactionId; + private final Lock fileReadLock; + + public ScrubInfo(RandomAccessReader dataFile, SSTableReader sstable, Lock fileReadLock) + { + this.dataFile = dataFile; + this.sstable = sstable; + this.fileReadLock = fileReadLock; + scrubCompactionId = nextTimeUUID(); + } + + public CompactionInfo getCompactionInfo() + { + fileReadLock.lock(); + try + { + return new CompactionInfo(sstable.metadata(), + OperationType.SCRUB, + dataFile.getFilePointer(), + dataFile.length(), + sstable.onDiskLength(), + scrubCompactionId, + ImmutableSet.of(sstable), + File.getPath(sstable.getFilename()).getParent().toString()); + } + catch (Exception e) + { + throw new RuntimeException(e); + } + finally + { + fileReadLock.unlock(); + } + } + + public boolean isGlobal() + { + return false; + } + } + + /** + * In some case like CASSANDRA-12127 the cells might have been stored in the wrong order. This decorator check the + * cells order and collect the out-of-order cells to correct the problem. + */ + private static final class OrderCheckerIterator extends AbstractIterator<Unfiltered> implements WrappingUnfilteredRowIterator + { + private final UnfilteredRowIterator iterator; + private final ClusteringComparator comparator; + + private Unfiltered previous; + + /** + * The partition containing the rows which are out of order. + */ + private Partition rowsOutOfOrder; + + public OrderCheckerIterator(UnfilteredRowIterator iterator, ClusteringComparator comparator) + { + this.iterator = iterator; + this.comparator = comparator; + } + + @Override + public UnfilteredRowIterator wrapped() + { + return iterator; + } + + public boolean hasRowsOutOfOrder() + { + return rowsOutOfOrder != null; + } + + public Partition getRowsOutOfOrder() + { + return rowsOutOfOrder; + } + + @Override + protected Unfiltered computeNext() + { + if (!iterator.hasNext()) + return endOfData(); + + Unfiltered next = iterator.next(); + + // If we detect that some rows are out of order we will store and sort the remaining ones to insert them + // in a separate SSTable. + if (previous != null && comparator.compare(next, previous) < 0) + { + rowsOutOfOrder = ImmutableBTreePartition.create(UnfilteredRowIterators.concat(next, iterator), false); + return endOfData(); + } + previous = next; + return next; + } + } + + /** + * During 2.x migration, under some circumstances rows might have gotten duplicated. + * Merging iterator merges rows with same clustering. + * <p> + * For more details, refer to CASSANDRA-12144. + */ + private static class RowMergingSSTableIterator implements WrappingUnfilteredRowIterator + { + Unfiltered nextToOffer = null; + private final OutputHandler output; + private final UnfilteredRowIterator wrapped; + private final Version sstableVersion; + private final boolean reinsertOverflowedTTLRows; + + RowMergingSSTableIterator(UnfilteredRowIterator source, OutputHandler output, Version sstableVersion, boolean reinsertOverflowedTTLRows) + { + this.wrapped = source; + this.output = output; + this.sstableVersion = sstableVersion; + this.reinsertOverflowedTTLRows = reinsertOverflowedTTLRows; + } + + @Override + public UnfilteredRowIterator wrapped() + { + return wrapped; + } + + @Override + public boolean hasNext() + { + return nextToOffer != null || wrapped.hasNext(); + } + + @Override + public Unfiltered next() + { + Unfiltered next = nextToOffer != null ? nextToOffer : wrapped.next(); + + if (next.isRow()) + { + boolean logged = false; + while (wrapped.hasNext()) + { + Unfiltered peek = wrapped.next(); + if (!peek.isRow() || !next.clustering().equals(peek.clustering())) + { + nextToOffer = peek; // Offer peek in next call + return computeFinalRow((Row) next); + } + + // Duplicate row, merge it. + next = Rows.merge((Row) next, (Row) peek); + + if (!logged) + { + String partitionKey = metadata().partitionKeyType.getString(partitionKey().getKey()); + output.warn("Duplicate row detected in %s.%s: %s %s", metadata().keyspace, metadata().name, partitionKey, next.clustering().toString(metadata())); + logged = true; + } + } + } + + nextToOffer = null; + return computeFinalRow((Row) next); + } + + private Row computeFinalRow(Row next) + { + // If the row has overflowed let rows skip them unless we need to keep them for the overflow policy + if (hasOverflowedLocalExpirationTimeRow(next) && !reinsertOverflowedTTLRows) + return null; + else if (reinsertOverflowedTTLRows) + return rebuildTimestamptsForOverflowedRows(next); + else + return next; + } + + /* + * When building ldt on deser it won't overflow now being a long as it used to. + * This causes row resurrection for old sstable formats! + * To prevent it we preserve the overflow to be backwards compatible and to feed into the overflow policy + */ + private Row rebuildTimestamptsForOverflowedRows(Row row) + { + if (sstableVersion.hasUIntDeletionTime()) + return row; + + LivenessInfo livenessInfo = row.primaryKeyLivenessInfo(); + if (livenessInfo.isExpiring() && livenessInfo.localExpirationTime() >= 0) + { + livenessInfo = livenessInfo.withUpdatedTimestampAndLocalDeletionTime(livenessInfo.timestamp(), livenessInfo.localExpirationTime(), false); + } + + return row.transformAndFilter(livenessInfo, row.deletion(), cd -> { + if (cd.column().isSimple()) + { + Cell<?> cell = (Cell<?>)cd; + return cell.isExpiring() && cell.localDeletionTime() >= 0 + ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp(), cell.localDeletionTime()) + : cell; + } + else + { + ComplexColumnData complexData = (ComplexColumnData)cd; + return complexData.transformAndFilter(cell -> cell.isExpiring() && cell.localDeletionTime() >= 0 + ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp(), cell.localDeletionTime()) + : cell); + } + }).clone(HeapCloner.instance); + } + + private boolean hasOverflowedLocalExpirationTimeRow(Row next) + { + if (sstableVersion.hasUIntDeletionTime()) + return false; + + if (next.primaryKeyLivenessInfo().isExpiring() && next.primaryKeyLivenessInfo().localExpirationTime() >= 0) + { + return true; + } + + for (ColumnData cd : next) + { + if (cd.column().isSimple()) + { + Cell<?> cell = (Cell<?>)cd; + if (cell.isExpiring() && cell.localDeletionTime() >= 0) + return true; + } + else + { + ComplexColumnData complexData = (ComplexColumnData)cd; + for (Cell<?> cell : complexData) + { + if (cell.isExpiring() && cell.localDeletionTime() >= 0) + return true; + } + } + } + + return false; + } + } + + /** + * This iterator converts negative {@link AbstractCell#localDeletionTime()} into {@link AbstractCell#MAX_DELETION_TIME} + * <p> + * This is to recover entries with overflowed localExpirationTime due to CASSANDRA-14092 + */ + private static final class FixNegativeLocalDeletionTimeIterator extends AbstractIterator<Unfiltered> implements WrappingUnfilteredRowIterator + { + /** + * The decorated iterator. + */ + private final UnfilteredRowIterator iterator; + + private final OutputHandler outputHandler; + private final NegativeLocalDeletionInfoMetrics negativeLocalExpirationTimeMetrics; + + public FixNegativeLocalDeletionTimeIterator(UnfilteredRowIterator iterator, OutputHandler outputHandler, + NegativeLocalDeletionInfoMetrics negativeLocalDeletionInfoMetrics) + { + this.iterator = iterator; + this.outputHandler = outputHandler; + this.negativeLocalExpirationTimeMetrics = negativeLocalDeletionInfoMetrics; + } + + @Override + public UnfilteredRowIterator wrapped() + { + return iterator; + } + + @Override + protected Unfiltered computeNext() + { + if (!iterator.hasNext()) + return endOfData(); + + Unfiltered next = iterator.next(); + if (!next.isRow()) + return next; + + if (hasNegativeLocalExpirationTime((Row) next)) + { + outputHandler.debug("Found row with negative local expiration time: %s", next.toString(metadata(), false)); + negativeLocalExpirationTimeMetrics.fixedRows++; + return fixNegativeLocalExpirationTime((Row) next); + } + + return next; + } + + private boolean hasNegativeLocalExpirationTime(Row next) + { + Row row = next; + if (row.primaryKeyLivenessInfo().isExpiring() && row.primaryKeyLivenessInfo().localExpirationTime() == Cell.INVALID_DELETION_TIME) + { + return true; + } + + for (ColumnData cd : row) + { + if (cd.column().isSimple()) + { + Cell<?> cell = (Cell<?>) cd; + if (cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME) + return true; + } + else + { + ComplexColumnData complexData = (ComplexColumnData) cd; + for (Cell<?> cell : complexData) + { + if (cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME) + return true; + } + } + } + + return false; + } + + private Unfiltered fixNegativeLocalExpirationTime(Row row) + { + LivenessInfo livenessInfo = row.primaryKeyLivenessInfo(); + if (livenessInfo.isExpiring() && livenessInfo.localExpirationTime() == Cell.INVALID_DELETION_TIME) + livenessInfo = livenessInfo.withUpdatedTimestampAndLocalDeletionTime(livenessInfo.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP); + + return row.transformAndFilter(livenessInfo, row.deletion(), cd -> { + if (cd.column().isSimple()) + { + Cell cell = (Cell) cd; + return cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME + ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP) + : cell; + } + else + { + ComplexColumnData complexData = (ComplexColumnData) cd; + return complexData.transformAndFilter(cell -> cell.isExpiring() && cell.localDeletionTime() == Cell.INVALID_DELETION_TIME + ? cell.withUpdatedTimestampAndLocalDeletionTime(cell.timestamp() + 1, AbstractCell.MAX_DELETION_TIME_2038_LEGACY_CAP) + : cell); + } + }).clone(HeapCloner.instance); + } + } + + private static class NegativeLocalDeletionInfoMetrics + { + public volatile int fixedRows = 0; + } +} diff --cc src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java index 46f8a3bf2b,b55b4a00cf..716b083674 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java @@@ -17,128 -17,33 +17,118 @@@ */ package org.apache.cassandra.io.sstable.format.big; -import java.util.Collection; - +import java.io.IOException; +import java.util.Iterator; - import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Iterables; - import com.google.common.collect.Lists; +import com.google.common.collect.Sets; + - import org.slf4j.Logger; - import org.slf4j.LoggerFactory; - +import org.apache.cassandra.cache.KeyCacheKey; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.SSTable; -import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.schema.TableMetadataRef; -import org.apache.cassandra.db.RowIndexEntry; -import org.apache.cassandra.db.SerializationHeader; -import org.apache.cassandra.db.lifecycle.LifecycleNewTracker; import org.apache.cassandra.io.sstable.Descriptor; -import org.apache.cassandra.io.sstable.format.*; -import org.apache.cassandra.io.sstable.metadata.MetadataCollector; +import org.apache.cassandra.io.sstable.GaugeProvider; +import org.apache.cassandra.io.sstable.IScrubber; +import org.apache.cassandra.io.sstable.MetricsProviders; +import org.apache.cassandra.io.sstable.filter.BloomFilterMetrics; +import org.apache.cassandra.io.sstable.format.AbstractSSTableFormat; +import org.apache.cassandra.io.sstable.format.SSTableFormat; +import org.apache.cassandra.io.sstable.format.SSTableReaderLoadingBuilder; +import org.apache.cassandra.io.sstable.format.SSTableWriter; - import org.apache.cassandra.io.sstable.format.SortedTableScrubber; +import org.apache.cassandra.io.sstable.format.Version; +import org.apache.cassandra.io.sstable.indexsummary.IndexSummaryMetrics; +import org.apache.cassandra.io.sstable.keycache.KeyCacheMetrics; +import org.apache.cassandra.io.util.DataInputPlus; +import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.net.MessagingService; -import org.apache.cassandra.utils.TimeUUID; +import org.apache.cassandra.schema.TableMetadataRef; +import org.apache.cassandra.service.CacheService; +import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.OutputHandler; +import org.apache.cassandra.utils.Pair; - import static org.apache.cassandra.io.sstable.format.SSTableFormat.Components.DATA; - /** * Legacy bigtable format */ -public class BigFormat implements SSTableFormat +public class BigFormat extends AbstractSSTableFormat<BigTableReader, BigTableWriter> { - private final static Logger logger = LoggerFactory.getLogger(BigFormat.class); - - public static final BigFormat instance = new BigFormat(); - public static final Version latestVersion = new BigVersion(BigVersion.current_version); - private static final SSTableReader.Factory readerFactory = new ReaderFactory(); - private static final SSTableWriter.Factory writerFactory = new WriterFactory(); + public static final String NAME = "big"; - private BigFormat() + private final Version latestVersion = new BigVersion(this, BigVersion.current_version); + private final BigTableReaderFactory readerFactory = new BigTableReaderFactory(); + private final BigTableWriterFactory writerFactory = new BigTableWriterFactory(); + + public static class Components extends SSTableFormat.Components { + public static class Types extends SSTableFormat.Components.Types + { + // index of the row keys with pointers to their positions in the data file + public static final Component.Type PRIMARY_INDEX = Component.Type.createSingleton("PRIMARY_INDEX", "Index.db", true, BigFormat.class); + // holds SSTable Index Summary (sampling of Index component) + public static final Component.Type SUMMARY = Component.Type.createSingleton("SUMMARY", "Summary.db", true, BigFormat.class); + } + public final static Component PRIMARY_INDEX = Types.PRIMARY_INDEX.getSingleton(); + public final static Component SUMMARY = Types.SUMMARY.getSingleton(); + + private static final Set<Component> BATCH_COMPONENTS = ImmutableSet.of(DATA, + PRIMARY_INDEX, + COMPRESSION_INFO, + FILTER, + STATS); + + private static final Set<Component> PRIMARY_COMPONENTS = ImmutableSet.of(DATA, + PRIMARY_INDEX); + + private static final Set<Component> GENERATED_ON_LOAD_COMPONENTS = ImmutableSet.of(FILTER, SUMMARY); + + private static final Set<Component> MUTABLE_COMPONENTS = ImmutableSet.of(STATS, + SUMMARY); + + private static final Set<Component> UPLOAD_COMPONENTS = ImmutableSet.of(DATA, + PRIMARY_INDEX, + SUMMARY, + COMPRESSION_INFO, + STATS); + private static final Set<Component> ALL_COMPONENTS = ImmutableSet.of(DATA, + PRIMARY_INDEX, + STATS, + COMPRESSION_INFO, + FILTER, + SUMMARY, + DIGEST, + CRC, + TOC); + } + + public BigFormat(Map<String, String> options) + { + super(NAME, options); + } + + public static boolean is(SSTableFormat<?, ?> format) + { + return format.name().equals(NAME); + } + + public static BigFormat getInstance() + { + return (BigFormat) Objects.requireNonNull(DatabaseDescriptor.getSSTableFormats().get(NAME), "Unknown SSTable format: " + NAME); + } + + public static boolean isSelected() + { + return is(DatabaseDescriptor.getSelectedSSTableFormat()); } @Override @@@ -166,121 -71,20 +156,99 @@@ } @Override - public RowIndexEntry.IndexSerializer getIndexSerializer(TableMetadata metadata, Version version, SerializationHeader header) + public Set<Component> allComponents() { - return new RowIndexEntry.Serializer(version, header); + return Components.ALL_COMPONENTS; } - static class WriterFactory extends SSTableWriter.Factory + @Override + public Set<Component> primaryComponents() { + return Components.PRIMARY_COMPONENTS; + } + + @Override + public Set<Component> batchComponents() + { + return Components.BATCH_COMPONENTS; + } + + @Override + public Set<Component> uploadComponents() + { + return Components.UPLOAD_COMPONENTS; + } + + @Override + public Set<Component> mutableComponents() + { + return Components.MUTABLE_COMPONENTS; + } + + @Override + public Set<Component> generatedOnLoadComponents() + { + return Components.GENERATED_ON_LOAD_COMPONENTS; + } + + @Override + public SSTableFormat.KeyCacheValueSerializer<BigTableReader, RowIndexEntry> getKeyCacheValueSerializer() + { + return KeyCacheValueSerializer.instance; + } + + @Override + public IScrubber getScrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, OutputHandler outputHandler, IScrubber.Options options) + { + Preconditions.checkArgument(cfs.metadata().equals(transaction.onlyOne().metadata()), "SSTable metadata does not match current definition"); + return new BigTableScrubber(cfs, transaction, outputHandler, options); + } + + @Override + public MetricsProviders getFormatSpecificMetricsProviders() + { + return BigTableSpecificMetricsProviders.instance; + } + - @Override - public void deleteOrphanedComponents(Descriptor descriptor, Set<Component> components) - { - SortedTableScrubber.deleteOrphanedComponents(descriptor, components); - } - - private void delete(Descriptor desc, List<Component> components) - { - logger.info("Deleting sstable: {}", desc); - - if (components.remove(DATA)) - components.add(0, DATA); // DATA component should be first - if (components.remove(Components.SUMMARY)) - components.add(Components.SUMMARY); // SUMMARY component should be last (IDK why) - - for (Component component : components) - { - logger.trace("Deleting component {} of {}", component, desc); - desc.fileFor(component).deleteIfExists(); - } - } - + @Override + public void delete(Descriptor desc) + { + try + { + if (DatabaseDescriptor.shouldInvalidateKeycacheOnSSTableDeletion()) + { + // remove key cache entries for the sstable being deleted + Iterator<KeyCacheKey> it = CacheService.instance.keyCache.keyIterator(); + while (it.hasNext()) + { + KeyCacheKey key = it.next(); + if (key.desc.equals(desc)) + it.remove(); + } + } + - delete(desc, Lists.newArrayList(Sets.intersection(allComponents(), desc.discoverComponents()))); ++ deleteComponentsOldestFirst(desc, Sets.intersection(allComponents(), desc.discoverComponents())); + } + catch (Throwable t) + { + JVMStabilityInspector.inspectThrowable(t); + } + } + + static class KeyCacheValueSerializer implements SSTableFormat.KeyCacheValueSerializer<BigTableReader, RowIndexEntry> + { + private final static KeyCacheValueSerializer instance = new KeyCacheValueSerializer(); + @Override - public long estimateSize(SSTableWriter.SSTableSizeParameters parameters) + public void skip(DataInputPlus input) throws IOException { - return (long) ((parameters.partitionKeysSize() // index entries - + parameters.partitionKeysSize() // keys in data file - + parameters.dataSize()) // data - * 1.2); // bloom filter and row index overhead + RowIndexEntry.Serializer.skipForCache(input, getInstance().latestVersion); + } + + @Override + public RowIndexEntry deserialize(BigTableReader reader, DataInputPlus input) throws IOException + { + return reader.deserializeKeyCacheValue(input); } @Override diff --cc src/java/org/apache/cassandra/io/sstable/format/bti/BtiFormat.java index e7703c6a06,0000000000..12d627323a mode 100644,000000..100644 --- a/src/java/org/apache/cassandra/io/sstable/format/bti/BtiFormat.java +++ b/src/java/org/apache/cassandra/io/sstable/format/bti/BtiFormat.java @@@ -1,450 -1,0 +1,423 @@@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.io.sstable.format.bti; + +import java.io.IOException; - import java.util.List; +import java.util.Map; +import java.util.Set; + +import com.google.common.base.Preconditions; +import com.google.common.collect.ImmutableSet; - import com.google.common.collect.Lists; +import com.google.common.collect.Sets; - import org.slf4j.Logger; - import org.slf4j.LoggerFactory; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.lifecycle.LifecycleTransaction; +import org.apache.cassandra.dht.IPartitioner; +import org.apache.cassandra.io.sstable.Component; +import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.GaugeProvider; +import org.apache.cassandra.io.sstable.IScrubber; +import org.apache.cassandra.io.sstable.MetricsProviders; +import org.apache.cassandra.io.sstable.SSTable; +import org.apache.cassandra.io.sstable.filter.BloomFilterMetrics; +import org.apache.cassandra.io.sstable.format.AbstractSSTableFormat; +import org.apache.cassandra.io.sstable.format.SSTableFormat; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.io.sstable.format.SSTableReaderLoadingBuilder; +import org.apache.cassandra.io.sstable.format.SSTableWriter; - import org.apache.cassandra.io.sstable.format.SortedTableScrubber; +import org.apache.cassandra.io.sstable.format.Version; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.TableMetadataRef; +import org.apache.cassandra.utils.JVMStabilityInspector; +import org.apache.cassandra.utils.OutputHandler; +import org.apache.cassandra.utils.Pair; + +/** + * Bigtable format with trie indices. See BTIFormat.md for the format documentation. + */ +public class BtiFormat extends AbstractSSTableFormat<BtiTableReader, BtiTableWriter> +{ - private final static Logger logger = LoggerFactory.getLogger(BtiFormat.class); - + public static final String NAME = "bti"; + + private final Version latestVersion = new BtiVersion(this, BtiVersion.current_version); + static final BtiTableReaderFactory readerFactory = new BtiTableReaderFactory(); + static final BtiTableWriterFactory writerFactory = new BtiTableWriterFactory(); + + public static class Components extends SSTableFormat.Components + { + public static class Types extends AbstractSSTableFormat.Components.Types + { + public static final Component.Type PARTITION_INDEX = Component.Type.createSingleton("PARTITION_INDEX", "Partitions.db", true, BtiFormat.class); + public static final Component.Type ROW_INDEX = Component.Type.createSingleton("ROW_INDEX", "Rows.db", true, BtiFormat.class); + } + + public final static Component PARTITION_INDEX = Types.PARTITION_INDEX.getSingleton(); + + public final static Component ROW_INDEX = Types.ROW_INDEX.getSingleton(); + + private final static Set<Component> PRIMARY_COMPONENTS = ImmutableSet.of(DATA, + PARTITION_INDEX); + + private final static Set<Component> MUTABLE_COMPONENTS = ImmutableSet.of(STATS); + + private static final Set<Component> UPLOAD_COMPONENTS = ImmutableSet.of(DATA, + PARTITION_INDEX, + ROW_INDEX, + COMPRESSION_INFO, + STATS); + + private static final Set<Component> BATCH_COMPONENTS = ImmutableSet.of(DATA, + PARTITION_INDEX, + ROW_INDEX, + COMPRESSION_INFO, + FILTER, + STATS); + + private final static Set<Component> ALL_COMPONENTS = ImmutableSet.of(DATA, + PARTITION_INDEX, + ROW_INDEX, + STATS, + COMPRESSION_INFO, + FILTER, + DIGEST, + CRC, + TOC); + + private final static Set<Component> GENERATED_ON_LOAD_COMPONENTS = ImmutableSet.of(FILTER); + } + + + public BtiFormat(Map<String, String> options) + { + super(NAME, options); + } + + public static boolean is(SSTableFormat<?, ?> format) + { + return format.name().equals(NAME); + } + + public static boolean isSelected() + { + return is(DatabaseDescriptor.getSelectedSSTableFormat()); + } + + @Override + public Version getLatestVersion() + { + return latestVersion; + } + + @Override + public Version getVersion(String version) + { + return new BtiVersion(this, version); + } + + @Override + public BtiTableWriterFactory getWriterFactory() + { + return writerFactory; + } + + @Override + public BtiTableReaderFactory getReaderFactory() + { + return readerFactory; + } + + @Override + public Set<Component> primaryComponents() + { + return Components.PRIMARY_COMPONENTS; + } + + @Override + public Set<Component> batchComponents() + { + return Components.BATCH_COMPONENTS; + } + + @Override + public Set<Component> uploadComponents() + { + return Components.UPLOAD_COMPONENTS; + } + + @Override + public Set<Component> mutableComponents() + { + return Components.MUTABLE_COMPONENTS; + } + + @Override + public Set<Component> allComponents() + { + return Components.ALL_COMPONENTS; + } + + @Override + public Set<Component> generatedOnLoadComponents() + { + return Components.GENERATED_ON_LOAD_COMPONENTS; + } + + @Override + public SSTableFormat.KeyCacheValueSerializer<BtiTableReader, TrieIndexEntry> getKeyCacheValueSerializer() + { + throw new AssertionError("BTI sstables do not use key cache"); + } + + @Override + public IScrubber getScrubber(ColumnFamilyStore cfs, LifecycleTransaction transaction, OutputHandler outputHandler, IScrubber.Options options) + { + Preconditions.checkArgument(cfs.metadata().equals(transaction.onlyOne().metadata()), "SSTable metadata does not match current definition"); + return new BtiTableScrubber(cfs, transaction, outputHandler, options); + } + + @Override + public MetricsProviders getFormatSpecificMetricsProviders() + { + return BtiTableSpecificMetricsProviders.instance; + } + - @Override - public void deleteOrphanedComponents(Descriptor descriptor, Set<Component> components) - { - SortedTableScrubber.deleteOrphanedComponents(descriptor, components); - } - - private void delete(Descriptor desc, List<Component> components) - { - logger.info("Deleting sstable: {}", desc); - - if (components.remove(SSTableFormat.Components.DATA)) - components.add(0, SSTableFormat.Components.DATA); // DATA component should be first - - for (Component component : components) - { - logger.trace("Deleting component {} of {}", component, desc); - desc.fileFor(component).deleteIfExists(); - } - } - + @Override + public void delete(Descriptor desc) + { + try + { - delete(desc, Lists.newArrayList(Sets.intersection(allComponents(), desc.discoverComponents()))); ++ deleteComponentsOldestFirst(desc, Sets.intersection(allComponents(), desc.discoverComponents())); + } + catch (Throwable t) + { + JVMStabilityInspector.inspectThrowable(t); + } + } + + static class BtiTableReaderFactory implements SSTableReaderFactory<BtiTableReader, BtiTableReader.Builder> + { + @Override + public SSTableReader.Builder<BtiTableReader, BtiTableReader.Builder> builder(Descriptor descriptor) + { + return new BtiTableReader.Builder(descriptor); + } + + @Override + public SSTableReaderLoadingBuilder<BtiTableReader, BtiTableReader.Builder> loadingBuilder(Descriptor descriptor, TableMetadataRef tableMetadataRef, Set<Component> components) + { + return new BtiTableReaderLoadingBuilder(new SSTable.Builder<>(descriptor).setTableMetadataRef(tableMetadataRef) + .setComponents(components)); + } + + @Override + public Pair<DecoratedKey, DecoratedKey> readKeyRange(Descriptor descriptor, IPartitioner partitioner) throws IOException + { + return PartitionIndex.readFirstAndLastKey(descriptor.fileFor(Components.PARTITION_INDEX), partitioner); + } + + @Override + public Class<BtiTableReader> getReaderClass() + { + return BtiTableReader.class; + } + } + + static class BtiTableWriterFactory implements SSTableWriterFactory<BtiTableWriter, BtiTableWriter.Builder> + { + @Override + public BtiTableWriter.Builder builder(Descriptor descriptor) + { + return new BtiTableWriter.Builder(descriptor); + } + + @Override + public long estimateSize(SSTableWriter.SSTableSizeParameters parameters) + { + return (long) ((parameters.partitionCount() * 8 // index entries + + parameters.partitionKeysSize() // keys in data file + + parameters.dataSize()) // data + * 1.2); // bloom filter and row index overhead + } + } + + static class BtiVersion extends Version + { + public static final String current_version = "da"; + public static final String earliest_supported_version = "da"; + + // versions aa-cz are not supported in OSS + // da (5.0): initial version of the BIT format + // NOTE: when adding a new version, please add that to LegacySSTableTest, too. + + private final boolean isLatestVersion; + + private final int correspondingMessagingVersion; + + BtiVersion(BtiFormat format, String version) + { + super(format, version); + + isLatestVersion = version.compareTo(current_version) == 0; + correspondingMessagingVersion = MessagingService.VERSION_50; + } + + @Override + public boolean isLatestVersion() + { + return isLatestVersion; + } + + @Override + public int correspondingMessagingVersion() + { + return correspondingMessagingVersion; + } + + @Override + public boolean hasCommitLogLowerBound() + { + return true; + } + + @Override + public boolean hasCommitLogIntervals() + { + return true; + } + + @Override + public boolean hasMaxCompressedLength() + { + return true; + } + + @Override + public boolean hasPendingRepair() + { + return true; + } + + @Override + public boolean hasIsTransient() + { + return true; + } + + @Override + public boolean hasMetadataChecksum() + { + return true; + } + + @Override + public boolean hasOldBfFormat() + { + return false; + } + + @Override + public boolean hasAccurateMinMax() + { + return true; + } + + public boolean hasLegacyMinMax() + { + return false; + } + + @Override + public boolean hasOriginatingHostId() + { + return true; + } + + @Override + public boolean hasImprovedMinMax() { + return true; + } + + @Override + public boolean hasTokenSpaceCoverage() + { + return true; + } + + @Override + public boolean hasPartitionLevelDeletionsPresenceMarker() + { + return true; + } + + @Override + public boolean hasKeyRange() + { + return true; + } + + @Override + public boolean isCompatible() + { + return version.compareTo(earliest_supported_version) >= 0 && version.charAt(0) <= current_version.charAt(0); + } + + @Override + public boolean isCompatibleForStreaming() + { + return isCompatible() && version.charAt(0) == current_version.charAt(0); + } + + @Override + public boolean hasUIntDeletionTime() + { + return true; + } + + } + + private static class BtiTableSpecificMetricsProviders implements MetricsProviders + { + private final static BtiTableSpecificMetricsProviders instance = new BtiTableSpecificMetricsProviders(); + + private final Iterable<GaugeProvider<?>> gaugeProviders = BloomFilterMetrics.instance.getGaugeProviders(); + + @Override + public Iterable<GaugeProvider<?>> getGaugeProviders() + { + return gaugeProviders; + } + } + + @SuppressWarnings("unused") + public static class BtiFormatFactory implements Factory + { + @Override + public String name() + { + return NAME; + } + + @Override + public SSTableFormat<?, ?> getInstance(Map<String, String> options) + { + return new BtiFormat(options); + } + } +} diff --cc test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java index 16787b3d1c,0ff2a1fda4..a7458d51f7 --- a/test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java +++ b/test/unit/org/apache/cassandra/db/lifecycle/LogTransactionTest.java @@@ -21,13 -21,13 +21,16 @@@ import java.io.IOException import java.io.UncheckedIOException; import java.nio.file.Files; import java.nio.file.NoSuchFileException; ++import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashSet; ++import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; ++import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Supplier; @@@ -1635,9 -1584,8 +1638,173 @@@ public class LogTransactionTest extend @Test(expected = TransactionAlreadyCompletedException.class) public void useAfterCompletedTest() { - LogTransaction txnFile = new LogTransaction(OperationType.STREAM); - txnFile.abort(); // this should complete the txn - txnFile.trackNew(dummySSTable()); // expect an IllegalStateException here + try (LogTransaction txnFile = new LogTransaction(OperationType.STREAM)) + { + txnFile.abort(); // this should complete the txn + txnFile.trackNew(dummySSTable()); // expect an IllegalStateException here + } - }} ++ } ++ ++ @Test ++ public void testBigFormatSSTableLastModifiedTimestampNeverChangesOnPartialDeletes() throws IOException ++ { ++ assertLastModifiedTimestampNeverChangesOnPartialDeletes(BigFormat.getInstance()); ++ } ++ ++ @Test ++ public void testBtiFormatSSTableLastModifiedTimestampNeverChangesOnPartialDeletes() throws IOException ++ { ++ assertLastModifiedTimestampNeverChangesOnPartialDeletes(DatabaseDescriptor.getSSTableFormats().get(BtiFormat.NAME)); ++ } ++ ++ // Verifies that a format's deletion of an sstable's component files never lets the highest ++ // update time among the sstable's files change, even mid-deletion. ++ private static void assertLastModifiedTimestampNeverChangesOnPartialDeletes(SSTableFormat<?, ?> format) throws IOException ++ { ++ // 1) create an sstable's real component files, with reversed update times to require a full sort for correct deletion order ++ Descriptor realDesc = newTempDescriptor(format); ++ Map<Component, File> realFiles = createComponentFilesWithReversedUpdateTimes(realDesc, format); ++ ++ // 2) get the highest timestamp and save it ++ long expectedMaxUpdateTime = realFiles.values().stream().mapToLong(File::lastModified).max().getAsLong(); ++ ++ Set<Component> deleted = new HashSet<>(); ++ List<Component> deletionOrder = new ArrayList<>(); ++ List<String> violations = new ArrayList<>(); ++ Descriptor stubDesc = stubDeletionOrder(realDesc, realFiles, component -> ++ { ++ deleted.add(component); ++ deletionOrder.add(component); ++ long remainingMax = realFiles.entrySet().stream() ++ .filter(e -> !deleted.contains(e.getKey())) ++ .mapToLong(e -> e.getValue().lastModified()) ++ .max().orElse(-1); ++ if (remainingMax >= 0 && remainingMax != expectedMaxUpdateTime) ++ violations.add("after deleting " + component + ": last-modified calculation changed from " ++ + expectedMaxUpdateTime + " to " + remainingMax); ++ }); ++ ++ // 3) run the real delete path and confirm each partial delete preserved the invariant ++ format.delete(stubDesc); ++ ++ assertTrue("expected no violations, got: " + violations, violations.isEmpty()); ++ assertEquals(realFiles.size(), deleted.size()); ++ ++ // 4) confirm the files were deleted oldest-first (the stub deletes nothing so the timestamps are unchanged) ++ assertEquals("each component should be deleted exactly once", realFiles.size(), deletionOrder.size()); ++ for (int i = 1; i < deletionOrder.size(); i++) ++ { ++ Component previous = deletionOrder.get(i - 1); ++ Component current = deletionOrder.get(i); ++ assertTrue("expected " + previous + " to be deleted before " + current + " (deletion order: " + deletionOrder + ')', ++ realFiles.get(previous).lastModified() <= realFiles.get(current).lastModified()); ++ } ++ List<Component> expectedOrder = new ArrayList<>(realFiles.keySet()); ++ Collections.reverse(expectedOrder); ++ assertEquals(expectedOrder, deletionOrder); ++ } ++ ++ @Test ++ public void testBigFormatPartialDeletesLeaveTransactionRecordValid() throws IOException ++ { ++ assertPartialDeletesLeaveTransactionRecordValid(BigFormat.getInstance()); ++ } ++ ++ @Test ++ public void testBtiFormatPartialDeletesLeaveTransactionRecordValid() throws IOException ++ { ++ assertPartialDeletesLeaveTransactionRecordValid(DatabaseDescriptor.getSSTableFormats().get(BtiFormat.NAME)); ++ } ++ ++ private static void assertPartialDeletesLeaveTransactionRecordValid(SSTableFormat<?, ?> format) throws IOException ++ { ++ int numComponents = format.allComponents().size(); ++ ++ // simulate a crash after each possible number of deleted files, including none and all of them ++ for (int deletesBeforeCrash = 0; deletesBeforeCrash <= numComponents; deletesBeforeCrash++) ++ { ++ Descriptor realDesc = newTempDescriptor(format); ++ Map<Component, File> realFiles = createComponentFilesWithReversedUpdateTimes(realDesc, format); ++ ++ // the REMOVE record is logged while all the sstable's files still exist ++ String absolutePath = realDesc.baseFile().withSuffix(String.valueOf(Component.separator)).canonicalPath(); ++ LogRecord record = LogRecord.make(LogRecord.Type.REMOVE, LogRecord.getExistingFiles(absolutePath), numComponents, absolutePath); ++ ++ format.delete(crashAfterDeletes(realDesc, realFiles, deletesBeforeCrash)); ++ assertEquals(numComponents - deletesBeforeCrash, realFiles.values().stream().filter(File::exists).count()); ++ ++ LogFile.verifyRecord(record, LogRecord.getExistingFiles(absolutePath)); ++ assertFalse("record is corrupt after deleting " + deletesBeforeCrash + " of " + numComponents + " files: " + record.status.error, ++ record.status.hasError()); ++ } ++ } ++ ++ private static Descriptor newTempDescriptor(SSTableFormat<?, ?> format) throws IOException ++ { ++ File dataDir = new File(Files.createTempDirectory("LogTransactionPartialDeletesTest").toFile()); ++ return new Descriptor(dataDir, "ks", "cf", new SequenceBasedSSTableId(1), format); ++ } ++ ++ private static Map<Component, File> createComponentFilesWithReversedUpdateTimes(Descriptor desc, SSTableFormat<?, ?> format) ++ { ++ Map<Component, File> files = new LinkedHashMap<>(); ++ for (Component component : format.allComponents()) ++ { ++ File file = desc.fileFor(component); ++ file.parent().createDirectoriesIfNotExists(); ++ assertTrue(file.createFileIfNotExists()); ++ files.put(component, file); ++ } ++ ++ long base = System.currentTimeMillis(); ++ long offset = files.size(); ++ for (File file : files.values()) ++ assertTrue(file.trySetLastModified(base + (offset--) * 1000)); ++ return files; ++ } ++ ++ // Returns a descriptor that crashes after the specified number of file deletions ++ private static Descriptor crashAfterDeletes(Descriptor realDesc, Map<Component, File> realFiles, int deletesBeforeCrash) ++ { ++ AtomicInteger remainingDeletes = new AtomicInteger(deletesBeforeCrash); ++ return new Descriptor(realDesc.version, realDesc.directory, realDesc.ksname, ++ realDesc.cfname, realDesc.id) ++ { ++ @Override ++ public File fileFor(Component component) ++ { ++ File realFile = realFiles.get(component); ++ return new File(realFile.toPath()) ++ { ++ @Override ++ public void deleteIfExists() ++ { ++ if (remainingDeletes.getAndDecrement() <= 0) ++ throw new RuntimeException("simulated crash while deleting " + component); ++ realFile.deleteIfExists(); ++ } ++ }; ++ } ++ }; ++ } ++ ++ private static Descriptor stubDeletionOrder(Descriptor realDesc, Map<Component, File> realFiles, Consumer<Component> onDelete) ++ { ++ return new Descriptor(realDesc.version, realDesc.directory, realDesc.ksname, ++ realDesc.cfname, realDesc.id) ++ { ++ @Override ++ public File fileFor(Component component) ++ { ++ return new File(realFiles.get(component).toPath()) ++ { ++ @Override ++ public void deleteIfExists() ++ { ++ onDelete.accept(component); ++ } ++ }; ++ } ++ }; + } + } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
