diff --git a/CHANGES.txt b/CHANGES.txt index 6e96738268..a51d4717e2 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -17,6 +17,7 @@ Merged from 3.11: * SASI's `max_compaction_flush_memory_in_mb` settings over 100GB revert to default of 1GB (CASSANDRA-16071) Merged from 3.0: + * 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) @@ -25,6 +26,8 @@ Merged from 3.0: * Wait for schema agreement when bootstrapping (CASSANDRA-15158) * Prevent invoking enable/disable gossip when not in NORMAL (CASSANDRA-16146) * Raise Dynamic Snitch Default Badness Threshold to 1.0 (CASSANDRA-16285) +Merged from 2.2: + * Fix the histogram merge of the table metrics (CASSANDRA-16259) 4.0-beta3 * Segregate Network and Chunk Cache BufferPools and Recirculate Partially Freed Chunks (CASSANDRA-15229) diff --git a/src/java/org/apache/cassandra/db/transform/BaseIterator.java b/src/java/org/apache/cassandra/db/transform/BaseIterator.java index d00e406d73..8d7de474e0 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 call 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,31 @@ 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); } diff --git a/src/java/org/apache/cassandra/metrics/TableMetrics.java b/src/java/org/apache/cassandra/metrics/TableMetrics.java index 0bc66b9b35..eb42331bb1 100644 --- a/src/java/org/apache/cassandra/metrics/TableMetrics.java +++ b/src/java/org/apache/cassandra/metrics/TableMetrics.java @@ -20,11 +20,7 @@ package org.apache.cassandra.metrics; import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.EnumMap; -import java.util.Iterator; -import java.util.List; -import java.util.Set; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.TimeUnit; @@ -35,6 +31,10 @@ import com.google.common.collect.Maps; import com.google.common.collect.Sets; import com.codahale.metrics.Timer; +import com.google.common.annotations.VisibleForTesting; + +import org.apache.commons.lang3.ArrayUtils; + import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.db.Memtable; @@ -62,8 +62,6 @@ import com.codahale.metrics.RatioGauge; */ public class TableMetrics { - public static final long[] EMPTY = new long[0]; - /** Total amount of data stored in the memtable that resides on-heap, including column related overhead and partitions overwritten. */ public final Gauge memtableOnHeapDataSize; /** Total amount of data stored in the memtable that resides off-heap, including column related overhead and partitions overwritten. */ @@ -343,36 +341,34 @@ public class TableMetrics Iterator iterator = sstables.iterator(); if (!iterator.hasNext()) { - return EMPTY; + 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 efd5017736..21417ed787 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java @@ -17,6 +17,8 @@ */ package org.apache.cassandra.db; +import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collection; import org.junit.BeforeClass; @@ -32,11 +34,14 @@ import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.metrics.CassandraMetricsRegistry; +import org.apache.cassandra.metrics.TableMetrics; import org.apache.cassandra.schema.KeyspaceParams; +import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; -import static org.junit.Assert.assertEquals; +import static org.apache.cassandra.utils.ByteBufferUtil.bytes; +import static org.junit.Assert.*; public class ColumnFamilyMetricTest { @@ -58,16 +63,12 @@ public class ColumnFamilyMetricTest cfs.truncateBlocking(); - assertEquals(0, cfs.metric.liveDiskSpaceUsed.getCount()); - assertEquals(0, cfs.metric.totalDiskSpaceUsed.getCount()); + Util.spinAssertEquals(0L, cfs.metric.liveDiskSpaceUsed::getCount, 30); + Util.spinAssertEquals(0L, cfs.metric.totalDiskSpaceUsed::getCount, 30); for (int j = 0; j < 10; j++) { - new RowUpdateBuilder(cfs.metadata(), FBUtilities.timestampMicros(), String.valueOf(j)) - .clustering("0") - .add("val", ByteBufferUtil.EMPTY_BYTE_BUFFER) - .build() - .applyUnsafe(); + applyMutation(cfs.metadata(), String.valueOf(j), ByteBufferUtil.EMPTY_BYTE_BUFFER, FBUtilities.timestampMicros()); } cfs.forceBlockingFlush(); Collection sstables = cfs.getLiveSSTables(); @@ -99,21 +100,13 @@ public class ColumnFamilyMetricTest // This confirms another test/set up did not overflow the histogram store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); - new RowUpdateBuilder(store.metadata(), 0, "4242") - .clustering("0") - .add("val", ByteBufferUtil.bytes("0")) - .build() - .applyUnsafe(); + applyMutation(store.metadata(), "4242", ByteBufferUtil.bytes("0"), 0); // The histogram should not have overflowed on the first write store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); // smallest time delta that would overflow the histogram if unfiltered - new RowUpdateBuilder(store.metadata(), 18165375903307L, "4242") - .clustering("0") - .add("val", ByteBufferUtil.bytes("0")) - .build() - .applyUnsafe(); + applyMutation(store.metadata(), "4242", ByteBufferUtil.bytes("1"), 18165375903307L); // CASSANDRA-11117 - update with large timestamp delta should not overflow the histogram store.metric.colUpdateTimeDeltaHistogram.cf.getSnapshot().get999thPercentile(); @@ -140,6 +133,101 @@ public class ColumnFamilyMetricTest } } + @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()); + + applyMutation(store.metadata(), "0", bytes(0), FBUtilities.timestampMicros()); + applyMutation(store.metadata(), "1", bytes(1), FBUtilities.timestampMicros()); + + // 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.estimatedPartitionSizeHistogram.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); + + applyMutation(store.metadata(), "2", bytes(2), FBUtilities.timestampMicros()); + + // 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.estimatedPartitionSizeHistogram.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 = TableMetrics.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 = TableMetrics.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 = TableMetrics.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 = TableMetrics.addHistogram(sums, greater); + assertFalse(result == sums); // Check that we created a new array + assertArrayEquals(new long[]{10, 11, 9, 1}, result); + } + + private static void applyMutation(TableMetadata metadata, Object pk, ByteBuffer value, long timestamp) + { + new RowUpdateBuilder(metadata, timestamp, pk).clustering("0") + .add("val", value) + .build() + .applyUnsafe(); + } + + private static void assertNumberOfNonZeroValue(long[] array, long expectedCount) + { + long actualCount = Arrays.stream(array).filter(v -> v != 0).count(); + if (expectedCount != actualCount) + fail("Unexpected number of non zero values. (expected: " + expectedCount + ", actual: " + actualCount + " array: " + Arrays.toString(array)+ " )"); + } + + private static long sumValues(long[] array) + { + return Arrays.stream(array).sum(); + } + private static class TestBase extends MetricRegistryListener.Base { @Override