mirror of https://github.com/apache/cassandra
Reduce memory allocations in miscellaneous places along read path
org.apache.cassandra.db.rows.RangeTombstoneMarker.Merger is allocated on demand Optional usage in org.apache.cassandra.db.rows.UnfilteredRowIteratorWithLowerBound is replaced with a boolean field Double in org.apache.cassandra.io.util.CompressedChunkReader#getCrcCheckChance - specialized supplier is used to avoid boxing Stack - adjust default from 5 to 6 to avoid extra resizings in org.apache.cassandra.db.transform.Stack#resize EnumSet for CL guardrail in SelectStatement - add a condition to allocate it only if the guardrail is enabled EnumSet in ResultMetadata - replace it with plain int flags patch by Dmitry Konstantinov; reviewed by Francisco Guerrero for CASSANDRA-21360
This commit is contained in:
Dmitry Konstantinov
parent
0d2160631a
commit
64e041788a
|
|
@ -1,4 +1,5 @@
|
|||
6.0-alpha2
|
||||
* Reduce memory allocations in miscellaneous places along read path (CASSANDRA-21360)
|
||||
* Avoid ByteBuffer wrapping in cql3.selection.Selector.InputRow to reduce memory allocation rate (CASSANDRA-21362)
|
||||
* Reduce cost to calculate BTreeRow.minDeletionTime (CASSANDRA-21414)
|
||||
* Implement custom CassandraThread to keep direct references to frequently used thread local objects (CASSANDRA-21020)
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ import java.util.ArrayList;
|
|||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumSet;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
|
|
@ -149,7 +148,7 @@ public class ResultSet
|
|||
else
|
||||
{
|
||||
sb.append(" | ");
|
||||
if (metadata.flags.contains(Flag.NO_METADATA))
|
||||
if (Flag.contains(metadata.flags, Flag.NO_METADATA))
|
||||
sb.append("0x").append(ByteBufferUtil.bytesToHex(ByteBuffer.wrap(v)));
|
||||
else
|
||||
sb.append(metadata.names.get(i).type.getString(ByteBuffer.wrap(v)));
|
||||
|
|
@ -220,9 +219,9 @@ public class ResultSet
|
|||
{
|
||||
public static final CBCodec<ResultMetadata> codec = new Codec();
|
||||
|
||||
public static final ResultMetadata EMPTY = new ResultMetadata(MD5Digest.compute(new byte[0]), EnumSet.of(Flag.NO_METADATA), null, 0, null);
|
||||
public static final ResultMetadata EMPTY = new ResultMetadata(MD5Digest.compute(new byte[0]), Flag.of(Flag.NO_METADATA), null, 0, null);
|
||||
|
||||
private final EnumSet<Flag> flags;
|
||||
private int flags;
|
||||
// Please note that columnCount can actually be smaller than names, even if names is not null. This is
|
||||
// used to include columns in the resultSet that we need to do post-query re-orderings
|
||||
// (SelectStatement.orderResults) but that shouldn't be sent to the user as they haven't been requested
|
||||
|
|
@ -234,9 +233,9 @@ public class ResultSet
|
|||
|
||||
public ResultMetadata(MD5Digest digest, List<ColumnSpecification> names)
|
||||
{
|
||||
this(digest, EnumSet.noneOf(Flag.class), names, names.size(), null);
|
||||
this(digest, Flag.none(), names, names.size(), null);
|
||||
if (!names.isEmpty() && ColumnSpecification.allInSameTable(names))
|
||||
flags.add(Flag.GLOBAL_TABLES_SPEC);
|
||||
flags = Flag.add(flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
}
|
||||
|
||||
// Problem is that we compute the metadata from the columns on creation;
|
||||
|
|
@ -250,12 +249,12 @@ public class ResultSet
|
|||
// when re-preparing we create the intermediate object
|
||||
public ResultMetadata(List<ColumnSpecification> names, PagingState pagingState)
|
||||
{
|
||||
this(computeResultMetadataId(names), EnumSet.noneOf(Flag.class), names, names.size(), pagingState);
|
||||
this(computeResultMetadataId(names), Flag.none(), names, names.size(), pagingState);
|
||||
if (!names.isEmpty() && ColumnSpecification.allInSameTable(names))
|
||||
flags.add(Flag.GLOBAL_TABLES_SPEC);
|
||||
flags = Flag.add(flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
}
|
||||
|
||||
private ResultMetadata(MD5Digest resultMetadataId, EnumSet<Flag> flags, List<ColumnSpecification> names, int columnCount, PagingState pagingState)
|
||||
private ResultMetadata(MD5Digest resultMetadataId, int flags, List<ColumnSpecification> names, int columnCount, PagingState pagingState)
|
||||
{
|
||||
this.resultMetadataId = resultMetadataId;
|
||||
this.flags = flags;
|
||||
|
|
@ -266,7 +265,7 @@ public class ResultSet
|
|||
|
||||
public ResultMetadata copy()
|
||||
{
|
||||
return new ResultMetadata(resultMetadataId, EnumSet.copyOf(flags), names, columnCount, pagingState);
|
||||
return new ResultMetadata(resultMetadataId, flags, names, columnCount, pagingState);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -285,7 +284,7 @@ public class ResultSet
|
|||
}
|
||||
|
||||
@VisibleForTesting
|
||||
public EnumSet<Flag> getFlags()
|
||||
public int getFlags()
|
||||
{
|
||||
return flags;
|
||||
}
|
||||
|
|
@ -319,19 +318,19 @@ public class ResultSet
|
|||
{
|
||||
this.pagingState = pagingState;
|
||||
if (pagingState == null)
|
||||
flags.remove(Flag.HAS_MORE_PAGES);
|
||||
flags = Flag.remove(flags, Flag.HAS_MORE_PAGES);
|
||||
else
|
||||
flags.add(Flag.HAS_MORE_PAGES);
|
||||
flags = Flag.add(flags, Flag.HAS_MORE_PAGES);
|
||||
}
|
||||
|
||||
public void setSkipMetadata()
|
||||
{
|
||||
flags.add(Flag.NO_METADATA);
|
||||
flags = Flag.add(flags, Flag.NO_METADATA);
|
||||
}
|
||||
|
||||
public void setMetadataChanged()
|
||||
{
|
||||
flags.add(Flag.METADATA_CHANGED);
|
||||
flags = Flag.add(flags, Flag.METADATA_CHANGED);
|
||||
}
|
||||
|
||||
public MD5Digest getResultMetadataId()
|
||||
|
|
@ -358,7 +357,7 @@ public class ResultSet
|
|||
|
||||
ResultMetadata that = (ResultMetadata) other;
|
||||
|
||||
return Objects.equals(flags, that.flags)
|
||||
return flags == that.flags
|
||||
&& Objects.equals(names, that.names)
|
||||
&& columnCount == that.columnCount
|
||||
&& Objects.equals(pagingState, that.pagingState);
|
||||
|
|
@ -388,7 +387,7 @@ public class ResultSet
|
|||
sb.append(", ").append(name.type).append("]");
|
||||
}
|
||||
}
|
||||
if (flags.contains(Flag.HAS_MORE_PAGES))
|
||||
if (Flag.contains(flags, Flag.HAS_MORE_PAGES))
|
||||
sb.append(" (to be continued)");
|
||||
return sb.toString();
|
||||
}
|
||||
|
|
@ -398,28 +397,26 @@ public class ResultSet
|
|||
public ResultMetadata decode(ByteBuf body, ProtocolVersion version)
|
||||
{
|
||||
// flags & column count
|
||||
int iflags = body.readInt();
|
||||
int flags = Flag.deserialize(body.readInt());
|
||||
int columnCount = body.readInt();
|
||||
|
||||
EnumSet<Flag> flags = Flag.deserialize(iflags);
|
||||
|
||||
MD5Digest resultMetadataId = null;
|
||||
if (flags.contains(Flag.METADATA_CHANGED))
|
||||
if (Flag.contains(flags, Flag.METADATA_CHANGED))
|
||||
{
|
||||
assert version.isGreaterOrEqualTo(ProtocolVersion.V5) : "MetadataChanged flag is not supported before native protocol v5";
|
||||
assert !flags.contains(Flag.NO_METADATA) : "MetadataChanged and NoMetadata are mutually exclusive flags";
|
||||
assert !Flag.contains(flags, Flag.NO_METADATA) : "MetadataChanged and NoMetadata are mutually exclusive flags";
|
||||
|
||||
resultMetadataId = MD5Digest.wrap(CBUtil.readBytes(body));
|
||||
}
|
||||
|
||||
PagingState state = null;
|
||||
if (flags.contains(Flag.HAS_MORE_PAGES))
|
||||
if (Flag.contains(flags, Flag.HAS_MORE_PAGES))
|
||||
state = PagingState.deserialize(CBUtil.readValueNoCopy(body), version);
|
||||
|
||||
if (flags.contains(Flag.NO_METADATA))
|
||||
if (Flag.contains(flags, Flag.NO_METADATA))
|
||||
return new ResultMetadata(null, flags, null, columnCount, state);
|
||||
|
||||
boolean globalTablesSpec = flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean globalTablesSpec = Flag.contains(flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
|
||||
String globalKsName = null;
|
||||
String globalCfName = null;
|
||||
|
|
@ -444,10 +441,10 @@ public class ResultSet
|
|||
|
||||
public void encode(ResultMetadata m, ByteBuf dest, ProtocolVersion version)
|
||||
{
|
||||
boolean noMetadata = m.flags.contains(Flag.NO_METADATA);
|
||||
boolean globalTablesSpec = m.flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean hasMorePages = m.flags.contains(Flag.HAS_MORE_PAGES);
|
||||
boolean metadataChanged = m.flags.contains(Flag.METADATA_CHANGED);
|
||||
boolean noMetadata = Flag.contains(m.flags, Flag.NO_METADATA);
|
||||
boolean globalTablesSpec = Flag.contains(m.flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean hasMorePages = Flag.contains(m.flags, Flag.HAS_MORE_PAGES);
|
||||
boolean metadataChanged = Flag.contains(m.flags, Flag.METADATA_CHANGED);
|
||||
assert version.isGreaterThan(ProtocolVersion.V1) || (!hasMorePages && !noMetadata)
|
||||
: "version = " + version + ", flags = " + m.flags;
|
||||
|
||||
|
|
@ -487,10 +484,10 @@ public class ResultSet
|
|||
|
||||
public int encodedSize(ResultMetadata m, ProtocolVersion version)
|
||||
{
|
||||
boolean noMetadata = m.flags.contains(Flag.NO_METADATA);
|
||||
boolean globalTablesSpec = m.flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean hasMorePages = m.flags.contains(Flag.HAS_MORE_PAGES);
|
||||
boolean metadataChanged = m.flags.contains(Flag.METADATA_CHANGED);
|
||||
boolean noMetadata = Flag.contains(m.flags, Flag.NO_METADATA);
|
||||
boolean globalTablesSpec = Flag.contains(m.flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean hasMorePages = Flag.contains(m.flags, Flag.HAS_MORE_PAGES);
|
||||
boolean metadataChanged = Flag.contains(m.flags, Flag.METADATA_CHANGED);
|
||||
|
||||
int size = 8;
|
||||
if (hasMorePages)
|
||||
|
|
@ -531,18 +528,18 @@ public class ResultSet
|
|||
{
|
||||
public static final CBCodec<PreparedMetadata> codec = new Codec();
|
||||
|
||||
private final EnumSet<Flag> flags;
|
||||
private int flags;
|
||||
public final List<ColumnSpecification> names;
|
||||
private final short[] partitionKeyBindIndexes;
|
||||
|
||||
public PreparedMetadata(List<ColumnSpecification> names, short[] partitionKeyBindIndexes)
|
||||
{
|
||||
this(EnumSet.noneOf(Flag.class), names, partitionKeyBindIndexes);
|
||||
this(Flag.none(), names, partitionKeyBindIndexes);
|
||||
if (!names.isEmpty() && ColumnSpecification.allInSameTable(names))
|
||||
flags.add(Flag.GLOBAL_TABLES_SPEC);
|
||||
flags = Flag.add(flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
}
|
||||
|
||||
private PreparedMetadata(EnumSet<Flag> flags, List<ColumnSpecification> names, short[] partitionKeyBindIndexes)
|
||||
private PreparedMetadata(int flags, List<ColumnSpecification> names, short[] partitionKeyBindIndexes)
|
||||
{
|
||||
this.flags = flags;
|
||||
this.names = names;
|
||||
|
|
@ -551,7 +548,7 @@ public class ResultSet
|
|||
|
||||
public PreparedMetadata copy()
|
||||
{
|
||||
return new PreparedMetadata(EnumSet.copyOf(flags), names, partitionKeyBindIndexes);
|
||||
return new PreparedMetadata(flags, names, partitionKeyBindIndexes);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -565,7 +562,7 @@ public class ResultSet
|
|||
|
||||
PreparedMetadata that = (PreparedMetadata) other;
|
||||
return this.names.equals(that.names) &&
|
||||
this.flags.equals(that.flags) &&
|
||||
this.flags == that.flags &&
|
||||
Arrays.equals(this.partitionKeyBindIndexes, that.partitionKeyBindIndexes);
|
||||
}
|
||||
|
||||
|
|
@ -610,11 +607,9 @@ public class ResultSet
|
|||
public PreparedMetadata decode(ByteBuf body, ProtocolVersion version)
|
||||
{
|
||||
// flags & column count
|
||||
int iflags = body.readInt();
|
||||
int flags = Flag.deserialize(body.readInt());
|
||||
int columnCount = body.readInt();
|
||||
|
||||
EnumSet<Flag> flags = Flag.deserialize(iflags);
|
||||
|
||||
short[] partitionKeyBindIndexes = null;
|
||||
if (version.isGreaterOrEqualTo(ProtocolVersion.V4))
|
||||
{
|
||||
|
|
@ -627,7 +622,7 @@ public class ResultSet
|
|||
}
|
||||
}
|
||||
|
||||
boolean globalTablesSpec = flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean globalTablesSpec = Flag.contains(flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
|
||||
String globalKsName = null;
|
||||
String globalCfName = null;
|
||||
|
|
@ -652,7 +647,7 @@ public class ResultSet
|
|||
|
||||
public void encode(PreparedMetadata m, ByteBuf dest, ProtocolVersion version)
|
||||
{
|
||||
boolean globalTablesSpec = m.flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean globalTablesSpec = Flag.contains(m.flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
dest.writeInt(Flag.serialize(m.flags));
|
||||
dest.writeInt(m.names.size());
|
||||
|
||||
|
|
@ -691,7 +686,7 @@ public class ResultSet
|
|||
|
||||
public int encodedSize(PreparedMetadata m, ProtocolVersion version)
|
||||
{
|
||||
boolean globalTablesSpec = m.flags.contains(Flag.GLOBAL_TABLES_SPEC);
|
||||
boolean globalTablesSpec = Flag.contains(m.flags, Flag.GLOBAL_TABLES_SPEC);
|
||||
int size = 8;
|
||||
if (globalTablesSpec)
|
||||
{
|
||||
|
|
@ -725,24 +720,51 @@ public class ResultSet
|
|||
NO_METADATA,
|
||||
METADATA_CHANGED;
|
||||
|
||||
public static EnumSet<Flag> deserialize(int flags)
|
||||
private final int mask;
|
||||
|
||||
Flag()
|
||||
{
|
||||
EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
|
||||
Flag[] values = Flag.values();
|
||||
for (int n = 0; n < values.length; n++)
|
||||
{
|
||||
if ((flags & (1 << n)) != 0)
|
||||
set.add(values[n]);
|
||||
}
|
||||
return set;
|
||||
this.mask = 1 << this.ordinal();
|
||||
}
|
||||
|
||||
public static int serialize(EnumSet<Flag> flags)
|
||||
public static int none()
|
||||
{
|
||||
int i = 0;
|
||||
for (Flag flag : flags)
|
||||
i |= 1 << flag.ordinal();
|
||||
return i;
|
||||
return 0;
|
||||
}
|
||||
|
||||
public static int of(Flag flag)
|
||||
{
|
||||
return add(none(), flag);
|
||||
}
|
||||
|
||||
public static int of(Flag flag1, Flag flag2)
|
||||
{
|
||||
return add(add(none(), flag1), flag2);
|
||||
}
|
||||
|
||||
public static int add(int flags, Flag flag)
|
||||
{
|
||||
return flags | flag.mask;
|
||||
}
|
||||
|
||||
public static int remove(int flags, Flag flag)
|
||||
{
|
||||
return flags & ~flag.mask;
|
||||
}
|
||||
|
||||
public static boolean contains(int flags, Flag flag)
|
||||
{
|
||||
return (flags & flag.mask) != 0;
|
||||
}
|
||||
|
||||
public static int serialize(int flags)
|
||||
{
|
||||
return flags;
|
||||
}
|
||||
|
||||
public static int deserialize(int flags)
|
||||
{
|
||||
return flags;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -366,7 +366,8 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
|
|||
checkNotNull(cl, "Invalid empty consistency level");
|
||||
|
||||
cl.validateForRead();
|
||||
Guardrails.readConsistencyLevels.guard(EnumSet.of(cl), state.getClientState());
|
||||
if (Guardrails.readConsistencyLevels.enabled(state.getClientState())) // to avoid EnumSet allocation
|
||||
Guardrails.readConsistencyLevels.guard(EnumSet.of(cl), state.getClientState());
|
||||
|
||||
long nowInSec = options.getNowInSeconds(state);
|
||||
int userLimit = getLimit(options);
|
||||
|
|
|
|||
|
|
@ -73,6 +73,7 @@ public interface RangeTombstoneMarker extends Unfiltered, IMeasurableMemory
|
|||
*/
|
||||
public static class Merger
|
||||
{
|
||||
static final RangeTombstoneMarker[] EMPTY_MARKERS = new RangeTombstoneMarker[]{};
|
||||
private final DeletionTime partitionDeletion;
|
||||
private final boolean reversed;
|
||||
|
||||
|
|
|
|||
|
|
@ -21,7 +21,6 @@
|
|||
package org.apache.cassandra.db.rows;
|
||||
|
||||
import java.util.Comparator;
|
||||
import java.util.Optional;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
|
||||
|
|
@ -58,7 +57,8 @@ public class UnfilteredRowIteratorWithLowerBound extends LazilyInitializedUnfilt
|
|||
private final boolean isReverseOrder;
|
||||
private final ColumnFilter selectedColumns;
|
||||
private final SSTableReadsListener listener;
|
||||
private Optional<Unfiltered> lowerBoundMarker;
|
||||
private boolean lowerBoundComputed;
|
||||
private Unfiltered lowerBoundMarker;
|
||||
private boolean firstItemRetrieved;
|
||||
|
||||
public UnfilteredRowIteratorWithLowerBound(DecoratedKey partitionKey,
|
||||
|
|
@ -89,8 +89,8 @@ public class UnfilteredRowIteratorWithLowerBound extends LazilyInitializedUnfilt
|
|||
|
||||
public Unfiltered lowerBound()
|
||||
{
|
||||
if (lowerBoundMarker != null)
|
||||
return lowerBoundMarker.orElse(null);
|
||||
if (lowerBoundComputed)
|
||||
return lowerBoundMarker;
|
||||
|
||||
// lower bound from cache may be more accurate as it stores information about clusterings range for that exact
|
||||
// row, so we try it first (without initializing iterator)
|
||||
|
|
@ -99,12 +99,10 @@ public class UnfilteredRowIteratorWithLowerBound extends LazilyInitializedUnfilt
|
|||
// If we couldn't get the lower bound from cache, we try with metadata
|
||||
lowerBound = maybeGetLowerBoundFromMetadata();
|
||||
|
||||
if (lowerBound != null)
|
||||
lowerBoundMarker = Optional.of(makeBound(lowerBound));
|
||||
else
|
||||
lowerBoundMarker = Optional.empty();
|
||||
lowerBoundMarker = lowerBound != null ? makeBound(lowerBound) : null;
|
||||
lowerBoundComputed = true;
|
||||
|
||||
return lowerBoundMarker.orElse(null);
|
||||
return lowerBoundMarker;
|
||||
}
|
||||
|
||||
private Unfiltered makeBound(ClusteringBound<?> bound)
|
||||
|
|
|
|||
|
|
@ -547,12 +547,15 @@ public abstract class UnfilteredRowIterators
|
|||
private Unfiltered.Kind nextKind;
|
||||
|
||||
private final Row.Merger rowMerger;
|
||||
private final RangeTombstoneMarker.Merger markerMerger;
|
||||
private RangeTombstoneMarker.Merger markerMerger;
|
||||
private final int size;
|
||||
private final boolean reversed;
|
||||
|
||||
private MergeReducer(int size, boolean reversed, MergeListener listener)
|
||||
{
|
||||
this.rowMerger = new Row.Merger(size, columns().regulars.hasComplex());
|
||||
this.markerMerger = new RangeTombstoneMarker.Merger(size, partitionLevelDeletion(), reversed);
|
||||
this.size = size;
|
||||
this.reversed = reversed;
|
||||
this.listener = listener;
|
||||
}
|
||||
|
||||
|
|
@ -569,23 +572,42 @@ public abstract class UnfilteredRowIterators
|
|||
if (nextKind == Unfiltered.Kind.ROW)
|
||||
rowMerger.add(idx, (Row)current);
|
||||
else
|
||||
{
|
||||
if (markerMerger == null)
|
||||
markerMerger = new RangeTombstoneMarker.Merger(size, partitionLevelDeletion(), reversed);
|
||||
markerMerger.add(idx, (RangeTombstoneMarker)current);
|
||||
}
|
||||
}
|
||||
|
||||
protected Unfiltered getReduced()
|
||||
{
|
||||
if (nextKind == null)
|
||||
return null;
|
||||
|
||||
if (nextKind == Unfiltered.Kind.ROW)
|
||||
{
|
||||
Row merged = rowMerger.merge(markerMerger.activeDeletion());
|
||||
DeletionTime activeDeletion = markerMerger != null ? markerMerger.activeDeletion() : partitionLevelDeletion();
|
||||
Row merged = rowMerger.merge(activeDeletion);
|
||||
if (listener != null)
|
||||
listener.onMergedRows(merged == null ? BTreeRow.emptyRow(rowMerger.mergedClustering()) : merged, rowMerger.mergedRows());
|
||||
return merged;
|
||||
}
|
||||
else
|
||||
{
|
||||
RangeTombstoneMarker merged = markerMerger.merge();
|
||||
RangeTombstoneMarker merged;
|
||||
RangeTombstoneMarker[] mergedMarkers;
|
||||
if (markerMerger != null)
|
||||
{
|
||||
merged = markerMerger.merge();
|
||||
mergedMarkers = markerMerger.mergedMarkers();
|
||||
}
|
||||
else
|
||||
{
|
||||
merged = null;
|
||||
mergedMarkers = RangeTombstoneMarker.Merger.EMPTY_MARKERS;
|
||||
}
|
||||
if (listener != null)
|
||||
listener.onMergedRangeTombstoneMarkers(merged, markerMerger.mergedMarkers());
|
||||
listener.onMergedRangeTombstoneMarkers(merged, mergedMarkers);
|
||||
return merged;
|
||||
}
|
||||
}
|
||||
|
|
@ -594,7 +616,7 @@ public abstract class UnfilteredRowIterators
|
|||
{
|
||||
if (nextKind == Unfiltered.Kind.ROW)
|
||||
rowMerger.clear();
|
||||
else
|
||||
else if (markerMerger != null)
|
||||
markerMerger.clear();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -26,6 +26,8 @@ class Stack
|
|||
{
|
||||
public static final Transformation[] EMPTY_TRANSFORMATIONS = new Transformation[0];
|
||||
public static final MoreContentsHolder[] EMPTY_MORE_CONTENTS_HOLDERS = new MoreContentsHolder[0];
|
||||
|
||||
static final int DEFAULT_NOT_EMPTY_SIZE = 6;
|
||||
static final Stack EMPTY = new Stack();
|
||||
|
||||
Transformation[] stack;
|
||||
|
|
@ -72,7 +74,7 @@ class Stack
|
|||
|
||||
private static <E> E[] resize(E[] array)
|
||||
{
|
||||
int newLen = array.length == 0 ? 5 : array.length * 2;
|
||||
int newLen = array.length == 0 ? DEFAULT_NOT_EMPTY_SIZE : array.length * 2;
|
||||
return Arrays.copyOf(array, newLen);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ import java.nio.BufferOverflowException;
|
|||
import java.util.Collection;
|
||||
import java.util.Set;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.DoubleSupplier;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
|
|
@ -87,7 +88,7 @@ public abstract class SortedTableWriter<P extends SortedTablePartitionWriter, I
|
|||
private DecoratedKey lastWrittenKey;
|
||||
private DataPosition dataMark;
|
||||
private long lastEarlyOpenLength;
|
||||
private final Supplier<Double> crcCheckChanceSupplier;
|
||||
private final DoubleSupplier crcCheckChanceSupplier;
|
||||
|
||||
private final boolean isPartitionSizeGuardEnabled;
|
||||
private final boolean isPartitionTombstonesGuardEnabled;
|
||||
|
|
|
|||
|
|
@ -21,7 +21,7 @@ package org.apache.cassandra.io.util;
|
|||
import java.io.IOException;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.ThreadLocalRandom;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.function.DoubleSupplier;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.primitives.Ints;
|
||||
|
|
@ -38,9 +38,9 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
|
|||
{
|
||||
final CompressionMetadata metadata;
|
||||
final int maxCompressedLength;
|
||||
final Supplier<Double> crcCheckChanceSupplier;
|
||||
final DoubleSupplier crcCheckChanceSupplier;
|
||||
|
||||
protected CompressedChunkReader(ChannelProxy channel, CompressionMetadata metadata, Supplier<Double> crcCheckChanceSupplier)
|
||||
protected CompressedChunkReader(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier)
|
||||
{
|
||||
super(channel, metadata.dataLength);
|
||||
this.metadata = metadata;
|
||||
|
|
@ -57,7 +57,7 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
|
|||
@VisibleForTesting
|
||||
public double getCrcCheckChance()
|
||||
{
|
||||
return crcCheckChanceSupplier.get();
|
||||
return crcCheckChanceSupplier.getAsDouble();
|
||||
}
|
||||
|
||||
boolean shouldCheckCrc()
|
||||
|
|
@ -279,7 +279,7 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
|
|||
private final CompressedReader reader;
|
||||
private final CompressedReader scanReader;
|
||||
|
||||
public Direct(ChannelProxy channel, CompressionMetadata metadata, Supplier<Double> crcCheckChanceSupplier)
|
||||
public Direct(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier)
|
||||
{
|
||||
super(channel, metadata, crcCheckChanceSupplier);
|
||||
int blockSize = FileUtils.getFileBlockSize(channel.file());
|
||||
|
|
@ -367,7 +367,7 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
|
|||
private final CompressedReader reader;
|
||||
private final CompressedReader scanReader;
|
||||
|
||||
public Standard(ChannelProxy channel, CompressionMetadata metadata, Supplier<Double> crcCheckChanceSupplier)
|
||||
public Standard(ChannelProxy channel, CompressionMetadata metadata, DoubleSupplier crcCheckChanceSupplier)
|
||||
{
|
||||
super(channel, metadata, crcCheckChanceSupplier);
|
||||
reader = new RandomAccessCompressedReader(channel, metadata);
|
||||
|
|
@ -465,7 +465,7 @@ public abstract class CompressedChunkReader extends AbstractReaderFileProxy impl
|
|||
{
|
||||
protected final MmappedRegions regions;
|
||||
|
||||
public Mmap(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions, Supplier<Double> crcCheckChanceSupplier)
|
||||
public Mmap(ChannelProxy channel, CompressionMetadata metadata, MmappedRegions regions, DoubleSupplier crcCheckChanceSupplier)
|
||||
{
|
||||
super(channel, metadata, crcCheckChanceSupplier);
|
||||
this.regions = regions;
|
||||
|
|
|
|||
|
|
@ -18,8 +18,8 @@
|
|||
package org.apache.cassandra.io.util;
|
||||
|
||||
import java.util.Optional;
|
||||
import java.util.function.DoubleSupplier;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import com.google.common.annotations.VisibleForTesting;
|
||||
import com.google.common.util.concurrent.RateLimiter;
|
||||
|
|
@ -75,7 +75,7 @@ public class FileHandle extends SharedCloseableImpl
|
|||
// Properties to support unbuilding via toBuilder
|
||||
private final ChunkCache chunkCache;
|
||||
private final MmappedRegionsCache mmappedRegionsCache;
|
||||
private final Supplier<Double> crcCheckChanceSupplier;
|
||||
private final DoubleSupplier crcCheckChanceSupplier;
|
||||
private final long lengthOverride;
|
||||
private final int bufferSize;
|
||||
private final BufferType bufferType;
|
||||
|
|
@ -88,7 +88,7 @@ public class FileHandle extends SharedCloseableImpl
|
|||
DiskAccessMode diskAccessMode,
|
||||
ChunkCache chunkCache,
|
||||
MmappedRegionsCache mmappedRegionsCache,
|
||||
Supplier<Double> crcCheckChanceSupplier,
|
||||
DoubleSupplier crcCheckChanceSupplier,
|
||||
long lengthOverride,
|
||||
int bufferSize,
|
||||
BufferType bufferType)
|
||||
|
|
@ -338,7 +338,7 @@ public class FileHandle extends SharedCloseableImpl
|
|||
public final File file;
|
||||
|
||||
private CompressionMetadata compressionMetadata;
|
||||
private Supplier<Double> crcCheckChanceSupplier = () -> 1.0;
|
||||
private DoubleSupplier crcCheckChanceSupplier = () -> 1.0;
|
||||
private ChunkCache chunkCache;
|
||||
private int bufferSize = RandomAccessReader.DEFAULT_BUFFER_SIZE;
|
||||
private BufferType bufferType = BufferType.OFF_HEAP;
|
||||
|
|
@ -378,7 +378,7 @@ public class FileHandle extends SharedCloseableImpl
|
|||
return this;
|
||||
}
|
||||
|
||||
public Builder withCrcCheckChance(Supplier<Double> crcCheckChanceSupplier)
|
||||
public Builder withCrcCheckChance(DoubleSupplier crcCheckChanceSupplier)
|
||||
{
|
||||
this.crcCheckChanceSupplier = crcCheckChanceSupplier;
|
||||
return this;
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ package org.apache.cassandra.cql3;
|
|||
import java.nio.ByteBuffer;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.EnumSet;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Collectors;
|
||||
|
|
@ -498,7 +497,7 @@ public class PreparedStatementsTest extends CQLTester
|
|||
Arrays.asList(Int32Serializer.instance.serialize(1),
|
||||
Int32Serializer.instance.serialize(10),
|
||||
Int32Serializer.instance.serialize(20)),
|
||||
EnumSet.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
|
||||
// This is an _unsuccessful_ LWT update (as the condition fails)
|
||||
verifyMetadataFlagsWithLWTsUpdate(simpleClient,
|
||||
|
|
@ -516,7 +515,7 @@ public class PreparedStatementsTest extends CQLTester
|
|||
Arrays.asList(Int32Serializer.instance.serialize(1),
|
||||
Int32Serializer.instance.serialize(10),
|
||||
Int32Serializer.instance.serialize(20)),
|
||||
EnumSet.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
|
||||
// force a schema change on that table
|
||||
simpleClient.execute(String.format("ALTER TABLE %s.%s ADD v3 int",
|
||||
|
|
@ -574,8 +573,10 @@ public class PreparedStatementsTest extends CQLTester
|
|||
Int32Serializer.instance.serialize(1),
|
||||
Int32Serializer.instance.serialize(30),
|
||||
null),
|
||||
EnumSet.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC,
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.METADATA_CHANGED));
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.of(
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC,
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.METADATA_CHANGED
|
||||
));
|
||||
|
||||
// This is an _unsuccessful_ LWT update (as the condition fails)
|
||||
verifyMetadataFlagsWithLWTsUpdate(simpleClient,
|
||||
|
|
@ -594,7 +595,7 @@ public class PreparedStatementsTest extends CQLTester
|
|||
Int32Serializer.instance.serialize(1),
|
||||
Int32Serializer.instance.serialize(30),
|
||||
null),
|
||||
EnumSet.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
org.apache.cassandra.cql3.ResultSet.Flag.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -602,13 +603,13 @@ public class PreparedStatementsTest extends CQLTester
|
|||
ResultMessage.Prepared prepSelect,
|
||||
List<String> columnNames,
|
||||
List<ByteBuffer> expectedRow,
|
||||
EnumSet<org.apache.cassandra.cql3.ResultSet.Flag> expectedFlags)
|
||||
int expectedFlags)
|
||||
{
|
||||
ResultMessage result = simpleClient.executePrepared(prepSelect,
|
||||
Collections.singletonList(Int32Serializer.instance.serialize(1)),
|
||||
ConsistencyLevel.LOCAL_ONE);
|
||||
ResultMessage.Rows rows = (ResultMessage.Rows) result;
|
||||
EnumSet<org.apache.cassandra.cql3.ResultSet.Flag> resultFlags = rows.result.metadata.getFlags();
|
||||
int resultFlags = rows.result.metadata.getFlags();
|
||||
assertEquals(expectedFlags,
|
||||
resultFlags);
|
||||
assertEquals(columnNames.size(),
|
||||
|
|
@ -620,7 +621,7 @@ public class PreparedStatementsTest extends CQLTester
|
|||
assertEqualRows(expectedRow,
|
||||
rows.result.rows.get(0));
|
||||
|
||||
if (resultFlags.contains(org.apache.cassandra.cql3.ResultSet.Flag.METADATA_CHANGED))
|
||||
if (org.apache.cassandra.cql3.ResultSet.Flag.contains(resultFlags, org.apache.cassandra.cql3.ResultSet.Flag.METADATA_CHANGED))
|
||||
prepSelect = prepSelect.withResultMetadata(rows.result.metadata);
|
||||
return prepSelect;
|
||||
}
|
||||
|
|
@ -635,8 +636,8 @@ public class PreparedStatementsTest extends CQLTester
|
|||
params,
|
||||
ConsistencyLevel.LOCAL_ONE);
|
||||
ResultMessage.Rows rows = (ResultMessage.Rows) result;
|
||||
EnumSet<org.apache.cassandra.cql3.ResultSet.Flag> resultFlags = rows.result.metadata.getFlags();
|
||||
assertEquals(EnumSet.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC),
|
||||
int resultFlags = rows.result.metadata.getFlags();
|
||||
assertEquals(org.apache.cassandra.cql3.ResultSet.Flag.of(org.apache.cassandra.cql3.ResultSet.Flag.GLOBAL_TABLES_SPEC),
|
||||
resultFlags);
|
||||
assertEquals(columnNames.size(),
|
||||
rows.result.metadata.getColumnCount());
|
||||
|
|
|
|||
|
|
@ -0,0 +1,94 @@
|
|||
/*
|
||||
* 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.transform;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
import org.apache.cassandra.SchemaLoader;
|
||||
import org.apache.cassandra.Util;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.db.ReadCommand;
|
||||
import org.apache.cassandra.db.ReadExecutionController;
|
||||
import org.apache.cassandra.db.marshal.UTF8Type;
|
||||
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
|
||||
import org.apache.cassandra.schema.KeyspaceParams;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
|
||||
import static org.hamcrest.Matchers.lessThanOrEqualTo;
|
||||
import static org.junit.Assert.assertThat;
|
||||
|
||||
/**
|
||||
* Verifies that {@link ReadCommand#executeLocally} does not resize stack of {@link Transformation}s.
|
||||
* It is used to ensure that the default value for the stack size is optimal.
|
||||
*
|
||||
* Transformations applied unconditionally to a SinglePartitionReadCommand
|
||||
* (default config, DataLimits.NONE, empty RowFilter):
|
||||
* 1. RTBoundValidator (Stage.MERGED)
|
||||
* 2. QueryCancellationChecker
|
||||
* 3. WithoutPurgeableTombstones
|
||||
* 4. RTBoundValidator (Stage.PURGED)
|
||||
* 5. MetricRecording
|
||||
* 6. RTBoundCloser
|
||||
*
|
||||
* Conditional transformations NOT applied in the default case tested here:
|
||||
* - QuerySizeTracking (requires trackWarnings=true AND a size threshold configured)
|
||||
* - DelayInjector (requires TEST_ITERATION_DELAY_MILLIS > 0)
|
||||
* - PurgeableTombstonesMetric (requires granularity != disabled; default is disabled)
|
||||
* - RowFilter transformation (requires non-empty RowFilter expressions)
|
||||
* - DataLimits counter (DataLimits.NONE.filter() returns iterator unchanged)
|
||||
*/
|
||||
public class TransformationOptimalSizeStackTest
|
||||
{
|
||||
private static final String KEYSPACE = "ReadCommandTransformationCountTest";
|
||||
private static final String TABLE = "tbl";
|
||||
@BeforeClass
|
||||
public static void setup() throws Exception
|
||||
{
|
||||
DatabaseDescriptor.daemonInitialization();
|
||||
TableMetadata.Builder table = TableMetadata.builder(KEYSPACE, TABLE)
|
||||
.addPartitionKeyColumn("k", UTF8Type.instance)
|
||||
.addRegularColumn("v", UTF8Type.instance);
|
||||
|
||||
SchemaLoader.prepareServer();
|
||||
SchemaLoader.createKeyspace(KEYSPACE, KeyspaceParams.simple(1), table);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNoResizingOfStackForSinglePartitionCommand()
|
||||
{
|
||||
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(TABLE);
|
||||
ReadCommand cmd = Util.cmd(cfs, "key1").build();
|
||||
|
||||
try (ReadExecutionController controller = cmd.executionController();
|
||||
UnfilteredPartitionIterator iter = cmd.executeLocally(controller))
|
||||
{
|
||||
// The goal is to ensure that the default capacity is enough
|
||||
// (not less than the typical number of transformations for a single partition read),
|
||||
// so no resizing is needed.
|
||||
// If due to some reason we want to have the default size more the test accepts it
|
||||
assertThat("Transformations in stack: " + Arrays.toString(((Stack) iter).stack),
|
||||
((Stack) iter).length, lessThanOrEqualTo(Stack.DEFAULT_NOT_EMPTY_SIZE));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue