Compare commits

...

12 Commits

Author SHA1 Message Date
Štefan Miklošovič 1f58878248
Merge ff5a619551 into 10557d7ffe 2026-08-01 14:09:26 +03:00
Sam Tunnicliffe 10557d7ffe Merge branch 'cassandra-6.0' into trunk 2026-07-30 11:21:35 +01:00
C. Scott Andreas 8fd77ffea3 Improve stability of CMSShutdownTest
Patch by C. Scott Andreas; reviewed by Sam Tunnicliffe for CASSANDRA-21501
2026-07-30 11:14:52 +01:00
Dmitry Konstantinov e0e4feb120 Merge branch 'cassandra-6.0' into trunk
* cassandra-6.0:
  Apply performance optimizations for rows merging logic
2026-07-29 23:42:15 +01:00
Dmitry Konstantinov 1e3d43b2dd Apply performance optimizations for rows merging logic
duplicate MergeIterator class to avoid megamorphic calls in hot path for Row/ComplexCell and UnfilteredRowIterators
disable inlining for sinkInBinaryHeap, move typical case outside to improve hot path inlining
replace List for versions with array
fast path for same column metadata in cell merger
avoid potential megamorphic cell method invocation in ALIVE DeletionTime
minDeletionTime calculation optimization for merge logic
do not add an empty iterator, merge 2 loops into 1

patch by Dmitry Konstantinov; reviewed by Francisco Guerrero for CASSANDRA-21524
2026-07-29 23:28:07 +01:00
Caleb Rackliffe bcd9e662d2 Merge branch 'cassandra-6.0' into trunk
* cassandra-6.0:
  Fix operationMode reporting DECOMMISSION_FAILED instead of LEAVING when resuming a failed decommission
2026-07-29 15:25:14 -05:00
Minal Kyada feef3fcf51 Fix operationMode reporting DECOMMISSION_FAILED instead of LEAVING when resuming a failed decommission
Patch by Minal Kyada; reviewed by Caleb Rackliffe and Sam Tunnicliffe for CASSANDRA-21493
2026-07-29 15:15:02 -05:00
Francisco Guerrero 5d1830775d Merge branch 'cassandra-6.0' into trunk
* cassandra-6.0:
  [CASSANDRA-21536] Profile pollution in AbstractType.writeValue makes serialization slow for all column types
2026-07-29 14:21:37 -05:00
koo.taejin aa1447571b [CASSANDRA-21536] Profile pollution in AbstractType.writeValue makes serialization slow for all column types
Description:
AbstractType.writeValue() is one shared method. All column types use it.

Inside writeValue(), it calls valueLengthIfFixed(). This is a virtual call. Many types override this method (Int32Type, LongType, UTF8Type, ...).

In a real cluster, many column types pass through writeValue(). So the profile always sees lots types.

Because of this, the JIT cannot inline the call. It stays as a vtable call. Also, the compiled body of writeValue() becomes big, so the JIT refuses to inline writeValue() itself ("already compiled into a big method").

We also see the same itable/vtable stubs in async-profiler output from a real cluster, running with default production options and a normal workload.

How to solve:
Add a final int field to AbstractType. Set it in the constructor. writeValue() reads this field instead of calling valueLengthIfFixed().

A field read needs no type profile. So profile pollution has no effect on it. The valueLengthIfFixed() method is not changed. Only the write path uses the field.

Result:
Production is always the polluted state, so this is the real-world comparison

 JMH, JDK 17 / arm64, 1 thread, 10 x 1s warmup and measurement:

                    before          after       improvement
   readValue        49.99M ops/s    56.44M ops/s  +12.90%
   writeValue       99.63M ops/s   114.46M ops/s  +14.88%

patch by Koo Taejin; reviewed by Dmitry Konstantinov, Francisco Guerrero for CASSANDRA-21536
2026-07-29 14:16:15 -05:00
Dmitry Konstantinov 565c366ee1 Merge branch 'cassandra-6.0' into trunk
* cassandra-6.0:
  Avoid megamorphic calls for Cell.timestamp/ttl/path/localDeletionTimeAsUnsignedInt methods
2026-07-29 09:24:51 +01:00
Dmitry Konstantinov 725c61c1f7 Avoid megamorphic calls for Cell.timestamp/ttl/path/localDeletionTimeAsUnsignedInt methods
minDeletionTime is also added to Cell to avoid double invocation of localDeletionTime method

patch by Dmitry Konstantinov; reviewed by Francisco Guerrero for CASSANDRA-21526
2026-07-29 09:15:24 +01:00
Stefan Miklosovic ff5a619551
condition javadoc installation via ant mvn-install 2025-04-17 14:42:27 +02:00
38 changed files with 510 additions and 215 deletions

View File

@ -6,6 +6,10 @@
* Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767)
* Add a guardrail for misprepared statements (CASSANDRA-21139) * Add a guardrail for misprepared statements (CASSANDRA-21139)
Merged from 6.0: Merged from 6.0:
* Apply performance optimizations for rows merging logic (CASSANDRA-21524)
* Fix operationMode reporting DECOMMISSION_FAILED instead of LEAVING when resuming a failed decommission (CASSANDRA-21493)
* Avoid megamorphic calls when serializing and deserializing fixed-length values (CASSANDRA-21536)
* Avoid megamorphic calls for Cell.timestamp/ttl/path/localDeletionTimeAsUnsignedInt methods (CASSANDRA-21526)
* Reduce allocations in DefaultQueryOptions (CASSANDRA-21467) * Reduce allocations in DefaultQueryOptions (CASSANDRA-21467)
* Allow unreserved keywords as user and identity names in USER and IDENTITY statements (CASSANDRA-21510) * Allow unreserved keywords as user and identity names in USER and IDENTITY statements (CASSANDRA-21510)
* Reduce allocations in DefaultQueryOptions (CASSANDRA-21467) * Reduce allocations in DefaultQueryOptions (CASSANDRA-21467)
@ -29,7 +33,7 @@ Merged from 6.0:
* Always send TCM commit failures as Messaging failures (CASSANDRA-21457) * Always send TCM commit failures as Messaging failures (CASSANDRA-21457)
* Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438) * Fix ReadCommand serializedSize() using incorrect epoch (CASSANDRA-21438)
* Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199) * Allocation improvements in ProtocolVersion, StorageProxy and MerkleTree (CASSANDRA-21199)
* Dont leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236) * Don't leave autocompaction disabled during bootstrap and replace (CASSANDRA-21236)
* Make nodetool abortbootstrap more robust (CASSANDRA-21235) * Make nodetool abortbootstrap more robust (CASSANDRA-21235)
* Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234) * Don't clear prepared statement cache on nodetool cms initialize (CASSANDRA-21234)
* Improve performance when deserializing cluster metadata (CASSANDRA-21224) * Improve performance when deserializing cluster metadata (CASSANDRA-21224)
@ -8666,4 +8670,3 @@ Full list of issues resolved in 0.4 is at https://issues.apache.org/jira/secure/
* Added FlushPeriodInMinutes configuration parameter to force * Added FlushPeriodInMinutes configuration parameter to force
flushing of infrequently-updated ColumnFamilies flushing of infrequently-updated ColumnFamilies

View File

@ -651,6 +651,64 @@
<jflex file="${build.src.java}/org/apache/cassandra/index/sasi/analyzer/StandardTokenizerImpl.jflex" destdir="${build.src.gen-java}/" /> <jflex file="${build.src.java}/org/apache/cassandra/index/sasi/analyzer/StandardTokenizerImpl.jflex" destdir="${build.src.gen-java}/" />
</target> </target>
<!--
Copies a source class into ${build.src.gen-java} under a NEW class name while keeping it
in its ORIGINAL package, so it compiles into an independent class that still enjoys
package-private access to its neighbours.
This lets a hot call site use its own dedicated copy of a class (e.g. a RowMergeIterator
copy of MergeIterator) so that the virtual calls inside it stay mono-/bi-morphic and
remain inlinable, instead of turning megamorphic once every caller in the code base
funnels the shared original through the same bytecode.
The transformation is purely textual: every whole-word occurrence of the original simple
class name is rewritten to the new name. Matching on word boundaries keeps substrings
such as IMergeIterator intact. The package declaration is left untouched, so the copy
stays in the same package as the original and no imports need to change.
-->
<macrodef name="gen-java-copy">
<!-- Fully-qualified name of the class to copy, e.g. org.apache.cassandra.utils.MergeIterator -->
<attribute name="class"/>
<!-- New simple class name for the copy, e.g. RowMergeIterator -->
<attribute name="newName"/>
<sequential>
<local name="gen.src.rel"/>
<local name="gen.simple.name"/>
<!-- source file path relative to ${build.src.java}: dots -> slashes -->
<loadresource property="gen.src.rel">
<string value="@{class}"/>
<filterchain><tokenfilter><replaceregex pattern="\." replace="/" flags="g"/></tokenfilter></filterchain>
</loadresource>
<!-- original simple class name = last segment of the fully-qualified name -->
<loadresource property="gen.simple.name">
<string value="@{class}"/>
<filterchain><tokenfilter><replaceregex pattern="^.*\.([^.]+)$" replace="\1"/></tokenfilter></filterchain>
</loadresource>
<copy todir="${build.src.gen-java}" preservelastmodified="true">
<fileset dir="${build.src.java}" includes="${gen.src.rel}.java"/>
<!-- keep the original package directory, only rename the file -->
<mapper type="regexp" from="^(.*[\\/])[^\\/]+$" to="\1@{newName}.java"/>
<filterchain>
<tokenfilter>
<replaceregex pattern="\b${gen.simple.name}\b" replace="@{newName}" flags="g"/>
</tokenfilter>
</filterchain>
</copy>
</sequential>
</macrodef>
<!--
Generate the copies of hand-picked classes before compilation. Add a
<gen-java-copy class="..." newName="..."/> line here for every class that needs a
dedicated copy.
-->
<target name="gen-java-copies" depends="init"
description="Copy selected classes under new names (e.g. MergeIterator -> RowMergeIterator) into src/gen-java to avoid megamorphic calls">
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="RowMergeIterator"/>
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="UnfilteredMergeIterator"/>
<gen-java-copy class="org.apache.cassandra.utils.MergeIterator" newName="ComplexCellMergeIterator"/>
</target>
<!-- create properties file with C version --> <!-- create properties file with C version -->
<target name="_createVersionPropFile" depends="_get-git-sha,set-cqlsh-version,_set-build-date"> <target name="_createVersionPropFile" depends="_get-git-sha,set-cqlsh-version,_set-build-date">
<taskdef name="propertyfile" classname="org.apache.tools.ant.taskdefs.optional.PropertyFile"/> <taskdef name="propertyfile" classname="org.apache.tools.ant.taskdefs.optional.PropertyFile"/>
@ -710,7 +768,7 @@
</javac> </javac>
</target> </target>
<target depends="init,gen-cql3-grammar,generate-cql-html,generate-jflex-java" <target depends="init,gen-cql3-grammar,generate-cql-html,generate-jflex-java,gen-java-copies"
name="build-project"> name="build-project">
<echo message="${ant.project.name}: ${ant.file}"/> <echo message="${ant.project.name}: ${ant.file}"/>
<!-- Order matters! --> <!-- Order matters! -->
@ -2173,7 +2231,7 @@
</target> </target>
<!-- Generate IDEA project description files --> <!-- Generate IDEA project description files -->
<target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,_createVersionPropFile" description="Generate IDEA files"> <target name="generate-idea-files" depends="init,resolver-dist-lib,gen-cql3-grammar,generate-jflex-java,gen-java-copies,_createVersionPropFile" description="Generate IDEA files">
<delete dir=".idea"/> <delete dir=".idea"/>
<delete file="${eclipse.project.name}.iml"/> <delete file="${eclipse.project.name}.iml"/>
<mkdir dir=".idea"/> <mkdir dir=".idea"/>
@ -2298,7 +2356,7 @@
<!-- Installs artifacts to local Maven repository --> <!-- Installs artifacts to local Maven repository -->
<target name="mvn-install" <target name="mvn-install"
depends="jar,sources-jar,javadoc-jar" depends="jar,sources-jar,javadoc-jar,install-javadoc"
description="Installs the artifacts in the Maven Local Repository"> description="Installs the artifacts in the Maven Local Repository">
<!-- This logic does not install cassandra-accord as the accord submodule already did so --> <!-- This logic does not install cassandra-accord as the accord submodule already did so -->
@ -2310,12 +2368,16 @@
<!-- the cassandra-all jar --> <!-- the cassandra-all jar -->
<install pomFile="${build.dir}/${final.name}.pom" <install pomFile="${build.dir}/${final.name}.pom"
file="${build.dir}/${final.name}.jar"/> file="${build.dir}/${final.name}.jar"/>
<install pomFile="${build.dir}/${final.name}.pom" <install pomFile="${build.dir}/${final.name}.pom"
file="${build.dir}/${final.name}-sources.jar" file="${build.dir}/${final.name}-sources.jar"
classifier="sources"/> classifier="sources"/>
</target>
<target name="install-javadoc" unless="no-javadoc">
<install pomFile="${build.dir}/${final.name}.pom" <install pomFile="${build.dir}/${final.name}.pom"
file="${build.dir}/${final.name}-javadoc.jar" file="${build.dir}/${final.name}-sources.jar"
classifier="javadoc"/> classifier="sources"/>
</target> </target>
<!-- Publish artifacts to remote Maven repository --> <!-- Publish artifacts to remote Maven repository -->

View File

@ -178,7 +178,8 @@ public abstract class DeletionTime implements Comparable<DeletionTime>, IMeasura
public boolean deletes(Cell<?> cell) public boolean deletes(Cell<?> cell)
{ {
return deletes(cell.timestamp()); // check for LIVE first to avoid a potential cell megamorphic call
return markedForDeleteAt() != MARKED_FOR_DELETE_AT_LIVE && deletes(cell.timestamp());
} }
public boolean deletes(long timestamp) public boolean deletes(long timestamp)

View File

@ -39,7 +39,7 @@ public abstract class AbstractTimeUUIDType<T> extends TemporalType<T>
{ {
AbstractTimeUUIDType() AbstractTimeUUIDType()
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, 16);
} // singleton } // singleton
@Override @Override
@ -193,12 +193,6 @@ public abstract class AbstractTimeUUIDType<T> extends TemporalType<T>
return super.decomposeUntyped(value); return super.decomposeUntyped(value);
} }
@Override
public int valueLengthIfFixed()
{
return 16;
}
@Override @Override
public long toTimeInMillis(ByteBuffer value) public long toTimeInMillis(ByteBuffer value)
{ {

View File

@ -90,11 +90,18 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public final ComparisonType comparisonType; public final ComparisonType comparisonType;
public final boolean isByteOrderComparable; public final boolean isByteOrderComparable;
public final ValueComparators comparatorSet; public final ValueComparators comparatorSet;
private final int valueLengthIfFixed;
protected AbstractType(ComparisonType comparisonType) protected AbstractType(ComparisonType comparisonType)
{
this(comparisonType, VARIABLE_LENGTH);
}
protected AbstractType(ComparisonType comparisonType, int valueLengthIfFixed)
{ {
this.comparisonType = comparisonType; this.comparisonType = comparisonType;
this.isByteOrderComparable = comparisonType == ComparisonType.BYTE_ORDER; this.isByteOrderComparable = comparisonType == ComparisonType.BYTE_ORDER;
this.valueLengthIfFixed = valueLengthIfFixed;
reverseComparator = (o1, o2) -> AbstractType.this.compare(o2, o1); reverseComparator = (o1, o2) -> AbstractType.this.compare(o2, o1);
try try
{ {
@ -384,12 +391,12 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
/** /**
* Similar to {@link #isValueCompatibleWith(AbstractType)}, but takes into account {@link Cell} encoding. * Similar to {@link #isValueCompatibleWith(AbstractType)}, but takes into account {@link Cell} encoding.
* In particular, this method doesn't consider two types serialization compatible if one of them has fixed * In particular, this method doesn't consider two types serialization compatible if one of them has fixed
* length (overrides {@link #valueLengthIfFixed()}, and the other one doesn't. * length, and the other one doesn't.
*/ */
public boolean isSerializationCompatibleWith(AbstractType<?> previous) public boolean isSerializationCompatibleWith(AbstractType<?> previous)
{ {
return isValueCompatibleWith(previous) return isValueCompatibleWith(previous)
&& valueLengthIfFixed() == previous.valueLengthIfFixed() && valueLengthIfFixed == previous.valueLengthIfFixed
&& isMultiCell() == previous.isMultiCell(); && isMultiCell() == previous.isMultiCell();
} }
@ -498,7 +505,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
*/ */
public int valueLengthIfFixed() public int valueLengthIfFixed()
{ {
return VARIABLE_LENGTH; return valueLengthIfFixed;
} }
/** /**
@ -508,7 +515,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
*/ */
public final boolean isValueLengthFixed() public final boolean isValueLengthFixed()
{ {
return valueLengthIfFixed() != VARIABLE_LENGTH; return valueLengthIfFixed != VARIABLE_LENGTH;
} }
/** /**
@ -570,7 +577,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public <V> void writeValue(V value, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException public <V> void writeValue(V value, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException
{ {
assert !isNull(value, accessor) : "bytes should not be null for type " + this; assert !isNull(value, accessor) : "bytes should not be null for type " + this;
int expectedValueLength = valueLengthIfFixed(); int expectedValueLength = valueLengthIfFixed;
if (expectedValueLength >= 0) if (expectedValueLength >= 0)
{ {
int actualValueLength = accessor.size(value); int actualValueLength = accessor.size(value);
@ -589,7 +596,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public <V> void writeValue(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException public <V> void writeValue(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor, DataOutputPlus out) throws IOException
{ {
assert !valueHolder.isNull(i) : "bytes should not be null for type " + this; assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
int expectedValueLength = valueLengthIfFixed(); int expectedValueLength = valueLengthIfFixed;
if (expectedValueLength >= 0) if (expectedValueLength >= 0)
{ {
int actualValueLength = valueHolder.size(i); int actualValueLength = valueHolder.size(i);
@ -613,7 +620,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public <V> long writtenLength(V value, ValueAccessor<V> accessor) public <V> long writtenLength(V value, ValueAccessor<V> accessor)
{ {
assert !accessor.isEmpty(value) : "bytes should not be empty for type " + this; assert !accessor.isEmpty(value) : "bytes should not be empty for type " + this;
return valueLengthIfFixed() >= 0 return valueLengthIfFixed >= 0
? accessor.size(value) // if the size is wrong, this will be detected in writeValue ? accessor.size(value) // if the size is wrong, this will be detected in writeValue
: accessor.sizeWithVIntLength(value); : accessor.sizeWithVIntLength(value);
} }
@ -621,7 +628,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public <V> long writtenLength(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor) public <V> long writtenLength(IndexedValueHolder<V> valueHolder, int i, ValueAccessor<V> accessor)
{ {
assert !valueHolder.isNull(i) : "bytes should not be null for type " + this; assert !valueHolder.isNull(i) : "bytes should not be null for type " + this;
return valueLengthIfFixed() >= 0 return valueLengthIfFixed >= 0
? valueHolder.size(i) // if the size is wrong, this will be detected in writeValue ? valueHolder.size(i) // if the size is wrong, this will be detected in writeValue
: accessor.sizeWithVIntLength(valueHolder, i); : accessor.sizeWithVIntLength(valueHolder, i);
} }
@ -643,7 +650,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public <V> V read(ValueAccessor<V> accessor, DataInputPlus in, int maxValueSize) throws IOException public <V> V read(ValueAccessor<V> accessor, DataInputPlus in, int maxValueSize) throws IOException
{ {
int length = valueLengthIfFixed(); int length = valueLengthIfFixed;
if (length >= 0) if (length >= 0)
return accessor.read(in, length); return accessor.read(in, length);
@ -664,7 +671,7 @@ public abstract class AbstractType<T> implements Comparator<ByteBuffer>, Assignm
public void skipValue(DataInputPlus in) throws IOException public void skipValue(DataInputPlus in) throws IOException
{ {
int length = valueLengthIfFixed(); int length = valueLengthIfFixed;
if (length >= 0) if (length >= 0)
in.skipBytesFully(length); in.skipBytesFully(length);
else else

View File

@ -36,7 +36,7 @@ public class BooleanType extends AbstractType<Boolean>
private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance); private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance);
private static final ByteBuffer MASKED_VALUE = instance.decompose(false); private static final ByteBuffer MASKED_VALUE = instance.decompose(false);
BooleanType() {super(ComparisonType.CUSTOM);} // singleton BooleanType() {super(ComparisonType.CUSTOM, 1);} // singleton
@Override @Override
public boolean allowsEmpty() public boolean allowsEmpty()
@ -127,12 +127,6 @@ public class BooleanType extends AbstractType<Boolean>
return ARGUMENT_DESERIALIZER; return ARGUMENT_DESERIALIZER;
} }
@Override
public int valueLengthIfFixed()
{
return 1;
}
@Override @Override
public ByteBuffer getMaskedValue() public ByteBuffer getMaskedValue()
{ {

View File

@ -49,7 +49,7 @@ public class DateType extends AbstractType<Date>
private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance); private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance);
private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0)); private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
DateType() {super(ComparisonType.BYTE_ORDER);} // singleton DateType() {super(ComparisonType.BYTE_ORDER, 8);} // singleton
public boolean isEmptyValueMeaningless() public boolean isEmptyValueMeaningless()
{ {
@ -143,12 +143,6 @@ public class DateType extends AbstractType<Date>
return ARGUMENT_DESERIALIZER; return ARGUMENT_DESERIALIZER;
} }
@Override
public int valueLengthIfFixed()
{
return 8;
}
@Override @Override
public ByteBuffer getMaskedValue() public ByteBuffer getMaskedValue()
{ {

View File

@ -40,7 +40,7 @@ public class DoubleType extends NumberType<Double>
private static final ByteBuffer MASKED_VALUE = instance.decompose(0d); private static final ByteBuffer MASKED_VALUE = instance.decompose(0d);
DoubleType() {super(ComparisonType.CUSTOM);} // singleton DoubleType() {super(ComparisonType.CUSTOM, 8);} // singleton
@Override @Override
public boolean allowsEmpty() public boolean allowsEmpty()
@ -145,12 +145,6 @@ public class DoubleType extends NumberType<Double>
}; };
} }
@Override
public int valueLengthIfFixed()
{
return 8;
}
@Override @Override
public ByteBuffer add(Number left, Number right) public ByteBuffer add(Number left, Number right)
{ {

View File

@ -71,7 +71,7 @@ public class EmptyType extends AbstractType<Void>
public static final EmptyType instance = new EmptyType(); public static final EmptyType instance = new EmptyType();
private EmptyType() {super(ComparisonType.CUSTOM);} // singleton private EmptyType() {super(ComparisonType.CUSTOM, 0);} // singleton
@Override @Override
public <V> ByteSource asComparableBytes(ValueAccessor<V> accessor, V data, ByteComparable.Version version) public <V> ByteSource asComparableBytes(ValueAccessor<V> accessor, V data, ByteComparable.Version version)
@ -137,12 +137,6 @@ public class EmptyType extends AbstractType<Void>
throw new UnsupportedOperationException(); throw new UnsupportedOperationException();
} }
@Override
public int valueLengthIfFixed()
{
return 0;
}
@Override @Override
public <V> long writtenLength(V value, ValueAccessor<V> accessor) public <V> long writtenLength(V value, ValueAccessor<V> accessor)
{ {

View File

@ -41,7 +41,7 @@ public class FloatType extends NumberType<Float>
private static final ByteBuffer MASKED_VALUE = instance.decompose(0f); private static final ByteBuffer MASKED_VALUE = instance.decompose(0f);
FloatType() {super(ComparisonType.CUSTOM);} // singleton FloatType() {super(ComparisonType.CUSTOM, 4);} // singleton
@Override @Override
public boolean allowsEmpty() public boolean allowsEmpty()
@ -146,12 +146,6 @@ public class FloatType extends NumberType<Float>
}; };
} }
@Override
public int valueLengthIfFixed()
{
return 4;
}
@Override @Override
public ByteBuffer add(Number left, Number right) public ByteBuffer add(Number left, Number right)
{ {

View File

@ -43,7 +43,7 @@ public class Int32Type extends NumberType<Integer>
Int32Type() Int32Type()
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, 4);
} // singleton } // singleton
@Override @Override
@ -152,12 +152,6 @@ public class Int32Type extends NumberType<Integer>
}; };
} }
@Override
public int valueLengthIfFixed()
{
return 4;
}
@Override @Override
public ByteBuffer add(Number left, Number right) public ByteBuffer add(Number left, Number right)
{ {

View File

@ -44,7 +44,7 @@ public class LexicalUUIDType extends AbstractType<UUID>
LexicalUUIDType() LexicalUUIDType()
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, 16);
} // singleton } // singleton
@Override @Override
@ -148,12 +148,6 @@ public class LexicalUUIDType extends AbstractType<UUID>
return ARGUMENT_DESERIALIZER; return ARGUMENT_DESERIALIZER;
} }
@Override
public int valueLengthIfFixed()
{
return 16;
}
@Override @Override
public ByteBuffer getMaskedValue() public ByteBuffer getMaskedValue()
{ {

View File

@ -41,7 +41,7 @@ public class LongType extends NumberType<Long>
private static final ByteBuffer MASKED_VALUE = instance.decompose(0L); private static final ByteBuffer MASKED_VALUE = instance.decompose(0L);
LongType() {super(ComparisonType.CUSTOM);} // singleton LongType() {super(ComparisonType.CUSTOM, 8);} // singleton
@Override @Override
public boolean allowsEmpty() public boolean allowsEmpty()
@ -170,12 +170,6 @@ public class LongType extends NumberType<Long>
}; };
} }
@Override
public int valueLengthIfFixed()
{
return 8;
}
@Override @Override
public ByteBuffer add(Number left, Number right) public ByteBuffer add(Number left, Number right)
{ {

View File

@ -38,6 +38,11 @@ public abstract class MultiElementType<T> extends AbstractType<T>
super(comparisonType); super(comparisonType);
} }
protected MultiElementType(ComparisonType comparisonType, int valueLengthIfFixed)
{
super(comparisonType, valueLengthIfFixed);
}
/** /**
* Returns the serialized representation of the value composed of the specified elements. * Returns the serialized representation of the value composed of the specified elements.
* *
@ -133,4 +138,3 @@ public abstract class MultiElementType<T> extends AbstractType<T>
throw new UnsupportedOperationException(this + " does not support retrieving elements by key or index"); throw new UnsupportedOperationException(this + " does not support retrieving elements by key or index");
} }
} }

View File

@ -34,6 +34,11 @@ public abstract class NumberType<T extends Number> extends AbstractType<T>
super(comparisonType); super(comparisonType);
} }
protected NumberType(ComparisonType comparisonType, int valueLengthIfFixed)
{
super(comparisonType, valueLengthIfFixed);
}
/** /**
* Checks if this type support floating point numbers. * Checks if this type support floating point numbers.
* @return {@code true} if this type support floating point numbers, {@code false} otherwise. * @return {@code true} if this type support floating point numbers, {@code false} otherwise.

View File

@ -57,7 +57,7 @@ public class ReversedType<T> extends AbstractType<T>
private ReversedType(AbstractType<T> baseType) private ReversedType(AbstractType<T> baseType)
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, baseType.valueLengthIfFixed());
this.baseType = baseType; this.baseType = baseType;
} }
@ -181,12 +181,6 @@ public class ReversedType<T> extends AbstractType<T>
return getInstance(baseType.withUpdatedUserType(udt)); return getInstance(baseType.withUpdatedUserType(udt));
} }
@Override
public int valueLengthIfFixed()
{
return baseType.valueLengthIfFixed();
}
@Override @Override
public boolean isReversed() public boolean isReversed()
{ {

View File

@ -37,6 +37,11 @@ public abstract class TemporalType<T> extends AbstractType<T>
super(comparisonType); super(comparisonType);
} }
protected TemporalType(ComparisonType comparisonType, int valueLengthIfFixed)
{
super(comparisonType, valueLengthIfFixed);
}
/** /**
* Returns the current temporal value. * Returns the current temporal value.
* @return the current temporal value. * @return the current temporal value.

View File

@ -53,7 +53,7 @@ public class TimestampType extends TemporalType<Date>
private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0)); private static final ByteBuffer MASKED_VALUE = instance.decompose(new Date(0));
private TimestampType() {super(ComparisonType.CUSTOM);} // singleton private TimestampType() {super(ComparisonType.CUSTOM, 8);} // singleton
@Override @Override
public boolean allowsEmpty() public boolean allowsEmpty()
@ -166,12 +166,6 @@ public class TimestampType extends TemporalType<Date>
return TimestampSerializer.instance; return TimestampSerializer.instance;
} }
@Override
public int valueLengthIfFixed()
{
return 8;
}
@Override @Override
protected void validateDuration(Duration duration) protected void validateDuration(Duration duration)
{ {

View File

@ -56,7 +56,7 @@ public class UUIDType extends AbstractType<UUID>
UUIDType() UUIDType()
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, 16);
} }
@Override @Override
@ -252,12 +252,6 @@ public class UUIDType extends AbstractType<UUID>
return (uuid.get(6) & 0xf0) >> 4; return (uuid.get(6) & 0xf0) >> 4;
} }
@Override
public int valueLengthIfFixed()
{
return 16;
}
@Override @Override
public ByteBuffer getMaskedValue() public ByteBuffer getMaskedValue()
{ {

View File

@ -82,20 +82,16 @@ public final class VectorType<T> extends MultiElementType<List<T>>
public final AbstractType<T> elementType; public final AbstractType<T> elementType;
public final int dimension; public final int dimension;
private final TypeSerializer<T> elementSerializer; private final TypeSerializer<T> elementSerializer;
private final int valueLengthIfFixed;
private final VectorSerializer serializer; private final VectorSerializer serializer;
private VectorType(AbstractType<T> elementType, int dimension) private VectorType(AbstractType<T> elementType, int dimension)
{ {
super(ComparisonType.CUSTOM); super(ComparisonType.CUSTOM, valueLengthIfFixed(elementType, dimension));
if (dimension <= 0) if (dimension <= 0)
throw new InvalidRequestException(String.format("vectors may only have positive dimensions; given %d", dimension)); throw new InvalidRequestException(String.format("vectors may only have positive dimensions; given %d", dimension));
this.elementType = elementType; this.elementType = elementType;
this.dimension = dimension; this.dimension = dimension;
this.elementSerializer = elementType.getSerializer(); this.elementSerializer = elementType.getSerializer();
this.valueLengthIfFixed = elementType.isValueLengthFixed() ?
elementType.valueLengthIfFixed() * dimension :
super.valueLengthIfFixed();
this.serializer = elementType.isValueLengthFixed() ? this.serializer = elementType.isValueLengthFixed() ?
new FixedLengthSerializer() : new FixedLengthSerializer() :
new VariableLengthSerializer(); new VariableLengthSerializer();
@ -126,10 +122,10 @@ public final class VectorType<T> extends MultiElementType<List<T>>
return getSerializer().compareCustom(left, accessorL, right, accessorR); return getSerializer().compareCustom(left, accessorL, right, accessorR);
} }
@Override private static int valueLengthIfFixed(AbstractType<?> elementType, int dimension)
public int valueLengthIfFixed()
{ {
return valueLengthIfFixed; int elementLength = elementType.valueLengthIfFixed();
return elementLength >= 0 ? elementLength * dimension : elementLength;
} }
@Override @Override

View File

@ -61,7 +61,18 @@ public abstract class AbstractCell<V> extends Cell<V>
public boolean isTombstone() public boolean isTombstone()
{ {
return localDeletionTime() != NO_DELETION_TIME && ttl() == NO_TTL; return isTombstone(localDeletionTime());
}
public long minDeletionTime()
{
long localDeletionTime = localDeletionTime();
return isTombstone(localDeletionTime) ? Long.MIN_VALUE : localDeletionTime;
}
private boolean isTombstone(long localDeletionTime)
{
return localDeletionTime != NO_DELETION_TIME && ttl() == NO_TTL;
} }
public boolean isExpiring() public boolean isExpiring()

View File

@ -31,17 +31,12 @@ import org.apache.cassandra.utils.memory.ByteBufferCloner;
import static org.apache.cassandra.utils.ByteArrayUtil.EMPTY_BYTE_ARRAY; import static org.apache.cassandra.utils.ByteArrayUtil.EMPTY_BYTE_ARRAY;
public class ArrayCell extends AbstractCell<byte[]> public class ArrayCell extends HeapAbstractCell<byte[]>
{ {
private static final long EMPTY_SIZE = ObjectSizes.measure(new ArrayCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, EMPTY_BYTE_ARRAY, null)); private static final long EMPTY_SIZE = ObjectSizes.measure(new ArrayCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, EMPTY_BYTE_ARRAY, null));
// Careful: Adding vars here has an impact on memtable size // Careful: Adding vars here has an impact on memtable size
private final long timestamp;
private final int ttl;
private final int localDeletionTimeUnsignedInteger;
private final byte[] value; private final byte[] value;
private final CellPath path;
// Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not // Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not
// available. // available.
@ -52,12 +47,8 @@ public class ArrayCell extends AbstractCell<byte[]>
public ArrayCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, byte[] value, CellPath path) public ArrayCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, byte[] value, CellPath path)
{ {
super(column); super(column, timestamp, ttl, localDeletionTimeUnsignedInteger, path);
this.timestamp = timestamp;
this.ttl = ttl;
this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
this.value = value; this.value = value;
this.path = path;
} }
public static ArrayCell live(ColumnMetadata column, long timestamp, byte[] value, CellPath path) public static ArrayCell live(ColumnMetadata column, long timestamp, byte[] value, CellPath path)
@ -71,16 +62,6 @@ public class ArrayCell extends AbstractCell<byte[]>
return new ArrayCell(column, timestamp, ttl, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl), value, path); return new ArrayCell(column, timestamp, ttl, ExpirationDateOverflowHandling.computeLocalExpirationTime(nowInSec, ttl), value, path);
} }
public long timestamp()
{
return timestamp;
}
public int ttl()
{
return ttl;
}
public byte[] value() public byte[] value()
{ {
return value; return value;
@ -91,10 +72,6 @@ public class ArrayCell extends AbstractCell<byte[]>
return ByteArrayAccessor.instance; return ByteArrayAccessor.instance;
} }
public CellPath path()
{
return path;
}
public Cell<?> withUpdatedColumn(ColumnMetadata newColumn) public Cell<?> withUpdatedColumn(ColumnMetadata newColumn)
{ {
@ -144,9 +121,4 @@ public class ArrayCell extends AbstractCell<byte[]>
return EMPTY_SIZE + ObjectSizes.sizeOfArray(value) - value.length + (path == null ? 0 : path.unsharedHeapSizeExcludingData()); return EMPTY_SIZE + ObjectSizes.sizeOfArray(value) - value.length + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
} }
@Override
protected int localDeletionTimeAsUnsignedInt()
{
return localDeletionTimeUnsignedInteger;
}
} }

View File

@ -172,7 +172,7 @@ public class BTreeRow extends AbstractRow
private static long minDeletionTime(Cell<?> cell) private static long minDeletionTime(Cell<?> cell)
{ {
return cell.isTombstone() ? Long.MIN_VALUE : cell.localDeletionTime(); return cell.minDeletionTime();
} }
private static long minDeletionTime(LivenessInfo info) private static long minDeletionTime(LivenessInfo info)
@ -439,6 +439,11 @@ public class BTreeRow extends AbstractRow
return nowInSec >= minLocalDeletionTime; return nowInSec >= minLocalDeletionTime;
} }
public long minLocalDeletionTime()
{
return minLocalDeletionTime;
}
public boolean hasInvalidDeletions() public boolean hasInvalidDeletions()
{ {
if (primaryKeyLivenessInfo().isExpiring() && (primaryKeyLivenessInfo().ttl() < 0 || primaryKeyLivenessInfo().localExpirationTime() < 0)) if (primaryKeyLivenessInfo().isExpiring() && (primaryKeyLivenessInfo().ttl() < 0 || primaryKeyLivenessInfo().localExpirationTime() < 0))

View File

@ -30,17 +30,12 @@ import org.apache.cassandra.utils.memory.ByteBufferCloner;
import static java.lang.String.format; import static java.lang.String.format;
public class BufferCell extends AbstractCell<ByteBuffer> public class BufferCell extends HeapAbstractCell<ByteBuffer>
{ {
private static final long EMPTY_SIZE = ObjectSizes.measure(new BufferCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, null)); private static final long EMPTY_SIZE = ObjectSizes.measure(new BufferCell(ColumnMetadata.regularColumn("", "", "", ByteType.instance, ColumnMetadata.NO_UNIQUE_ID), 0L, 0, 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, null));
// Careful: Adding vars here has an impact on memtable size // Careful: Adding vars here has an impact on memtable size
private final long timestamp;
private final int ttl;
private final int localDeletionTimeUnsignedInteger;
private final ByteBuffer value; private final ByteBuffer value;
private final CellPath path;
// Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not // Please keep both int/long overloaded ctros public. Otherwise silent casts will mess timestamps when one is not
// available. // available.
@ -51,14 +46,10 @@ public class BufferCell extends AbstractCell<ByteBuffer>
public BufferCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, ByteBuffer value, CellPath path) public BufferCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, ByteBuffer value, CellPath path)
{ {
super(column); super(column, timestamp, ttl, localDeletionTimeUnsignedInteger, path);
assert !column.isPrimaryKeyColumn(); assert !column.isPrimaryKeyColumn();
assert column.isComplex() == (path != null) : format("Column %s.%s(%s: %s) isComplex: %b with cellpath: %s", column.ksName, column.cfName, column.name, column.type.toString(), column.isComplex(), path); assert column.isComplex() == (path != null) : format("Column %s.%s(%s: %s) isComplex: %b with cellpath: %s", column.ksName, column.cfName, column.name, column.type.toString(), column.isComplex(), path);
this.timestamp = timestamp;
this.ttl = ttl;
this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
this.value = value; this.value = value;
this.path = path;
} }
public static BufferCell live(ColumnMetadata column, long timestamp, ByteBuffer value) public static BufferCell live(ColumnMetadata column, long timestamp, ByteBuffer value)
@ -92,16 +83,6 @@ public class BufferCell extends AbstractCell<ByteBuffer>
return new BufferCell(column, timestamp, NO_TTL, nowInSec, ByteBufferUtil.EMPTY_BYTE_BUFFER, path); return new BufferCell(column, timestamp, NO_TTL, nowInSec, ByteBufferUtil.EMPTY_BYTE_BUFFER, path);
} }
public long timestamp()
{
return timestamp;
}
public int ttl()
{
return ttl;
}
public ByteBuffer value() public ByteBuffer value()
{ {
return value; return value;
@ -112,10 +93,6 @@ public class BufferCell extends AbstractCell<ByteBuffer>
return ByteBufferAccessor.instance; return ByteBufferAccessor.instance;
} }
public CellPath path()
{
return path;
}
public Cell<?> withUpdatedColumn(ColumnMetadata newColumn) public Cell<?> withUpdatedColumn(ColumnMetadata newColumn)
{ {
@ -163,10 +140,4 @@ public class BufferCell extends AbstractCell<ByteBuffer>
{ {
return EMPTY_SIZE + ObjectSizes.sizeOnHeapExcludingDataOf(value) + (path == null ? 0 : path.unsharedHeapSizeExcludingData()); return EMPTY_SIZE + ObjectSizes.sizeOnHeapExcludingDataOf(value) + (path == null ? 0 : path.unsharedHeapSizeExcludingData());
} }
@Override
protected int localDeletionTimeAsUnsignedInt()
{
return localDeletionTimeUnsignedInteger;
}
} }

View File

@ -150,6 +150,8 @@ public abstract class Cell<V> extends ColumnData
return deletionTimeUnsignedIntegerToLong(localDeletionTimeAsUnsignedInt()); return deletionTimeUnsignedIntegerToLong(localDeletionTimeAsUnsignedInt());
} }
public abstract long minDeletionTime();
/** /**
* Whether the cell is a tombstone or not. * Whether the cell is a tombstone or not.
* *

View File

@ -0,0 +1,64 @@
/*
* 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 org.apache.cassandra.schema.ColumnMetadata;
public abstract class HeapAbstractCell<V> extends AbstractCell<V>
{
// Careful: Adding vars here has an impact on memtable size
protected final long timestamp;
protected final int ttl;
protected final int localDeletionTimeUnsignedInteger;
protected final CellPath path;
protected HeapAbstractCell(ColumnMetadata column, long timestamp, int ttl, int localDeletionTimeUnsignedInteger, CellPath path)
{
super(column);
this.timestamp = timestamp;
this.ttl = ttl;
this.localDeletionTimeUnsignedInteger = localDeletionTimeUnsignedInteger;
this.path = path;
}
@Override
public long timestamp()
{
return timestamp;
}
@Override
public int ttl()
{
return ttl;
}
@Override
protected int localDeletionTimeAsUnsignedInt()
{
return localDeletionTimeUnsignedInteger;
}
@Override
public CellPath path()
{
return path;
}
}

View File

@ -20,7 +20,6 @@ package org.apache.cassandra.db.rows;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collection; import java.util.Collection;
import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
@ -44,9 +43,10 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.paxos.Commit; import org.apache.cassandra.service.paxos.Commit;
import org.apache.cassandra.utils.BiLongAccumulator; import org.apache.cassandra.utils.BiLongAccumulator;
import org.apache.cassandra.utils.BulkIterator; import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.ComplexCellMergeIterator;
import org.apache.cassandra.utils.LongAccumulator; import org.apache.cassandra.utils.LongAccumulator;
import org.apache.cassandra.utils.MergeIterator;
import org.apache.cassandra.utils.ObjectSizes; import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.RowMergeIterator;
import org.apache.cassandra.utils.SearchIterator; import org.apache.cassandra.utils.SearchIterator;
import org.apache.cassandra.utils.btree.BTree; import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction; import org.apache.cassandra.utils.btree.UpdateFunction;
@ -221,6 +221,15 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
*/ */
public boolean hasDeletion(long nowInSec); public boolean hasDeletion(long nowInSec);
/**
* The smallest local deletion time of all the data in this row (row deletion, primary key liveness, cells and
* complex deletions), or {@link Cell#MAX_DELETION_TIME} if the row has no deletion nor expiring data.
* <p>
* Unlike {@link #hasDeletion(long)}, this value is independent of the current time. In particular, a value of
* {@link Cell#MAX_DELETION_TIME} guarantees the row carries neither tombstones nor expiring data.
*/
public long minLocalDeletionTime();
/** /**
* An iterator to efficiently search data for a given column. * An iterator to efficiently search data for a given column.
* *
@ -762,6 +771,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
LivenessInfo rowInfo = LivenessInfo.EMPTY; LivenessInfo rowInfo = LivenessInfo.EMPTY;
Deletion rowDeletion = Deletion.LIVE; Deletion rowDeletion = Deletion.LIVE;
int columnsCountEstimation = 0;
// Track the smallest local deletion time across all inputs: if none of them carries any deletion or
// expiring data (i.e. this stays at MAX_DELETION_TIME), the merged row can't either, so we can hand the
// value to BTreeRow.create() below and skip the full btree scan it would otherwise do to recompute it.
long minDeletionTime = Cell.MAX_DELETION_TIME;
for (Row row : rows) for (Row row : rows)
{ {
if (row == null) if (row == null)
@ -771,6 +785,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
rowInfo = row.primaryKeyLivenessInfo(); rowInfo = row.primaryKeyLivenessInfo();
if (row.deletion().supersedes(rowDeletion)) if (row.deletion().supersedes(rowDeletion))
rowDeletion = row.deletion(); rowDeletion = row.deletion();
minDeletionTime = Math.min(minDeletionTime, row.minLocalDeletionTime());
columnDataIterators.add(row.iterator());
columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
} }
if (rowDeletion.isShadowedBy(rowInfo)) if (rowDeletion.isShadowedBy(rowInfo))
@ -784,25 +803,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (activeDeletion.deletes(rowInfo)) if (activeDeletion.deletes(rowInfo))
rowInfo = LivenessInfo.EMPTY; rowInfo = LivenessInfo.EMPTY;
int columnsCountEstimation = 0;
for (Row row : rows)
{
if (row != null)
{
columnDataIterators.add(row.iterator());
columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
}
else
{
columnDataIterators.add(Collections.emptyIterator());
}
}
// try to estimate and set a potential target capacity // try to estimate and set a potential target capacity
if (dataBuffer.length < columnsCountEstimation) if (dataBuffer.length < columnsCountEstimation)
dataBuffer = new ColumnData[columnsCountEstimation]; dataBuffer = new ColumnData[columnsCountEstimation];
columnDataReducer.setActiveDeletion(activeDeletion); columnDataReducer.setActiveDeletion(activeDeletion);
Iterator<ColumnData> merged = MergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer); Iterator<ColumnData> merged = RowMergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
while (merged.hasNext()) while (merged.hasNext())
{ {
ColumnData data = merged.next(); ColumnData data = merged.next();
@ -819,8 +825,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer)) try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer))
{ {
return BTreeRow.create(clustering, rowInfo, rowDeletion, Object[] tree = BTree.build(it, dataBufferSize, UpdateFunction.noOp());
BTree.build(it, dataBufferSize, UpdateFunction.noOp())); // If none of the merged rows had any deletion or expiring data, neither does the result, so we can
// pass the already-known min local deletion time and avoid rescanning the whole btree to recompute it.
return minDeletionTime == Cell.MAX_DELETION_TIME
? BTreeRow.create(clustering, rowInfo, rowDeletion, tree, Cell.MAX_DELETION_TIME)
: BTreeRow.create(clustering, rowInfo, rowDeletion, tree);
} }
} }
@ -841,10 +851,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
return rows; return rows;
} }
private static class ColumnDataReducer extends MergeIterator.Reducer<ColumnData, ColumnData> private static class ColumnDataReducer extends RowMergeIterator.Reducer<ColumnData, ColumnData>
{ {
private ColumnMetadata column; private ColumnMetadata column;
private final List<ColumnData> versions; private final ColumnData[] versions;
private int versionsSize;
private DeletionTime activeDeletion; private DeletionTime activeDeletion;
@ -854,7 +865,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
public ColumnDataReducer(int size, boolean hasComplex) public ColumnDataReducer(int size, boolean hasComplex)
{ {
this.versions = new ArrayList<>(size); this.versions = new ColumnData[size];
this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null; this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null;
this.complexCells = hasComplex ? new ArrayList<>(size) : null; this.complexCells = hasComplex ? new ArrayList<>(size) : null;
this.cellReducer = new CellReducer(); this.cellReducer = new CellReducer();
@ -870,20 +881,24 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (useColumnMetadata(data.column())) if (useColumnMetadata(data.column()))
column = data.column(); column = data.column();
versions.add(data); versions[versionsSize++] = data;
} }
/** /**
* Determines it the {@code ColumnMetadata} is the one that should be used. * Determines whether {@code dataColumn} should replace the currently selected column metadata,
* @param dataColumn the {@code ColumnMetadata} to use. * i.e. whether no column has been selected yet or {@code dataColumn} is a newer version.
* @return {@code true} if the {@code ColumnMetadata} is the one that should be used, {@code false} otherwise. * @param dataColumn the candidate {@code ColumnMetadata} to evaluate.
* @return {@code true} if {@code dataColumn} should be used, {@code false} otherwise.
*/ */
private boolean useColumnMetadata(ColumnMetadata dataColumn) private boolean useColumnMetadata(ColumnMetadata dataColumn)
{ {
if (column == null) ColumnMetadata currentColumn = column;
if (currentColumn == null)
return true; return true;
if (currentColumn == dataColumn)
return false;
return ColumnMetadataVersionComparator.INSTANCE.compare(column, dataColumn) < 0; return ColumnMetadataVersionComparator.INSTANCE.compare(currentColumn, dataColumn) < 0;
} }
protected ColumnData getReduced() protected ColumnData getReduced()
@ -891,9 +906,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
if (column.isSimple()) if (column.isSimple())
{ {
Cell<?> merged = null; Cell<?> merged = null;
for (int i=0, isize=versions.size(); i<isize; i++) for (int i = 0; i < versionsSize; i++)
{ {
Cell<?> cell = (Cell<?>) versions.get(i); Cell<?> cell = (Cell<?>) versions[i];
if (!activeDeletion.deletes(cell)) if (!activeDeletion.deletes(cell))
merged = merged == null ? cell : Cells.reconcile(merged, cell); merged = merged == null ? cell : Cells.reconcile(merged, cell);
} }
@ -904,9 +919,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
complexBuilder.newColumn(column); complexBuilder.newColumn(column);
complexCells.clear(); complexCells.clear();
DeletionTime complexDeletion = DeletionTime.LIVE; DeletionTime complexDeletion = DeletionTime.LIVE;
for (int i=0, isize=versions.size(); i<isize; i++) for (int i = 0; i < versionsSize; i++)
{ {
ColumnData data = versions.get(i); ColumnData data = versions[i];
ComplexColumnData cd = (ComplexColumnData)data; ComplexColumnData cd = (ComplexColumnData)data;
if (cd.complexDeletion().supersedes(complexDeletion)) if (cd.complexDeletion().supersedes(complexDeletion))
complexDeletion = cd.complexDeletion(); complexDeletion = cd.complexDeletion();
@ -923,7 +938,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
cellReducer.setActiveDeletion(activeDeletion); cellReducer.setActiveDeletion(activeDeletion);
} }
Iterator<Cell<?>> cells = MergeIterator.get(complexCells, Cell.comparator, cellReducer); Iterator<Cell<?>> cells = ComplexCellMergeIterator.get(complexCells, Cell.comparator, cellReducer);
while (cells.hasNext()) while (cells.hasNext())
{ {
Cell<?> merged = cells.next(); Cell<?> merged = cells.next();
@ -937,11 +952,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
protected void onKeyChange() protected void onKeyChange()
{ {
column = null; column = null;
versions.clear(); Arrays.fill(versions, 0, versionsSize, null);
versionsSize = 0;
} }
} }
private static class CellReducer extends MergeIterator.Reducer<Cell<?>, Cell<?>> private static class CellReducer extends ComplexCellMergeIterator.Reducer<Cell<?>, Cell<?>>
{ {
private DeletionTime activeDeletion; private DeletionTime activeDeletion;
private Cell<?> merged; private Cell<?> merged;

View File

@ -38,7 +38,7 @@ import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.serializers.MarshalException; import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.IMergeIterator; import org.apache.cassandra.utils.IMergeIterator;
import org.apache.cassandra.utils.MergeIterator; import org.apache.cassandra.utils.UnfilteredMergeIterator;
/** /**
* Static methods to work with atom iterators. * Static methods to work with atom iterators.
@ -415,7 +415,7 @@ public abstract class UnfilteredRowIterators
reversed, reversed,
EncodingStats.merge(iterators, UnfilteredRowIterator::stats)); EncodingStats.merge(iterators, UnfilteredRowIterator::stats));
this.mergeIterator = MergeIterator.get(iterators, this.mergeIterator = UnfilteredMergeIterator.get(iterators,
reversed ? metadata.comparator.reversed() : metadata.comparator, reversed ? metadata.comparator.reversed() : metadata.comparator,
new MergeReducer(iterators.size(), reversed, listener)); new MergeReducer(iterators.size(), reversed, listener));
this.listener = listener; this.listener = listener;
@ -540,7 +540,7 @@ public abstract class UnfilteredRowIterators
listener.close(); listener.close();
} }
private class MergeReducer extends MergeIterator.Reducer<Unfiltered, Unfiltered> private class MergeReducer extends UnfilteredMergeIterator.Reducer<Unfiltered, Unfiltered>
{ {
private final MergeListener listener; private final MergeListener listener;

View File

@ -103,6 +103,12 @@ public class CellWithSource<T> extends Cell<T>
return cell.localDeletionTime(); return cell.localDeletionTime();
} }
@Override
public long minDeletionTime()
{
return cell.minDeletionTime();
}
@Override @Override
public boolean isTombstone() public boolean isTombstone()
{ {

View File

@ -213,6 +213,12 @@ public class RowWithSource implements Row
return row.hasDeletion(nowInSec); return row.hasDeletion(nowInSec);
} }
@Override
public long minLocalDeletionTime()
{
return row.minLocalDeletionTime();
}
@Override @Override
public SearchIterator<ColumnMetadata, ColumnData> searchIterator() public SearchIterator<ColumnMetadata, ColumnData> searchIterator()
{ {

View File

@ -92,6 +92,7 @@ public interface SingleNodeSequences
else if (InProgressSequences.isLeave(inProgress)) else if (InProgressSequences.isLeave(inProgress))
{ {
logger.info("Resuming decommission @ {} (current epoch = {}): {}", inProgress.latestModification, metadata.epoch, inProgress.status()); logger.info("Resuming decommission @ {} (current epoch = {}): {}", inProgress.latestModification, metadata.epoch, inProgress.status());
StorageService.instance.clearTransientMode();
} }
else else
{ {

View File

@ -21,6 +21,8 @@ import java.util.Comparator;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
import net.nicoulaj.compilecommand.annotations.DontInline;
import accord.utils.Invariants; import accord.utils.Invariants;
/** Merges sorted input iterators which individually contain unique items. */ /** Merges sorted input iterators which individually contain unique items. */
@ -307,9 +309,36 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
heap[currIdx] = heap[nextIdx]; heap[currIdx] = heap[nextIdx];
currIdx = nextIdx; currIdx = nextIdx;
} }
// If the candidate has no children in the binary-heap section it is already in its final position, so we can
// skip the (rarely needed) sink below. nextIdx is the candidate's left child; anything >= size means there
// are no children. The single-child (nextIdx == size - 1) and two-children cases still have to go through
// sinkInBinaryHeap, which compares against the child(ren) and may swap or update the equalParent flag.
nextIdx = (currIdx * 2) - (sortedSectionSize - 1);
if (nextIdx >= size)
{
heap[currIdx] = candidate;
return;
}
// The candidate did not settle within the sorted section; sink it through the binary-heap section. This is
// rarely reached for typical narrow, lightly-overlapping merges, so it is split into its own method to keep
// the hot sorted-section path above small enough to be inlined into advance().
sinkInBinaryHeap(candidate, currIdx, size, sortedSectionSize);
}
/**
* Sink {@code candidate} down the binary-heap section of the queue (the part below the sorted section), pulling
* up the lighter child at each level, and place it in its final position.
*
* Split out of {@link #replaceAndSink} so that the common case, where the candidate settles within the sorted
* section, stays compact and inlinable.
*/
@DontInline
private void sinkInBinaryHeap(Candidate<In> candidate, int currIdx, final int size, final int sortedSectionSize)
{
// If size <= SORTED_SECTION_SIZE, nextIdx below will be no less than size, // If size <= SORTED_SECTION_SIZE, nextIdx below will be no less than size,
// because currIdx == sortedSectionSize == size - 1 and nextIdx becomes // because currIdx == sortedSectionSize == size - 1 and nextIdx becomes
// (size - 1) * 2) - (size - 1 - 1) == size. // (size - 1) * 2) - (size - 1 - 1) == size.
int nextIdx;
// Advance in the binary heap, pulling up the lighter element from the two at each level. // Advance in the binary heap, pulling up the lighter element from the two at each level.
while ((nextIdx = (currIdx * 2) - (sortedSectionSize - 1)) + 1 < size) while ((nextIdx = (currIdx * 2) - (sortedSectionSize - 1)) + 1 < size)

View File

@ -21,6 +21,7 @@ package org.apache.cassandra.distributed.test;
import java.io.IOException; import java.io.IOException;
import java.util.concurrent.Callable; import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import net.bytebuddy.ByteBuddy; import net.bytebuddy.ByteBuddy;
import net.bytebuddy.dynamic.loading.ClassLoadingStrategy; import net.bytebuddy.dynamic.loading.ClassLoadingStrategy;
@ -141,6 +142,7 @@ public class DecommissionTest extends TestBaseImpl
public static class BB public static class BB
{ {
static final AtomicBoolean injected = new AtomicBoolean(false);
public static void install(ClassLoader classLoader, Integer num) public static void install(ClassLoader classLoader, Integer num)
{ {
if (num == 2) if (num == 2)
@ -157,7 +159,7 @@ public class DecommissionTest extends TestBaseImpl
public static void execute(NodeId leaving, PlacementDeltas startLeave, PlacementDeltas midLeave, PlacementDeltas finishLeave, public static void execute(NodeId leaving, PlacementDeltas startLeave, PlacementDeltas midLeave, PlacementDeltas finishLeave,
@SuperCall Callable<?> zuper) throws ExecutionException, InterruptedException @SuperCall Callable<?> zuper) throws ExecutionException, InterruptedException
{ {
if (!StorageService.instance.isDecommissionFailed()) if (injected.compareAndSet(false, true))
throw new ExecutionException(new RuntimeException("simulated error in prepareUnbootstrapStreaming")); throw new ExecutionException(new RuntimeException("simulated error in prepareUnbootstrapStreaming"));
try try

View File

@ -53,8 +53,10 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamSession; import org.apache.cassandra.streaming.StreamSession;
import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService; import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeId; import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.tcm.membership.NodeVersion; import org.apache.cassandra.tcm.membership.NodeVersion;
import org.apache.cassandra.tcm.transformations.PrepareLeave;
import org.apache.cassandra.tcm.transformations.Startup; import org.apache.cassandra.tcm.transformations.Startup;
import org.apache.cassandra.utils.CassandraVersion; import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.FBUtilities;
@ -62,6 +64,8 @@ import org.apache.cassandra.utils.FBUtilities;
import static net.bytebuddy.matcher.ElementMatchers.named; import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.cassandra.distributed.api.Feature.GOSSIP; import static org.apache.cassandra.distributed.api.Feature.GOSSIP;
import static org.apache.cassandra.distributed.api.Feature.NETWORK; import static org.apache.cassandra.distributed.api.Feature.NETWORK;
import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeCommit;
import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseCommits;
import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack; import static org.apache.cassandra.distributed.shared.NetworkTopology.dcAndRack;
import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology; import static org.apache.cassandra.distributed.shared.NetworkTopology.networkTopology;
import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate; import static org.apache.cassandra.distributed.test.ring.BootstrapTest.populate;
@ -84,6 +88,43 @@ public class DecommissionTest extends TestBaseImpl
} }
} }
@Test
public void testOperationModeOnDecomResume() throws Exception
{
// Node 2's first decomission attempt fails mid-stream (injected by BB), leaving it in
// DECOMISSION_FAILED. On resume, operationMode() must transition back to LEAVING before
// MID_LEAVE is committed. We pause the CMS just before that commit to assert the mode
// in the window where streaming has finished but the epoch has not yet advanced.
try (Cluster cluster = builder().withNodes(3)
.withConfig(config -> config.with(NETWORK, GOSSIP))
.withInstanceInitializer(BB::install)
.start())
{
populate(cluster, 0, 100, 1, 2, ConsistencyLevel.QUORUM);
IInvokableInstance cmsNode = cluster.get(1);
IInvokableInstance leavingNode = cluster.get(2);
// --force required: cie_internal keyspace has RF=3, decommission would fail replication check otherwise
leavingNode.nodetoolResult("decommission", "--force").asserts().failure();
leavingNode.runOnInstance(() -> assertEquals(StorageService.Mode.DECOMMISSION_FAILED, StorageService.instance.operationMode()));
Callable<Epoch> midLeavePaused = pauseBeforeCommit(cmsNode, e -> e instanceof PrepareLeave.MidLeave);
Thread resumeThread = new Thread(() -> leavingNode.nodetoolResult("decommission").asserts().success());
resumeThread.start();
midLeavePaused.call();
leavingNode.runOnInstance(() ->
assertEquals("operationMode during resumed decommission streaming should be LEAVING, not DECOMMISSION_FAILED",
StorageService.Mode.LEAVING,
StorageService.instance.operationMode()));
unpauseCommits(cmsNode);
resumeThread.join(TimeUnit.MINUTES.toMillis(2));
}
}
@Test @Test
public void testAddressReuseAfterDecommission() throws IOException, ExecutionException, InterruptedException public void testAddressReuseAfterDecommission() throws IOException, ExecutionException, InterruptedException
{ {

View File

@ -50,7 +50,7 @@ public class CMSShutdownTest extends TestBaseImpl
// This test simulates a CMS node attempting to commit an entry to the log but being unable // This test simulates a CMS node attempting to commit an entry to the log but being unable
// to obtain consensus from other CMS members while it is also shutting down itself. // to obtain consensus from other CMS members while it is also shutting down itself.
try (Cluster cluster = Cluster.build(2) try (Cluster cluster = Cluster.build(2)
.withConfig(c -> c.with(Feature.values())) .withConfig(c -> c.with(Feature.GOSSIP, Feature.NETWORK))
.withInstanceInitializer(BBHelper::install) .withInstanceInitializer(BBHelper::install)
.start()) .start())
{ {
@ -67,8 +67,15 @@ public class CMSShutdownTest extends TestBaseImpl
// latch ensures that every commit will fail as if unable to obtain consensus // latch ensures that every commit will fail as if unable to obtain consensus
// from other CMS members // from other CMS members
State.latch.countDown(); State.latch.countDown();
for (; ;) try
ClusterMetadataService.instance().commit(TriggerSnapshot.instance); {
while (!Thread.currentThread().isInterrupted())
ClusterMetadataService.instance().commit(TriggerSnapshot.instance);
}
catch (Throwable t)
{
// Commits fail by design; an interrupt or a stopped processor on shutdown ends them.
}
} }
public static void scheduleCommits() public static void scheduleCommits()

View File

@ -0,0 +1,94 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.test.microbench;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeUnit;
import org.openjdk.jmh.annotations.Benchmark;
import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.Fork;
import org.openjdk.jmh.annotations.Measurement;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.Setup;
import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.Threads;
import org.openjdk.jmh.annotations.Warmup;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.BooleanType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.utils.ByteBufferUtil;
/**
* Measures value serialization when one call site sees the fixed- and variable-width types commonly used by a table.
*/
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@Warmup(iterations = 5, time = 1)
@Measurement(iterations = 5, time = 1)
@Fork(value = 1, jvmArgsAppend = { "-Xmx1G", "-Djmh.executor=CUSTOM", "-Djmh.executor.class=org.apache.cassandra.test.microbench.FastThreadExecutor" })
@Threads(1)
@State(Scope.Thread)
public class AbstractTypeSerializationBench
{
private final AbstractType<?>[] types = { Int32Type.instance, LongType.instance, BooleanType.instance, UUIDType.instance,
UTF8Type.instance, Int32Type.instance, LongType.instance, UTF8Type.instance };
private final ByteBuffer[] values = { ByteBufferUtil.bytes(42), ByteBufferUtil.bytes(42L), ByteBuffer.wrap(new byte[] { 1 }),
ByteBuffer.wrap(new byte[16]), ByteBufferUtil.bytes("cassandra"), ByteBufferUtil.bytes(42),
ByteBufferUtil.bytes(42L), ByteBufferUtil.bytes("cassandra") };
private final ByteBuffer[] serializedValues = new ByteBuffer[types.length];
private final DataOutputBuffer out = new DataOutputBuffer();
private int index;
@Setup
public void setup() throws IOException
{
for (int i = 0; i < types.length; i++)
{
out.clear();
types[i].writeValue(values[i], out);
serializedValues[i] = out.asNewBuffer();
}
}
@Benchmark
public int writeValue() throws IOException
{
int i = index++ & (types.length - 1);
out.clear();
types[i].writeValue(values[i], out);
return out.getLength();
}
@Benchmark
public ByteBuffer readValue() throws IOException
{
int i = index++ & (types.length - 1);
return types[i].readBuffer(new DataInputBuffer(serializedValues[i], true));
}
}

View File

@ -739,6 +739,12 @@ public class AbstractTypeTest
}}); }});
} }
@Test
public void valueLengthIfFixedIsNotFinal() throws NoSuchMethodException
{
assertThat(Modifier.isFinal(AbstractType.class.getMethod("valueLengthIfFixed").getModifiers())).isFalse();
}
@Test @Test
@SuppressWarnings({"rawtypes", "unchecked"}) @SuppressWarnings({"rawtypes", "unchecked"})
public void serde() public void serde()

View File

@ -282,6 +282,57 @@ public class RowsTest
} }
} }
private static Row liveRow(Clustering<?> c, long ts, ByteBuffer vVal)
{
return rowWithCell(c, ts, BufferCell.live(v, ts, vVal));
}
private static Row rowWithCell(Clustering<?> c, long ts, Cell<?> cell)
{
Row.Builder builder = createBuilder(c);
builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(ts));
builder.addCell(cell);
return builder.build();
}
private static Row mergeRows(Row... rows)
{
boolean hasComplex = false;
for (Row row : rows)
hasComplex |= row.hasComplex();
Row.Merger merger = new Row.Merger(rows.length, hasComplex);
for (int i = 0; i < rows.length; i++)
merger.add(i, rows[i]);
return merger.merge(DeletionTime.LIVE);
}
@Test
public void testMergerMinLocalDeletionTime()
{
long now = FBUtilities.nowInSeconds();
long ts = secondToTs(now);
// All inputs are free of deletions and expiring data, so the merged row must be too. This is the fast path in
// Row.Merger#merge that reuses Cell.MAX_DELETION_TIME instead of recomputing it by scanning the merged btree.
Row mergedLive = mergeRows(liveRow(c1, ts, BB1), liveRow(c1, ts + 1, BB2));
Assert.assertEquals(Cell.MAX_DELETION_TIME, mergedLive.minLocalDeletionTime());
Assert.assertFalse(mergedLive.hasDeletion(now));
// One input carries an expiring cell that wins reconciliation (higher timestamp): the merged row must keep
// its expiration time rather than being reported as deletion-free.
Cell<?> expiringCell = BufferCell.expiring(v, ts + 2, 100, now, BB3);
Row mergedExpiring = mergeRows(liveRow(c1, ts, BB1), rowWithCell(c1, ts + 2, expiringCell));
Assert.assertEquals(expiringCell.localDeletionTime(), mergedExpiring.minLocalDeletionTime());
Assert.assertFalse(mergedExpiring.hasDeletion(now));
Assert.assertTrue(mergedExpiring.hasDeletion(expiringCell.localDeletionTime()));
// Same for a tombstone cell winning reconciliation: the merged row must report a deletion.
Cell<?> tombstoneCell = BufferCell.tombstone(v, ts + 2, now);
Row mergedTombstone = mergeRows(liveRow(c1, ts, BB1), rowWithCell(c1, ts + 2, tombstoneCell));
Assert.assertEquals(Long.MIN_VALUE, mergedTombstone.minLocalDeletionTime());
Assert.assertTrue(mergedTombstone.hasDeletion(now));
}
@Test @Test
public void diff() public void diff()
{ {