From 8576e769d13b3e887ea604074641fd4c42af5e8a Mon Sep 17 00:00:00 2001 From: Blake Eggleston Date: Tue, 15 Oct 2019 20:10:42 -0700 Subject: [PATCH] Minimize BTree iterator allocations Patch by Blake Eggleston; Reviewed by Benedict Elliott Smith for CASSANDRA-15389 --- CHANGES.txt | 1 + .../org/apache/cassandra/db/ColumnIndex.java | 6 +- src/java/org/apache/cassandra/db/Columns.java | 9 +- .../org/apache/cassandra/db/Mutation.java | 6 +- .../org/apache/cassandra/db/ReadResponse.java | 8 +- .../cassandra/db/SerializationHeader.java | 1 + .../cassandra/db/UnfilteredDeserializer.java | 6 +- .../AbstractSSTableIterator.java | 6 +- .../db/commitlog/CommitLogReader.java | 4 +- .../db/partitions/CachedBTreePartition.java | 4 +- .../db/partitions/PartitionUpdate.java | 7 +- .../UnfilteredPartitionIterators.java | 2 +- .../apache/cassandra/db/rows/AbstractRow.java | 18 +- .../apache/cassandra/db/rows/BTreeRow.java | 103 +++++---- .../org/apache/cassandra/db/rows/Cell.java | 2 +- .../apache/cassandra/db/rows/ColumnData.java | 5 + .../cassandra/db/rows/ComplexColumnData.java | 18 +- .../db/rows/DeserializationHelper.java | 149 ++++++++++++ .../org/apache/cassandra/db/rows/Row.java | 23 +- .../org/apache/cassandra/db/rows/Rows.java | 72 +++--- .../db/rows/SerializationHelper.java | 136 ++--------- .../rows/UnfilteredRowIteratorSerializer.java | 21 +- .../db/rows/UnfilteredSerializer.java | 95 ++++---- .../db/streaming/CassandraStreamReader.java | 4 +- .../io/sstable/SSTableIdentityIterator.java | 4 +- .../io/sstable/SSTableSimpleIterator.java | 12 +- .../sstable/SSTableSimpleUnsortedWriter.java | 5 +- .../cassandra/service/paxos/Commit.java | 2 +- ...WrappedInt.java => BiLongAccumulator.java} | 32 +-- .../cassandra/utils/LongAccumulator.java | 24 ++ .../apache/cassandra/utils/btree/BTree.java | 218 +++++++++++------- .../utils/btree/BTreeSearchIterator.java | 4 + .../utils/btree/LeafBTreeSearchIterator.java | 29 ++- .../db/commitlog/CommitLogStressTest.java | 5 +- .../apache/cassandra/db/ReadCommandTest.java | 3 +- .../cassandra/db/RowIndexEntryTest.java | 7 +- .../db/commitlog/CDCTestReplayer.java | 4 +- .../db/commitlog/CommitLogTestReplayer.java | 4 +- .../cassandra/utils/btree/BTreeTest.java | 55 ++++- 39 files changed, 667 insertions(+), 447 deletions(-) create mode 100644 src/java/org/apache/cassandra/db/rows/DeserializationHelper.java rename src/java/org/apache/cassandra/utils/{WrappedInt.java => BiLongAccumulator.java} (67%) create mode 100644 src/java/org/apache/cassandra/utils/LongAccumulator.java diff --git a/CHANGES.txt b/CHANGES.txt index 96ce286884..c1e5aeb70c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -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) diff --git a/src/java/org/apache/cassandra/db/ColumnIndex.java b/src/java/org/apache/cassandra/db/ColumnIndex.java index 74ad2646ea..e11f784ec5 100644 --- a/src/java/org/apache/cassandra/db/ColumnIndex.java +++ b/src/java/org/apache/cassandra/db/ColumnIndex.java @@ -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 observers, ISerializer 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()) diff --git a/src/java/org/apache/cassandra/db/Columns.java b/src/java/org/apache/cassandra/db/Columns.java index f56072e49b..fe13919130 100644 --- a/src/java/org/apache/cassandra/db/Columns.java +++ b/src/java/org/apache/cassandra/db/Columns.java @@ -53,7 +53,7 @@ public class Columns extends AbstractCollection 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 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 implements Colle /** * Apply a function to each column definition in forwards or reversed order. * @param function - * @param reversed */ - public void apply(Consumer function, boolean reversed) + public void apply(Consumer function) { - BTree.apply(columns, function, reversed); + BTree.apply(columns, function); } @Override diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index 22c4ed8376..3d27ef31d8 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -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) diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index affbbbe1a0..3f6481d6c4 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/SerializationHeader.java b/src/java/org/apache/cassandra/db/SerializationHeader.java index 15ef268f8c..1c22febc44 100644 --- a/src/java/org/apache/cassandra/db/SerializationHeader.java +++ b/src/java/org/apache/cassandra/db/SerializationHeader.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/UnfilteredDeserializer.java b/src/java/org/apache/cassandra/db/UnfilteredDeserializer.java index 84ff691663..f9ff1d7175 100644 --- a/src/java/org/apache/cassandra/db/UnfilteredDeserializer.java +++ b/src/java/org/apache/cassandra/db/UnfilteredDeserializer.java @@ -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); } diff --git a/src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java b/src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java index cfc7da2cd0..c631f1ce67 100644 --- a/src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java +++ b/src/java/org/apache/cassandra/db/columniterator/AbstractSSTableIterator.java @@ -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()) diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogReader.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogReader.java index c91841f6e4..5123580194 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogReader.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogReader.java @@ -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(); diff --git a/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java b/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java index 4d9c227f99..9a2b331515 100644 --- a/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java +++ b/src/java/org/apache/cassandra/db/partitions/CachedBTreePartition.java @@ -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); } diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java index ec01fa6a99..076c975a54 100644 --- a/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java +++ b/src/java/org/apache/cassandra/db/partitions/PartitionUpdate.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java index 945bcb41de..30b4d9eb77 100644 --- a/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java +++ b/src/java/org/apache/cassandra/db/partitions/UnfilteredPartitionIterators.java @@ -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(); diff --git a/src/java/org/apache/cassandra/db/rows/AbstractRow.java b/src/java/org/apache/cassandra/db/rows/AbstractRow.java index 957ffd4242..fc90e34038 100644 --- a/src/java/org/apache/cassandra/db/rows/AbstractRow.java +++ b/src/java/org/apache/cassandra/db/rows/AbstractRow.java @@ -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() diff --git a/src/java/org/apache/cassandra/db/rows/BTreeRow.java b/src/java/org/apache/cassandra/db/rows/BTreeRow.java index dc3219aa7c..6689c77ff5 100644 --- a/src/java/org/apache/cassandra/db/rows/BTreeRow.java +++ b/src/java/org/apache/cassandra/db/rows/BTreeRow.java @@ -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 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.iterable(btree)) - minDeletionTime = Math.min(minDeletionTime, minDeletionTime(cd)); + long result = BTree.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 function, boolean reversed) + public void apply(Consumer function) { - BTree.apply(btree, function, reversed); + BTree.apply(btree, function); } - public void apply(Consumer funtion, com.google.common.base.Predicate stopCondition, boolean reversed) + public void apply(BiConsumer function, A arg) { - BTree.apply(btree, funtion, stopCondition, reversed); + BTree.apply(btree, function, arg); + } + + public long accumulate(LongAccumulator accumulator, long initialValue) + { + return BTree.accumulate(btree, accumulator, initialValue); + } + + public long accumulate(LongAccumulator accumulator, Comparator comparator, ColumnData from, long initialValue) + { + return BTree.accumulate(btree, accumulator, comparator, from, initialValue); + } + + public long accumulate(BiLongAccumulator accumulator, A arg, long initialValue) + { + return BTree.accumulate(btree, accumulator, arg, initialValue); + } + + public long accumulate(BiLongAccumulator accumulator, A arg, Comparator 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.apply(btree, cd -> min.set( Math.min(min.get(), minDeletionTime(cd)) ), cd -> min.get() == Integer.MIN_VALUE, false); - return min.get(); + min = BTree.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.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() diff --git a/src/java/org/apache/cassandra/db/rows/Cell.java b/src/java/org/apache/cassandra/db/rows/Cell.java index 300bbce12a..959676a192 100644 --- a/src/java/org/apache/cassandra/db/rows/Cell.java +++ b/src/java/org/apache/cassandra/db/rows/Cell.java @@ -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; diff --git a/src/java/org/apache/cassandra/db/rows/ColumnData.java b/src/java/org/apache/cassandra/db/rows/ColumnData.java index e5f555063d..36aad974cd 100644 --- a/src/java/org/apache/cassandra/db/rows/ColumnData.java +++ b/src/java/org/apache/cassandra/db/rows/ColumnData.java @@ -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}. diff --git a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java index 832167f03d..5b0350439d 100644 --- a/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java +++ b/src/java/org/apache/cassandra/db/rows/ComplexColumnData.java @@ -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 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 return BTree.iterator(cells, BTree.Dir.DESC); } + public long accumulate(LongAccumulator accumulator, long initialValue) + { + return BTree.accumulate(cells, accumulator, initialValue); + } + + public long accumulate(BiLongAccumulator accumulator, A arg, long initialValue) + { + return BTree.accumulate(cells, accumulator, arg, initialValue); + } + public int dataSize() { int size = complexDeletion.dataSize(); diff --git a/src/java/org/apache/cassandra/db/rows/DeserializationHelper.java b/src/java/org/apache/cassandra/db/rows/DeserializationHelper.java new file mode 100644 index 0000000000..386e6ef9bd --- /dev/null +++ b/src/java/org/apache/cassandra/db/rows/DeserializationHelper.java @@ -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 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; + } +} diff --git a/src/java/org/apache/cassandra/db/rows/Row.java b/src/java/org/apache/cassandra/db/rows/Row.java index 2f752b8908..ee93da46c7 100644 --- a/src/java/org/apache/cassandra/db/rows/Row.java +++ b/src/java/org/apache/cassandra/db/rows/Row.java @@ -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 /** * Apply a function to every column in a row */ - public void apply(Consumer function, boolean reverse); + public void apply(Consumer 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 function, Predicate stopCondition, boolean reverse); + public void apply(BiConsumer function, A arg); + + /** + * Apply an accumulation funtion to every column in a row + */ + + public long accumulate(LongAccumulator accumulator, long initialValue); + + public long accumulate(LongAccumulator accumulator, Comparator comparator, ColumnData from, long initialValue); + + public long accumulate(BiLongAccumulator accumulator, A arg, long initialValue); + + public long accumulate(BiLongAccumulator accumulator, A arg, Comparator comparator, ColumnData from, long initialValue); /** * A row deletion/tombstone. diff --git a/src/java/org/apache/cassandra/db/rows/Rows.java b/src/java/org/apache/cassandra/db/rows/Rows.java index d62d3b578f..58284ac4b9 100644 --- a/src/java/org/apache/cassandra/db/rows/Rows.java +++ b/src/java/org/apache/cassandra/db/rows/Rows.java @@ -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); } /** diff --git a/src/java/org/apache/cassandra/db/rows/SerializationHelper.java b/src/java/org/apache/cassandra/db/rows/SerializationHelper.java index db23cb868c..dca4240dd0 100644 --- a/src/java/org/apache/cassandra/db/rows/SerializationHelper.java +++ b/src/java/org/apache/cassandra/db/rows/SerializationHelper.java @@ -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 statics = null; + private BTreeSearchIterator 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 droppedColumns; - private DroppedColumn currentDroppedComplex; - - - public SerializationHelper(TableMetadata metadata, int version, Flag flag, ColumnFilter columnsToFetch) + private BTreeSearchIterator 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 regulars() { - this(metadata, version, flag, null); + if (regulars == null) + regulars = header.columns().regulars.iterator(); + return regulars; } - public boolean includes(ColumnMetadata column) + public SearchIterator 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 iterator = isStatic ? statics() : regulars(); + iterator.rewind(); + return iterator; } } diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java index 7cac5e6548..df67754b1b 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredRowIteratorSerializer.java @@ -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)); } diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java index 7b486526b2..a5fad1435b 100644 --- a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java +++ b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java @@ -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 si = headerColumns.iterator(); + SearchIterator 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 si = headerColumns.iterator(); - for (ColumnData data : row) - { + SearchIterator 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; diff --git a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java index 190f1360bb..686d874d39 100644 --- a/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java +++ b/src/java/org/apache/cassandra/db/streaming/CassandraStreamReader.java @@ -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; } diff --git a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java index 1846aa5155..76e12c891a 100644 --- a/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java +++ b/src/java/org/apache/cassandra/io/sstable/SSTableIdentityIterator.java @@ -57,7 +57,7 @@ public class SSTableIdentityIterator implements Comparable +{ + long apply(T obj, long v); +} diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java index 97e935e42b..6c546ac95e 100644 --- a/src/java/org/apache/cassandra/utils/btree/BTree.java +++ b/src/java/org/apache/cassandra/utils/btree/BTree.java @@ -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 void apply(Object[] btree, Consumer function, boolean reversed) + private static void applyValue(V value, BiConsumer 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 void apply(Object[] btree, Consumer function, Predicate stopCondition, boolean reversed) + public static void applyLeaf(Object[] btree, BiConsumer 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 boolean applyForwards(Object[] btree, Consumer function, Predicate stopCondition) + public static void apply(Object[] btree, BiConsumer 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 boolean applyReverse(Object[] btree, Consumer function, Predicate stopCondition) + public static void apply(Object[] btree, Consumer 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.>apply(btree, Consumer::accept, function); + } + + private static int find(Object[] btree, V from, Comparator 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 long accumulateLeaf(Object[] btree, BiLongAccumulator accumulator, A arg, Comparator 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 long accumulate(Object[] btree, BiLongAccumulator accumulator, A arg, Comparator 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 long accumulate(Object[] btree, LongAccumulator accumulator, Comparator comparator, V from, long initialValue) + { + return accumulate(btree, LongAccumulator::apply, accumulator, comparator, from, initialValue); + } + + public static long accumulate(Object[] btree, LongAccumulator accumulator, long initialValue) + { + return accumulate(btree, accumulator, null, null, initialValue); + } + + public static long accumulate(Object[] btree, BiLongAccumulator accumulator, A arg, long initialValue) + { + return accumulate(btree, accumulator, arg, null, null, initialValue); } } diff --git a/src/java/org/apache/cassandra/utils/btree/BTreeSearchIterator.java b/src/java/org/apache/cassandra/utils/btree/BTreeSearchIterator.java index 2fcece6219..2ad7f40148 100644 --- a/src/java/org/apache/cassandra/utils/btree/BTreeSearchIterator.java +++ b/src/java/org/apache/cassandra/utils/btree/BTreeSearchIterator.java @@ -25,4 +25,8 @@ import org.apache.cassandra.utils.IndexedSearchIterator; public interface BTreeSearchIterator extends IndexedSearchIterator, Iterator { + /** + * Reset this Iterator to its starting position + */ + public void rewind(); } diff --git a/src/java/org/apache/cassandra/utils/btree/LeafBTreeSearchIterator.java b/src/java/org/apache/cassandra/utils/btree/LeafBTreeSearchIterator.java index a6197f8f7d..29aed4beb0 100644 --- a/src/java/org/apache/cassandra/utils/btree/LeafBTreeSearchIterator.java +++ b/src/java/org/apache/cassandra/utils/btree/LeafBTreeSearchIterator.java @@ -45,8 +45,13 @@ public class LeafBTreeSearchIterator implements BTreeSearchIterator 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 implements BTreeSearchIterator 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 implements BTreeSearchIterator nextPos = (forwards ? -1 : -2) - find; hasCurrent = false; } - hasNext = nextPos >= lowerBound && nextPos <= upperBound; + updateHasNext(); return result; } diff --git a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java index e2c6e33789..7fadac4382 100644 --- a/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java +++ b/test/long/org/apache/cassandra/db/commitlog/CommitLogStressTest.java @@ -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) { diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java index 87097753b1..e0215b7fef 100644 --- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java +++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java @@ -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)); } } diff --git a/test/unit/org/apache/cassandra/db/RowIndexEntryTest.java b/test/unit/org/apache/cassandra/db/RowIndexEntryTest.java index c52fd3de50..392a1a041d 100644 --- a/test/unit/org/apache/cassandra/db/RowIndexEntryTest.java +++ b/test/unit/org/apache/cassandra/db/RowIndexEntryTest.java @@ -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()) diff --git a/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java index 18bc6e0a95..fa3295a484 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java @@ -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; } diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java index 9a22b04d2e..5b87d68781 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java @@ -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) diff --git a/test/unit/org/apache/cassandra/utils/btree/BTreeTest.java b/test/unit/org/apache/cassandra/utils/btree/BTreeTest.java index 7cc12914b8..e60fb64560 100644 --- a/test/unit/org/apache/cassandra/utils/btree/BTreeTest.java +++ b/test/unit/org/apache/cassandra/utils/btree/BTreeTest.java @@ -88,14 +88,20 @@ public class BTreeTest } }; - private static List seq(int count) + private static List seq(int count, int interval) { List 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 seq(int count) + { + return seq(count, 1); + } + private static List rand(int count) { Random rand = ThreadLocalRandom.current(); @@ -133,27 +139,60 @@ public class BTreeTest } @Test - public void testApplyForwards() + public void testApply() { List input = seq(71); Object[] btree = BTree.build(input, noOp); final List result = new ArrayList<>(); - BTree.apply(btree, i -> result.add(i), false); + BTree.apply(btree, i -> result.add(i)); org.junit.Assert.assertArrayEquals(input.toArray(),result.toArray()); } @Test - public void testApplyReverse() + public void inOrderAccumulation() { List input = seq(71); Object[] btree = BTree.build(input, noOp); + long result = BTree.accumulate(btree, (o, l) -> { + Assert.assertEquals((long) o, l + 1); + return o; + }, -1); + Assert.assertEquals(result, 70); + } - final List result = new ArrayList<>(); - BTree.apply(btree, i -> result.add(i), true); + @Test + public void accumulateFrom() + { + int limit = 100; + for (int interval=1; interval<=5; interval++) + { + List 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 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); } /**