diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index 2be55fbe5c..09e9a85e8b 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -212,6 +212,11 @@ public enum CassandraRelevantProperties DATA_OUTPUT_STREAM_PLUS_TEMP_BUFFER_SIZE("cassandra.data_output_stream_plus_temp_buffer_size", "8192"), DATA_RESPONSE_BUFFER_INITIAL_SIZE_MAX("cassandra.data_response_buffer_initial_size_max", "4096"), DATA_RESPONSE_BUFFER_INITIAL_SIZE_MIN("cassandra.data_response_buffer_initial_size_min", "128"), + /** higher in-memory limits used for ONE/LOCAL_ONE reads, when local result is consumed directly without waiting for other replicas **/ + DATA_RESPONSE_IN_MEMORY_MAX_ROWS("cassandra.data_response_in_memory_max_rows", "128"), + DATA_RESPONSE_IN_MEMORY_MAX_ROWS_CL_ONE("cassandra.data_response_in_memory_max_rows_cl_one", "512"), + DATA_RESPONSE_IN_MEMORY_MAX_SIZE("cassandra.data_response_in_memory_max_size", "16KiB"), + DATA_RESPONSE_IN_MEMORY_MAX_SIZE_CL_ONE("cassandra.data_response_in_memory_max_size_cl_one", "64KiB"), DECAYING_ESTIMATED_HISTOGRAM_RESERVOIR_STRIPE_COUNT("cassandra.dehr_stripe_count", "2"), DEFAULT_PROVIDE_OVERLAPPING_TOMBSTONES("default.provide.overlapping.tombstones"), /** determinism properties for testing */ diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java index d848e23b6b..4607085049 100644 --- a/src/java/org/apache/cassandra/db/ReadCommand.java +++ b/src/java/org/apache/cassandra/db/ReadCommand.java @@ -443,14 +443,27 @@ public abstract class ReadCommand extends AbstractReadQuery public abstract boolean isReversed(); public ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi) + { + return createResponse(iterator, rdi, false, false); + } + + private ReadResponse createResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localRead, boolean localReplicaOnly) { // validate that the sequence of RT markers is correct: open is followed by close, deletion times for both // ends equal, and there are no dangling RT bound in any partition. iterator = RTBoundValidator.validate(iterator, Stage.PROCESSED, true); + if (isDigestQuery()) + return ReadResponse.createDigestResponse(iterator, this); - return isDigestQuery() - ? ReadResponse.createDigestResponse(iterator, this) - : ReadResponse.createDataResponse(iterator, this, rdi); + if (localRead && ReadResponse.inMemoryLocalResponseEnabled(localReplicaOnly)) + return ReadResponse.createInMemoryDataResponse(iterator, this, rdi, localReplicaOnly); + + return ReadResponse.createDataResponse(iterator, this, rdi); + } + + public ReadResponse createLocalObjectResponse(UnfilteredPartitionIterator iterator, RepairedDataInfo rdi, boolean localReplicaOnly) + { + return createResponse(iterator, rdi, true, localReplicaOnly); } public ReadResponse createEmptyResponse() diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index 7edd743ad4..b35917412c 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -21,21 +21,30 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; +import java.util.NoSuchElementException; import com.google.common.annotations.VisibleForTesting; import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.db.filter.ColumnFilter; +import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator; +import org.apache.cassandra.db.partitions.ImmutableBTreePartition; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.db.rows.AbstractUnfilteredRowIterator; import org.apache.cassandra.db.rows.DeserializationHelper; +import org.apache.cassandra.db.rows.RangeTombstoneMarker; +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.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.rows.UnfilteredRowIterators; import org.apache.cassandra.io.IVersionedSerializer; 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.DataOutputPlus; +import org.apache.cassandra.metrics.ReadResponseMetrics; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; @@ -49,10 +58,43 @@ public abstract class ReadResponse // Serializer for single partition read response public static final IVersionedSerializer serializer = new Serializer(); + // Per-request limits for the in-memory portion of a local single-partition read response. Unfiltered objects are + // kept as an in-memory partition until either enabled limit is reached; the remainder is serialized into an + // overflow buffer. Each limit is disabled independently by setting it to <= 0. + // - IN_MEMORY_MAX_ROWS: maximum number of in-memory Unfiltered objects (rows and range tombstone markers). + // - IN_MEMORY_MAX_SIZE: maximum accumulated unshared heap size (in bytes) of the in-memory Unfiltered objects. + // ONE/LOCAL_ONE reads use separate higher limits: the local read is consumed directly + // without waiting for cross-replica interactions, so it is worth keeping more of it in memory. + static final int IN_MEMORY_MAX_ROWS = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_ROWS.getInt(); + static final long IN_MEMORY_MAX_SIZE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_SIZE.getSizeInBytes(); + static final int IN_MEMORY_MAX_ROWS_CL_ONE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_ROWS_CL_ONE.getInt(); + static final long IN_MEMORY_MAX_SIZE_CL_ONE = CassandraRelevantProperties.DATA_RESPONSE_IN_MEMORY_MAX_SIZE_CL_ONE.getSizeInBytes(); + protected ReadResponse() { } + private static int maxRows(boolean localReplicaOnly) + { + return localReplicaOnly ? IN_MEMORY_MAX_ROWS_CL_ONE : IN_MEMORY_MAX_ROWS; + } + + private static long maxSize(boolean localReplicaOnly) + { + return localReplicaOnly ? IN_MEMORY_MAX_SIZE_CL_ONE : IN_MEMORY_MAX_SIZE; + } + + static boolean inMemoryLocalResponseEnabled(boolean localReplicaOnly) + { + return maxRows(localReplicaOnly) > 0 || maxSize(localReplicaOnly) > 0; + } + + @VisibleForTesting + int inMemoryUnfilteredCount() + { + return 0; + } + public static ReadResponse createDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi) { return new LocalDataResponse(data, command, rdi); @@ -63,9 +105,15 @@ public abstract class ReadResponse return new LocalDataResponse(data, command, NO_OP_REPAIRED_DATA_INFO); } - public static ReadResponse createSimpleDataResponse(UnfilteredPartitionIterator data, ColumnFilter selection) + static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, boolean oneOrLocalOne) { - return new LocalDataResponse(data, selection); + return createInMemoryDataResponse(data, command, rdi, maxRows(oneOrLocalOne), maxSize(oneOrLocalOne)); + } + + @VisibleForTesting + static ReadResponse createInMemoryDataResponse(UnfilteredPartitionIterator data, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize) + { + return InMemoryDataResponse.build(data, command, rdi, maxRows, maxHeapSize); } @VisibleForTesting @@ -240,9 +288,11 @@ public abstract class ReadResponse DeserializationHelper.Flag.LOCAL); } - private LocalDataResponse(UnfilteredPartitionIterator iter, ColumnFilter selection) + // Wraps an already-serialized response buffer; used when a local in-memory response outgrows its limits and + // is serialized fully (see InMemoryDataResponse.build). + private LocalDataResponse(ByteBuffer data, ByteBuffer repairedDataDigest, boolean isRepairedDigestConclusive) { - super(build(iter, selection), null, false, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + super(data, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); } private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection) @@ -354,6 +404,214 @@ public abstract class ReadResponse { return false; } + + protected ByteBuffer getSerializedData() + { + return data; + } + } + + // built on the coordinator for local single-partition reads; never sent over the network. + // Holds the whole response as an in-memory object graph. Only used when the response fits within the per-request + // limits; larger responses are serialized in full and returned as an ordinary LocalDataResponse instead + private static class InMemoryDataResponse extends DataResponse + { + // the whole response as an in-memory partition; null if the response was empty + private final ImmutableBTreePartition partition; + + static ReadResponse build(UnfilteredPartitionIterator iter, ReadCommand command, RepairedDataInfo rdi, int maxRows, long maxHeapSize) + { + if (!iter.hasNext()) + { + // Empty response + return new InMemoryDataResponse(null, rdi.getDigest(), rdi.isConclusive()); + } + + ImmutableBTreePartition partition; + ByteBuffer serialized; + try (UnfilteredRowIterator rowIter = iter.next()) + { + LimitedUnfilteredRowIterator limitedIter = new LimitedUnfilteredRowIterator(rowIter, maxRows, maxHeapSize); + partition = ImmutableBTreePartition.create(limitedIter); + + if (limitedIter.hasOverflow()) + { + // if a per-request limit is crossed then fall back to the ordinary serialized representation. + (limitedIter.overflowedByRowLimit() ? ReadResponseMetrics.inMemoryRowLimitHits + : ReadResponseMetrics.inMemorySizeLimitHits).inc(); + serialized = serialize(command, partition, rowIter); + } + else + { + serialized = null; + } + } + + // Capture digest after consuming and closing the iterator so any RepairedDataInfo transformations are reflected. + return serialized != null + ? new LocalDataResponse(serialized, rdi.getDigest(), rdi.isConclusive()) + : new InMemoryDataResponse(partition, rdi.getDigest(), rdi.isConclusive()); + } + + // Serializes the full response (the already-consumed in-memory prefix followed by the remaining rows) into a + // buffer in the intra-node partition format, so it can be read back like any other serialized DataResponse. + private static ByteBuffer serialize(ReadCommand command, ImmutableBTreePartition prefix, UnfilteredRowIterator suffix) + { + UnfilteredRowIterator prefixIter = prefix.unfilteredIterator(command.columnFilter(), Slices.ALL, suffix.isReverseOrder()); + UnfilteredRowIterator combined = UnfilteredRowIterators.concat(prefixIter, suffix); + UnfilteredPartitionIterator partitionIter = new SingletonUnfilteredPartitionIterator(command.metadata(), combined); + return LocalDataResponse.build(partitionIter, command.columnFilter()); + } + + private InMemoryDataResponse(ImmutableBTreePartition partition, + ByteBuffer repairedDataDigest, boolean isRepairedDigestConclusive) + { + super(null, repairedDataDigest, isRepairedDigestConclusive, MessagingService.current_version, DeserializationHelper.Flag.LOCAL); + this.partition = partition; + } + + @Override + @VisibleForTesting + int inMemoryUnfilteredCount() + { + if (partition == null) + return 0; + int count = 0; + try (UnfilteredRowIterator iter = partition.unfilteredIterator()) + { + while (iter.hasNext()) { iter.next(); count++; } + } + return count; + } + + // Decorating iterator that stops once a per-request limit is reached and records whether overflow occurred and which limit caused it. + // Does NOT close the wrapped iterator on close() so the caller can continue reading overflow rows. + private static class LimitedUnfilteredRowIterator extends AbstractUnfilteredRowIterator + { + private final UnfilteredRowIterator wrapped; + private final int maxRows; + private final long maxHeapSize; + + private int unfilteredCount = 0; + private long accumulatedHeapSize = 0; + private boolean hasOverflow = false; + private boolean overflowByRowLimit = false; + private boolean insideOpenMarker = false; + + private LimitedUnfilteredRowIterator(UnfilteredRowIterator wrapped, int maxRows, long maxHeapSize) + { + super(wrapped.metadata(), + wrapped.partitionKey(), + wrapped.partitionLevelDeletion(), + wrapped.columns(), + wrapped.staticRow(), + wrapped.isReverseOrder(), + wrapped.stats()); + this.wrapped = wrapped; + this.maxRows = maxRows; + this.maxHeapSize = maxHeapSize; + } + + boolean hasOverflow() + { + return hasOverflow; + } + + boolean overflowedByRowLimit() + { + return overflowByRowLimit; + } + + @Override + protected Unfiltered computeNext() + { + if (!wrapped.hasNext()) + return endOfData(); + + if (!insideOpenMarker) + { + if (maxRows > 0 && unfilteredCount >= maxRows) + return overflow(true); + if (maxHeapSize > 0 && accumulatedHeapSize >= maxHeapSize) + return overflow(false); + } + + Unfiltered next = wrapped.next(); + if (next.isRangeTombstoneMarker()) + { + RangeTombstoneMarker marker = (RangeTombstoneMarker) next; + insideOpenMarker = marker.isOpen(isReverseOrder); + } + unfilteredCount++; + // Compute the heap size only when the size limit is enabled to avoid overheads if it is not needed + if (maxHeapSize > 0) + accumulatedHeapSize += unsharedHeapSize(next); + return next; + } + + private Unfiltered overflow(boolean byRowLimit) + { + hasOverflow = true; + overflowByRowLimit = byRowLimit; + return endOfData(); + } + + private static long unsharedHeapSize(Unfiltered unfiltered) + { + return unfiltered.isRow() + ? ((Row) unfiltered).unsharedHeapSize() + : ((RangeTombstoneMarker) unfiltered).unsharedHeapSize(); + } + } + + @Override + public UnfilteredPartitionIterator makeIterator(ReadCommand command) + { + if (partition == null) + return EmptyIterators.unfilteredPartition(command.metadata()); + + UnfilteredRowIterator inMemoryIter = partition.unfilteredIterator(command.columnFilter(), Slices.ALL, command.isReversed()); + return new SingletonUnfilteredPartitionIterator(command.metadata(), inMemoryIter); + } + + @Override + protected ByteBuffer getSerializedData() + { + throw new UnsupportedOperationException("InMemoryDataResponse cannot be serialized over the network"); + } + } + + private static class SingletonUnfilteredPartitionIterator extends AbstractUnfilteredPartitionIterator + { + private final TableMetadata metadata; + private final UnfilteredRowIterator partition; + private boolean returned = false; + + private SingletonUnfilteredPartitionIterator(TableMetadata metadata, UnfilteredRowIterator partition) + { + this.metadata = metadata; + this.partition = partition; + } + + public TableMetadata metadata() { return metadata; } + + public boolean hasNext() + { + return !returned; + } + + public UnfilteredRowIterator next() + { + if (returned) throw new NoSuchElementException(); + returned = true; + return partition; + } + + @Override + public void close() + { + partition.close(); + } } private static class Serializer implements IVersionedSerializer @@ -378,7 +636,7 @@ public abstract class ReadResponse ByteBufferUtil.writeWithVIntLength(response.repairedDataDigest(), out); out.writeBoolean(response.isRepairedDigestConclusive()); - ByteBuffer data = ((DataResponse)response).data; + ByteBuffer data = ((DataResponse)response).getSerializedData(); ByteBufferUtil.writeWithVIntLength(data, out); } } @@ -418,7 +676,7 @@ public abstract class ReadResponse // In theory, we should deserialize/re-serialize if the version asked is different from the current // version as the content could have a different serialization format. So far though, we haven't made // change to partition iterators serialization since 3.0 so we skip this. - ByteBuffer data = ((DataResponse)response).data; + ByteBuffer data = ((DataResponse)response).getSerializedData(); size += ByteBufferUtil.serializedSizeWithVIntLength(data); } return size; diff --git a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java index 7c69c77d88..312f449f13 100644 --- a/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java +++ b/src/java/org/apache/cassandra/metrics/CassandraMetricsRegistry.java @@ -146,6 +146,7 @@ public class CassandraMetricsRegistry extends MetricRegistry .add(MutualTlsMetrics.TYPE_NAME) .add(PaxosMetrics.TYPE_NAME) .add(ReadRepairMetrics.TYPE_NAME) + .add(ReadResponseMetrics.TYPE_NAME) .add(RepairMetrics.TYPE_NAME) .add(RowIndexEntry.TYPE_NAME) .add(StorageMetrics.TYPE_NAME) diff --git a/src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java b/src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java new file mode 100644 index 0000000000..a138bd5871 --- /dev/null +++ b/src/java/org/apache/cassandra/metrics/ReadResponseMetrics.java @@ -0,0 +1,42 @@ +/* + * 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.metrics; + +import com.codahale.metrics.Counter; + +import static org.apache.cassandra.metrics.CassandraMetricsRegistry.Metrics; + +public class ReadResponseMetrics +{ + public static final String TYPE_NAME = "ReadResponse"; + private static final MetricNameFactory factory = new DefaultNameFactory(TYPE_NAME); + + /** + * Incremented when a local single-partition read response hit the per-request in-memory row-count limit + */ + public static final Counter inMemoryRowLimitHits = Metrics.counter(factory.createMetricName("InMemoryRowLimitHits")); + + /** + * Incremented when a local single-partition read response hit the per-request in-memory heap-size limit. + * The row-count limit is checked first, so this + * counts only responses that passed the row-count limit (or have it disabled) but crossed the size limit. + */ + public static final Counter inMemorySizeLimitHits = Metrics.counter(factory.createMetricName("InMemorySizeLimitHits")); + + public static void init() {} +} diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index d10f8bb36b..f049446b69 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -127,6 +127,7 @@ import org.apache.cassandra.metrics.CASClientRequestMetrics; import org.apache.cassandra.metrics.ClientRequestSizeMetrics; import org.apache.cassandra.metrics.DenylistMetrics; import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.metrics.ReadResponseMetrics; import org.apache.cassandra.metrics.StorageMetrics; import org.apache.cassandra.net.ArtificialLatency; import org.apache.cassandra.net.ForwardingInfo; @@ -311,6 +312,7 @@ public class StorageProxy implements StorageProxyMBean ReadRepairMetrics.init(); + ReadResponseMetrics.init(); if (!Paxos.isLinearizable()) { @@ -2753,7 +2755,14 @@ public class StorageProxy implements StorageProxyMBean try (ReadExecutionController controller = command.executionController(trackRepairedStatus); UnfilteredPartitionIterator iterator = command.executeLocally(controller)) { - response = command.createResponse(iterator, controller.getRepairedDataInfo()); + if (command.isLimitedToOnePartition() && !command.isDigestQuery()) + { + ConsistencyLevel cl = handler.consistencyLevel(); + boolean localReplicaOnly = cl == ConsistencyLevel.ONE || cl == ConsistencyLevel.LOCAL_ONE; + response = command.createLocalObjectResponse(iterator, controller.getRepairedDataInfo(), localReplicaOnly); + } + else + response = command.createResponse(iterator, controller.getRepairedDataInfo()); } catch (RejectException e) { diff --git a/src/java/org/apache/cassandra/service/reads/ReadCallback.java b/src/java/org/apache/cassandra/service/reads/ReadCallback.java index 7f68b3e4d5..10b19d23a3 100644 --- a/src/java/org/apache/cassandra/service/reads/ReadCallback.java +++ b/src/java/org/apache/cassandra/service/reads/ReadCallback.java @@ -30,6 +30,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ConsistencyLevel; import org.apache.cassandra.db.MessageParams; import org.apache.cassandra.db.PartitionRangeReadCommand; import org.apache.cassandra.db.ReadCommand; @@ -101,6 +102,11 @@ public class ReadCallback, P extends ReplicaPlan.ForRead< return replicaPlan.get(); } + public ConsistencyLevel consistencyLevel() + { + return replicaPlan().consistencyLevel(); + } + public boolean await(long commandTimeout, TimeUnit unit) { return awaitUntil(requestTime.computeDeadline(unit.toNanos(commandTimeout))); diff --git a/test/unit/org/apache/cassandra/db/ReadResponseTest.java b/test/unit/org/apache/cassandra/db/ReadResponseTest.java index ebe1cf0322..e148d8a49b 100644 --- a/test/unit/org/apache/cassandra/db/ReadResponseTest.java +++ b/test/unit/org/apache/cassandra/db/ReadResponseTest.java @@ -20,6 +20,8 @@ package org.apache.cassandra.db; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; import java.util.Random; import org.junit.Before; @@ -27,16 +29,28 @@ import org.junit.BeforeClass; import org.junit.Test; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter; import org.apache.cassandra.db.filter.ColumnFilter; import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.filter.RowFilter; import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator; +import org.apache.cassandra.db.partitions.PartitionUpdate; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; +import org.apache.cassandra.db.rows.BTreeRow; +import org.apache.cassandra.db.rows.BufferCell; +import org.apache.cassandra.db.rows.RangeTombstoneMarker; +import org.apache.cassandra.db.rows.Row; +import org.apache.cassandra.db.rows.Unfiltered; +import org.apache.cassandra.db.rows.UnfilteredRowIterator; +import org.apache.cassandra.db.rows.WrappingUnfilteredRowIterator; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper; import org.apache.cassandra.io.util.DataInputBuffer; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.metrics.ReadResponseMetrics; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; @@ -50,6 +64,7 @@ public class ReadResponseTest { private final Random random = new Random(); private TableMetadata metadata; + private TableMetadata metadataWithClustering; @BeforeClass public static void beforeClass() @@ -60,13 +75,20 @@ public class ReadResponseTest @Before public void setup() { - metadata = TableMetadata.builder("ks", "t1") .offline() .addPartitionKeyColumn("p", Int32Type.instance) .addRegularColumn("v", Int32Type.instance) .partitioner(Murmur3Partitioner.instance) .build(); + + metadataWithClustering = TableMetadata.builder("ks", "t2") + .offline() + .addPartitionKeyColumn("p", Int32Type.instance) + .addClusteringColumn("c", Int32Type.instance) + .addRegularColumn("v", Int32Type.instance) + .partitioner(Murmur3Partitioner.instance) + .build(); } @Test @@ -172,6 +194,505 @@ public class ReadResponseTest assertEquals(response1.digest(command1), response2.digest(command2)); } + @Test + public void inMemoryResponseEmptyIteratorMatchesLocalDataResponse() + { + ReadCommand command = command(key(), metadata); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + + ReadResponse localResponse = command.createResponse(EmptyIterators.unfilteredPartition(metadata), rdi); + ReadResponse inMemoryResponse = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false); + + assertIteratorsEqual(command, localResponse, inMemoryResponse); + } + + @Test + public void inMemoryResponseWithRowsMatchesLocalDataResponse() + { + int key = key(); + ReadCommand command = command(key, metadata); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + + DecoratedKey dk = metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key)); + Row row = buildRow(metadata, dk); + PartitionUpdate update = PartitionUpdate.singleRowUpdate(metadata, dk, row); + + ReadResponse localResponse = command.createResponse(singlePartitionIterator(update), rdi); + ReadResponse inMemoryResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, false); + + assertIteratorsEqual(command, localResponse, inMemoryResponse); + } + + @Test(expected = UnsupportedOperationException.class) + public void inMemoryResponseCannotBeSerialized() + { + ReadCommand command = command(key(), metadata); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + ReadResponse response = command.createLocalObjectResponse(EmptyIterators.unfilteredPartition(metadata), rdi, false); + + try (DataOutputBuffer out = new DataOutputBuffer()) + { + ReadResponse.serializer.serialize(response, out, MessagingService.current_version); + } + catch (IOException e) + { + fail("Unexpected IOException: " + e.getMessage()); + } + } + + @Test + public void inMemoryResponseUsesHigherLimitsForOneConsistency() + { + int key = key(); + ReadCommand command = command(key, metadataWithClustering); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + // More rows than the default row limit (128) but fewer than the ONE/LOCAL_ONE limit (512). + int rowCount = 150; + PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rowCount); + + ReadResponse otherResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, false); + ReadResponse oneResponse = command.createLocalObjectResponse(singlePartitionIterator(update), rdi, true); + + // The default-tier limits overflow this partition; ONE/LOCAL_ONE keeps more of it in memory. + assertTrue("default tier should overflow", otherResponse.inMemoryUnfilteredCount() < rowCount); + assertTrue("ONE/LOCAL_ONE tier should keep more rows in memory", + oneResponse.inMemoryUnfilteredCount() > otherResponse.inMemoryUnfilteredCount()); + } + + @Test + public void inMemoryResponseCountsRowLimitHitOnOverflow() + { + // Row limit 1, size limit disabled: the rows beyond the first overflow, hitting the row-count limit. + limitHitAssertion() + .rows(5).maxRows(1) + .expectRowLimitHit() + .verify(); + } + + @Test + public void inMemoryResponseCountsSizeLimitHitOnOverflow() + { + // Row limit disabled, tiny size limit: the rows beyond the first overflow, hitting the size limit. + limitHitAssertion() + .rows(5).maxSize(1) + .expectSizeLimitHit() + .verify(); + } + + @Test + public void inMemoryResponseCountsRowLimitFirstWhenBothLimitsCrossed() + { + // Both limits would be crossed; the row-count limit is checked first, so only it is counted. + limitHitAssertion() + .rows(5).maxRows(1).maxSize(1) + .expectRowLimitHit() + .verify(); + } + + @Test + public void inMemoryResponseDoesNotCountLimitHitWhenAllInMemory() + { + // Limits above the response size: nothing overflows. + limitHitAssertion() + .rows(3).maxRows(100).maxSize(Long.MAX_VALUE) + .expectNoLimitHit() + .verify(); + } + + private LimitHitAssertionBuilder limitHitAssertion() + { + return new LimitHitAssertionBuilder(); + } + + /** + * Builds an in-memory response for a partition of {@link #rows} rows with the configured per-request limits + * ({@code 0} disables a limit) and asserts which limit-hit counter(s) were incremented (each by one). + */ + private class LimitHitAssertionBuilder + { + private int rows = 0; + private int maxRows = 0; // 0 = row-count limit disabled + private long maxSize = 0; // 0 = heap-size limit disabled + private boolean expectRowHit = false; + private boolean expectSizeHit = false; + + LimitHitAssertionBuilder rows(int rows) { this.rows = rows; return this; } + LimitHitAssertionBuilder maxRows(int maxRows) { this.maxRows = maxRows; return this; } + LimitHitAssertionBuilder maxSize(long maxSize) { this.maxSize = maxSize; return this; } + LimitHitAssertionBuilder expectRowLimitHit() { this.expectRowHit = true; return this; } + LimitHitAssertionBuilder expectSizeLimitHit() { this.expectSizeHit = true; return this; } + LimitHitAssertionBuilder expectNoLimitHit() { this.expectRowHit = false; this.expectSizeHit = false; return this; } + + void verify() + { + int key = key(); + ReadCommand command = command(key, metadataWithClustering); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, rows); + + long rowHitsBefore = ReadResponseMetrics.inMemoryRowLimitHits.getCount(); + long sizeHitsBefore = ReadResponseMetrics.inMemorySizeLimitHits.getCount(); + ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update), command, rdi, maxRows, maxSize); + assertEquals("unexpected row limit hit count", rowHitsBefore + (expectRowHit ? 1 : 0), ReadResponseMetrics.inMemoryRowLimitHits.getCount()); + assertEquals("unexpected size limit hit count", sizeHitsBefore + (expectSizeHit ? 1 : 0), ReadResponseMetrics.inMemorySizeLimitHits.getCount()); + } + } + + @Test + public void inMemoryResponseWithOverflowMatchesLocalDataResponse() + { + // Row limit crossed: the whole response is serialized. + inMemoryAssertion() + .maxRows(2).rows(5) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseAllRowsInMemoryWhenUnderLimit() + { + inMemoryAssertion() + .maxRows(10).rows(3) + .expectInMemory() + .verify(); + } + + @Test + public void inMemoryResponseByteLimitWithOverflow() + { + // Row limit disabled; the byte limit is crossed, so the whole response is serialized. + inMemoryAssertion() + .maxBytesAsSizeOfFirstNRows(2).rows(5) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseByteLimitAllRowsInMemoryWhenUnderLimit() + { + // Byte limit sized above the whole response: it stays in memory. + inMemoryAssertion() + .maxBytesAsSizeOfFirstNRows(10).rows(3) + .expectInMemory() + .verify(); + } + + @Test + public void inMemoryResponseByteLimitReachedBeforeRowLimit() + { + // Row limit (10) is not reached; the byte limit (sized for 2 rows) is crossed first, so the response is serialized. + inMemoryAssertion() + .maxRows(10).maxBytesAsSizeOfFirstNRows(2).rows(5) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseRowLimitReachedBeforeByteLimit() + { + // Byte limit (sized for all 5 rows) is not reached; the row limit (2) is crossed first, so the response is serialized. + inMemoryAssertion() + .maxRows(2).maxBytesAsSizeOfFirstNRows(5).rows(5) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseKeepsEverythingWhenBothLimitsDisabled() + { + // Both limits disabled: the whole partition is kept in memory. + inMemoryAssertion() + .maxRows(0).rows(4) + .expectInMemory() + .verify(); + } + + @Test + public void inMemoryResponseCapturesRepairedDigestAfterIteratorIsConsumed() + { + // RepairedDataInfo updates its digest lazily as rows are consumed via withRepairedDataInfo + // transformations; this test uses a stub that simulates the same timing. + int key = key(); + ReadCommand command = command(key, metadataWithClustering); + PartitionUpdate update = buildMultiRowUpdate(metadataWithClustering, key, 3); + + ByteBuffer expectedDigest = digest(); + // Returns EMPTY before any iteration; returns expectedDigest once the iterator has been consumed. + LazyRepairedDataInfo rdi = new LazyRepairedDataInfo(expectedDigest); + + // large heap budget (row limit disabled) so all rows are kept in memory; this test only cares about digest capture timing + ReadResponse response = ReadResponse.createInMemoryDataResponse(lazyRdiWrappedIterator(update, rdi), command, rdi, 0, Long.MAX_VALUE); + + assertTrue("digest should be non-empty after iterator was consumed", rdi.wasConsumedBeforeDigestCaptured()); + assertEquals(expectedDigest, response.repairedDataDigest()); + assertTrue(response.isRepairedDigestConclusive()); + } + + @Test + public void inMemoryResponseWithRangeTombstoneAllInMemory() + { + // A single range tombstone (open+close markers) well under the limit: kept in memory. + inMemoryAssertion() + .maxRows(10).rows(0) + .withTombstone(rangeTombstone(0, 5)) + .expectInMemory() + .verify(); + } + + @Test + public void inMemoryResponseWithRangeTombstoneOverflow() + { + // Rows plus a trailing range tombstone cross the row limit: the whole response is serialized. + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(6, 9)) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseWithTombstonesOnlyOverflow() + { + // Two range tombstones (open+close markers each) cross the row limit: the whole response is serialized. + inMemoryAssertion() + .maxRows(2).rows(0) + .withTombstone(rangeTombstone(0, 3)) + .withTombstone(rangeTombstone(5, 8)) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseWithRangeTombstoneAmongRowsOverflow() + { + // A range tombstone at [1, 2) interleaved with rows 0-4, crossing the row limit: serialized in full. + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(1, 2)) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseWithRangeTombstoneAmongRowsOverflowReversed() + { + // Same as above with reversed read order; the range tombstone is at [2, 3). + inMemoryAssertion() + .maxRows(2).rows(5).reversed() + .withTombstone(rangeTombstone(2, 3)) + .expectSerialized() + .verify(); + } + + @Test + public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestamps() + { + // Three overlapping RTs: [0,3) ts=100, [1,4) ts=200, [2,5) ts=50 form a single continuous open range + // (bound, boundary, boundary, bound markers). Emission cannot stop while inside the open marker, so even + // with a row limit of 2 the whole response is kept in memory. + inMemoryAssertion() + .maxRows(2).rows(5) + .withTombstone(rangeTombstone(0, 3).timestamp(100)) + .withTombstone(rangeTombstone(1, 4).timestamp(200)) + .withTombstone(rangeTombstone(2, 5).timestamp(50)) + .expectInMemory() + .verify(); + } + + @Test + public void inMemoryResponseWithOverlappingRangeTombstonesAtDifferentTimestampsReversed() + { + // Same tombstones as above with reversed read order; still kept in memory (open marker cannot be split). + inMemoryAssertion() + .maxRows(2).rows(5).reversed() + .withTombstone(rangeTombstone(0, 3).timestamp(100)) + .withTombstone(rangeTombstone(1, 4).timestamp(200)) + .withTombstone(rangeTombstone(2, 5).timestamp(50)) + .expectInMemory() + .verify(); + } + + private InMemoryAssertionBuilder inMemoryAssertion() + { + return new InMemoryAssertionBuilder(); + } + + private class InMemoryAssertionBuilder + { + private int maxRows = 0; // per-request row limit; 0 = disabled + private Integer maxBytes = null; // if set, the byte limit is sized to hold this many leading Unfiltered objects; null = disabled + private int rows = 0; + private boolean reversed = false; + private final List tombstones = new ArrayList<>(); + // true = the whole response is expected to be kept as an object graph, false = serialized in full + private Boolean expectInMemory = null; + + InMemoryAssertionBuilder maxRows(int maxRows) { this.maxRows = maxRows; return this; } + // Enables the byte limit, sizing it to exactly hold the first n Unfiltered objects the read produces + InMemoryAssertionBuilder maxBytesAsSizeOfFirstNRows(int n) {this.maxBytes = n; return this; } + InMemoryAssertionBuilder rows(int rows) { this.rows = rows; return this; } + InMemoryAssertionBuilder reversed() { this.reversed = true; return this; } + InMemoryAssertionBuilder withTombstone(RangeTombstoneSpec rt) { tombstones.add(rt); return this; } + // The response stays within the limits and is kept in memory as an object graph. + InMemoryAssertionBuilder expectInMemory() { this.expectInMemory = true; return this; } + // A limit is crossed, so the whole response is serialized into a buffer. + InMemoryAssertionBuilder expectSerialized() { this.expectInMemory = false; return this; } + + void verify() + { + int partitionKey = key(); + ReadCommand command = command(partitionKey, metadataWithClustering, reversed); + StubRepairedDataInfo rdi = new StubRepairedDataInfo(ByteBufferUtil.EMPTY_BYTE_BUFFER, true); + + PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadataWithClustering, ByteBufferUtil.bytes(partitionKey)).timestamp(0); + for (int i = 0; i < rows; i++) + builder.row(i).add("v", i); + for (RangeTombstoneSpec rt : tombstones) + { + Slice slice = Slice.make(Clustering.make(ByteBufferUtil.bytes(rt.start)), + Clustering.make(ByteBufferUtil.bytes(rt.end))); + builder.addRangeTombstone(new RangeTombstone(slice, + DeletionTime.build(rt.markedForDeleteAt, FBUtilities.nowInSeconds()) + ) + ); + } + PartitionUpdate update = builder.build(); + + long maxHeapSize = maxBytes == null ? 0 : heapSizeOfFirstNRows(update, reversed, maxBytes); + + ReadResponse localResponse = command.createResponse(singlePartitionIterator(update, reversed), rdi); + ReadResponse inMemoryResponse = ReadResponse.createInMemoryDataResponse(singlePartitionIterator(update, reversed), command, rdi, maxRows, maxHeapSize); + + // The reconstructed response must match the plain serialized response regardless of representation. + List expected = collectUnfiltered(command, localResponse); + assertEquals(expected, collectUnfiltered(command, inMemoryResponse)); + + // When kept in memory the whole response is an object graph (inMemoryUnfilteredCount == all unfiltereds); + // when serialized nothing is kept in memory (inMemoryUnfilteredCount == 0). + if (expectInMemory != null) + assertEquals("unexpected inMemoryUnfilteredCount", + expectInMemory ? expected.size() : 0, + inMemoryResponse.inMemoryUnfilteredCount()); + } + } + + /** + * Returns a heap size of the first {@code n} Unfiltered objects the read of + * {@code update} produces. Feeding this to the in-memory response keeps those {@code n} objects in memory (the + * (n+1)th crosses the >= threshold and overflows), which lets the assertions above express the byte-based split + * as a count of leading Unfiltered objects even though the limit itself is a heap size. + */ + private long heapSizeOfFirstNRows(PartitionUpdate update, boolean reversed, int n) + { + if (n <= 0) + return 0; + + long size = 0; + int count = 0; + try (UnfilteredRowIterator rowIter = update.unfilteredIterator(ColumnFilter.SelectionColumnFilter.all(update.columns()), Slices.ALL, reversed)) + { + while (rowIter.hasNext() && count < n) + { + size += unsharedHeapSize(rowIter.next()); + count++; + } + } + return size; + } + + private static long unsharedHeapSize(Unfiltered unfiltered) + { + return unfiltered.isRow() + ? ((Row) unfiltered).unsharedHeapSize() + : ((RangeTombstoneMarker) unfiltered).unsharedHeapSize(); + } + + private static RangeTombstoneSpec rangeTombstone(int start, int end) + { + return new RangeTombstoneSpec(start, end); + } + + private static class RangeTombstoneSpec + { + final int start; + final int end; + long markedForDeleteAt; + + RangeTombstoneSpec(int start, int end) + { + this.start = start; + this.end = end; + } + + RangeTombstoneSpec timestamp(long ts) + { + this.markedForDeleteAt = ts; + return this; + } + } + + private void assertIteratorsEqual(ReadCommand command, ReadResponse expected, ReadResponse actual) + { + List expectedUnfiltered = collectUnfiltered(command, expected); + List actualUnfiltered = collectUnfiltered(command, actual); + assertEquals(expectedUnfiltered, actualUnfiltered); + } + + private List collectUnfiltered(ReadCommand command, ReadResponse response) + { + List result = new ArrayList<>(); + try (UnfilteredPartitionIterator iter = response.makeIterator(command)) + { + while (iter.hasNext()) + { + try (UnfilteredRowIterator partition = iter.next()) + { + while (partition.hasNext()) + result.add(partition.next().toString(partition.metadata(), true)); + } + } + } + return result; + } + + private Row buildRow(TableMetadata metadata, DecoratedKey key) + { + ColumnMetadata col = metadata.getColumn(ByteBufferUtil.bytes("v")); + Clustering clustering = Clustering.EMPTY; + return BTreeRow.singleCellRow(clustering, BufferCell.live(col, FBUtilities.timestampMicros(), ByteBufferUtil.bytes(42))); + } + + private PartitionUpdate buildMultiRowUpdate(TableMetadata metadata, int partitionKey, int rowCount) + { + PartitionUpdate.SimpleBuilder builder = PartitionUpdate.simpleBuilder(metadata, ByteBufferUtil.bytes(partitionKey)).timestamp(0); + for (int i = 0; i < rowCount; i++) + builder.row(i).add("v", i); + return builder.build(); + } + + private UnfilteredPartitionIterator singlePartitionIterator(PartitionUpdate update) + { + return singlePartitionIterator(update, false); + } + private UnfilteredPartitionIterator singlePartitionIterator(PartitionUpdate update, boolean reversed) + { + UnfilteredRowIterator rowIter = update.unfilteredIterator(ColumnFilter.SelectionColumnFilter.all(update.columns()), Slices.ALL, reversed); + return new AbstractUnfilteredPartitionIterator() + { + private boolean returned = false; + + public TableMetadata metadata() { return update.metadata(); } + + public boolean hasNext() { return !returned; } + + public UnfilteredRowIterator next() + { + returned = true; + return rowIter; + } + }; + } + private void verifySerDe(ReadResponse response) { // check that roundtripping through ReadResponse.serializer behaves as expected for (MessagingService.Version version : MessagingService.Version.supportedVersions()) @@ -214,12 +735,18 @@ public class ReadResponseTest private ReadCommand digestCommand(int key, TableMetadata metadata) { - return new StubReadCommand(key, metadata, true); + return new StubReadCommand(key, metadata, true, false); } private ReadCommand command(int key, TableMetadata metadata) { - return new StubReadCommand(key, metadata, false); + return command(key, metadata, false); + } + + private ReadCommand command(int key, TableMetadata metadata, boolean reversed) + { + return new StubReadCommand(key, metadata, false, reversed); + } private static class StubRepairedDataInfo extends RepairedDataInfo @@ -247,9 +774,84 @@ public class ReadResponseTest } } + /** + * Simulates a RepairedDataInfo that updates its digest lazily as rows are consumed. + * Returns EMPTY_BYTE_BUFFER until the partition iterator wrapping it is fully consumed, + * at which point getDigest() returns the provided expected digest. + * This lets us verify that InMemoryDataResponse captures the digest *after* consumption. + */ + private static class LazyRepairedDataInfo extends RepairedDataInfo + { + private final ByteBuffer finalDigest; + private boolean iteratorConsumed = false; + private boolean digestCapturedAfterConsumption = false; + + LazyRepairedDataInfo(ByteBuffer finalDigest) + { + super(null); + this.finalDigest = finalDigest; + } + + void markConsumed() + { + iteratorConsumed = true; + } + + boolean wasConsumedBeforeDigestCaptured() + { + return digestCapturedAfterConsumption; + } + + @Override + public ByteBuffer getDigest() + { + if (iteratorConsumed) + digestCapturedAfterConsumption = true; + return iteratorConsumed ? finalDigest : ByteBufferUtil.EMPTY_BYTE_BUFFER; + } + + @Override + public boolean isConclusive() + { + return true; + } + } + + private UnfilteredPartitionIterator lazyRdiWrappedIterator(PartitionUpdate update, LazyRepairedDataInfo rdi) + { + // Wraps the row iterator so that close() marks rdi as consumed, simulating RepairedDataInfo + // updating its digest lazily after the partition has been fully read. + UnfilteredRowIterator baseRowIter = update.unfilteredIterator(); + UnfilteredRowIterator wrappedRowIter = new WrappingUnfilteredRowIterator() + { + public UnfilteredRowIterator wrapped() { return baseRowIter; } + + @Override + public void close() + { + rdi.markConsumed(); + baseRowIter.close(); + } + }; + return new AbstractUnfilteredPartitionIterator() + { + private boolean returned = false; + + public TableMetadata metadata() { return update.metadata(); } + + public boolean hasNext() { return !returned; } + + public UnfilteredRowIterator next() + { + returned = true; + return wrappedRowIter; + } + }; + } + private static class StubReadCommand extends SinglePartitionReadCommand { - StubReadCommand(int key, TableMetadata metadata, boolean isDigest) + StubReadCommand(int key, TableMetadata metadata, boolean isDigest, boolean reversed) { super(metadata.epoch, isDigest, @@ -262,11 +864,11 @@ public class ReadResponseTest RowFilter.none(), DataLimits.NONE, metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key)), - null, + new ClusteringIndexSliceFilter(Slices.ALL, reversed), null, false, null); - + } @Override