natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RemoteToLocalVirtualKeyspace: supporting access to all nodes' local virtual tables from any node in the cluster 

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20900
This commit is contained in:
Benedict Elliott Smith 2025-09-13 10:50:32 +01:00
parent f7743fe3ab
commit b712cfb930
15 changed files with 1025 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
src/java/org/apache/cassandra/db/filter/ColumnFilter.java
View File

@ -33,6 +33,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.btree.BTree;
/**
* Represents which (non-PK) columns (and optionally which sub-part of a column for complex columns) are selected
@ -64,7 +65,6 @@ import org.apache.cassandra.schema.TableMetadata;
*/
public abstract class ColumnFilter
{
public static final ColumnFilter NONE = selection(RegularAndStaticColumns.NONE);
public static final Serializer serializer = new Serializer();
@ -305,6 +305,20 @@ public abstract class ColumnFilter
return false;
}
/**
* Rebinds matching columns into a new filter; ignores any missing but fails if any are a different type
*/
abstract ColumnFilter rebind(TableMetadata newTable);
public static ColumnFilter rebindVirtual(ColumnFilter filter, TableMetadata newTable)
{
// review feedback; nothing actually preventing its use with other tables,
// but unclear utility/rationale so just some protection against incorrect usage
if (!newTable.isVirtual())
throw new UnsupportedOperationException("This feature is intended only to be used with virtual keyspaces");
return filter.rebind(newTable);
}
/**
* Returns the CQL string corresponding to this {@code ColumnFilter}.
*
@ -630,6 +644,12 @@ public abstract class ColumnFilter
return true;
}
@Override
ColumnFilter rebind(TableMetadata newTable)
{
return new WildCardColumnFilter(ColumnFilter.rebind(newTable, fetchedAndQueried));
}
@Override
protected SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections()
{
@ -779,6 +799,21 @@ public abstract class ColumnFilter
return new Tester(fetchingStrategy.fetchesAllColumns(column.isStatic()), s.iterator());
}
@Override
ColumnFilter rebind(TableMetadata newTable)
{
RegularAndStaticColumns queried = ColumnFilter.rebind(newTable, this.queried);
RegularAndStaticColumns fetched = this.queried == this.fetched ? queried : ColumnFilter.rebind(newTable, this.fetched);
SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections = null;
if (this.subSelections != null)
{
subSelections = TreeMultimap.create();
for (Map.Entry<ColumnIdentifier, ColumnSubselection> e : this.subSelections.entries())
subSelections.put(e.getKey(), e.getValue().rebind(newTable));
}
return new SelectionColumnFilter(fetchingStrategy, queried, fetched, subSelections);
}
@Override
protected SortedSetMultimap<ColumnIdentifier, ColumnSubselection> subSelections()
{
@ -1003,4 +1038,26 @@ public abstract class ColumnFilter
return size;
}
}
private static RegularAndStaticColumns rebind(TableMetadata newTable, RegularAndStaticColumns columns)
{
return new RegularAndStaticColumns(rebind(newTable, columns.statics), rebind(newTable, columns.regulars));
}
private static Columns rebind(TableMetadata newTable, Columns columns)
{
if (columns.isEmpty())
return columns;
try (BTree.FastBuilder<ColumnMetadata> builder = BTree.fastBuilder())
{
for (ColumnMetadata in : columns)
{
ColumnMetadata out = newTable.getColumn(in.name);
if (out != null)
builder.add(out);
}
return Columns.from(builder);
}
}
}

14
src/java/org/apache/cassandra/db/filter/ColumnSubselection.java
View File

@ -76,6 +76,8 @@ public abstract class ColumnSubselection implements Comparable<ColumnSubselectio
protected abstract CellPath comparisonPath();
protected abstract ColumnSubselection rebind(TableMetadata newTable);
public int compareTo(ColumnSubselection other)
{
assert other.column().name.equals(column().name);
@ -118,6 +120,12 @@ public abstract class ColumnSubselection implements Comparable<ColumnSubselectio
return from;
}
@Override
protected ColumnSubselection rebind(TableMetadata newTable)
{
return new Slice(newTable.getColumn(column.name), from, to);
}
public int compareInclusionOf(CellPath path)
{
Comparator<CellPath> cmp = column.cellPathComparator();
@ -160,6 +168,12 @@ public abstract class ColumnSubselection implements Comparable<ColumnSubselectio
return element;
}
@Override
protected ColumnSubselection rebind(TableMetadata newTable)
{
return new Element(newTable.getColumn(column.name), element);
}
public int compareInclusionOf(CellPath path)
{
return column.cellPathComparator().compare(path, element);

50
src/java/org/apache/cassandra/db/filter/RowFilter.java
View File

@ -141,7 +141,7 @@ public class RowFilter implements Iterable<RowFilter.Expression>
add(new CustomExpression(metadata, targetIndex, value));
}
private void add(Expression expression)
public void add(Expression expression)
{
expression.validate();
expressions.add(expression);
@ -549,6 +549,28 @@ public class RowFilter implements Iterable<RowFilter.Expression>
"Index expression values may not be larger than 64K");
}
/**
* Rebind this expression to a table metadata that is expected to have equivalent columns.
* If any referenced column is missing, returns null;
* if any referenced column has a different type throws an exception
*/
public Expression rebind(TableMetadata newTable)
{
throw new UnsupportedOperationException("Expression " + toString(true) + " does not support rebinding to another table definition");
}
protected static ColumnMetadata rebind(ColumnMetadata in, TableMetadata newTable)
{
ColumnMetadata out = newTable.getColumn(in.name);
if (out == null)
return null;
if (!out.type.equals(in.type) && !out.type.isCompatibleWith(in.type) || !in.type.isCompatibleWith(out.type))
throw new IllegalArgumentException("The provided TableMetadata is not compatible with the expression");
return out;
}
/**
* Returns whether the provided row satisfied this expression or not.
*
@ -734,6 +756,16 @@ public class RowFilter implements Iterable<RowFilter.Expression>
super(column, operator, value);
}
@Override
public Expression rebind(TableMetadata newTable)
{
ColumnMetadata out = rebind(column, newTable);
if (out == null)
return null;
return new SimpleExpression(out, operator, value);
}
@Override
public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)
{
@ -853,6 +885,16 @@ public class RowFilter implements Iterable<RowFilter.Expression>
checkBindValueSet(value, "Unsupported unset map value for column %s", column.name);
}
@Override
public Expression rebind(TableMetadata newTable)
{
ColumnMetadata out = rebind(column, newTable);
if (out == null)
return null;
return new MapElementExpression(out, key, operator, value);
}
@Override
public ByteBuffer getIndexValue()
{
@ -978,6 +1020,12 @@ public class RowFilter implements Iterable<RowFilter.Expression>
return Kind.CUSTOM;
}
@Override
public Expression rebind(TableMetadata newTable)
{
return new CustomExpression(table, targetIndex, value);
}
// Filtering by custom expressions isn't supported yet, so just accept any row
@Override
public boolean isSatisfiedBy(TableMetadata metadata, DecoratedKey partitionKey, Row row, long nowInSec)

31
src/java/org/apache/cassandra/db/virtual/AccordDebugRemoteKeyspace.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.virtual;
import org.apache.cassandra.schema.SchemaConstants;
public class AccordDebugRemoteKeyspace extends RemoteToLocalVirtualKeyspace
{
public static final AccordDebugRemoteKeyspace instance = new AccordDebugRemoteKeyspace(SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.instance);
public AccordDebugRemoteKeyspace(String name, VirtualKeyspace wrap)
{
super(name, wrap);
}
}

35
src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualKeyspace.java Normal file
View File

@ -0,0 +1,35 @@
/*
* 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.virtual;
import java.util.function.Predicate;
import java.util.stream.Collectors;
public class RemoteToLocalVirtualKeyspace extends VirtualKeyspace
{
public RemoteToLocalVirtualKeyspace(String name, VirtualKeyspace wrap)
{
this(name, wrap, ignore -> true);
}
public RemoteToLocalVirtualKeyspace(String name, VirtualKeyspace wrap, Predicate<VirtualTable> include)
{
super(name, wrap.tables().stream().filter(include).map(vt -> new RemoteToLocalVirtualTable(name, vt)).collect(Collectors.toList()));
}
}

625
src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualTable.java Normal file
View File

@ -0,0 +1,625 @@
/*
* 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.virtual;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.NavigableSet;
import java.util.function.Function;
import accord.utils.Invariants;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.BufferClusteringBound;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ClusteringBound;
import org.apache.cassandra.db.ClusteringPrefix;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionInfo;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.PartitionRangeReadCommand;
import org.apache.cassandra.db.RangeTombstone;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadResponse;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.TruncateRequest;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.CompositeType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.RequestCallback;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Promise;
import org.apache.cassandra.utils.concurrent.SyncPromise;
import static org.apache.cassandra.db.ClusteringBound.BOTTOM;
import static org.apache.cassandra.db.ClusteringBound.TOP;
import static org.apache.cassandra.db.ClusteringBound.boundKind;
import static org.apache.cassandra.db.ReadCommand.PotentialTxnConflicts.ALLOW;
import static org.apache.cassandra.db.virtual.VirtualTable.Sorted.SORTED;
public class RemoteToLocalVirtualTable extends AbstractLazyVirtualTable
{
private static final int MAX_CONCURRENCY = 8;
final TableMetadata local;
final boolean allowFilteringImplicitly;
final boolean allowFilteringLocalPartitionKeysImplicitly;
public RemoteToLocalVirtualTable(String keyspace, VirtualTable virtualTable)
{
super(wrap(keyspace, virtualTable.name(), virtualTable.metadata()), virtualTable instanceof AbstractLazyVirtualTable ? ((AbstractLazyVirtualTable) virtualTable).onTimeout() : OnTimeout.FAIL, virtualTable.sorted(), SORTED);
this.local = virtualTable.metadata();
this.allowFilteringImplicitly = virtualTable.allowFilteringImplicitly();
this.allowFilteringLocalPartitionKeysImplicitly = virtualTable.allowFilteringPrimaryKeysImplicitly();
}
@Override
public boolean allowFilteringImplicitly()
{
return allowFilteringImplicitly;
}
@Override
public boolean allowFilteringPrimaryKeysImplicitly()
{
return true;
}
private static TableMetadata wrap(String keyspace, String name, TableMetadata local)
{
if (local.partitionKeyColumns().size() != 1 && !(local.partitionKeyType instanceof CompositeType))
throw new IllegalArgumentException("Underlying table must have a single partition key, else use CompositeType for its partitioner");
TableMetadata.Builder builder = TableMetadata.builder(keyspace, name);
builder.partitioner(new LocalPartitioner(Int32Type.instance));
builder.addPartitionKeyColumn("node_id", Int32Type.instance);
for (ColumnMetadata cm : local.partitionKeyColumns())
builder.addClusteringColumn(cm.name, cm.type, cm.getMask(), cm.getColumnConstraints());
for (ColumnMetadata cm : local.clusteringColumns())
builder.addClusteringColumn(cm.name, cm.type, cm.getMask(), cm.getColumnConstraints());
// we don't add static columns as they can't be modelled correctly with the insertion of a prefix partition column
for (ColumnMetadata cm : local.regularColumns())
{
if (cm.isComplex())
throw new IllegalArgumentException("Complex columns are not currently supported by " + RemoteToLocalVirtualTable.class.getSimpleName());
builder.addRegularColumn(cm.name, cm.type, cm.getMask(), cm.getColumnConstraints());
}
builder.kind(TableMetadata.Kind.VIRTUAL);
return builder.build();
}
@Override
protected void collect(PartitionsCollector collector)
{
ClusterMetadata cm = ClusterMetadata.current();
NavigableSet<NodeId> matchingIds = cm.directory.peerIds();
DataRange dataRange = collector.dataRange();
AbstractBounds<PartitionPosition> bounds = dataRange.keyRange();
{
NodeId start = null;
if (!bounds.left.isMinimum())
{
if (!(bounds.left instanceof DecoratedKey))
throw new InvalidRequestException(metadata + " does not support filtering by token or incomplete partition keys");
start = new NodeId(Int32Type.instance.compose(((DecoratedKey) bounds.left).getKey()));
}
NodeId end = null;
if (!bounds.right.isMinimum())
{
if (!(bounds.right instanceof DecoratedKey))
throw new InvalidRequestException(metadata + " does not support filtering by token or incomplete partition keys");
end = new NodeId(Int32Type.instance.compose(((DecoratedKey) bounds.right).getKey()));
}
if (start != null && end != null) matchingIds = matchingIds.subSet(start, bounds.isStartInclusive(), end, bounds.isEndInclusive());
else if (start != null) matchingIds = matchingIds.tailSet(start, bounds.isStartInclusive());
else if (end != null) matchingIds = matchingIds.headSet(end, bounds.isEndInclusive());
}
if (dataRange.isReversed())
matchingIds = matchingIds.descendingSet();
RowFilter rowFilter = rebind(local, collector.rowFilter());
ColumnFilter columnFilter = ColumnFilter.rebindVirtual(collector.columnFilter(), local);
// TODO (expected): count this down as we progress where possible (or have AbstractLazyVirtualTable do it for us)
DataLimits limits = collector.limits();
Function<DecoratedKey, ByteBuffer[]> pksToCks = partitionKeyToClusterings(metadata, local);
ArrayDeque<RequestAndResponse> pending = new ArrayDeque<>();
matchingIds.forEach(id -> {
InetAddressAndPort endpoint = cm.directory.endpoint(id);
DecoratedKey remoteKey = metadata.partitioner.decorateKey(Int32Type.instance.decompose(id.id()));
ClusteringIndexFilter filter = dataRange.clusteringIndexFilter(remoteKey);
Slices slices = filter.getSlices(metadata);
int i = 0, advance = 1, end = slices.size();
if (dataRange.isReversed())
{
i = slices.size() - 1;
end = -1;
advance = -1;
}
PartitionCollector partition = collector.partition(id.id());
while (i != end)
{
List<Request> request = rebind(local, slices.get(i), dataRange.isReversed(), rowFilter, columnFilter);
for (Request send : request)
{
ReadCommand readCommand;
if (send.dataRange.startKey().equals(send.dataRange.stopKey()) && !send.dataRange.startKey().isMinimum())
readCommand = SinglePartitionReadCommand.create(local, collector.nowInSeconds(), send.columnFilter, send.rowFilter, limits, (DecoratedKey) send.dataRange.startKey(), send.dataRange.clusteringIndexFilter(remoteKey), ALLOW);
else
readCommand = PartitionRangeReadCommand.create(local, collector.nowInSeconds(), send.columnFilter, send.rowFilter, limits, send.dataRange);
RequestAndResponse rr = new RequestAndResponse(id, partition, readCommand);
send(rr, endpoint);
pending.addLast(rr);
boolean selectsOneRow = selectsOneRow(local, send.dataRange, remoteKey);
while (pending.size() >= (selectsOneRow ? 1 : MAX_CONCURRENCY))
collect(collector, pending.pollFirst(), pksToCks);
}
i += advance;
}
});
while (!pending.isEmpty())
collect(collector, pending.pollFirst(), pksToCks);
}
private static class RequestAndResponse extends SyncPromise<ReadResponse>
{
final NodeId nodeId;
final PartitionCollector partition;
final ReadCommand readCommand;
private RequestAndResponse(NodeId nodeId, PartitionCollector partition, ReadCommand readCommand)
{
this.nodeId = nodeId;
this.partition = partition;
this.readCommand = readCommand;
}
}
private static class Request
{
final DataRange dataRange;
final RowFilter rowFilter;
final ColumnFilter columnFilter;
private Request(DataRange dataRange, RowFilter rowFilter, ColumnFilter columnFilter)
{
this.dataRange = dataRange;
this.rowFilter = rowFilter;
this.columnFilter = columnFilter;
}
}
private void send(RequestAndResponse rr, InetAddressAndPort endpoint)
{
send(Verb.READ_REQ, rr.readCommand, rr, endpoint);
}
private <Reply> Promise<Reply> send(Verb verb, Object payload, InetAddressAndPort endpoint)
{
Promise<Reply> promise = new AsyncPromise<>();
send(verb, payload, promise, endpoint);
return promise;
}
private <Reply> void send(Verb verb, Object payload, Promise<Reply> promise, InetAddressAndPort endpoint)
{
if (!FailureDetector.instance.isAlive(endpoint))
{
promise.trySuccess(null);
return;
}
// we have to send inline some of the MessagingService logic to circumvent the requirement to use AbstractWriteResponseHandler
Message<?> message = Message.out(verb, payload);
RequestCallback<?> callback = new RequestCallback<Reply>()
{
@Override public void onResponse(Message<Reply> msg) { promise.trySuccess(msg.payload); }
@Override public boolean invokeOnFailure() { return true; }
@Override public void onFailure(InetAddressAndPort from, RequestFailure failure)
{
if (failure.failure == null) promise.tryFailure(new RuntimeException(failure.reason.toString()));
else promise.tryFailure(failure.failure);
}
};
MessagingService.instance().sendWithCallback(message, endpoint, callback);
}
private void collect(PartitionsCollector collector, RequestAndResponse rr, Function<DecoratedKey, ByteBuffer[]> pksToCks)
{
if (!rr.awaitUntilThrowUncheckedOnInterrupt(collector.deadlineNanos()))
throw new InternalTimeoutException();
rr.rethrowIfFailed();
ReadResponse response = rr.getNow();
if (response == null)
{
ClientWarn.instance.warn("No response from " + rr.nodeId);
return;
}
int pkCount = local.partitionKeyColumns().size();
try (UnfilteredPartitionIterator partitions = response.makeIterator(rr.readCommand))
{
while (partitions.hasNext())
{
try (UnfilteredRowIterator iter = partitions.next())
{
ByteBuffer[] clusterings = pksToCks.apply(iter.partitionKey());
while (iter.hasNext())
{
Unfiltered next = iter.next();
if (!next.isRow())
throw new UnsupportedOperationException("Range tombstones not supported");
Row row = (Row)next;
{
Clustering<?> clustering = row.clustering();
for (int j = 0 ; j < clustering.size(); ++j)
clusterings[pkCount + j] = clustering.bufferAt(j);
}
rr.partition.collect(rows -> {
rows.add((Object[])clusterings)
.lazyCollect(columns -> {
row.forEach(cd -> {
Invariants.require(cd instanceof Cell);
columns.add(cd.column().name.toString(), ((Cell<?>) cd).buffer());
});
});
});
}
}
}
}
}
private static boolean selectsOneRow(TableMetadata metadata, DataRange dataRange, DecoratedKey key)
{
if (dataRange.startKey().isMinimum() || !dataRange.startKey().equals(dataRange.stopKey()))
return false;
if (metadata.clusteringColumns().isEmpty())
return true;
Slices slices = dataRange.clusteringIndexFilter(key).getSlices(metadata);
if (slices.size() != 1)
return false;
Slice slice = slices.get(0);
return slice.start().equals(slice.end());
}
private static Function<DecoratedKey, ByteBuffer[]> partitionKeyToClusterings(TableMetadata distributed, TableMetadata local)
{
ByteBuffer[] cks = new ByteBuffer[distributed.clusteringColumns().size()];
if (local.partitionKeyColumns().size() == 1)
{
return pk -> {
cks[0] = pk.getKey();
return cks.clone();
};
}
CompositeType type = (CompositeType) local.partitionKeyType;
int pkCount = type.types.size();
return (pk) -> {
System.arraycopy(type.split(pk.getKey()), 0, cks, 0, pkCount);
return cks.clone();
};
}
private static RowFilter rebind(TableMetadata local, RowFilter rowFilter)
{
if (rowFilter.isEmpty())
return rowFilter;
RowFilter result = RowFilter.create(false);
for (RowFilter.Expression in : rowFilter.getExpressions())
{
RowFilter.Expression out = in.rebind(local);
if (out != null)
result.add(out);
}
return result;
}
private List<Request> rebind(TableMetadata local, Slice slice, boolean reversed, RowFilter rowFilter, ColumnFilter columnFilter)
{
ClusteringBound<?> start = slice.start();
ClusteringBound<?> end = slice.end();
int pkCount = local.partitionKeyColumns().size();
// TODO (expected): we can filter by partition key by inserting a new row filter, but need to impose ALLOW FILTERING restrictions
if (((start.size() > 0 && start.size() < pkCount) || (end.size() > 0 && end.size() < pkCount)))
{
if (!allowFilteringLocalPartitionKeysImplicitly)
throw new InvalidRequestException("Must specify full partition key bounds for the underlying table");
List<ColumnMetadata> pks = local.partitionKeyColumns();
ByteBuffer[] starts = start.getBufferArray();
ByteBuffer[] ends = end.getBufferArray();
int minCount = Math.min(start.size(), end.size());
int maxCount = Math.max(start.size(), end.size());
int commonPrefixLength = 0;
while (commonPrefixLength < minCount && equalPart(start, end, commonPrefixLength))
++commonPrefixLength;
RowFilter commonRowFilter = rowFilter;
if (commonPrefixLength > 0)
{
commonRowFilter = copy(commonRowFilter);
for (int i = 0 ; i < commonPrefixLength ; ++i)
commonRowFilter.add(pks.get(i), Operator.EQ, starts[i]);
}
Operator lastStartOp = start.isInclusive() ? Operator.GTE : Operator.GT;
Operator lastEndOp = end.isInclusive() ? Operator.LTE : Operator.LT;
if (commonPrefixLength == Math.max(minCount, maxCount - 1))
{
// can simply add our remaining filters and continue on our way
addExpressions(commonRowFilter, pks, commonPrefixLength, starts, Operator.GTE, lastStartOp);
addExpressions(commonRowFilter, pks, commonPrefixLength, ends, Operator.LTE, lastEndOp);
return List.of(new Request(DataRange.allData(local.partitioner), commonRowFilter, columnFilter));
}
throw new InvalidRequestException("This table currently does not support the complex partial partition key filters implied for the underlying table");
}
ByteBuffer[] startBuffers = start.getBufferArray();
PartitionPosition startBound;
if (start.size() == 0) startBound = local.partitioner.getMinimumToken().minKeyBound();
else if (pkCount == 1) startBound = local.partitioner.decorateKey(startBuffers[0]);
else startBound = local.partitioner.decorateKey(CompositeType.build(ByteBufferAccessor.instance, Arrays.copyOf(startBuffers, pkCount)));
ByteBuffer[] endBuffers = end.getBufferArray();
PartitionPosition endBound;
if (end.size() == 0) endBound = local.partitioner.getMinimumToken().maxKeyBound();
else if (pkCount == 1) endBound = local.partitioner.decorateKey(endBuffers[0]);
else endBound = local.partitioner.decorateKey(CompositeType.build(ByteBufferAccessor.instance, Arrays.copyOf(endBuffers, pkCount)));
AbstractBounds<PartitionPosition> bounds = AbstractBounds.bounds(startBound, start.isEmpty() || start.size() > pkCount || start.isInclusive(),
endBound, end.isEmpty() || end.size() > pkCount || end.isInclusive());
boolean hasSlices = start.size() > pkCount || end.size() > pkCount;
if (!hasSlices)
return List.of(new Request(new DataRange(bounds, new ClusteringIndexSliceFilter(Slices.ALL, reversed)), rowFilter, columnFilter));
ClusteringBound<?> startSlice = ClusteringBound.BOTTOM;
if (start.size() > pkCount)
startSlice = BufferClusteringBound.create(boundKind(true, start.isInclusive()), Arrays.copyOfRange(startBuffers, pkCount, startBuffers.length));
ClusteringBound<?> endSlice = ClusteringBound.TOP;
if (end.size() > pkCount)
endSlice = BufferClusteringBound.create(boundKind(false, end.isInclusive()), Arrays.copyOfRange(endBuffers, pkCount, endBuffers.length));
if (startBound.equals(endBound))
return List.of(new Request(new DataRange(bounds, filter(local, startSlice, endSlice, reversed)), rowFilter, columnFilter));
List<Request> result = new ArrayList<>(3);
if (startSlice != BOTTOM)
{
AbstractBounds<PartitionPosition> startBoundOnly = AbstractBounds.bounds(startBound, true, startBound, true);
result.add(new Request(new DataRange(startBoundOnly, filter(local, startSlice, TOP, reversed)), rowFilter, columnFilter));
}
result.add(new Request(new DataRange(AbstractBounds.bounds(bounds.left, bounds.inclusiveLeft() && startSlice == BOTTOM,
bounds.right, bounds.inclusiveRight() && endSlice == TOP),
new ClusteringIndexSliceFilter(Slices.ALL, reversed)), rowFilter, columnFilter)
);
if (endSlice != TOP)
{
AbstractBounds<PartitionPosition> endBoundOnly = AbstractBounds.bounds(endBound, true, endBound, true);
result.add(new Request(new DataRange(endBoundOnly, filter(local, BOTTOM, endSlice, reversed)), rowFilter, columnFilter));
}
if (reversed)
Collections.reverse(result);
return result;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private static boolean equalPart(ClusteringBound start, ClusteringBound end, int i)
{
return 0 == start.accessor().compare(start.get(i), end.get(i), end.accessor());
}
private static RowFilter copy(RowFilter copy)
{
RowFilter newRowFilter = RowFilter.create(false);
for (RowFilter.Expression expression : copy)
newRowFilter.add(expression);
return newRowFilter;
}
private static void addExpressions(RowFilter rowFilter, List<ColumnMetadata> cms, int start, ByteBuffer[] values, Operator op, Operator lastOp)
{
for (int i = start ; i < values.length ; ++i)
rowFilter.add(cms.get(i), i + 1 == values.length ? lastOp : op, values[i]);
}
private static ClusteringIndexSliceFilter filter(TableMetadata metadata, ClusteringBound<?> start, ClusteringBound<?> end, boolean reversed)
{
return new ClusteringIndexSliceFilter(Slices.with(metadata.comparator, Slice.make(start, end)), reversed);
}
@Override
public void apply(PartitionUpdate update)
{
int nodeId = Int32Type.instance.compose(update.partitionKey().getKey());
InetAddressAndPort endpoint = ClusterMetadata.current().directory.endpoint(new NodeId(nodeId));
if (endpoint == null)
throw new InvalidRequestException("Unknown node " + nodeId);
DeletionInfo deletionInfo = update.deletionInfo();
if (!deletionInfo.getPartitionDeletion().isLive())
{
truncate(endpoint).syncThrowUncheckedOnInterrupt();
return;
}
int pkCount = local.partitionKeyColumns().size();
ByteBuffer[] pkBuffer, ckBuffer;
{
int ckCount = local.clusteringColumns().size();
pkBuffer = pkCount == 1 ? null : new ByteBuffer[pkCount];
ckBuffer = new ByteBuffer[ckCount];
}
PartitionUpdate.Builder builder = null;
ArrayDeque<Promise<Void>> results = new ArrayDeque<>();
if (deletionInfo.hasRanges())
{
Iterator<RangeTombstone> iterator = deletionInfo.rangeIterator(false);
while (iterator.hasNext())
{
RangeTombstone rt = iterator.next();
ClusteringBound start = rt.deletedSlice().start();
ClusteringBound end = rt.deletedSlice().end();
if (start.size() < pkCount || end.size() < pkCount)
throw new InvalidRequestException("Range deletions must specify a complete partition key in the underlying table " + metadata);
for (int i = 0 ; i < pkCount ; ++i)
{
if (0 != start.accessor().compare(start.get(i), end.get(i), end.accessor()))
throw new InvalidRequestException("Range deletions must specify a single partition key in the underlying table " + metadata);
}
DecoratedKey key = remoteClusteringToLocalPartitionKey(local, start, pkCount, pkBuffer);
builder = maybeRolloverAndWait(key, builder, results, endpoint);
if (start.size() == pkCount && end.size() == pkCount)
{
builder.addPartitionDeletion(rt.deletionTime());
}
else
{
start = ClusteringBound.create(start.kind(), Clustering.make(remoteClusteringToLocalClustering(start.clustering(), pkCount, ckBuffer)));
end = ClusteringBound.create(end.kind(), Clustering.make(remoteClusteringToLocalClustering(end.clustering(), pkCount, ckBuffer)));
builder.add(new RangeTombstone(Slice.make(start, end), rt.deletionTime()));
}
}
}
if (!update.staticRow().isEmpty())
throw new InvalidRequestException("Static rows are not supported for remote table " + metadata);
try (BTree.FastBuilder<ColumnData> columns = BTree.fastBuilder())
{
for (Row row : update)
{
Clustering<?> clustering = row.clustering();
DecoratedKey key = remoteClusteringToLocalPartitionKey(local, clustering, pkCount, pkBuffer);
builder = maybeRolloverAndWait(key, builder, results, endpoint);
Clustering<?> newClustering = Clustering.make(remoteClusteringToLocalClustering(clustering, pkCount, ckBuffer));
columns.reset();
for (ColumnData cd : row)
columns.add(rebind(local, cd));
builder.add(BTreeRow.create(newClustering, row.primaryKeyLivenessInfo(), row.deletion(), columns.build()));
}
}
if (builder != null)
results.add(send(Verb.VIRTUAL_MUTATION_REQ, new VirtualMutation(builder.build()), endpoint));
while (!results.isEmpty())
results.pollFirst().syncThrowUncheckedOnInterrupt();
}
private PartitionUpdate.Builder maybeRolloverAndWait(DecoratedKey key, PartitionUpdate.Builder builder, ArrayDeque<Promise<Void>> waiting, InetAddressAndPort endpoint)
{
if (builder == null || !builder.partitionKey().equals(key))
{
if (builder != null)
waiting.add(send(Verb.VIRTUAL_MUTATION_REQ, new VirtualMutation(builder.build()), endpoint));
builder = new PartitionUpdate.Builder(local, key, local.regularAndStaticColumns(), 8);
while (waiting.size() >= MAX_CONCURRENCY)
waiting.pollFirst().syncThrowUncheckedOnInterrupt();
}
return builder;
}
private Promise<Void> truncate(InetAddressAndPort endpoint)
{
return send(Verb.TRUNCATE_REQ, new TruncateRequest(local.keyspace, local.name), endpoint);
}
private static ColumnData rebind(TableMetadata local, ColumnData cd)
{
ColumnMetadata column = local.getColumn(cd.column().name);
Invariants.require(column != null, cd.column() + " not found in " + local);
Invariants.require(!column.isComplex(), "Complex column " + column + " not supported; should have been removed from metadata");
return ((Cell<?>) cd).withUpdatedColumn(column);
}
private static DecoratedKey remoteClusteringToLocalPartitionKey(TableMetadata local, ClusteringPrefix<?> clustering, int pkCount, ByteBuffer[] pkBuffer)
{
ByteBuffer bytes;
if (pkCount == 1) bytes = clustering.bufferAt(0);
else
{
for (int i = 0 ; i < pkBuffer.length ; ++i)
pkBuffer[i] = clustering.bufferAt(i);
bytes = CompositeType.build(ByteBufferAccessor.instance, pkBuffer);
}
return local.partitioner.decorateKey(bytes);
}
private static ByteBuffer[] remoteClusteringToLocalClustering(ClusteringPrefix<?> clustering, int pkCount, ByteBuffer[] ckBuffer)
{
for (int i = pkCount ; i < clustering.size(); ++i)
ckBuffer[i - pkCount] = clustering.bufferAt(i);
return Arrays.copyOf(ckBuffer, clustering.size() - pkCount);
}
}

38
src/java/org/apache/cassandra/db/virtual/VirtualMutation.java
View File

@ -17,6 +17,7 @@
*/
package org.apache.cassandra.db.virtual;
import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
@ -26,12 +27,20 @@ import javax.annotation.Nullable;
import com.google.common.base.MoreObjects;
import com.google.common.collect.ImmutableMap;
import accord.utils.Invariants;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.IMutation;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.ReadCommand.PotentialTxnConflicts;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.DeserializationHelper;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.IVerbHandler;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.net.NoPayload;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.ClientState;
@ -42,6 +51,35 @@ import org.apache.cassandra.service.ClientState;
*/
public final class VirtualMutation implements IMutation
{
public static final IVersionedSerializer<VirtualMutation> serializer = new IVersionedSerializer<VirtualMutation>()
{
@Override
public void serialize(VirtualMutation t, DataOutputPlus out, int version) throws IOException
{
Invariants.require(t.modifications.size() == 1);
PartitionUpdate.serializer.serialize(t.modifications.values().iterator().next(), out, version);
}
@Override
public VirtualMutation deserialize(DataInputPlus in, int version) throws IOException
{
PartitionUpdate update = PartitionUpdate.serializer.deserialize(in, version, DeserializationHelper.Flag.FROM_REMOTE);
return new VirtualMutation(update);
}
@Override
public long serializedSize(VirtualMutation t, int version)
{
Invariants.require(t.modifications.size() == 1);
return PartitionUpdate.serializer.serializedSize(t.modifications.values().iterator().next(), version);
}
};
public static final IVerbHandler<VirtualMutation> handler = message -> {
message.payload.apply();
MessagingService.instance().respond(NoPayload.noPayload, message);
};
private final String keyspaceName;
private final DecoratedKey partitionKey;
private final ImmutableMap<TableId, PartitionUpdate> modifications;

2
src/java/org/apache/cassandra/db/virtual/VirtualTable.java
View File

@ -100,4 +100,6 @@ public interface VirtualTable
{
return allowFilteringImplicitly();
}
default Sorted sorted() { return Sorted.UNSORTED; }
}

2
src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java
View File

@ -46,7 +46,7 @@ public class ExceptionSerializer
{
public static class RemoteException extends RuntimeException
{
private final String originalClass;
public final String originalClass;
public RemoteException(String originalClass, String originalMessage, StackTraceElement[] stackTrace)
{

3
src/java/org/apache/cassandra/exceptions/RequestFailure.java
View File

@ -31,6 +31,7 @@ import org.apache.cassandra.tcm.NotCMSException;
import static com.google.common.base.Preconditions.checkNotNull;
import static org.apache.cassandra.exceptions.ExceptionSerializer.nullableRemoteExceptionSerializer;
import static org.apache.cassandra.exceptions.RequestFailureReason.UNKNOWN;
/**
* Allow inclusion of a serialized exception in failure response messages
@ -125,7 +126,7 @@ public class RequestFailure
if (t instanceof CoordinatorBehindException)
return COORDINATOR_BEHIND;
return UNKNOWN;
return new RequestFailure(t);
}
public static RequestFailure forReason(RequestFailureReason reason)

29
src/java/org/apache/cassandra/locator/RemoteStrategy.java Normal file
View File

@ -0,0 +1,29 @@
/*
* 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.locator;
import java.util.Map;
public class RemoteStrategy extends LocalStrategy
{
public RemoteStrategy(String keyspaceName, Map<String, String> configOptions)
{
super(keyspaceName, configOptions);
}
}

3
src/java/org/apache/cassandra/net/Verb.java
View File

@ -44,6 +44,7 @@ import org.apache.cassandra.db.SnapshotCommand;
import org.apache.cassandra.db.TruncateRequest;
import org.apache.cassandra.db.TruncateResponse;
import org.apache.cassandra.db.TruncateVerbHandler;
import org.apache.cassandra.db.virtual.VirtualMutation;
import org.apache.cassandra.exceptions.RequestFailure;
import org.apache.cassandra.gms.GossipDigestAck;
import org.apache.cassandra.gms.GossipDigestAck2;
@ -195,6 +196,8 @@ public enum Verb
{
MUTATION_RSP (60, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
MUTATION_REQ (0, P3, writeTimeout, MUTATION, () -> Mutation.serializer, () -> MutationVerbHandler.instance, MUTATION_RSP ),
VIRTUAL_MUTATION_RSP (200, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
VIRTUAL_MUTATION_REQ (201, P3, writeTimeout, MUTATION, () -> VirtualMutation.serializer, () -> VirtualMutation.handler, VIRTUAL_MUTATION_RSP),
HINT_RSP (61, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),
HINT_REQ (1, P4, writeTimeout, MUTATION, () -> HintMessage.serializer, () -> HintVerbHandler.instance, HINT_RSP ),
READ_REPAIR_RSP (62, P1, writeTimeout, REQUEST_RESPONSE, () -> NoPayload.serializer, RESPONSE_HANDLER ),

3
src/java/org/apache/cassandra/schema/SchemaConstants.java
View File

@ -57,6 +57,7 @@ public final class SchemaConstants
public static final String VIRTUAL_VIEWS = "system_views";
public static final String VIRTUAL_METRICS = "system_metrics";
public static final String VIRTUAL_ACCORD_DEBUG = "system_accord_debug";
public static final String VIRTUAL_ACCORD_DEBUG_REMOTE = "system_accord_debug_remote";
public static final String DUMMY_KEYSPACE_OR_TABLE_NAME = "--dummy--";
@ -66,7 +67,7 @@ public final class SchemaConstants
/* virtual table system keyspace names */
public static final Set<String> VIRTUAL_SYSTEM_KEYSPACE_NAMES =
ImmutableSet.of(VIRTUAL_SCHEMA, VIRTUAL_VIEWS, VIRTUAL_METRICS);
ImmutableSet.of(VIRTUAL_SCHEMA, VIRTUAL_VIEWS, VIRTUAL_METRICS, VIRTUAL_ACCORD_DEBUG, VIRTUAL_ACCORD_DEBUG_REMOTE);
/* replicate system keyspace names (the ones with a "true" replication strategy) */
public static final Set<String> REPLICATED_SYSTEM_KEYSPACE_NAMES =

4
src/java/org/apache/cassandra/service/CassandraDaemon.java
View File

@ -61,6 +61,7 @@ import org.apache.cassandra.db.SystemKeyspaceMigrator41;
import org.apache.cassandra.db.commitlog.CommitLog;
import org.apache.cassandra.db.virtual.AccordDebugKeyspace;
import org.apache.cassandra.db.virtual.ExceptionsTable;
import org.apache.cassandra.db.virtual.AccordDebugRemoteKeyspace;
import org.apache.cassandra.db.virtual.LogMessagesTable;
import org.apache.cassandra.db.virtual.SlowQueriesTable;
import org.apache.cassandra.db.virtual.SystemViewsKeyspace;
@ -559,7 +560,10 @@ public class CassandraDaemon
VirtualKeyspaceRegistry.instance.register(new VirtualKeyspace(VIRTUAL_METRICS, createMetricsKeyspaceTables()));
if (DatabaseDescriptor.getAccord().enable_virtual_debug_only_keyspace)
{
VirtualKeyspaceRegistry.instance.register(AccordDebugKeyspace.instance);
VirtualKeyspaceRegistry.instance.register(AccordDebugRemoteKeyspace.instance);
}
// Flush log messages to system_views.system_logs virtual table as there were messages already logged
// before that virtual table was instantiated.

138
test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java
View File

@ -52,6 +52,8 @@ import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.exceptions.ExceptionSerializer;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
@ -88,54 +90,108 @@ public class AccordDebugKeyspaceTest extends CQLTester
private static final String QUERY_TXN_BLOCKED_BY =
String.format("SELECT * FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_BLOCKED_BY);
private static final String QUERY_TXN_BLOCKED_BY_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_BLOCKED_BY);
private static final String QUERY_COMMANDS_FOR_KEY =
String.format("SELECT txn_id, status FROM %s.%s WHERE key=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.COMMANDS_FOR_KEY);
private static final String QUERY_COMMANDS_FOR_KEY_REMOTE =
String.format("SELECT txn_id, status FROM %s.%s WHERE node_id = ? AND key=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.COMMANDS_FOR_KEY);
private static final String QUERY_TXN =
String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN);
private static final String QUERY_TXN_REMOTE =
String.format("SELECT txn_id, save_status FROM %s.%s WHERE node_id = ? AND txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN);
private static final String QUERY_TXNS =
String.format("SELECT save_status FROM %s.%s WHERE command_store_id = ? LIMIT 5", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN);
private static final String QUERY_TXNS_REMOTE =
String.format("SELECT save_status FROM %s.%s WHERE node_id = ? AND command_store_id = ? LIMIT 5", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN);
private static final String QUERY_TXNS_SEARCH =
String.format("SELECT save_status FROM %s.%s WHERE command_store_id = ? AND txn_id > ? LIMIT 5", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN);
private static final String QUERY_TXNS_SEARCH_REMOTE =
String.format("SELECT save_status FROM %s.%s WHERE node_id = ? AND command_store_id = ? AND txn_id > ? LIMIT 5", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN);
private static final String QUERY_JOURNAL =
String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);
private static final String ERASE_JOURNAL_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND command_store_id = ? AND txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.JOURNAL);
private static final String QUERY_JOURNAL_REMOTE =
String.format("SELECT txn_id, save_status FROM %s.%s WHERE node_id = ? AND txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.JOURNAL);
private static final String SET_TRACE =
String.format("UPDATE %s.%s SET permits = ? WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String SET_TRACE_REMOTE =
String.format("UPDATE %s.%s SET permits = ? WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_TRACE =
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_TRACE_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE1 =
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE1_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE2 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACE);
private static final String UNSET_TRACE2_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACE);
private static final String QUERY_TRACES =
String.format("SELECT * FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String QUERY_TRACES_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES1 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES1_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ? AND id_micros < ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES2 =
String.format("DELETE FROM %s.%s WHERE txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES2_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ? AND event_type = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES3 =
String.format("DELETE FROM %s.%s WHERE txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.TXN_TRACES);
private static final String ERASE_TRACES3_REMOTE =
String.format("DELETE FROM %s.%s WHERE node_id = ? AND txn_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.TXN_TRACES);
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_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ =
String.format("SELECT * FROM %s.%s WHERE quorum_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.REDUNDANT_BEFORE);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND quorum_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, 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);
private static final String QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ_REMOTE =
String.format("SELECT * FROM %s.%s WHERE node_id = ? AND shard_applied >= ?", SchemaConstants.VIRTUAL_ACCORD_DEBUG_REMOTE, AccordDebugKeyspace.REDUNDANT_BEFORE);
@BeforeClass
public static void setUpClass()
{
@ -170,43 +226,55 @@ public class AccordDebugKeyspaceTest extends CQLTester
public void tracing()
{
// simple test to confirm basic tracing functionality works, doesn't validate specific behaviours only requesting/querying/erasing
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
DatabaseDescriptor.getAccord().fetch_txn = "1s";
int nodeId = accord.nodeId().id;
AccordMsgFilter filter = new AccordMsgFilter();
MessagingService.instance().outboundSink.add(filter);
try
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
DatabaseDescriptor.getAccord().fetch_txn = "1s";
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
filter.appliesTo(id);
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(SET_TRACE, 0, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(UNSET_TRACE1, id.toString());
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
execute(UNSET_TRACE2, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"));
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"));
execute(SET_TRACE, 1, id.toString(), "WAIT_PROGRESS");
assertRows(execute(QUERY_TRACE, id.toString(), "WAIT_PROGRESS"), row(id.toString(), "WAIT_PROGRESS", 1));
accord.node().coordinate(id, txn).beginAsResult();
assertRows(execute(QUERY_TRACE_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS"), row(nodeId, id.toString(), "WAIT_PROGRESS", 1));
filter.appliesTo(id);
accord.node().coordinate(id, txn).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
filter.apply.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isGreaterThan(0));
spinUntilSuccess(() -> Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS").size()).isGreaterThan(0));
execute(ERASE_TRACES1, id.toString(), "FETCH", Long.MAX_VALUE);
execute(ERASE_TRACES2, id.toString(), "FETCH");
execute(ERASE_TRACES1, id.toString(), "WAIT_PROGRESS", Long.MAX_VALUE);
Assertions.assertThat(execute(QUERY_TRACES, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0);
Assertions.assertThat(execute(QUERY_TRACES_REMOTE, nodeId, id.toString(), "WAIT_PROGRESS").size()).isEqualTo(0);
// just check other variants don't fail
execute(ERASE_TRACES2, id.toString(), "WAIT_PROGRESS");
execute(ERASE_TRACES3, id.toString());
}
finally
{
@ -219,6 +287,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
var accord = accord();
int nodeId = accord.nodeId().id;
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());
@ -234,6 +303,27 @@ public class AccordDebugKeyspaceTest extends CQLTester
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_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);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_REMOTE, nodeId).size()).isGreaterThan(0);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ_REMOTE, nodeId, syncId1.toString()).size()).isEqualTo(2);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_QUORUM_APPLIED_GEQ_REMOTE, nodeId, syncId2.toString()).size()).isEqualTo(1);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ_REMOTE, nodeId, syncId1.toString()).size()).isEqualTo(1);
Assertions.assertThat(execute(QUERY_REDUNDANT_BEFORE_FILTER_SHARD_APPLIED_GEQ_REMOTE, nodeId, syncId2.toString()).size()).isEqualTo(0);
}
@Test
public void reportInvalidRequestForUnsupportedRemoteToLocal()
{
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
try
{
execute(ERASE_JOURNAL_REMOTE, nodeId, 1, id.toString());
}
catch (ExceptionSerializer.RemoteException t)
{
Assertions.assertThat(t.originalClass).isEqualTo(InvalidRequestException.class.getName());
}
}
@Test
@ -241,6 +331,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
String keyStr = txn.keys().get(0).toUnseekable().toString();
@ -249,15 +340,19 @@ public class AccordDebugKeyspaceTest extends CQLTester
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), anyInt(), 0, "", "", any(), anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied"));
assertRows(execute(QUERY_TXN_REMOTE, nodeId, id.toString()), row(id.toString(), "Applied"));
assertRows(execute(QUERY_JOURNAL, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_JOURNAL_REMOTE, nodeId, id.toString()), row(id.toString(), "PreAccepted"), row(id.toString(), "Applying"), row(id.toString(), "Applied"), row(id.toString(), null));
assertRows(execute(QUERY_COMMANDS_FOR_KEY, keyStr), row(id.toString(), "APPLIED_DURABLE"));
assertRows(execute(QUERY_COMMANDS_FOR_KEY_REMOTE, nodeId, keyStr), row(id.toString(), "APPLIED_DURABLE"));
}
@Test
public void manyTxns() throws ExecutionException, InterruptedException
public void manyTxns()
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
List<IAccordService.IAccordResult> await = new ArrayList<>();
Txn txn = createTxn(wrapInTxn(String.format("INSERT INTO %s.%s(k, c, v) VALUES (?, ?, ?)", KEYSPACE, tableName)), 0, 0, 0);
for (int i = 0 ; i < 100; ++i)
@ -281,6 +376,22 @@ public class AccordDebugKeyspaceTest extends CQLTester
row("Applied"),
row("Applied")
);
assertRows(execute(QUERY_TXNS_REMOTE, nodeId, commandStore.id()),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied")
);
assertRows(execute(QUERY_TXNS_SEARCH_REMOTE, nodeId, commandStore.id(), TxnId.NONE.toString()),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied")
);
}
@Test
@ -293,6 +404,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId id = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
@ -307,9 +419,13 @@ public class AccordDebugKeyspaceTest extends CQLTester
filter.preAccept.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), anyInt(), 0, "", "", any(), anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE, nodeId, id.toString()),
row(nodeId, id.toString(), anyInt(), 0, "", "", any(), anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
filter.apply.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), anyInt(), 0, "", "", any(), SaveStatus.ReadyToExecute.name()));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE, nodeId, id.toString()),
row(nodeId, id.toString(), anyInt(), 0, "", "", any(), SaveStatus.ReadyToExecute.name()));
}
finally
{
@ -329,6 +445,7 @@ public class AccordDebugKeyspaceTest extends CQLTester
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
int nodeId = accord.nodeId().id;
TxnId first = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
String insertTxn = String.format("BEGIN TRANSACTION\n" +
" LET r = (SELECT * FROM %s.%s WHERE k = ? AND c = ?);\n" +
@ -343,9 +460,13 @@ public class AccordDebugKeyspaceTest extends CQLTester
filter.preAccept.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), anyInt(), 0, "", any(), any(), anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE, nodeId, first.toString()),
row(nodeId, first.toString(), anyInt(), 0, "", any(), any(), anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
filter.apply.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), anyInt(), 0, "", any(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE, nodeId, first.toString()),
row(nodeId, first.toString(), anyInt(), 0, "", any(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
filter.reset();
@ -366,6 +487,11 @@ public class AccordDebugKeyspaceTest extends CQLTester
row(second.toString(), anyInt(), 1, any(), first.toString(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
assertRows(execute(QUERY_TXN_BLOCKED_BY + " AND depth < 1", second.toString()),
row(second.toString(), anyInt(), 0, any(), "", anyNonNull(), SaveStatus.Stable.name()));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE, nodeId, second.toString()),
row(nodeId, second.toString(), anyInt(), 0, "", "", anyNonNull(), SaveStatus.Stable.name()),
row(nodeId, second.toString(), anyInt(), 1, any(), first.toString(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
assertRows(execute(QUERY_TXN_BLOCKED_BY_REMOTE + " AND depth < 1", nodeId, second.toString()),
row(nodeId, second.toString(), anyInt(), 0, any(), "", anyNonNull(), SaveStatus.Stable.name()));
}
finally
{