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

Lazy Virtual Tables 

Also Improve:
 - Searchable system_accord_debug.txn
 - Integrate txn_blocked_by deadline and depth filter with execution logic, to ensure we terminate promptly and get a best effort reply

patch by Benedict; reviewed by Aleksey Yeschenko for CASSANDRA-20899
This commit is contained in:
Benedict Elliott Smith 2025-09-08 14:42:32 +01:00
parent 3d2b0efc34
commit f7743fe3ab
28 changed files with 1855 additions and 1054 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
View File

@ -156,13 +156,13 @@ public final class StatementRestrictions
return new StatementRestrictions(type, table, IndexHints.NONE, false);
}
private StatementRestrictions(StatementType type, TableMetadata table, IndexHints indexHints, boolean allowFiltering)
private StatementRestrictions(StatementType type, TableMetadata table, IndexHints indexHints, boolean allowFilteringOfPrimaryKeys)
{
this.type = type;
this.table = table;
this.indexHints = indexHints;
this.partitionKeyRestrictions = new PartitionKeyRestrictions(table.partitionKeyAsClusteringComparator());
this.clusteringColumnsRestrictions = new ClusteringColumnRestrictions(table, allowFiltering);
this.clusteringColumnsRestrictions = new ClusteringColumnRestrictions(table, allowFilteringOfPrimaryKeys);
this.nonPrimaryKeyRestrictions = RestrictionSet.empty();
this.notNullColumns = new HashSet<>();
}
@ -370,7 +370,7 @@ public final class StatementRestrictions
}
else
{
if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table))
if (!allowFiltering && requiresAllowFilteringIfNotSpecified(table, false))
throw invalidRequest(allowFilteringMessage(state));
}
@ -381,14 +381,14 @@ public final class StatementRestrictions
validateSecondaryIndexSelections();
}
public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata)
public static boolean requiresAllowFilteringIfNotSpecified(TableMetadata metadata, boolean isPrimaryKey)
{
if (!metadata.isVirtual())
return true;
VirtualTable tableNullable = VirtualKeyspaceRegistry.instance.getTableNullable(metadata.id);
assert tableNullable != null;
return !tableNullable.allowFilteringImplicitly();
return isPrimaryKey ? !tableNullable.allowFilteringPrimaryKeysImplicitly() : !tableNullable.allowFilteringImplicitly();
}
private void addRestriction(Restriction restriction, IndexRegistry indexRegistry, IndexHints indexHints)
@ -593,7 +593,7 @@ public final class StatementRestrictions
// components must have a EQ. Only the last partition key component can be in IN relation.
if (partitionKeyRestrictions.needFiltering())
{
if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified(table))
if (!allowFiltering && !forView && !hasQueriableIndex && requiresAllowFilteringIfNotSpecified(table, true))
throw new InvalidRequestException(allowFilteringMessage(state));
isKeyRange = true;

4
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -1474,7 +1474,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
boundNames,
orderings,
selectsOnlyStaticColumns,
parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata),
parameters.allowFiltering || !requiresAllowFilteringIfNotSpecified(metadata, true),
forView);
}
@ -1700,7 +1700,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
{
// We will potentially filter data if the row filter is not the identity and there isn't any index group
// supporting all the expressions in the filter.
if (requiresAllowFilteringIfNotSpecified(table))
if (requiresAllowFilteringIfNotSpecified(table, true))
checkFalse(restrictions.needFiltering(table), StatementRestrictions.REQUIRES_ALLOW_FILTERING_MESSAGE);
}
}

2
src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
View File

@ -647,7 +647,7 @@ public class PartitionRangeReadCommand extends ReadCommand implements PartitionR
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter(), rowFilter());
UnfilteredPartitionIterator resultIterator = view.select(dataRange, columnFilter(), rowFilter(), limits());
return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
}

2
src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
View File

@ -1516,7 +1516,7 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
{
VirtualTable view = VirtualKeyspaceRegistry.instance.getTableNullable(metadata().id);
UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter(), rowFilter());
UnfilteredPartitionIterator resultIterator = view.select(partitionKey, clusteringIndexFilter, columnFilter(), rowFilter(), limits());
return limits().filter(rowFilter().filter(resultIterator, nowInSec()), nowInSec(), selectsFullPartition());
}

8
src/java/org/apache/cassandra/db/marshal/TxnIdUtf8Type.java
View File

@ -36,7 +36,7 @@ public class TxnIdUtf8Type extends PseudoUtf8Type
{
super.validate(value, accessor);
String str = deserialize(value, accessor);
if (null == TxnId.tryParse(str))
if (!str.isEmpty() && null == TxnId.tryParse(str))
throw new MarshalException("Invalid TxnId: " + str);
}
};
@ -59,6 +59,12 @@ public class TxnIdUtf8Type extends PseudoUtf8Type
{
String leftStr = UTF8Serializer.instance.deserialize(left, accessorL);
String rightStr = UTF8Serializer.instance.deserialize(right, accessorR);
if (leftStr.isEmpty() || rightStr.isEmpty())
{
if (leftStr.isEmpty() && rightStr.isEmpty())
return 0;
return leftStr.isEmpty() ? -1 : 1;
}
TxnId leftId = TxnId.parse(leftStr);
TxnId rightId = TxnId.parse(rightStr);
return leftId.compareTo(rightId);

767
src/java/org/apache/cassandra/db/virtual/AbstractLazyVirtualTable.java Normal file
View File

@ -0,0 +1,767 @@
/*
* 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.Arrays;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.UnaryOperator;
import javax.annotation.Nullable;
import accord.utils.Invariants;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Clusterable;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ClusteringPrefix;
import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.LivenessInfo;
import org.apache.cassandra.db.RegularAndStaticColumns;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
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.AbstractType;
import org.apache.cassandra.db.marshal.CompositeType;
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.BufferCell;
import org.apache.cassandra.db.rows.ColumnData;
import org.apache.cassandra.db.rows.EncodingStats;
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.Bounds;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.utils.BulkIterator;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.UpdateFunction;
import static org.apache.cassandra.db.ClusteringPrefix.Kind.STATIC_CLUSTERING;
import static org.apache.cassandra.db.ConsistencyLevel.ONE;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
/**
* An abstract virtual table implementation that builds the resultset on demand.
*/
public abstract class AbstractLazyVirtualTable implements VirtualTable
{
public enum OnTimeout { BEST_EFFORT, FAIL }
// in the special case where we know we have enough rows in the collector, throw this exception to terminate early
public static class InternalDoneException extends RuntimeException {}
// in the special case where we have timed out, throw this exception to terminate early
public static class InternalTimeoutException extends RuntimeException {}
public static class FilterRange<V>
{
final V min, max;
public FilterRange(V min, V max)
{
this.min = min;
this.max = max;
}
}
public interface PartitionsCollector
{
DataRange dataRange();
RowFilter rowFilter();
ColumnFilter columnFilter();
DataLimits limits();
long nowInSeconds();
long timestampMicros();
long deadlineNanos();
boolean isEmpty();
RowCollector row(Object... primaryKeys);
PartitionCollector partition(Object... partitionKeys);
UnfilteredPartitionIterator finish();
@Nullable Object[] singlePartitionKey();
<I, O> FilterRange<O> filters(String column, Function<I, O> translate, UnaryOperator<O> exclusiveStart, UnaryOperator<O> exclusiveEnd);
}
public interface PartitionCollector
{
void collect(Consumer<RowsCollector> addTo);
}
public interface RowsCollector
{
RowCollector add(Object... clusteringKeys);
}
public interface RowCollector
{
default void lazyCollect(Consumer<ColumnsCollector> addToIfNeeded) { eagerCollect(addToIfNeeded); }
void eagerCollect(Consumer<ColumnsCollector> addToNow);
}
public interface ColumnsCollector
{
/**
* equivalent to
* {@code
* if (value == null) add(columnName, null);
* else if (f1.apply(value) == null) add(columnName, f1.apply(value));
* else add(columnName, f2.apply(f1.apply(value)));
* }
*/
<V1, V2> ColumnsCollector add(String columnName, V1 value, Function<? super V1, ? extends V2> f1, Function<? super V2, ?> f2);
default <V> ColumnsCollector add(String columnName, V value, Function<? super V, ?> transform)
{
return add(columnName, value, Function.identity(), transform);
}
default ColumnsCollector add(String columnName, Object value)
{
return add(columnName, value, Function.identity());
}
}
public static class SimplePartitionsCollector implements PartitionsCollector
{
final TableMetadata metadata;
final boolean isSorted;
final boolean isSortedByPartitionKey;
final Map<String, ColumnMetadata> columnLookup = new HashMap<>();
final NavigableMap<DecoratedKey, SimplePartition> partitions;
final DataRange dataRange;
final ColumnFilter columnFilter;
final RowFilter rowFilter;
final DataLimits limits;
final long startedAtNanos = Clock.Global.nanoTime();
final long deadlineNanos;
final long nowInSeconds = Clock.Global.nowInSeconds();
final long timestampMicros;
int totalRowCount;
int lastFilteredTotalRowCount, lastFilteredPartitionCount;
@Override public DataRange dataRange() { return dataRange; }
@Override public RowFilter rowFilter() { return rowFilter; }
@Override public ColumnFilter columnFilter() { return columnFilter; }
@Override public DataLimits limits() { return limits; }
@Override public long nowInSeconds() { return nowInSeconds; }
@Override public long timestampMicros() { return timestampMicros; }
@Override public long deadlineNanos() { return deadlineNanos; }
@Override public boolean isEmpty() { return totalRowCount == 0; }
public SimplePartitionsCollector(TableMetadata metadata, boolean isSorted, boolean isSortedByPartitionKey,
DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
this.metadata = metadata;
this.isSorted = isSorted;
this.isSortedByPartitionKey = isSortedByPartitionKey;
this.dataRange = dataRange;
this.columnFilter = columnFilter;
this.rowFilter = rowFilter;
this.limits = limits;
this.timestampMicros = FBUtilities.timestampMicros();
this.deadlineNanos = startedAtNanos + DatabaseDescriptor.getReadRpcTimeout(TimeUnit.NANOSECONDS);
this.partitions = new TreeMap<>(dataRange.isReversed() ? DecoratedKey.comparator.reversed() : DecoratedKey.comparator);
for (ColumnMetadata cm : metadata.columns())
columnLookup.put(cm.name.toString(), cm);
}
public Object[] singlePartitionKey()
{
AbstractBounds<?> bounds = dataRange().keyRange();
if (!bounds.isStartInclusive() || !bounds.isEndInclusive() || !bounds.left.equals(bounds.right) || !(bounds.left instanceof DecoratedKey))
return null;
return composePartitionKeys((DecoratedKey) bounds.left, metadata);
}
@Override
public PartitionCollector partition(Object ... partitionKeys)
{
int pkSize = metadata.partitionKeyColumns().size();
if (pkSize != partitionKeys.length)
throw new IllegalArgumentException();
DecoratedKey partitionKey = decomposePartitionKeys(metadata, partitionKeys);
if (!dataRange.contains(partitionKey))
return dropCks -> {};
return partitions.computeIfAbsent(partitionKey, SimplePartition::new);
}
@Override
public UnfilteredPartitionIterator finish()
{
final Iterator<SimplePartition> partitions = this.partitions.values().iterator();
return new UnfilteredPartitionIterator()
{
@Override public TableMetadata metadata() { return metadata; }
@Override public void close() {}
@Override
public boolean hasNext()
{
return partitions.hasNext();
}
@Override
public UnfilteredRowIterator next()
{
SimplePartition partition = partitions.next();
Iterator<Row> rows = partition.rows();
return new UnfilteredRowIterator()
{
@Override public TableMetadata metadata() { return metadata; }
@Override public boolean isReverseOrder() { return dataRange.isReversed(); }
@Override public RegularAndStaticColumns columns() { return columnFilter.fetchedColumns(); }
@Override public DecoratedKey partitionKey() { return partition.key; }
@Override public Row staticRow() { return partition.staticRow(); }
@Override public boolean hasNext() { return rows.hasNext(); }
@Override public Unfiltered next() { return rows.next(); }
@Override public void close() {}
@Override public DeletionTime partitionLevelDeletion() { return DeletionTime.LIVE; }
@Override public EncodingStats stats() { return EncodingStats.NO_STATS; }
};
}
};
}
@Override
@Nullable
public <I, O> FilterRange<O> filters(String columnName, Function<I, O> translate, UnaryOperator<O> exclusiveStart, UnaryOperator<O> exclusiveEnd)
{
ColumnMetadata column = columnLookup.get(columnName);
O min = null, max = null;
for (RowFilter.Expression expression : rowFilter().getExpressions())
{
if (!expression.column().equals(column))
continue;
O bound = translate.apply((I)column.type.compose(expression.getIndexValue()));
switch (expression.operator())
{
default: throw new InvalidRequestException("Operator " + expression.operator() + " not supported for txn_id");
case EQ: min = max = bound; break;
case LTE: max = bound; break;
case LT: max = exclusiveEnd.apply(bound); break;
case GTE: min = bound; break;
case GT: min = exclusiveStart.apply(bound); break;
}
}
return new FilterRange<>(min, max);
}
@Override
public RowCollector row(Object... primaryKeys)
{
int pkSize = metadata.partitionKeyColumns().size();
int ckSize = metadata.clusteringColumns().size();
if (pkSize + ckSize != primaryKeys.length)
throw new IllegalArgumentException();
Object[] partitionKeyValues = new Object[pkSize];
Object[] clusteringValues = new Object[ckSize];
System.arraycopy(primaryKeys, 0, partitionKeyValues, 0, pkSize);
System.arraycopy(primaryKeys, pkSize, clusteringValues, 0, ckSize);
DecoratedKey partitionKey = decomposePartitionKeys(metadata, partitionKeyValues);
Clustering<?> clustering = decomposeClusterings(metadata, clusteringValues);
if (!dataRange.contains(partitionKey) || !dataRange.clusteringIndexFilter(partitionKey).selects(clustering))
return drop -> {};
return partitions.computeIfAbsent(partitionKey, SimplePartition::new).row(clustering);
}
private final class SimplePartition implements PartitionCollector, RowsCollector
{
private final DecoratedKey key;
// we assume no duplicate rows, and impose the condition lazily
private SimpleRow[] rows;
private int rowCount;
private SimpleRow staticRow;
private boolean dropRows;
private SimplePartition(DecoratedKey key)
{
this.key = key;
this.rows = new SimpleRow[1];
}
@Override
public void collect(Consumer<RowsCollector> addTo)
{
addTo.accept(this);
}
@Override
public RowCollector add(Object... clusteringKeys)
{
int ckSize = metadata.clusteringColumns().size();
if (ckSize != clusteringKeys.length)
throw new IllegalArgumentException();
return row(decomposeClusterings(metadata, clusteringKeys));
}
RowCollector row(Clustering<?> clustering)
{
if (nanoTime() > deadlineNanos)
throw new InternalTimeoutException();
if (dropRows || !dataRange.clusteringIndexFilter(key).selects(clustering))
return drop -> {};
if (totalRowCount >= limits.count())
{
boolean filter;
if (!isSortedByPartitionKey || lastFilteredPartitionCount == partitions.size())
{
filter = totalRowCount / 2 >= Math.max(1024, limits.count());
}
else
{
int rowsAddedSinceLastFiltered = totalRowCount - lastFilteredTotalRowCount;
int threshold = Math.max(32, Math.min(1024, lastFilteredTotalRowCount / 2));
filter = lastFilteredTotalRowCount == 0 || rowsAddedSinceLastFiltered >= threshold;
}
if (filter)
{
// first filter within each partition
for (SimplePartition partition : partitions.values())
partition.truncate(limits.perPartitionCount());
// then drop any partitions that completely fall outside our limit
Iterator<SimplePartition> iter = partitions.descendingMap().values().iterator();
SimplePartition last;
while (true)
{
SimplePartition next = last = iter.next();
if (totalRowCount - next.rowCount < limits.count())
break;
iter.remove();
totalRowCount -= next.rowCount;
if (next == this)
dropRows = true;
}
// possibly truncate the last partition if it partially falls outside the limit
int overflow = Math.max(0, totalRowCount - limits.count());
int newCount = last.truncate(last.rowCount - overflow);
lastFilteredTotalRowCount = totalRowCount;
lastFilteredPartitionCount = partitions.size();
if (isSortedByPartitionKey && totalRowCount - newCount >= limits.count())
throw new InternalDoneException();
if (isSorted && totalRowCount >= limits.count())
throw new InternalDoneException();
if (dropRows)
return drop -> {};
}
}
SimpleRow result = new SimpleRow(clustering);
if (clustering.kind() == STATIC_CLUSTERING)
{
Invariants.require(staticRow == null);
staticRow = result;
}
else
{
totalRowCount++;
if (rowCount == rows.length)
rows = Arrays.copyOf(rows, Math.max(8, rowCount * 2));
rows[rowCount++] = result;
}
return result;
}
void filterAndSort()
{
int newCount = 0;
for (int i = 0 ; i < rowCount; ++i)
{
if (rows[i].rowFilterIncludes())
{
if (newCount != i)
rows[newCount] = rows[i];
newCount++;
}
}
if (newCount != rowCount)
{
Arrays.fill(rows, newCount, rowCount, null);
totalRowCount -= (rowCount - newCount);
rowCount = newCount;
}
Arrays.sort(rows, 0, newCount, rowComparator());
}
int truncate(int newCount)
{
if (rowCount <= newCount)
return rowCount;
filterAndSort();
if (rowCount <= newCount)
return rowCount;
Arrays.fill(rows, newCount, rowCount, null);
totalRowCount -= (rowCount - newCount);
rowCount = newCount;
return newCount;
}
private Comparator<SimpleRow> rowComparator()
{
Comparator<Clusterable> cmp = dataRange.isReversed() ? metadata.comparator.reversed() : metadata.comparator;
return (a, b) -> cmp.compare(a.clustering, b.clustering);
}
Row staticRow()
{
if (staticRow == null)
return null;
return staticRow.materialiseAndFilter();
}
Iterator<Row> rows()
{
filterAndSort();
return Arrays.stream(rows, 0, rowCount).map(SimpleRow::materialiseAndFilter).iterator();
}
private final class SimpleRow implements RowCollector
{
final Clustering<?> clustering;
SomeColumns state;
private SimpleRow(Clustering<?> clustering)
{
this.clustering = clustering;
}
@Override
public void lazyCollect(Consumer<ColumnsCollector> addToIfNeeded)
{
Invariants.require(state == null);
state = new LazyColumnsCollector(addToIfNeeded);
}
@Override
public void eagerCollect(Consumer<ColumnsCollector> addToNow)
{
Invariants.require(state == null);
state = new EagerColumnsCollector(addToNow);
}
boolean rowFilterIncludes()
{
return null != materialiseAndFilter();
}
Row materialiseAndFilter()
{
if (state == null)
return null;
FilteredRow filtered = state.materialiseAndFilter(this);
state = filtered;
return filtered == null ? null : filtered.row;
}
DecoratedKey partitionKey()
{
return SimplePartition.this.key;
}
SimplePartitionsCollector collector()
{
return SimplePartitionsCollector.this;
}
}
}
static abstract class SomeColumns
{
abstract FilteredRow materialiseAndFilter(SimplePartition.SimpleRow parent);
}
static class LazyColumnsCollector extends SomeColumns
{
final Consumer<ColumnsCollector> lazy;
LazyColumnsCollector(Consumer<ColumnsCollector> lazy)
{
this.lazy = lazy;
}
@Override
FilteredRow materialiseAndFilter(SimplePartition.SimpleRow parent)
{
return parent.collector().new EagerColumnsCollector(lazy).materialiseAndFilter(parent);
}
}
class EagerColumnsCollector extends SomeColumns implements ColumnsCollector
{
Object[] columns = new Object[4];
int columnCount;
public EagerColumnsCollector(Consumer<ColumnsCollector> add)
{
add.accept(this);
}
@Override
public <V1, V2> ColumnsCollector add(String name, V1 v1, Function<? super V1, ? extends V2> f1, Function<? super V2, ?> f2)
{
if (v1 == null)
return this;
ColumnMetadata cm = columnLookup.get(name);
if (cm == null)
throw new IllegalArgumentException("Unknown column name " + name);
if (!columnFilter.fetches(cm))
return this;
V2 v2 = f1.apply(v1);
if (v2 == null)
return this;
Object result = f2.apply(v2);
if (result == null)
return this;
if (columnCount * 2 == columns.length)
columns = Arrays.copyOf(columns, columnCount * 4);
columns[columnCount * 2] = cm;
columns[columnCount * 2 + 1] = result;
++columnCount;
return this;
}
@Override
FilteredRow materialiseAndFilter(SimplePartition.SimpleRow parent)
{
for (int i = 0 ; i < columnCount ; i++)
{
ColumnMetadata cm = (ColumnMetadata) columns[i * 2];
Object value = columns[i * 2 + 1];
ByteBuffer bb = value instanceof ByteBuffer ? (ByteBuffer)value : decompose(cm.type, value);
columns[i] = BufferCell.live(cm, timestampMicros, bb);
}
Arrays.sort(columns, 0, columnCount, (a, b) -> ColumnData.comparator.compare((BufferCell)a, (BufferCell)b));
Object[] btree = BTree.build(BulkIterator.of(columns), columnCount, UpdateFunction.noOp);
BTreeRow row = BTreeRow.create(parent.clustering, LivenessInfo.create(timestampMicros, nowInSeconds), Row.Deletion.LIVE, btree);
if (!rowFilter.isSatisfiedBy(metadata, parent.partitionKey(), row, nowInSeconds))
return null;
return new FilteredRow(row);
}
}
static class FilteredRow extends SomeColumns
{
final Row row;
FilteredRow(Row row)
{
this.row = row;
}
@Override
FilteredRow materialiseAndFilter(SimplePartition.SimpleRow parent)
{
return this;
}
}
}
protected final TableMetadata metadata;
private final OnTimeout onTimeout;
private final Sorted sorted, sortedByPartitionKey;
protected AbstractLazyVirtualTable(TableMetadata metadata, OnTimeout onTimeout, Sorted sorted)
{
this(metadata, onTimeout, sorted, sorted);
}
protected AbstractLazyVirtualTable(TableMetadata metadata, OnTimeout onTimeout, Sorted sorted, Sorted sortedByPartitionKey)
{
if (!metadata.isVirtual())
throw new IllegalArgumentException("Cannot instantiate a non-virtual table");
if (!metadata.keyspace.startsWith(SchemaConstants.ACCORD_KEYSPACE_NAME))
{
// NOTE: there is nothing stopping other use cases from using this facility, but there was
// feedback on the ticket questioning the reliance on Accord integration tests for validating the API.
// If another use case wishes to use the facility, simply satisfy reviewers in this regard. See PR #4373 for details.
throw new IllegalArgumentException("This facility is only currently supported by Accord keyspaces");
}
this.metadata = metadata;
this.onTimeout = onTimeout;
this.sorted = sorted;
this.sortedByPartitionKey = sortedByPartitionKey;
}
@Override
public TableMetadata metadata()
{
return metadata;
}
public OnTimeout onTimeout() { return onTimeout; }
protected PartitionsCollector collector(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
boolean isSorted = isSorted(sorted, !dataRange.isReversed());
boolean isSortedByPartitionKey = isSorted || isSorted(sortedByPartitionKey, !dataRange.isReversed());
return new SimplePartitionsCollector(metadata, isSorted, isSortedByPartitionKey, dataRange, columnFilter, rowFilter, limits);
}
private static boolean isSorted(Sorted sorted, boolean asc)
{
return sorted == Sorted.SORTED || sorted == (asc ? Sorted.ASC : Sorted.DESC);
}
protected abstract void collect(PartitionsCollector collector);
@Override
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
return select(new DataRange(new Bounds<>(partitionKey, partitionKey), clusteringIndexFilter), columnFilter, rowFilter, limits);
}
@Override
public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
PartitionsCollector collector = collector(dataRange, columnFilter, rowFilter, limits);
try
{
collect(collector);
}
catch (InternalDoneException ignore) {}
catch (InternalTimeoutException ignore)
{
if (onTimeout != OnTimeout.BEST_EFFORT || collector.isEmpty())
throw new ReadTimeoutException(ONE, 0, 1, false);
ClientWarn.instance.warn("Ran out of time. Returning best effort.");
}
return collector.finish();
}
@Override
public void apply(PartitionUpdate update)
{
throw new InvalidRequestException("Modification is not supported by table " + metadata);
}
@Override
public void truncate()
{
throw new InvalidRequestException("Truncation is not supported by table " + metadata);
}
@Override
public String toString()
{
return metadata().toString();
}
static Object[] composePartitionKeys(DecoratedKey decoratedKey, TableMetadata metadata)
{
if (metadata.partitionKeyColumns().size() == 1)
return new Object[] { metadata.partitionKeyType.compose(decoratedKey.getKey()) };
ByteBuffer[] split = ((CompositeType)metadata.partitionKeyType).split(decoratedKey.getKey());
Object[] result = new Object[split.length];
for (int i = 0 ; i < split.length ; ++i)
result[i] = metadata.partitionKeyColumns().get(i).type.compose(split[i]);
return result;
}
static Object[] composeClusterings(ClusteringPrefix clustering, TableMetadata metadata)
{
Object[] result = new Object[clustering.size()];
for (int i = 0 ; i < result.length ; ++i)
result[i] = metadata.clusteringColumns().get(i).type.compose(clustering.get(i), clustering.accessor());
return result;
}
private static ByteBuffer decompose(AbstractType<?> type, Object value)
{
return type.decomposeUntyped(value);
}
static DecoratedKey decomposePartitionKeys(TableMetadata metadata, Object... partitionKeys)
{
ByteBuffer partitionKey = partitionKeys.length == 1
? decompose(metadata.partitionKeyType, partitionKeys[0])
: ((CompositeType) metadata.partitionKeyType).decompose(partitionKeys);
return metadata.partitioner.decorateKey(partitionKey);
}
static Clustering<?> decomposeClusterings(TableMetadata metadata, Object... clusteringKeys)
{
if (clusteringKeys.length == 0)
return Clustering.EMPTY;
ByteBuffer[] clusteringByteBuffers = new ByteBuffer[clusteringKeys.length];
for (int i = 0; i < clusteringKeys.length; i++)
{
if (clusteringKeys[i] instanceof ByteBuffer) clusteringByteBuffers[i] = (ByteBuffer) clusteringKeys[i];
else clusteringByteBuffers[i] = decompose(metadata.clusteringColumns().get(i).type, clusteringKeys[i]);
}
return Clustering.make(clusteringByteBuffers);
}
}

137
src/java/org/apache/cassandra/db/virtual/AbstractMutableLazyVirtualTable.java Normal file
View File

@ -0,0 +1,137 @@
/*
* 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.Iterator;
import javax.annotation.Nullable;
import accord.utils.Invariants;
import org.apache.cassandra.db.DeletionInfo;
import org.apache.cassandra.db.RangeTombstone;
import org.apache.cassandra.db.Slice;
import org.apache.cassandra.db.partitions.PartitionUpdate;
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.exceptions.InvalidRequestException;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
import static org.apache.cassandra.db.ClusteringPrefix.Kind.STATIC_CLUSTERING;
/**
* An abstract virtual table implementation that builds the resultset on demand and allows fine-grained source
* modification via INSERT/UPDATE, DELETE and TRUNCATE operations.
*
* Virtual table implementation need to be thread-safe has they can be called from different threads.
*/
public abstract class AbstractMutableLazyVirtualTable extends AbstractLazyVirtualTable
{
protected AbstractMutableLazyVirtualTable(TableMetadata metadata, OnTimeout onTimeout, Sorted sorted)
{
super(metadata, onTimeout, sorted);
}
protected AbstractMutableLazyVirtualTable(TableMetadata metadata, OnTimeout onTimeout, Sorted sorted, Sorted sortedByPartitionKey)
{
super(metadata, onTimeout, sorted, sortedByPartitionKey);
}
protected void applyPartitionDeletion(Object[] partitionKeys)
{
throw invalidRequest("Partition deletion is not supported by table %s", metadata());
}
protected void applyRangeTombstone(Object[] partitionKey, Object[] start, boolean startInclusive, Object[] end, boolean endInclusive)
{
throw invalidRequest("Range deletion is not supported by table %s", metadata());
}
protected void applyRowDeletion(Object[] partitionKey, @Nullable Object[] clusteringKeys)
{
throw invalidRequest("Row deletion is not supported by table %s", metadata());
}
protected void applyRowUpdate(Object[] partitionKeys, @Nullable Object[] clusteringKeys, ColumnMetadata[] columns, Object[] values)
{
throw invalidRequest("Column modification is not supported by table %s", metadata());
}
private void applyRangeTombstone(Object[] pks, RangeTombstone rt)
{
Slice slice = rt.deletedSlice();
Object[] starts = composeClusterings(slice.start(), metadata());
Object[] ends = composeClusterings(slice.end(), metadata());
applyRangeTombstone(pks, starts, slice.start().isInclusive(), ends, slice.end().isInclusive());
}
private void applyRow(Object[] pks, Row row)
{
Object[] cks = row.clustering().kind() == STATIC_CLUSTERING ? null : composeClusterings(row.clustering(), metadata());
if (!row.deletion().isLive())
{
applyRowDeletion(pks, cks);
}
else
{
ColumnMetadata[] columns = new ColumnMetadata[row.columnCount()];
Object[] values = new Object[row.columnCount()];
int i = 0;
for (ColumnData cd : row)
{
ColumnMetadata cm = cd.column();
if (cm.isComplex())
throw new InvalidRequestException(metadata() + " does not support complex column updates");
Cell cell = (Cell)cd;
columns[i] = cm;
if (!cell.isTombstone())
values[i] = cm.type.compose(cell.value(), cell.accessor());
++i;
}
Invariants.require(i == columns.length);
applyRowUpdate(pks, cks, columns, values);
}
}
public void apply(PartitionUpdate update)
{
TableMetadata metadata = metadata();
Object[] pks = composePartitionKeys(update.partitionKey(), metadata);
DeletionInfo deletionInfo = update.deletionInfo();
if (!deletionInfo.getPartitionDeletion().isLive())
{
applyPartitionDeletion(pks);
}
else if (deletionInfo.hasRanges())
{
Iterator<RangeTombstone> iter = deletionInfo.rangeIterator(false);
while (iter.hasNext())
applyRangeTombstone(pks, iter.next());
}
else
{
for (Row row : update)
applyRow(pks, row);
if (!update.staticRow().isEmpty())
applyRow(pks, update.staticRow());
}
}
}

5
src/java/org/apache/cassandra/db/virtual/AbstractVirtualTable.java
View File

@ -29,6 +29,7 @@ import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
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.partitions.AbstractUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
@ -76,7 +77,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
}
@Override
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter)
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
Partition partition = data(partitionKey).getPartition(partitionKey);
@ -89,7 +90,7 @@ public abstract class AbstractVirtualTable implements VirtualTable
}
@Override
public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
public final UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
DataSet data = data();

1167
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

File diff suppressed because it is too large Load Diff

5
src/java/org/apache/cassandra/db/virtual/CollectionVirtualTableAdapter.java
View File

@ -50,6 +50,7 @@ import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
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.AbstractType;
import org.apache.cassandra.db.marshal.BooleanType;
@ -308,7 +309,7 @@ public class CollectionVirtualTableAdapter<R> implements VirtualTable
public UnfilteredPartitionIterator select(DecoratedKey partitionKey,
ClusteringIndexFilter clusteringFilter,
ColumnFilter columnFilter,
RowFilter rowFilter)
RowFilter rowFilter, DataLimits limits)
{
if (!data.iterator().hasNext())
return EmptyIterators.unfilteredPartition(metadata);
@ -349,7 +350,7 @@ public class CollectionVirtualTableAdapter<R> implements VirtualTable
}
@Override
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
return createPartitionIterator(metadata, new AbstractIterator<>()
{

5
src/java/org/apache/cassandra/db/virtual/PartitionKeyStatsTable.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
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.CompositeType;
import org.apache.cassandra.db.marshal.CounterColumnType;
@ -146,7 +147,7 @@ public class PartitionKeyStatsTable implements VirtualTable
}
@Override
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter)
public UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
if (clusteringIndexFilter.isReversed())
throw new InvalidRequestException(REVERSED_QUERY_ERROR);
@ -345,7 +346,7 @@ public class PartitionKeyStatsTable implements VirtualTable
}
@Override
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter)
public UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits)
{
throw new InvalidRequestException(UNSUPPORTED_RANGE_QUERY_ERROR);
}

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

@ -21,6 +21,7 @@ import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.filter.ClusteringIndexFilter;
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.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
@ -31,6 +32,8 @@ import org.apache.cassandra.schema.TableMetadata;
*/
public interface VirtualTable
{
enum Sorted { UNSORTED, ASC, DESC, SORTED }
/**
* Returns the view name.
*
@ -57,23 +60,25 @@ public interface VirtualTable
/**
* Selects the rows from a single partition.
*
* @param partitionKey the partition key
* @param partitionKey the partition key
* @param clusteringIndexFilter the clustering columns to selected
* @param columnFilter the selected columns
* @param rowFilter filter on which rows a given query should include or exclude
* @param columnFilter the selected columns
* @param rowFilter filter on which rows a given query should include or exclude
* @param limits result limits to apply
* @return the rows corresponding to the requested data.
*/
UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter);
UnfilteredPartitionIterator select(DecoratedKey partitionKey, ClusteringIndexFilter clusteringIndexFilter, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits);
/**
* Selects the rows from a range of partitions.
*
* @param dataRange the range of data to retrieve
* @param dataRange the range of data to retrieve
* @param columnFilter the selected columns
* @param rowFilter filter on which rows a given query should include or exclude
* @param rowFilter filter on which rows a given query should include or exclude
* @param limits
* @return the rows corresponding to the requested data.
*/
UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter);
UnfilteredPartitionIterator select(DataRange dataRange, ColumnFilter columnFilter, RowFilter rowFilter, DataLimits limits);
/**
* Truncates data from the underlying source, if supported.
@ -90,4 +95,9 @@ public interface VirtualTable
{
return true;
}
default boolean allowFilteringPrimaryKeysImplicitly()
{
return allowFilteringImplicitly();
}
}

11
src/java/org/apache/cassandra/journal/InMemoryIndex.java
View File

@ -18,6 +18,7 @@
package org.apache.cassandra.journal;
import java.io.IOException;
import java.util.Iterator;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
@ -126,6 +127,16 @@ public final class InMemoryIndex<K> extends Index<K>
return lookUp(id);
}
public Iterator<K> keyIterator(@Nullable K min, @Nullable K max)
{
NavigableMap<K, long[]> m;
if (min == null && max == null) m = index;
else if (min == null) m = index.headMap(max, true);
else if (max == null) m = index.tailMap(min, true);
else m = index.subMap(min, true, max, true);
return m.keySet().iterator();
}
public void persist(Descriptor descriptor)
{
File tmpFile = descriptor.tmpFileFor(Component.INDEX);

87
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -934,11 +934,11 @@ public class Journal<K, V> implements Shutdownable
}
/**
* Static segment iterator iterates all keys in _static_ segments in order.
* segment iterator iterates all keys in order.
*/
public StaticSegmentKeyIterator staticSegmentKeyIterator(K min, K max)
public SegmentKeyIterator segmentKeyIterator(K min, K max, Predicate<Segment<?, ?>> include)
{
return new StaticSegmentKeyIterator(min, max);
return new SegmentKeyIterator(min, max, include);
}
/**
@ -1000,53 +1000,36 @@ public class Journal<K, V> implements Shutdownable
}
}
public class StaticSegmentKeyIterator implements CloseableIterator<KeyRefs<K>>
public class SegmentKeyIterator implements CloseableIterator<KeyRefs<K>>
{
private final ReferencedSegments<K, V> segments;
private final MergeIterator<Head, KeyRefs<K>> iterator;
public StaticSegmentKeyIterator(K min, K max)
public SegmentKeyIterator(K min, K max, Predicate<Segment<?, ?>> include)
{
this.segments = selectAndReference(s -> s.isStatic()
&& s.asStatic().index().entryCount() > 0
this.segments = selectAndReference(s -> include.test(s) && !s.isEmpty()
&& (min == null || keySupport.compare(s.index().lastId(), min) >= 0)
&& (max == null || keySupport.compare(s.index().firstId(), max) <= 0));
List<Iterator<Head>> iterators = new ArrayList<>(segments.count());
for (Segment<K, V> segment : segments.allSorted(true))
{
final StaticSegment<K, V> staticSegment = (StaticSegment<K, V>) segment;
final OnDiskIndex<K>.IndexReader iter = staticSegment.index().reader();
if (min != null) iter.seek(min);
if (max != null) iter.seekEnd(max);
if (!iter.hasNext())
continue;
iterators.add(new AbstractIterator<>()
if (segment.isStatic())
{
final Head head = new Head(staticSegment.descriptor.timestamp);
@Override
protected Head computeNext()
{
if (!iter.hasNext())
return endOfData();
K next = iter.next();
while (next.equals(head.key))
{
if (!iter.hasNext())
return endOfData();
next = iter.next();
}
Invariants.require(!next.equals(head.key),
"%s == %s", next, head.key);
head.key = next;
return head;
}
});
final StaticSegment<K, V> staticSegment = (StaticSegment<K, V>) segment;
final OnDiskIndex<K>.IndexReader iter = staticSegment.index().reader();
if (min != null) iter.seek(min);
if (max != null) iter.seekEnd(max);
if (iter.hasNext())
iterators.add(keyIterator(segment.descriptor.timestamp, iter));
}
else
{
final ActiveSegment<K, V> activeSegment = (ActiveSegment<K, V>) segment;
final Iterator<K> iter = activeSegment.index().keyIterator(min, max);
if (iter.hasNext())
iterators.add(keyIterator(segment.descriptor.timestamp, iter));
}
}
this.iterator = MergeIterator.get(iterators,
@ -1077,6 +1060,34 @@ public class Journal<K, V> implements Shutdownable
});
}
private Iterator<Head> keyIterator(long segment, Iterator<K> iter)
{
final Head head = new Head(segment);
return new AbstractIterator<>()
{
@Override
protected Head computeNext()
{
if (!iter.hasNext())
return endOfData();
K next = iter.next();
while (next.equals(head.key))
{
if (!iter.hasNext())
return endOfData();
next = iter.next();
}
Invariants.require(!next.equals(head.key),
"%s == %s", next, head.key);
head.key = next;
return head;
}
};
}
@Override
public void close()
{

3
src/java/org/apache/cassandra/journal/Segment.java
View File

@ -66,7 +66,8 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
abstract boolean isActive();
abstract boolean isFlushed(long position);
boolean isStatic() { return !isActive(); }
public boolean isStatic() { return !isActive(); }
abstract boolean isEmpty();
abstract ActiveSegment<K, V> asActive();
abstract StaticSegment<K, V> asStatic();

6
src/java/org/apache/cassandra/journal/StaticSegment.java
View File

@ -246,6 +246,12 @@ public final class StaticSegment<K, V> extends Segment<K, V>
return index.entryCount();
}
@Override
boolean isEmpty()
{
return entryCount() == 0;
}
@Override
boolean isActive()
{

5
src/java/org/apache/cassandra/service/accord/AccordExecutor.java
View File

@ -1655,7 +1655,8 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
public static class TaskInfo implements Comparable<TaskInfo>
{
public enum Status { WAITING_TO_LOAD, SCANNING_RANGES, LOADING, WAITING_TO_RUN, RUNNING }
// sorted in name order for reporting to virtual tables
public enum Status { LOADING, RUNNING, SCANNING_RANGES, WAITING_TO_LOAD, WAITING_TO_RUN }
final Status status;
final int commandStoreId;
@ -1706,7 +1707,7 @@ public abstract class AccordExecutor implements CacheSize, LoadExecutor<AccordTa
public int compareTo(TaskInfo that)
{
int c = this.status.compareTo(that.status);
if (c == 0) c = this.position() - that.position();
if (c == 0) c = Integer.compare(this.position(), that.position());
return c;
}
}

14
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -376,7 +376,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
try (CloseableIterator<TopologyUpdate> iter = new CloseableIterator<>()
{
final CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator(topologyUpdateKey(0L),
topologyUpdateKey(Timestamp.MAX_EPOCH));
topologyUpdateKey(Timestamp.MAX_EPOCH),
true);
TopologyImage prev = null;
@Override
@ -571,9 +572,14 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
journalTable.forceCompaction();
}
public void forEach(Consumer<JournalKey> consumer)
public void forEach(Consumer<JournalKey> consumer, boolean includeActive)
{
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator(null, null))
forEach(consumer, null, null, includeActive);
}
public void forEach(Consumer<JournalKey> consumer, @Nullable JournalKey min, @Nullable JournalKey max, boolean includeActive)
{
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator(min, max, includeActive))
{
while (iter.hasNext())
{
@ -610,7 +616,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
this.commandStore = commandStore;
this.replayer = commandStore.replayer();
// Keys in the index are sorted by command store id, so index iteration will be sequential
this.iter = journalTable.keyIterator(new JournalKey(TxnId.NONE, COMMAND_DIFF, commandStore.id()), new JournalKey(TxnId.MAX.withoutNonIdentityFlags(), COMMAND_DIFF, commandStore.id()));
this.iter = journalTable.keyIterator(new JournalKey(TxnId.NONE, COMMAND_DIFF, commandStore.id()), new JournalKey(TxnId.MAX.withoutNonIdentityFlags(), COMMAND_DIFF, commandStore.id()), false);
}
boolean replay()

11
src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
View File

@ -75,6 +75,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.KeySupport;
import org.apache.cassandra.journal.RecordConsumer;
import org.apache.cassandra.journal.Segment;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.service.RetryStrategy;
import org.apache.cassandra.service.accord.AccordKeyspace.JournalColumns;
@ -457,9 +458,9 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
}
@SuppressWarnings("resource") // Auto-closeable iterator will release related resources
public CloseableIterator<Journal.KeyRefs<K>> keyIterator(@Nullable K min, @Nullable K max)
public CloseableIterator<Journal.KeyRefs<K>> keyIterator(@Nullable K min, @Nullable K max, boolean includeActive)
{
return new JournalAndTableKeyIterator(min, max);
return new JournalAndTableKeyIterator(min, max, includeActive);
}
private class TableIterator extends AbstractIterator<K> implements CloseableIterator<K>
@ -515,12 +516,12 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
private class JournalAndTableKeyIterator extends AbstractIterator<Journal.KeyRefs<K>> implements CloseableIterator<Journal.KeyRefs<K>>
{
final TableIterator tableIterator;
final Journal<K, V>.StaticSegmentKeyIterator journalIterator;
final Journal<K, V>.SegmentKeyIterator journalIterator;
private JournalAndTableKeyIterator(K min, K max)
private JournalAndTableKeyIterator(K min, K max, boolean includeActive)
{
this.tableIterator = new TableIterator(min, max);
this.journalIterator = journal.staticSegmentKeyIterator(min, max);
this.journalIterator = journal.segmentKeyIterator(min, max, includeActive ? ignore -> true : Segment::isStatic);
}
K prevFromTable = null;

149
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -21,7 +21,6 @@ package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@ -40,6 +39,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.api.ConfigurationService.EpochReady;
import accord.primitives.Txn;
import org.apache.cassandra.metrics.AccordReplicaMetrics;
import org.apache.cassandra.service.accord.api.AccordViolationHandler;
import org.apache.cassandra.utils.Clock;
@ -58,17 +58,11 @@ import accord.impl.DefaultRemoteListeners;
import accord.impl.RequestCallbacks;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.Node;
import accord.local.Node.Id;
import accord.local.PreLoadContext;
import accord.local.SafeCommand;
import accord.local.ShardDistributor.EvenSplit;
import accord.local.UniqueTimeService.AtomicUniqueTimeWithStaleReservation;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.local.durability.DurabilityService;
import accord.local.durability.ShardDurability;
import accord.messages.Reply;
@ -76,13 +70,9 @@ import accord.messages.Request;
import accord.primitives.FullRoute;
import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.RoutingKeys;
import accord.primitives.SaveStatus;
import accord.primitives.Seekable;
import accord.primitives.Seekables;
import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.Shard;
import accord.topology.Topology;
@ -90,7 +80,6 @@ import accord.topology.TopologyManager;
import accord.utils.DefaultRandom;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.Shutdownable;
@ -115,7 +104,6 @@ import org.apache.cassandra.service.accord.api.AccordScheduler;
import org.apache.cassandra.service.accord.api.AccordTimeService;
import org.apache.cassandra.service.accord.api.AccordTopologySorter;
import org.apache.cassandra.service.accord.api.CompositeTopologySorter;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.api.TokenKey.KeyspaceSplitter;
import org.apache.cassandra.service.accord.interop.AccordInteropAdapter.AccordInteropFactory;
import org.apache.cassandra.service.accord.serializers.TableMetadatas;
@ -140,8 +128,6 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import static accord.api.Journal.TopologyUpdate;
import static accord.api.ProtocolModifiers.Toggles.FastExec.MAY_BYPASS_SAFESTORE;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static accord.local.durability.DurabilityService.SyncLocal.Self;
import static accord.local.durability.DurabilityService.SyncRemote.All;
import static accord.messages.SimpleReply.Ok;
@ -875,139 +861,6 @@ public class AccordService implements IAccordService, Shutdownable
return node.id();
}
@Override
public List<CommandStoreTxnBlockedGraph> debugTxnBlockedGraph(TxnId txnId)
{
return getBlocking(loadDebug(txnId));
}
public AsyncChain<List<CommandStoreTxnBlockedGraph>> loadDebug(TxnId original)
{
CommandStores commandStores = node.commandStores();
if (commandStores.count() == 0)
return AsyncChains.success(Collections.emptyList());
int[] ids = commandStores.ids();
List<AsyncChain<CommandStoreTxnBlockedGraph>> chains = new ArrayList<>(ids.length);
for (int id : ids)
chains.add(loadDebug(original, commandStores.forId(id)).chain());
return AsyncChains.allOf(chains);
}
private AsyncResult<CommandStoreTxnBlockedGraph> loadDebug(TxnId txnId, CommandStore store)
{
CommandStoreTxnBlockedGraph.Builder state = new CommandStoreTxnBlockedGraph.Builder(store.id());
populateAsync(state, store, txnId);
return state;
}
private static void populate(CommandStoreTxnBlockedGraph.Builder state, AccordSafeCommandStore safeStore, TxnId blockedBy)
{
if (safeStore.ifLoadedAndInitialised(blockedBy) != null) populateSync(state, safeStore, blockedBy);
else populateAsync(state, safeStore.commandStore(), blockedBy);
}
private static void populateAsync(CommandStoreTxnBlockedGraph.Builder state, CommandStore store, TxnId txnId)
{
state.asyncTxns.incrementAndGet();
store.execute(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), in -> {
populateSync(state, (AccordSafeCommandStore) in, txnId);
if (0 == state.asyncTxns.decrementAndGet() && 0 == state.asyncKeys.get())
state.complete();
});
}
@Nullable
private static void populateSync(CommandStoreTxnBlockedGraph.Builder state, AccordSafeCommandStore safeStore, TxnId txnId)
{
try
{
if (state.txns.containsKey(txnId))
return; // could plausibly request same txn twice
SafeCommand safeCommand = safeStore.unsafeGet(txnId);
Invariants.nonNull(safeCommand, "Txn %s is not in the cache", txnId);
if (safeCommand.current() == null || safeCommand.current().saveStatus() == SaveStatus.Uninitialised)
return;
CommandStoreTxnBlockedGraph.TxnState cmdTxnState = populateSync(state, safeCommand.current());
if (cmdTxnState.notBlocked())
return;
for (TxnId blockedBy : cmdTxnState.blockedBy)
{
if (!state.knows(blockedBy))
populate(state, safeStore, blockedBy);
}
for (TokenKey blockedBy : cmdTxnState.blockedByKey)
{
if (!state.keys.containsKey(blockedBy))
populate(state, safeStore, blockedBy, txnId, safeCommand.current().executeAt());
}
}
catch (Throwable t)
{
state.tryFailure(t);
}
}
private static void populate(CommandStoreTxnBlockedGraph.Builder state, AccordSafeCommandStore safeStore, TokenKey blockedBy, TxnId txnId, Timestamp executeAt)
{
if (safeStore.ifLoadedAndInitialised(txnId) != null && safeStore.ifLoadedAndInitialised(blockedBy) != null) populateSync(state, safeStore, blockedBy, txnId, executeAt);
else populateAsync(state, safeStore.commandStore(), blockedBy, txnId, executeAt);
}
private static void populateAsync(CommandStoreTxnBlockedGraph.Builder state, CommandStore commandStore, TokenKey blockedBy, TxnId txnId, Timestamp executeAt)
{
state.asyncKeys.incrementAndGet();
commandStore.execute(PreLoadContext.contextFor(txnId, RoutingKeys.of(blockedBy.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), in -> {
populateSync(state, (AccordSafeCommandStore) in, blockedBy, txnId, executeAt);
if (0 == state.asyncKeys.decrementAndGet() && 0 == state.asyncTxns.get())
state.complete();
});
}
private static void populateSync(CommandStoreTxnBlockedGraph.Builder state, AccordSafeCommandStore safeStore, TokenKey pk, TxnId txnId, Timestamp executeAt)
{
try
{
SafeCommandsForKey commandsForKey = safeStore.ifLoadedAndInitialised(pk);
TxnId blocking = commandsForKey.current().blockedOnTxnId(txnId, executeAt);
if (blocking instanceof CommandsForKey.TxnInfo)
blocking = ((CommandsForKey.TxnInfo) blocking).plainTxnId();
state.keys.put(pk, blocking);
if (state.txns.containsKey(blocking))
return;
populate(state, safeStore, blocking);
}
catch (Throwable t)
{
state.tryFailure(t);
}
}
private static CommandStoreTxnBlockedGraph.TxnState populateSync(CommandStoreTxnBlockedGraph.Builder state, Command cmd)
{
CommandStoreTxnBlockedGraph.Builder.TxnBuilder cmdTxnState = state.txn(cmd.txnId(), cmd.executeAt(), cmd.saveStatus());
if (!cmd.hasBeen(Status.Applied) && cmd.hasBeen(Status.Stable))
{
// check blocking state
Command.WaitingOn waitingOn = cmd.asCommitted().waitingOn();
waitingOn.waitingOn.reverseForEach(null, null, null, null, (i1, i2, i3, i4, i) -> {
if (i < waitingOn.txnIdCount())
{
// blocked on txn
cmdTxnState.blockedBy.add(waitingOn.txnId(i));
}
else
{
// blocked on key
cmdTxnState.blockedByKey.add((TokenKey) waitingOn.keys.get(i - waitingOn.txnIdCount()));
}
});
}
return cmdTxnState.build();
}
@Override
public long minEpoch()
{

135
src/java/org/apache/cassandra/service/accord/CommandStoreTxnBlockedGraph.java
View File

@ -1,135 +0,0 @@
/*
* 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.service.accord;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import accord.primitives.SaveStatus;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.service.accord.api.TokenKey;
public class CommandStoreTxnBlockedGraph
{
public final int commandStoreId;
public final Map<TxnId, TxnState> txns;
public final Map<TokenKey, TxnId> keys;
public CommandStoreTxnBlockedGraph(Builder builder)
{
commandStoreId = builder.storeId;
txns = ImmutableMap.copyOf(builder.txns);
keys = ImmutableMap.copyOf(builder.keys);
}
public static class TxnState
{
public final TxnId txnId;
public final Timestamp executeAt;
public final SaveStatus saveStatus;
public final List<TxnId> blockedBy;
public final Set<TokenKey> blockedByKey;
public TxnState(Builder.TxnBuilder builder)
{
txnId = builder.txnId;
executeAt = builder.executeAt;
saveStatus = builder.saveStatus;
blockedBy = ImmutableList.copyOf(builder.blockedBy);
blockedByKey = ImmutableSet.copyOf(builder.blockedByKey);
}
public boolean isBlocked()
{
return !notBlocked();
}
public boolean notBlocked()
{
return blockedBy.isEmpty() && blockedByKey.isEmpty();
}
}
public static class Builder extends AsyncResults.SettableResult<CommandStoreTxnBlockedGraph>
{
final AtomicInteger asyncTxns = new AtomicInteger(), asyncKeys = new AtomicInteger();
final int storeId;
final Map<TxnId, TxnState> txns = new LinkedHashMap<>();
final Map<TokenKey, TxnId> keys = new LinkedHashMap<>();
public Builder(int storeId)
{
this.storeId = storeId;
}
boolean knows(TxnId id)
{
return txns.containsKey(id);
}
public void complete()
{
trySuccess(build());
}
public CommandStoreTxnBlockedGraph build()
{
return new CommandStoreTxnBlockedGraph(this);
}
public TxnBuilder txn(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus)
{
return new TxnBuilder(txnId, executeAt, saveStatus);
}
public class TxnBuilder
{
final TxnId txnId;
final Timestamp executeAt;
final SaveStatus saveStatus;
List<TxnId> blockedBy = new ArrayList<>();
Set<TokenKey> blockedByKey = new LinkedHashSet<>();
public TxnBuilder(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus)
{
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
}
public TxnState build()
{
TxnState state = new TxnState(this);
txns.put(txnId, state);
return state;
}
}
}
}

249
src/java/org/apache/cassandra/service/accord/DebugBlockedTxns.java Normal file
View File

@ -0,0 +1,249 @@
/*
* 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.service.accord;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
import accord.local.PreLoadContext;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.RoutingKeys;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.utils.concurrent.Future;
import static accord.local.LoadKeys.SYNC;
import static accord.local.LoadKeysFor.READ_WRITE;
import static java.util.Collections.emptyList;
public class DebugBlockedTxns
{
public static class Txn implements Comparable<Txn>
{
public final int commandStoreId;
public final int depth;
public final TxnId txnId;
public final Timestamp executeAt;
public final SaveStatus saveStatus;
public final RoutingKey blockedViaKey;
public final List<TxnId> blockedBy;
public final List<TokenKey> blockedByKey;
public Txn(int commandStoreId, int depth, TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, RoutingKey blockedViaKey, List<TxnId> blockedBy, List<TokenKey> blockedByKey)
{
this.commandStoreId = commandStoreId;
this.depth = depth;
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
this.blockedViaKey = blockedViaKey;
this.blockedBy = blockedBy;
this.blockedByKey = blockedByKey;
}
public boolean isBlocked()
{
return !notBlocked();
}
public boolean notBlocked()
{
return blockedBy.isEmpty() && blockedByKey.isEmpty();
}
@Override
public int compareTo(Txn that)
{
int c = Integer.compare(this.commandStoreId, that.commandStoreId);
if (c == 0) c = Integer.compare(this.depth, that.depth);
if (c == 0) c = this.txnId.compareTo(that.txnId);
return c;
}
}
final IAccordService service;
final Consumer<Txn> visit;
final TxnId root;
final int maxDepth;
final Set<Object> visited = Collections.newSetFromMap(new ConcurrentHashMap<>());
final ConcurrentLinkedQueue<AsyncChain<Void>> queuedKeys = new ConcurrentLinkedQueue<>();
final ConcurrentLinkedQueue<AsyncChain<Txn>> queuedTxn = new ConcurrentLinkedQueue<>();
public DebugBlockedTxns(IAccordService service, TxnId root, int maxDepth, Consumer<Txn> visit)
{
this.service = service;
this.visit = visit;
this.root = root;
this.maxDepth = maxDepth;
}
public static void visit(IAccordService accord, TxnId txnId, int maxDepth, long deadlineNanos, Consumer<Txn> visit) throws TimeoutException
{
new DebugBlockedTxns(accord, txnId, maxDepth, visit).visit(deadlineNanos);
}
private void visit(long deadlineNanos) throws TimeoutException
{
CommandStores commandStores = service.node().commandStores();
if (commandStores.count() == 0)
return;
int[] ids = commandStores.ids();
List<AsyncChain<Txn>> chains = new ArrayList<>(ids.length);
for (int id : ids)
chains.add(visitRootTxnAsync(commandStores.forId(id), root));
List<AsyncChain> tmp = new ArrayList<>();
Future<List<Txn>> next = AccordService.toFuture(AsyncChains.allOf(chains));
while (next != null)
{
if (!next.awaitUntilThrowUncheckedOnInterrupt(deadlineNanos))
throw new TimeoutException();
next.rethrowIfFailed();
List<Txn> process = next.getNow().stream()
.filter(Objects::nonNull)
.sorted(Comparator.naturalOrder())
.collect(Collectors.toList());
for (Txn txn : process)
visit.accept(txn);
Future<List<Void>> awaitKeys = drainToFuture(queuedKeys, (List<AsyncChain<Void>>)(List)tmp);
if (awaitKeys != null && !awaitKeys.awaitUntilThrowUncheckedOnInterrupt(deadlineNanos))
throw new TimeoutException();
next = drainToFuture(queuedTxn, (List<AsyncChain<Txn>>)(List)tmp);
}
}
private <T> Future<List<T>> drainToFuture(Queue<AsyncChain<T>> drain, List<AsyncChain<T>> tmp)
{
AsyncChain<T> next;
while (null != (next = drain.poll()))
tmp.add(next);
if (tmp.isEmpty())
return null;
Future<List<T>> result = AccordService.toFuture(AsyncChains.allOf(List.copyOf(tmp)));
tmp.clear();
return result;
}
private AsyncChain<Txn> visitRootTxnAsync(CommandStore commandStore, TxnId txnId)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
return visitTxnSync(safeStore, command, command.executeAt(), null, 0);
});
}
private AsyncChain<Txn> visitTxnAsync(CommandStore commandStore, TxnId txnId, Timestamp rootExecuteAt, @Nullable TokenKey byKey, int depth, boolean recurse)
{
return commandStore.chain(PreLoadContext.contextFor(txnId, "Populate txn_blocked_by"), safeStore -> {
Command command = safeStore.unsafeGetNoCleanup(txnId).current();
if (command == null || command.saveStatus() == SaveStatus.Uninitialised)
return null;
return visitTxnSync(safeStore, command, rootExecuteAt, byKey, depth);
});
}
private Txn visitTxnSync(SafeCommandStore safeStore, Command command, Timestamp rootExecuteAt, @Nullable TokenKey byKey, int depth)
{
List<TxnId> waitingOnTxnId = new ArrayList<>();
List<TokenKey> waitingOnKey = new ArrayList<>();
if (!command.hasBeen(Status.Applied) && command.hasBeen(Status.Stable))
{
// check blocking state
Command.WaitingOn waitingOn = command.asCommitted().waitingOn();
waitingOn.waitingOn.reverseForEach(null, null, null, null, (i1, i2, i3, i4, i) -> {
if (i < waitingOn.txnIdCount()) waitingOnTxnId.add(waitingOn.txnId(i));
else waitingOnKey.add((TokenKey) waitingOn.keys.get(i - waitingOn.txnIdCount()));
});
}
CommandStore commandStore = safeStore.commandStore();
if (depth < maxDepth)
{
for (TxnId waitingOn : waitingOnTxnId)
{
if (visited.add(waitingOn))
queuedTxn.add(visitTxnAsync(commandStore, waitingOn, rootExecuteAt, null, depth + 1, true));
}
for (TokenKey key : waitingOnKey)
{
if (visited.add(key))
queuedKeys.add(visitKeysAsync(commandStore, key, rootExecuteAt, depth + 1));
}
}
return new Txn(commandStore.id(), depth, command.txnId(), command.executeAt(), command.saveStatus(), byKey, waitingOnTxnId, waitingOnKey);
}
private AsyncChain<Void> visitKeysAsync(CommandStore commandStore, TokenKey key, Timestamp rootExecuteAt, int depth)
{
return commandStore.chain(PreLoadContext.contextFor(RoutingKeys.of(key.toUnseekable()), SYNC, READ_WRITE, "Populate txn_blocked_by"), safeStore -> {
visitKeysSync(safeStore, key, rootExecuteAt, depth);
});
}
private void visitKeysSync(SafeCommandStore safeStore, TokenKey key, Timestamp rootExecuteAt, int depth)
{
SafeCommandsForKey commandsForKey = safeStore.ifLoadedAndInitialised(key);
TxnId blocking = commandsForKey.current().blockedOnTxnId(root, rootExecuteAt);
CommandStore commandStore = safeStore.commandStore();
if (blocking == null)
{
queuedTxn.add(AsyncChains.success(new Txn(commandStore.id(), depth, null, null, null, key, emptyList(), emptyList())));
}
else
{
// TODO (required): this type check should not be needed; release accord version that fixes it at origin
if (blocking instanceof CommandsForKey.TxnInfo)
blocking = ((CommandsForKey.TxnInfo) blocking).plainTxnId();
boolean recurse = visited.add(blocking);
queuedTxn.add(visitTxnAsync(commandStore, blocking, rootExecuteAt, key, depth, recurse));
}
}
}

16
src/java/org/apache/cassandra/service/accord/IAccordService.java
View File

@ -19,7 +19,6 @@
package org.apache.cassandra.service.accord;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.List;
import java.util.concurrent.TimeUnit;
@ -49,7 +48,6 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.topology.TopologyManager;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
@ -178,8 +176,6 @@ public interface IAccordService
Id nodeId();
List<CommandStoreTxnBlockedGraph> debugTxnBlockedGraph(TxnId txnId);
long minEpoch();
void awaitDone(TableId id, long epoch);
@ -341,12 +337,6 @@ public interface IAccordService
throw new UnsupportedOperationException();
}
@Override
public List<CommandStoreTxnBlockedGraph> debugTxnBlockedGraph(TxnId txnId)
{
return Collections.emptyList();
}
@Override
public long minEpoch()
{
@ -551,12 +541,6 @@ public interface IAccordService
return delegate.nodeId();
}
@Override
public List<CommandStoreTxnBlockedGraph> debugTxnBlockedGraph(TxnId txnId)
{
return delegate.debugTxnBlockedGraph(txnId);
}
@Override
public long minEpoch()
{

2
src/java/org/apache/cassandra/tools/StandaloneJournalUtil.java
View File

@ -274,7 +274,7 @@ public class StandaloneJournalUtil implements Runnable
Map<Integer, RedundantBefore> cache = new HashMap<>();
journal.start(null);
journal.forEach(key -> processKey(cache, journal, key, txnId, sinceTimestamp, untilTimestamp, skipAllErrors, skipExceptionTypes));
journal.forEach(key -> processKey(cache, journal, key, txnId, sinceTimestamp, untilTimestamp, skipAllErrors, skipExceptionTypes), false);
}
private void processKey(Map<Integer, RedundantBefore> redundantBeforeCache, AccordJournal journal, JournalKey key, Timestamp txnId, Timestamp minTimestamp, Timestamp maxTimestamp, boolean skipAllErrors, Set<String> skipExceptionTypes)

4
test/distributed/org/apache/cassandra/fuzz/topology/JournalGCTest.java
View File

@ -113,7 +113,7 @@ public class JournalGCTest extends FuzzTestBase
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF && (a.get() == null || v.id.compareTo(a.get()) > 0))
a.set(v.id);
});
}, false);
return a.get() == null ? "" : a.get().toString();
});
@ -123,7 +123,7 @@ public class JournalGCTest extends FuzzTestBase
((AccordService) AccordService.instance()).journal().forEach((v) -> {
if (v.type == JournalKey.Type.COMMAND_DIFF && v.id.compareTo(maxId) <= 0)
a.incrementAndGet();
});
}, false);
return a.get();
}, maximumId);

2
test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
View File

@ -332,7 +332,7 @@ public class AccordJournalBurnTest extends BurnTestBase
private TreeMap<JournalKey, Command> read(CommandStores commandStores)
{
TreeMap<JournalKey, Command> result = new TreeMap<>(JournalKey.SUPPORT::compare);
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator(null, null))
try (CloseableIterator<Journal.KeyRefs<JournalKey>> iter = journalTable.keyIterator(null, null, false))
{
JournalKey prev = null;
while (iter.hasNext())

2
test/unit/org/apache/cassandra/cql3/CQLTester.java
View File

@ -3185,7 +3185,7 @@ public abstract class CQLTester
private static String formatValue(ByteBuffer bb, AbstractType<?> type)
{
if (bb == null)
if (bb == null || (!bb.hasRemaining() && type.isEmptyValueMeaningless()))
return "null";
if (type instanceof CollectionType)

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

@ -19,6 +19,8 @@
package org.apache.cassandra.db.virtual;
import java.util.Collections;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -35,6 +37,7 @@ import org.slf4j.LoggerFactory;
import accord.api.ProtocolModifiers;
import accord.messages.NoWaitRequest;
import accord.api.RoutingKey;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.SaveStatus;
@ -47,6 +50,7 @@ import org.apache.cassandra.config.OptionaldPositiveInt;
import org.apache.cassandra.config.YamlConfigurationLoader;
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.locator.InetAddressAndPort;
import org.apache.cassandra.net.Message;
@ -56,11 +60,14 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.service.CassandraDaemon;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.transport.Dispatcher;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.concurrent.Condition;
import org.assertj.core.api.Assertions;
import org.awaitility.Awaitility;
@ -87,6 +94,12 @@ public class AccordDebugKeyspaceTest extends CQLTester
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_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_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_JOURNAL =
String.format("SELECT txn_id, save_status FROM %s.%s WHERE txn_id=?", SchemaConstants.VIRTUAL_ACCORD_DEBUG, AccordDebugKeyspace.JOURNAL);
@ -234,12 +247,42 @@ public class AccordDebugKeyspaceTest extends CQLTester
getBlocking(accord.node().coordinate(id, txn));
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.ReadyToExecute.name(), SaveStatus.Applying.name(), SaveStatus.Applied.name()))));
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN, id.toString()), row(id.toString(), "Applied")));
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_JOURNAL, 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"));
}
@Test
public void manyTxns() throws ExecutionException, InterruptedException
{
String tableName = createTable("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'");
AccordService accord = accord();
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)
await.add(accord.coordinateAsync(0, 0, txn, ConsistencyLevel.QUORUM, new Dispatcher.RequestTime(Clock.Global.nanoTime())));
AccordCommandStore commandStore = (AccordCommandStore) accord.node().commandStores().unsafeForKey((RoutingKey) txn.keys().get(0).toUnseekable());
await.forEach(IAccordService.IAccordResult::awaitAndGet);
assertRows(execute(QUERY_TXNS, commandStore.id()),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied")
);
assertRows(execute(QUERY_TXNS_SEARCH, commandStore.id(), TxnId.NONE.toString()),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied"),
row("Applied")
);
}
@Test
public void inflight() throws ExecutionException, InterruptedException
{
@ -263,11 +306,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
filter.preAccept.awaitThrowUncheckedOnInterrupt();
assertRows(execute(QUERY_TXN_BLOCKED_BY, id.toString()),
row(id.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
row(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(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, SaveStatus.ReadyToExecute.name()));
row(id.toString(), anyInt(), 0, "", "", any(), SaveStatus.ReadyToExecute.name()));
}
finally
{
@ -299,12 +341,13 @@ public class AccordDebugKeyspaceTest extends CQLTester
accord.node().coordinate(first, createTxn(insertTxn, 0, 0, 0, 0, 0)).beginAsResult();
filter.preAccept.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() ->assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", any(), null, anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name()))));
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), anyInt(), 0, "", any(), any(), anyOf(SaveStatus.PreAccepted.name(), SaveStatus.ReadyToExecute.name())));
filter.apply.awaitThrowUncheckedOnInterrupt();
spinUntilSuccess(() -> assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", anyNonNull(), null, SaveStatus.ReadyToExecute.name())));
assertRows(execute(QUERY_TXN_BLOCKED_BY, first.toString()),
row(first.toString(), anyInt(), 0, "", any(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
filter.reset();
TxnId second = accord.node().nextTxnIdWithDefaultFlags(Txn.Kind.Write, Routable.Domain.Key);
filter.reset();
@ -319,12 +362,10 @@ public class AccordDebugKeyspaceTest extends CQLTester
return rs.size() == 2;
});
assertRows(execute(QUERY_TXN_BLOCKED_BY, second.toString()),
row(second.toString(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", anyNonNull(), null, SaveStatus.Stable.name()),
row(second.toString(), KEYSPACE, tableName, anyInt(), 1, first.toString(), "Key", anyNonNull(), anyNonNull(), SaveStatus.ReadyToExecute.name()));
row(second.toString(), anyInt(), 0, "", "", anyNonNull(), SaveStatus.Stable.name()),
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(), KEYSPACE, tableName, anyInt(), 0, ByteBufferUtil.EMPTY_BYTE_BUFFER, "Self", anyNonNull(), null, SaveStatus.Stable.name()));
row(second.toString(), anyInt(), 0, any(), "", anyNonNull(), SaveStatus.Stable.name()));
}
finally
{
@ -463,4 +504,6 @@ public class AccordDebugKeyspaceTest extends CQLTester
return !dropVerbs.contains(msg.verb());
}
}
}