natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch cassandra-3.11 into trunk

This commit is contained in:
Benjamin Lerer 2020-12-02 14:47:18 +01:00
parent 90b2f3e068 e94459c695
commit 39d2974379
4 changed files with 157 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -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)

29
src/java/org/apache/cassandra/db/transform/BaseIterator.java
View File

@ -43,6 +43,10 @@ abstract class BaseIterator<V, I extends CloseableIterator<? extends V>, 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<V, I extends CloseableIterator<? extends V>, 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);
}

50
src/java/org/apache/cassandra/metrics/TableMetrics.java
View File

@ -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<Long> 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<SSTableReader> 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}.
*

124
test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java
View File

@ -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<SSTableReader> 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