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

Minimize BTree iterator allocations 

Patch by Blake Eggleston; Reviewed by Benedict Elliott Smith for CASSANDRA-15389
This commit is contained in:
Blake Eggleston 2019-10-15 20:10:42 -07:00
parent d00c004cc1
commit 8576e769d1
39 changed files with 667 additions and 447 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
4.0-alpha4
* Minimize BTree iterator allocations (CASSANDRA-15389)
* Add client request size server metrics (CASSANDRA-15704)
* Add additional logging around FileUtils and compaction leftover cleanup (CASSANDRA-15705)
* Mark system_views/system_virtual_schema as non-alterable keyspaces in cqlsh (CASSANDRA-15711)

6
src/java/org/apache/cassandra/db/ColumnIndex.java
View File

@ -53,6 +53,7 @@ public class ColumnIndex
public int columnIndexCount;
private int[] indexOffsets;
private final SerializationHelper helper;
private final SerializationHeader header;
private final int version;
private final SequentialWriter writer;
@ -79,6 +80,7 @@ public class ColumnIndex
Collection<SSTableFlushObserver> observers,
ISerializer<IndexInfo> indexInfoSerializer)
{
this.helper = new SerializationHelper(header);
this.header = header;
this.writer = writer;
this.version = version.correspondingMessagingVersion();
@ -126,7 +128,7 @@ public class ColumnIndex
{
Row staticRow = iterator.staticRow();
UnfilteredSerializer.serializer.serializeStaticRow(staticRow, header, writer, version);
UnfilteredSerializer.serializer.serializeStaticRow(staticRow, helper, writer, version);
if (!observers.isEmpty())
observers.forEach((o) -> o.nextUnfilteredCluster(staticRow));
}
@ -248,7 +250,7 @@ public class ColumnIndex
startPosition = pos;
}
UnfilteredSerializer.serializer.serialize(unfiltered, header, writer, pos - previousRowStart, version);
UnfilteredSerializer.serializer.serialize(unfiltered, helper, writer, pos - previousRowStart, version);
// notify observers about each new row
if (!observers.isEmpty())

9
src/java/org/apache/cassandra/db/Columns.java
View File

@ -53,7 +53,7 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
public static final Serializer serializer = new Serializer();
public static final Columns NONE = new Columns(BTree.empty(), 0);
private static final ColumnMetadata FIRST_COMPLEX_STATIC =
public static final ColumnMetadata FIRST_COMPLEX_STATIC =
new ColumnMetadata("",
"",
ColumnIdentifier.getInterned(ByteBufferUtil.EMPTY_BYTE_BUFFER, UTF8Type.instance),
@ -61,7 +61,7 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
ColumnMetadata.NO_POSITION,
ColumnMetadata.Kind.STATIC);
private static final ColumnMetadata FIRST_COMPLEX_REGULAR =
public static final ColumnMetadata FIRST_COMPLEX_REGULAR =
new ColumnMetadata("",
"",
ColumnIdentifier.getInterned(ByteBufferUtil.EMPTY_BYTE_BUFFER, UTF8Type.instance),
@ -382,11 +382,10 @@ public class Columns extends AbstractCollection<ColumnMetadata> implements Colle
/**
* Apply a function to each column definition in forwards or reversed order.
* @param function
* @param reversed
*/
public void apply(Consumer<ColumnMetadata> function, boolean reversed)
public void apply(Consumer<ColumnMetadata> function)
{
BTree.apply(columns, function, reversed);
BTree.apply(columns, function);
}
@Override

6
src/java/org/apache/cassandra/db/Mutation.java
View File

@ -29,7 +29,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -336,7 +336,7 @@ public class Mutation implements IMutation
PartitionUpdate.serializer.serialize(entry.getValue(), out, version);
}
public Mutation deserialize(DataInputPlus in, int version, SerializationHelper.Flag flag) throws IOException
public Mutation deserialize(DataInputPlus in, int version, DeserializationHelper.Flag flag) throws IOException
{
int size = (int)in.readUnsignedVInt();
assert size > 0;
@ -359,7 +359,7 @@ public class Mutation implements IMutation
public Mutation deserialize(DataInputPlus in, int version) throws IOException
{
return deserialize(in, version, SerializationHelper.Flag.FROM_REMOTE);
return deserialize(in, version, DeserializationHelper.Flag.FROM_REMOTE);
}
public long serializedSize(Mutation mutation, int version)

8
src/java/org/apache/cassandra/db/ReadResponse.java
View File

@ -182,7 +182,7 @@ public abstract class ReadResponse
command.getRepairedDataDigest(),
command.isRepairedDataDigestConclusive(),
MessagingService.current_version,
SerializationHelper.Flag.LOCAL);
DeserializationHelper.Flag.LOCAL);
}
private static ByteBuffer build(UnfilteredPartitionIterator iter, ColumnFilter selection)
@ -208,7 +208,7 @@ public abstract class ReadResponse
boolean isRepairedDigestConclusive,
int version)
{
super(data, repairedDataDigest, isRepairedDigestConclusive, version, SerializationHelper.Flag.FROM_REMOTE);
super(data, repairedDataDigest, isRepairedDigestConclusive, version, DeserializationHelper.Flag.FROM_REMOTE);
}
}
@ -220,13 +220,13 @@ public abstract class ReadResponse
private final ByteBuffer repairedDataDigest;
private final boolean isRepairedDigestConclusive;
private final int dataSerializationVersion;
private final SerializationHelper.Flag flag;
private final DeserializationHelper.Flag flag;
protected DataResponse(ByteBuffer data,
ByteBuffer repairedDataDigest,
boolean isRepairedDigestConclusive,
int dataSerializationVersion,
SerializationHelper.Flag flag)
DeserializationHelper.Flag flag)
{
super();
this.data = data;

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

@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.TypeParser;

6
src/java/org/apache/cassandra/db/UnfilteredDeserializer.java
View File

@ -34,7 +34,7 @@ public class UnfilteredDeserializer
{
protected final TableMetadata metadata;
protected final DataInputPlus in;
protected final SerializationHelper helper;
protected final DeserializationHelper helper;
private final ClusteringPrefix.Deserializer clusteringDeserializer;
private final SerializationHeader header;
@ -49,7 +49,7 @@ public class UnfilteredDeserializer
private UnfilteredDeserializer(TableMetadata metadata,
DataInputPlus in,
SerializationHeader header,
SerializationHelper helper)
DeserializationHelper helper)
{
this.metadata = metadata;
this.in = in;
@ -62,7 +62,7 @@ public class UnfilteredDeserializer
public static UnfilteredDeserializer create(TableMetadata metadata,
DataInputPlus in,
SerializationHeader header,
SerializationHelper helper)
DeserializationHelper helper)
{
return new UnfilteredDeserializer(metadata, in, header, helper);
}

6
src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java
View File

@ -43,7 +43,7 @@ public abstract class AbstractSSTableIterator implements UnfilteredRowIterator
protected final DecoratedKey key;
protected final DeletionTime partitionLevelDeletion;
protected final ColumnFilter columns;
protected final SerializationHelper helper;
protected final DeserializationHelper helper;
protected final Row staticRow;
protected final Reader reader;
@ -70,7 +70,7 @@ public abstract class AbstractSSTableIterator implements UnfilteredRowIterator
this.key = key;
this.columns = columnFilter;
this.slices = slices;
this.helper = new SerializationHelper(metadata, sstable.descriptor.version.correspondingMessagingVersion(), SerializationHelper.Flag.LOCAL, columnFilter);
this.helper = new DeserializationHelper(metadata, sstable.descriptor.version.correspondingMessagingVersion(), DeserializationHelper.Flag.LOCAL, columnFilter);
if (indexEntry == null)
{
@ -159,7 +159,7 @@ public abstract class AbstractSSTableIterator implements UnfilteredRowIterator
private static Row readStaticRow(SSTableReader sstable,
FileDataInput file,
SerializationHelper helper,
DeserializationHelper helper,
Columns statics) throws IOException
{
if (!sstable.header.hasStatic())

4
src/java/org/apache/cassandra/db/commitlog/CommitLogReader.java
View File

@ -34,7 +34,7 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.commitlog.CommitLogReadHandler.CommitLogReadErrorReason;
import org.apache.cassandra.db.commitlog.CommitLogReadHandler.CommitLogReadException;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.exceptions.UnknownTableException;
import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.io.util.DataInputBuffer;
@ -429,7 +429,7 @@ public class CommitLogReader
{
mutation = Mutation.serializer.deserialize(bufIn,
desc.getMessagingVersion(),
SerializationHelper.Flag.LOCAL);
DeserializationHelper.Flag.LOCAL);
// doublecheck that what we read is still] valid for the current schema
for (PartitionUpdate upd : mutation.getPartitionUpdates())
upd.validate();

4
src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java
View File

@ -177,11 +177,11 @@ public class CachedBTreePartition extends ImmutableBTreePartition implements Cac
TableMetadata metadata = Schema.instance.getExistingTableMetadata(TableId.deserialize(in));
UnfilteredRowIteratorSerializer.Header header = UnfilteredRowIteratorSerializer.serializer.deserializeHeader(metadata, null, in, version, SerializationHelper.Flag.LOCAL);
UnfilteredRowIteratorSerializer.Header header = UnfilteredRowIteratorSerializer.serializer.deserializeHeader(metadata, null, in, version, DeserializationHelper.Flag.LOCAL);
assert !header.isReversed && header.rowEstimate >= 0;
Holder holder;
try (UnfilteredRowIterator partition = UnfilteredRowIteratorSerializer.serializer.deserialize(in, version, metadata, SerializationHelper.Flag.LOCAL, header))
try (UnfilteredRowIterator partition = UnfilteredRowIteratorSerializer.serializer.deserialize(in, version, metadata, DeserializationHelper.Flag.LOCAL, header))
{
holder = ImmutableBTreePartition.build(partition, header.rowEstimate);
}

7
src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java
View File

@ -20,8 +20,6 @@ package org.apache.cassandra.db.partitions;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import com.google.common.collect.Iterables;
@ -38,7 +36,6 @@ import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
@ -230,7 +227,7 @@ public class PartitionUpdate extends AbstractBTreePartition
{
return serializer.deserialize(new DataInputBuffer(bytes, true),
version,
SerializationHelper.Flag.LOCAL);
DeserializationHelper.Flag.LOCAL);
}
catch (IOException e)
{
@ -636,7 +633,7 @@ public class PartitionUpdate extends AbstractBTreePartition
}
}
public PartitionUpdate deserialize(DataInputPlus in, int version, SerializationHelper.Flag flag) throws IOException
public PartitionUpdate deserialize(DataInputPlus in, int version, DeserializationHelper.Flag flag) throws IOException
{
TableMetadata metadata = Schema.instance.getExistingTableMetadata(TableId.deserialize(in));
UnfilteredRowIteratorSerializer.Header header = UnfilteredRowIteratorSerializer.serializer.deserializeHeader(metadata, null, in, version, flag);

2
src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java
View File

@ -316,7 +316,7 @@ public abstract class UnfilteredPartitionIterators
out.writeBoolean(false);
}
public UnfilteredPartitionIterator deserialize(final DataInputPlus in, final int version, final TableMetadata metadata, final ColumnFilter selection, final SerializationHelper.Flag flag) throws IOException
public UnfilteredPartitionIterator deserialize(final DataInputPlus in, final int version, final TableMetadata metadata, final ColumnFilter selection, final DeserializationHelper.Flag flag) throws IOException
{
// Skip now unused isForThrift boolean
in.readBoolean();

18
src/java/org/apache/cassandra/db/rows/AbstractRow.java
View File

@ -24,10 +24,11 @@ import java.util.stream.StreamSupport;
import com.google.common.collect.Iterables;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.Digest;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.db.marshal.UserType;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.serializers.MarshalException;
/**
@ -66,8 +67,7 @@ public abstract class AbstractRow implements Row
deletion().digest(digest);
primaryKeyLivenessInfo().digest(digest);
for (ColumnData cd : this)
cd.digest(digest);
apply(ColumnData::digest, digest);
}
public void validateData(TableMetadata metadata)
@ -93,15 +93,7 @@ public abstract class AbstractRow implements Row
if (deletion().time().localDeletionTime() < 0)
throw new MarshalException("A local deletion time should not be negative in '" + metadata + "'");
for (ColumnData cd : this)
try
{
cd.validate();
}
catch (Exception e)
{
throw new MarshalException("data for '" + cd.column.debugString() + "', " + cd + " in '" + metadata + "' didn't validate", e);
}
apply(cd -> cd.validate());
}
public boolean hasInvalidDeletions()

103
src/java/org/apache/cassandra/db/rows/BTreeRow.java
View File

@ -19,13 +19,14 @@ package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Predicate;
import com.google.common.base.Function;
import com.google.common.collect.Collections2;
import com.google.common.collect.Iterables;
import com.google.common.collect.Iterators;
import com.google.common.primitives.Ints;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
@ -53,6 +54,11 @@ public class BTreeRow extends AbstractRow
// The data for each columns present in this row in column sorted order.
private final Object[] btree;
private static final ColumnData FIRST_COMPLEX_STATIC = new ComplexColumnData(Columns.FIRST_COMPLEX_STATIC, new Object[0], new DeletionTime(0, 0));
private static final ColumnData FIRST_COMPLEX_REGULAR = new ComplexColumnData(Columns.FIRST_COMPLEX_REGULAR, new Object[0], new DeletionTime(0, 0));
private static final Comparator<ColumnData> COLUMN_COMPARATOR = (cd1, cd2) -> cd1.column.compareTo(cd2.column);
// We need to filter the tombstones of a row on every read (twice in fact: first to remove purgeable tombstone, and then after reconciliation to remove
// all tombstone since we don't return them to the client) as well as on compaction. But it's likely that many rows won't have any tombstone at all, so
// we want to speed up that case by not having to iterate/copy the row in this case. We could keep a single boolean telling us if we have tombstones,
@ -90,8 +96,8 @@ public class BTreeRow extends AbstractRow
int minDeletionTime = Math.min(minDeletionTime(primaryKeyLivenessInfo), minDeletionTime(deletion.time()));
if (minDeletionTime != Integer.MIN_VALUE)
{
for (ColumnData cd : BTree.<ColumnData>iterable(btree))
minDeletionTime = Math.min(minDeletionTime, minDeletionTime(cd));
long result = BTree.<ColumnData>accumulate(btree, (cd, l) -> Math.min(l, minDeletionTime(cd)) , minDeletionTime);
minDeletionTime = Ints.checkedCast(result);
}
return create(clustering, primaryKeyLivenessInfo, deletion, btree, minDeletionTime);
@ -168,23 +174,49 @@ public class BTreeRow extends AbstractRow
return cd.column().isSimple() ? minDeletionTime((Cell) cd) : minDeletionTime((ComplexColumnData)cd);
}
public void apply(Consumer<ColumnData> function, boolean reversed)
public void apply(Consumer<ColumnData> function)
{
BTree.apply(btree, function, reversed);
BTree.apply(btree, function);
}
public void apply(Consumer<ColumnData> funtion, com.google.common.base.Predicate<ColumnData> stopCondition, boolean reversed)
public <A> void apply(BiConsumer<A, ColumnData> function, A arg)
{
BTree.apply(btree, funtion, stopCondition, reversed);
BTree.apply(btree, function, arg);
}
public long accumulate(LongAccumulator<ColumnData> accumulator, long initialValue)
{
return BTree.accumulate(btree, accumulator, initialValue);
}
public long accumulate(LongAccumulator<ColumnData> accumulator, Comparator<ColumnData> comparator, ColumnData from, long initialValue)
{
return BTree.accumulate(btree, accumulator, comparator, from, initialValue);
}
public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, long initialValue)
{
return BTree.accumulate(btree, accumulator, arg, initialValue);
}
public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, Comparator<ColumnData> comparator, ColumnData from, long initialValue)
{
return BTree.accumulate(btree, accumulator, arg, comparator, from, initialValue);
}
private static int minDeletionTime(Object[] btree, LivenessInfo info, DeletionTime rowDeletion)
{
//we have to wrap this for the lambda
final WrappedInt min = new WrappedInt(Math.min(minDeletionTime(info), minDeletionTime(rowDeletion)));
long min = Math.min(minDeletionTime(info), minDeletionTime(rowDeletion));
BTree.<ColumnData>apply(btree, cd -> min.set( Math.min(min.get(), minDeletionTime(cd)) ), cd -> min.get() == Integer.MIN_VALUE, false);
return min.get();
min = BTree.<ColumnData>accumulate(btree, (cd, l) -> {
int m = Math.min((int) l, minDeletionTime(cd));
return m != Integer.MIN_VALUE ? m : Long.MAX_VALUE;
}, min);
if (min == Long.MAX_VALUE)
return Integer.MIN_VALUE;
return Ints.checkedCast(min);
}
public Clustering clustering()
@ -334,33 +366,19 @@ public class BTreeRow extends AbstractRow
public boolean hasComplex()
{
// We start by the end cause we know complex columns sort after the simple ones
ColumnData cd = Iterables.getFirst(BTree.<ColumnData>iterable(btree, BTree.Dir.DESC), null);
return cd != null && cd.column.isComplex();
if (BTree.isEmpty(btree))
return false;
int size = BTree.size(btree);
ColumnData last = BTree.findByIndex(btree, size - 1);
return last.column.isComplex();
}
public boolean hasComplexDeletion()
{
final WrappedBoolean result = new WrappedBoolean(false);
// We start by the end cause we know complex columns sort before simple ones
apply(c -> {}, cd -> {
if (cd.column.isSimple())
{
result.set(false);
return true;
}
if (!((ComplexColumnData) cd).complexDeletion().isLive())
{
result.set(true);
return true;
}
return false;
}, true);
return result.get();
long result = accumulate((cd, v) -> ((ComplexColumnData) cd).complexDeletion().isLive() ? 0 : Long.MAX_VALUE,
COLUMN_COMPARATOR, isStatic() ? FIRST_COMPLEX_STATIC : FIRST_COMPLEX_REGULAR, 0L);
return result == Long.MAX_VALUE;
}
public Row markCounterLocalToBeCleared()
@ -375,6 +393,15 @@ public class BTreeRow extends AbstractRow
return nowInSec >= minLocalDeletionTime;
}
public boolean hasInvalidDeletions()
{
if (primaryKeyLivenessInfo().isExpiring() && (primaryKeyLivenessInfo().ttl() < 0 || primaryKeyLivenessInfo().localExpirationTime() < 0))
return true;
if (!deletion().time().validate())
return true;
return accumulate((cd, v) -> cd.hasInvalidDeletions() ? Long.MAX_VALUE : v, 0) != 0;
}
/**
* Returns a copy of the row where all timestamps for live data have replaced by {@code newTimestamp} and
* all deletion timestamp by {@code newTimestamp - 1}.
@ -440,9 +467,7 @@ public class BTreeRow extends AbstractRow
+ primaryKeyLivenessInfo.dataSize()
+ deletion.dataSize();
for (ColumnData cd : this)
dataSize += cd.dataSize();
return dataSize;
return Ints.checkedCast(accumulate((cd, v) -> v + cd.dataSize(), dataSize));
}
public long unsharedHeapSizeExcludingData()
@ -451,9 +476,7 @@ public class BTreeRow extends AbstractRow
+ clustering.unsharedHeapSizeExcludingData()
+ BTree.sizeOfStructureOnHeap(btree);
for (ColumnData cd : this)
heapSize += cd.unsharedHeapSizeExcludingData();
return heapSize;
return accumulate((cd, v) -> v + cd.unsharedHeapSizeExcludingData(), heapSize);
}
public static Row.Builder sortedBuilder()

2
src/java/org/apache/cassandra/db/rows/Cell.java
View File

@ -214,7 +214,7 @@ public abstract class Cell extends ColumnData
header.getType(column).writeValue(cell.value(), out);
}
public Cell deserialize(DataInputPlus in, LivenessInfo rowLiveness, ColumnMetadata column, SerializationHeader header, SerializationHelper helper) throws IOException
public Cell deserialize(DataInputPlus in, LivenessInfo rowLiveness, ColumnMetadata column, SerializationHeader header, DeserializationHelper helper) throws IOException
{
int flags = in.readUnsignedByte();
boolean hasValue = (flags & HAS_EMPTY_VALUE_MASK) == 0;

5
src/java/org/apache/cassandra/db/rows/ColumnData.java
View File

@ -78,6 +78,11 @@ public abstract class ColumnData
*/
public abstract void digest(Digest digest);
public static void digest(Digest digest, ColumnData cd)
{
cd.digest(digest);
}
/**
* Returns a copy of the data where all timestamps for live data have replaced by {@code newTimestamp} and
* all deletion timestamp by {@code newTimestamp - 1}.

18
src/java/org/apache/cassandra/db/rows/ComplexColumnData.java
View File

@ -30,8 +30,11 @@ import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.marshal.ByteType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.DroppedColumn;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.btree.BTree;
@ -60,11 +63,6 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell>
this.complexDeletion = complexDeletion;
}
public boolean hasCells()
{
return !BTree.isEmpty(cells);
}
public int cellsCount()
{
return BTree.size(cells);
@ -106,6 +104,16 @@ public class ComplexColumnData extends ColumnData implements Iterable<Cell>
return BTree.iterator(cells, BTree.Dir.DESC);
}
public long accumulate(LongAccumulator<Cell> accumulator, long initialValue)
{
return BTree.accumulate(cells, accumulator, initialValue);
}
public <A> long accumulate(BiLongAccumulator<A, Cell> accumulator, A arg, long initialValue)
{
return BTree.accumulate(cells, accumulator, arg, initialValue);
}
public int dataSize()
{
int size = complexDeletion.dataSize();

149
src/java/org/apache/cassandra/db/rows/DeserializationHelper.java Normal file
View File

@ -0,0 +1,149 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.schema.DroppedColumn;
public class DeserializationHelper
{
/**
* Flag affecting deserialization behavior (this only affect counters in practice).
* - LOCAL: for deserialization of local data (Expired columns are
* converted to tombstones (to gain disk space)).
* - FROM_REMOTE: for deserialization of data received from remote hosts
* (Expired columns are converted to tombstone and counters have
* their delta cleared)
* - PRESERVE_SIZE: used when no transformation must be performed, i.e,
* when we must ensure that deserializing and reserializing the
* result yield the exact same bytes. Streaming uses this.
*/
public enum Flag
{
LOCAL, FROM_REMOTE, PRESERVE_SIZE
}
private final Flag flag;
public final int version;
private final ColumnFilter columnsToFetch;
private ColumnFilter.Tester tester;
private final boolean hasDroppedColumns;
private final Map<ByteBuffer, DroppedColumn> droppedColumns;
private DroppedColumn currentDroppedComplex;
public DeserializationHelper(TableMetadata metadata, int version, Flag flag, ColumnFilter columnsToFetch)
{
this.flag = flag;
this.version = version;
this.columnsToFetch = columnsToFetch;
this.droppedColumns = metadata.droppedColumns;
this.hasDroppedColumns = droppedColumns.size() > 0;
}
public DeserializationHelper(TableMetadata metadata, int version, Flag flag)
{
this(metadata, version, flag, null);
}
public boolean includes(ColumnMetadata column)
{
return columnsToFetch == null || columnsToFetch.fetches(column);
}
public boolean includes(Cell cell, LivenessInfo rowLiveness)
{
if (columnsToFetch == null)
return true;
// During queries, some columns are included even though they are not queried by the user because
// we always need to distinguish between having a row (with potentially only null values) and not
// having a row at all (see #CASSANDRA-7085 for background). In the case where the column is not
// actually requested by the user however (canSkipValue), we can skip the full cell if the cell
// timestamp is lower than the row one, because in that case, the row timestamp is enough proof
// of the liveness of the row. Otherwise, we'll only be able to skip the values of those cells.
ColumnMetadata column = cell.column();
if (column.isComplex())
{
if (!includes(cell.path()))
return false;
return !canSkipValue(cell.path()) || cell.timestamp() >= rowLiveness.timestamp();
}
else
{
return columnsToFetch.fetchedColumnIsQueried(column) || cell.timestamp() >= rowLiveness.timestamp();
}
}
public boolean includes(CellPath path)
{
return path == null || tester == null || tester.fetches(path);
}
public boolean canSkipValue(ColumnMetadata column)
{
return columnsToFetch != null && !columnsToFetch.fetchedColumnIsQueried(column);
}
public boolean canSkipValue(CellPath path)
{
return path != null && tester != null && !tester.fetchedCellIsQueried(path);
}
public void startOfComplexColumn(ColumnMetadata column)
{
this.tester = columnsToFetch == null ? null : columnsToFetch.newTester(column);
this.currentDroppedComplex = droppedColumns.get(column.name.bytes);
}
public void endOfComplexColumn()
{
this.tester = null;
}
public boolean isDropped(Cell cell, boolean isComplex)
{
if (!hasDroppedColumns)
return false;
DroppedColumn dropped = isComplex ? currentDroppedComplex : droppedColumns.get(cell.column().name.bytes);
return dropped != null && cell.timestamp() <= dropped.droppedTime;
}
public boolean isDroppedComplexDeletion(DeletionTime complexDeletion)
{
return currentDroppedComplex != null && complexDeletion.markedForDeleteAt() <= currentDroppedComplex.droppedTime;
}
public ByteBuffer maybeClearCounterValue(ByteBuffer value)
{
return flag == Flag.FROM_REMOTE || (flag == Flag.LOCAL && CounterContext.instance().shouldClearLocal(value))
? CounterContext.instance().clearAllLocal(value)
: value;
}
}

23
src/java/org/apache/cassandra/db/rows/Row.java
View File

@ -18,15 +18,16 @@
package org.apache.cassandra.db.rows;
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import com.google.common.base.Predicate;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree;
@ -282,12 +283,24 @@ public interface Row extends Unfiltered, Iterable<ColumnData>
/**
* Apply a function to every column in a row
*/
public void apply(Consumer<ColumnData> function, boolean reverse);
public void apply(Consumer<ColumnData> function);
/**
* Apply a funtion to every column in a row until a stop condition is reached
* Apply a function to every column in a row
*/
public void apply(Consumer<ColumnData> function, Predicate<ColumnData> stopCondition, boolean reverse);
public <A> void apply(BiConsumer<A, ColumnData> function, A arg);
/**
* Apply an accumulation funtion to every column in a row
*/
public long accumulate(LongAccumulator<ColumnData> accumulator, long initialValue);
public long accumulate(LongAccumulator<ColumnData> accumulator, Comparator<ColumnData> comparator, ColumnData from, long initialValue);
public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, long initialValue);
public <A> long accumulate(BiLongAccumulator<A, ColumnData> accumulator, A arg, Comparator<ColumnData> comparator, ColumnData from, long initialValue);
/**
* A row deletion/tombstone.

72
src/java/org/apache/cassandra/db/rows/Rows.java
View File

@ -27,7 +27,6 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.PartitionStatisticsCollector;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.WrappedInt;
/**
* Static utilities to work on Row objects.
@ -75,6 +74,46 @@ public abstract class Rows
return new SimpleBuilders.RowBuilder(metadata, clusteringValues);
}
private static class StatsAccumulation
{
private static final long COLUMN_INCR = 1L << 32;
private static final long CELL_INCR = 1L;
private static long accumulateOnCell(PartitionStatisticsCollector collector, Cell cell, long l)
{
Cells.collectStats(cell, collector);
return l + CELL_INCR;
}
private static long accumulateOnColumnData(PartitionStatisticsCollector collector, ColumnData cd, long l)
{
if (cd.column().isSimple())
{
l = accumulateOnCell(collector, (Cell) cd, l) + COLUMN_INCR;
}
else
{
ComplexColumnData complexData = (ComplexColumnData)cd;
collector.update(complexData.complexDeletion());
int startingCells = unpackCellCount(l);
l = complexData.accumulate(StatsAccumulation::accumulateOnCell, collector, l);
if (unpackCellCount(l) > startingCells)
l += COLUMN_INCR;
}
return l;
}
private static int unpackCellCount(long v)
{
return (int) (v & 0xFFFFFFFFL);
}
private static int unpackColumnCount(long v)
{
return (int) (v >>> 32);
}
}
/**
* Collect statistics on a given row.
*
@ -89,35 +128,10 @@ public abstract class Rows
collector.update(row.primaryKeyLivenessInfo());
collector.update(row.deletion().time());
//we have to wrap these for the lambda
final WrappedInt columnCount = new WrappedInt(0);
final WrappedInt cellCount = new WrappedInt(0);
long result = row.accumulate(StatsAccumulation::accumulateOnColumnData, collector, 0);
row.apply(cd -> {
if (cd.column().isSimple())
{
columnCount.increment();
cellCount.increment();
Cells.collectStats((Cell) cd, collector);
}
else
{
ComplexColumnData complexData = (ComplexColumnData)cd;
collector.update(complexData.complexDeletion());
if (complexData.hasCells())
{
columnCount.increment();
for (Cell cell : complexData)
{
cellCount.increment();
Cells.collectStats(cell, collector);
}
}
}
}, false);
collector.updateColumnSetPerRow(columnCount.get());
return cellCount.get();
collector.updateColumnSetPerRow(StatsAccumulation.unpackColumnCount(result));
return StatsAccumulation.unpackCellCount(result);
}
/**

136
src/java/org/apache/cassandra/db/rows/SerializationHelper.java
View File

@ -15,135 +15,43 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.db.rows;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.SerializationHeader;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.schema.DroppedColumn;
import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTreeSearchIterator;
public class SerializationHelper
{
/**
* Flag affecting deserialization behavior (this only affect counters in practice).
* - LOCAL: for deserialization of local data (Expired columns are
* converted to tombstones (to gain disk space)).
* - FROM_REMOTE: for deserialization of data received from remote hosts
* (Expired columns are converted to tombstone and counters have
* their delta cleared)
* - PRESERVE_SIZE: used when no transformation must be performed, i.e,
* when we must ensure that deserializing and reserializing the
* result yield the exact same bytes. Streaming uses this.
*/
public enum Flag
public final SerializationHeader header;
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> statics = null;
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> regulars = null;
public SerializationHelper(SerializationHeader header)
{
LOCAL, FROM_REMOTE, PRESERVE_SIZE
this.header = header;
}
private final Flag flag;
public final int version;
private final ColumnFilter columnsToFetch;
private ColumnFilter.Tester tester;
private final boolean hasDroppedColumns;
private final Map<ByteBuffer, DroppedColumn> droppedColumns;
private DroppedColumn currentDroppedComplex;
public SerializationHelper(TableMetadata metadata, int version, Flag flag, ColumnFilter columnsToFetch)
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> statics()
{
this.flag = flag;
this.version = version;
this.columnsToFetch = columnsToFetch;
this.droppedColumns = metadata.droppedColumns;
this.hasDroppedColumns = droppedColumns.size() > 0;
if (statics == null)
statics = header.columns().statics.iterator();
return statics;
}
public SerializationHelper(TableMetadata metadata, int version, Flag flag)
private BTreeSearchIterator<ColumnMetadata, ColumnMetadata> regulars()
{
this(metadata, version, flag, null);
if (regulars == null)
regulars = header.columns().regulars.iterator();
return regulars;
}
public boolean includes(ColumnMetadata column)
public SearchIterator<ColumnMetadata, ColumnMetadata> iterator(boolean isStatic)
{
return columnsToFetch == null || columnsToFetch.fetches(column);
}
public boolean includes(Cell cell, LivenessInfo rowLiveness)
{
if (columnsToFetch == null)
return true;
// During queries, some columns are included even though they are not queried by the user because
// we always need to distinguish between having a row (with potentially only null values) and not
// having a row at all (see #CASSANDRA-7085 for background). In the case where the column is not
// actually requested by the user however (canSkipValue), we can skip the full cell if the cell
// timestamp is lower than the row one, because in that case, the row timestamp is enough proof
// of the liveness of the row. Otherwise, we'll only be able to skip the values of those cells.
ColumnMetadata column = cell.column();
if (column.isComplex())
{
if (!includes(cell.path()))
return false;
return !canSkipValue(cell.path()) || cell.timestamp() >= rowLiveness.timestamp();
}
else
{
return columnsToFetch.fetchedColumnIsQueried(column) || cell.timestamp() >= rowLiveness.timestamp();
}
}
public boolean includes(CellPath path)
{
return path == null || tester == null || tester.fetches(path);
}
public boolean canSkipValue(ColumnMetadata column)
{
return columnsToFetch != null && !columnsToFetch.fetchedColumnIsQueried(column);
}
public boolean canSkipValue(CellPath path)
{
return path != null && tester != null && !tester.fetchedCellIsQueried(path);
}
public void startOfComplexColumn(ColumnMetadata column)
{
this.tester = columnsToFetch == null ? null : columnsToFetch.newTester(column);
this.currentDroppedComplex = droppedColumns.get(column.name.bytes);
}
public void endOfComplexColumn()
{
this.tester = null;
}
public boolean isDropped(Cell cell, boolean isComplex)
{
if (!hasDroppedColumns)
return false;
DroppedColumn dropped = isComplex ? currentDroppedComplex : droppedColumns.get(cell.column().name.bytes);
return dropped != null && cell.timestamp() <= dropped.droppedTime;
}
public boolean isDroppedComplexDeletion(DeletionTime complexDeletion)
{
return currentDroppedComplex != null && complexDeletion.markedForDeleteAt() <= currentDroppedComplex.droppedTime;
}
public ByteBuffer maybeClearCounterValue(ByteBuffer value)
{
return flag == Flag.FROM_REMOTE || (flag == Flag.LOCAL && CounterContext.instance().shouldClearLocal(value))
? CounterContext.instance().clearAllLocal(value)
: value;
BTreeSearchIterator<ColumnMetadata, ColumnMetadata> iterator = isStatic ? statics() : regulars();
iterator.rewind();
return iterator;
}
}

21
src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java
View File

@ -123,18 +123,19 @@ public class UnfilteredRowIteratorSerializer
out.writeByte((byte)flags);
SerializationHeader.serializer.serializeForMessaging(header, selection, out, hasStatic);
SerializationHelper helper = new SerializationHelper(header);
if (!partitionDeletion.isLive())
header.writeDeletionTime(partitionDeletion, out);
if (hasStatic)
UnfilteredSerializer.serializer.serialize(staticRow, header, out, version);
UnfilteredSerializer.serializer.serialize(staticRow, helper, out, version);
if (rowEstimate >= 0)
out.writeUnsignedVInt(rowEstimate);
while (iterator.hasNext())
UnfilteredSerializer.serializer.serialize(iterator.next(), header, out, version);
UnfilteredSerializer.serializer.serialize(iterator.next(), helper, out, version);
UnfilteredSerializer.serializer.writeEndOfPartition(out);
}
@ -147,6 +148,8 @@ public class UnfilteredRowIteratorSerializer
iterator.columns(),
iterator.stats());
SerializationHelper helper = new SerializationHelper(header);
assert rowEstimate >= 0;
long size = ByteBufferUtil.serializedSizeWithVIntLength(iterator.partitionKey().getKey())
@ -165,19 +168,19 @@ public class UnfilteredRowIteratorSerializer
size += header.deletionTimeSerializedSize(partitionDeletion);
if (hasStatic)
size += UnfilteredSerializer.serializer.serializedSize(staticRow, header, version);
size += UnfilteredSerializer.serializer.serializedSize(staticRow, helper, version);
if (rowEstimate >= 0)
size += TypeSizes.sizeofUnsignedVInt(rowEstimate);
while (iterator.hasNext())
size += UnfilteredSerializer.serializer.serializedSize(iterator.next(), header, version);
size += UnfilteredSerializer.serializer.serializedSize(iterator.next(), helper, version);
size += UnfilteredSerializer.serializer.serializedSizeEndOfPartition();
return size;
}
public Header deserializeHeader(TableMetadata metadata, ColumnFilter selection, DataInputPlus in, int version, SerializationHelper.Flag flag) throws IOException
public Header deserializeHeader(TableMetadata metadata, ColumnFilter selection, DataInputPlus in, int version, DeserializationHelper.Flag flag) throws IOException
{
DecoratedKey key = metadata.partitioner.decorateKey(ByteBufferUtil.readWithVIntLength(in));
int flags = in.readUnsignedByte();
@ -198,18 +201,18 @@ public class UnfilteredRowIteratorSerializer
Row staticRow = Rows.EMPTY_STATIC_ROW;
if (hasStatic)
staticRow = UnfilteredSerializer.serializer.deserializeStaticRow(in, header, new SerializationHelper(metadata, version, flag));
staticRow = UnfilteredSerializer.serializer.deserializeStaticRow(in, header, new DeserializationHelper(metadata, version, flag));
int rowEstimate = hasRowEstimate ? (int)in.readUnsignedVInt() : -1;
return new Header(header, key, isReversed, false, partitionDeletion, staticRow, rowEstimate);
}
public UnfilteredRowIterator deserialize(DataInputPlus in, int version, TableMetadata metadata, SerializationHelper.Flag flag, Header header) throws IOException
public UnfilteredRowIterator deserialize(DataInputPlus in, int version, TableMetadata metadata, DeserializationHelper.Flag flag, Header header) throws IOException
{
if (header.isEmpty)
return EmptyIterators.unfilteredRow(metadata, header.key, header.isReversed);
final SerializationHelper helper = new SerializationHelper(metadata, version, flag);
final DeserializationHelper helper = new DeserializationHelper(metadata, version, flag);
final SerializationHeader sHeader = header.sHeader;
return new AbstractUnfilteredRowIterator(metadata, header.key, header.partitionDeletion, sHeader.columns(), header.staticRow, header.isReversed, sHeader.stats())
{
@ -230,7 +233,7 @@ public class UnfilteredRowIteratorSerializer
};
}
public UnfilteredRowIterator deserialize(DataInputPlus in, int version, TableMetadata metadata, ColumnFilter selection, SerializationHelper.Flag flag) throws IOException
public UnfilteredRowIterator deserialize(DataInputPlus in, int version, TableMetadata metadata, ColumnFilter selection, DeserializationHelper.Flag flag) throws IOException
{
return deserialize(in, version, metadata, flag, deserializeHeader(metadata, selection, in, version, flag));
}

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

@ -19,8 +19,6 @@ package org.apache.cassandra.db.rows;
import java.io.IOException;
import com.google.common.collect.Collections2;
import net.nicoulaj.compilecommand.annotations.Inline;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.db.*;
@ -118,40 +116,41 @@ public class UnfilteredSerializer
@Deprecated
private final static int HAS_SHADOWABLE_DELETION = 0x02; // Whether the row deletion is shadowable. If there is no extended flag (or no row deletion), the deletion is assumed not shadowable.
public void serialize(Unfiltered unfiltered, SerializationHeader header, DataOutputPlus out, int version)
public void serialize(Unfiltered unfiltered, SerializationHelper helper, DataOutputPlus out, int version)
throws IOException
{
assert !header.isForSSTable();
serialize(unfiltered, header, out, 0, version);
assert !helper.header.isForSSTable();
serialize(unfiltered, helper, out, 0, version);
}
public void serialize(Unfiltered unfiltered, SerializationHeader header, DataOutputPlus out, long previousUnfilteredSize, int version)
public void serialize(Unfiltered unfiltered, SerializationHelper helper, DataOutputPlus out, long previousUnfilteredSize, int version)
throws IOException
{
if (unfiltered.kind() == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER)
{
serialize((RangeTombstoneMarker) unfiltered, header, out, previousUnfilteredSize, version);
serialize((RangeTombstoneMarker) unfiltered, helper, out, previousUnfilteredSize, version);
}
else
{
serialize((Row) unfiltered, header, out, previousUnfilteredSize, version);
serialize((Row) unfiltered, helper, out, previousUnfilteredSize, version);
}
}
public void serializeStaticRow(Row row, SerializationHeader header, DataOutputPlus out, int version)
public void serializeStaticRow(Row row, SerializationHelper helper, DataOutputPlus out, int version)
throws IOException
{
assert row.isStatic();
serialize(row, header, out, 0, version);
serialize(row, helper, out, 0, version);
}
private void serialize(Row row, SerializationHeader header, DataOutputPlus out, long previousUnfilteredSize, int version)
private void serialize(Row row, SerializationHelper helper, DataOutputPlus out, long previousUnfilteredSize, int version)
throws IOException
{
int flags = 0;
int extendedFlags = 0;
boolean isStatic = row.isStatic();
SerializationHeader header = helper.header;
Columns headerColumns = header.columns(isStatic);
LivenessInfo pkLiveness = row.primaryKeyLivenessInfo();
Row.Deletion deletion = row.deletion();
@ -191,7 +190,7 @@ public class UnfilteredSerializer
{
try (DataOutputBuffer dob = DataOutputBuffer.scratchBuffer.get())
{
serializeRowBody(row, flags, header, dob);
serializeRowBody(row, flags, helper, dob);
out.writeUnsignedVInt(dob.position() + TypeSizes.sizeofUnsignedVInt(previousUnfilteredSize));
// We write the size of the previous unfiltered to make reverse queries more efficient (and simpler).
@ -202,16 +201,17 @@ public class UnfilteredSerializer
}
else
{
serializeRowBody(row, flags, header, out);
serializeRowBody(row, flags, helper, out);
}
}
@Inline
private void serializeRowBody(Row row, int flags, SerializationHeader header, DataOutputPlus out)
private void serializeRowBody(Row row, int flags, SerializationHelper helper, DataOutputPlus out)
throws IOException
{
boolean isStatic = row.isStatic();
SerializationHeader header = helper.header;
Columns headerColumns = header.columns(isStatic);
LivenessInfo pkLiveness = row.primaryKeyLivenessInfo();
Row.Deletion deletion = row.deletion();
@ -229,7 +229,7 @@ public class UnfilteredSerializer
if ((flags & HAS_ALL_COLUMNS) == 0)
Columns.serializer.serializeSubset(row.columns(), headerColumns, out);
SearchIterator<ColumnMetadata, ColumnMetadata> si = headerColumns.iterator();
SearchIterator<ColumnMetadata, ColumnMetadata> si = helper.iterator(isStatic);
try
{
@ -253,7 +253,7 @@ public class UnfilteredSerializer
{
throw new WrappedException(e);
}
}, false);
});
}
catch (WrappedException e)
{
@ -275,9 +275,10 @@ public class UnfilteredSerializer
Cell.serializer.serialize(cell, column, out, rowLiveness, header);
}
private void serialize(RangeTombstoneMarker marker, SerializationHeader header, DataOutputPlus out, long previousUnfilteredSize, int version)
private void serialize(RangeTombstoneMarker marker, SerializationHelper helper, DataOutputPlus out, long previousUnfilteredSize, int version)
throws IOException
{
SerializationHeader header = helper.header;
out.writeByte((byte)IS_MARKER);
ClusteringBoundOrBoundary.serializer.serialize(marker.clustering(), out, version, header.clusteringTypes());
@ -299,20 +300,20 @@ public class UnfilteredSerializer
}
}
public long serializedSize(Unfiltered unfiltered, SerializationHeader header, int version)
public long serializedSize(Unfiltered unfiltered, SerializationHelper helper, int version)
{
assert !header.isForSSTable();
return serializedSize(unfiltered, header, 0, version);
assert !helper.header.isForSSTable();
return serializedSize(unfiltered, helper, 0, version);
}
public long serializedSize(Unfiltered unfiltered, SerializationHeader header, long previousUnfilteredSize,int version)
public long serializedSize(Unfiltered unfiltered, SerializationHelper helper, long previousUnfilteredSize,int version)
{
return unfiltered.kind() == Unfiltered.Kind.RANGE_TOMBSTONE_MARKER
? serializedSize((RangeTombstoneMarker) unfiltered, header, previousUnfilteredSize, version)
: serializedSize((Row) unfiltered, header, previousUnfilteredSize, version);
? serializedSize((RangeTombstoneMarker) unfiltered, helper, previousUnfilteredSize, version)
: serializedSize((Row) unfiltered, helper, previousUnfilteredSize, version);
}
private long serializedSize(Row row, SerializationHeader header, long previousUnfilteredSize, int version)
private long serializedSize(Row row, SerializationHelper helper, long previousUnfilteredSize, int version)
{
long size = 1; // flags
@ -320,15 +321,16 @@ public class UnfilteredSerializer
size += 1; // extended flags
if (!row.isStatic())
size += Clustering.serializer.serializedSize(row.clustering(), version, header.clusteringTypes());
size += Clustering.serializer.serializedSize(row.clustering(), version, helper.header.clusteringTypes());
return size + serializedRowBodySize(row, header, previousUnfilteredSize, version);
return size + serializedRowBodySize(row, helper, previousUnfilteredSize, version);
}
private long serializedRowBodySize(Row row, SerializationHeader header, long previousUnfilteredSize, int version)
private long serializedRowBodySize(Row row, SerializationHelper helper, long previousUnfilteredSize, int version)
{
long size = 0;
SerializationHeader header = helper.header;
if (header.isForSSTable())
size += TypeSizes.sizeofUnsignedVInt(previousUnfilteredSize);
@ -352,19 +354,16 @@ public class UnfilteredSerializer
if (!hasAllColumns)
size += Columns.serializer.serializedSubsetSize(row.columns(), header.columns(isStatic));
SearchIterator<ColumnMetadata, ColumnMetadata> si = headerColumns.iterator();
for (ColumnData data : row)
{
SearchIterator<ColumnMetadata, ColumnMetadata> si = helper.iterator(isStatic);
return row.accumulate((data, v) -> {
ColumnMetadata column = si.next(data.column());
assert column != null;
if (data.column.isSimple())
size += Cell.serializer.serializedSize((Cell) data, column, pkLiveness, header);
return v + Cell.serializer.serializedSize((Cell) data, column, pkLiveness, header);
else
size += sizeOfComplexColumn((ComplexColumnData) data, column, hasComplexDeletion, pkLiveness, header);
}
return size;
return v + sizeOfComplexColumn((ComplexColumnData) data, column, hasComplexDeletion, pkLiveness, header);
}, size);
}
private long sizeOfComplexColumn(ComplexColumnData data, ColumnMetadata column, boolean hasComplexDeletion, LivenessInfo rowLiveness, SerializationHeader header)
@ -381,12 +380,12 @@ public class UnfilteredSerializer
return size;
}
private long serializedSize(RangeTombstoneMarker marker, SerializationHeader header, long previousUnfilteredSize, int version)
private long serializedSize(RangeTombstoneMarker marker, SerializationHelper helper, long previousUnfilteredSize, int version)
{
assert !header.isForSSTable();
assert !helper.header.isForSSTable();
return 1 // flags
+ ClusteringBoundOrBoundary.serializer.serializedSize(marker.clustering(), version, header.clusteringTypes())
+ serializedMarkerBodySize(marker, header, previousUnfilteredSize, version);
+ ClusteringBoundOrBoundary.serializer.serializedSize(marker.clustering(), version, helper.header.clusteringTypes())
+ serializedMarkerBodySize(marker, helper.header, previousUnfilteredSize, version);
}
private long serializedMarkerBodySize(RangeTombstoneMarker marker, SerializationHeader header, long previousUnfilteredSize, int version)
@ -428,7 +427,7 @@ public class UnfilteredSerializer
* @return the deserialized {@link Unfiltered} or {@code null} if we've read the end of a partition. This method is
* guaranteed to never return empty rows.
*/
public Unfiltered deserialize(DataInputPlus in, SerializationHeader header, SerializationHelper helper, Row.Builder builder)
public Unfiltered deserialize(DataInputPlus in, SerializationHeader header, DeserializationHelper helper, Row.Builder builder)
throws IOException
{
while (true)
@ -453,7 +452,7 @@ public class UnfilteredSerializer
* But as {@link UnfilteredRowIterator} should not return empty
* rows, this mean consumer of this method should make sure to skip said empty rows.
*/
private Unfiltered deserializeOne(DataInputPlus in, SerializationHeader header, SerializationHelper helper, Row.Builder builder)
private Unfiltered deserializeOne(DataInputPlus in, SerializationHeader header, DeserializationHelper helper, Row.Builder builder)
throws IOException
{
// It wouldn't be wrong per-se to use an unsorted builder, but it would be inefficient so make sure we don't do it by mistake
@ -481,7 +480,7 @@ public class UnfilteredSerializer
}
}
public Unfiltered deserializeTombstonesOnly(FileDataInput in, SerializationHeader header, SerializationHelper helper)
public Unfiltered deserializeTombstonesOnly(FileDataInput in, SerializationHeader header, DeserializationHelper helper)
throws IOException
{
while (true)
@ -533,7 +532,7 @@ public class UnfilteredSerializer
}
}
public Row deserializeStaticRow(DataInputPlus in, SerializationHeader header, SerializationHelper helper)
public Row deserializeStaticRow(DataInputPlus in, SerializationHeader header, DeserializationHelper helper)
throws IOException
{
int flags = in.readUnsignedByte();
@ -561,7 +560,7 @@ public class UnfilteredSerializer
public Row deserializeRowBody(DataInputPlus in,
SerializationHeader header,
SerializationHelper helper,
DeserializationHelper helper,
int flags,
int extendedFlags,
Row.Builder builder)
@ -614,7 +613,7 @@ public class UnfilteredSerializer
{
throw new WrappedException(e);
}
}, false);
});
}
catch (WrappedException e)
{
@ -636,7 +635,7 @@ public class UnfilteredSerializer
}
}
private void readSimpleColumn(ColumnMetadata column, DataInputPlus in, SerializationHeader header, SerializationHelper helper, Row.Builder builder, LivenessInfo rowLiveness)
private void readSimpleColumn(ColumnMetadata column, DataInputPlus in, SerializationHeader header, DeserializationHelper helper, Row.Builder builder, LivenessInfo rowLiveness)
throws IOException
{
if (helper.includes(column))
@ -651,7 +650,7 @@ public class UnfilteredSerializer
}
}
private void readComplexColumn(ColumnMetadata column, DataInputPlus in, SerializationHeader header, SerializationHelper helper, boolean hasComplexDeletion, Row.Builder builder, LivenessInfo rowLiveness)
private void readComplexColumn(ColumnMetadata column, DataInputPlus in, SerializationHeader header, DeserializationHelper helper, boolean hasComplexDeletion, Row.Builder builder, LivenessInfo rowLiveness)
throws IOException
{
if (helper.includes(column))
@ -686,7 +685,7 @@ public class UnfilteredSerializer
in.skipBytesFully(rowSize);
}
public void skipStaticRow(DataInputPlus in, SerializationHeader header, SerializationHelper helper) throws IOException
public void skipStaticRow(DataInputPlus in, SerializationHeader header, DeserializationHelper helper) throws IOException
{
int flags = in.readUnsignedByte();
assert !isEndOfPartition(flags) && kind(flags) == Unfiltered.Kind.ROW && isExtended(flags) : "Flags is " + flags;

4
src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java
View File

@ -181,7 +181,7 @@ public class CassandraStreamReader implements IStreamReader
private final TableMetadata metadata;
private final DataInputPlus in;
private final SerializationHeader header;
private final SerializationHelper helper;
private final DeserializationHelper helper;
private DecoratedKey key;
private DeletionTime partitionLevelDeletion;
@ -193,7 +193,7 @@ public class CassandraStreamReader implements IStreamReader
{
this.metadata = metadata;
this.in = in;
this.helper = new SerializationHelper(metadata, version.correspondingMessagingVersion(), SerializationHelper.Flag.PRESERVE_SIZE);
this.helper = new DeserializationHelper(metadata, version.correspondingMessagingVersion(), DeserializationHelper.Flag.PRESERVE_SIZE);
this.header = header;
}

4
src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java
View File

@ -57,7 +57,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
DeletionTime partitionLevelDeletion = DeletionTime.serializer.deserialize(file);
if (!partitionLevelDeletion.validate())
UnfilteredValidation.handleInvalid(sstable.metadata(), key, sstable, "partitionLevelDeletion="+partitionLevelDeletion.toString());
SerializationHelper helper = new SerializationHelper(sstable.metadata(), sstable.descriptor.version.correspondingMessagingVersion(), SerializationHelper.Flag.LOCAL);
DeserializationHelper helper = new DeserializationHelper(sstable.metadata(), sstable.descriptor.version.correspondingMessagingVersion(), DeserializationHelper.Flag.LOCAL);
SSTableSimpleIterator iterator = SSTableSimpleIterator.create(sstable.metadata(), file, sstable.header, helper, partitionLevelDeletion);
return new SSTableIdentityIterator(sstable, key, partitionLevelDeletion, file.getPath(), iterator);
}
@ -76,7 +76,7 @@ public class SSTableIdentityIterator implements Comparable<SSTableIdentityIterat
dfile.seek(indexEntry.position);
ByteBufferUtil.skipShortLength(dfile); // Skip partition key
DeletionTime partitionLevelDeletion = DeletionTime.serializer.deserialize(dfile);
SerializationHelper helper = new SerializationHelper(sstable.metadata(), sstable.descriptor.version.correspondingMessagingVersion(), SerializationHelper.Flag.LOCAL);
DeserializationHelper helper = new DeserializationHelper(sstable.metadata(), sstable.descriptor.version.correspondingMessagingVersion(), DeserializationHelper.Flag.LOCAL);
SSTableSimpleIterator iterator = tombstoneOnly
? SSTableSimpleIterator.createTombstoneOnly(sstable.metadata(), dfile, sstable.header, helper, partitionLevelDeletion)
: SSTableSimpleIterator.create(sstable.metadata(), dfile, sstable.header, helper, partitionLevelDeletion);

12
src/java/org/apache/cassandra/io/sstable/SSTableSimpleIterator.java
View File

@ -38,21 +38,21 @@ public abstract class SSTableSimpleIterator extends AbstractIterator<Unfiltered>
{
final TableMetadata metadata;
protected final DataInputPlus in;
protected final SerializationHelper helper;
protected final DeserializationHelper helper;
private SSTableSimpleIterator(TableMetadata metadata, DataInputPlus in, SerializationHelper helper)
private SSTableSimpleIterator(TableMetadata metadata, DataInputPlus in, DeserializationHelper helper)
{
this.metadata = metadata;
this.in = in;
this.helper = helper;
}
public static SSTableSimpleIterator create(TableMetadata metadata, DataInputPlus in, SerializationHeader header, SerializationHelper helper, DeletionTime partitionDeletion)
public static SSTableSimpleIterator create(TableMetadata metadata, DataInputPlus in, SerializationHeader header, DeserializationHelper helper, DeletionTime partitionDeletion)
{
return new CurrentFormatIterator(metadata, in, header, helper);
}
public static SSTableSimpleIterator createTombstoneOnly(TableMetadata metadata, DataInputPlus in, SerializationHeader header, SerializationHelper helper, DeletionTime partitionDeletion)
public static SSTableSimpleIterator createTombstoneOnly(TableMetadata metadata, DataInputPlus in, SerializationHeader header, DeserializationHelper helper, DeletionTime partitionDeletion)
{
return new CurrentFormatTombstoneIterator(metadata, in, header, helper);
}
@ -65,7 +65,7 @@ public abstract class SSTableSimpleIterator extends AbstractIterator<Unfiltered>
private final Row.Builder builder;
private CurrentFormatIterator(TableMetadata metadata, DataInputPlus in, SerializationHeader header, SerializationHelper helper)
private CurrentFormatIterator(TableMetadata metadata, DataInputPlus in, SerializationHeader header, DeserializationHelper helper)
{
super(metadata, in, helper);
this.header = header;
@ -95,7 +95,7 @@ public abstract class SSTableSimpleIterator extends AbstractIterator<Unfiltered>
{
private final SerializationHeader header;
private CurrentFormatTombstoneIterator(TableMetadata metadata, DataInputPlus in, SerializationHeader header, SerializationHelper helper)
private CurrentFormatTombstoneIterator(TableMetadata metadata, DataInputPlus in, SerializationHeader header, DeserializationHelper helper)
{
super(metadata, in, helper);
this.header = header;

5
src/java/org/apache/cassandra/io/sstable/SSTableSimpleUnsortedWriter.java
View File

@ -31,6 +31,7 @@ import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.EncodingStats;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.UnfilteredSerializer;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.schema.TableMetadataRef;
@ -56,6 +57,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
// Used to compute the row serialized size
private final SerializationHeader header;
private final SerializationHelper helper;
private final BlockingQueue<Buffer> writeQueue = new SynchronousQueue<Buffer>();
private final DiskWriter diskWriter = new DiskWriter();
@ -65,6 +67,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
super(directory, metadata, columns);
this.bufferSize = bufferSizeInMB * 1024L * 1024L;
this.header = new SerializationHeader(true, metadata.get(), columns, EncodingStats.NO_STATS);
this.helper = new SerializationHelper(this.header);
diskWriter.start();
}
@ -90,7 +93,7 @@ class SSTableSimpleUnsortedWriter extends AbstractSSTableSimpleWriter
// improve that. In particular, what we count is closer to the serialized value, but it's debatable that it's the right thing
// to count since it will take a lot more space in memory and the bufferSize if first and foremost used to avoid OOM when
// using this writer.
currentSize += UnfilteredSerializer.serializer.serializedSize(row, header, 0, formatType.info.getLatestVersion().correspondingMessagingVersion());
currentSize += UnfilteredSerializer.serializer.serializedSize(row, helper, 0, formatType.info.getLatestVersion().correspondingMessagingVersion());
}
private void maybeSync() throws SyncException

2
src/java/org/apache/cassandra/service/paxos/Commit.java
View File

@ -117,7 +117,7 @@ public class Commit
public Commit deserialize(DataInputPlus in, int version) throws IOException
{
UUID ballot = UUIDSerializer.serializer.deserialize(in, version);
PartitionUpdate update = PartitionUpdate.serializer.deserialize(in, version, SerializationHelper.Flag.LOCAL);
PartitionUpdate update = PartitionUpdate.serializer.deserialize(in, version, DeserializationHelper.Flag.LOCAL);
return new Commit(ballot, update);
}

32
src/java/org/apache/cassandra/utils/WrappedInt.java → src/java/org/apache/cassandra/utils/BiLongAccumulator.java
View File

@ -18,35 +18,7 @@
package org.apache.cassandra.utils;
/**
* Simple wrapper for native int type
*/
public class WrappedInt
public interface BiLongAccumulator<T, A>
{
private int value;
public WrappedInt(int initial)
{
this.value = initial;
}
public int get()
{
return value;
}
public void set(int value)
{
this.value = value;
}
public void increment()
{
++value;
}
public void decrement()
{
--value;
}
long apply(T obj, A arguemnt, long v);
}

24
src/java/org/apache/cassandra/utils/LongAccumulator.java Normal file
View File

@ -0,0 +1,24 @@
/*
* 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.utils;
public interface LongAccumulator<T>
{
long apply(T obj, long v);
}

218
src/java/org/apache/cassandra/utils/btree/BTree.java
View File

@ -19,14 +19,18 @@
package org.apache.cassandra.utils.btree;
import java.util.*;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterators;
import com.google.common.collect.Ordering;
import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.ObjectSizes;
import static com.google.common.collect.Iterables.concat;
@ -691,7 +695,7 @@ public class BTree
}
// returns true if the provided node is a leaf, false if it is a branch
static boolean isLeaf(Object[] node)
public static boolean isLeaf(Object[] node)
{
return (node.length & 1) == 1;
}
@ -1285,37 +1289,19 @@ public class BTree
return compare(cmp, previous, max) < 0;
}
/**
* Simple method to walk the btree forwards or reversed and apply a function to each element
*
* Public method
*
*/
public static <V> void apply(Object[] btree, Consumer<V> function, boolean reversed)
private static <V, A> void applyValue(V value, BiConsumer<A, V> function, A argument)
{
if (reversed)
applyReverse(btree, function, null);
else
applyForwards(btree, function, null);
function.accept(argument, value);
}
/**
* Simple method to walk the btree forwards or reversed and apply a function till a stop condition is reached
*
* Public method
*
*/
public static <V> void apply(Object[] btree, Consumer<V> function, Predicate<V> stopCondition, boolean reversed)
public static <V, A> void applyLeaf(Object[] btree, BiConsumer<A, V> function, A argument)
{
if (reversed)
applyReverse(btree, function, stopCondition);
else
applyForwards(btree, function, stopCondition);
Preconditions.checkArgument(isLeaf(btree));
int limit = getLeafKeyEnd(btree);
for (int i=0; i<limit; i++)
applyValue((V) btree[i], function, argument);
}
/**
* Simple method to walk the btree forwards and apply a function till a stop condition is reached
*
@ -1323,89 +1309,143 @@ public class BTree
*
* @param btree
* @param function
* @param stopCondition
*/
private static <V> boolean applyForwards(Object[] btree, Consumer<V> function, Predicate<V> stopCondition)
public static <V, A> void apply(Object[] btree, BiConsumer<A, V> function, A argument)
{
boolean isLeaf = isLeaf(btree);
int childOffset = isLeaf ? Integer.MAX_VALUE : getChildStart(btree);
int limit = isLeaf ? getLeafKeyEnd(btree) : btree.length - 1;
for (int i = 0 ; i < limit ; i++)
if (isLeaf(btree))
{
// we want to visit in iteration order, so we visit our key nodes inbetween our children
int idx = isLeaf ? i : (i / 2) + (i % 2 == 0 ? childOffset : 0);
Object current = btree[idx];
if (idx < childOffset)
{
V castedCurrent = (V) current;
if (stopCondition != null && stopCondition.apply(castedCurrent))
return true;
function.accept(castedCurrent);
}
else
{
if (applyForwards((Object[]) current, function, stopCondition))
return true;
}
applyLeaf(btree, function, argument);
return;
}
return false;
int childOffset = getChildStart(btree);
int limit = btree.length - 1 - childOffset;
for (int i = 0 ; i < limit ; i++)
{
apply((Object[]) btree[childOffset + i], function, argument);
if (i < childOffset)
applyValue((V) btree[i], function, argument);
}
}
/**
* Simple method to walk the btree in reverse and apply a function till a stop condition is reached
* Simple method to walk the btree forwards and apply a function till a stop condition is reached
*
* Private method
*
* @param btree
* @param function
* @param stopCondition
*/
private static <V> boolean applyReverse(Object[] btree, Consumer<V> function, Predicate<V> stopCondition)
public static <V> void apply(Object[] btree, Consumer<V> function)
{
boolean isLeaf = isLeaf(btree);
int childOffset = isLeaf ? 0 : getChildStart(btree);
int limit = isLeaf ? getLeafKeyEnd(btree) : btree.length - 1;
for (int i = limit - 1, visited = 0; i >= 0 ; i--, visited++)
BTree.<V, Consumer<V>>apply(btree, Consumer::accept, function);
}
private static <V> int find(Object[] btree, V from, Comparator<V> comparator)
{
// find the start index in iteration order
Preconditions.checkNotNull(comparator);
int keyEnd = getKeyEnd(btree);
return Arrays.binarySearch((V[]) btree, 0, keyEnd, from, comparator);
}
private static boolean isStopSentinel(long v)
{
return v == Long.MAX_VALUE;
}
private static <V, A> long accumulateLeaf(Object[] btree, BiLongAccumulator<A, V> accumulator, A arg, Comparator<V> comparator, V from, long initialValue)
{
Preconditions.checkArgument(isLeaf(btree));
long value = initialValue;
int limit = getLeafKeyEnd(btree);
int startIdx = 0;
if (from != null)
{
int idx = i;
int i = find(btree, from, comparator);
boolean isExact = i >= 0;
startIdx = isExact ? i : (-1 - i);
}
// we want to visit in reverse iteration order, so we visit our children nodes inbetween our keys
if (!isLeaf)
for (int i = startIdx; i < limit; i++)
{
value = accumulator.apply(arg, (V) btree[i], value);
if (isStopSentinel(value))
break;
}
return value;
}
/**
* Walk the btree and accumulate a long value using the supplied accumulator function. Iteration will stop if the
* accumulator function returns the sentinel values Long.MIN_VALUE or Long.MAX_VALUE
*
* If the optional from argument is not null, iteration will start from that value (or the one after it's insertion
* point if an exact match isn't found)
*/
public static <V, A> long accumulate(Object[] btree, BiLongAccumulator<A, V> accumulator, A arg, Comparator<V> comparator, V from, long initialValue)
{
if (isLeaf(btree))
return accumulateLeaf(btree, accumulator, arg, comparator, from, initialValue);
long value = initialValue;
int childOffset = getChildStart(btree);
int startChild = 0;
if (from != null)
{
int i = find(btree, from, comparator);
boolean isExact = i >= 0;
startChild = isExact ? i + 1 : -1 - i;
if (isExact)
{
int typeOffset = visited / 2;
if (i % 2 == 0)
{
// This is a child branch. Since children are in the second half of the array, we must
// adjust for the key's we've visited along the way
idx += typeOffset;
}
else
{
// This is a key. Since the keys are in the first half of the array and we are iterating
// in reverse we subtract the childOffset and adjust for children we've walked so far
idx = i - childOffset + typeOffset;
}
}
Object current = btree[idx];
if (isLeaf || idx < childOffset)
{
V castedCurrent = (V) current;
if (stopCondition != null && stopCondition.apply(castedCurrent))
return true;
function.accept(castedCurrent);
}
else
{
if (applyReverse((Object[]) current, function, stopCondition))
return true;
value = accumulator.apply(arg, (V) btree[i], value);
if (isStopSentinel(value))
return value;
from = null;
}
}
return false;
int limit = btree.length - 1 - childOffset;
for (int i=startChild; i<limit; i++)
{
value = accumulate((Object[]) btree[childOffset + i], accumulator, arg, comparator, from, value);
if (isStopSentinel(value))
break;
if (i < childOffset)
{
value = accumulator.apply(arg, (V) btree[i], value);
// stop if a sentinel stop value was returned
if (isStopSentinel(value))
break;
}
if (from != null)
from = null;
}
return value;
}
public static <V> long accumulate(Object[] btree, LongAccumulator<V> accumulator, Comparator<V> comparator, V from, long initialValue)
{
return accumulate(btree, LongAccumulator::apply, accumulator, comparator, from, initialValue);
}
public static <V> long accumulate(Object[] btree, LongAccumulator<V> accumulator, long initialValue)
{
return accumulate(btree, accumulator, null, null, initialValue);
}
public static <V, A> long accumulate(Object[] btree, BiLongAccumulator<A, V> accumulator, A arg, long initialValue)
{
return accumulate(btree, accumulator, arg, null, null, initialValue);
}
}

4
src/java/org/apache/cassandra/utils/btree/BTreeSearchIterator.java
View File

@ -25,4 +25,8 @@ import org.apache.cassandra.utils.IndexedSearchIterator;
public interface BTreeSearchIterator<K, V> extends IndexedSearchIterator<K, V>, Iterator<V>
{
/**
* Reset this Iterator to its starting position
*/
public void rewind();
}

29
src/java/org/apache/cassandra/utils/btree/LeafBTreeSearchIterator.java
View File

@ -45,8 +45,13 @@ public class LeafBTreeSearchIterator<K, V> implements BTreeSearchIterator<K, V>
this.comparator = comparator;
this.lowerBound = lowerBound;
this.upperBound = upperBound;
this.nextPos = forwards ? lowerBound : upperBound;
this.hasNext = nextPos >= lowerBound && nextPos <= upperBound;
rewind();
}
public void rewind()
{
nextPos = forwards ? lowerBound : upperBound;
hasNext = nextPos >= lowerBound && nextPos <= upperBound;
}
public V next()
@ -73,12 +78,30 @@ public class LeafBTreeSearchIterator<K, V> implements BTreeSearchIterator<K, V>
return Arrays.binarySearch(keys, lb, ub + 1, key, comparator);
}
private void updateHasNext()
{
hasNext = nextPos >= lowerBound && nextPos <= upperBound;
}
public V next(K key)
{
if (!hasNext)
return null;
V result = null;
// first check the current position in case of sequential access
if (comparator.compare(key, keys[nextPos]) == 0)
{
hasCurrent = true;
result = (V) keys[nextPos];
nextPos += forwards ? 1 : -1;
}
updateHasNext();
if (result != null || !hasNext)
return result;
// otherwise search against the remaining values
int find = searchNext(key);
if (find >= 0)
{
@ -91,7 +114,7 @@ public class LeafBTreeSearchIterator<K, V> implements BTreeSearchIterator<K, V>
nextPos = (forwards ? -1 : -2) - find;
hasCurrent = false;
}
hasNext = nextPos >= lowerBound && nextPos <= upperBound;
updateHasNext();
return result;
}

5
test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java
View File

@ -35,15 +35,12 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import io.netty.util.concurrent.FastThreadLocalThread;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.config.Config.CommitLogSync;
import org.apache.cassandra.config.*;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.marshal.UTF8Type;
@ -451,7 +448,7 @@ public abstract class CommitLogStressTest
{
mutation = Mutation.serializer.deserialize(bufIn,
desc.getMessagingVersion(),
SerializationHelper.Flag.LOCAL);
DeserializationHelper.Flag.LOCAL);
}
catch (IOException e)
{

3
test/unit/org/apache/cassandra/db/ReadCommandTest.java
View File

@ -47,6 +47,7 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
import org.apache.cassandra.dht.Range;
@ -368,7 +369,7 @@ public class ReadCommandTest
MessagingService.current_version,
cfs.metadata(),
columnFilter,
SerializationHelper.Flag.LOCAL));
DeserializationHelper.Flag.LOCAL));
}
}

7
test/unit/org/apache/cassandra/db/RowIndexEntryTest.java
View File

@ -34,6 +34,7 @@ import org.junit.Test;
import org.apache.cassandra.Util;
import org.apache.cassandra.cache.IMeasurableMemory;
import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
@ -279,6 +280,7 @@ public class RowIndexEntryTest extends CQLTester
{
private final UnfilteredRowIterator iterator;
private final SequentialWriter writer;
private final SerializationHelper helper;
private final SerializationHeader header;
private final int version;
@ -306,6 +308,7 @@ public class RowIndexEntryTest extends CQLTester
{
this.iterator = iterator;
this.writer = writer;
this.helper = new SerializationHelper(header);
this.header = header;
this.version = version;
this.observers = observers == null ? Collections.emptyList() : observers;
@ -317,7 +320,7 @@ public class RowIndexEntryTest extends CQLTester
ByteBufferUtil.writeWithShortLength(iterator.partitionKey().getKey(), writer);
DeletionTime.serializer.serialize(iterator.partitionLevelDeletion(), writer);
if (header.hasStatic())
UnfilteredSerializer.serializer.serializeStaticRow(iterator.staticRow(), header, writer, version);
UnfilteredSerializer.serializer.serializeStaticRow(iterator.staticRow(), helper, writer, version);
}
public ColumnIndex build() throws IOException
@ -358,7 +361,7 @@ public class RowIndexEntryTest extends CQLTester
startPosition = pos;
}
UnfilteredSerializer.serializer.serialize(unfiltered, header, writer, pos - previousRowStart, version);
UnfilteredSerializer.serializer.serialize(unfiltered, helper, writer, pos - previousRowStart, version);
// notify observers about each new row
if (!observers.isEmpty())

4
test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java
View File

@ -25,7 +25,7 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.RebufferingInputStream;
@ -62,7 +62,7 @@ public class CDCTestReplayer extends CommitLogReplayer
Mutation mutation;
try
{
mutation = Mutation.serializer.deserialize(bufIn, desc.getMessagingVersion(), SerializationHelper.Flag.LOCAL);
mutation = Mutation.serializer.deserialize(bufIn, desc.getMessagingVersion(), DeserializationHelper.Flag.LOCAL);
if (mutation.trackedByCDC())
sawCDCMutation = true;
}

4
test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java
View File

@ -26,7 +26,7 @@ import org.junit.Assert;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.rows.SerializationHelper;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.RebufferingInputStream;
@ -65,7 +65,7 @@ public class CommitLogTestReplayer extends CommitLogReplayer
Mutation mutation;
try
{
mutation = Mutation.serializer.deserialize(bufIn, desc.getMessagingVersion(), SerializationHelper.Flag.LOCAL);
mutation = Mutation.serializer.deserialize(bufIn, desc.getMessagingVersion(), DeserializationHelper.Flag.LOCAL);
Assert.assertTrue(processor.apply(mutation));
}
catch (IOException e)

55
test/unit/org/apache/cassandra/utils/btree/BTreeTest.java
View File

@ -88,14 +88,20 @@ public class BTreeTest
}
};
private static List<Integer> seq(int count)
private static List<Integer> seq(int count, int interval)
{
List<Integer> r = new ArrayList<>();
for (int i = 0 ; i < count ; i++)
r.add(i);
if (i % interval == 0)
r.add(i);
return r;
}
private static List<Integer> seq(int count)
{
return seq(count, 1);
}
private static List<Integer> rand(int count)
{
Random rand = ThreadLocalRandom.current();
@ -133,27 +139,60 @@ public class BTreeTest
}
@Test
public void testApplyForwards()
public void testApply()
{
List<Integer> input = seq(71);
Object[] btree = BTree.build(input, noOp);
final List<Integer> result = new ArrayList<>();
BTree.<Integer>apply(btree, i -> result.add(i), false);
BTree.<Integer>apply(btree, i -> result.add(i));
org.junit.Assert.assertArrayEquals(input.toArray(),result.toArray());
}
@Test
public void testApplyReverse()
public void inOrderAccumulation()
{
List<Integer> input = seq(71);
Object[] btree = BTree.build(input, noOp);
long result = BTree.<Integer>accumulate(btree, (o, l) -> {
Assert.assertEquals((long) o, l + 1);
return o;
}, -1);
Assert.assertEquals(result, 70);
}
final List<Integer> result = new ArrayList<>();
BTree.<Integer>apply(btree, i -> result.add(i), true);
@Test
public void accumulateFrom()
{
int limit = 100;
for (int interval=1; interval<=5; interval++)
{
List<Integer> input = seq(limit, interval);
Object[] btree = BTree.build(input, noOp);
for (int start=0; start<=limit; start+=interval)
{
int thisInterval = interval;
String errMsg = String.format("interval=%s, start=%s", interval, start);
long result = BTree.accumulate(btree, (o, l) -> {
Assert.assertEquals(errMsg, (long) o, l + thisInterval);
return o;
}, Comparator.naturalOrder(), start, start - thisInterval);
Assert.assertEquals(errMsg, result, (limit-1)/interval*interval);
}
}
}
org.junit.Assert.assertArrayEquals(Lists.reverse(input).toArray(),result.toArray());
/**
* accumulate function should not be called if we ask it to start past the end of the btree
*/
@Test
public void accumulateFromEnd()
{
List<Integer> input = seq(100);
Object[] btree = BTree.build(input, noOp);
long result = BTree.accumulate(btree, (o, l) -> 1, Integer::compareTo, 101, 0L);
Assert.assertEquals(0, result);
}
/**