diff --git a/build.xml b/build.xml index addc48b69fd0..061854d521b0 100644 --- a/build.xml +++ b/build.xml @@ -752,6 +752,7 @@ + @@ -946,6 +947,7 @@ + diff --git a/src/java/org/apache/cassandra/cache/AutoSavingCache.java b/src/java/org/apache/cassandra/cache/AutoSavingCache.java index 8037bd90b880..efe9b4944277 100644 --- a/src/java/org/apache/cassandra/cache/AutoSavingCache.java +++ b/src/java/org/apache/cassandra/cache/AutoSavingCache.java @@ -118,7 +118,7 @@ public InputStream getInputStream(File dataPath, File crcPath) throws IOExceptio public OutputStream getOutputStream(File dataPath, File crcPath) { - return new ChecksummedSequentialWriter(dataPath, crcPath, null, writerOption); + return new ChecksummedSequentialWriter(dataPath, crcPath, writerOption); } }; diff --git a/src/java/org/apache/cassandra/db/compaction/Verifier.java b/src/java/org/apache/cassandra/db/compaction/Verifier.java index fd6cba56d3c1..011f9010a8fa 100644 --- a/src/java/org/apache/cassandra/db/compaction/Verifier.java +++ b/src/java/org/apache/cassandra/db/compaction/Verifier.java @@ -80,6 +80,8 @@ public class Verifier implements Closeable { + private static final List DIGEST_COMPONENTS = ImmutableList.of(Component.DIGEST, Component.DIGEST_CRC32C, Component.DIGEST_CRC64NVME); + private final @Nullable CompactionRealm realm; private final SSTableReader sstable; @@ -250,12 +252,17 @@ public void verify() { validator = null; - if (sstable.descriptor.fileFor(Component.DIGEST).exists()) + for (Component component : DIGEST_COMPONENTS) { - validator = DataIntegrityMetadata.fileDigestValidator(sstable.descriptor); - validator.validate(); + if (sstable.descriptor.fileFor(component).exists()) + { + validator = DataIntegrityMetadata.fileDigestValidator(sstable.descriptor, component); + validator.validate(); + break; + } } - else + + if (validator == null) { outputHandler.output("Data digest missing, assuming extended verification of disk values"); extended = true; diff --git a/src/java/org/apache/cassandra/db/streaming/ComponentManifest.java b/src/java/org/apache/cassandra/db/streaming/ComponentManifest.java index 06cdf357613c..2aa157c1f905 100644 --- a/src/java/org/apache/cassandra/db/streaming/ComponentManifest.java +++ b/src/java/org/apache/cassandra/db/streaming/ComponentManifest.java @@ -39,7 +39,7 @@ public final class ComponentManifest implements Iterable { private static final List STREAM_COMPONENTS = ImmutableList.of(Component.DATA, Component.PRIMARY_INDEX, Component.PARTITION_INDEX, Component.ROW_INDEX, Component.STATS, Component.COMPRESSION_INFO, Component.FILTER, Component.SUMMARY, - Component.DIGEST, Component.CRC); + Component.DIGEST, Component.DIGEST_CRC32C, Component.DIGEST_CRC64NVME, Component.CRC); private final LinkedHashMap components; diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java index e9f2e91dd280..3a58e2ef0dd9 100644 --- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java @@ -27,7 +27,9 @@ import java.util.Optional; import java.util.zip.CRC32; import java.util.zip.CheckedInputStream; +import java.util.zip.Checksum; +import com.google.common.annotations.VisibleForTesting; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -40,6 +42,7 @@ import org.apache.cassandra.io.util.File; import org.apache.cassandra.schema.CompressionParams; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.ChecksumType; import static org.apache.cassandra.utils.Throwables.merge; @@ -70,6 +73,8 @@ public class CompressedSequentialWriter extends SequentialWriter private final int maxCompressedLength; + private final ChecksumType checksumType; + /** * When corruption is found, the file writer is reset to previous data point but we can't reset the CRC checksum. * So we have to recompute digest value. @@ -82,6 +87,7 @@ public class CompressedSequentialWriter extends SequentialWriter * @param file File to write * @param offsetsPath File name to write compression metadata * @param digestFile File to write digest + * @param checksumType The type of checksum in the digest file * @param option Write option (buffer size and type will be set the same as compression params) * @param parameters Compression mparameters * @param sstableMetadataCollector Metadata collector @@ -89,6 +95,7 @@ public class CompressedSequentialWriter extends SequentialWriter public CompressedSequentialWriter(File file, File offsetsPath, File digestFile, + ChecksumType checksumType, SequentialWriterOption option, CompressionParams parameters, MetadataCollector sstableMetadataCollector) @@ -112,7 +119,18 @@ public CompressedSequentialWriter(File file, metadataWriter = CompressionMetadata.Writer.open(parameters, offsetsPath); this.sstableMetadataCollector = sstableMetadataCollector; - crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel))); + this.checksumType = checksumType; + crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel)), checksumType); + } + + public CompressedSequentialWriter(File file, + File offsetsPath, + File digestFile, + SequentialWriterOption option, + CompressionParams parameters, + MetadataCollector sstableMetadataCollector) + { + this(file, offsetsPath, digestFile, ChecksumType.CRC32, option, parameters, sstableMetadataCollector); } @Override @@ -460,7 +478,7 @@ private void maybeWriteChecksum() try (FileChannel fileChannel = StorageProvider.instance.writeTimeReadFileChannelFor(file); InputStream stream = Channels.newInputStream(fileChannel)) { - CRC32 checksum = new CRC32(); + Checksum checksum = checksumType.newInstance(); try (CheckedInputStream checkedInputStream = new CheckedInputStream(stream, checksum)) { byte[] chunk = new byte[64 * 1024]; diff --git a/src/java/org/apache/cassandra/io/compress/EncryptedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/EncryptedSequentialWriter.java index 68f49b7b25cb..cd62c3cfba50 100644 --- a/src/java/org/apache/cassandra/io/compress/EncryptedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/compress/EncryptedSequentialWriter.java @@ -35,6 +35,7 @@ import org.apache.cassandra.io.util.SequentialWriter; import org.apache.cassandra.io.util.SequentialWriterOption; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.PageAware; /** @@ -103,7 +104,7 @@ public EncryptedSequentialWriter(File file, this.encrypted = BufferType.preferredForCompression().allocate(CHUNK_SIZE); maxBytesInChunk = buffer.capacity(); - crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel))); + crcMetadata = new ChecksumWriter(new DataOutputStream(Channels.newOutputStream(channel)), ChecksumType.CRC32); } public static int maxBytesInPage(ICompressor encryptor) diff --git a/src/java/org/apache/cassandra/io/sstable/Component.java b/src/java/org/apache/cassandra/io/sstable/Component.java index ee9f34b221b4..8c7a201b3d83 100644 --- a/src/java/org/apache/cassandra/io/sstable/Component.java +++ b/src/java/org/apache/cassandra/io/sstable/Component.java @@ -61,6 +61,10 @@ public enum Type STATS("Statistics.db"), // holds CRC32 checksum of the data file DIGEST("Digest.crc32"), + // holds CRC32C checksum of the data file + DIGEST_CRC32C("Digest.crc32c"), + // holds CRC64-NVMe checksum of the data file + DIGEST_CRC64NMVE("Digest.crc64nvme"), // holds the CRC32 for chunks in an a uncompressed file. CRC("CRC.db"), // holds SSTable Index Summary (sampling of Index component) @@ -104,6 +108,8 @@ public static Type fromRepresentation(String repr) public final static Component COMPRESSION_INFO = new Component(Type.COMPRESSION_INFO); public final static Component STATS = new Component(Type.STATS); public final static Component DIGEST = new Component(Type.DIGEST); + public final static Component DIGEST_CRC32C = new Component(Type.DIGEST_CRC32C); + public final static Component DIGEST_CRC64NVME = new Component(Type.DIGEST_CRC64NMVE); public final static Component CRC = new Component(Type.CRC); public final static Component SUMMARY = new Component(Type.SUMMARY); public final static Component TOC = new Component(Type.TOC); @@ -156,6 +162,8 @@ public static Component parse(String name) case COMPRESSION_INFO: return Component.COMPRESSION_INFO; case STATS: return Component.STATS; case DIGEST: return Component.DIGEST; + case DIGEST_CRC32C: return Component.DIGEST_CRC32C; + case DIGEST_CRC64NMVE: return Component.DIGEST_CRC64NVME; case CRC: return Component.CRC; case SUMMARY: return Component.SUMMARY; case TOC: return Component.TOC; diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java index 3ba16108fd41..906794f60b34 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java @@ -42,6 +42,7 @@ import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.db.rows.Rows; import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.guardrails.Guardrails; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.compress.CompressedSequentialWriter; @@ -66,13 +67,18 @@ import org.apache.cassandra.schema.CompressionParams; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableMetadataRef; +import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.FBUtilities; -import org.apache.cassandra.utils.IFilter; import org.apache.cassandra.utils.Throwables; public abstract class SortedTableWriter extends SSTableWriter { protected static final Logger logger = LoggerFactory.getLogger(SortedTableWriter.class); + + private static final ChecksumType checksumType = getChecksumType(); + private static final Component digestComponent = getDigestComponent(); + public static final String SSTABLE_CHECKSUM_TYPE_PROPERTY = "cassandra.sstable_checksum_type"; + protected final FileHandle.Builder dbuilder; protected final SequentialWriter dataFile; protected DataPosition dataMark; @@ -104,6 +110,34 @@ protected SortedTableWriter(Descriptor descriptor, isInternalKeyspace = SchemaConstants.isInternalKeyspace(metadata.keyspace); } + private static ChecksumType getChecksumType() + { + String checksumTypeProp = System.getProperty(SSTABLE_CHECKSUM_TYPE_PROPERTY, ChecksumType.CRC32.name()); + try + { + return ChecksumType.valueOf(checksumTypeProp.toUpperCase()); + } + catch (IllegalArgumentException e) + { + throw new ConfigurationException(String.format("Invalid value for system property 'cassandra.sstable_checksum_type': %s", checksumTypeProp), e); + } + } + + private static Component getDigestComponent() + { + switch (checksumType) + { + case CRC32: + return Component.DIGEST; + case CRC32C: + return Component.DIGEST_CRC32C; + case CRC64NVME: + return Component.DIGEST_CRC64NVME; + default: + throw new IllegalStateException("Unexpected checksumType for digest file: " + checksumType); + } + } + protected static SequentialWriter constructDataFileWriter(Descriptor descriptor, TableMetadataRef metadata, MetadataCollector metadataCollector, @@ -116,7 +150,8 @@ protected static SequentialWriter constructDataFileWriter(Descriptor descriptor, return new CompressedSequentialWriter(descriptor.fileFor(Component.DATA), descriptor.fileFor(Component.COMPRESSION_INFO), - descriptor.fileFor(Component.DIGEST), + descriptor.fileFor(digestComponent), + checksumType, writerOption, compressionParams, metadataCollector); @@ -125,7 +160,8 @@ protected static SequentialWriter constructDataFileWriter(Descriptor descriptor, { return new ChecksummedSequentialWriter(descriptor.fileFor(Component.DATA), descriptor.fileFor(Component.CRC), - descriptor.fileFor(Component.DIGEST), + descriptor.fileFor(digestComponent), + checksumType, writerOption); } } diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java index c2d18cb3cdc9..e3bf6609549e 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java @@ -65,6 +65,8 @@ public class BigFormat implements SSTableFormat Component.COMPRESSION_INFO, Component.STATS, Component.DIGEST, + Component.DIGEST_CRC32C, + Component.DIGEST_CRC64NVME, Component.CRC, Component.SUMMARY, Component.TOC); @@ -76,6 +78,8 @@ public class BigFormat implements SSTableFormat Component.COMPRESSION_INFO, Component.FILTER, Component.DIGEST, + Component.DIGEST_CRC32C, + Component.DIGEST_CRC64NVME, Component.CRC); private final static Set PRIMARY_INDEX_COMPONENTS = ImmutableSet.of(Component.PRIMARY_INDEX); diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java index 120cb4663201..05c2c6746fa3 100644 --- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java +++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableWriter.java @@ -110,7 +110,9 @@ private static Set components(TableMetadata metadata, Collection PRIMARY_INDEX_COMPONENTS = ImmutableSet.of(Component.PARTITION_INDEX, diff --git a/src/java/org/apache/cassandra/io/util/ChecksumWriter.java b/src/java/org/apache/cassandra/io/util/ChecksumWriter.java index 76cd00ecbbdc..4f9ffab039c7 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksumWriter.java +++ b/src/java/org/apache/cassandra/io/util/ChecksumWriter.java @@ -25,21 +25,23 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; -import java.util.zip.CRC32; +import java.util.zip.Checksum; import javax.annotation.Nonnull; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.compress.CompressedSequentialWriter; +import org.apache.cassandra.utils.ChecksumType; public class ChecksumWriter { - private final CRC32 incrementalChecksum = new CRC32(); + private final Checksum incrementalChecksum = ChecksumType.CRC32.newInstance(); private final DataOutput incrementalOut; - private final CRC32 fullChecksum = new CRC32(); + private final Checksum fullChecksum; - public ChecksumWriter(DataOutput incrementalOut) + public ChecksumWriter(DataOutput incrementalOut, ChecksumType fullChecksumType) { this.incrementalOut = incrementalOut; + this.fullChecksum = fullChecksumType.newInstance(); } public void writeChunkSize(int length) diff --git a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java index fa8fad7d5771..0a9614c5b6c2 100644 --- a/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java +++ b/src/java/org/apache/cassandra/io/util/ChecksummedSequentialWriter.java @@ -20,6 +20,8 @@ import java.nio.ByteBuffer; import java.util.Optional; +import org.apache.cassandra.utils.ChecksumType; + public class ChecksummedSequentialWriter extends SequentialWriter { private static final SequentialWriterOption CRC_WRITER_OPTION = SequentialWriterOption.newBuilder() @@ -30,15 +32,20 @@ public class ChecksummedSequentialWriter extends SequentialWriter private final ChecksumWriter crcMetadata; private final Optional digestFile; - public ChecksummedSequentialWriter(File file, File crcPath, File digestFile, SequentialWriterOption option) + public ChecksummedSequentialWriter(File file, File crcPath, File digestFile, ChecksumType checksumType, SequentialWriterOption option) { super(file, option); crcWriter = new SequentialWriter(crcPath, CRC_WRITER_OPTION); - crcMetadata = new ChecksumWriter(crcWriter); + crcMetadata = new ChecksumWriter(crcWriter, checksumType); crcMetadata.writeChunkSize(buffer.capacity()); this.digestFile = Optional.ofNullable(digestFile); } + public ChecksummedSequentialWriter(File file, File crcPath, SequentialWriterOption option) + { + this(file, crcPath, null, ChecksumType.CRC32, option); + } + @Override protected void flushData() { diff --git a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java index 4efc2cec8736..65c219c9f532 100644 --- a/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java +++ b/src/java/org/apache/cassandra/io/util/DataIntegrityMetadata.java @@ -108,9 +108,9 @@ public void close() } } - public static FileDigestValidator fileDigestValidator(Descriptor desc) throws IOException + public static FileDigestValidator fileDigestValidator(Descriptor desc, Component digestComponent) throws IOException { - return new FileDigestValidator(desc); + return new FileDigestValidator(desc, digestComponent); } public static class FileDigestValidator implements Closeable @@ -121,11 +121,11 @@ public static class FileDigestValidator implements Closeable private final Descriptor descriptor; private long storedDigestValue; - public FileDigestValidator(Descriptor descriptor) throws IOException + public FileDigestValidator(Descriptor descriptor, Component digestComponent) throws IOException { this.descriptor = descriptor; - checksum = ChecksumType.CRC32.newInstance(); - digestReader = RandomAccessReader.open(descriptor.fileFor(Component.DIGEST)); + checksum = checksumFor(digestComponent).newInstance(); + digestReader = RandomAccessReader.open(descriptor.fileFor(digestComponent)); dataReader = RandomAccessReader.open(descriptor.fileFor(Component.DATA)); try { @@ -139,6 +139,18 @@ public FileDigestValidator(Descriptor descriptor) throws IOException } } + private static ChecksumType checksumFor(Component digestComponent) + { + if (digestComponent == Component.DIGEST) + return ChecksumType.CRC32; + else if (digestComponent == Component.DIGEST_CRC32C) + return ChecksumType.CRC32C; + else if (digestComponent == Component.DIGEST_CRC64NVME) + return ChecksumType.CRC64NVME; + else + throw new IllegalArgumentException("Unsupported digest component " + digestComponent); + } + // Validate the entire file public void validate() throws IOException { diff --git a/src/java/org/apache/cassandra/utils/ChecksumType.java b/src/java/org/apache/cassandra/utils/ChecksumType.java index fa920aaca8d2..217a78193bd3 100644 --- a/src/java/org/apache/cassandra/utils/ChecksumType.java +++ b/src/java/org/apache/cassandra/utils/ChecksumType.java @@ -18,11 +18,16 @@ package org.apache.cassandra.utils; import java.nio.ByteBuffer; +import java.util.zip.CRC32C; import java.util.zip.Checksum; import java.util.zip.CRC32; import java.util.zip.Adler32; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import io.netty.util.concurrent.FastThreadLocal; +import software.amazon.awssdk.crt.checksums.CRC64NVME; public enum ChecksumType { @@ -38,7 +43,7 @@ public Checksum newInstance() @Override public void update(Checksum checksum, ByteBuffer buf) { - ((Adler32)checksum).update(buf); + checksum.update(buf); } }, @@ -54,11 +59,60 @@ public Checksum newInstance() @Override public void update(Checksum checksum, ByteBuffer buf) { - ((CRC32)checksum).update(buf); + checksum.update(buf); + } + + }, + CRC32C + { + + @Override + public Checksum newInstance() + { + return new CRC32C(); + } + + @Override + public void update(Checksum checksum, ByteBuffer buf) + { + checksum.update(buf); + } + + }, + CRC64NVME + { + + @Override + public Checksum newInstance() + { + if (HAS_AWS_CRT_CRC64NVME) + return new CRC64NVME(); + return new PureJavaCRC64NVME(); + } + + @Override + public void update(Checksum checksum, ByteBuffer buf) + { + checksum.update(buf); } }; + private static final Logger logger = LoggerFactory.getLogger(ChecksumType.class); + private static final boolean HAS_AWS_CRT_CRC64NVME; + + static { + boolean available = false; + try { + Class.forName("software.amazon.awssdk.crt.checksums.CRC64NVME"); + available = true; + } catch (ClassNotFoundException e) { + logger.debug("software.amazon.awssdk.crt.checksums.CRC64NVME not found, " + + "falling back to PureJavaCRC64NVME for CRC64NVME checksum"); + } + HAS_AWS_CRT_CRC64NVME = available; + } + public abstract Checksum newInstance(); public abstract void update(Checksum checksum, ByteBuffer buf); diff --git a/src/java/org/apache/cassandra/utils/PureJavaCRC64NVME.java b/src/java/org/apache/cassandra/utils/PureJavaCRC64NVME.java new file mode 100644 index 000000000000..c748740665ec --- /dev/null +++ b/src/java/org/apache/cassandra/utils/PureJavaCRC64NVME.java @@ -0,0 +1,112 @@ +/* + * 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.utils; + +import java.nio.ByteOrder; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; +import java.util.zip.Checksum; + +public final class PureJavaCRC64NVME implements Checksum +{ + private static final long POLY = 0x9A6C9329AC4BC9B5L; + + private static final long[][] TABLE = new long[8][256]; + + private static final VarHandle LONG_LE = MethodHandles.byteArrayViewVarHandle(long[].class, ByteOrder.LITTLE_ENDIAN); + + static + { + for (int i = 0; i < 256; i++) + { + long crc = i; + + for (int j = 0; j < 8; j++) + { + crc = ((crc & 1) != 0) + ? (crc >>> 1) ^ POLY + : (crc >>> 1); + } + + TABLE[0][i] = crc; + } + + for (int t = 1; t < 8; t++) + { + for (int i = 0; i < 256; i++) + { + long crc = TABLE[t - 1][i]; + TABLE[t][i] = TABLE[0][(int) (crc & 0xFF)] ^ (crc >>> 8); + } + } + } + + private long crc; + + @Override + public void update(int b) + { + long c = crc ^ ~0L; + c = (c >>> 8) ^ TABLE[0][(int) ((c ^ b) & 0xFF)]; + crc = c ^ ~0L; + } + + @Override + public void update(byte[] b, int off, int len) + { + long c = crc ^ ~0L; + + int end = off + len; + + while (off + 8 <= end) + { + long x = (long) LONG_LE.get(b, off) ^ c; + + c = TABLE[7][(int) (x & 0xFF)] ^ + TABLE[6][(int) ((x >>> 8) & 0xFF)] ^ + TABLE[5][(int) ((x >>> 16) & 0xFF)] ^ + TABLE[4][(int) ((x >>> 24) & 0xFF)] ^ + TABLE[3][(int) ((x >>> 32) & 0xFF)] ^ + TABLE[2][(int) ((x >>> 40) & 0xFF)] ^ + TABLE[1][(int) ((x >>> 48) & 0xFF)] ^ + TABLE[0][(int) ((x >>> 56) & 0xFF)]; + + off += 8; + } + + while (off < end) + { + c = (c >>> 8) ^ TABLE[0][(int) ((c ^ b[off++]) & 0xFF)]; + } + + crc = c ^ ~0L; + } + + @Override + public long getValue() + { + return crc; + } + + @Override + public void reset() + { + crc = 0; + } +} \ No newline at end of file diff --git a/test/distributed/org/apache/cassandra/distributed/test/SSTableDigestTest.java b/test/distributed/org/apache/cassandra/distributed/test/SSTableDigestTest.java new file mode 100644 index 000000000000..8ebb8cb9bd0e --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/SSTableDigestTest.java @@ -0,0 +1,117 @@ +/* + * 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.distributed.test; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; + +import org.junit.Test; + +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.shared.WithProperties; +import org.apache.cassandra.io.sstable.Component; +import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.utils.ChecksumType; + +import static java.lang.String.format; +import static org.apache.cassandra.io.sstable.format.SortedTableWriter.SSTABLE_CHECKSUM_TYPE_PROPERTY; +import static org.assertj.core.api.Assertions.assertThat; + +public class SSTableDigestTest extends TestBaseImpl +{ + public void testDigestFile(ChecksumType checksumType, Component digestComponent) throws IOException + { + try (WithProperties properties = new WithProperties()) + { + if (checksumType != null) + properties.setProperty(SSTABLE_CHECKSUM_TYPE_PROPERTY, checksumType.toString()); + + try (Cluster cluster = init(Cluster.build(1) + .withDataDirCount(1) + .start())) + { + cluster.disableAutoCompaction(KEYSPACE); + cluster.schemaChange(createTableStmt(KEYSPACE)); + cluster.get(1).executeInternal(format("INSERT INTO %s.%s (pk, ck, v) VALUES (?, ?, ?)", KEYSPACE, "tbl"), 1, 1, 1); + cluster.get(1).flush(KEYSPACE); + + String digestFileName = digestComponent.type.repr; + + cluster.get(1).runOnInstance(() -> { + try + { + SSTableReader ssTable = new ArrayList<>(Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").getLiveSSTables()).get(0); + + Path digestPath = ssTable.descriptor.pathFor(Component.parse(digestFileName)); + + assertThat(digestPath).exists(); + + long digest = Long.parseLong(Files.readString(digestPath)); + + Path data = ssTable.descriptor.pathFor(Component.DATA); + byte[] bytes = Files.readAllBytes(data); + + ChecksumType effectiveChecksumType = checksumType == null ? ChecksumType.CRC32 : checksumType; + long checksum = effectiveChecksumType.of(bytes, 0, bytes.length); + + assertThat(checksum).isEqualTo(digest); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + }); + } + } + } + + @Test + public void testDigestFileDefault() throws IOException + { + testDigestFile(null, Component.DIGEST); + } + + @Test + public void testDigestFileCRC32() throws IOException + { + testDigestFile(ChecksumType.CRC32, Component.DIGEST); + } + + @Test + public void testDigestFileCRC32C() throws IOException + { + testDigestFile(ChecksumType.CRC32C, Component.DIGEST_CRC32C); + } + + @Test + public void testDigestFileCRC64NVME() throws IOException + { + testDigestFile(ChecksumType.CRC64NVME, Component.DIGEST_CRC64NVME); + } + + private static String createTableStmt(String ks) + { + return format("CREATE TABLE %s.%s (pk int, ck int, v int, PRIMARY KEY (pk, ck)) " + + "WITH compaction = {'class':'SizeTieredCompactionStrategy', 'enabled':'false'}", + ks, "tbl"); + } +} diff --git a/test/microbench/org/apache/cassandra/test/microbench/ChecksumBench.java b/test/microbench/org/apache/cassandra/test/microbench/ChecksumBench.java index 5615f8c41f67..78b0cd3ec43f 100644 --- a/test/microbench/org/apache/cassandra/test/microbench/ChecksumBench.java +++ b/test/microbench/org/apache/cassandra/test/microbench/ChecksumBench.java @@ -21,7 +21,11 @@ import com.google.common.hash.Hasher; import com.google.common.hash.Hashing; import com.google.common.primitives.Longs; + +import org.apache.cassandra.utils.PureJavaCRC64NVME; +import software.amazon.awssdk.crt.checksums.CRC64NVME; import org.apache.cassandra.utils.ChecksumType; +import org.apache.cassandra.utils.FBUtilities; import org.openjdk.jmh.annotations.Benchmark; import org.openjdk.jmh.annotations.BenchmarkMode; import org.openjdk.jmh.annotations.Fork; @@ -36,9 +40,11 @@ import org.openjdk.jmh.annotations.Warmup; import org.xerial.snappy.PureJavaCrc32C; +import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.Random; import java.util.concurrent.TimeUnit; +import java.util.zip.CRC32C; @BenchmarkMode(Mode.AverageTime) @OutputTimeUnit(TimeUnit.NANOSECONDS) @@ -104,27 +110,59 @@ public byte[] benchPureJavaCrc32c() return Longs.toByteArray(pureJavaCrc32C.getValue()); } - // Below benchmarks are commented because CRC32C is unavailable in Java 8. -// @Benchmark -// @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", -// "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", -// }) -// public byte[] benchCrc32c() -// { -// CRC32C crc32C = new CRC32C(); -// crc32C.update(array); -// return Longs.toByteArray(crc32C.getValue()); -// } -// -// @Benchmark -// @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", -// "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", -// "-XX:+UnlockDiagnosticVMOptions", "-XX:-UseCRC32CIntrinsics", -// }) -// public byte[] benchCrc32cNoIntrinsic() -// { -// CRC32C crc32C = new CRC32C(); -// crc32C.update(array); -// return Longs.toByteArray(crc32C.getValue()); -// } + @Benchmark + @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", + "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", + }) + public byte[] benchCrc32c() + { + CRC32C crc32C = new CRC32C(); + crc32C.update(array); + return Longs.toByteArray(crc32C.getValue()); + } + + @Benchmark + @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", + "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", + "-XX:+UnlockDiagnosticVMOptions", "-XX:-UseCRC32CIntrinsics", + }) + public byte[] benchCrc32cNoIntrinsic() + { + CRC32C crc32C = new CRC32C(); + crc32C.update(array); + return Longs.toByteArray(crc32C.getValue()); + } + + @Benchmark + @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", + "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", + }) + public byte[] benchAwsCrtCrc64nvme() + { + CRC64NVME crc64nvme = new CRC64NVME(); + crc64nvme.update(array, 0, array.length); + return Longs.toByteArray(crc64nvme.getValue()); + } + + @Benchmark + @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", + "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", + }) + public byte[] benchPureJavaCrc64nvme() + { + PureJavaCRC64NVME crc64nvme = new PureJavaCRC64NVME(); + crc64nvme.update(array, 0, array.length); + return Longs.toByteArray(crc64nvme.getValue()); + } + + @Benchmark + @Fork(value = 1, jvmArgsAppend = { "-Xmx512M", "-Djmh.executor=CUSTOM", + "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor", + }) + public byte[] benchMd5() + { + MessageDigest md5 = FBUtilities.newMessageDigest("MD5"); + md5.update(array, 0, array.length); + return md5.digest(); + } } diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java index 9b92867df05e..be8c9c2c3ee9 100644 --- a/test/unit/org/apache/cassandra/db/ScrubTest.java +++ b/test/unit/org/apache/cassandra/db/ScrubTest.java @@ -266,7 +266,7 @@ public void testScrubOneBrokenPartitionInTheMiddleOfCompressedFile() throws Exec SSTableReader outputSSTable = cfs.getLiveSSTables().iterator().next(); assertThat(outputSSTable.descriptor.fileFor(Component.DIGEST).exists()).isTrue(); - try (DataIntegrityMetadata.FileDigestValidator validator = DataIntegrityMetadata.fileDigestValidator(outputSSTable.descriptor)) + try (DataIntegrityMetadata.FileDigestValidator validator = DataIntegrityMetadata.fileDigestValidator(outputSSTable.descriptor, Component.DIGEST)) { validator.validate(); } diff --git a/test/unit/org/apache/cassandra/db/VerifyTest.java b/test/unit/org/apache/cassandra/db/VerifyTest.java index 7123727eef08..075ef156be72 100644 --- a/test/unit/org/apache/cassandra/db/VerifyTest.java +++ b/test/unit/org/apache/cassandra/db/VerifyTest.java @@ -77,6 +77,7 @@ import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.ChecksumType; import org.apache.cassandra.utils.OutputHandler; import org.apache.cassandra.utils.UUIDGen; @@ -370,6 +371,57 @@ public void testVerifyIncorrectDigest() throws IOException, WriteTimeoutExceptio catch (RuntimeException err) {} } + private void testVerifyDigest(ChecksumType checksumType, Component digestComponent) throws Exception + { + CompactionManager.instance.disableAutoCompaction(); + Keyspace keyspace = Keyspace.open(KEYSPACE); + ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF); + + fillCF(cfs, 2); + + SSTableReader sstable = cfs.getLiveSSTables().iterator().next(); + + // Delete default digest file + sstable.descriptor.fileFor(Component.DIGEST).delete(); + + File digestFile = sstable.descriptor.fileFor(digestComponent); + File dataFile = sstable.descriptor.fileFor(Component.DATA); + + byte[] bytes = Files.readAllBytes(dataFile.toPath()); + long correctChecksum = checksumType.of(bytes, 0, bytes.length); + writeChecksum(correctChecksum, digestFile); + + try (Verifier verifier = new Verifier(cfs, sstable, false, Verifier.options().invokeDiskFailurePolicy(true).build())) + { + verifier.verify(); + } + catch (CorruptSSTableException err) + { + fail("Unexpected CorruptSSTableException"); + } + + writeChecksum(++correctChecksum, digestFile); + + try (Verifier verifier = new Verifier(cfs, sstable, false, Verifier.options().invokeDiskFailurePolicy(true).build())) + { + verifier.verify(); + fail("Expected a CorruptSSTableException to be thrown"); + } + catch (CorruptSSTableException err) {} + + } + + @Test + public void testVerifyDigestCRC32C() throws Exception + { + testVerifyDigest(ChecksumType.CRC32C, Component.DIGEST_CRC32C); + } + + @Test + public void testVerifyDigestCRC64NVME() throws Exception + { + testVerifyDigest(ChecksumType.CRC64NVME, Component.DIGEST_CRC64NVME); + } @Test public void testVerifyCorruptRowCorrectDigest() throws IOException, WriteTimeoutException diff --git a/test/unit/org/apache/cassandra/io/sstable/ComponentTest.java b/test/unit/org/apache/cassandra/io/sstable/ComponentTest.java index 5900fefc7fc7..0ba22b5ff7ab 100644 --- a/test/unit/org/apache/cassandra/io/sstable/ComponentTest.java +++ b/test/unit/org/apache/cassandra/io/sstable/ComponentTest.java @@ -39,6 +39,8 @@ public void testFromRepresentationExactMatches() assertEquals(Component.Type.COMPRESSION_INFO, Component.Type.fromRepresentation("CompressionInfo.db")); assertEquals(Component.Type.STATS, Component.Type.fromRepresentation("Statistics.db")); assertEquals(Component.Type.DIGEST, Component.Type.fromRepresentation("Digest.crc32")); + assertEquals(Component.Type.DIGEST_CRC32C, Component.Type.fromRepresentation("Digest.crc32c")); + assertEquals(Component.Type.DIGEST_CRC64NMVE, Component.Type.fromRepresentation("Digest.crc64nmve")); assertEquals(Component.Type.CRC, Component.Type.fromRepresentation("CRC.db")); assertEquals(Component.Type.SUMMARY, Component.Type.fromRepresentation("Summary.db")); assertEquals(Component.Type.TOC, Component.Type.fromRepresentation("TOC.txt")); diff --git a/test/unit/org/apache/cassandra/io/util/ChecksummedRandomAccessReaderTest.java b/test/unit/org/apache/cassandra/io/util/ChecksummedRandomAccessReaderTest.java index 8eee977b5622..fc1285432d44 100644 --- a/test/unit/org/apache/cassandra/io/util/ChecksummedRandomAccessReaderTest.java +++ b/test/unit/org/apache/cassandra/io/util/ChecksummedRandomAccessReaderTest.java @@ -48,7 +48,7 @@ public void readFully() throws IOException final byte[] expected = new byte[70 * 1024]; // bit more than crc chunk size, so we can test rebuffering. ThreadLocalRandom.current().nextBytes(expected); - try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, null, SequentialWriterOption.DEFAULT)) + try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, SequentialWriterOption.DEFAULT)) { writer.write(expected); writer.finish(); @@ -76,7 +76,7 @@ public void seek() throws IOException final byte[] dataBytes = new byte[70 * 1024]; // bit more than crc chunk size ThreadLocalRandom.current().nextBytes(dataBytes); - try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, null, SequentialWriterOption.DEFAULT)) + try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, SequentialWriterOption.DEFAULT)) { writer.write(dataBytes); writer.finish(); @@ -110,7 +110,7 @@ public void corruptionDetection() throws IOException final byte[] expected = new byte[5 * 1024]; Arrays.fill(expected, (byte) 0); - try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, null, SequentialWriterOption.DEFAULT)) + try (SequentialWriter writer = new ChecksummedSequentialWriter(data, crc, SequentialWriterOption.DEFAULT)) { writer.write(expected); writer.finish(); diff --git a/test/unit/org/apache/cassandra/io/util/ChecksummedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/util/ChecksummedSequentialWriterTest.java index 991d716e43ea..38236e3406bc 100644 --- a/test/unit/org/apache/cassandra/io/util/ChecksummedSequentialWriterTest.java +++ b/test/unit/org/apache/cassandra/io/util/ChecksummedSequentialWriterTest.java @@ -66,7 +66,7 @@ private TestableCSW() throws IOException private TestableCSW(File file, File crcFile) throws IOException { - this(file, crcFile, new ChecksummedSequentialWriter(file, crcFile, null, SequentialWriterOption.newBuilder() + this(file, crcFile, new ChecksummedSequentialWriter(file, crcFile, SequentialWriterOption.newBuilder() .bufferSize(DEFAULT_BUFFER_SIZE) .build())); } diff --git a/test/unit/org/apache/cassandra/utils/PureJavaCRC64NVMETest.java b/test/unit/org/apache/cassandra/utils/PureJavaCRC64NVMETest.java new file mode 100644 index 000000000000..c11463fb6e2f --- /dev/null +++ b/test/unit/org/apache/cassandra/utils/PureJavaCRC64NVMETest.java @@ -0,0 +1,98 @@ +/* + * 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.utils; + +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.Base64; + +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +// We use the same values to test as in AWS S3 SDK tests. +// See https://github.com/aws/aws-sdk-java-v2/blob/0c5e331bd2544e7d43fabd3db046a95b94d0a2dd/core/checksums/src/test/java/software/amazon/awssdk/checksums/internal/Crc64NvmeChecksumTest.java +public class PureJavaCRC64NVMETest +{ + + private PureJavaCRC64NVME crc64NVME; + private static final String TEST_STRING = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789"; + + @Before + public void setUp() + { + crc64NVME = new PureJavaCRC64NVME(); + } + + @Test + public void validateCrcChecksumValues() + { + byte[] bytes = TEST_STRING.getBytes(StandardCharsets.UTF_8); + crc64NVME.update(bytes, 0, bytes.length); + assertThat(getAsString(getAsBytes(crc64NVME.getValue()))).isEqualTo("0000000000000000000000008b8f30cfc6f16409"); + } + + @Test + public void validateEncodedBase64ForCrc() + { + crc64NVME.update("abc".getBytes(StandardCharsets.UTF_8)); + String toBase64 = toBase64(crc64NVME.getValue()); + assertThat(toBase64).isEqualTo("BeXKuz/B+us="); + } + + @Test + public void validateEncodedBase64ForCrcSingleByte() + { + for (byte value : "abc".getBytes(StandardCharsets.UTF_8)) + { + crc64NVME.update(value); + } + String toBase64 = toBase64(crc64NVME.getValue()); + assertThat(toBase64).isEqualTo("BeXKuz/B+us="); + } + + @Test + public void validateResetForCrc() { + crc64NVME.update("beta".getBytes(StandardCharsets.UTF_8)); + crc64NVME.reset(); + crc64NVME.update("alpha".getBytes(StandardCharsets.UTF_8)); + String toBase64 = toBase64(crc64NVME.getValue()); + // Checksum of "alpha" + assertThat(toBase64).isEqualTo("Ehnh98TMQlQ="); + } + + private String toBase64(long value) + { + byte[] bytes = getAsBytes(value); + return Base64.getEncoder().encodeToString(bytes); + } + + private byte[] getAsBytes(long value) + { + ByteBuffer buffer = ByteBuffer.allocate(Long.BYTES); + buffer.putLong(value); + return buffer.array(); + } + + private String getAsString(byte[] checksumBytes) { + return String.format("%040x", new BigInteger(1, checksumBytes)); + } +}