diff --git a/CHANGES.txt b/CHANGES.txt index 87d815b299..0283be667c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 4.0 + * Add Zstd compressor (CASSANDRA-14482) * Fix IR prepare anti-compaction race (CASSANDRA-15027) * Fix SimpleStrategy option validation (CASSANDRA-15007) * Don't try to cancel 2i compactions when starting anticompaction (CASSANDRA-15024) diff --git a/build.xml b/build.xml index 5488f3857d..efcc9a99a2 100644 --- a/build.xml +++ b/build.xml @@ -64,6 +64,7 @@ + @@ -412,6 +413,7 @@ + @@ -1483,7 +1485,7 @@ - + diff --git a/doc/source/architecture/storage_engine.rst b/doc/source/architecture/storage_engine.rst index 2a95d93c7b..23b738de75 100644 --- a/doc/source/architecture/storage_engine.rst +++ b/doc/source/architecture/storage_engine.rst @@ -55,7 +55,7 @@ Commitlogs are an append only log of all mutations local to a Cassandra node. An *Default Value:* /var/lib/cassandra/commitlog -- ``commitlog_compression``: Compression to apply to the commitlog. If omitted, the commit log will be written uncompressed. LZ4, Snappy, and Deflate compressors are supported. +- ``commitlog_compression``: Compression to apply to the commitlog. If omitted, the commit log will be written uncompressed. LZ4, Snappy, Deflate and Zstd compressors are supported. (Default Value: (complex option):: diff --git a/doc/source/operating/compression.rst b/doc/source/operating/compression.rst index 01da34b6df..42a057b242 100644 --- a/doc/source/operating/compression.rst +++ b/doc/source/operating/compression.rst @@ -33,11 +33,12 @@ Configuring Compression Compression is configured on a per-table basis as an optional argument to ``CREATE TABLE`` or ``ALTER TABLE``. By default, three options are relevant: -- ``class`` specifies the compression class - Cassandra provides three classes (``LZ4Compressor``, - ``SnappyCompressor``, and ``DeflateCompressor`` ). The default is ``LZ4Compressor``. +- ``class`` specifies the compression class - Cassandra provides four classes (``LZ4Compressor``, + ``SnappyCompressor``, ``DeflateCompressor`` and ``ZstdCompressor``). The default is ``LZ4Compressor``. - ``chunk_length_in_kb`` specifies the number of kilobytes of data per compression chunk. The default is 64KB. - ``crc_check_chance`` determines how likely Cassandra is to verify the checksum on each compression chunk during reads. The default is 1.0. +- ``compression_level`` is only applicable for ``ZstdCompressor`` and accepts values between ``-131072`` and ``2``. Users can set compression using the following syntax: diff --git a/lib/licenses/zstd-jni-1.3.8-3.txt b/lib/licenses/zstd-jni-1.3.8-3.txt new file mode 100644 index 0000000000..66abb8ae78 --- /dev/null +++ b/lib/licenses/zstd-jni-1.3.8-3.txt @@ -0,0 +1,26 @@ +Zstd-jni: JNI bindings to Zstd Library + +Copyright (c) 2015-present, Luben Karavelov/ All rights reserved. + +BSD License + +Redistribution and use in source and binary forms, with or without modification, +are permitted provided that the following conditions are met: + +* Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +* Redistributions in binary form must reproduce the above copyright notice, this + list of conditions and the following disclaimer in the documentation and/or + other materials provided with the distribution. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR +ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES +(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; +LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON +ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/lib/zstd-jni-1.3.8-5.jar b/lib/zstd-jni-1.3.8-5.jar new file mode 100644 index 0000000000..f68ce66684 Binary files /dev/null and b/lib/zstd-jni-1.3.8-5.jar differ diff --git a/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java b/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java new file mode 100644 index 0000000000..1fd741571c --- /dev/null +++ b/src/java/org/apache/cassandra/io/compress/ZstdCompressor.java @@ -0,0 +1,239 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.compress; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +import com.google.common.annotations.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.luben.zstd.Zstd; + +/** + * ZSTD Compressor + */ +public class ZstdCompressor implements ICompressor +{ + private static final Logger logger = LoggerFactory.getLogger(ZstdCompressor.class); + + // These might change with the version of Zstd we're using + public static final int FAST_COMPRESSION_LEVEL = Zstd.minCompressionLevel(); + public static final int BEST_COMPRESSION_LEVEL = Zstd.maxCompressionLevel(); + + // Compressor Defaults + public static final int DEFAULT_COMPRESSION_LEVEL = 3; + private static final boolean ENABLE_CHECKSUM_FLAG = true; + + @VisibleForTesting + public static final String COMPRESSION_LEVEL_OPTION_NAME = "compression_level"; + + private static final ConcurrentHashMap instances = new ConcurrentHashMap<>(); + + private final int compressionLevel; + + /** + * Create a Zstd compressor with the given options + * + * @param options + * @return + */ + public static ZstdCompressor create(Map options) + { + int level = getOrDefaultCompressionLevel(options); + + if (!isValid(level)) + throw new IllegalArgumentException(String.format("%s=%d is invalid", COMPRESSION_LEVEL_OPTION_NAME, level)); + + return getOrCreate(level); + } + + /** + * Private constructor + * + * @param compressionLevel + */ + private ZstdCompressor(int compressionLevel) + { + this.compressionLevel = compressionLevel; + logger.trace("Creating Zstd Compressor with compression level={}", compressionLevel); + } + + /** + * Get a cached instance or return a new one + * + * @param level + * @return + */ + public static ZstdCompressor getOrCreate(int level) + { + return instances.computeIfAbsent(level, l -> new ZstdCompressor(level)); + } + + /** + * Get initial compressed buffer length + * + * @param chunkLength + * @return + */ + @Override + public int initialCompressedBufferLength(int chunkLength) + { + return (int) Zstd.compressBound(chunkLength); + } + + /** + * Decompress data using arrays + * + * @param input + * @param inputOffset + * @param inputLength + * @param output + * @param outputOffset + * @return + * @throws IOException + */ + @Override + public int uncompress(byte[] input, int inputOffset, int inputLength, byte[] output, int outputOffset) + throws IOException + { + long dsz = Zstd.decompressByteArray(output, outputOffset, output.length - outputOffset, + input, inputOffset, inputLength); + + if (Zstd.isError(dsz)) + throw new IOException(String.format("Decompression failed due to %s", Zstd.getErrorName(dsz))); + + return (int) dsz; + } + + /** + * Decompress data via ByteBuffers + * + * @param input + * @param output + * @throws IOException + */ + @Override + public void uncompress(ByteBuffer input, ByteBuffer output) throws IOException + { + try + { + Zstd.decompress(output, input); + } catch (Exception e) + { + throw new IOException("Decompression failed", e); + } + } + + /** + * Compress using ByteBuffers + * + * @param input + * @param output + * @throws IOException + */ + @Override + public void compress(ByteBuffer input, ByteBuffer output) throws IOException + { + try + { + Zstd.compress(output, input, compressionLevel, ENABLE_CHECKSUM_FLAG); + } catch (Exception e) + { + throw new IOException("Compression failed", e); + } + } + + /** + * Check if the given compression level is valid. This can be a negative value as well. + * + * @param level + * @return + */ + private static boolean isValid(int level) + { + return (level >= FAST_COMPRESSION_LEVEL && level <= BEST_COMPRESSION_LEVEL); + } + + /** + * Parse the compression options + * + * @param options + * @return + */ + private static int getOrDefaultCompressionLevel(Map options) + { + if (options == null) + return DEFAULT_COMPRESSION_LEVEL; + + String val = options.get(COMPRESSION_LEVEL_OPTION_NAME); + + if (val == null) + return DEFAULT_COMPRESSION_LEVEL; + + return Integer.valueOf(val); + } + + /** + * Return the preferred BufferType + * + * @return + */ + @Override + public BufferType preferredBufferType() + { + return BufferType.OFF_HEAP; + } + + /** + * Check whether the given BufferType is supported + * + * @param bufferType + * @return + */ + @Override + public boolean supports(BufferType bufferType) + { + return bufferType == BufferType.OFF_HEAP; + } + + /** + * Lists the supported options by this compressor + * + * @return + */ + @Override + public Set supportedOptions() + { + return new HashSet<>(Collections.singletonList(COMPRESSION_LEVEL_OPTION_NAME)); + } + + + @VisibleForTesting + public int getCompressionLevel() + { + return compressionLevel; + } +} diff --git a/src/java/org/apache/cassandra/schema/CompressionParams.java b/src/java/org/apache/cassandra/schema/CompressionParams.java index 25631116de..40a4be351b 100644 --- a/src/java/org/apache/cassandra/schema/CompressionParams.java +++ b/src/java/org/apache/cassandra/schema/CompressionParams.java @@ -29,7 +29,6 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Objects; import com.google.common.collect.ImmutableMap; -import org.apache.commons.lang3.builder.EqualsBuilder; import org.apache.commons.lang3.builder.HashCodeBuilder; import org.slf4j.Logger; @@ -179,6 +178,17 @@ public final class CompressionParams return new CompressionParams(LZ4Compressor.create(Collections.emptyMap()), chunkLength, maxCompressedLength, calcMinCompressRatio(chunkLength, maxCompressedLength), Collections.emptyMap()); } + public static CompressionParams zstd() + { + return zstd(DEFAULT_CHUNK_LENGTH); + } + + public static CompressionParams zstd(Integer chunkLength) + { + ZstdCompressor compressor = ZstdCompressor.create(Collections.emptyMap()); + return new CompressionParams(compressor, chunkLength, Integer.MAX_VALUE, DEFAULT_MIN_COMPRESS_RATIO, Collections.emptyMap()); + } + public CompressionParams(String sstableCompressorClass, Map otherOptions, int chunkLength, double minCompressRatio) throws ConfigurationException { this(createCompressor(parseCompressorClass(sstableCompressorClass), otherOptions), chunkLength, calcMaxCompressedLength(chunkLength, minCompressRatio), minCompressRatio, otherOptions); diff --git a/test/conf/commitlog_compression.yaml b/test/conf/commitlog_compression_LZ4.yaml similarity index 100% rename from test/conf/commitlog_compression.yaml rename to test/conf/commitlog_compression_LZ4.yaml diff --git a/test/conf/commitlog_compression_Zstd.yaml b/test/conf/commitlog_compression_Zstd.yaml new file mode 100644 index 0000000000..0c440aee38 --- /dev/null +++ b/test/conf/commitlog_compression_Zstd.yaml @@ -0,0 +1,2 @@ +commitlog_compression: + - class_name: ZstdCompressor diff --git a/test/long/org/apache/cassandra/io/compress/CompressorPerformance.java b/test/long/org/apache/cassandra/io/compress/CompressorPerformance.java index e703839c51..b3cdaa1fc2 100644 --- a/test/long/org/apache/cassandra/io/compress/CompressorPerformance.java +++ b/test/long/org/apache/cassandra/io/compress/CompressorPerformance.java @@ -35,7 +35,9 @@ public class CompressorPerformance SnappyCompressor.instance, // warm up DeflateCompressor.instance, LZ4Compressor.create(Collections.emptyMap()), - SnappyCompressor.instance + SnappyCompressor.instance, + ZstdCompressor.getOrCreate(ZstdCompressor.FAST_COMPRESSION_LEVEL), + ZstdCompressor.getOrCreate(ZstdCompressor.DEFAULT_COMPRESSION_LEVEL) }) { for (BufferType in: BufferType.values()) @@ -70,10 +72,15 @@ public class CompressorPerformance int count = 100; long time = System.nanoTime(); + long uncompressedBytes = 0; + long compressedBytes = 0; for (int i=0; i%s compress %.3f ns/b %.3f mb/s uncompress %.3f ns/b %.3f mb/s.%s\n", + System.out.format("Compressor %s %s->%s compress %.3f ns/b %.3f mb/s uncompress %.3f ns/b %.3f mb/s ratio %.2f:1.%s\n", compressor.getClass().getSimpleName(), in, out, @@ -101,6 +108,7 @@ public class CompressorPerformance Math.scalb(1.0e9, -20) * count * len / timec, 1.0 * timed / (count * len), Math.scalb(1.0e9, -20) * count * len / timed, + ((double) uncompressedBytes) / ((double) compressedBytes), checksum == 0 ? " " : ""); } diff --git a/test/unit/org/apache/cassandra/db/RecoveryManagerFlushedTest.java b/test/unit/org/apache/cassandra/db/RecoveryManagerFlushedTest.java index 72255609fd..fc34942340 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManagerFlushedTest.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManagerFlushedTest.java @@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.ParameterizedClass; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.commitlog.CommitLog; @@ -71,7 +72,8 @@ public class RecoveryManagerFlushedTest {null, EncryptionContextGenerator.createContext(true)}, // Encryption {new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, {new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, - {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); + {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, + {new ParameterizedClass(ZstdCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); } @Before diff --git a/test/unit/org/apache/cassandra/db/RecoveryManagerMissingHeaderTest.java b/test/unit/org/apache/cassandra/db/RecoveryManagerMissingHeaderTest.java index d60574faa2..4044fff612 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManagerMissingHeaderTest.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManagerMissingHeaderTest.java @@ -42,6 +42,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.compress.DeflateCompressor; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.compress.SnappyCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.security.EncryptionContext; @@ -70,7 +71,8 @@ public class RecoveryManagerMissingHeaderTest {null, EncryptionContextGenerator.createContext(true)}, // Encryption {new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, {new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, - {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); + {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, + {new ParameterizedClass(ZstdCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); } @Before diff --git a/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java b/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java index 164253d1e1..7e397c1c27 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java @@ -55,6 +55,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.compress.DeflateCompressor; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.compress.SnappyCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.security.EncryptionContext; @@ -89,7 +90,8 @@ public class RecoveryManagerTest {null, EncryptionContextGenerator.createContext(true)}, // Encryption {new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, {new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, - {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); + {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, + {new ParameterizedClass(ZstdCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); } @Before diff --git a/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java b/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java index ef00b22d1c..a51cd21510 100644 --- a/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java +++ b/test/unit/org/apache/cassandra/db/RecoveryManagerTruncateTest.java @@ -32,6 +32,7 @@ import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.io.compress.DeflateCompressor; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.compress.SnappyCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.security.EncryptionContext; import org.apache.cassandra.security.EncryptionContextGenerator; @@ -68,7 +69,8 @@ public class RecoveryManagerTruncateTest {null, EncryptionContextGenerator.createContext(true)}, // Encryption {new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, {new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, - {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); + {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, + {new ParameterizedClass(ZstdCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); } @Before diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java index e1bd96c1a3..25e2f30625 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java @@ -38,6 +38,7 @@ import org.junit.runners.Parameterized.Parameters; import org.apache.cassandra.SchemaLoader; import org.apache.cassandra.Util; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; @@ -98,7 +99,8 @@ public abstract class CommitLogTest {null, EncryptionContextGenerator.createContext(true)}, // Encryption {new ParameterizedClass(LZ4Compressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, {new ParameterizedClass(SnappyCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, - {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); + {new ParameterizedClass(DeflateCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}, + {new ParameterizedClass(ZstdCompressor.class.getName(), Collections.emptyMap()), EncryptionContextGenerator.createDisabledContext()}}); } public static void beforeClass() throws ConfigurationException diff --git a/test/unit/org/apache/cassandra/db/commitlog/SegmentReaderTest.java b/test/unit/org/apache/cassandra/db/commitlog/SegmentReaderTest.java index 47e1afa898..ce209351fa 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/SegmentReaderTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/SegmentReaderTest.java @@ -41,6 +41,7 @@ import org.apache.cassandra.io.compress.DeflateCompressor; import org.apache.cassandra.io.compress.ICompressor; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.compress.SnappyCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; import org.apache.cassandra.io.util.FileDataInput; import org.apache.cassandra.io.util.FileUtils; import org.apache.cassandra.io.util.RandomAccessReader; @@ -78,6 +79,12 @@ public class SegmentReaderTest compressedSegmenter(DeflateCompressor.create(null)); } + @Test + public void compressedSegmenter_Zstd() throws IOException + { + compressedSegmenter(ZstdCompressor.create(Collections.emptyMap())); + } + private void compressedSegmenter(ICompressor compressor) throws IOException { int rawSize = (1 << 15) - 137; diff --git a/test/unit/org/apache/cassandra/hints/HintsCompressionTest.java b/test/unit/org/apache/cassandra/hints/HintsCompressionTest.java index f82db49a82..faa8d2791d 100644 --- a/test/unit/org/apache/cassandra/hints/HintsCompressionTest.java +++ b/test/unit/org/apache/cassandra/hints/HintsCompressionTest.java @@ -26,6 +26,7 @@ import org.apache.cassandra.io.compress.DeflateCompressor; import org.apache.cassandra.io.compress.ICompressor; import org.apache.cassandra.io.compress.LZ4Compressor; import org.apache.cassandra.io.compress.SnappyCompressor; +import org.apache.cassandra.io.compress.ZstdCompressor; public class HintsCompressionTest extends AlteredHints { @@ -77,4 +78,12 @@ public class HintsCompressionTest extends AlteredHints compressorClass = DeflateCompressor.class; multiFlushAndDeserializeTest(); } + + @Test + public void zstdCompressor() throws Exception + { + compressorClass = ZstdCompressor.class; + multiFlushAndDeserializeTest(); + } + } diff --git a/test/unit/org/apache/cassandra/io/compress/CQLCompressionTest.java b/test/unit/org/apache/cassandra/io/compress/CQLCompressionTest.java index 1efccd34e6..d4b0b7bebf 100644 --- a/test/unit/org/apache/cassandra/io/compress/CQLCompressionTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CQLCompressionTest.java @@ -53,4 +53,29 @@ public class CQLCompressionTest extends CQLTester throw e.getCause(); } } + + @Test + public void zstdParamsTest() + { + createTable("create table %s (id int primary key, uh text) with compression = {'class':'ZstdCompressor', 'compression_level':-22}"); + assertTrue(((ZstdCompressor)getCurrentColumnFamilyStore().metadata().params.compression.getSstableCompressor()).getClass().equals(ZstdCompressor.class)); + assertEquals(((ZstdCompressor)getCurrentColumnFamilyStore().metadata().params.compression.getSstableCompressor()).getCompressionLevel(), -22); + + createTable("create table %s (id int primary key, uh text) with compression = {'class':'ZstdCompressor'}"); + assertTrue(((ZstdCompressor)getCurrentColumnFamilyStore().metadata().params.compression.getSstableCompressor()).getClass().equals(ZstdCompressor.class)); + assertEquals(((ZstdCompressor)getCurrentColumnFamilyStore().metadata().params.compression.getSstableCompressor()).getCompressionLevel(), ZstdCompressor.DEFAULT_COMPRESSION_LEVEL); + } + + @Test(expected = ConfigurationException.class) + public void zstdBadParamsTest() throws Throwable + { + try + { + createTable("create table %s (id int primary key, uh text) with compression = {'class':'ZstdCompressor', 'compression_level':'100'}"); + } + catch (RuntimeException e) + { + throw e.getCause(); + } + } } diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java index 557bc32109..a965410813 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java @@ -97,6 +97,13 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest runTests("Snappy"); } + @Test + public void testZSTDWriter() throws IOException + { + compressionParameters = CompressionParams.zstd(); + runTests("ZSTD"); + } + private void testWrite(File f, int bytesToTest, boolean useMemmap) throws IOException { final String filename = f.getAbsolutePath(); diff --git a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java index 1d37ad4eab..b649f526d9 100644 --- a/test/unit/org/apache/cassandra/io/compress/CompressorTest.java +++ b/test/unit/org/apache/cassandra/io/compress/CompressorTest.java @@ -42,7 +42,8 @@ public class CompressorTest ICompressor[] compressors = new ICompressor[] { LZ4Compressor.create(Collections.emptyMap()), DeflateCompressor.create(Collections.emptyMap()), - SnappyCompressor.create(Collections.emptyMap()) + SnappyCompressor.create(Collections.emptyMap()), + ZstdCompressor.create(Collections.emptyMap()) }; @Test @@ -177,6 +178,13 @@ public class CompressorTest testByteBuffers(); } + @Test + public void testZstdByteBuffers() throws IOException + { + compressor = ZstdCompressor.create(Collections.emptyMap()); + testByteBuffers(); + } + private void testByteBuffers() throws IOException { assert compressor.supports(BufferType.OFF_HEAP); diff --git a/test/unit/org/apache/cassandra/io/compress/ZstdCompressorTest.java b/test/unit/org/apache/cassandra/io/compress/ZstdCompressorTest.java new file mode 100644 index 0000000000..70e32ad4d2 --- /dev/null +++ b/test/unit/org/apache/cassandra/io/compress/ZstdCompressorTest.java @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.io.compress; + +import java.util.Collections; + +import com.google.common.collect.ImmutableMap; +import org.junit.Test; + +import com.github.luben.zstd.Zstd; + +import static org.junit.Assert.assertEquals; + +/** + * Zstd Compressor specific tests. General compressor tests are in {@link CompressorTest} + */ +public class ZstdCompressorTest +{ + @Test + public void emptyConfigurationUsesDefaultCompressionLevel() + { + ZstdCompressor compressor = ZstdCompressor.create(Collections.emptyMap()); + assertEquals(ZstdCompressor.DEFAULT_COMPRESSION_LEVEL, compressor.getCompressionLevel()); + } + + @Test(expected = IllegalArgumentException.class) + public void badCompressionLevelParamThrowsExceptionMin() + { + ZstdCompressor.create(ImmutableMap.of(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME, Integer.toString(Zstd.minCompressionLevel() - 1))); + } + + @Test(expected = IllegalArgumentException.class) + public void badCompressionLevelParamThrowsExceptionMax() + { + ZstdCompressor.create(ImmutableMap.of(ZstdCompressor.COMPRESSION_LEVEL_OPTION_NAME, Integer.toString(Zstd.maxCompressionLevel() + 1))); + } +}