From e96b60bd9f814ee1a911e8820b7e255d23e2f24e Mon Sep 17 00:00:00 2001 From: Benjamin Lerer Date: Fri, 13 Nov 2020 14:58:30 +0100 Subject: [PATCH 1/2] Fix the histogram merge of the table metrics patch by Benjamin Lerer; reviewed by Berenguer Blasi and ZhaoYang for CASSANDRA-16259 The logic to merge two histograms in the table metrics was wrong. It was causing an IndexArrayOutOfBondsException if one of the SSTable histograms has less buckets than the previous SSTable. --- CHANGES.txt | 2 +- .../metrics/ColumnFamilyMetrics.java | 43 ++--- .../cassandra/db/ColumnFamilyMetricTest.java | 170 ++++++++++++++---- 3 files changed, 162 insertions(+), 53 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 166cca46ac..a2890e81e2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,5 @@ 2.2.20 - + * Fix the histogram merge of the table metrics (CASSANDRA-16259) 2.2.19 * Package tools/bin scripts as executable (CASSANDRA-16151) diff --git a/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java b/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java index 40ed2e4ed9..3b62e9f819 100644 --- a/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ColumnFamilyMetrics.java @@ -21,9 +21,12 @@ import java.nio.ByteBuffer; import java.util.*; import java.util.concurrent.ConcurrentMap; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.Maps; import com.google.common.collect.Sets; +import org.apache.commons.lang3.ArrayUtils; + import com.codahale.metrics.*; import com.codahale.metrics.Timer; import org.apache.cassandra.db.ColumnFamilyStore; @@ -166,36 +169,34 @@ public class ColumnFamilyMetrics Iterator iterator = sstables.iterator(); if (!iterator.hasNext()) { - return new long[0]; + return ArrayUtils.EMPTY_LONG_ARRAY; } long[] firstBucket = getHistogram.getHistogram(iterator.next()).getBuckets(false); - long[] values = new long[firstBucket.length]; - System.arraycopy(firstBucket, 0, values, 0, values.length); + long[] values = Arrays.copyOf(firstBucket, firstBucket.length); while (iterator.hasNext()) { long[] nextBucket = getHistogram.getHistogram(iterator.next()).getBuckets(false); - if (nextBucket.length > values.length) - { - long[] newValues = new long[nextBucket.length]; - System.arraycopy(firstBucket, 0, newValues, 0, firstBucket.length); - for (int i = 0; i < newValues.length; i++) - { - newValues[i] += nextBucket[i]; - } - values = newValues; - } - else - { - for (int i = 0; i < values.length; i++) - { - values[i] += nextBucket[i]; - } - } + values = addHistogram(values, nextBucket); } return values; } - + + @VisibleForTesting + public static long[] addHistogram(long[] sums, long[] buckets) + { + if (buckets.length > sums.length) + { + sums = Arrays.copyOf(sums, buckets.length); + } + + for (int i = 0; i < buckets.length; i++) + { + sums[i] += buckets[i]; + } + return sums; + } + /** * Creates metrics for given {@link ColumnFamilyStore}. * diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java index 2d89e09528..565bb67bed 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java @@ -20,6 +20,7 @@ package org.apache.cassandra.db; import java.nio.ByteBuffer; import java.util.Collection; + import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.SchemaLoader; @@ -27,12 +28,17 @@ import org.apache.cassandra.Util; import org.apache.cassandra.config.KSMetaData; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.locator.SimpleStrategy; +import org.apache.cassandra.metrics.ColumnFamilyMetrics; import org.apache.cassandra.utils.ByteBufferUtil; import com.google.common.base.Supplier; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.apache.cassandra.Util.cellname; +import static org.apache.cassandra.utils.ByteBufferUtil.bytes; public class ColumnFamilyMetricTest { @@ -55,15 +61,29 @@ public class ColumnFamilyMetricTest store.truncateBlocking(); - assertEquals(0, store.metric.liveDiskSpaceUsed.getCount()); - assertEquals(0, store.metric.totalDiskSpaceUsed.getCount()); + Supplier getLiveDiskSpaceUsed = new Supplier() + { + public Long get() + { + return store.metric.liveDiskSpaceUsed.getCount(); + } + }; + + Supplier getTotalDiskSpaceUsed = new Supplier() + { + public Long get() + { + return store.metric.totalDiskSpaceUsed.getCount(); + } + }; + + Util.spinAssertEquals(0L, getLiveDiskSpaceUsed, 30); + Util.spinAssertEquals(0L, getTotalDiskSpaceUsed, 30); for (int j = 0; j < 10; j++) { ByteBuffer key = ByteBufferUtil.bytes(String.valueOf(j)); - Mutation rm = new Mutation("Keyspace1", key); - rm.add("Standard2", cellname("0"), ByteBufferUtil.EMPTY_BYTE_BUFFER, j); - rm.apply(); + applyStandard2Mutation(key, ByteBufferUtil.EMPTY_BYTE_BUFFER, j); } store.forceBlockingFlush(); Collection sstables = store.getSSTables(); @@ -80,26 +100,8 @@ public class ColumnFamilyMetricTest store.truncateBlocking(); // after truncate, size metrics should be down to 0 - Util.spinAssertEquals( - 0L, - new Supplier() - { - public Long get() - { - return store.metric.liveDiskSpaceUsed.getCount(); - } - }, - 30); - Util.spinAssertEquals( - 0L, - new Supplier() - { - public Long get() - { - return store.metric.totalDiskSpaceUsed.getCount(); - } - }, - 30); + Util.spinAssertEquals(0L, getLiveDiskSpaceUsed, 30); + Util.spinAssertEquals(0L, getTotalDiskSpaceUsed, 30); store.enableAutoCompaction(); } @@ -114,19 +116,125 @@ public class ColumnFamilyMetricTest store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); ByteBuffer key = ByteBufferUtil.bytes(4242); - Mutation m = new Mutation("Keyspace1", key); - m.add("Standard2", cellname("0"), ByteBufferUtil.bytes("0"), 0); - m.apply(); + applyStandard2Mutation(key, ByteBufferUtil.bytes("0"), 0); // The histogram should not have overflowed on the first write store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); - m = new Mutation("Keyspace1", key); // smallest time delta that would overflow the histogram if unfiltered - m.add("Standard2", cellname("0"), ByteBufferUtil.bytes("1"), 18165375903307L); - m.apply(); + applyStandard2Mutation(key, ByteBufferUtil.bytes("1"), 18165375903307L); // CASSANDRA-11117 - update with large timestamp delta should not overflow the histogram store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); } + + @Test + public void testEstimatedColumnCountHistogramAndEstimatedRowSizeHistogram() + { + Keyspace keyspace = Keyspace.open("Keyspace1"); + ColumnFamilyStore store = keyspace.getColumnFamilyStore("Standard2"); + + store.disableAutoCompaction(); + + try + { + // Ensure that there is no SSTables + store.truncateBlocking(); + + assertArrayEquals(new long[0], store.metric.estimatedColumnCountHistogram.getValue()); + + applyStandard2Mutation(bytes(0), bytes(0), 0); + applyStandard2Mutation(bytes(1), bytes(1), 0); + + // Flushing first SSTable + store.forceBlockingFlush(); + + long[] estimatedColumnCountHistogram = store.metric.estimatedColumnCountHistogram.getValue(); + assertNumberOfNonZeroValue(estimatedColumnCountHistogram, 1); + assertEquals(2, estimatedColumnCountHistogram[0]); //2 rows of one cell in 1 SSTable + + long[] estimatedRowSizeHistogram = store.metric.estimatedRowSizeHistogram.getValue(); + // Due to the timestamps we cannot guaranty the size of the row. So we can only check the number of histogram updates. + assertEquals(sumValues(estimatedRowSizeHistogram), 2); + + applyStandard2Mutation(bytes(2), bytes(2), 0); + + // Flushing second SSTable + store.forceBlockingFlush(); + + estimatedColumnCountHistogram = store.metric.estimatedColumnCountHistogram.getValue(); + assertNumberOfNonZeroValue(estimatedColumnCountHistogram, 1); + assertEquals(3, estimatedColumnCountHistogram[0]); //2 rows of one cell in the first SSTable and 1 row of one cell int the second sstable + + estimatedRowSizeHistogram = store.metric.estimatedRowSizeHistogram.getValue(); + assertEquals(sumValues(estimatedRowSizeHistogram), 3); + } + finally + { + store.enableAutoCompaction(); + } + } + + @Test + public void testAddHistogram() + { + long[] sums = new long[] {0, 0, 0}; + long[] smaller = new long[] {1, 2}; + + long[] result = ColumnFamilyMetrics.addHistogram(sums, smaller); + assertTrue(result == sums); // Check that we did not create a new array + assertArrayEquals(new long[]{1, 2, 0}, result); + + long[] equal = new long[] {5, 6, 7}; + + result = ColumnFamilyMetrics.addHistogram(sums, equal); + assertTrue(result == sums); // Check that we did not create a new array + assertArrayEquals(new long[]{6, 8, 7}, result); + + long[] empty = new long[0]; + + result = ColumnFamilyMetrics.addHistogram(sums, empty); + assertTrue(result == sums); // Check that we did not create a new array + assertArrayEquals(new long[]{6, 8, 7}, result); + + long[] greater = new long[] {4, 3, 2, 1}; + result = ColumnFamilyMetrics.addHistogram(sums, greater); + assertFalse(result == sums); // Check that we did not create a new array + assertArrayEquals(new long[]{10, 11, 9, 1}, result); + } + + private static void applyStandard2Mutation(ByteBuffer pk, ByteBuffer value, long timestamp) + { + Mutation m = new Mutation("Keyspace1", pk); + m.add("Standard2", cellname("0"), value, timestamp); + m.apply(); + } + + private static void assertNumberOfNonZeroValue(long[] array, int expectedCount) + { + int actualCount = countNonZeroValues(array); + assertEquals("Unexpected number of non zero values. (expected: " + expectedCount + ", actual: " + actualCount + ")", + expectedCount, actualCount); + } + + private static int countNonZeroValues(long[] array) + { + int count = 0; + for (long value : array) + { + if (value != 0) + count++; + } + return count; + } + + private static long sumValues(long[] array) + { + long sum = 0; + for (long value : array) + { + sum += value; + } + return sum; + } } From 6baeb9247b9156c08305326a77d8c27510d6cb40 Mon Sep 17 00:00:00 2001 From: Benjamin Lerer Date: Thu, 19 Nov 2020 16:08:52 +0100 Subject: [PATCH 2/2] Fix the counting of cells per partition patch by Benjamin Lerer; reviewed by Berenguer Blasi and ZhaoYang for CASSANDRA-16259 The number of cells per partition is updated upon the closing of the RowIterator. As UnfilteredRowIterator.close() was called twice the number of cells could have been counted twice (depending on the RowIterator implementation). --- CHANGES.txt | 1 + .../cassandra/db/transform/BaseIterator.java | 30 ++++++++++++++++--- 2 files changed, 27 insertions(+), 4 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 1a5c72d48f..bb968e8138 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.24: + * Fix the counting of cells per partition (CASSANDRA-16259) * Fix serial read/non-applying CAS linearizability (CASSANDRA-12126) * Avoid potential NPE in JVMStabilityInspector (CASSANDRA-16294) * Improved check of num_tokens against the length of initial_token (CASSANDRA-14477) diff --git a/src/java/org/apache/cassandra/db/transform/BaseIterator.java b/src/java/org/apache/cassandra/db/transform/BaseIterator.java index d00e406d73..8c938a38f8 100644 --- a/src/java/org/apache/cassandra/db/transform/BaseIterator.java +++ b/src/java/org/apache/cassandra/db/transform/BaseIterator.java @@ -43,6 +43,10 @@ abstract class BaseIterator, O exten // Signals that the current child iterator has been signalled to stop. Stop stopChild; + // Multiple calls to close can have some side effects on the Transformations. By consequence if the iterator is + // already closed we want to ignore extra calls to close. + private boolean closed; + static class Stop { // TODO: consider moving "next" into here, so that a stop() when signalled outside of a function call (e.g. in attach) @@ -83,14 +87,32 @@ abstract class BaseIterator, O exten public final void close() { + // If close has already been called we want to ignore other calls + if (closed) + return; + + closed = true; + Throwable fail = runOnClose(length); if (next instanceof AutoCloseable) { - try { ((AutoCloseable) next).close(); } - catch (Throwable t) { fail = merge(fail, t); } + try + { + ((AutoCloseable) next).close(); + } + catch (Throwable t) + { + fail = merge(fail, t); + } + } + try + { + input.close(); + } + catch (Throwable t) + { + fail = merge(fail, t); } - try { input.close(); } - catch (Throwable t) { fail = merge(fail, t); } maybeFail(fail); }