mirror of https://github.com/apache/cassandra
Improved DeletionTime serialization
This commit is contained in:
Bereng
parent
d08fd1ebd8
commit
2a458ba01d
|
|
@ -195,42 +195,103 @@ public class DeletionTime implements Comparable<DeletionTime>, IMeasurableMemory
|
|||
return legacySerializer;
|
||||
}
|
||||
|
||||
// Serializer for Usigned Integer ldt
|
||||
/* Serializer for Usigned Integer ldt
|
||||
*
|
||||
* ldt is encoded as a uint in seconds since unix epoch, it can go up o 2106-02-07T06:28:13+00:00 only.
|
||||
* mfda is a positive timestamp. We use the sign bit to encode LIVE DeletionTimes
|
||||
* Throw IOException to mark sstable as corrupt.
|
||||
*/
|
||||
public static class Serializer implements ISerializer<DeletionTime>
|
||||
{
|
||||
// We use the sign bit to signal LIVE DeletionTimes
|
||||
private final static int IS_LIVE_DELETION = 0b1000_0000;
|
||||
|
||||
public void serialize(DeletionTime delTime, DataOutputPlus out) throws IOException
|
||||
{
|
||||
out.writeInt(delTime.localDeletionTimeUnsignedInteger);
|
||||
out.writeLong(delTime.markedForDeleteAt());
|
||||
if (delTime == LIVE)
|
||||
out.writeByte(IS_LIVE_DELETION);
|
||||
else
|
||||
{
|
||||
// The sign bit is zero here, so we can write a long directly
|
||||
out.writeLong(delTime.markedForDeleteAt());
|
||||
out.writeInt(delTime.localDeletionTimeUnsignedInteger);
|
||||
}
|
||||
}
|
||||
|
||||
public DeletionTime deserialize(DataInputPlus in) throws IOException
|
||||
{
|
||||
int localDeletionTimeUnsignedInteger = in.readInt();
|
||||
long mfda = in.readLong();
|
||||
return mfda == Long.MIN_VALUE && localDeletionTimeUnsignedInteger == Cell.NO_DELETION_TIME_UNSIGNED_INTEGER
|
||||
? LIVE
|
||||
: new DeletionTime(mfda, localDeletionTimeUnsignedInteger);
|
||||
int flags = in.readByte();
|
||||
if ((flags & IS_LIVE_DELETION) != 0)
|
||||
{
|
||||
if ((flags & 0xFF) != IS_LIVE_DELETION)
|
||||
throw new IOException("Corrupted sstable. Invalid flags found deserializing DeletionTime: " + Integer.toBinaryString(flags & 0xFF));
|
||||
return LIVE;
|
||||
}
|
||||
else
|
||||
{
|
||||
// Read the remaining 7 bytes
|
||||
int bytes1 = in.readByte();
|
||||
int bytes2 = in.readShort();
|
||||
int bytes4 = in.readInt();
|
||||
|
||||
long mfda = readBytesToMFDA(flags, bytes1, bytes2, bytes4);
|
||||
int localDeletionTimeUnsignedInteger = in.readInt();
|
||||
|
||||
return new DeletionTime(mfda, localDeletionTimeUnsignedInteger);
|
||||
}
|
||||
}
|
||||
|
||||
public DeletionTime deserialize(ByteBuffer buf, int offset)
|
||||
public DeletionTime deserialize(ByteBuffer buf, int offset) throws IOException
|
||||
{
|
||||
int localDeletionTimeUnsignedInteger = buf.getInt(offset);
|
||||
long mfda = buf.getLong(offset + 4);
|
||||
return mfda == Long.MIN_VALUE && localDeletionTimeUnsignedInteger == Cell.NO_DELETION_TIME_UNSIGNED_INTEGER
|
||||
? LIVE
|
||||
: new DeletionTime(mfda, localDeletionTimeUnsignedInteger);
|
||||
int flags = buf.get(offset);
|
||||
if ((flags & IS_LIVE_DELETION) != 0)
|
||||
{
|
||||
if ((flags & 0xFF) != IS_LIVE_DELETION)
|
||||
throw new IOException("Corrupted sstable. Invalid flags found deserializing DeletionTime: " + Integer.toBinaryString(flags & 0xFF));
|
||||
return LIVE;
|
||||
}
|
||||
else
|
||||
{
|
||||
long mfda = buf.getLong(offset);
|
||||
int localDeletionTimeUnsignedInteger = buf.getInt(offset + TypeSizes.LONG_SIZE);
|
||||
|
||||
return new DeletionTime(mfda, localDeletionTimeUnsignedInteger);
|
||||
}
|
||||
}
|
||||
|
||||
public void skip(DataInputPlus in) throws IOException
|
||||
{
|
||||
in.skipBytesFully(4 + 8);
|
||||
int flags = in.readByte();
|
||||
if ((flags & IS_LIVE_DELETION) != 0)
|
||||
{
|
||||
if ((flags & 0xFF) != IS_LIVE_DELETION)
|
||||
throw new IOException("Corrupted sstable. Invalid flags found deserializing DeletionTime: " + Integer.toBinaryString(flags & 0xFF));
|
||||
// We read the flags already and there's nothing left to skip over
|
||||
return;
|
||||
}
|
||||
else
|
||||
// We read the flags already but there is mfda and ldt to skip over still
|
||||
in.skipBytesFully(TypeSizes.LONG_SIZE - 1 + TypeSizes.INT_SIZE);
|
||||
}
|
||||
|
||||
public long serializedSize(DeletionTime delTime)
|
||||
{
|
||||
return TypeSizes.sizeof(Integer.MAX_VALUE)
|
||||
+ TypeSizes.sizeof(delTime.markedForDeleteAt());
|
||||
if (delTime == LIVE)
|
||||
// Just the flags
|
||||
return 1;
|
||||
else
|
||||
// 1 for the flags, 7 for mfda and 4 for ldt
|
||||
return TypeSizes.LONG_SIZE
|
||||
+ TypeSizes.INT_SIZE;
|
||||
}
|
||||
|
||||
private long readBytesToMFDA(int flagsByte, int bytes1, int bytes2, int bytes4)
|
||||
{
|
||||
long mfda = flagsByte & 0xFFL;
|
||||
mfda = (mfda << 8) + (bytes1 & 0xFFL);
|
||||
mfda = (mfda << 16) + (bytes2 & 0xFFFFL);
|
||||
mfda = (mfda << 32) + (bytes4 & 0xFFFFFFFFL);
|
||||
return mfda;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -443,7 +443,7 @@ public class PartitionIndex implements SharedCloseable
|
|||
}
|
||||
}
|
||||
|
||||
private void dumpTrie(PrintStream out)
|
||||
private void dumpTrie(PrintStream out) throws IOException
|
||||
{
|
||||
try (Reader rdr = openReader())
|
||||
{
|
||||
|
|
|
|||
|
|
@ -76,8 +76,9 @@ public class RowIndexReader extends Walker<RowIndexReader>
|
|||
|
||||
/**
|
||||
* Computes the floor for a given key.
|
||||
* @throws IOException
|
||||
*/
|
||||
public IndexInfo separatorFloor(ByteComparable key)
|
||||
public IndexInfo separatorFloor(ByteComparable key) throws IOException
|
||||
{
|
||||
// Check for a prefix and find closest smaller branch.
|
||||
IndexInfo res = prefixAndNeighbours(key, RowIndexReader::readPayload);
|
||||
|
|
@ -96,23 +97,23 @@ public class RowIndexReader extends Walker<RowIndexReader>
|
|||
return getCurrentIndexInfo();
|
||||
}
|
||||
|
||||
public IndexInfo min()
|
||||
public IndexInfo min() throws IOException
|
||||
{
|
||||
goMin(root);
|
||||
return getCurrentIndexInfo();
|
||||
}
|
||||
|
||||
protected IndexInfo getCurrentIndexInfo()
|
||||
protected IndexInfo getCurrentIndexInfo() throws IOException
|
||||
{
|
||||
return readPayload(payloadPosition(), payloadFlags());
|
||||
}
|
||||
|
||||
protected IndexInfo readPayload(int ppos, int bits)
|
||||
protected IndexInfo readPayload(int ppos, int bits) throws IOException
|
||||
{
|
||||
return readPayload(buf, ppos, bits, version);
|
||||
}
|
||||
|
||||
static IndexInfo readPayload(ByteBuffer buf, int ppos, int bits, Version version)
|
||||
static IndexInfo readPayload(ByteBuffer buf, int ppos, int bits, Version version) throws IOException
|
||||
{
|
||||
long dataOffset;
|
||||
if (bits == 0)
|
||||
|
|
@ -183,12 +184,12 @@ public class RowIndexReader extends Walker<RowIndexReader>
|
|||
|
||||
// debug/test code
|
||||
@SuppressWarnings("unused")
|
||||
public void dumpTrie(PrintStream out)
|
||||
public void dumpTrie(PrintStream out) throws IOException
|
||||
{
|
||||
dumpTrie(out, RowIndexReader::dumpRowIndexEntry, version);
|
||||
}
|
||||
|
||||
static String dumpRowIndexEntry(ByteBuffer buf, int ppos, int bits, Version version)
|
||||
static String dumpRowIndexEntry(ByteBuffer buf, int ppos, int bits, Version version) throws IOException
|
||||
{
|
||||
IndexInfo ii = readPayload(buf, ppos, bits, version);
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
*/
|
||||
package org.apache.cassandra.io.sstable.format.bti;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.PrintStream;
|
||||
|
||||
import org.apache.cassandra.io.sstable.format.Version;
|
||||
|
|
@ -44,7 +45,7 @@ class RowIndexReverseIterator extends ReverseValueIterator<RowIndexReverseIterat
|
|||
this(file, entry.indexTrieRoot, ByteComparable.EMPTY, end, version);
|
||||
}
|
||||
|
||||
public IndexInfo nextIndexInfo()
|
||||
public IndexInfo nextIndexInfo() throws IOException
|
||||
{
|
||||
if (currentNode == -1)
|
||||
{
|
||||
|
|
@ -62,7 +63,7 @@ class RowIndexReverseIterator extends ReverseValueIterator<RowIndexReverseIterat
|
|||
|
||||
// debug/test code
|
||||
@SuppressWarnings("unused")
|
||||
public void dumpTrie(PrintStream out)
|
||||
public void dumpTrie(PrintStream out) throws IOException
|
||||
{
|
||||
dumpTrie(out, RowIndexReader::dumpRowIndexEntry, version);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
*/
|
||||
package org.apache.cassandra.io.tries;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.PrintStream;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
|
|
@ -184,7 +185,7 @@ public class Walker<CONCRETE extends Walker<CONCRETE>> implements AutoCloseable
|
|||
|
||||
public interface Extractor<RESULT, VALUE>
|
||||
{
|
||||
RESULT extract(VALUE walker, int payloadPosition, int payloadFlags);
|
||||
RESULT extract(VALUE walker, int payloadPosition, int payloadFlags) throws IOException;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -268,9 +269,10 @@ public class Walker<CONCRETE extends Walker<CONCRETE>> implements AutoCloseable
|
|||
* is in the trie when looking for 'abc' or 'ac', but accepted when looking for 'aa').
|
||||
* In order to not have to go back to data that may have exited cache, payloads are extracted when the node is
|
||||
* visited (instead of saving the node's position), which requires an extractor to be passed as parameter.
|
||||
* @throws IOException
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <RESULT> RESULT prefix(ByteComparable key, Extractor<RESULT, CONCRETE> extractor)
|
||||
public <RESULT> RESULT prefix(ByteComparable key, Extractor<RESULT, CONCRETE> extractor) throws IOException
|
||||
{
|
||||
RESULT payload = null;
|
||||
|
||||
|
|
@ -304,9 +306,10 @@ public class Walker<CONCRETE extends Walker<CONCRETE>> implements AutoCloseable
|
|||
* does not take that into account. E.g. if trie contains "abba", "as" and "ask", looking for "asking" will find
|
||||
* "ask" as the match, but max(lesserBranch) will point to "abba" instead of the correct "as". This problem can
|
||||
* only occur if there is a valid prefix match.
|
||||
* @throws IOException
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public <RESULT> RESULT prefixAndNeighbours(ByteComparable key, Extractor<RESULT, CONCRETE> extractor)
|
||||
public <RESULT> RESULT prefixAndNeighbours(ByteComparable key, Extractor<RESULT, CONCRETE> extractor) throws IOException
|
||||
{
|
||||
RESULT payload = null;
|
||||
greaterBranch = NONE;
|
||||
|
|
@ -358,16 +361,16 @@ public class Walker<CONCRETE extends Walker<CONCRETE>> implements AutoCloseable
|
|||
|
||||
public interface PayloadToString
|
||||
{
|
||||
String payloadAsString(ByteBuffer buf, int payloadPos, int payloadFlags, Version version);
|
||||
String payloadAsString(ByteBuffer buf, int payloadPos, int payloadFlags, Version version) throws IOException;
|
||||
}
|
||||
|
||||
public void dumpTrie(PrintStream out, PayloadToString payloadReader, Version version)
|
||||
public void dumpTrie(PrintStream out, PayloadToString payloadReader, Version version) throws IOException
|
||||
{
|
||||
out.print("ROOT");
|
||||
dumpTrie(out, payloadReader, root, "", version);
|
||||
}
|
||||
|
||||
private void dumpTrie(PrintStream out, PayloadToString payloadReader, long node, String indent, Version version)
|
||||
private void dumpTrie(PrintStream out, PayloadToString payloadReader, long node, String indent, Version version) throws IOException
|
||||
{
|
||||
go(node);
|
||||
int bits = payloadFlags();
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1178851237
|
||||
1983729251
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1260299575
|
||||
2020062012
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
2099476497
|
||||
1689908368
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1213045656
|
||||
2128362302
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1331566588
|
||||
103182460
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1886131535
|
||||
3206431625
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
1899914505
|
||||
309360903
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +1 @@
|
|||
3169912797
|
||||
2437993698
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,366 @@
|
|||
/*
|
||||
* 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.test.microbench;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.io.RandomAccessFile;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.channels.FileChannel;
|
||||
import java.nio.file.NoSuchFileException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Random;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.cassandra.db.DeletionTime;
|
||||
import org.apache.cassandra.db.DeletionTime.Serializer;
|
||||
import org.apache.cassandra.io.util.DataInputBuffer;
|
||||
import org.apache.cassandra.io.util.DataInputPlus;
|
||||
import org.apache.cassandra.io.util.DataOutputBuffer;
|
||||
import org.apache.cassandra.io.util.DataOutputStreamPlus;
|
||||
import org.apache.cassandra.io.util.FileInputStreamPlus;
|
||||
import org.apache.cassandra.io.util.FileOutputStreamPlus;
|
||||
import org.openjdk.jmh.annotations.Benchmark;
|
||||
import org.openjdk.jmh.annotations.BenchmarkMode;
|
||||
import org.openjdk.jmh.annotations.Fork;
|
||||
import org.openjdk.jmh.annotations.Measurement;
|
||||
import org.openjdk.jmh.annotations.Mode;
|
||||
import org.openjdk.jmh.annotations.OutputTimeUnit;
|
||||
import org.openjdk.jmh.annotations.Param;
|
||||
import org.openjdk.jmh.annotations.Scope;
|
||||
import org.openjdk.jmh.annotations.State;
|
||||
import org.openjdk.jmh.annotations.TearDown;
|
||||
import org.openjdk.jmh.annotations.Threads;
|
||||
import org.openjdk.jmh.annotations.Warmup;
|
||||
import org.openjdk.jmh.infra.Blackhole;
|
||||
|
||||
@BenchmarkMode(Mode.AverageTime)
|
||||
@OutputTimeUnit(TimeUnit.NANOSECONDS)
|
||||
@Warmup(iterations = 2, time = 1, timeUnit = TimeUnit.SECONDS)
|
||||
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
|
||||
@Fork(value = 3, jvmArgsAppend = "-Xmx512M")
|
||||
@Threads(1)
|
||||
@State(Scope.Benchmark)
|
||||
public class DeletionTimeDeSerBench
|
||||
{
|
||||
final static int BUFFER_SIZE = 1024 * 1024;
|
||||
final static int DT_STD_SIZE = 8 + 4; // Long + Int
|
||||
|
||||
final static Random random = new Random(100);
|
||||
final static ArrayList<DeletionTime> TEST_DTS_70PC_LIVE = generateDTs(70);
|
||||
final static ArrayList<DeletionTime> TEST_DTS_30PC_LIVE = generateDTs(30);
|
||||
|
||||
// Parameters
|
||||
final static String ncSstableParam = "NC";
|
||||
final static String oaSstableParam = "OA";
|
||||
@Param({ ncSstableParam, oaSstableParam })
|
||||
String sstableParam;
|
||||
|
||||
final static String live70PcParam = "70PcLive";
|
||||
final static String live30PcParam = "30PcLive";
|
||||
@Param({ live70PcParam, live30PcParam })
|
||||
String liveDTPcParam;
|
||||
|
||||
final static String RAMParam = "RAM";
|
||||
final static String diskParam = "Disk";
|
||||
@Param({ RAMParam, diskParam })
|
||||
String diskRAMParam;
|
||||
|
||||
// Files
|
||||
File serMMapedFile = new File(DeletionTimeDeSerBench.class + "_Sermmap");
|
||||
ByteBuffer serMMap = allocateMmapedByteBuffer(serMMapedFile, BUFFER_SIZE);
|
||||
File deserMMapedFile = new File(DeletionTimeDeSerBench.class + "_Desermmap");
|
||||
ByteBuffer deserMMap = allocateMmapedByteBuffer(deserMMapedFile, BUFFER_SIZE);
|
||||
File diskFile = new File(DeletionTimeDeSerBench.class + "_DTBenchTest");
|
||||
|
||||
@TearDown
|
||||
public void tearDown()
|
||||
{
|
||||
serMMapedFile.delete();
|
||||
deserMMapedFile.delete();
|
||||
diskFile.delete();
|
||||
}
|
||||
|
||||
@Benchmark
|
||||
public void testE2ESerializeDT(final Blackhole bh) throws IOException
|
||||
{
|
||||
ByteBuffer buffer = serMMap;
|
||||
Serializer serializer = getSerializer(sstableParam);
|
||||
ArrayList<DeletionTime> dts = getDTs(liveDTPcParam);
|
||||
|
||||
try(DataOutputStreamPlus out = getSerOut(diskRAMParam, serMMap))
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
serializer.serialize(dts.get(i), out);
|
||||
bh.consume(out);
|
||||
}
|
||||
|
||||
bh.consume(buffer);
|
||||
bh.consume(serializer);
|
||||
bh.consume(dts);
|
||||
buffer.clear();
|
||||
}
|
||||
|
||||
@Benchmark
|
||||
public void testE2EDeSerializeDT(final Blackhole bh) throws IOException
|
||||
{
|
||||
ByteBuffer buffer = deserMMap;
|
||||
Serializer serializer = getSerializer(sstableParam);
|
||||
ArrayList<DeletionTime> dts = getDTs(liveDTPcParam);
|
||||
|
||||
try(DataOutputStreamPlus out = getSerOut(diskRAMParam, deserMMap))
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
serializer.serialize(dts.get(i), out);
|
||||
bh.consume(out);
|
||||
}
|
||||
|
||||
DataInputPlus in = getDeSerIn(diskRAMParam, deserMMap);
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
serializer.deserialize(in);
|
||||
|
||||
bh.consume(buffer);
|
||||
bh.consume(serializer);
|
||||
bh.consume(dts);
|
||||
bh.consume(in);
|
||||
buffer.clear();
|
||||
}
|
||||
|
||||
/**
|
||||
* Test the impact of the different bit ops and write calls. Against memory only.
|
||||
*/
|
||||
@Benchmark
|
||||
public void testRawAlgWrites(final Blackhole bh)
|
||||
{
|
||||
if (diskRAMParam != RAMParam)
|
||||
return;
|
||||
|
||||
ArrayList<DeletionTime> dts = getDTs(liveDTPcParam);
|
||||
ByteBuffer buffer = ByteBuffer.allocate(DT_STD_SIZE);
|
||||
|
||||
if (sstableParam == oaSstableParam)
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
{
|
||||
buffer.clear();
|
||||
if (dts.get(i).equals(DeletionTime.LIVE))
|
||||
buffer.put((byte) 0b1000_0000);
|
||||
else
|
||||
{
|
||||
buffer.putLong(Math.abs(random.nextLong()));
|
||||
buffer.putInt(Math.abs(random.nextInt()));
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
{
|
||||
buffer.clear();
|
||||
buffer.putLong(Math.abs(random.nextLong()));
|
||||
buffer.putInt(Math.abs(random.nextInt()));
|
||||
}
|
||||
}
|
||||
|
||||
bh.consume(dts);
|
||||
bh.consume(buffer);
|
||||
}
|
||||
|
||||
/**
|
||||
* Test the impact of the different bit ops and read calls. Against memory only.
|
||||
*/
|
||||
@Benchmark
|
||||
public void testRawAlgReads(final Blackhole bh)
|
||||
{
|
||||
if (diskRAMParam != RAMParam)
|
||||
return;
|
||||
|
||||
ArrayList<DeletionTime> dts = getDTs(liveDTPcParam);
|
||||
ByteBuffer buffer = ByteBuffer.allocate(DT_STD_SIZE);
|
||||
buffer.putLong(random.nextLong() & Long.MAX_VALUE); // to prevent Match.abs(MIN_VALUE) == MIN_VALUE
|
||||
buffer.putInt(random.nextInt() & Integer.MAX_VALUE);
|
||||
|
||||
if (sstableParam == oaSstableParam)
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
{
|
||||
buffer.clear();
|
||||
|
||||
if (dts.get(i).equals(DeletionTime.LIVE))
|
||||
{
|
||||
// LIVE: we only read the flags Byte
|
||||
byte b = buffer.get();
|
||||
b = (byte) (b & 0b1000_0000 & 0xFF);
|
||||
bh.consume(b);
|
||||
}
|
||||
else
|
||||
{
|
||||
// The flag
|
||||
byte flag = buffer.get();
|
||||
|
||||
// 7 bytes mfda
|
||||
int bytes4 = buffer.getInt();
|
||||
int bytes2 = buffer.getShort();
|
||||
int bytes1 = buffer.get();
|
||||
|
||||
long mfda = flag & 0xFFL;
|
||||
mfda = (mfda << 32) + (bytes4 & 0xFFFFFFFFL);
|
||||
mfda = (mfda << 16) + (bytes2 & 0xFFFFL);
|
||||
mfda = (mfda << 8) + (bytes1 & 0xFFL);
|
||||
|
||||
// The ldt
|
||||
int ldt = buffer.getInt();
|
||||
|
||||
bh.consume(flag);
|
||||
bh.consume(bytes4);
|
||||
bh.consume(bytes2);
|
||||
bh.consume(bytes1);
|
||||
bh.consume(mfda);
|
||||
bh.consume(ldt);
|
||||
}
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
for (int i = 0, m = dts.size(); i < m; i++)
|
||||
{
|
||||
buffer.clear();
|
||||
long mfda = buffer.getLong();
|
||||
int ldt = buffer.getInt();
|
||||
|
||||
bh.consume(mfda);
|
||||
bh.consume(ldt);
|
||||
}
|
||||
}
|
||||
|
||||
bh.consume(dts);
|
||||
bh.consume(buffer);
|
||||
}
|
||||
|
||||
private DataInputPlus getDeSerIn(String value, ByteBuffer maybeBackingBB)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case RAMParam:
|
||||
maybeBackingBB.rewind();
|
||||
return new DataInputBuffer(maybeBackingBB, false);
|
||||
|
||||
case diskParam:
|
||||
try
|
||||
{
|
||||
return new FileInputStreamPlus(new org.apache.cassandra.io.util.File(diskFile));
|
||||
}
|
||||
catch(NoSuchFileException e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
default:
|
||||
throw new IllegalStateException("Unknown backing media for deser.");
|
||||
}
|
||||
}
|
||||
|
||||
private DataOutputStreamPlus getSerOut(String value, ByteBuffer maybeBackingBB)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case RAMParam:
|
||||
maybeBackingBB.rewind();
|
||||
return new DataOutputBuffer(maybeBackingBB);
|
||||
|
||||
case diskParam:
|
||||
try
|
||||
{
|
||||
return new FileOutputStreamPlus(new org.apache.cassandra.io.util.File(diskFile));
|
||||
}
|
||||
catch(NoSuchFileException e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
|
||||
default:
|
||||
throw new IllegalStateException("Unknown backing media for ser.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates DeletionTimes where livePercentage are DeletionTime.LIVE
|
||||
* @param livePercentage
|
||||
* @return
|
||||
*/
|
||||
private static ArrayList<DeletionTime> generateDTs(int livePercentage)
|
||||
{
|
||||
ArrayList<DeletionTime> dts = new ArrayList<>(BUFFER_SIZE / DT_STD_SIZE);
|
||||
for (int i = 0; i < (BUFFER_SIZE / (DT_STD_SIZE)); i++)
|
||||
{
|
||||
DeletionTime dt;
|
||||
|
||||
if ((random.nextInt() % 100) < livePercentage)
|
||||
dt = DeletionTime.LIVE;
|
||||
else
|
||||
dt = DeletionTime.build(Math.abs(random.nextLong()), Math.abs(random.nextInt()));
|
||||
|
||||
dts.add(dt);
|
||||
}
|
||||
|
||||
return dts;
|
||||
}
|
||||
|
||||
private static ArrayList<DeletionTime> getDTs(String value)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case live70PcParam:
|
||||
return TEST_DTS_70PC_LIVE;
|
||||
|
||||
case live30PcParam:
|
||||
return TEST_DTS_30PC_LIVE;
|
||||
|
||||
default:
|
||||
throw new IllegalStateException("Unknown live DT PC");
|
||||
}
|
||||
}
|
||||
|
||||
private static Serializer getSerializer(String value)
|
||||
{
|
||||
switch (value)
|
||||
{
|
||||
case ncSstableParam:
|
||||
return new DeletionTime.LegacySerializer();
|
||||
|
||||
case oaSstableParam:
|
||||
return new DeletionTime.Serializer();
|
||||
|
||||
default:
|
||||
throw new IllegalStateException("Unknown serializer");
|
||||
}
|
||||
}
|
||||
|
||||
private static ByteBuffer allocateMmapedByteBuffer(File mmapFile, int bufferSize)
|
||||
{
|
||||
try(RandomAccessFile file = new RandomAccessFile(mmapFile, "rw"); FileChannel ch = file.getChannel())
|
||||
{
|
||||
return ch.map(FileChannel.MapMode.READ_WRITE, 0, bufferSize);
|
||||
}
|
||||
catch(IOException e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -186,12 +186,14 @@ public class CompactionsTest
|
|||
// disable compaction while flushing
|
||||
store.disableAutoCompaction();
|
||||
|
||||
//Populate sstable1 with with keys [0a..29a]
|
||||
populate(KEYSPACE1, CF_STANDARD1, 0, 29, "a",3); //ttl=3s
|
||||
//Populate sstable1 with with keys [0a..29aaaaaaaaaaa] Partitions have to be big enough
|
||||
//that prevent the size dependent AbstractCompactionStrategy.worthDroppingTombstones to trigger
|
||||
//a compaction.
|
||||
populate(KEYSPACE1, CF_STANDARD1, 0, 29, "aaaaaaaaaaa",3); //ttl=3s
|
||||
Util.flush(store);
|
||||
|
||||
//Populate sstable2 with with keys [0b..29b] (keys do not overlap with SSTable1, but the range is almost fully covered)
|
||||
long timestamp2 = populate(KEYSPACE1, CF_STANDARD1, 0, 29, "b", 3); //ttl=3s
|
||||
long timestamp2 = populate(KEYSPACE1, CF_STANDARD1, 0, 29, "bbbbbbbbbbb", 3); //ttl=3s
|
||||
Util.flush(store);
|
||||
|
||||
assertEquals(2, store.getLiveSSTables().size());
|
||||
|
|
|
|||
|
|
@ -135,7 +135,13 @@ public class RowIndexEntryTest extends CQLTester
|
|||
doubleSerializer.build(null, partitionKey(42L),
|
||||
Arrays.asList(cn(42), cn(43), cn(44)),
|
||||
0L);
|
||||
assertEquals(doubleSerializer.rieOldSerialized, doubleSerializer.rieNewSerialized);
|
||||
|
||||
RowIndexEntry newRie = doubleSerializer.rieSerializer.deserialize(new DataInputBuffer(doubleSerializer.rieNewSerialized, false), 0);
|
||||
Pre_C_11206_RowIndexEntry oldRie = doubleSerializer.oldSerializer.deserialize(new DataInputBuffer(doubleSerializer.rieOldSerialized, false));
|
||||
|
||||
assertEquals(oldRie.position, newRie.getPosition());
|
||||
assertEquals(oldRie.deletionTime(), newRie.deletionTime());
|
||||
assertEquals(oldRie.columnsIndex().size(), newRie.blockCount());
|
||||
}
|
||||
|
||||
// partition with multiple IndexInfo
|
||||
|
|
@ -144,7 +150,13 @@ public class RowIndexEntryTest extends CQLTester
|
|||
doubleSerializer.build(null, partitionKey(42L),
|
||||
Arrays.asList(cn(42), cn(43), cn(44), cn(45), cn(46), cn(47), cn(48), cn(49), cn(50), cn(51)),
|
||||
0L);
|
||||
assertEquals(doubleSerializer.rieOldSerialized, doubleSerializer.rieNewSerialized);
|
||||
|
||||
RowIndexEntry newRie = doubleSerializer.rieSerializer.deserialize(new DataInputBuffer(doubleSerializer.rieNewSerialized, false), 0);
|
||||
Pre_C_11206_RowIndexEntry oldRie = doubleSerializer.oldSerializer.deserialize(new DataInputBuffer(doubleSerializer.rieOldSerialized, false));
|
||||
|
||||
assertEquals(oldRie.position, newRie.getPosition());
|
||||
assertEquals(oldRie.deletionTime(), newRie.deletionTime());
|
||||
assertEquals(oldRie.columnsIndex().size(), newRie.blockCount());
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,97 @@
|
|||
/*
|
||||
* 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.transport;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.DeletionTime;
|
||||
import org.apache.cassandra.io.sstable.format.big.BigFormat;
|
||||
import org.apache.cassandra.io.util.DataInputBuffer;
|
||||
import org.apache.cassandra.io.util.DataOutputBuffer;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
/**
|
||||
* We test here serialization of DTs and some of it's sentinel values
|
||||
*
|
||||
*/
|
||||
public class DeletionTimeDeSerTest
|
||||
{
|
||||
@BeforeClass
|
||||
public static void setupDD()
|
||||
{
|
||||
DatabaseDescriptor.daemonInitialization();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLDTDeserLive() throws IOException
|
||||
{
|
||||
assertEquals(DeletionTime.LIVE, serDeser(DeletionTime.LIVE));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLDTDeserLongMAX_VALUE() throws IOException
|
||||
{
|
||||
DeletionTime dt = DeletionTime.build(Long.MAX_VALUE, 2147483600);
|
||||
assertEquals(dt, serDeser(dt));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testLDTDeser() throws IOException
|
||||
{
|
||||
// Test Serialization in all bit possible positions
|
||||
for (int i = 0; i < 63; i++)
|
||||
{
|
||||
DeletionTime dt = DeletionTime.build(1L << i, 2147483600);
|
||||
assertEquals(dt, serDeser(dt));
|
||||
}
|
||||
}
|
||||
|
||||
private DeletionTime serDeser(DeletionTime dt) throws IOException
|
||||
{
|
||||
DeletionTime readDt = null;
|
||||
int offset = 2;
|
||||
|
||||
try(DataOutputBuffer out = new DataOutputBuffer(12)) // Long + Int = 8 + 4
|
||||
{
|
||||
DeletionTime.getSerializer(BigFormat.getInstance().getLatestVersion()).serialize(dt, out);
|
||||
|
||||
try(DataInputBuffer in = new DataInputBuffer(out.toByteArray()))
|
||||
{
|
||||
// Test *both* deserializers
|
||||
|
||||
readDt = DeletionTime.getSerializer(BigFormat.getInstance().getLatestVersion()).deserialize(in);
|
||||
|
||||
ByteBuffer bbOrig = out.buffer();
|
||||
bbOrig.rewind();
|
||||
ByteBuffer bb = ByteBuffer.allocate(bbOrig.capacity() + offset);
|
||||
bb.position(offset);
|
||||
bb.put(bbOrig);
|
||||
DeletionTime readDt2 = DeletionTime.getSerializer(BigFormat.getInstance().getLatestVersion()).deserialize(bb, offset);
|
||||
assertEquals(readDt, readDt2);
|
||||
}
|
||||
}
|
||||
|
||||
return readDt;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue