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()