Skip to content

Commit b6f175b

Browse files
netudimanetudima
committed
Merge branch 'cassandra-6.0' into trunk
* cassandra-6.0: Uncompressed size is being used for compressed tables in maintenance operations
2 parents 0625a4d + 51a58ef commit b6f175b

29 files changed

Lines changed: 164 additions & 67 deletions

‎CHANGES.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ Merged from 6.0:
66
* Differentiate between legitimate cases where the first entry is the same as the last entry and empty bounds in SSTableCursorWriter#addIndexBlock() (CASSANDRA-21255)
77
* Introduce minimum_threshold for data resurrection startup check (CASSANDRA-21293)
88
Merged from 5.0:
9+
* Use estimated compressed size for tables to check if there is enough free space for a compaction (CASSANDRA-21245)
910
* Fix failing select on system_views.settings for non-string keys (CASSANDRA-21348)
1011

1112

‎src/java/org/apache/cassandra/cache/AutoSavingCache.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -330,6 +330,7 @@ else if (cacheType == CacheService.CacheType.COUNTER_CACHE)
330330
type,
331331
0,
332332
keysEstimate,
333+
keysEstimate,
333334
Unit.KEYS,
334335
nextTimeUUID(),
335336
getCacheDataPath(CURRENT_VERSION).toPath().toString());
@@ -344,7 +345,8 @@ public CompactionInfo getCompactionInfo()
344345
{
345346
// keyset can change in size, thus total can too
346347
// TODO need to check for this one... was: info.forProgress(keysWritten, Math.max(keysWritten, keys.size()));
347-
return info.forProgress(keysWritten, Math.max(keysWritten, keysEstimate));
348+
long totalKeys = Math.max(keysWritten, keysEstimate);
349+
return info.forProgress(keysWritten, totalKeys, totalKeys);
348350
}
349351

350352
public void saveCache()

‎src/java/org/apache/cassandra/db/compaction/ActiveCompactions.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,13 +49,14 @@ public void finishCompaction(CompactionInfo.Holder ci)
4949
{
5050
compactions.remove(ci);
5151
CompactionManager.instance.getMetrics().bytesCompacted.inc(ci.getCompactionInfo().getTotal());
52+
CompactionManager.instance.getMetrics().compressedBytesCompacted.inc(ci.getCompactionInfo().getTotalCompressed());
5253
CompactionManager.instance.getMetrics().totalCompactionsCompleted.mark();
5354
}
5455

5556
/**
5657
* Get the estimated number of bytes remaining to write per sstable directory
5758
*/
58-
public Map<File, Long> estimatedRemainingWriteBytes()
59+
public Map<File, Long> estimatedRemainingWriteToDiskBytes()
5960
{
6061
synchronized (compactions)
6162
{
@@ -66,7 +67,7 @@ public Map<File, Long> estimatedRemainingWriteBytes()
6667
List<File> directories = compactionInfo.getTargetDirectories();
6768
if (directories == null || directories.isEmpty())
6869
continue;
69-
long remainingWriteBytesPerDataDir = compactionInfo.estimatedRemainingWriteBytes() / directories.size();
70+
long remainingWriteBytesPerDataDir = compactionInfo.estimatedRemainingWriteToDiskBytes() / directories.size();
7071
for (File directory : directories)
7172
writeBytesPerSSTableDir.merge(directory, remainingWriteBytesPerDataDir, Long::sum);
7273
}

‎src/java/org/apache/cassandra/db/compaction/CompactionInfo.java‎

Lines changed: 27 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,7 @@ public final class CompactionInfo
4242
public static final String COLUMNFAMILY = "columnfamily";
4343
public static final String COMPLETED = "completed";
4444
public static final String TOTAL = "total";
45+
public static final String TOTAL_COMPRESSED = "totalCompressed";
4546
public static final String TASK_TYPE = "taskType";
4647
public static final String UNIT = "unit";
4748
public static final String COMPACTION_ID = "compactionId";
@@ -52,55 +53,57 @@ public final class CompactionInfo
5253
private final OperationType tasktype;
5354
private final long completed;
5455
private final long total;
56+
private final long totalCompressed;
5557
private final Unit unit;
5658
private final TimeUUID compactionId;
5759
private final ImmutableSet<SSTableReader> sstables;
5860
private final String targetDirectory;
5961

60-
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, Unit unit, TimeUUID compactionId, Collection<? extends SSTableReader> sstables, String targetDirectory)
62+
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, long totalCompressed, Unit unit, TimeUUID compactionId, Collection<? extends SSTableReader> sstables, String targetDirectory)
6163
{
6264
this.tasktype = tasktype;
6365
this.completed = completed;
6466
this.total = total;
67+
this.totalCompressed = totalCompressed;
6568
this.metadata = metadata;
6669
this.unit = unit;
6770
this.compactionId = compactionId;
6871
this.sstables = ImmutableSet.copyOf(sstables);
6972
this.targetDirectory = targetDirectory;
7073
}
7174

72-
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, TimeUUID compactionId, Collection<SSTableReader> sstables, String targetDirectory)
75+
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, long totalCompressed, TimeUUID compactionId, Collection<SSTableReader> sstables, String targetDirectory)
7376
{
74-
this(metadata, tasktype, completed, total, Unit.BYTES, compactionId, sstables, targetDirectory);
77+
this(metadata, tasktype, completed, total, totalCompressed, Unit.BYTES, compactionId, sstables, targetDirectory);
7578
}
7679

77-
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, TimeUUID compactionId, Collection<? extends SSTableReader> sstables)
80+
public CompactionInfo(TableMetadata metadata, OperationType tasktype, long completed, long total, long totalCompressed, TimeUUID compactionId, Collection<? extends SSTableReader> sstables)
7881
{
79-
this(metadata, tasktype, completed, total, Unit.BYTES, compactionId, sstables, null);
82+
this(metadata, tasktype, completed, total, totalCompressed, Unit.BYTES, compactionId, sstables, null);
8083
}
8184

8285
/**
8386
* Special compaction info where we always need to cancel the compaction - for example ViewBuilderTask where we don't know
8487
* the sstables at construction
8588
*/
86-
public static CompactionInfo withoutSSTables(TableMetadata metadata, OperationType tasktype, long completed, long total, Unit unit, TimeUUID compactionId)
89+
public static CompactionInfo withoutSSTables(TableMetadata metadata, OperationType tasktype, long completed, long total, long totalCompressed, Unit unit, TimeUUID compactionId)
8790
{
88-
return withoutSSTables(metadata, tasktype, completed, total, unit, compactionId, null);
91+
return withoutSSTables(metadata, tasktype, completed, total, totalCompressed, unit, compactionId, null);
8992
}
9093

9194
/**
9295
* Special compaction info where we always need to cancel the compaction - for example AutoSavingCache where we don't know
9396
* the sstables at construction
9497
*/
95-
public static CompactionInfo withoutSSTables(TableMetadata metadata, OperationType tasktype, long completed, long total, Unit unit, TimeUUID compactionId, String targetDirectory)
98+
public static CompactionInfo withoutSSTables(TableMetadata metadata, OperationType tasktype, long completed, long total, long totalCompressed, Unit unit, TimeUUID compactionId, String targetDirectory)
9699
{
97-
return new CompactionInfo(metadata, tasktype, completed, total, unit, compactionId, ImmutableSet.of(), targetDirectory);
100+
return new CompactionInfo(metadata, tasktype, completed, total, totalCompressed, unit, compactionId, ImmutableSet.of(), targetDirectory);
98101
}
99102

100103
/** @return A copy of this CompactionInfo with updated progress. */
101-
public CompactionInfo forProgress(long complete, long total)
104+
public CompactionInfo forProgress(long complete, long total, long totalCompressed)
102105
{
103-
return new CompactionInfo(metadata, tasktype, complete, total, unit, compactionId, sstables, targetDirectory);
106+
return new CompactionInfo(metadata, tasktype, complete, total, totalCompressed, unit, compactionId, sstables, targetDirectory);
104107
}
105108

106109
public Optional<String> getKeyspace()
@@ -128,6 +131,11 @@ public long getTotal()
128131
return total;
129132
}
130133

134+
public long getTotalCompressed()
135+
{
136+
return totalCompressed;
137+
}
138+
131139
public OperationType getTaskType()
132140
{
133141
return tasktype;
@@ -183,12 +191,16 @@ public String targetDirectory()
183191
/**
184192
* Note that this estimate is based on the amount of data we have left to read - it assumes input
185193
* size == output size for a compaction, which is not really true, but should most often provide a worst case
186-
* remaining write size.
194+
* remaining write size. We also scale by the effective compression ratio since total/completed are for the uncompressed size.
187195
*/
188-
public long estimatedRemainingWriteBytes()
196+
public long estimatedRemainingWriteToDiskBytes()
189197
{
190198
if (unit == Unit.BYTES && tasktype.writesData)
191-
return getTotal() - getCompleted();
199+
{
200+
final long total = getTotal();
201+
double compressionRatio = total == 0 ? 1 : ((double) totalCompressed / (double)total);
202+
return (long)(compressionRatio * (total - getCompleted()));
203+
}
192204
return 0;
193205
}
194206

@@ -216,6 +228,7 @@ public Map<String, String> asMap()
216228
ret.put(COLUMNFAMILY, getTable().orElse(null));
217229
ret.put(COMPLETED, Long.toString(completed));
218230
ret.put(TOTAL, Long.toString(total));
231+
ret.put(TOTAL_COMPRESSED, Long.toString(totalCompressed));
219232
ret.put(TASK_TYPE, tasktype.toString());
220233
ret.put(UNIT, unit.toString());
221234
ret.put(COMPACTION_ID, compactionId == null ? "" : compactionId.toString());

‎src/java/org/apache/cassandra/db/compaction/CompactionIterator.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -158,6 +158,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
158158
private final long nowInSec;
159159
private final TimeUUID compactionId;
160160
private final long totalBytes;
161+
private final long totalCompressedBytes;
161162
private long bytesRead;
162163
private long totalSourceCQLRows;
163164

@@ -231,9 +232,14 @@ public CompactionIterator(OperationType type,
231232
this.bytesRead = 0;
232233

233234
long bytes = 0;
235+
long compressedBytes = 0;
234236
for (ISSTableScanner scanner : scanners)
237+
{
235238
bytes += scanner.getLengthInBytes();
239+
compressedBytes += scanner.getCompressedLengthInBytes();
240+
}
236241
this.totalBytes = bytes;
242+
this.totalCompressedBytes = compressedBytes;
237243
this.mergeCounters = new long[scanners.size()];
238244
// note that we leak `this` from the constructor when calling beginCompaction below, this means we have to get the sstables before
239245
// calling that to avoid a NPE.
@@ -281,6 +287,7 @@ public CompactionInfo getCompactionInfo()
281287
type,
282288
bytesRead,
283289
totalBytes,
290+
totalCompressedBytes,
284291
compactionId,
285292
sstables,
286293
targetDirectory);

‎src/java/org/apache/cassandra/db/compaction/CompactionTask.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -489,7 +489,7 @@ protected boolean buildCompactionCandidatesForAvailableDiskSpace(final Set<SSTab
489489
for (File directory : newCompactionDatadirs)
490490
expectedNewWriteSize.put(directory, writeSizePerOutputDatadir);
491491

492-
Map<File, Long> expectedWriteSize = CompactionManager.instance.active.estimatedRemainingWriteBytes();
492+
Map<File, Long> expectedWriteSize = CompactionManager.instance.active.estimatedRemainingWriteToDiskBytes();
493493

494494
// todo: abort streams if they block compactions
495495
if (cfs.getDirectories().hasDiskSpaceForCompactionsAndStreams(expectedNewWriteSize, expectedWriteSize))

‎src/java/org/apache/cassandra/db/compaction/CursorCompactor.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -210,6 +210,7 @@ private static void logDebugReason(TableMetadata metadata, String reason)
210210
private final long nowInSec;
211211
private final TimeUUID compactionId;
212212
private final long totalInputBytes;
213+
private final long totalCompressedInputBytes;
213214
private final StatefulCursor[] sstableCursors;
214215
private final boolean[] sstableCursorsEqualsNext;
215216
private final boolean hasStaticColumns;
@@ -268,9 +269,14 @@ private CursorCompactor(OperationType type,
268269
this.compactionId = compactionId;
269270

270271
long inputBytes = 0;
272+
long compressedInputBytes = 0;
271273
for (ISSTableScanner scanner : scanners)
274+
{
272275
inputBytes += scanner.getLengthInBytes();
276+
compressedInputBytes += scanner.getCompressedLengthInBytes();
277+
}
273278
this.totalInputBytes = inputBytes;
279+
this.totalCompressedInputBytes = compressedInputBytes;
274280
this.partitionMergeCounters = new long[scanners.size()];
275281
this.staticRowMergeCounters = new long[partitionMergeCounters.length];
276282
this.rowMergeCounters = new long[partitionMergeCounters.length];
@@ -1461,6 +1467,7 @@ public CompactionInfo getCompactionInfo()
14611467
type,
14621468
getBytesRead(),
14631469
totalInputBytes,
1470+
totalCompressedInputBytes,
14641471
compactionId,
14651472
sstables,
14661473
targetDirectory);
@@ -1666,4 +1673,4 @@ else if (cmp == 0) {
16661673
}
16671674
preSortedArray[insertInto] = newElement;
16681675
}
1669-
}
1676+
}

‎src/java/org/apache/cassandra/db/view/ViewBuilderTask.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -205,13 +205,13 @@ public CompactionInfo getCompactionInfo()
205205
if (range.left.getPartitioner().splitter().isPresent())
206206
{
207207
long progress = prevToken == null ? 0 : Math.round(prevToken.getPartitioner().splitter().get().positionInRange(prevToken, range) * 1000);
208-
return CompactionInfo.withoutSSTables(baseCfs.metadata(), OperationType.VIEW_BUILD, progress, 1000, Unit.RANGES, compactionId);
208+
return CompactionInfo.withoutSSTables(baseCfs.metadata(), OperationType.VIEW_BUILD, progress, 1000, 1000, Unit.RANGES, compactionId);
209209
}
210210

211211
// When there is no splitter, estimate based on number of total keys but
212212
// take the max with keysBuilt + 1 to avoid having more completed than total
213213
long keysTotal = Math.max(keysBuilt + 1, baseCfs.estimatedKeysForRange(range));
214-
return CompactionInfo.withoutSSTables(baseCfs.metadata(), OperationType.VIEW_BUILD, keysBuilt, keysTotal, Unit.KEYS, compactionId);
214+
return CompactionInfo.withoutSSTables(baseCfs.metadata(), OperationType.VIEW_BUILD, keysBuilt, keysTotal, keysTotal, Unit.KEYS, compactionId);
215215
}
216216

217217
@Override

‎src/java/org/apache/cassandra/db/virtual/SSTableTasksTable.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@ final class SSTableTasksTable extends AbstractVirtualTable
3939
private final static String PROGRESS = "progress";
4040
private final static String SSTABLES = "sstables";
4141
private final static String TOTAL = "total";
42+
private final static String TOTAL_COMPRESSED = "total_compressed";
4243
private final static String UNIT = "unit";
4344
private final static String TARGET_DIRECTORY = "target_directory";
4445

@@ -56,6 +57,7 @@ final class SSTableTasksTable extends AbstractVirtualTable
5657
.addRegularColumn(PROGRESS, LongType.instance)
5758
.addRegularColumn(SSTABLES, Int32Type.instance)
5859
.addRegularColumn(TOTAL, LongType.instance)
60+
.addRegularColumn(TOTAL_COMPRESSED, LongType.instance)
5961
.addRegularColumn(UNIT, UTF8Type.instance)
6062
.addRegularColumn(TARGET_DIRECTORY, UTF8Type.instance)
6163
.build());
@@ -69,6 +71,7 @@ public DataSet data()
6971
{
7072
long completed = task.getCompleted();
7173
long total = task.getTotal();
74+
long totalCompressed = task.getTotalCompressed();
7275

7376
double completionRatio = total == 0L ? 1.0 : (((double) completed) / total);
7477

@@ -80,6 +83,7 @@ public DataSet data()
8083
.column(PROGRESS, completed)
8184
.column(SSTABLES, task.getSSTables().size())
8285
.column(TOTAL, total)
86+
.column(TOTAL_COMPRESSED, totalCompressed)
8387
.column(UNIT, toLowerCaseLocalized(task.getUnit().toString()))
8488
.column(TARGET_DIRECTORY, task.targetDirectory());
8589
}

‎src/java/org/apache/cassandra/index/accord/RouteSecondaryIndexBuilder.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,7 @@ public class RouteSecondaryIndexBuilder extends SecondaryIndexBuilder
5757
private final boolean isFullRebuild;
5858
private final boolean isInitialBuild;
5959
private final long totalSizeInBytes;
60+
private final long totalCompressedSizeInBytes;
6061
private long bytesProcessed = 0;
6162

6263
public RouteSecondaryIndexBuilder(RouteJournalIndex index,
@@ -72,7 +73,15 @@ public RouteSecondaryIndexBuilder(RouteJournalIndex index,
7273
this.sstables = sstables;
7374
this.isFullRebuild = isFullRebuild;
7475
this.isInitialBuild = isInitialBuild;
75-
this.totalSizeInBytes = sstables.stream().mapToLong(SSTableReader::uncompressedLength).sum();
76+
long uncompressedSum = 0L;
77+
long compressedSum = 0L;
78+
for (SSTableReader sstable : sstables)
79+
{
80+
uncompressedSum += sstable.uncompressedLength();
81+
compressedSum += sstable.onDiskLength();
82+
}
83+
this.totalSizeInBytes = uncompressedSum;
84+
this.totalCompressedSizeInBytes = compressedSum;
7685
}
7786

7887
@Override
@@ -82,6 +91,7 @@ public CompactionInfo getCompactionInfo()
8291
OperationType.INDEX_BUILD,
8392
bytesProcessed,
8493
totalSizeInBytes,
94+
totalCompressedSizeInBytes,
8595
compactionId,
8696
sstables);
8797
}

0 commit comments

Comments
 (0)