From 10e065ffe5453a3ba9453d5fa1333ab74ead312f Mon Sep 17 00:00:00 2001 From: David Capwell Date: Wed, 16 Aug 2023 15:36:01 -0700 Subject: [PATCH] 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 --- modules/accord | 2 +- .../db/compaction/CompactionIterator.java | 112 +++++---- .../cassandra/db/marshal/TupleType.java | 5 + .../service/accord/AccordKeyspace.java | 237 ++++++++++++++---- test/unit/org/apache/cassandra/Util.java | 37 ++- .../CompactionAccordIteratorsTest.java | 183 +++++++++++--- .../db/compaction/CompactionIteratorTest.java | 56 +++-- .../ByteSourceComparisonTest.java | 2 +- .../ByteSourceConversionTest.java | 2 +- 9 files changed, 483 insertions(+), 153 deletions(-) diff --git a/modules/accord b/modules/accord index 8c7a3c9ef4..1d6028ca20 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 8c7a3c9ef4209d635b186189e17a2d9e728e9871 +Subproject commit 1d6028ca20553d1c1a6fe2809b204254955da3b3 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 85f5245969..200d8687f3 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -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); - } } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/db/marshal/TupleType.java b/src/java/org/apache/cassandra/db/marshal/TupleType.java index d6ce2da0f4..ff0b943078 100644 --- a/src/java/org/apache/cassandra/db/marshal/TupleType.java +++ b/src/java/org/apache/cassandra/db/marshal/TupleType.java @@ -351,6 +351,11 @@ public class TupleType extends MultiElementType return "component"; } + public static V pack(ValueAccessor accessor, V... components) + { + return pack(accessor, Arrays.asList(components)); + } + public static V pack(ValueAccessor accessor, Collection components) { int totalLength = 0; diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index 5f31f86276..bfaf764592 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -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 deserializeTimestampOrNull(ByteBuffer bytes, TimestampFactory 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 Timestamp deserializeTimestampOrNull(Cell cell) + { + if (cell == null) + return null; + ValueAccessor accessor = cell.accessor(); + V value = cell.value(); + if (accessor.isEmpty(value)) + return null; + List 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 T deserializeTimestampOrNull(V value, ValueAccessor accessor, TimestampFactory 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 deserializeWithVersionOr(UntypedResultSet.Row row, String dataColumn, LocalVersionedSerializer serializer, Supplier 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 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 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) diff --git a/test/unit/org/apache/cassandra/Util.java b/test/unit/org/apache/cassandra/Util.java index 60203f5e25..22b0eb9999 100644 --- a/test/unit/org/apache/cassandra/Util.java +++ b/test/unit/org/apache/cassandra/Util.java @@ -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 call) + { + spinAssertEquals(null, expected, timeoutInSeconds, TimeUnit.SECONDS, call); + } + + public static void spinAssertEquals(String message, T expected, long timeout, TimeUnit timeUnit, Callable 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); diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index e09a183b15..c51e161d4d 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -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> expectedResult) throws Throwable + private void testAccordCommandsPurger(RedundantBefore redundantBefore, DurableBefore durableBefore, Consumer> 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> expectedResult) throws Throwable + private void testAccordCommandsForKeyPurger(RedundantBefore redundantBefore, Consumer> 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 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 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 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 compactCFS(IAccordService mockAccordService, ColumnFamilyStore cfs) + private List compactCFS(IAccordService mockAccordService, ColumnFamilyStore cfs) { - cfs.forceBlockingFlush(FlushReason.UNIT_TESTS); List scanners = cfs.getLiveSSTables().stream().map(SSTableReader::getScanner).collect(Collectors.toList()); - List 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 result = null; + do { - while (compactionIterator.hasNext()) + List outputPartitions = new ArrayList<>(); + List 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; } } diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java index 076ef9876f..d09c955173 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionIteratorTest.java @@ -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 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 iter; + TableMetadata metadata; - Scanner(Iterable content) + Scanner(TableMetadata metadata, Iterable 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)) diff --git a/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceComparisonTest.java b/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceComparisonTest.java index 496f6355c1..e1f11ac15a 100644 --- a/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceComparisonTest.java +++ b/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceComparisonTest.java @@ -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() diff --git a/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceConversionTest.java b/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceConversionTest.java index 7ab30adfa7..a6f34acadf 100644 --- a/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceConversionTest.java +++ b/test/unit/org/apache/cassandra/utils/bytecomparable/ByteSourceConversionTest.java @@ -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()