You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jz...@apache.org on 2018/05/01 22:30:00 UTC
[03/10] cassandra git commit: Fix compaction failure caused by
reading un-flushed data
Fix compaction failure caused by reading un-flushed data
patch by Jay Zhuang; reviewed by Marcus Eriksson for CASSANDRA-12743
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/3a713827
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/3a713827
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/3a713827
Branch: refs/heads/cassandra-3.11
Commit: 3a713827f48399f389ea851a19b8ec8cd2cc5773
Parents: 334dca9
Author: Jay Zhuang <ja...@yahoo.com>
Authored: Sat Apr 21 11:15:06 2018 -0700
Committer: Jay Zhuang <ja...@yahoo.com>
Committed: Tue May 1 15:07:01 2018 -0700
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../io/compress/CompressedSequentialWriter.java | 17 ++-
.../cassandra/io/util/SequentialWriter.java | 2 +
.../CompressedSequentialWriterReopenTest.java | 153 +++++++++++++++++++
.../CompressedSequentialWriterTest.java | 52 +++++++
.../cassandra/io/util/SequentialWriterTest.java | 41 +++++
6 files changed, 264 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 5f6189f..22ee346 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.2.13
+ * Fix compaction failure caused by reading un-flushed data (CASSANDRA-12743)
* Use Bounds instead of Range for sstables in anticompaction (CASSANDRA-14411)
* Fix JSON queries with IN restrictions and ORDER BY clause (CASSANDRA-14286)
* CQL fromJson(null) throws NullPointerException (CASSANDRA-13891)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
index 9c7c776..a7f9bb4 100644
--- a/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
+++ b/src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
@@ -129,7 +129,7 @@ public class CompressedSequentialWriter extends SequentialWriter
// write corresponding checksum
compressed.rewind();
crcMetadata.appendDirect(compressed, true);
- lastFlushOffset += compressedLength + 4;
+ lastFlushOffset = uncompressedSize;
// adjust our bufferOffset to account for the new uncompressed data we've now written out
resetBuffer();
@@ -235,10 +235,23 @@ public class CompressedSequentialWriter extends SequentialWriter
chunkCount = realMark.nextChunkIndex - 1;
// truncate data and index file
- truncate(chunkOffset);
+ truncate(chunkOffset, bufferOffset);
metadataWriter.resetAndTruncate(realMark.nextChunkIndex - 1);
}
+ private void truncate(long toFileSize, long toBufferOffset)
+ {
+ try
+ {
+ channel.truncate(toFileSize);
+ lastFlushOffset = toBufferOffset;
+ }
+ catch (IOException e)
+ {
+ throw new FSWriteError(e, getPath());
+ }
+ }
+
/**
* Seek to the offset where next compressed data chunk should be stored.
*/
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/src/java/org/apache/cassandra/io/util/SequentialWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/util/SequentialWriter.java b/src/java/org/apache/cassandra/io/util/SequentialWriter.java
index 0c39469..452318e 100644
--- a/src/java/org/apache/cassandra/io/util/SequentialWriter.java
+++ b/src/java/org/apache/cassandra/io/util/SequentialWriter.java
@@ -430,6 +430,7 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
throw new FSReadError(e, getPath());
}
+ bufferOffset = truncateTarget;
resetBuffer();
}
@@ -443,6 +444,7 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
try
{
channel.truncate(toSize);
+ lastFlushOffset = toSize;
}
catch (IOException e)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterReopenTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterReopenTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterReopenTest.java
new file mode 100644
index 0000000..33b4957
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterReopenTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+
+import org.junit.Test;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static org.junit.Assert.assertEquals;
+
+public class CompressedSequentialWriterReopenTest extends CQLTester
+{
+ @Test
+ public void badCompressor1() throws IOException
+ {
+ BadCompressor bad = new BadCompressor();
+ byte [] test = new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19};
+ byte [] out = new byte[10];
+ bad.uncompress(test, 0, 20, out, 0);
+ for (int i = 0; i < 10; i++)
+ assertEquals(out[i], (byte)i);
+ }
+
+ @Test
+ public void badCompressor2() throws IOException
+ {
+ BadCompressor bad = new BadCompressor();
+ ByteBuffer input = ByteBuffer.wrap(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19});
+ ByteBuffer output = ByteBuffer.allocate(40);
+ bad.compress(input, output);
+ for (int i = 0; i < 40; i++)
+ assertEquals(i % 20, output.get(i));
+ }
+
+ @Test
+ public void badCompressor3() throws IOException
+ {
+ BadCompressor bad = new BadCompressor();
+ ByteBuffer input = ByteBuffer.wrap(new byte[] {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19});
+ ByteBuffer output = ByteBuffer.allocate(10);
+ bad.uncompress(input, output);
+ for (int i = 0; i < 10; i++)
+ assertEquals(i, output.get(i));
+ }
+
+ @Test
+ public void compressionEnabled() throws Throwable
+ {
+ createTable("create table %s (id int primary key, t blob) with compression = {'sstable_compression':'org.apache.cassandra.io.compress.CompressedSequentialWriterReopenTest$BadCompressor'}");
+ byte [] blob = new byte[1000];
+ (new Random()).nextBytes(blob);
+ Keyspace keyspace = Keyspace.open(keyspace());
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(currentTable());
+ cfs.disableAutoCompaction();
+ for (int i = 0; i < 10000; i++)
+ {
+ execute("insert into %s (id, t) values (?, ?)", i, ByteBuffer.wrap(blob));
+ }
+ cfs.forceBlockingFlush();
+ for (int i = 0; i < 10000; i++)
+ {
+ execute("insert into %s (id, t) values (?, ?)", i, ByteBuffer.wrap(blob));
+ }
+ cfs.forceBlockingFlush();
+ DatabaseDescriptor.setSSTablePreempiveOpenIntervalInMB(1);
+ cfs.forceMajorCompaction();
+ }
+
+ public static class BadCompressor implements ICompressor
+ {
+ public static ICompressor create(Map<String, String> options)
+ {
+ return new BadCompressor();
+ }
+
+ @Override
+ public int initialCompressedBufferLength(int chunkLength)
+ {
+ return chunkLength * 2;
+ }
+
+ @Override
+ public int uncompress(byte[] input, int inputOffset, int inputLength, byte[] output, int outputOffset) throws IOException
+ {
+ System.arraycopy(input, inputOffset, output, outputOffset, inputLength / 2);
+ return inputLength / 2;
+ }
+
+ @Override
+ public void compress(ByteBuffer input, ByteBuffer output) throws IOException
+ {
+ int len = input.remaining();
+ byte [] arr = ByteBufferUtil.getArray(input);
+ output.put(arr);
+ output.put(arr);
+ input.position(len);
+ }
+
+ @Override
+ public void uncompress(ByteBuffer input, ByteBuffer output) throws IOException
+ {
+ byte [] arr = ByteBufferUtil.getArray(input);
+ output.put(arr, 0, arr.length / 2);
+ input.position(arr.length);
+ }
+
+ @Override
+ public BufferType preferredBufferType()
+ {
+ return BufferType.ON_HEAP;
+ }
+
+ @Override
+ public boolean supports(BufferType bufferType)
+ {
+ return true;
+ }
+
+ @Override
+ public Set<String> supportedOptions()
+ {
+ return new HashSet<>(Arrays.asList(CompressionParameters.CRC_CHECK_CHANCE));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
index 43c44fd..bca0354 100644
--- a/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
@@ -17,6 +17,7 @@
*/
package org.apache.cassandra.io.compress;
+import com.google.common.io.Files;
import java.io.ByteArrayInputStream;
import java.io.DataInputStream;
import java.io.File;
@@ -26,6 +27,7 @@ import java.util.*;
import static org.apache.commons.io.FileUtils.readFileToByteArray;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import org.junit.After;
import org.junit.Test;
@@ -38,7 +40,9 @@ import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.io.sstable.metadata.MetadataCollector;
import org.apache.cassandra.io.util.ChannelProxy;
import org.apache.cassandra.io.util.FileMark;
+import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.io.util.RandomAccessReader;
+import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.io.util.SequentialWriterTest;
public class CompressedSequentialWriterTest extends SequentialWriterTest
@@ -107,6 +111,12 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest
{
writer.write((byte)i);
}
+
+ if (bytesToTest <= CompressionParameters.DEFAULT_CHUNK_LENGTH)
+ assertEquals(writer.getLastFlushOffset(), CompressionParameters.DEFAULT_CHUNK_LENGTH);
+ else
+ assertTrue(writer.getLastFlushOffset() % CompressionParameters.DEFAULT_CHUNK_LENGTH == 0);
+
writer.resetAndTruncate(mark);
writer.write(dataPost);
writer.finish();
@@ -155,6 +165,48 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest
writers.clear();
}
+ @Test
+ @Override
+ public void resetAndTruncateTest()
+ {
+ File tempFile = new File(Files.createTempDir(), "reset.txt");
+ File offsetsFile = FileUtils.createTempFile("compressedsequentialwriter.offset", "test");
+ final int bufferSize = 48;
+ final int writeSize = 64;
+ byte[] toWrite = new byte[writeSize];
+
+ try (SequentialWriter writer = new CompressedSequentialWriter(tempFile, offsetsFile.getPath(),
+ new CompressionParameters(LZ4Compressor.instance),new MetadataCollector(CellNames.fromAbstractType(UTF8Type.instance, false))))
+ {
+ // write bytes greather than buffer
+ writer.write(toWrite);
+ long flushedOffset = writer.getLastFlushOffset();
+ assertEquals(writeSize, writer.getFilePointer());
+ // mark thi position
+ FileMark pos = writer.mark();
+ // write another
+ writer.write(toWrite);
+ // another buffer should be flushed
+ assertEquals(flushedOffset * 2, writer.getLastFlushOffset());
+ assertEquals(writeSize * 2, writer.getFilePointer());
+ // reset writer
+ writer.resetAndTruncate(pos);
+ // current position and flushed size should be changed
+ assertEquals(writeSize, writer.getFilePointer());
+ assertEquals(flushedOffset, writer.getLastFlushOffset());
+ // write another byte less than buffer
+ writer.write(new byte[]{0});
+ assertEquals(writeSize + 1, writer.getFilePointer());
+ // flush off set should not be increase
+ assertEquals(flushedOffset, writer.getLastFlushOffset());
+ writer.finish();
+ }
+ catch (IOException e)
+ {
+ Assert.fail();
+ }
+ }
+
protected TestableTransaction newTest() throws IOException
{
TestableCSW sw = new TestableCSW();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/3a713827/test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java b/test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java
index fd38427..15d6160 100644
--- a/test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java
@@ -36,6 +36,7 @@ import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.utils.concurrent.AbstractTransactionalTest;
import static org.apache.commons.io.FileUtils.*;
+import static org.junit.Assert.assertEquals;
public class SequentialWriterTest extends AbstractTransactionalTest
{
@@ -119,6 +120,46 @@ public class SequentialWriterTest extends AbstractTransactionalTest
}
}
+ @Test
+ public void resetAndTruncateTest()
+ {
+ File tempFile = new File(Files.createTempDir(), "reset.txt");
+ final int bufferSize = 48;
+ final int writeSize = 64;
+ byte[] toWrite = new byte[writeSize];
+ try (SequentialWriter writer = new SequentialWriter(tempFile, bufferSize, BufferType.OFF_HEAP))
+ {
+ // write bytes greather than buffer
+ writer.write(toWrite);
+ assertEquals(bufferSize, writer.getLastFlushOffset());
+ assertEquals(writeSize, writer.getFilePointer());
+ // mark thi position
+ FileMark pos = writer.mark();
+ // write another
+ writer.write(toWrite);
+ // another buffer should be flushed
+ assertEquals(bufferSize * 2, writer.getLastFlushOffset());
+ assertEquals(writeSize * 2, writer.getFilePointer());
+ // reset writer
+ writer.resetAndTruncate(pos);
+ // current position and flushed size should be changed
+ assertEquals(writeSize, writer.getFilePointer());
+ assertEquals(writeSize, writer.getLastFlushOffset());
+ // write another byte less than buffer
+ writer.write(new byte[]{0});
+ assertEquals(writeSize + 1, writer.getFilePointer());
+ // flush off set should not be increase
+ assertEquals(writeSize, writer.getLastFlushOffset());
+ writer.finish();
+ }
+ catch (IOException e)
+ {
+ Assert.fail();
+ }
+ // final file size check
+ assertEquals(writeSize + 1, tempFile.length());
+ }
+
/**
* Tests that the output stream exposed by SequentialWriter behaves as expected
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org