Skip to content

[CASSANDRA-21360][trunk] Reduce memory allocations in miscellaneous places along read path #4797

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
netudima wants to merge 1 commit into
base: trunk
Choose a base branch
from
Open

[CASSANDRA-21360][trunk] Reduce memory allocations in miscellaneous places along read path #4797

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
140 changes: 81 additions & 59 deletions src/java/org/apache/cassandra/cql3/ResultSet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,6 @@
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.Objects;

Expand Down Expand Up @@ -122,7 +121,7 @@ public String toString()
else
{
sb.append(" | ");
if (metadata.flags.contains(Flag.NO_METADATA))
if (Flag.contains(metadata.flags, Flag.NO_METADATA))
sb.append("0x").append(ByteBufferUtil.bytesToHex(v));
else
sb.append(metadata.names.get(i).type.getString(v));
Expand Down Expand Up @@ -193,9 +192,9 @@ public static class ResultMetadata
{
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.add(Flag.none(), 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
Expand All @@ -207,9 +206,9 @@ public static class ResultMetadata

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;
Expand All @@ -223,12 +222,12 @@ public ResultMetadata(List<ColumnSpecification> names)
// 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;
Expand All @@ -239,7 +238,7 @@ private ResultMetadata(MD5Digest resultMetadataId, EnumSet<Flag> flags, List<Col

public ResultMetadata copy()
{
return new ResultMetadata(resultMetadataId, EnumSet.copyOf(flags), names, columnCount, pagingState);
return new ResultMetadata(resultMetadataId, flags, names, columnCount, pagingState);
}

/**
Expand All @@ -258,7 +257,7 @@ public int valueCount()
}

@VisibleForTesting
public EnumSet<Flag> getFlags()
public int getFlags()
{
return flags;
}
Expand Down Expand Up @@ -292,19 +291,19 @@ public void setHasMorePages(PagingState pagingState)
{
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()
Expand All @@ -331,7 +330,7 @@ public boolean equals(Object other)

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);
Expand Down Expand Up @@ -361,7 +360,7 @@ public String toString()
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();
}
Expand All @@ -371,28 +370,26 @@ private static class Codec implements CBCodec<ResultMetadata>
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;
Expand All @@ -417,10 +414,10 @@ public ResultMetadata decode(ByteBuf body, ProtocolVersion version)

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;

Expand Down Expand Up @@ -460,10 +457,10 @@ public void encode(ResultMetadata m, ByteBuf dest, ProtocolVersion version)

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)
Expand Down Expand Up @@ -504,18 +501,18 @@ public static class PreparedMetadata
{
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;
Expand All @@ -524,7 +521,7 @@ private PreparedMetadata(EnumSet<Flag> flags, List<ColumnSpecification> names, s

public PreparedMetadata copy()
{
return new PreparedMetadata(EnumSet.copyOf(flags), names, partitionKeyBindIndexes);
return new PreparedMetadata(flags, names, partitionKeyBindIndexes);
}

@Override
Expand All @@ -538,7 +535,7 @@ public boolean equals(Object other)

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);
}

Expand Down Expand Up @@ -583,11 +580,9 @@ private static class Codec implements CBCodec<PreparedMetadata>
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))
{
Expand All @@ -600,7 +595,7 @@ public PreparedMetadata decode(ByteBuf body, ProtocolVersion version)
}
}

boolean globalTablesSpec = flags.contains(Flag.GLOBAL_TABLES_SPEC);
boolean globalTablesSpec = Flag.contains(flags, Flag.GLOBAL_TABLES_SPEC);

String globalKsName = null;
String globalCfName = null;
Expand All @@ -625,7 +620,7 @@ public PreparedMetadata decode(ByteBuf body, ProtocolVersion version)

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());

Expand Down Expand Up @@ -664,7 +659,7 @@ public void encode(PreparedMetadata m, ByteBuf dest, ProtocolVersion version)

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)
{
Expand Down Expand Up @@ -698,24 +693,51 @@ public enum Flag
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 none()
{
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 serialize(EnumSet<Flag> flags)
public static int add(int flags, Flag flag)
{
int i = 0;
for (Flag flag : flags)
i |= 1 << flag.ordinal();
return i;
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;
}
}

Expand Down
3 changes: 2 additions & 1 deletion src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,8 @@ public ResultMessage.Rows execute(QueryState state, QueryOptions options, Dispat
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);
Expand Down
16 changes: 7 additions & 9 deletions src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorWithLowerBound.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@
package org.apache.cassandra.db.rows;

import java.util.Comparator;
import java.util.Optional;

import com.google.common.annotations.VisibleForTesting;

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -89,8 +89,8 @@ public UnfilteredRowIteratorWithLowerBound(DecoratedKey partitionKey,

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)
Expand All @@ -99,12 +99,10 @@ public Unfiltered lowerBound()
// 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)
Expand Down
Loading