diff --git a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java index bc17dd0158..86d66c59ff 100644 --- a/src/java/org/apache/cassandra/db/filter/ColumnFilter.java +++ b/src/java/org/apache/cassandra/db/filter/ColumnFilter.java @@ -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 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 subSelections = null; + if (this.subSelections != null) + { + subSelections = TreeMultimap.create(); + for (Map.Entry e : this.subSelections.entries()) + subSelections.put(e.getKey(), e.getValue().rebind(newTable)); + } + return new SelectionColumnFilter(fetchingStrategy, queried, fetched, subSelections); + } + @Override protected SortedSetMultimap 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 builder = BTree.fastBuilder()) + { + for (ColumnMetadata in : columns) + { + ColumnMetadata out = newTable.getColumn(in.name); + if (out != null) + builder.add(out); + } + return Columns.from(builder); + } + } } diff --git a/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java b/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java index 459b636a5a..b4f0346c76 100644 --- a/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java +++ b/src/java/org/apache/cassandra/db/filter/ColumnSubselection.java @@ -76,6 +76,8 @@ public abstract class ColumnSubselection implements Comparable cmp = column.cellPathComparator(); @@ -160,6 +168,12 @@ public abstract class ColumnSubselection implements Comparable 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 "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 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 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 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) diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugRemoteKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugRemoteKeyspace.java new file mode 100644 index 0000000000..8698b3ccd8 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugRemoteKeyspace.java @@ -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); + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualKeyspace.java b/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualKeyspace.java new file mode 100644 index 0000000000..171f494036 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualKeyspace.java @@ -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 include) + { + super(name, wrap.tables().stream().filter(include).map(vt -> new RemoteToLocalVirtualTable(name, vt)).collect(Collectors.toList())); + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualTable.java b/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualTable.java new file mode 100644 index 0000000000..837c1feeb1 --- /dev/null +++ b/src/java/org/apache/cassandra/db/virtual/RemoteToLocalVirtualTable.java @@ -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 matchingIds = cm.directory.peerIds(); + DataRange dataRange = collector.dataRange(); + AbstractBounds 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 pksToCks = partitionKeyToClusterings(metadata, local); + ArrayDeque 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 = 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 + { + 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 Promise send(Verb verb, Object payload, InetAddressAndPort endpoint) + { + Promise promise = new AsyncPromise<>(); + send(verb, payload, promise, endpoint); + return promise; + } + + private void send(Verb verb, Object payload, Promise 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() + { + @Override public void onResponse(Message 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 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 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 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 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 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 result = new ArrayList<>(3); + if (startSlice != BOTTOM) + { + AbstractBounds 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 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 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> results = new ArrayDeque<>(); + + if (deletionInfo.hasRanges()) + { + Iterator 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 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> 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 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); + } +} diff --git a/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java b/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java index ee98da29c1..41e9e96266 100644 --- a/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java +++ b/src/java/org/apache/cassandra/db/virtual/VirtualMutation.java @@ -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 serializer = new IVersionedSerializer() + { + @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 handler = message -> { + message.payload.apply(); + MessagingService.instance().respond(NoPayload.noPayload, message); + }; + private final String keyspaceName; private final DecoratedKey partitionKey; private final ImmutableMap modifications; diff --git a/src/java/org/apache/cassandra/db/virtual/VirtualTable.java b/src/java/org/apache/cassandra/db/virtual/VirtualTable.java index 3f05ed3a69..0423b12a4d 100644 --- a/src/java/org/apache/cassandra/db/virtual/VirtualTable.java +++ b/src/java/org/apache/cassandra/db/virtual/VirtualTable.java @@ -100,4 +100,6 @@ public interface VirtualTable { return allowFilteringImplicitly(); } + + default Sorted sorted() { return Sorted.UNSORTED; } } diff --git a/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java b/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java index de379739a3..834abca49d 100644 --- a/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java +++ b/src/java/org/apache/cassandra/exceptions/ExceptionSerializer.java @@ -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) { diff --git a/src/java/org/apache/cassandra/exceptions/RequestFailure.java b/src/java/org/apache/cassandra/exceptions/RequestFailure.java index b9bba7fc70..03d088a239 100644 --- a/src/java/org/apache/cassandra/exceptions/RequestFailure.java +++ b/src/java/org/apache/cassandra/exceptions/RequestFailure.java @@ -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) diff --git a/src/java/org/apache/cassandra/locator/RemoteStrategy.java b/src/java/org/apache/cassandra/locator/RemoteStrategy.java new file mode 100644 index 0000000000..515462ee98 --- /dev/null +++ b/src/java/org/apache/cassandra/locator/RemoteStrategy.java @@ -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 configOptions) + { + super(keyspaceName, configOptions); + } +} diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index d24c9e64ad..4a0ebb2c22 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -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 ), diff --git a/src/java/org/apache/cassandra/schema/SchemaConstants.java b/src/java/org/apache/cassandra/schema/SchemaConstants.java index 6e3cb7a72e..ada413537d 100644 --- a/src/java/org/apache/cassandra/schema/SchemaConstants.java +++ b/src/java/org/apache/cassandra/schema/SchemaConstants.java @@ -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 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 REPLICATED_SYSTEM_KEYSPACE_NAMES = diff --git a/src/java/org/apache/cassandra/service/CassandraDaemon.java b/src/java/org/apache/cassandra/service/CassandraDaemon.java index f4ed67a440..2f85e6e58e 100644 --- a/src/java/org/apache/cassandra/service/CassandraDaemon.java +++ b/src/java/org/apache/cassandra/service/CassandraDaemon.java @@ -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. diff --git a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java index 6cac2f25fc..0889ebffe5 100644 --- a/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java +++ b/test/unit/org/apache/cassandra/db/virtual/AccordDebugKeyspaceTest.java @@ -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 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 {