mirror of https://github.com/apache/cassandra
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
This commit is contained in:
Dmitry Konstantinov
parent
feef3fcf51
commit
1e3d43b2dd
|
|
@ -1,4 +1,5 @@
|
|||
6.0-alpha2
|
||||
* 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)
|
||||
|
|
|
|||
62
build.xml
62
build.xml
|
|
@ -651,6 +651,64 @@
|
|||
<jflex file="${build.src.java}/org/apache/cassandra/index/sasi/analyzer/StandardTokenizerImpl.jflex" destdir="${build.src.gen-java}/" />
|
||||
</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 -->
|
||||
<target name="_createVersionPropFile" depends="_get-git-sha,set-cqlsh-version,_set-build-date">
|
||||
<taskdef name="propertyfile" classname="org.apache.tools.ant.taskdefs.optional.PropertyFile"/>
|
||||
|
|
@ -710,7 +768,7 @@
|
|||
</javac>
|
||||
</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">
|
||||
<echo message="${ant.project.name}: ${ant.file}"/>
|
||||
<!-- Order matters! -->
|
||||
|
|
@ -2165,7 +2223,7 @@
|
|||
</target>
|
||||
|
||||
<!-- 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 file="${eclipse.project.name}.iml"/>
|
||||
<mkdir dir=".idea"/>
|
||||
|
|
|
|||
|
|
@ -178,7 +178,8 @@ public abstract class DeletionTime implements Comparable<DeletionTime>, IMeasura
|
|||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -439,6 +439,11 @@ public class BTreeRow extends AbstractRow
|
|||
return nowInSec >= minLocalDeletionTime;
|
||||
}
|
||||
|
||||
public long minLocalDeletionTime()
|
||||
{
|
||||
return minLocalDeletionTime;
|
||||
}
|
||||
|
||||
public boolean hasInvalidDeletions()
|
||||
{
|
||||
if (primaryKeyLivenessInfo().isExpiring() && (primaryKeyLivenessInfo().ttl() < 0 || primaryKeyLivenessInfo().localExpirationTime() < 0))
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ package org.apache.cassandra.db.rows;
|
|||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.Comparator;
|
||||
import java.util.Iterator;
|
||||
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.utils.BiLongAccumulator;
|
||||
import org.apache.cassandra.utils.BulkIterator;
|
||||
import org.apache.cassandra.utils.ComplexCellMergeIterator;
|
||||
import org.apache.cassandra.utils.LongAccumulator;
|
||||
import org.apache.cassandra.utils.MergeIterator;
|
||||
import org.apache.cassandra.utils.ObjectSizes;
|
||||
import org.apache.cassandra.utils.RowMergeIterator;
|
||||
import org.apache.cassandra.utils.SearchIterator;
|
||||
import org.apache.cassandra.utils.btree.BTree;
|
||||
import org.apache.cassandra.utils.btree.UpdateFunction;
|
||||
|
|
@ -221,6 +221,15 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
*/
|
||||
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.
|
||||
*
|
||||
|
|
@ -762,6 +771,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
|
||||
LivenessInfo rowInfo = LivenessInfo.EMPTY;
|
||||
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)
|
||||
{
|
||||
if (row == null)
|
||||
|
|
@ -771,6 +785,11 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
rowInfo = row.primaryKeyLivenessInfo();
|
||||
if (row.deletion().supersedes(rowDeletion))
|
||||
rowDeletion = row.deletion();
|
||||
|
||||
minDeletionTime = Math.min(minDeletionTime, row.minLocalDeletionTime());
|
||||
|
||||
columnDataIterators.add(row.iterator());
|
||||
columnsCountEstimation = Math.max(columnsCountEstimation, row.columnCount());
|
||||
}
|
||||
|
||||
if (rowDeletion.isShadowedBy(rowInfo))
|
||||
|
|
@ -784,25 +803,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
if (activeDeletion.deletes(rowInfo))
|
||||
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
|
||||
if (dataBuffer.length < columnsCountEstimation)
|
||||
dataBuffer = new ColumnData[columnsCountEstimation];
|
||||
|
||||
columnDataReducer.setActiveDeletion(activeDeletion);
|
||||
Iterator<ColumnData> merged = MergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
|
||||
Iterator<ColumnData> merged = RowMergeIterator.get(columnDataIterators, ColumnData.comparator, columnDataReducer);
|
||||
while (merged.hasNext())
|
||||
{
|
||||
ColumnData data = merged.next();
|
||||
|
|
@ -819,8 +825,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
|
||||
try (BulkIterator<ColumnData> it = BulkIterator.of(dataBuffer))
|
||||
{
|
||||
return BTreeRow.create(clustering, rowInfo, rowDeletion,
|
||||
BTree.build(it, dataBufferSize, UpdateFunction.noOp()));
|
||||
Object[] tree = 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;
|
||||
}
|
||||
|
||||
private static class ColumnDataReducer extends MergeIterator.Reducer<ColumnData, ColumnData>
|
||||
private static class ColumnDataReducer extends RowMergeIterator.Reducer<ColumnData, ColumnData>
|
||||
{
|
||||
private ColumnMetadata column;
|
||||
private final List<ColumnData> versions;
|
||||
private final ColumnData[] versions;
|
||||
private int versionsSize;
|
||||
|
||||
private DeletionTime activeDeletion;
|
||||
|
||||
|
|
@ -854,7 +865,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
|
||||
public ColumnDataReducer(int size, boolean hasComplex)
|
||||
{
|
||||
this.versions = new ArrayList<>(size);
|
||||
this.versions = new ColumnData[size];
|
||||
this.complexBuilder = hasComplex ? ComplexColumnData.builder() : null;
|
||||
this.complexCells = hasComplex ? new ArrayList<>(size) : null;
|
||||
this.cellReducer = new CellReducer();
|
||||
|
|
@ -870,20 +881,24 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
if (useColumnMetadata(data.column()))
|
||||
column = data.column();
|
||||
|
||||
versions.add(data);
|
||||
versions[versionsSize++] = data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines it the {@code ColumnMetadata} is the one that should be used.
|
||||
* @param dataColumn the {@code ColumnMetadata} to use.
|
||||
* @return {@code true} if the {@code ColumnMetadata} is the one that should be used, {@code false} otherwise.
|
||||
* Determines whether {@code dataColumn} should replace the currently selected column metadata,
|
||||
* i.e. whether no column has been selected yet or {@code dataColumn} is a newer version.
|
||||
* @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)
|
||||
{
|
||||
if (column == null)
|
||||
ColumnMetadata currentColumn = column;
|
||||
if (currentColumn == null)
|
||||
return true;
|
||||
if (currentColumn == dataColumn)
|
||||
return false;
|
||||
|
||||
return ColumnMetadataVersionComparator.INSTANCE.compare(column, dataColumn) < 0;
|
||||
return ColumnMetadataVersionComparator.INSTANCE.compare(currentColumn, dataColumn) < 0;
|
||||
}
|
||||
|
||||
protected ColumnData getReduced()
|
||||
|
|
@ -891,9 +906,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
if (column.isSimple())
|
||||
{
|
||||
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))
|
||||
merged = merged == null ? cell : Cells.reconcile(merged, cell);
|
||||
}
|
||||
|
|
@ -904,9 +919,9 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
complexBuilder.newColumn(column);
|
||||
complexCells.clear();
|
||||
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;
|
||||
if (cd.complexDeletion().supersedes(complexDeletion))
|
||||
complexDeletion = cd.complexDeletion();
|
||||
|
|
@ -923,7 +938,7 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
cellReducer.setActiveDeletion(activeDeletion);
|
||||
}
|
||||
|
||||
Iterator<Cell<?>> cells = MergeIterator.get(complexCells, Cell.comparator, cellReducer);
|
||||
Iterator<Cell<?>> cells = ComplexCellMergeIterator.get(complexCells, Cell.comparator, cellReducer);
|
||||
while (cells.hasNext())
|
||||
{
|
||||
Cell<?> merged = cells.next();
|
||||
|
|
@ -937,11 +952,12 @@ public interface Row extends Unfiltered, Iterable<ColumnData>, IMeasurableMemory
|
|||
protected void onKeyChange()
|
||||
{
|
||||
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 Cell<?> merged;
|
||||
|
|
|
|||
|
|
@ -38,7 +38,7 @@ import org.apache.cassandra.schema.TableMetadata;
|
|||
import org.apache.cassandra.serializers.MarshalException;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
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.
|
||||
|
|
@ -415,7 +415,7 @@ public abstract class UnfilteredRowIterators
|
|||
reversed,
|
||||
EncodingStats.merge(iterators, UnfilteredRowIterator::stats));
|
||||
|
||||
this.mergeIterator = MergeIterator.get(iterators,
|
||||
this.mergeIterator = UnfilteredMergeIterator.get(iterators,
|
||||
reversed ? metadata.comparator.reversed() : metadata.comparator,
|
||||
new MergeReducer(iterators.size(), reversed, listener));
|
||||
this.listener = listener;
|
||||
|
|
@ -540,7 +540,7 @@ public abstract class UnfilteredRowIterators
|
|||
listener.close();
|
||||
}
|
||||
|
||||
private class MergeReducer extends MergeIterator.Reducer<Unfiltered, Unfiltered>
|
||||
private class MergeReducer extends UnfilteredMergeIterator.Reducer<Unfiltered, Unfiltered>
|
||||
{
|
||||
private final MergeListener listener;
|
||||
|
||||
|
|
|
|||
|
|
@ -213,6 +213,12 @@ public class RowWithSource implements Row
|
|||
return row.hasDeletion(nowInSec);
|
||||
}
|
||||
|
||||
@Override
|
||||
public long minLocalDeletionTime()
|
||||
{
|
||||
return row.minLocalDeletionTime();
|
||||
}
|
||||
|
||||
@Override
|
||||
public SearchIterator<ColumnMetadata, ColumnData> searchIterator()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -21,6 +21,8 @@ import java.util.Comparator;
|
|||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
|
||||
import net.nicoulaj.compilecommand.annotations.DontInline;
|
||||
|
||||
import accord.utils.Invariants;
|
||||
|
||||
/** 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];
|
||||
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,
|
||||
// because currIdx == sortedSectionSize == size - 1 and nextIdx becomes
|
||||
// (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.
|
||||
while ((nextIdx = (currIdx * 2) - (sortedSectionSize - 1)) + 1 < size)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
public void diff()
|
||||
{
|
||||
|
|
|
|||
Loading…
Reference in New Issue