diff --git a/CHANGES.txt b/CHANGES.txt index adea342a69..b3af28ac01 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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: diff --git a/src/java/org/apache/cassandra/db/TypeSizes.java b/src/java/org/apache/cassandra/db/TypeSizes.java index aed18a6c04..8df83b6ba2 100644 --- a/src/java/org/apache/cassandra/db/TypeSizes.java +++ b/src/java/org/apache/cassandra/db/TypeSizes.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java index 712c9cfb54..c1802cf012 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java @@ -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) { diff --git a/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java b/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java index dc5bbb6599..d319a86e4f 100644 --- a/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java +++ b/src/java/org/apache/cassandra/io/util/TrackedDataInputPlus.java @@ -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); + } + } } diff --git a/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java b/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java new file mode 100644 index 0000000000..835e613b9e --- /dev/null +++ b/test/unit/org/apache/cassandra/db/rows/UnfilteredSerializerTest.java @@ -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 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 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); + } + } +}