diff --git a/src/java/org/apache/cassandra/cql3/CQL3Type.java b/src/java/org/apache/cassandra/cql3/CQL3Type.java index 6e2fdcc22e..aba76c0fa3 100644 --- a/src/java/org/apache/cassandra/cql3/CQL3Type.java +++ b/src/java/org/apache/cassandra/cql3/CQL3Type.java @@ -166,8 +166,13 @@ public interface CQL3Type @Override public String toCQLLiteral(ByteBuffer buffer) { - // *always* use the 'blob' syntax to express custom types in CQL - return Native.BLOB.toCQLLiteral(buffer); + CQL3Type asCql3 = type.asCQL3Type(); + if (asCql3 instanceof Custom) + { + // use the 'blob' syntax to express custom types in CQL if not overridden + return Native.BLOB.toCQLLiteral(buffer); + } + return asCql3.toCQLLiteral(buffer); } @Override diff --git a/src/java/org/apache/cassandra/db/marshal/PseudoUtf8Type.java b/src/java/org/apache/cassandra/db/marshal/PseudoUtf8Type.java new file mode 100644 index 0000000000..6c3542808c --- /dev/null +++ b/src/java/org/apache/cassandra/db/marshal/PseudoUtf8Type.java @@ -0,0 +1,82 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.db.marshal; + +import java.nio.ByteBuffer; +import java.nio.charset.CharacterCodingException; +import java.nio.charset.StandardCharsets; + +import org.apache.cassandra.cql3.CQL3Type; +import org.apache.cassandra.cql3.terms.Constants; +import org.apache.cassandra.cql3.terms.Term; +import org.apache.cassandra.serializers.MarshalException; +import org.apache.cassandra.transport.ProtocolVersion; +import org.apache.cassandra.utils.ByteBufferUtil; +import org.apache.cassandra.utils.JsonUtils; + +public abstract class PseudoUtf8Type extends StringType +{ + PseudoUtf8Type() {super(ComparisonType.CUSTOM);} // singleton + + abstract String describe(); + + public ByteBuffer fromString(String source) + { + return decompose(source); + } + + @Override + public Term fromJSONObject(Object parsed) throws MarshalException + { + try + { + return new Constants.Value(fromString((String) parsed)); + } + catch (ClassCastException exc) + { + throw new MarshalException(String.format( + "Expected a %s string, but got a %s: %s", describe(), parsed.getClass().getSimpleName(), parsed)); + } + } + + @Override + public String toJSONString(ByteBuffer buffer, ProtocolVersion protocolVersion) + { + try + { + return '"' + JsonUtils.quoteAsJsonString(ByteBufferUtil.string(buffer, StandardCharsets.UTF_8)) + '"'; + } + catch (CharacterCodingException exc) + { + throw new AssertionError(describe() + " value contained non-utf8 characters: ", exc); + } + } + + @Override + public boolean isCompatibleWith(AbstractType previous) + { + // Anything that is ascii is also utf8, and they both use bytes comparison + return previous == AsciiType.instance || previous == UTF8Type.instance || previous instanceof PseudoUtf8Type; + } + + @Override + public CQL3Type asCQL3Type() + { + return CQL3Type.Native.TEXT; + } +} diff --git a/src/java/org/apache/cassandra/db/marshal/TokenUtf8Type.java b/src/java/org/apache/cassandra/db/marshal/TokenUtf8Type.java new file mode 100644 index 0000000000..caf8980910 --- /dev/null +++ b/src/java/org/apache/cassandra/db/marshal/TokenUtf8Type.java @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.db.marshal; + +import java.nio.ByteBuffer; + +import org.apache.cassandra.cql3.functions.ArgumentDeserializer; +import org.apache.cassandra.dht.Token; +import org.apache.cassandra.serializers.MarshalException; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.serializers.UTF8Serializer; +import org.apache.cassandra.utils.ByteBufferUtil; + +import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; + +public class TokenUtf8Type extends PseudoUtf8Type +{ + public static final TokenUtf8Type instance = new TokenUtf8Type(); + static final TypeSerializer tokenSerializer = new UTF8Serializer() + { + @Override + public void validate(V value, ValueAccessor accessor) throws MarshalException + { + super.validate(value, accessor); + String str = deserialize(value, accessor); + if (null == getPartitioner().getTokenFactory().fromString(str)) + throw new MarshalException("Invalid Token: " + str); + } + }; + + private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance); + private static final ByteBuffer MASKED_VALUE = ByteBufferUtil.EMPTY_BYTE_BUFFER; + + TokenUtf8Type() {} // singleton + + String describe() { return "Token"; } + + @Override + public TypeSerializer getSerializer() + { + return tokenSerializer; + } + + @Override + public int compareCustom(VL left, ValueAccessor accessorL, VR right, ValueAccessor accessorR) + { + String leftStr = UTF8Serializer.instance.deserialize(left, accessorL); + String rightStr = UTF8Serializer.instance.deserialize(right, accessorR); + Token leftToken = getPartitioner().getTokenFactory().fromString(leftStr); + Token rightToken = getPartitioner().getTokenFactory().fromString(rightStr); + return leftToken.compareTo(rightToken); + } + + @Override + public ArgumentDeserializer getArgumentDeserializer() + { + return ARGUMENT_DESERIALIZER; + } + + @Override + public ByteBuffer getMaskedValue() + { + return MASKED_VALUE; + } +} diff --git a/src/java/org/apache/cassandra/db/marshal/TxnIdUtf8Type.java b/src/java/org/apache/cassandra/db/marshal/TxnIdUtf8Type.java new file mode 100644 index 0000000000..784969ead7 --- /dev/null +++ b/src/java/org/apache/cassandra/db/marshal/TxnIdUtf8Type.java @@ -0,0 +1,78 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.db.marshal; + +import java.nio.ByteBuffer; + +import accord.primitives.TxnId; +import org.apache.cassandra.cql3.functions.ArgumentDeserializer; +import org.apache.cassandra.serializers.MarshalException; +import org.apache.cassandra.serializers.TypeSerializer; +import org.apache.cassandra.serializers.UTF8Serializer; +import org.apache.cassandra.utils.ByteBufferUtil; + +public class TxnIdUtf8Type extends PseudoUtf8Type +{ + public static final TxnIdUtf8Type instance = new TxnIdUtf8Type(); + static final TypeSerializer txnIdSerializer = new UTF8Serializer() + { + @Override + public void validate(V value, ValueAccessor accessor) throws MarshalException + { + super.validate(value, accessor); + String str = deserialize(value, accessor); + if (null == TxnId.tryParse(str)) + throw new MarshalException("Invalid TxnId: " + str); + } + }; + + private static final ArgumentDeserializer ARGUMENT_DESERIALIZER = new DefaultArgumentDeserializer(instance); + private static final ByteBuffer MASKED_VALUE = ByteBufferUtil.EMPTY_BYTE_BUFFER; + + TxnIdUtf8Type() {} // singleton + + String describe() { return "TxnId"; } + + @Override + public TypeSerializer getSerializer() + { + return txnIdSerializer; + } + + @Override + public int compareCustom(VL left, ValueAccessor accessorL, VR right, ValueAccessor accessorR) + { + String leftStr = UTF8Serializer.instance.deserialize(left, accessorL); + String rightStr = UTF8Serializer.instance.deserialize(right, accessorR); + TxnId leftId = TxnId.parse(leftStr); + TxnId rightId = TxnId.parse(rightStr); + return leftId.compareTo(rightId); + } + + @Override + public ArgumentDeserializer getArgumentDeserializer() + { + return ARGUMENT_DESERIALIZER; + } + + @Override + public ByteBuffer getMaskedValue() + { + return MASKED_VALUE; + } +} diff --git a/src/java/org/apache/cassandra/db/marshal/TypeParser.java b/src/java/org/apache/cassandra/db/marshal/TypeParser.java index 87df2c3805..2fd1e9e053 100644 --- a/src/java/org/apache/cassandra/db/marshal/TypeParser.java +++ b/src/java/org/apache/cassandra/db/marshal/TypeParser.java @@ -35,6 +35,7 @@ import org.apache.cassandra.cql3.FieldIdentifier; import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.exceptions.ConfigurationException; import org.apache.cassandra.exceptions.SyntaxException; +import org.apache.cassandra.service.StorageService; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; @@ -465,6 +466,11 @@ public class TypeParser { String className = compareWith.contains(".") ? compareWith : "org.apache.cassandra.db.marshal." + compareWith; Class> typeClass = FBUtilities.>classForName(className, "abstract-type"); + if (PseudoUtf8Type.class.isAssignableFrom(typeClass)) + { + if (StorageService.instance.isDaemonSetupCompleted()) + throw new ConfigurationException(typeClass.getName() + " is reserved for internal functionality"); + } try { Method method = typeClass.getDeclaredMethod("getInstance", TypeParser.class); diff --git a/src/java/org/apache/cassandra/db/marshal/UTF8Type.java b/src/java/org/apache/cassandra/db/marshal/UTF8Type.java index 5258321bee..babd6a52da 100644 --- a/src/java/org/apache/cassandra/db/marshal/UTF8Type.java +++ b/src/java/org/apache/cassandra/db/marshal/UTF8Type.java @@ -80,7 +80,7 @@ public class UTF8Type extends StringType { // Anything that is ascii is also utf8, and they both use bytes // comparison - return this == previous || previous == AsciiType.instance; + return this == previous || previous == AsciiType.instance || previous instanceof PseudoUtf8Type; } @Override diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 896c71c0ae..685e44a061 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -149,9 +149,8 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + - " token_sort blob,\n" + - " token_start text,\n" + - " token_end text,\n" + + " token_start 'TokenUtf8Type',\n" + + " token_end 'TokenUtf8Type',\n" + " last_started_at bigint,\n" + " cycle_started_at bigint,\n" + " retries int,\n" + @@ -161,10 +160,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace " waiting text,\n" + " node_offset int,\n" + " cycle_offset int,\n" + - " activeIndex int,\n" + - " nextIndex int,\n" + - " nextToIndex int,\n" + - " endIndex int,\n" + + " active_index int,\n" + + " next_index int,\n" + + " next_to_index int,\n" + + " end_index int,\n" + " current_splits int,\n" + " stopping boolean,\n" + " stopped boolean,\n" + @@ -182,8 +181,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace { TableId tableId = (TableId) view.shard().range.start().prefix(); TableMetadata tableMetadata = tableMetadata(tableId); - ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), sortToken(view.shard().range.start())) - .column("token_start", printToken(view.shard().range.start())) + ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), printToken(view.shard().range.start())) .column("token_end", printToken(view.shard().range.end())) .column("last_started_at", approxTime.translate().toMillisSinceEpoch(view.lastStartedAtMicros() * 1000)) .column("cycle_started_at", approxTime.translate().toMillisSinceEpoch(view.cycleStartedAtMicros() * 1000)) @@ -194,10 +192,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace .column("waiting", Objects.toString(view.waiting())) .column("node_offset", view.nodeOffset()) .column("cycle_offset", view.cycleOffset()) - .column("activeIndex", view.activeIndex()) - .column("nextIndex", view.nextIndex()) - .column("nextToIndex", view.toIndex()) - .column("endIndex", view.cycleLength()) + .column("active_index", view.activeIndex()) + .column("next_index", view.nextIndex()) + .column("next_to_index", view.toIndex()) + .column("end_index", view.cycleLength()) .column("current_splits", view.currentSplits()) .column("stopping", view.stopping()) .column("stopped", view.stopped()) @@ -216,12 +214,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + - " token_sort blob,\n" + - " token_start text,\n" + - " token_end text,\n" + + " token_start 'TokenUtf8Type',\n" + + " token_end 'TokenUtf8Type',\n" + " majority_before text,\n" + " universal_before text,\n" + - " PRIMARY KEY (keyspace_name, table_name, token_sort)" + + " PRIMARY KEY (keyspace_name, table_name, token_start)" + ')', UTF8Type.instance)); } @@ -233,8 +230,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace (entry, ds, start, end) -> { TableId tableId = (TableId) start.prefix(); TableMetadata tableMetadata = tableMetadata(tableId); - ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), sortToken(start)) - .column("token_start", printToken(start)) + ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), printToken(start)) .column("token_end", printToken(end)) .column("majority_before", entry.majorityBefore.toString()) .column("universal_before", entry.universalBefore.toString()); @@ -297,12 +293,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + - " token_sort blob,\n" + - " token_start text,\n" + - " token_end text,\n" + " command_store_id bigint,\n" + + " token_start 'TokenUtf8Type',\n" + + " token_end 'TokenUtf8Type',\n" + " timestamp text,\n" + - " PRIMARY KEY (keyspace_name, table_name, token_sort, command_store_id)" + + " PRIMARY KEY (keyspace_name, table_name, command_store_id, token_start)" + ')', UTF8Type.instance)); } @@ -321,8 +316,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace maxConflicts.foldlWithBounds( (timestamp, ds, start, end) -> { - return ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), sortToken(start), commandStoreId) - .column("token_start", printToken(start)) + return ds.row(keyspace(tableMetadata), table(tableId, tableMetadata), commandStoreId, printToken(start)) .column("token_end", printToken(end)) .column("timestamp", timestamp.toString()) ; @@ -434,7 +428,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace " keyspace_name text,\n" + " table_name text,\n" + " command_store_id int,\n" + - " txn_id text,\n" + + " txn_id 'TxnIdUtf8Type',\n" + // Timer + BaseTxnState " contact_everyone boolean,\n" + // WaitingState @@ -504,22 +498,21 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + - " token_sort blob,\n" + - " token_start text,\n" + - " token_end text,\n" + + " token_start 'TokenUtf8Type',\n" + + " token_end 'TokenUtf8Type',\n" + " command_store_id int,\n" + " start_epoch bigint,\n" + " end_epoch bigint,\n" + - " gc_before text,\n" + - " shard_applied text,\n" + - " majority_applied text,\n" + - " locally_applied text,\n" + - " locally_synced text,\n" + - " locally_redundant text,\n" + - " locally_witnessed text,\n" + - " pre_bootstrap text,\n" + - " stale_until_at_least text,\n" + - " PRIMARY KEY (keyspace_name, table_name, token_sort, command_store_id)" + + " gc_before 'TxnIdUtf8Type',\n" + + " shard_applied 'TxnIdUtf8Type',\n" + + " majority_applied 'TxnIdUtf8Type',\n" + + " locally_applied 'TxnIdUtf8Type',\n" + + " locally_synced 'TxnIdUtf8Type',\n" + + " locally_redundant 'TxnIdUtf8Type',\n" + + " locally_witnessed 'TxnIdUtf8Type',\n" + + " pre_bootstrap 'TxnIdUtf8Type',\n" + + " stale_until_at_least 'TxnIdUtf8Type',\n" + + " PRIMARY KEY (keyspace_name, table_name, command_store_id, token_start)" + ')', UTF8Type.instance)); } @@ -538,8 +531,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace String table = table(tableId, tableMetadata); commandStore.unsafeGetRedundantBefore().foldl( (entry, ds) -> { - ds.row(keyspace, table, sortToken(entry.range.start()), commandStoreId) - .column("token_start", printToken(entry.range.start())) + ds.row(keyspace, table, commandStoreId, printToken(entry.range.start())) .column("token_end", printToken(entry.range.end())) .column("start_epoch", entry.startEpoch) .column("end_epoch", entry.endEpoch) @@ -571,12 +563,11 @@ public class AccordDebugKeyspace extends VirtualKeyspace "CREATE TABLE %s (\n" + " keyspace_name text,\n" + " table_name text,\n" + - " token_sort blob,\n" + - " token_start text,\n" + - " token_end text,\n" + " command_store_id int,\n" + - " txn_id text,\n" + - " PRIMARY KEY (keyspace_name, table_name, token_sort, command_store_id)" + + " token_start 'TokenUtf8Type',\n" + + " token_end 'TokenUtf8Type',\n" + + " timestamp text,\n" + + " PRIMARY KEY (keyspace_name, table_name, command_store_id, token_start)" + ')', UTF8Type.instance)); } @@ -596,10 +587,10 @@ public class AccordDebugKeyspace extends VirtualKeyspace String keyspace = keyspace(tableMetadata); String table = table(tableId, tableMetadata); rejectBefore.foldlWithBounds( - (txnId, ds, start, end) -> ds.row(keyspace, table, sortToken(start), commandStore.id()) + (timestamp, ds, start, end) -> ds.row(keyspace, table, sortToken(start), commandStore.id()) .column("token_start", printToken(start)) .column("token_end", printToken(end)) - .column("txn_id", txnId.toString()) + .column("timestamp", timestamp.toString()) , dataSet, ignore -> false diff --git a/src/java/org/apache/cassandra/serializers/UTF8Serializer.java b/src/java/org/apache/cassandra/serializers/UTF8Serializer.java index 38faf430b7..7b071e7131 100644 --- a/src/java/org/apache/cassandra/serializers/UTF8Serializer.java +++ b/src/java/org/apache/cassandra/serializers/UTF8Serializer.java @@ -25,7 +25,7 @@ public class UTF8Serializer extends AbstractTextSerializer { public static final UTF8Serializer instance = new UTF8Serializer(); - private UTF8Serializer() + protected UTF8Serializer() { super(StandardCharsets.UTF_8); } diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index bb30e914b4..4d4067786a 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -33,8 +33,10 @@ import org.slf4j.LoggerFactory; import accord.api.ProtocolModifiers; import accord.messages.TxnRequest; +import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.SaveStatus; +import accord.primitives.Status; import accord.primitives.Txn; import accord.primitives.TxnId; import accord.utils.async.AsyncChains; @@ -42,12 +44,17 @@ import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.config.OptionaldPositiveInt; import org.apache.cassandra.cql3.CQLTester; import org.apache.cassandra.cql3.UntypedResultSet; +import org.apache.cassandra.dht.Murmur3Partitioner.LongToken; import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.net.Verb; +import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; +import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.accord.AccordService; +import org.apache.cassandra.service.accord.TokenRange; +import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.concurrent.Condition; @@ -99,6 +106,12 @@ public class AccordDebugKeyspaceTest extends CQLTester private static final String QUERY_REDUNDANT_BEFORE = String.format("SELECT * FROM %s.%s", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE); + private static final String QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ = + String.format("SELECT * FROM %s.%s WHERE majority_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE); + + private static final String QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ = + String.format("SELECT * FROM %s.%s WHERE shard_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE); + @BeforeClass public static void setUpClass() { @@ -171,11 +184,25 @@ public class AccordDebugKeyspaceTest extends CQLTester } @Test - public void redundantBefore() + public void redundantBefore() throws ExecutionException, InterruptedException { String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'"); var accord = accord(); + TableId tableId = Schema.instance.getTableMetadata(KEYSPACE, tableName).id; + TxnId syncId1 = new TxnId(100, 200, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, accord.nodeId()); + TxnId syncId2 = new TxnId(101, 300, Txn.Kind.ExclusiveSyncPoint, Routable.Domain.Range, accord.nodeId()); + Ranges ranges1 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(1)), new TokenKey(tableId, new LongToken(100)))); + Ranges ranges2 = Ranges.of(TokenRange.create(new TokenKey(tableId, new LongToken(100)), new TokenKey(tableId, new LongToken(200)))); + AsyncChains.getBlocking(accord.node().commandStores().forEach(safeStore -> { + safeStore.commandStore().markShardDurable(safeStore, syncId1, ranges1, Status.Durability.Universal); + safeStore.commandStore().markShardDurable(safeStore, syncId2, ranges2, Status.Durability.Majority); + })); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE).size()).isGreaterThan(0); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ, syncId1.toString()).size()).isEqualTo(2); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_MAJORITY_APPLIED_GEQ, syncId2.toString()).size()).isEqualTo(1); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ, syncId1.toString()).size()).isEqualTo(1); + Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ, syncId2.toString()).size()).isEqualTo(0); } @Test diff --git a/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java b/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java index 6c85ad98fd..e6c904dd88 100644 --- a/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java +++ b/test/unit/org/apache/cassandra/utils/AbstractTypeGenerators.java @@ -88,7 +88,9 @@ import org.apache.cassandra.db.marshal.StringType; import org.apache.cassandra.db.marshal.TimeType; import org.apache.cassandra.db.marshal.TimeUUIDType; import org.apache.cassandra.db.marshal.TimestampType; +import org.apache.cassandra.db.marshal.TokenUtf8Type; import org.apache.cassandra.db.marshal.TupleType; +import org.apache.cassandra.db.marshal.TxnIdUtf8Type; import org.apache.cassandra.db.marshal.TypeParser; import org.apache.cassandra.db.marshal.UTF8Type; import org.apache.cassandra.db.marshal.UUIDType; @@ -120,6 +122,8 @@ public final class AbstractTypeGenerators .put((Class>) (Class) ReversedType.class, "Implementation detail for cluster ordering... its expected the caller will unwrap the clustering type to always get access to the real type") .put(DynamicCompositeType.FixedValueComparator.class, "Hack type used for special ordering case, not a real/valid type") .put(FrozenType.class, "Fake class only used during parsing... the parsing creates this and the real type under it, then this gets swapped for the real type") + .put(TxnIdUtf8Type.class, "Used only internally by accord debug virtual tables - could be tested, but class initialisation order prevents easy reuse of the relevant type generators") + .put(TokenUtf8Type.class, "Used only internally by accord debug virtual tables - could be tested, but class initialisation order prevents easy reuse of the relevant type generators") .build(); /** @@ -136,7 +140,6 @@ public final class AbstractTypeGenerators return (String a, String b) -> FastByteOperations.compareUnsigned(st.decompose(a), st.decompose(b)); } - private static final Map, TypeSupport> PRIMITIVE_TYPE_DATA_GENS = Stream.of(TypeSupport.of(BooleanType.instance, BOOLEAN_GEN), TypeSupport.of(ByteType.instance, SourceDSL.integers().between(0, Byte.MAX_VALUE * 2 + 1).map(Integer::byteValue)),