Repository: cassandra Updated Branches: refs/heads/trunk 716264c72 -> fbbedcee8
Improve compaction log patch by tjake; reviewed by Marcus Eriksson for CASSANDRA-12080 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/fbbedcee Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/fbbedcee Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/fbbedcee Branch: refs/heads/trunk Commit: fbbedcee81920d17ac2b857eefe1c2b4a9d94b9f Parents: 716264c Author: T Jake Luciani <j...@apache.org> Authored: Thu Jun 23 12:23:29 2016 -0400 Committer: T Jake Luciani <j...@apache.org> Committed: Mon Jun 27 10:41:48 2016 -0400 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../cassandra/db/compaction/CompactionIterator.java | 7 +++++++ .../apache/cassandra/db/compaction/CompactionTask.java | 10 +++++++++- 3 files changed, 17 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbbedcee/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index f441f8b..12a6ddb 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.8 + * Improve details in compaction log message (CASSANDRA-12080) * Allow unset values in CQLSSTableWriter (CASSANDRA-11911) * Chunk cache to request compressor-compatible buffers if pool space is exhausted (CASSANDRA-11993) * Remove DatabaseDescriptor dependencies from SequentialWriter (CASSANDRA-11579) http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbbedcee/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index d39da2a..0111aec 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -63,6 +63,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte private final long totalBytes; private long bytesRead; + private long totalSourceCQLRows; /* * counters for merged rows. @@ -136,6 +137,11 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte return mergeCounters; } + public long getTotalSourceCQLRows() + { + return totalSourceCQLRows; + } + private UnfilteredPartitionIterators.MergeListener listener() { return new UnfilteredPartitionIterators.MergeListener() @@ -287,6 +293,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte @Override protected void updateProgress() { + totalSourceCQLRows++; if ((++compactedUnfiltered) % UNFILTERED_TO_UPDATE_PROGRESS == 0) updateBytesRead(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbbedcee/src/java/org/apache/cassandra/db/compaction/CompactionTask.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java index b3a94bb..cb6faab 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionTask.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionTask.java @@ -160,6 +160,7 @@ public class CompactionTask extends AbstractCompactionTask Collection<SSTableReader> newSStables; long[] mergedRowCounts; + long totalSourceCQLRows; // SSTableScanners need to be closed before markCompactedSSTablesReplaced call as scanners contain references // to both ifile and dfile and SSTR will throw deletion errors on Windows if it tries to delete before scanner is closed. @@ -203,6 +204,8 @@ public class CompactionTask extends AbstractCompactionTask collector.finishCompaction(ci); mergedRowCounts = ci.getMergedRowCounts(); + + totalSourceCQLRows = ci.getTotalSourceCQLRows(); } } @@ -219,8 +222,11 @@ public class CompactionTask extends AbstractCompactionTask newSSTableNames.append(reader.descriptor.baseFilename()).append(","); long totalSourceRows = 0; + for (int i = 0; i < mergedRowCounts.length; i++) + totalSourceRows += mergedRowCounts[i] * (i + 1); + String mergeSummary = updateCompactionHistory(cfs.keyspace.getName(), cfs.getColumnFamilyName(), mergedRowCounts, startsize, endsize); - logger.debug(String.format("Compacted (%s) %d sstables to [%s] to level=%d. %s to %s (~%d%% of original) in %,dms. Throughput = %s. %,d total partitions merged to %,d. Partition merge counts were {%s}", + logger.debug(String.format("Compacted (%s) %d sstables to [%s] to level=%d. %s to %s (~%d%% of original) in %,dms. Read Throughput = %s, Write Throughput = %s, Row Throughput = ~%,d/s. %,d total partitions merged to %,d. Partition merge counts were {%s}", taskId, transaction.originals().size(), newSSTableNames.toString(), @@ -229,7 +235,9 @@ public class CompactionTask extends AbstractCompactionTask FBUtilities.prettyPrintMemory(endsize), (int) (ratio * 100), dTime, + FBUtilities.prettyPrintMemoryPerSecond(startsize, durationInNano), FBUtilities.prettyPrintMemoryPerSecond(endsize, durationInNano), + (int) totalSourceCQLRows / (TimeUnit.NANOSECONDS.toSeconds(durationInNano) + 1), totalSourceRows, totalKeysWritten, mergeSummary));