natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-4.1' into trunk 

* cassandra-4.1:
  Track the amount of read data per row
This commit is contained in:
Jacek Lewandowski 2023-06-12 11:26:03 +02:00
parent dcb763ec2c b0cf9ba219
commit ba3ad7487a
5 changed files with 195 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGES.txt
View File

@ -148,7 +148,8 @@ Merged from 4.0:
4.1.3
Merged from 4.0:
Merged from 4.0:
* Track the amount of read data per row (CASSANDRA-18513)
* Fix Down nodes counter in nodetool describecluster (CASSANDRA-18512)
* Remove unnecessary shuffling of GossipDigests in Gossiper#makeRandomGossipDigest (CASSANDRA-18546)
Merged from 3.11:

1
src/java/org/apache/cassandra/db/TypeSizes.java
View File

@ -29,6 +29,7 @@ public final class TypeSizes
public static final int BOOL_SIZE = 1;
public static final int BYTE_SIZE = 1;
public static final int CHAR_SIZE = 2;
public static final int SHORT_SIZE = 2;
public static final int INT_SIZE = 4;
public static final int LONG_SIZE = 8;

9
src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
View File

@ -27,6 +27,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.io.util.FileDataInput;
import org.apache.cassandra.io.util.TrackedDataInputPlus;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.utils.SearchIterator;
@ -579,8 +580,9 @@ public class UnfilteredSerializer
if (header.isForSSTable())
{
in.readUnsignedVInt(); // Skip row size
long rowSize = in.readUnsignedVInt();
in.readUnsignedVInt(); // previous unfiltered size
in = new TrackedDataInputPlus(in, rowSize);
}
LivenessInfo rowLiveness = LivenessInfo.EMPTY;
@ -606,13 +608,14 @@ public class UnfilteredSerializer
try
{
DataInputPlus finalIn = in;
columns.apply(column -> {
try
{
if (column.isSimple())
readSimpleColumn(column, in, header, helper, builder, livenessInfo);
readSimpleColumn(column, finalIn, header, helper, builder, livenessInfo);
else
readComplexColumn(column, in, header, helper, hasComplexDeletion, builder, livenessInfo);
readComplexColumn(column, finalIn, header, helper, hasComplexDeletion, builder, livenessInfo);
}
catch (IOException e)
{

62
src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java
View File

@ -19,19 +19,37 @@ package org.apache.cassandra.io.util;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.EOFException;
import java.io.IOException;
import net.nicoulaj.compilecommand.annotations.Inline;
import org.apache.cassandra.db.TypeSizes;
/**
* This class is to track bytes read from given DataInput
*/
public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker
{
private long bytesRead;
private final long limit;
final DataInput source;
/**
* Create a TrackedDataInputPlus from given DataInput with no limit of bytes to read
*/
public TrackedDataInputPlus(DataInput source)
{
this(source, -1);
}
/**
* Create a TrackedDataInputPlus from given DataInput with limit of bytes to read. If limit is reached
* {@link IOException} will be thrown when trying to read more bytes.
*/
public TrackedDataInputPlus(DataInput source, long limit)
{
this.source = source;
this.limit = limit;
}
public long getBytesRead()
@ -49,55 +67,63 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker
public boolean readBoolean() throws IOException
{
checkCanRead(TypeSizes.BOOL_SIZE);
boolean bool = source.readBoolean();
bytesRead += 1;
bytesRead += TypeSizes.BOOL_SIZE;
return bool;
}
public byte readByte() throws IOException
{
checkCanRead(TypeSizes.BYTE_SIZE);
byte b = source.readByte();
bytesRead += 1;
bytesRead += TypeSizes.BYTE_SIZE;
return b;
}
public char readChar() throws IOException
{
checkCanRead(TypeSizes.CHAR_SIZE);
char c = source.readChar();
bytesRead += 2;
bytesRead += TypeSizes.CHAR_SIZE;
return c;
}
public double readDouble() throws IOException
{
checkCanRead(TypeSizes.DOUBLE_SIZE);
double d = source.readDouble();
bytesRead += 8;
bytesRead += TypeSizes.DOUBLE_SIZE;
return d;
}
public float readFloat() throws IOException
{
checkCanRead(TypeSizes.FLOAT_SIZE);
float f = source.readFloat();
bytesRead += 4;
bytesRead += TypeSizes.FLOAT_SIZE;
return f;
}
public void readFully(byte[] b, int off, int len) throws IOException
{
checkCanRead(len);
source.readFully(b, off, len);
bytesRead += len;
}
public void readFully(byte[] b) throws IOException
{
checkCanRead(b.length);
source.readFully(b);
bytesRead += b.length;
}
public int readInt() throws IOException
{
checkCanRead(TypeSizes.INT_SIZE);
int i = source.readInt();
bytesRead += 4;
bytesRead += TypeSizes.INT_SIZE;
return i;
}
@ -110,15 +136,17 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker
public long readLong() throws IOException
{
checkCanRead(TypeSizes.LONG_SIZE);
long l = source.readLong();
bytesRead += 8;
bytesRead += TypeSizes.LONG_SIZE;
return l;
}
public short readShort() throws IOException
{
checkCanRead(TypeSizes.SHORT_SIZE);
short s = source.readShort();
bytesRead += 2;
bytesRead += TypeSizes.SHORT_SIZE;
return s;
}
@ -129,22 +157,34 @@ public class TrackedDataInputPlus implements DataInputPlus, BytesReadTracker
public int readUnsignedByte() throws IOException
{
checkCanRead(TypeSizes.BYTE_SIZE);
int i = source.readUnsignedByte();
bytesRead += 1;
bytesRead += TypeSizes.BYTE_SIZE;
return i;
}
public int readUnsignedShort() throws IOException
{
checkCanRead(TypeSizes.SHORT_SIZE);
int i = source.readUnsignedShort();
bytesRead += 2;
bytesRead += TypeSizes.SHORT_SIZE;
return i;
}
public int skipBytes(int n) throws IOException
{
int skipped = source.skipBytes(n);
int skipped = source.skipBytes(limit < 0 ? n : (int) Math.min(limit - bytesRead, n));
bytesRead += skipped;
return skipped;
}
@Inline
private void checkCanRead(int size) throws IOException
{
if (limit >= 0 && bytesRead + size > limit)
{
skipBytes((int) (limit - bytesRead));
throw new EOFException("EOF after " + (limit - bytesRead) + " bytes out of " + size);
}
}
}

135
test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java Normal file
View File

@ -0,0 +1,135 @@
/*
* 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.db.rows;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.TableMetadata;
import static org.assertj.core.api.Assertions.assertThatIOException;
import static org.junit.Assert.assertEquals;
public class UnfilteredSerializerTest
{
private static TableMetadata md;
@BeforeClass
public static void beforeClass()
{
DatabaseDescriptor.daemonInitialization();
md = TableMetadata.builder("ks", "cf")
.addPartitionKeyColumn("pk", IntegerType.instance)
.addRegularColumn("v1", BytesType.instance)
.addRegularColumn("v2", BytesType.instance)
.build();
}
@Test
public void testRowSerDe() throws IOException
{
// test serialization and deserialization of a row body
testRowBodySerDe(10, Function.identity());
}
@Test
public void testRowSerDeWithCorruption() throws IOException
{
// test serialization and deserialization of a row body when the row is corrupted in the way that the actual
// row content is larger than the row size serialized in the preamble
ByteBuffer largeRow = getSerializedRow(50);
assertThatIOException().isThrownBy(() -> testRowBodySerDe(10, buf -> replaceRowContent(buf, largeRow)))
.withMessageMatching("EOF after \\d+ bytes out of 50");
}
public static void testRowBodySerDe(int cellSize, Function<ByteBuffer, ByteBuffer> transform) throws IOException
{
Random random = new Random();
ByteBuffer data1 = ByteBuffer.allocate(cellSize);
ByteBuffer data2 = ByteBuffer.allocate(cellSize);
random.nextBytes(data1.array());
random.nextBytes(data2.array());
Row.Builder builder = BTreeRow.sortedBuilder();
builder.newRow(Clustering.EMPTY);
builder.addCell(BufferCell.live(md.regularColumns().getSimple(0), TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()), data1.duplicate()));
builder.addCell(BufferCell.live(md.regularColumns().getSimple(1), TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis()), data2.duplicate()));
Row writtenRow = builder.build();
try (DataOutputBuffer out = new DataOutputBuffer())
{
UnfilteredSerializer.serializer.serialize(writtenRow, new SerializationHelper(SerializationHeader.makeWithoutStats(md)), out, 0, MessagingService.current_version);
out.flush();
try (DataInputBuffer in = new DataInputBuffer(transform.apply(out.asNewBuffer()), false))
{
builder = BTreeRow.sortedBuilder();
DeserializationHelper helper = new DeserializationHelper(md, MessagingService.current_version, DeserializationHelper.Flag.LOCAL, ColumnFilter.all(md));
Unfiltered readRow = UnfilteredSerializer.serializer.deserialize(in, SerializationHeader.makeWithoutStats(md), helper, builder);
assertEquals(writtenRow, readRow);
}
}
}
private ByteBuffer getSerializedRow(int cellSize) throws IOException
{
AtomicReference<ByteBuffer> rowData = new AtomicReference<>();
testRowBodySerDe(cellSize, buf -> {
rowData.set(buf.duplicate());
return buf;
});
return rowData.get();
}
private static ByteBuffer replaceRowContent(ByteBuffer original, ByteBuffer replacement)
{
try (DataOutputBuffer out = new DataOutputBuffer();
DataInputBuffer origIn = new DataInputBuffer(original, true);
DataInputBuffer replIn = new DataInputBuffer(replacement, false))
{
out.writeByte(origIn.readByte()); // flag
out.writeUnsignedVInt(origIn.readUnsignedVInt()); // row size
replIn.readByte(); // skip flag
replIn.readUnsignedVInt(); // skip row size
out.write(replacement);
out.close();
return out.asNewBuffer();
}
catch (IOException ex)
{
throw new RuntimeException(ex);
}
}
}