diff --git a/CHANGES.txt b/CHANGES.txt index 430654948a..4a7d8ca817 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -99,6 +99,7 @@ * Log failed host when preparing incremental repair (CASSANDRA-8228) * Force config client mode in CQLSSTableWriter (CASSANDRA-8281) Merged from 2.0: + * Check for available disk space before starting a compaction (CASSANDRA-8562) * Fix DISTINCT queries with LIMITs or paging when some partitions contain only tombstones (CASSANDRA-8490) * Introduce background cache refreshing to permissions cache diff --git a/src/java/org/apache/cassandra/db/Directories.java b/src/java/org/apache/cassandra/db/Directories.java index 4c4078c106..fa9b3206e7 100644 --- a/src/java/org/apache/cassandra/db/Directories.java +++ b/src/java/org/apache/cassandra/db/Directories.java @@ -345,6 +345,24 @@ public class Directories Collections.sort(candidates); } + public boolean hasAvailableDiskSpace(long estimatedSSTables, long expectedTotalWriteSize) + { + long writeSize = expectedTotalWriteSize / estimatedSSTables; + long totalAvailable = 0L; + + for (DataDirectory dataDir : dataDirectories) + { + if (BlacklistedDirectories.isUnwritable(getLocationForDisk(dataDir))) + continue; + DataDirectoryCandidate candidate = new DataDirectoryCandidate(dataDir); + // exclude directory if its total writeSize does not fit to data directory + if (candidate.availableSpace < writeSize) + continue; + totalAvailable += candidate.availableSpace; + } + return totalAvailable > expectedTotalWriteSize; + } + public static File getSnapshotDirectory(Descriptor desc, String snapshotName) { return getSnapshotDirectory(desc.directory, snapshotName); diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index 8b5058be18..dfbdc22946 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -19,6 +19,7 @@ package org.apache.cassandra.db.compaction; import java.io.File; import java.io.IOException; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; import java.util.Iterator; @@ -88,12 +89,14 @@ public class CompactionTask extends AbstractCompactionTask logger.warn("insufficient space to compact all requested files {}", StringUtils.join(sstables, ", ")); // Note that we have removed files that are still marked as compacting. // This suboptimal but ok since the caller will unmark all the sstables at the end. - return sstables.remove(cfs.getMaxSizeFile(sstables)); - } - else - { - return false; + SSTableReader removedSSTable = cfs.getMaxSizeFile(sstables); + if (sstables.remove(removedSSTable)) + { + cfs.getDataTracker().unmarkCompacting(Arrays.asList(removedSSTable)); + return true; + } } + return false; } /** @@ -117,6 +120,11 @@ public class CompactionTask extends AbstractCompactionTask if (DatabaseDescriptor.isSnapshotBeforeCompaction()) cfs.snapshotWithoutFlush(System.currentTimeMillis() + "-compact-" + cfs.name); + // note that we need to do a rough estimate early if we can fit the compaction on disk - this is pessimistic, but + // since we might remove sstables from the compaction in checkAvailableDiskSpace it needs to be done here + long earlySSTableEstimate = Math.max(1, cfs.getExpectedCompactedFileSize(sstables, compactionType) / strategy.getMaxSSTableBytes()); + checkAvailableDiskSpace(earlySSTableEstimate); + // sanity check: all sstables must belong to the same cfs assert !Iterables.any(sstables, new Predicate() { @@ -149,7 +157,7 @@ public class CompactionTask extends AbstractCompactionTask Set actuallyCompact = Sets.difference(sstables, controller.getFullyExpiredSSTables()); long estimatedTotalKeys = Math.max(cfs.metadata.getMinIndexInterval(), SSTableReader.getApproximateKeyCount(actuallyCompact)); - long estimatedSSTables = Math.max(1, SSTableReader.getTotalBytes(actuallyCompact) / strategy.getMaxSSTableBytes()); + long estimatedSSTables = Math.max(1, cfs.getExpectedCompactedFileSize(actuallyCompact, compactionType) / strategy.getMaxSSTableBytes()); long keysPerSSTable = (long) Math.ceil((double) estimatedTotalKeys / estimatedSSTables); SSTableFormat.Type sstableFormat = getFormatType(sstables); @@ -278,6 +286,15 @@ public class CompactionTask extends AbstractCompactionTask return minRepairedAt; } + protected void checkAvailableDiskSpace(long estimatedSSTables) + { + while (!getDirectories().hasAvailableDiskSpace(estimatedSSTables, getExpectedWriteSize())) + { + if (!reduceScopeForLimitedSpace()) + throw new RuntimeException(String.format("Not enough space for compaction, estimated sstables = %d, expected write size = %d", estimatedSSTables, getExpectedWriteSize())); + } + } + private SSTableWriter createCompactionWriter(File sstableDirectory, long keysPerSSTable, long repairedAt, SSTableFormat.Type type) { return SSTableWriter.create(Descriptor.fromFilename(cfs.getTempSSTablePath(sstableDirectory), type), diff --git a/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java b/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java index 4188f6e3fc..925efd68ac 100644 --- a/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java +++ b/src/java/org/apache/cassandra/io/util/DiskAwareRunnable.java @@ -24,13 +24,7 @@ public abstract class DiskAwareRunnable extends WrappedRunnable { protected Directories.DataDirectory getWriteDirectory(long writeSize) { - Directories.DataDirectory directory; - while (true) - { - directory = getDirectories().getWriteableLocation(writeSize); - if (directory != null || !reduceScopeForLimitedSpace()) - break; - } + Directories.DataDirectory directory = getDirectories().getWriteableLocation(writeSize); if (directory == null) throw new RuntimeException("Insufficient disk space to write " + writeSize + " bytes"); @@ -42,13 +36,4 @@ public abstract class DiskAwareRunnable extends WrappedRunnable * @return Directories instance for the CF. */ protected abstract Directories getDirectories(); - - /** - * Called if no disk is available with free space for the full write size. - * @return true if the scope of the task was successfully reduced. - */ - public boolean reduceScopeForLimitedSpace() - { - return false; - } }