mirror of https://github.com/apache/cassandra
Fix Accord compaction purger tombstone logic
Accord compaction purgers see random slices of Accord state during compaction (based on randomly selected compaction inputs). For at least the `durability` column in the `commands` table the tombstone being created when truncating was deleting the latest value since we can get enough information to truncate without actuall yhaving the latest `durability` value. To fix we can wait to emit a tombstone until we are erasing the entire command row when truncating or truncating with outcome and meanwhile we can drop the extra columns that are no longer needed instead of using a tombstone. We don't need to emit cell tombstones we can drop them from the purger when processing each row. patch by Ariel Weisberg; reviewed by David Capwell for CASSANDRA-18795
This commit is contained in:
David Capwell
David Capwell
parent
ba02bd0bc7
commit
10e065ffe5
|
|
@ -1 +1 @@
|
|||
Subproject commit 8c7a3c9ef4209d635b186189e17a2d9e728e9871
|
||||
Subproject commit 1d6028ca20553d1c1a6fe2809b204254955da3b3
|
||||
|
|
@ -25,9 +25,9 @@ import java.util.List;
|
|||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.LongPredicate;
|
||||
import java.util.function.Supplier;
|
||||
import javax.annotation.Nonnull;
|
||||
|
||||
import com.google.common.base.Supplier;
|
||||
import com.google.common.collect.ImmutableSet;
|
||||
import com.google.common.collect.Ordering;
|
||||
|
||||
|
|
@ -54,6 +54,8 @@ import org.apache.cassandra.db.filter.ColumnFilter;
|
|||
import org.apache.cassandra.db.partitions.PurgeFunction;
|
||||
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
|
||||
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
|
||||
import org.apache.cassandra.db.rows.BTreeRow;
|
||||
import org.apache.cassandra.db.rows.Cell;
|
||||
import org.apache.cassandra.db.rows.RangeTombstoneBoundMarker;
|
||||
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
|
||||
import org.apache.cassandra.db.rows.Row;
|
||||
|
|
@ -77,6 +79,7 @@ import org.apache.cassandra.schema.TableId;
|
|||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyRows;
|
||||
import org.apache.cassandra.service.accord.AccordService;
|
||||
import org.apache.cassandra.service.accord.IAccordService;
|
||||
|
|
@ -86,14 +89,23 @@ import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
|
|||
import org.apache.cassandra.utils.Pair;
|
||||
import org.apache.cassandra.utils.TimeUUID;
|
||||
|
||||
import static accord.impl.CommandsForKey.NO_LAST_EXECUTED_HLC;
|
||||
import static accord.local.Commands.Cleanup.TRUNCATE_WITH_OUTCOME;
|
||||
import static accord.local.Status.Durability.Universal;
|
||||
import static com.google.common.base.Preconditions.checkState;
|
||||
import static java.util.concurrent.TimeUnit.MICROSECONDS;
|
||||
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
|
||||
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.maybeDropTruncatedCommandColumns;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.truncatedApply;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyColumns.last_executed_micros;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyColumns.last_executed_timestamp;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyColumns.last_write_timestamp;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyColumns.max_timestamp;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyRows.truncateStaticRow;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeDurabilityOrNull;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeRouteOrNull;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeSaveStatusOrNull;
|
||||
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeTimestampOrNull;
|
||||
|
||||
/**
|
||||
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
|
||||
|
|
@ -785,26 +797,43 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
|
|||
if (redundantBefore == null)
|
||||
return row;
|
||||
|
||||
Timestamp executeAt = CommandRows.getExecuteAt(row);
|
||||
Durability durability = CommandRows.getDurability(row);
|
||||
SaveStatus saveStatus = CommandRows.getStatus(row);
|
||||
Route<?> route = CommandRows.getRoute(row);
|
||||
// When commands end up being sliced by compaction we need this to discard tombstones and slices
|
||||
// without enough information to run the rest of the cleanup logic
|
||||
if (durableBefore.min(txnId) == Universal)
|
||||
return null;
|
||||
|
||||
Commands.Cleanup cleanup = Commands.shouldCleanup(txnId, saveStatus.status, durability, executeAt, route, redundantBefore, durableBefore);
|
||||
Cell durabilityCell = row.getCell(CommandsColumns.durability);
|
||||
Durability durability = deserializeDurabilityOrNull(durabilityCell);
|
||||
Cell executeAtCell = row.getCell(CommandsColumns.execute_at);
|
||||
Timestamp executeAt = deserializeTimestampOrNull(executeAtCell);
|
||||
Cell routeCell = row.getCell(CommandsColumns.route);
|
||||
Route<?> route = deserializeRouteOrNull(routeCell);
|
||||
Cell statusCell = row.getCell(CommandsColumns.status);
|
||||
SaveStatus saveStatus = deserializeSaveStatusOrNull(statusCell);
|
||||
|
||||
// With a sliced row we might not have enough columns to determine what to do so output the
|
||||
// the row unmodified and we will try again later once it merges with the rest of the command state
|
||||
// or is dropped by `durableBefore.min(txnId) == Universal`
|
||||
if (executeAt == null || durability == null || saveStatus == null || route == null)
|
||||
return row;
|
||||
|
||||
Commands.Cleanup cleanup = Commands.shouldCleanup(txnId, saveStatus.status, durability, executeAt, route, redundantBefore, durableBefore, false);
|
||||
switch (cleanup)
|
||||
{
|
||||
default: throw new AssertionError(String.format("Unexpected cleanup task: %s", cleanup));
|
||||
case ERASE:
|
||||
return null;
|
||||
// Emit a tombstone so if this is slicing the command and making it not possible to determine if it
|
||||
// can be truncated later it can still be dropped via the tombstone.
|
||||
// Eventually the tombstone can be dropped by `durableBefore.min(txnId) == Universal`
|
||||
// We can still encounter sliced command state just because compaction inputs are random
|
||||
return BTreeRow.emptyDeletedRow(row.clustering(), new Row.Deletion(DeletionTime.build(row.primaryKeyLivenessInfo().timestamp(), nowInSec), false));
|
||||
|
||||
case TRUNCATE_WITH_OUTCOME:
|
||||
if (saveStatus.compareTo(cleanup.appliesIfNot) >= 0)
|
||||
return row;
|
||||
|
||||
case TRUNCATE:
|
||||
if (saveStatus.compareTo(cleanup.appliesIfNot) >= 0)
|
||||
return row;
|
||||
return truncatedApply(cleanup.appliesIfNot, row, nowInSec, cleanup == TRUNCATE_WITH_OUTCOME);
|
||||
return maybeDropTruncatedCommandColumns(row, cleanup == TRUNCATE_WITH_OUTCOME, durabilityCell, executeAtCell, routeCell, statusCell);
|
||||
return truncatedApply(cleanup.appliesIfNot,
|
||||
row, nowInSec, durability, durabilityCell, executeAtCell, routeCell, cleanup == TRUNCATE_WITH_OUTCOME);
|
||||
|
||||
case NO:
|
||||
return row;
|
||||
|
|
@ -853,45 +882,45 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
|
|||
|
||||
TxnId redundantBeforeTxnId = redundantBeforeEntry.redundantBefore;
|
||||
|
||||
boolean updatedColumn = false;
|
||||
Timestamp max_timestamp = CommandsForKeyRows.getMaxTimestamp(row);
|
||||
if (max_timestamp.compareTo(redundantBeforeTxnId) < 0)
|
||||
Cell lastExecuteMicrosCell = row.getCell(last_executed_micros);
|
||||
Long last_execute_micros = null;
|
||||
if (lastExecuteMicrosCell != null && !lastExecuteMicrosCell.accessor().isEmpty(lastExecuteMicrosCell.value()))
|
||||
last_execute_micros = lastExecuteMicrosCell.accessor().getLong(lastExecuteMicrosCell.value(), 0);
|
||||
if (last_execute_micros != null && last_execute_micros < redundantBeforeTxnId.hlc())
|
||||
{
|
||||
max_timestamp = Timestamp.NONE;
|
||||
updatedColumn = true;
|
||||
lastExecuteMicrosCell = null;
|
||||
}
|
||||
|
||||
Timestamp last_execute = CommandsForKeyRows.getLastExecutedTimestamp(row);
|
||||
if (last_execute.compareTo(redundantBeforeTxnId) < 0)
|
||||
Cell lastExecuteCell = row.getCell(last_executed_timestamp);
|
||||
Timestamp last_execute = deserializeTimestampOrNull(lastExecuteCell);
|
||||
if (last_execute != null && last_execute.compareTo(redundantBeforeTxnId) < 0)
|
||||
{
|
||||
last_execute = Timestamp.NONE;
|
||||
updatedColumn = true;
|
||||
lastExecuteCell = null;
|
||||
}
|
||||
|
||||
Timestamp last_write = CommandsForKeyRows.getLastWriteTimestamp(row);
|
||||
if (last_write.compareTo(redundantBeforeTxnId) < 0)
|
||||
Cell lastWriteCell = row.getCell(last_write_timestamp);
|
||||
Timestamp last_write = deserializeTimestampOrNull(lastWriteCell);
|
||||
if (last_write != null && last_write.compareTo(redundantBeforeTxnId) < 0)
|
||||
{
|
||||
last_write = Timestamp.NONE;
|
||||
updatedColumn = true;
|
||||
lastWriteCell = null;
|
||||
}
|
||||
|
||||
long last_execute_micros = CommandsForKeyRows.getLastExecutedMicros(row);
|
||||
if (last_execute_micros < redundantBeforeTxnId.hlc())
|
||||
Cell<?> maxTimestampCell = row.getCell(max_timestamp);
|
||||
Timestamp max_timestamp = deserializeTimestampOrNull(maxTimestampCell);
|
||||
if (max_timestamp != null && max_timestamp.compareTo(redundantBeforeTxnId) < 0)
|
||||
{
|
||||
last_execute_micros = NO_LAST_EXECUTED_HLC;
|
||||
updatedColumn = true;
|
||||
maxTimestampCell = null;
|
||||
}
|
||||
|
||||
if (max_timestamp == Timestamp.NONE &&
|
||||
last_execute == Timestamp.NONE &&
|
||||
last_write == Timestamp.NONE &&
|
||||
last_execute_micros == NO_LAST_EXECUTED_HLC)
|
||||
// No need to emit a tombstone as earlier versions of the row will also be nulled out
|
||||
// when compacted later or loaded into a commands for key
|
||||
if (lastExecuteMicrosCell == null &&
|
||||
lastExecuteCell == null &&
|
||||
lastWriteCell == null &&
|
||||
maxTimestampCell == null)
|
||||
return null;
|
||||
|
||||
if (updatedColumn)
|
||||
return truncateStaticRow(nowInSec, row, last_execute_micros, last_execute, last_write, max_timestamp);
|
||||
|
||||
return row;
|
||||
return truncateStaticRow(nowInSec, row, lastExecuteMicrosCell, lastExecuteCell, lastWriteCell, maxTimestampCell);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -910,7 +939,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
|
|||
|
||||
TxnId redundantBeforeTxnId = redundantBeforeEntry.redundantBefore;
|
||||
Timestamp timestamp = CommandsForKeyRows.getTimestamp(row);
|
||||
if (timestamp.compareTo(redundantBeforeTxnId) < 0)
|
||||
if (timestamp != null && timestamp.compareTo(redundantBeforeTxnId) < 0)
|
||||
return null;
|
||||
|
||||
return row;
|
||||
|
|
@ -966,9 +995,4 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
|
|||
{
|
||||
return cfs.name.equals(AccordKeyspace.COMMANDS_FOR_KEY) && cfs.keyspace.getName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME);
|
||||
}
|
||||
|
||||
private static boolean isAccordCommandsOrAccordCommandsForKey(ColumnFamilyStore cfs)
|
||||
{
|
||||
return isAccordCommands(cfs) || isAccordCommandsForKey(cfs);
|
||||
}
|
||||
}
|
||||
|
|
@ -351,6 +351,11 @@ public class TupleType extends MultiElementType<ByteBuffer>
|
|||
return "component";
|
||||
}
|
||||
|
||||
public static <V> V pack(ValueAccessor<V> accessor, V... components)
|
||||
{
|
||||
return pack(accessor, Arrays.asList(components));
|
||||
}
|
||||
|
||||
public static <V> V pack(ValueAccessor<V> accessor, Collection<V> components)
|
||||
{
|
||||
int totalLength = 0;
|
||||
|
|
|
|||
|
|
@ -62,6 +62,7 @@ import accord.local.Node;
|
|||
import accord.local.RedundantBefore;
|
||||
import accord.local.SaveStatus;
|
||||
import accord.local.Status;
|
||||
import accord.local.Status.Durability;
|
||||
import accord.primitives.Ballot;
|
||||
import accord.primitives.Deps;
|
||||
import accord.primitives.PartialDeps;
|
||||
|
|
@ -119,6 +120,7 @@ import org.apache.cassandra.db.rows.BufferCell;
|
|||
import org.apache.cassandra.db.rows.Cell;
|
||||
import org.apache.cassandra.db.rows.CellPath;
|
||||
import org.apache.cassandra.db.rows.Row;
|
||||
import org.apache.cassandra.db.rows.Row.Deletion;
|
||||
import org.apache.cassandra.db.rows.RowIterator;
|
||||
import org.apache.cassandra.db.transform.FilteredPartitions;
|
||||
import org.apache.cassandra.dht.ByteOrderedPartitioner;
|
||||
|
|
@ -158,6 +160,8 @@ import org.apache.cassandra.utils.Clock;
|
|||
import org.apache.cassandra.utils.btree.BTree;
|
||||
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
|
||||
|
||||
import static accord.utils.Invariants.checkArgument;
|
||||
import static accord.utils.Invariants.checkState;
|
||||
import static java.lang.String.format;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
|
||||
import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal;
|
||||
|
|
@ -286,17 +290,17 @@ public class AccordKeyspace
|
|||
return column;
|
||||
}
|
||||
|
||||
private static class CommandsColumns
|
||||
public static class CommandsColumns
|
||||
{
|
||||
static final ClusteringComparator keyComparator = Commands.partitionKeyAsClusteringComparator();
|
||||
static final CompositeType partitionKeyType = (CompositeType) Commands.partitionKeyType;
|
||||
static final ColumnMetadata txn_id = getColumn(Commands, "txn_id");
|
||||
static final ColumnMetadata store_id = getColumn(Commands, "store_id");
|
||||
static final ColumnMetadata status = getColumn(Commands, "status");
|
||||
static final ColumnMetadata route = getColumn(Commands, "route");
|
||||
static final ColumnMetadata durability = getColumn(Commands, "durability");
|
||||
public static final ColumnMetadata status = getColumn(Commands, "status");
|
||||
public static final ColumnMetadata route = getColumn(Commands, "route");
|
||||
public static final ColumnMetadata durability = getColumn(Commands, "durability");
|
||||
static final ColumnMetadata txn = getColumn(Commands, "txn");
|
||||
static final ColumnMetadata execute_at = getColumn(Commands, "execute_at");
|
||||
public static final ColumnMetadata execute_at = getColumn(Commands, "execute_at");
|
||||
static final ColumnMetadata promised_ballot = getColumn(Commands, "promised_ballot");
|
||||
static final ColumnMetadata accepted_ballot = getColumn(Commands, "accepted_ballot");
|
||||
static final ColumnMetadata dependencies = getColumn(Commands, "dependencies");
|
||||
|
|
@ -305,7 +309,7 @@ public class AccordKeyspace
|
|||
static final ColumnMetadata waiting_on = getColumn(Commands, "waiting_on");
|
||||
static final ColumnMetadata listeners = getColumn(Commands, "listeners");
|
||||
|
||||
static ColumnMetadata[][] TRUNCATE_FIELDS = new ColumnMetadata[][] {
|
||||
public static ColumnMetadata[][] TRUNCATE_FIELDS = new ColumnMetadata[][] {
|
||||
new ColumnMetadata[] { durability, execute_at, route, status },
|
||||
new ColumnMetadata[] { durability, execute_at, result, route, status, writes },
|
||||
};
|
||||
|
|
@ -315,10 +319,9 @@ public class AccordKeyspace
|
|||
for (ColumnMetadata[] cds : TRUNCATE_FIELDS)
|
||||
{
|
||||
for (int i = 1 ; i < cds.length ; ++i)
|
||||
Invariants.checkState(cds[i - 1].compareTo(cds[i]) < 0);
|
||||
checkState(cds[i - 1].compareTo(cds[i]) < 0);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public static class CommandRows extends CommandsColumns
|
||||
|
|
@ -338,29 +341,41 @@ public class AccordKeyspace
|
|||
return deserializeTimestampOrNull(partitionKeyComponents[txn_id.position()], ByteBufferAccessor.instance, TxnId::fromBits);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Timestamp getExecuteAt(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(execute_at);
|
||||
if (cell == null)
|
||||
return null;
|
||||
return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static SaveStatus getStatus(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(status);
|
||||
if (cell == null)
|
||||
return null;
|
||||
int ordinal = cell.accessor().getInt(cell.value(), 0);
|
||||
return CommandSerializers.saveStatus.forOrdinal(ordinal);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Status.Durability getDurability(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(durability);
|
||||
if (cell == null)
|
||||
return null;
|
||||
int ordinal = cell.accessor().getInt(cell.value(), 0);
|
||||
return CommandSerializers.durability.forOrdinal(ordinal);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Route<?> getRoute(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(route);
|
||||
if (cell == null)
|
||||
return null;
|
||||
try
|
||||
{
|
||||
return deserializeOrNull(cell.buffer(), LocalVersionedSerializers.route);
|
||||
|
|
@ -371,22 +386,94 @@ public class AccordKeyspace
|
|||
}
|
||||
}
|
||||
|
||||
public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, boolean withOutcome)
|
||||
private static Object[] truncatedApplyLeaf(long newTimestamp, SaveStatus newSaveStatus, Cell durabilityCell, Cell executeAtCell, @Nullable Cell resultCell, Cell routeCell, @Nullable Cell writesCell, boolean updateTimestamps)
|
||||
{
|
||||
checkArgument(durabilityCell.column() == CommandsColumns.durability);
|
||||
checkArgument(executeAtCell.column() == CommandsColumns.execute_at);
|
||||
checkArgument(resultCell == null || resultCell.column() == CommandsColumns.result);
|
||||
checkArgument(routeCell.column() == CommandsColumns.route);
|
||||
checkArgument(writesCell == null || writesCell.column() == CommandsColumns.writes);
|
||||
boolean includeOutcome = resultCell != null;
|
||||
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(TRUNCATE_FIELDS[includeOutcome ? 1 : 0].length);
|
||||
int colIndex = 0;
|
||||
newLeaf[colIndex++] = updateTimestamps ? durabilityCell.withUpdatedTimestamp(newTimestamp) : durabilityCell;
|
||||
newLeaf[colIndex++] = updateTimestamps ? executeAtCell.withUpdatedTimestamp(newTimestamp) : executeAtCell;
|
||||
if (includeOutcome)
|
||||
newLeaf[colIndex++] = updateTimestamps ? resultCell.withUpdatedTimestamp(newTimestamp) : resultCell;
|
||||
newLeaf[colIndex++] = updateTimestamps ? routeCell.withUpdatedTimestamp(newTimestamp) : routeCell;
|
||||
// Status always needs to use the new timestamp since we are replacing the existing value
|
||||
// All the other columns are being retained unmodified with at most updated timestamps to accomdate deletion
|
||||
newLeaf[colIndex++] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
|
||||
if (includeOutcome)
|
||||
newLeaf[colIndex++] = updateTimestamps ? writesCell.withUpdatedTimestamp(newTimestamp) : writesCell;
|
||||
return newLeaf;
|
||||
}
|
||||
|
||||
public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, Cell durabilityCell, Cell executeAtCell, Cell routeCell, boolean withOutcome)
|
||||
{
|
||||
checkArgument(durabilityCell.column() == CommandsColumns.durability);
|
||||
checkArgument(executeAtCell.column() == CommandsColumns.execute_at);
|
||||
checkArgument(routeCell.column() == CommandsColumns.route);
|
||||
long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
|
||||
long newTimestamp = oldTimestamp + 1;
|
||||
Cell resultCell = withOutcome ? row.getCell(CommandsColumns.result) : null;
|
||||
Cell writesCell = withOutcome ? row.getCell(CommandsColumns.writes) : null;
|
||||
checkState((resultCell != null) == (writesCell != null), "result and writes should always be set together");
|
||||
boolean doDeletion = true;
|
||||
// If durability is not universal we don't want to delete older versions of the row that might have recorded
|
||||
// a higher durability value. maybeDropTruncatedCommandColumns will take care of dropping things even if we don't drop via tombstones.
|
||||
// durability should be the only column that could have an older value that is insufficient for propagating forward
|
||||
if (durability != Durability.Universal)
|
||||
doDeletion = false;
|
||||
// We may not have what we need to generate a deletion and include the outcome in the truncated row
|
||||
// so need to wait until we can have the outcome to issue the deletion otherwise it would be shadowed and lost
|
||||
if (withOutcome && resultCell == null)
|
||||
doDeletion = false;
|
||||
|
||||
ColumnMetadata[] fields = TRUNCATE_FIELDS[withOutcome ? 1 : 0];
|
||||
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(fields.length);
|
||||
for (int i = 0 ; i < fields.length ; ++i)
|
||||
{
|
||||
if (fields[i] == status) newLeaf[i] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
|
||||
else newLeaf[i] = row.getCell(fields[i]).withUpdatedTimestamp(newTimestamp);
|
||||
}
|
||||
Object[] newLeaf = truncatedApplyLeaf(newTimestamp, newSaveStatus, durabilityCell, executeAtCell, resultCell, routeCell, writesCell, doDeletion);
|
||||
|
||||
// Including a deletion allows future compactions to drop data before it gets to the purger
|
||||
// but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns
|
||||
// regardless
|
||||
Row.Deletion deletion = doDeletion ? new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false) : Deletion.LIVE;
|
||||
return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec),
|
||||
new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false),
|
||||
newLeaf);
|
||||
deletion, newLeaf);
|
||||
}
|
||||
|
||||
public static Row maybeDropTruncatedCommandColumns(Row row, boolean withOutcome, Cell durabilityCell, Cell executeAtCell, Cell routeCell, Cell statusCell)
|
||||
{
|
||||
checkArgument(durabilityCell.column() == CommandsColumns.durability);
|
||||
checkArgument(executeAtCell.column() == CommandsColumns.execute_at);
|
||||
checkArgument(routeCell.column() == CommandsColumns.route);
|
||||
checkArgument(statusCell.column() == CommandsColumns.status);
|
||||
int colCount = row.columnCount();
|
||||
// If it's the exact length of the post truncate column count without outcome fields
|
||||
// then it is exactly the columns needed for getting this far and withOutcome doesn't matter since
|
||||
// nothing additional is available to include anyways
|
||||
if (colCount == TRUNCATE_FIELDS[0].length)
|
||||
return row;
|
||||
|
||||
Cell resultCell = row.getCell(CommandsColumns.result);
|
||||
Cell writesCell = row.getCell(CommandsColumns.writes);
|
||||
checkState((resultCell != null) == (writesCell != null), "result and writes should always be set together");
|
||||
boolean includeOutcome = withOutcome && resultCell != null;
|
||||
// This has just the columns needed for truncation with outcome so return it unmodified
|
||||
if (colCount == TRUNCATE_FIELDS[1].length && includeOutcome)
|
||||
return row;
|
||||
|
||||
// Construct a replacement with just the available columns that are still needed
|
||||
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(TRUNCATE_FIELDS[includeOutcome ? 1 : 0].length);
|
||||
int colIndex = 0;
|
||||
newLeaf[colIndex++] = durabilityCell;
|
||||
newLeaf[colIndex++] = executeAtCell;
|
||||
if (includeOutcome)
|
||||
newLeaf[colIndex++] = resultCell;
|
||||
newLeaf[colIndex++] = routeCell;
|
||||
newLeaf[colIndex++] = statusCell;
|
||||
if (includeOutcome)
|
||||
newLeaf[colIndex++] = writesCell;
|
||||
|
||||
return BTreeRow.create(row.clustering(), row.primaryKeyLivenessInfo(), row.deletion(), newLeaf);
|
||||
}
|
||||
|
||||
public static Result getResult(Row row) throws IOException
|
||||
|
|
@ -419,7 +506,7 @@ public class AccordKeyspace
|
|||
.partitioner(new LocalPartitioner(CompositeType.getInstance(Int32Type.instance, BytesType.instance, KEY_TYPE)))
|
||||
.build();
|
||||
|
||||
private static class CommandsForKeyColumns
|
||||
public static class CommandsForKeyColumns
|
||||
{
|
||||
static final ClusteringComparator keyComparator = CommandsForKeys.partitionKeyAsClusteringComparator();
|
||||
static final CompositeType partitionKeyType = (CompositeType) CommandsForKeys.partitionKeyType;
|
||||
|
|
@ -428,10 +515,10 @@ public class AccordKeyspace
|
|||
static final ColumnMetadata key_token = getColumn(CommandsForKeys, "key_token");
|
||||
static final ColumnMetadata key = getColumn(CommandsForKeys, "key");
|
||||
static final ColumnMetadata timestamp = getColumn(CommandsForKeys, "timestamp");
|
||||
static final ColumnMetadata max_timestamp = getColumn(CommandsForKeys, "max_timestamp");
|
||||
static final ColumnMetadata last_executed_timestamp = getColumn(CommandsForKeys, "last_executed_timestamp");
|
||||
static final ColumnMetadata last_executed_micros = getColumn(CommandsForKeys, "last_executed_micros");
|
||||
static final ColumnMetadata last_write_timestamp = getColumn(CommandsForKeys, "last_write_timestamp");
|
||||
public static final ColumnMetadata max_timestamp = getColumn(CommandsForKeys, "max_timestamp");
|
||||
public static final ColumnMetadata last_executed_timestamp = getColumn(CommandsForKeys, "last_executed_timestamp");
|
||||
public static final ColumnMetadata last_executed_micros = getColumn(CommandsForKeys, "last_executed_micros");
|
||||
public static final ColumnMetadata last_write_timestamp = getColumn(CommandsForKeys, "last_write_timestamp");
|
||||
|
||||
static final ColumnMetadata data = getColumn(CommandsForKeys, "data");
|
||||
|
||||
|
|
@ -486,6 +573,7 @@ public class AccordKeyspace
|
|||
return Int32Type.instance.compose(partitionKeyComponents[store_id.position()]);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Timestamp getMaxTimestamp(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(max_timestamp);
|
||||
|
|
@ -494,6 +582,7 @@ public class AccordKeyspace
|
|||
return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Timestamp getLastExecutedTimestamp(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(last_executed_timestamp);
|
||||
|
|
@ -515,11 +604,13 @@ public class AccordKeyspace
|
|||
return deserializeKey(partitionKeyComponents[key.position()]);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Timestamp getTimestamp(Row row)
|
||||
{
|
||||
return deserializeTimestampOrNull(row.clustering().bufferAt(CommandsForKeyColumns.timestamp.position()), Timestamp::fromBits);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static Timestamp getLastWriteTimestamp(Row row)
|
||||
{
|
||||
Cell cell = row.getCell(last_write_timestamp);
|
||||
|
|
@ -528,21 +619,41 @@ public class AccordKeyspace
|
|||
return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
|
||||
}
|
||||
|
||||
public static Row truncateStaticRow(long nowInSec, Row row, long last_execute_micros, Timestamp last_execute, Timestamp last_write, Timestamp max_timestamp)
|
||||
public static Row truncateStaticRow(long nowInSec, Row row, Cell lastExecuteMicrosCell, Cell lastExecuteCell, Cell lastWriteCell, Cell maxTimestampCell)
|
||||
{
|
||||
long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
|
||||
long newTimestamp = oldTimestamp + 1;
|
||||
checkArgument(lastExecuteMicrosCell == null || lastExecuteMicrosCell.column() == CommandsForKeyColumns.last_executed_micros);
|
||||
checkArgument(lastExecuteCell == null || lastExecuteCell.column() == CommandsForKeyColumns.last_executed_timestamp);
|
||||
checkArgument(lastWriteCell == null || lastWriteCell.column() == CommandsForKeyColumns.last_write_timestamp);
|
||||
checkArgument(maxTimestampCell == null || maxTimestampCell.column() == CommandsForKeyColumns.max_timestamp);
|
||||
|
||||
long timestamp = row.primaryKeyLivenessInfo().timestamp();
|
||||
|
||||
int colCount = 0;
|
||||
if (lastExecuteMicrosCell != null)
|
||||
colCount++;
|
||||
if (lastExecuteCell != null)
|
||||
colCount++;
|
||||
if (lastWriteCell != null)
|
||||
colCount++;
|
||||
if (maxTimestampCell != null)
|
||||
colCount++;
|
||||
|
||||
ColumnMetadata[] fields = CommandsForKeyColumns.static_columns_metadata;
|
||||
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(fields.length);
|
||||
newLeaf[0] = BufferCell.live(fields[0], newTimestamp, ByteBufferAccessor.instance.valueOf(last_execute_micros));
|
||||
newLeaf[1] = BufferCell.live(fields[1], newTimestamp, serializeTimestamp(last_execute));
|
||||
newLeaf[2] = BufferCell.live(fields[2], newTimestamp, serializeTimestamp(last_write));
|
||||
newLeaf[3] = BufferCell.live(fields[3], newTimestamp, serializeTimestamp(max_timestamp));
|
||||
checkState(fields.length >= colCount, "CommandsForKeyColumns.static_columns_metadata should include all the columns");
|
||||
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(colCount);
|
||||
int colIndex = 0;
|
||||
|
||||
return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec),
|
||||
new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false),
|
||||
newLeaf);
|
||||
if (lastExecuteMicrosCell != null)
|
||||
newLeaf[colIndex++] = lastExecuteMicrosCell;
|
||||
if (lastExecuteCell != null)
|
||||
newLeaf[colIndex++] = lastExecuteCell;
|
||||
if (lastWriteCell != null)
|
||||
newLeaf[colIndex++] = lastWriteCell;
|
||||
if (maxTimestampCell != null)
|
||||
newLeaf[colIndex++] = maxTimestampCell;
|
||||
|
||||
return BTreeRow.create(row.clustering(), LivenessInfo.create(timestamp, nowInSec),
|
||||
Deletion.LIVE, newLeaf);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -791,6 +902,7 @@ public class AccordKeyspace
|
|||
Row.Builder builder = BTreeRow.unsortedBuilder();
|
||||
builder.newRow(Clustering.EMPTY);
|
||||
int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(timestampMicros);
|
||||
builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(timestampMicros, nowInSeconds));
|
||||
|
||||
addEnumCellIfModified(CommandsColumns.status, Command::saveStatus, builder, timestampMicros, nowInSeconds, original, command);
|
||||
addCellIfModified(CommandsColumns.route, Command::route, LocalVersionedSerializers.route, builder, timestampMicros, nowInSeconds, original, command);
|
||||
|
|
@ -864,6 +976,7 @@ public class AccordKeyspace
|
|||
T create(long msb, long lsb, Node.Id node);
|
||||
}
|
||||
|
||||
@Nullable
|
||||
public static <T extends Timestamp> T deserializeTimestampOrNull(ByteBuffer bytes, TimestampFactory<T> factory)
|
||||
{
|
||||
if (bytes == null || ByteBufferAccessor.instance.isEmpty(bytes))
|
||||
|
|
@ -872,6 +985,18 @@ public class AccordKeyspace
|
|||
return factory.create(split.get(0).getLong(), split.get(1).getLong(), new Node.Id(split.get(2).getInt()));
|
||||
}
|
||||
|
||||
public static <V> Timestamp deserializeTimestampOrNull(Cell<V> cell)
|
||||
{
|
||||
if (cell == null)
|
||||
return null;
|
||||
ValueAccessor<V> accessor = cell.accessor();
|
||||
V value = cell.value();
|
||||
if (accessor.isEmpty(value))
|
||||
return null;
|
||||
List<V> split = TIMESTAMP_TYPE.unpack(value, accessor);
|
||||
return Timestamp.fromBits(accessor.getLong(split.get(0), 0), accessor.getLong(split.get(1), 0), new Node.Id(accessor.getInt(split.get(2), 0)));
|
||||
}
|
||||
|
||||
public static <V, T extends Timestamp> T deserializeTimestampOrNull(V value, ValueAccessor<V> accessor, TimestampFactory<T> factory)
|
||||
{
|
||||
if (value == null || accessor.isEmpty(value))
|
||||
|
|
@ -902,6 +1027,30 @@ public class AccordKeyspace
|
|||
return result;
|
||||
}
|
||||
|
||||
public static Durability deserializeDurabilityOrNull(Cell cell)
|
||||
{
|
||||
return cell == null ? null : CommandSerializers.durability.forOrdinal(cell.accessor().getInt(cell.value(), 0));
|
||||
}
|
||||
|
||||
public static SaveStatus deserializeSaveStatusOrNull(Cell cell)
|
||||
{
|
||||
return cell == null ? null : CommandSerializers.saveStatus.forOrdinal(cell.accessor().getInt(cell.value(), 0));
|
||||
}
|
||||
|
||||
public static Route<?> deserializeRouteOrNull(Cell cell)
|
||||
{
|
||||
if (cell == null)
|
||||
return null;
|
||||
try
|
||||
{
|
||||
return deserializeOrNull(cell.buffer(), LocalVersionedSerializers.route);
|
||||
}
|
||||
catch (IOException e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private static <T> T deserializeWithVersionOr(UntypedResultSet.Row row, String dataColumn, LocalVersionedSerializer<T> serializer, Supplier<T> defaultSupplier) throws IOException
|
||||
{
|
||||
if (!row.has(dataColumn))
|
||||
|
|
@ -1149,7 +1298,7 @@ public class AccordKeyspace
|
|||
try
|
||||
{
|
||||
UntypedResultSet.Row row = rows.one();
|
||||
Invariants.checkState(deserializeTxnId(row).equals(txnId));
|
||||
checkState(deserializeTxnId(row).equals(txnId));
|
||||
SaveStatus status = deserializeStatus(row);
|
||||
CommonAttributes.Mutable attributes = new CommonAttributes.Mutable(txnId);
|
||||
// TODO: something less brittle than ordinal, more efficient than values()
|
||||
|
|
@ -1250,12 +1399,13 @@ public class AccordKeyspace
|
|||
SeriesKind kind,
|
||||
PartitionUpdate.Builder partitionBuilder,
|
||||
Row.Builder rowBuilder,
|
||||
long timestampMicros,
|
||||
LivenessInfo livenessInfo,
|
||||
int nowInSeconds)
|
||||
{
|
||||
if (prev == value)
|
||||
return;
|
||||
|
||||
long timestampMicros = livenessInfo.timestamp();
|
||||
Set<Timestamp> deletions = Sets.difference(prev.keySet(), value.keySet());
|
||||
|
||||
Row.Deletion deletion = !deletions.isEmpty() ?
|
||||
|
|
@ -1267,6 +1417,7 @@ public class AccordKeyspace
|
|||
return;
|
||||
rowBuilder.newRow(Clustering.make(ordinalBytes, serializeTimestamp(timestamp)));
|
||||
rowBuilder.addCell(live(CommandsForKeyColumns.data, timestampMicros, bytes));
|
||||
rowBuilder.addPrimaryKeyLivenessInfo(livenessInfo);
|
||||
partitionBuilder.add(rowBuilder.build());
|
||||
});
|
||||
deletions.forEach(timestamp -> {
|
||||
|
|
@ -1281,10 +1432,10 @@ public class AccordKeyspace
|
|||
SeriesKind kind,
|
||||
PartitionUpdate.Builder partitionBuilder,
|
||||
Row.Builder rowBuilder,
|
||||
long timestampMicros,
|
||||
LivenessInfo livenessInfo,
|
||||
int nowInSeconds)
|
||||
{
|
||||
addSeriesMutations(kind.getValues(original), kind.getValues(cfk), kind, partitionBuilder, rowBuilder, timestampMicros, nowInSeconds);
|
||||
addSeriesMutations(kind.getValues(original), kind.getValues(cfk), kind, partitionBuilder, rowBuilder, livenessInfo, nowInSeconds);
|
||||
}
|
||||
|
||||
private static DecoratedKey makeKey(int storeId, PartitionKey key)
|
||||
|
|
@ -1315,6 +1466,7 @@ public class AccordKeyspace
|
|||
ValueAccessor<ByteBuffer> accessor = ByteBufferAccessor.instance;
|
||||
|
||||
int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(timestampMicros);
|
||||
LivenessInfo livenessInfo = LivenessInfo.create(timestampMicros, nowInSeconds);
|
||||
|
||||
boolean hasStaticChanges = CommandsForKeyColumns.hasStaticChanges(original, cfk);
|
||||
int expectedRows = (hasStaticChanges ? 1 : 0)
|
||||
|
|
@ -1335,13 +1487,14 @@ public class AccordKeyspace
|
|||
addCellIfModified(CommandsForKeyColumns.last_executed_timestamp, CommandsForKey::lastExecutedTimestamp, AccordKeyspace::serializeTimestamp, rowBuilder, timestampMicros, nowInSeconds, original, cfk);
|
||||
addCellIfModified(CommandsForKeyColumns.last_executed_micros, CommandsForKey::rawLastExecutedHlc, accessor::valueOf, rowBuilder, timestampMicros, nowInSeconds, original, cfk);
|
||||
addCellIfModified(CommandsForKeyColumns.last_write_timestamp, CommandsForKey::lastWriteTimestamp, AccordKeyspace::serializeTimestamp, rowBuilder, timestampMicros, nowInSeconds, original, cfk);
|
||||
rowBuilder.addPrimaryKeyLivenessInfo(livenessInfo);
|
||||
Row row = rowBuilder.build();
|
||||
if (!row.isEmpty())
|
||||
partitionBuilder.add(row);
|
||||
}
|
||||
|
||||
addSeriesMutations(original, cfk, SeriesKind.BY_ID, partitionBuilder, rowBuilder, timestampMicros, nowInSeconds);
|
||||
addSeriesMutations(original, cfk, SeriesKind.BY_EXECUTE_AT, partitionBuilder, rowBuilder, timestampMicros, nowInSeconds);
|
||||
addSeriesMutations(original, cfk, SeriesKind.BY_ID, partitionBuilder, rowBuilder, livenessInfo, nowInSeconds);
|
||||
addSeriesMutations(original, cfk, SeriesKind.BY_EXECUTE_AT, partitionBuilder, rowBuilder, livenessInfo, nowInSeconds);
|
||||
|
||||
PartitionUpdate update = partitionBuilder.build();
|
||||
if (update.isEmpty())
|
||||
|
|
@ -1431,7 +1584,7 @@ public class AccordKeyspace
|
|||
seriesMaps.get(SeriesKind.values()[ordinal]).put(timestamp, data);
|
||||
}
|
||||
}
|
||||
Invariants.checkState(!partitions.hasNext());
|
||||
checkState(!partitions.hasNext());
|
||||
|
||||
return CommandsForKey.SerializerSupport.create(key, max, lastExecutedTimestamp, lastExecutedMicros, lastWriteTimestamp,
|
||||
CommandsForKeySerializer.loader,
|
||||
|
|
@ -1657,7 +1810,7 @@ public class AccordKeyspace
|
|||
String cql = format("SELECT * FROM %s.%s WHERE epoch=?", ACCORD_KEYSPACE_NAME, TOPOLOGIES);
|
||||
|
||||
UntypedResultSet result = executeInternal(cql, epoch);
|
||||
Invariants.checkState(!result.isEmpty(), "Nothing found for epoch %d", epoch);
|
||||
checkState(!result.isEmpty(), "Nothing found for epoch %d", epoch);
|
||||
UntypedResultSet.Row row = result.one();
|
||||
Topology topology = row.has("topology")
|
||||
? deserialize(row.getBytes("topology"), LocalVersionedSerializers.topology)
|
||||
|
|
|
|||
|
|
@ -55,12 +55,6 @@ import com.google.common.base.Function;
|
|||
import com.google.common.base.Preconditions;
|
||||
import com.google.common.collect.Iterables;
|
||||
import com.google.common.collect.Iterators;
|
||||
|
||||
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
|
||||
import org.apache.cassandra.gms.ApplicationState;
|
||||
import org.apache.cassandra.gms.Gossiper;
|
||||
import org.apache.cassandra.gms.VersionedValue;
|
||||
import org.apache.cassandra.io.util.File;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.junit.Assume;
|
||||
import org.slf4j.Logger;
|
||||
|
|
@ -91,12 +85,6 @@ import org.apache.cassandra.db.compaction.CompactionManager;
|
|||
import org.apache.cassandra.db.compaction.CompactionTasks;
|
||||
import org.apache.cassandra.db.compaction.OperationType;
|
||||
import org.apache.cassandra.db.lifecycle.LifecycleTransaction;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
import org.apache.cassandra.locator.ReplicaCollection;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.schema.ColumnMetadata;
|
||||
import org.apache.cassandra.schema.TableId;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.db.marshal.AbstractType;
|
||||
import org.apache.cassandra.db.marshal.AsciiType;
|
||||
import org.apache.cassandra.db.marshal.Int32Type;
|
||||
|
|
@ -123,6 +111,10 @@ import org.apache.cassandra.dht.RandomPartitioner.BigIntegerToken;
|
|||
import org.apache.cassandra.dht.Range;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.index.internal.CassandraIndex;
|
||||
import org.apache.cassandra.distributed.test.log.ClusterMetadataTestHelper;
|
||||
import org.apache.cassandra.gms.ApplicationState;
|
||||
import org.apache.cassandra.gms.Gossiper;
|
||||
import org.apache.cassandra.gms.VersionedValue;
|
||||
import org.apache.cassandra.io.sstable.Descriptor;
|
||||
import org.apache.cassandra.io.sstable.SSTableId;
|
||||
import org.apache.cassandra.io.sstable.SSTableLoader;
|
||||
|
|
@ -131,8 +123,15 @@ import org.apache.cassandra.io.sstable.UUIDBasedSSTableId;
|
|||
import org.apache.cassandra.io.sstable.format.SSTableReader;
|
||||
import org.apache.cassandra.io.sstable.format.SSTableReaderWithFilter;
|
||||
import org.apache.cassandra.io.util.DataInputPlus;
|
||||
import org.apache.cassandra.io.util.File;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
import org.apache.cassandra.locator.Replica;
|
||||
import org.apache.cassandra.locator.ReplicaCollection;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.schema.ColumnMetadata;
|
||||
import org.apache.cassandra.schema.Schema;
|
||||
import org.apache.cassandra.schema.TableId;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.schema.TableMetadataRef;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.service.pager.PagingState;
|
||||
|
|
@ -735,6 +734,20 @@ public class Util
|
|||
.untilAsserted(() -> assertThat(message, actualSupplier.get(), matcher));
|
||||
}
|
||||
|
||||
public static void spinAssertEquals(Object expected, int timeoutInSeconds, Callable<Object> call)
|
||||
{
|
||||
spinAssertEquals(null, expected, timeoutInSeconds, TimeUnit.SECONDS, call);
|
||||
}
|
||||
|
||||
public static <T> void spinAssertEquals(String message, T expected, long timeout, TimeUnit timeUnit, Callable<? extends T> call)
|
||||
{
|
||||
Awaitility.await()
|
||||
.pollInterval(Duration.ofMillis(100))
|
||||
.pollDelay(0, TimeUnit.MILLISECONDS)
|
||||
.atMost(timeout, timeUnit)
|
||||
.untilAsserted(() -> assertThat(message, call.call(), equalTo(expected)));
|
||||
}
|
||||
|
||||
public static void joinThread(Thread thread) throws InterruptedException
|
||||
{
|
||||
thread.join(10000);
|
||||
|
|
|
|||
|
|
@ -24,24 +24,27 @@ import java.util.ArrayList;
|
|||
import java.util.Collections;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Random;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import com.google.common.collect.Iterables;
|
||||
import com.google.common.collect.Iterators;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import accord.api.Result;
|
||||
import accord.api.RoutingKey;
|
||||
import accord.local.CheckedCommands;
|
||||
import accord.local.CommandStore;
|
||||
import accord.local.DurableBefore;
|
||||
import accord.local.RedundantBefore;
|
||||
import accord.local.SaveStatus;
|
||||
import accord.local.Status;
|
||||
import accord.local.Status.Durability;
|
||||
import accord.primitives.Ballot;
|
||||
import accord.primitives.Deps;
|
||||
|
|
@ -51,7 +54,6 @@ import accord.primitives.PartialRoute;
|
|||
import accord.primitives.PartialTxn;
|
||||
import accord.primitives.Ranges;
|
||||
import accord.primitives.Seekable;
|
||||
import accord.primitives.Timestamp;
|
||||
import accord.primitives.Txn;
|
||||
import accord.primitives.Txn.Kind;
|
||||
import accord.primitives.TxnId;
|
||||
|
|
@ -64,18 +66,22 @@ import org.apache.cassandra.db.Clustering;
|
|||
import org.apache.cassandra.db.ColumnFamilyStore;
|
||||
import org.apache.cassandra.db.ColumnFamilyStore.FlushReason;
|
||||
import org.apache.cassandra.db.Keyspace;
|
||||
import org.apache.cassandra.db.compaction.CompactionIteratorTest.Scanner;
|
||||
import org.apache.cassandra.db.partitions.ImmutableBTreePartition;
|
||||
import org.apache.cassandra.db.partitions.Partition;
|
||||
import org.apache.cassandra.db.rows.Row;
|
||||
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
|
||||
import org.apache.cassandra.io.sstable.ISSTableScanner;
|
||||
import org.apache.cassandra.io.sstable.format.SSTableReader;
|
||||
import org.apache.cassandra.schema.ColumnMetadata;
|
||||
import org.apache.cassandra.schema.KeyspaceParams;
|
||||
import org.apache.cassandra.schema.SchemaConstants;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.cassandra.service.accord.AccordCommandStore;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
|
||||
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyRows;
|
||||
import org.apache.cassandra.service.accord.AccordTestUtils;
|
||||
import org.apache.cassandra.service.accord.IAccordService;
|
||||
|
|
@ -86,6 +92,7 @@ import org.apache.cassandra.utils.Pair;
|
|||
import static accord.impl.CommandsForKey.NO_LAST_EXECUTED_HLC;
|
||||
import static accord.local.PreLoadContext.contextFor;
|
||||
import static accord.utils.async.AsyncChains.getUninterruptibly;
|
||||
import static org.apache.cassandra.Util.spinAssertEquals;
|
||||
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
|
||||
import static org.apache.cassandra.db.compaction.CompactionAccordIteratorsTest.DurableBeforeType.MAJORITY;
|
||||
import static org.apache.cassandra.db.compaction.CompactionAccordIteratorsTest.DurableBeforeType.NOT_DURABLE;
|
||||
|
|
@ -95,6 +102,7 @@ import static org.apache.cassandra.service.accord.AccordKeyspace.COMMANDS;
|
|||
import static org.apache.cassandra.service.accord.AccordKeyspace.COMMANDS_FOR_KEY;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.times;
|
||||
|
|
@ -104,6 +112,7 @@ import static org.mockito.Mockito.when;
|
|||
public class CompactionAccordIteratorsTest
|
||||
{
|
||||
private static final Logger logger = LoggerFactory.getLogger(CompactionAccordIteratorsTest.class);
|
||||
|
||||
private static final long CLOCK_START = 44;
|
||||
private static final long HLC_START = 41;
|
||||
private static final int NODE = 1;
|
||||
|
|
@ -117,8 +126,17 @@ public class CompactionAccordIteratorsTest
|
|||
private static final TxnId[] TXN_IDS = new TxnId[] {TXN_ID, SECOND_TXN_ID};
|
||||
private static final TxnId GT_SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, SECOND_TXN_ID.hlc() + 1, NODE);
|
||||
|
||||
static ColumnFamilyStore commands;
|
||||
static ColumnFamilyStore commandsForKey;
|
||||
static TableMetadata table;
|
||||
static FullRoute route;
|
||||
Random random;
|
||||
|
||||
/*
|
||||
* Whether to compact all tables at once in a single merge or forcing two random tables
|
||||
* to merge at a time
|
||||
*/
|
||||
private boolean singleCompaction;
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() throws Throwable
|
||||
|
|
@ -128,15 +146,40 @@ public class CompactionAccordIteratorsTest
|
|||
SchemaLoader.createKeyspace("ks", KeyspaceParams.simple(1),
|
||||
parse("CREATE TABLE tbl (k int, c int, v int, primary key (k, c))", "ks"));
|
||||
StorageService.instance.initServer();
|
||||
commands = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, COMMANDS);
|
||||
commands.disableAutoCompaction();
|
||||
commandsForKey = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, COMMANDS_FOR_KEY);
|
||||
commandsForKey.disableAutoCompaction();
|
||||
table = ColumnFamilyStore.getIfExists("ks", "tbl").metadata();
|
||||
route = AccordTestUtils.keys(table, 42).toRoute(AccordTestUtils.key(table, 42).toUnseekable());
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setUp()
|
||||
{
|
||||
// This attempt at determinism doesn't work because the order of the SSTableScanners is not determinisitc
|
||||
long seed = System.nanoTime();
|
||||
logger.info("Seed " + seed + "L");
|
||||
random = new Random(seed);
|
||||
}
|
||||
|
||||
// This isn't attempting to be an exhaustive test of Commands.shouldCleanup just that the return values
|
||||
// are handled correctly and that the interaction between the CompactionIterator and shoudCleanup seems reasonable
|
||||
@Test
|
||||
public void testAccordCommandsPurger() throws Throwable
|
||||
public void testAccordCommandsPurgerSingleCompaction() throws Throwable
|
||||
{
|
||||
testAccordCommandsPurger(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAccordCommandsPurgerMultipleCompactions() throws Throwable
|
||||
{
|
||||
testAccordCommandsPurger(false);
|
||||
}
|
||||
|
||||
private void testAccordCommandsPurger(boolean singleCompaction) throws Throwable
|
||||
{
|
||||
this.singleCompaction = singleCompaction;
|
||||
// Null redudnant before should make no change since we have no information on this CommandStore
|
||||
testAccordCommandsPurger(null, DurableBefore.EMPTY, expectAccordCommandsNoChange());
|
||||
// Universally durable (and global to boot) should be erased since literally everyone knows about it
|
||||
|
|
@ -156,7 +199,7 @@ public class CompactionAccordIteratorsTest
|
|||
testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(DurableBeforeType.EMPTY), expectAccordCommandsNoChange());
|
||||
}
|
||||
|
||||
private static void testAccordCommandsPurger(RedundantBefore redundantBefore, DurableBefore durableBefore, Consumer<List<Partition>> expectedResult) throws Throwable
|
||||
private void testAccordCommandsPurger(RedundantBefore redundantBefore, DurableBefore durableBefore, Consumer<List<Partition>> expectedResult) throws Throwable
|
||||
{
|
||||
testWithCommandStore((commandStore) -> {
|
||||
IAccordService mockAccordService = mockAccordService(commandStore, redundantBefore, durableBefore);
|
||||
|
|
@ -167,8 +210,20 @@ public class CompactionAccordIteratorsTest
|
|||
}
|
||||
|
||||
@Test
|
||||
public void testAccordCommandsForKeyPurger() throws Throwable
|
||||
public void testAccordCommandsForKeyPurgerSingleCompaction() throws Throwable
|
||||
{
|
||||
testAccordCommandsForKeyPurger(true);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAccordCommandsForKeyPurgerMultipleCompactions() throws Throwable
|
||||
{
|
||||
testAccordCommandsForKeyPurger(false);
|
||||
}
|
||||
|
||||
private void testAccordCommandsForKeyPurger(boolean singleCompaction) throws Throwable
|
||||
{
|
||||
this.singleCompaction = singleCompaction;
|
||||
testAccordCommandsForKeyPurger(null, expectedAccordCommandsForKeyNoChange());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordCommandsForKeyNoChange());
|
||||
testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyNoChange());
|
||||
|
|
@ -189,7 +244,7 @@ public class CompactionAccordIteratorsTest
|
|||
assertEquals(TXN_ID.hlc(), CommandsForKeyRows.getLastExecutedMicros(staticRow));
|
||||
assertEquals(4, Iterators.size(partition.unfilteredIterator()));
|
||||
UnfilteredRowIterator rows = partition.unfilteredIterator();
|
||||
// One row per series
|
||||
// One row per txn per series
|
||||
for (int i = 0; i < 2; i++)
|
||||
for (TxnId txnId : TXN_IDS)
|
||||
assertEquals(txnId, CommandsForKeyRows.getTimestamp((Row)rows.next()));
|
||||
|
|
@ -202,10 +257,11 @@ public class CompactionAccordIteratorsTest
|
|||
assertEquals(1, partitions.size());
|
||||
Partition partition = partitions.get(0);
|
||||
Row staticRow = partition.getRow(Clustering.STATIC_CLUSTERING);
|
||||
assertEquals(4, Iterables.size(staticRow));
|
||||
// Only expect one column to remain because the second transaction is a read
|
||||
assertEquals(1, Iterables.size(staticRow));
|
||||
assertEquals(SECOND_TXN_ID, CommandsForKeyRows.getMaxTimestamp(staticRow));
|
||||
assertEquals(Timestamp.NONE, CommandsForKeyRows.getLastExecutedTimestamp(staticRow));
|
||||
assertEquals(Timestamp.NONE, CommandsForKeyRows.getLastWriteTimestamp(staticRow));
|
||||
assertNull(CommandsForKeyRows.getLastExecutedTimestamp(staticRow));
|
||||
assertNull(CommandsForKeyRows.getLastWriteTimestamp(staticRow));
|
||||
assertEquals(NO_LAST_EXECUTED_HLC, CommandsForKeyRows.getLastExecutedMicros(staticRow));
|
||||
assertEquals(2, Iterators.size(partition.unfilteredIterator()));
|
||||
UnfilteredRowIterator rows = partition.unfilteredIterator();
|
||||
|
|
@ -219,7 +275,7 @@ public class CompactionAccordIteratorsTest
|
|||
return partitions -> assertEquals(0, partitions.size());
|
||||
}
|
||||
|
||||
private static void testAccordCommandsForKeyPurger(RedundantBefore redundantBefore, Consumer<List<Partition>> expectedResult) throws Throwable
|
||||
private void testAccordCommandsForKeyPurger(RedundantBefore redundantBefore, Consumer<List<Partition>> expectedResult) throws Throwable
|
||||
{
|
||||
testWithCommandStore((commandStore) -> {
|
||||
IAccordService mockAccordService = mockAccordService(commandStore, redundantBefore, DurableBefore.EMPTY);
|
||||
|
|
@ -244,7 +300,9 @@ public class CompactionAccordIteratorsTest
|
|||
assertEquals(1, Iterators.size(partition.unfilteredIterator()));
|
||||
ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
|
||||
Row row = (Row) partition.unfilteredIterator().next();
|
||||
assertEquals(6, row.columnCount());
|
||||
assertEquals(CommandsColumns.TRUNCATE_FIELDS[1].length, row.columnCount());
|
||||
for (ColumnMetadata cm : CommandsColumns.TRUNCATE_FIELDS[1])
|
||||
assertNotNull(row.getColumnData(cm));
|
||||
assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
|
||||
assertEquals(1, ((TxnData)CommandRows.getResult(row)).entrySet().size());
|
||||
assertNotNull(CommandRows.getWrites(row));
|
||||
|
|
@ -268,7 +326,9 @@ public class CompactionAccordIteratorsTest
|
|||
assertEquals(1, Iterators.size(partition.unfilteredIterator()));
|
||||
ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
|
||||
Row row = (Row)partition.unfilteredIterator().next();
|
||||
assertEquals(4, row.columnCount());
|
||||
assertEquals(CommandsColumns.TRUNCATE_FIELDS[0].length, row.columnCount());
|
||||
for (ColumnMetadata cm : CommandsColumns.TRUNCATE_FIELDS[0])
|
||||
assertNotNull(row.getColumnData(cm));
|
||||
assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
|
||||
assertEquals(Durability.Local, CommandRows.getDurability(row));
|
||||
assertEquals(TXN_ID, CommandRows.getExecuteAt(row));
|
||||
|
|
@ -285,6 +345,9 @@ public class CompactionAccordIteratorsTest
|
|||
assertEquals(1, Iterators.size(partition.unfilteredIterator()));
|
||||
ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
|
||||
Row row = (Row)partition.unfilteredIterator().next();
|
||||
assertEquals(commands.metadata().regularColumns().size(), row.columnCount());
|
||||
for (ColumnMetadata cm : commands.metadata().regularColumns())
|
||||
assertNotNull(row.getColumnData(cm));
|
||||
assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
|
||||
assertEquals(SaveStatus.Applied, AccordKeyspace.CommandRows.getStatus(row));
|
||||
};
|
||||
|
|
@ -338,7 +401,21 @@ public class CompactionAccordIteratorsTest
|
|||
void test(AccordCommandStore commandStore) throws Throwable;
|
||||
}
|
||||
|
||||
private static void testWithCommandStore(TestWithCommandStore test, boolean additionalCommand) throws Throwable
|
||||
|
||||
private static void flush(AccordCommandStore commandStore)
|
||||
{
|
||||
commandStore.executeBlocking(() -> {
|
||||
// clear cache and wait for post-eviction writes to complete
|
||||
long cacheSize = commandStore.getCacheSize();
|
||||
commandStore.setCacheSize(0);
|
||||
commandStore.setCacheSize(cacheSize);
|
||||
commandStore.cache().awaitSaveResults();
|
||||
});
|
||||
commands.forceBlockingFlush(FlushReason.UNIT_TESTS);
|
||||
commandsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
|
||||
}
|
||||
|
||||
private void testWithCommandStore(TestWithCommandStore test, boolean additionalCommand) throws Throwable
|
||||
{
|
||||
Keyspace.open(ACCORD_KEYSPACE_NAME).getColumnFamilyStores().forEach(ColumnFamilyStore::truncateBlocking);
|
||||
clock.set(CLOCK_START);
|
||||
|
|
@ -350,57 +427,93 @@ public class CompactionAccordIteratorsTest
|
|||
Seekable key = txn.keys().get(0);
|
||||
PartialDeps partialDeps = Deps.NONE.slice(AccordTestUtils.fullRange(txn));
|
||||
PartialTxn partialTxn = txn.slice(commandStore.unsafeRangesForEpoch().currentRanges(), true);
|
||||
RoutingKey homeKey = key.someIntersectingRoutingKey(commandStore.unsafeRangesForEpoch().currentRanges());
|
||||
PartialRoute partialRoute = route.slice(commandStore.unsafeRangesForEpoch().currentRanges());
|
||||
getUninterruptibly(commandStore.submit(contextFor(txnId, txn.keys()), safe -> {
|
||||
long originalCacheSize = getUninterruptibly(commandStore.submit(contextFor(txnId, txn.keys()), safe -> {
|
||||
// clear cache
|
||||
long cacheSize = commandStore.getCacheSize();
|
||||
commandStore.setCacheSize(0);
|
||||
CheckedCommands.preaccept(safe, txnId, partialTxn, route, null);
|
||||
return cacheSize;
|
||||
}).beginAsResult());
|
||||
flush(commandStore);
|
||||
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {
|
||||
CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, partialTxn.keys(), null, txnId, partialDeps);
|
||||
}).beginAsResult());
|
||||
flush(commandStore);
|
||||
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {
|
||||
CheckedCommands.commit(safe, txnId, route, null, partialTxn, txnId, partialDeps);
|
||||
}).beginAsResult());
|
||||
flush(commandStore);
|
||||
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {
|
||||
Pair<Writes, Result> result = AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
|
||||
CheckedCommands.apply(safe, txnId, route, null, txnId, partialDeps, partialTxn, result.left, result.right);
|
||||
return safe.get(txnId, homeKey).current();
|
||||
}).beginAsResult());
|
||||
flush(commandStore);
|
||||
// The apply chain is asychronous so it is easiest to just spin until it is applied
|
||||
// in order to have the updated state in the system table
|
||||
spinAssertEquals(true, 5, () ->
|
||||
getUninterruptibly(commandStore.submit(contextFor(txnId, txn.keys()), safe -> safe.get(txnId, route.homeKey()).current().hasBeen(Status.Applied)
|
||||
).beginAsResult()));
|
||||
flush(commandStore);
|
||||
getUninterruptibly(commandStore.execute(contextFor(txnId, txn.keys()), safe -> {
|
||||
commandStore.setCacheSize(originalCacheSize);
|
||||
}).beginAsResult());
|
||||
}
|
||||
|
||||
commandStore.executeBlocking(() -> {
|
||||
// clear cache and wait for post-eviction writes to complete
|
||||
long cacheSize = commandStore.getCacheSize();
|
||||
commandStore.setCacheSize(0);
|
||||
commandStore.setCacheSize(cacheSize);
|
||||
commandStore.cache().awaitSaveResults();
|
||||
});
|
||||
|
||||
UntypedResultSet commandsTable = QueryProcessor.executeInternal("SELECT * FROM " + ACCORD_KEYSPACE_NAME + "." + COMMANDS + ";");
|
||||
logger.info(commandsTable.toStringUnsafe());
|
||||
assertEquals(txnIds.length, commandsTable.size());
|
||||
Iterator<UntypedResultSet.Row> commandsTableIterator = commandsTable.iterator();
|
||||
for (TxnId txnId : txnIds)
|
||||
assertEquals(txnId, AccordKeyspace.deserializeTimestampOrNull(commandsTableIterator.next().getBytes("txn_id"), TxnId::fromBits));
|
||||
UntypedResultSet commandsForKeyTable = QueryProcessor.executeInternal("SELECT * FROM " + ACCORD_KEYSPACE_NAME + "." + COMMANDS_FOR_KEY + ";");
|
||||
logger.info(commandsForKeyTable.toStringUnsafe());
|
||||
assertEquals(txnIds.length * 2, commandsForKeyTable.size());
|
||||
Iterator<UntypedResultSet.Row> commandsForKeyTableIterator = commandsTable.iterator();
|
||||
for (TxnId txnId : txnIds)
|
||||
assertEquals(txnId, AccordKeyspace.deserializeTimestampOrNull(commandsForKeyTableIterator.next().getBytes("txn_id"), TxnId::fromBits));
|
||||
System.out.println(commandsForKeyTable);
|
||||
test.test(commandStore);
|
||||
}
|
||||
|
||||
private static List<Partition> compactCFS(IAccordService mockAccordService, ColumnFamilyStore cfs)
|
||||
private List<Partition> compactCFS(IAccordService mockAccordService, ColumnFamilyStore cfs)
|
||||
{
|
||||
cfs.forceBlockingFlush(FlushReason.UNIT_TESTS);
|
||||
List<ISSTableScanner> scanners = cfs.getLiveSSTables().stream().map(SSTableReader::getScanner).collect(Collectors.toList());
|
||||
List<Partition> result = new ArrayList<>();
|
||||
try (CompactionController controller = new CompactionController(ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, cfs.name), Collections.emptySet(), 0);
|
||||
CompactionIterator compactionIterator = new CompactionIterator(OperationType.COMPACTION, scanners, controller, FBUtilities.nowInSeconds(), null, ActiveCompactionsTracker.NOOP, null, () -> mockAccordService))
|
||||
int numScanners = scanners.size();
|
||||
List<Partition> result = null;
|
||||
do
|
||||
{
|
||||
while (compactionIterator.hasNext())
|
||||
List<Partition> outputPartitions = new ArrayList<>();
|
||||
List<ISSTableScanner> nextInputScanners = new ArrayList<>();
|
||||
if (singleCompaction)
|
||||
{
|
||||
try (UnfilteredRowIterator partition = compactionIterator.next())
|
||||
nextInputScanners = ImmutableList.copyOf(scanners);
|
||||
scanners.clear();
|
||||
}
|
||||
else
|
||||
{
|
||||
// Process the rows only two sstables at a time to force compacting random slices of command state
|
||||
nextInputScanners.add(scanners.remove(random.nextInt(scanners.size())));
|
||||
nextInputScanners.add(scanners.remove(random.nextInt(scanners.size())));
|
||||
}
|
||||
try (CompactionController controller = new CompactionController(ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, cfs.name), Collections.emptySet(), 0);
|
||||
CompactionIterator compactionIterator = new CompactionIterator(OperationType.COMPACTION, nextInputScanners, controller, FBUtilities.nowInSeconds(), null, ActiveCompactionsTracker.NOOP, null, () -> mockAccordService))
|
||||
{
|
||||
while (compactionIterator.hasNext())
|
||||
{
|
||||
result.add(ImmutableBTreePartition.create(partition));
|
||||
try (UnfilteredRowIterator partition = compactionIterator.next())
|
||||
{
|
||||
outputPartitions.add(ImmutableBTreePartition.create(partition));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
verify(mockAccordService, times(1)).getRedundantBeforesAndDurableBefore();
|
||||
|
||||
if (scanners.isEmpty())
|
||||
result = outputPartitions;
|
||||
else
|
||||
scanners.add(random.nextInt(scanners.size()), new Scanner(cfs.metadata(), outputPartitions.stream().map(Partition::unfilteredIterator).collect(Collectors.toList())));
|
||||
} while (!scanners.isEmpty());
|
||||
|
||||
verify(mockAccordService, times(singleCompaction ? 1 : numScanners - 1)).getRedundantBeforesAndDurableBefore();
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,18 +17,25 @@
|
|||
*/
|
||||
package org.apache.cassandra.db.compaction;
|
||||
|
||||
import static org.apache.cassandra.config.CassandraRelevantProperties.DIAGNOSTIC_SNAPSHOT_INTERVAL_NANOS;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.assertCommandIssued;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.makeRow;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.partition;
|
||||
import static org.junit.Assert.*;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.NavigableMap;
|
||||
import java.util.Random;
|
||||
import java.util.Set;
|
||||
import java.util.TreeMap;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import com.google.common.collect.*;
|
||||
|
||||
import com.google.common.collect.ImmutableList;
|
||||
import com.google.common.collect.ImmutableSet;
|
||||
import com.google.common.collect.Iterables;
|
||||
import com.google.common.collect.Iterators;
|
||||
import com.google.common.collect.Lists;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
|
||||
|
|
@ -44,17 +51,30 @@ import org.apache.cassandra.db.Keyspace;
|
|||
import org.apache.cassandra.db.marshal.Int32Type;
|
||||
import org.apache.cassandra.db.marshal.UTF8Type;
|
||||
import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
|
||||
import org.apache.cassandra.db.rows.*;
|
||||
import org.apache.cassandra.db.rows.RangeTombstoneBoundaryMarker;
|
||||
import org.apache.cassandra.db.rows.Unfiltered;
|
||||
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
|
||||
import org.apache.cassandra.db.rows.UnfilteredRowsGenerator;
|
||||
import org.apache.cassandra.io.sstable.ISSTableScanner;
|
||||
import org.apache.cassandra.io.sstable.format.SSTableReader;
|
||||
import org.apache.cassandra.locator.InetAddressAndPort;
|
||||
import org.apache.cassandra.net.Message;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.schema.KeyspaceParams;
|
||||
import org.apache.cassandra.schema.TableMetadata;
|
||||
import org.apache.cassandra.net.MessagingService;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.junit.Assert.fail;
|
||||
|
||||
import static org.apache.cassandra.config.CassandraRelevantProperties.DIAGNOSTIC_SNAPSHOT_INTERVAL_NANOS;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.assertCommandIssued;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.makeRow;
|
||||
import static org.apache.cassandra.db.transform.DuplicateRowCheckerTest.partition;
|
||||
|
||||
public class CompactionIteratorTest extends CQLTester
|
||||
{
|
||||
|
||||
|
|
@ -274,7 +294,7 @@ public class CompactionIteratorTest extends CQLTester
|
|||
transformedSources.put(kk, Iterables.transform(tombstoneSources, list -> listToIterator(list, kk)));
|
||||
try (CompactionController controller = new Controller(Keyspace.openAndGetStore(metadata), transformedSources, GC_BEFORE);
|
||||
CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
|
||||
Lists.transform(content, x -> new Scanner(x)),
|
||||
Lists.transform(content, x -> new Scanner(metadata, x)),
|
||||
controller, NOW, null))
|
||||
{
|
||||
List<Unfiltered> result = new ArrayList<>();
|
||||
|
|
@ -336,7 +356,7 @@ public class CompactionIteratorTest extends CQLTester
|
|||
transformedSources.put(kk, Iterables.transform(tombstoneLists, list -> listToIterator(list, kk)));
|
||||
try (CompactionController controller = new Controller(Keyspace.openAndGetStore(metadata), transformedSources, GC_BEFORE);
|
||||
CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
|
||||
Lists.transform(content, x -> new Scanner(x)),
|
||||
Lists.transform(content, x -> new Scanner(metadata, x)),
|
||||
controller, NOW, null))
|
||||
{
|
||||
assertTrue(iter.hasNext());
|
||||
|
|
@ -369,7 +389,7 @@ public class CompactionIteratorTest extends CQLTester
|
|||
transformedSources.put(kk, Iterables.transform(tombstoneLists, list -> listToIterator(list, kk)));
|
||||
try (CompactionController controller = new Controller(Keyspace.openAndGetStore(metadata), transformedSources, GC_BEFORE);
|
||||
CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
|
||||
Lists.transform(content, x -> new Scanner(x)),
|
||||
Lists.transform(content, x -> new Scanner(metadata, x)),
|
||||
controller, NOW, null))
|
||||
{
|
||||
iter.stop();
|
||||
|
|
@ -404,12 +424,14 @@ public class CompactionIteratorTest extends CQLTester
|
|||
}
|
||||
}
|
||||
|
||||
class Scanner extends AbstractUnfilteredPartitionIterator implements ISSTableScanner
|
||||
static class Scanner extends AbstractUnfilteredPartitionIterator implements ISSTableScanner
|
||||
{
|
||||
Iterator<UnfilteredRowIterator> iter;
|
||||
TableMetadata metadata;
|
||||
|
||||
Scanner(Iterable<UnfilteredRowIterator> content)
|
||||
Scanner(TableMetadata metadata, Iterable<UnfilteredRowIterator> content)
|
||||
{
|
||||
this.metadata = metadata;
|
||||
iter = content.iterator();
|
||||
}
|
||||
|
||||
|
|
@ -500,7 +522,7 @@ public class CompactionIteratorTest extends CQLTester
|
|||
DecoratedKey key = cfs.getPartitioner().decorateKey(ByteBufferUtil.bytes("key"));
|
||||
try (CompactionController controller = new CompactionController(cfs, Integer.MAX_VALUE);
|
||||
UnfilteredRowIterator rows = partition(cfs.metadata(), key, false, unfiltereds);
|
||||
ISSTableScanner scanner = new Scanner(Collections.singletonList(rows));
|
||||
ISSTableScanner scanner = new Scanner(cfs.metadata(), Collections.singletonList(rows));
|
||||
CompactionIterator iter = new CompactionIterator(OperationType.COMPACTION,
|
||||
Collections.singletonList(scanner),
|
||||
controller, FBUtilities.nowInSeconds(), null))
|
||||
|
|
|
|||
|
|
@ -494,7 +494,7 @@ public class ByteSourceComparisonTest extends ByteSourceTestBase
|
|||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, ""), decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
// Note: a decomposed null (e.g. decomposeAndRandomPad(Int32Type.instance, null)) should not reach a tuple
|
||||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, ""), null),
|
||||
tt.pack(null, decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
tt.pack((ByteBuffer) null, decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, "")),
|
||||
tt.pack((ByteBuffer) null),
|
||||
tt.pack()
|
||||
|
|
|
|||
|
|
@ -442,7 +442,7 @@ public class ByteSourceConversionTest extends ByteSourceTestBase
|
|||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, ""), decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
// Note: a decomposed null (e.g. decomposeAndRandomPad(Int32Type.instance, null)) should not reach a tuple
|
||||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, ""), null),
|
||||
tt.pack(null, decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
tt.pack((ByteBuffer) null, decomposeAndRandomPad(Int32Type.instance, 0)),
|
||||
tt.pack(decomposeAndRandomPad(UTF8Type.instance, "")),
|
||||
tt.pack((ByteBuffer) null),
|
||||
tt.pack()
|
||||
|
|
|
|||
Loading…
Reference in New Issue