diff --git a/CHANGES.txt b/CHANGES.txt index 93e7a87810..f3df5d052d 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 3.0.23: + * Avoid failing compactions with very large partitions (CASSANDRA-15164) * Use IF NOT EXISTS for index and UDT create statements in snapshot schema files (CASSANDRA-13935) 3.0.22: diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java index 867e9a1855..ff4b06ff77 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java @@ -49,8 +49,8 @@ public class MetadataCollector implements PartitionStatisticsCollector static EstimatedHistogram defaultCellPerPartitionCountHistogram() { - // EH of 114 can track a max value of 2395318855, i.e., > 2B columns - return new EstimatedHistogram(114); + // EH of 118 can track a max value of 4139110981, i.e., > 4B cells + return new EstimatedHistogram(118); } static EstimatedHistogram defaultPartitionSizeHistogram() diff --git a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java index 1994bca198..be6b4307af 100644 --- a/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java +++ b/src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java @@ -26,6 +26,9 @@ import org.apache.cassandra.io.ISerializer; import org.apache.cassandra.io.sstable.format.Version; import org.apache.commons.lang3.builder.EqualsBuilder; import org.apache.commons.lang3.builder.HashCodeBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.db.commitlog.IntervalSet; import org.apache.cassandra.db.commitlog.ReplayPosition; @@ -230,6 +233,8 @@ public class StatsMetadata extends MetadataComponent public static class StatsMetadataSerializer implements IMetadataComponentSerializer { + private static final Logger logger = LoggerFactory.getLogger(StatsMetadataSerializer.class); + public int serializedSize(Version version, StatsMetadata component) throws IOException { int size = 0; @@ -302,9 +307,29 @@ public class StatsMetadata extends MetadataComponent public StatsMetadata deserialize(Version version, DataInputPlus in) throws IOException { EstimatedHistogram partitionSizes = EstimatedHistogram.serializer.deserialize(in); + + if (partitionSizes.isOverflowed()) + { + logger.warn("Deserialized partition size histogram with {} values greater than the maximum of {}. " + + "Clearing the overflow bucket to allow for degraded mean and percentile calculations...", + partitionSizes.overflowCount(), partitionSizes.getLargestBucketOffset()); + + partitionSizes.clearOverflow(); + } + EstimatedHistogram columnCounts = EstimatedHistogram.serializer.deserialize(in); + if (columnCounts.isOverflowed()) + { + logger.warn("Deserialized partition cell count histogram with {} values greater than the maximum of {}. " + + "Clearing the overflow bucket to allow for degraded mean and percentile calculations...", + columnCounts.overflowCount(), columnCounts.getLargestBucketOffset()); + + columnCounts.clearOverflow(); + } + ReplayPosition commitLogLowerBound = ReplayPosition.NONE, commitLogUpperBound; commitLogUpperBound = ReplayPosition.serializer.deserialize(in); + long minTimestamp = in.readLong(); long maxTimestamp = in.readLong(); // We use MAX_VALUE as that's the default value for "no deletion time" diff --git a/src/java/org/apache/cassandra/utils/EstimatedHistogram.java b/src/java/org/apache/cassandra/utils/EstimatedHistogram.java index 8109c98d23..07a41bc1fe 100644 --- a/src/java/org/apache/cassandra/utils/EstimatedHistogram.java +++ b/src/java/org/apache/cassandra/utils/EstimatedHistogram.java @@ -22,12 +22,13 @@ import java.util.Arrays; import java.util.concurrent.atomic.AtomicLongArray; import com.google.common.base.Objects; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.ISerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.slf4j.Logger; public class EstimatedHistogram { @@ -262,11 +263,27 @@ public class EstimatedHistogram } /** - * @return true if this histogram has overflowed -- that is, a value larger than our largest bucket could bound was added + * @return true if a value larger than our largest bucket offset has been recorded, and false otherwise */ public boolean isOverflowed() { - return buckets.get(buckets.length() - 1) > 0; + return overflowCount() > 0; + } + + /** + * @return the number of recorded values larger than the largest bucket offset + */ + public long overflowCount() + { + return buckets.get(buckets.length() - 1); + } + + /** + * Resets the count in the overflow bucket to zero. Subsequent calls to {@link #isOverflowed()} will return false. + */ + public void clearOverflow() + { + buckets.set(buckets.length() - 1, 0); } /** @@ -354,8 +371,16 @@ public class EstimatedHistogram public static class EstimatedHistogramSerializer implements ISerializer { + private static final Logger logger = LoggerFactory.getLogger(EstimatedHistogramSerializer.class); + public void serialize(EstimatedHistogram eh, DataOutputPlus out) throws IOException { + if (eh.isOverflowed()) + { + logger.warn("Serializing a histogram with {} values greater than the maximum of {}...", + eh.overflowCount(), eh.getLargestBucketOffset()); + } + long[] offsets = eh.getBucketOffsets(); long[] buckets = eh.getBuckets(false); out.writeInt(buckets.length); diff --git a/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java b/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java index de12d572a1..3c9a13bb09 100644 --- a/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/metadata/MetadataSerializerTest.java @@ -37,6 +37,7 @@ import org.apache.cassandra.db.commitlog.ReplayPosition; import org.apache.cassandra.dht.RandomPartitioner; import org.apache.cassandra.io.sstable.Component; import org.apache.cassandra.io.sstable.Descriptor; +import org.apache.cassandra.io.sstable.format.SSTableFormat; import org.apache.cassandra.io.sstable.format.Version; import org.apache.cassandra.io.sstable.format.big.BigFormat; import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus; @@ -44,6 +45,8 @@ import org.apache.cassandra.io.util.DataOutputStreamPlus; import org.apache.cassandra.io.util.RandomAccessReader; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; public class MetadataSerializerTest { @@ -67,6 +70,33 @@ public class MetadataSerializerTest } } + @Test + public void testHistogramSterilization() throws IOException + { + Map originalMetadata = constructMetadata(); + + // Modify the histograms to overflow: + StatsMetadata originalStats = (StatsMetadata) originalMetadata.get(MetadataType.STATS); + originalStats.estimatedColumnCount.add(Long.MAX_VALUE); + originalStats.estimatedPartitionSize.add(Long.MAX_VALUE); + assertTrue(originalStats.estimatedColumnCount.isOverflowed()); + assertTrue(originalStats.estimatedPartitionSize.isOverflowed()); + + // Serialize w/ overflowed histograms: + MetadataSerializer serializer = new MetadataSerializer(); + File statsFile = serialize(originalMetadata, serializer, BigFormat.latestVersion); + Descriptor desc = new Descriptor(statsFile.getParentFile(), "", "", 0, SSTableFormat.Type.BIG); + + try (RandomAccessReader in = RandomAccessReader.open(statsFile)) + { + // Deserialie and verify that the two histograms have had their overflow buckets cleared: + Map deserialized = serializer.deserialize(desc, in, EnumSet.allOf(MetadataType.class)); + StatsMetadata deserializedStats = (StatsMetadata)deserialized.get(MetadataType.STATS); + assertFalse(deserializedStats.estimatedColumnCount.isOverflowed()); + assertFalse(deserializedStats.estimatedPartitionSize.isOverflowed()); + } + } + public File serialize(Map metadata, MetadataSerializer serializer, Version version) throws IOException, FileNotFoundException { diff --git a/test/unit/org/apache/cassandra/utils/EstimatedHistogramTest.java b/test/unit/org/apache/cassandra/utils/EstimatedHistogramTest.java index b3fbfb6a28..edc297a641 100644 --- a/test/unit/org/apache/cassandra/utils/EstimatedHistogramTest.java +++ b/test/unit/org/apache/cassandra/utils/EstimatedHistogramTest.java @@ -21,6 +21,8 @@ package org.apache.cassandra.utils; import org.junit.Test; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; public class EstimatedHistogramTest @@ -167,4 +169,14 @@ public class EstimatedHistogramTest assertEquals(1, histogram.percentile(0.99)); } } + + @Test + public void testClearOverflow() + { + EstimatedHistogram histogram = new EstimatedHistogram(1); + histogram.add(100); + assertTrue(histogram.isOverflowed()); + histogram.clearOverflow(); + assertFalse(histogram.isOverflowed()); + } } diff --git a/test/unit/org/apache/cassandra/utils/SerializationsTest.java b/test/unit/org/apache/cassandra/utils/SerializationsTest.java index cf507695bb..9accaa6ce6 100644 --- a/test/unit/org/apache/cassandra/utils/SerializationsTest.java +++ b/test/unit/org/apache/cassandra/utils/SerializationsTest.java @@ -202,7 +202,7 @@ public class SerializationsTest extends AbstractSerializationsTester offsets[i] = i; data[i] = 10 * i; } - data[offsets.length] = 100000; + data[offsets.length] = 100000; // write into the overflow bucket EstimatedHistogram hist2 = new EstimatedHistogram(offsets, data); try (DataOutputStreamPlus out = getOutput("utils.EstimatedHistogram.bin"))