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

CEP-45: Fix ALLOW FILTERING queries for mutation tracking 

Patch by Blake Eggleston; Reviewed by Aleksey Yeschenko for CASSANDRA-20555
This commit is contained in:
Blake Eggleston 2025-04-30 10:52:30 -07:00
parent 65caf3f58b
commit 14a743f8f7
19 changed files with 1596 additions and 434 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -544,6 +544,24 @@ public abstract class ReadCommand extends AbstractReadQuery
}
}
public RowFilter rowFilter(Index.Searcher searcher)
{
// If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
// no point in checking it again.
return (null == searcher) ? rowFilter() : indexQueryPlan.postIndexQueryFilter();
}
private UnfilteredPartitionIterator withRowFilter(UnfilteredPartitionIterator iterator, Index.Searcher searcher)
{
/*
* TODO: We'll currently do filtering by the rowFilter here because it's convenient. However,
* we'll probably want to optimize by pushing it down the layer (like for dropped columns) as it
* would be more efficient (the sooner we discard stuff we know we don't care, the less useless
* processing we do on it).
*/
return rowFilter(searcher).filter(iterator, nowInSec());
}
private UnfilteredPartitionIterator completeRead(UnfilteredPartitionIterator iterator, ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos)
{
COMMAND.set(this);
@ -556,18 +574,7 @@ public abstract class ReadCommand extends AbstractReadQuery
iterator = maybeRecordPurgeableTombstones(iterator, cfs);
iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false);
iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos);
// If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
// no point in checking it again.
RowFilter filter = (null == searcher) ? rowFilter() : indexQueryPlan.postIndexQueryFilter();
/*
* TODO: We'll currently do filtering by the rowFilter here because it's convenient. However,
* we'll probably want to optimize by pushing it down the layer (like for dropped columns) as it
* would be more efficient (the sooner we discard stuff we know we don't care, the less useless
* processing we do on it).
*/
iterator = filter.filter(iterator, nowInSec());
iterator = withRowFilter(iterator, searcher);
// apply the limits/row counter; this transformation is stopping and would close the iterator as soon
// as the count is observed; if that happens in the middle of an open RT, its end bound will not be included.

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

@ -182,6 +182,23 @@ public class SinglePartitionReadCommand extends ReadCommand implements SinglePar
dataRange);
}
public static SinglePartitionReadCommand fromRangeRead(DecoratedKey key, PartitionRangeReadCommand command, DataLimits limits)
{
return create(command.serializedAtEpoch(),
command.isDigestQuery(),
command.digestVersion(),
command.acceptsTransient(),
command.metadata(),
command.nowInSec(),
command.columnFilter(),
command.rowFilter(),
limits,
key,
command.clusteringIndexFilter(key),
command.indexQueryPlan(),
command.isTrackingWarnings());
}
/**
* Creates a new read command on a single partition.
*

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

@ -54,6 +54,7 @@ import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.rows.Cell;
@ -61,6 +62,7 @@ import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.db.rows.ComplexColumnData;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.exceptions.InvalidRequestException;
@ -216,6 +218,112 @@ public class RowFilter implements Iterable<RowFilter.Expression>
return false;
}
public static class RowFilterTransformation extends Transformation<BaseRowIterator<?>>
{
private final TableMetadata metadata;
private final long nowInSec;
private final List<Expression> partitionLevelExpressions = new ArrayList<>();
private final List<Expression> rowLevelExpressions = new ArrayList<>();
private final boolean needsReconciliation;
private DecoratedKey pk;
private RowFilterTransformation(RowFilter filter, TableMetadata metadata, long nowInSec)
{
this.metadata = metadata;
this.nowInSec = nowInSec;
for (Expression e: filter.expressions)
{
if (e.column.isStatic() || e.column.isPartitionKey())
partitionLevelExpressions.add(e);
else
rowLevelExpressions.add(e);
}
this.needsReconciliation = filter.needsReconciliation();
}
public int potentialMatches(PartitionUpdate update)
{
try (UnfilteredRowIterator partition = update.unfilteredIterator())
{
int matches = 0;
for (Expression e : partitionLevelExpressions)
{
if (e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
{
matches++;
break;
}
}
while (partition.hasNext())
{
Unfiltered unfiltered = partition.next();
if (unfiltered instanceof Row)
{
Row row = (Row) unfiltered;
for (Expression e : rowLevelExpressions)
{
if (e.isSatisfiedBy(metadata, pk, row, nowInSec))
{
matches++;
break;
}
}
}
}
return matches;
}
}
@Override
protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
{
pk = partition.partitionKey();
// Short-circuit all partitions that won't match based on static and partition keys
for (Expression e : partitionLevelExpressions)
{
if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
{
partition.close();
return null;
}
}
BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
? Transformation.apply((UnfilteredRowIterator) partition, this)
: Transformation.apply((RowIterator) partition, this);
boolean filterNonStaticColumns = !rowLevelExpressions.isEmpty();
if (filterNonStaticColumns && !iterator.hasNext())
{
iterator.close();
return null;
}
return iterator;
}
@Override
public Row applyToRow(Row row)
{
// If we purge deletions when reconciliation is required, we hide information replica filtering
// protection would require to filter rows that are no longer matches are the coordinator.
Row purged = needsReconciliation ? row : row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
if (purged == null)
return null;
for (Expression e : rowLevelExpressions)
if (!e.isSatisfiedBy(metadata, pk, purged, nowInSec))
return null;
return row;
}
}
/**
* Note that the application of this transformation does not yet take {@link #isStrict()} into account. This means
* that even when strict filtering is not safe, expressions will be applied as intersections rather than unions.
@ -224,68 +332,9 @@ public class RowFilter implements Iterable<RowFilter.Expression>
*
* @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-190007">CASSANDRA-19007</a>
*/
protected Transformation<BaseRowIterator<?>> filter(TableMetadata metadata, long nowInSec)
public RowFilterTransformation filter(TableMetadata metadata, long nowInSec)
{
List<Expression> partitionLevelExpressions = new ArrayList<>();
List<Expression> rowLevelExpressions = new ArrayList<>();
for (Expression e: expressions)
{
if (e.column.isStatic() || e.column.isPartitionKey())
partitionLevelExpressions.add(e);
else
rowLevelExpressions.add(e);
}
long numberOfRegularColumnExpressions = rowLevelExpressions.size();
final boolean filterNonStaticColumns = numberOfRegularColumnExpressions > 0;
return new Transformation<>()
{
DecoratedKey pk;
@Override
protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
{
pk = partition.partitionKey();
// Short-circuit all partitions that won't match based on static and partition keys
for (Expression e : partitionLevelExpressions)
if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
{
partition.close();
return null;
}
BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
? Transformation.apply((UnfilteredRowIterator) partition, this)
: Transformation.apply((RowIterator) partition, this);
if (filterNonStaticColumns && !iterator.hasNext())
{
iterator.close();
return null;
}
return iterator;
}
@Override
public Row applyToRow(Row row)
{
// If we purge deletions when reconciliation is required, we hide information replica filtering
// protection would require to filter rows that are no longer matches are the coordinator.
Row purged = needsReconciliation() ? row : row.purge(DeletionPurger.PURGE_ALL, nowInSec, metadata.enforceStrictLiveness());
if (purged == null)
return null;
for (Expression e : rowLevelExpressions)
if (!e.isSatisfiedBy(metadata, pk, purged, nowInSec))
return null;
return row;
}
};
return new RowFilterTransformation( this, metadata, nowInSec);
}
/**

66
src/java/org/apache/cassandra/db/partitions/PartitionIterators.java
View File

@ -19,11 +19,13 @@ package org.apache.cassandra.db.partitions;
import java.io.IOError;
import java.io.IOException;
import java.util.Comparator;
import java.util.List;
import org.apache.cassandra.db.EmptyIterators;
import org.apache.cassandra.db.SinglePartitionReadQuery;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.rows.BaseRowIterator;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.rows.RowIterators;
import org.apache.cassandra.db.transform.MorePartitions;
@ -32,6 +34,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.utils.AbstractIterator;
import org.apache.cassandra.utils.MergeIterator;
public abstract class PartitionIterators
{
@ -99,6 +102,67 @@ public abstract class PartitionIterators
}
}
/**
* Merges multiple partition iterators with the requirement that there are no keys in common between any
* of the iterators
*/
public static PartitionIterator mergeNonOverlapping(List<PartitionIterator> iterators)
{
MergeIterator.Reducer<RowIterator, RowIterator> reducer = new MergeIterator.Reducer<>()
{
RowIterator current;
@Override
protected void onKeyChange()
{
current = null;
}
@Override
public void reduce(int idx, RowIterator partition)
{
if (current != null)
{
throw new IllegalStateException("Multiple partitions received for " + current.partitionKey());
}
current = partition;
}
@Override
protected RowIterator getReduced()
{
return current;
}
};
MergeIterator<RowIterator, RowIterator> mergeIterator = MergeIterator.get(iterators, rowIteratorComparator, reducer);
return new AbstractPartitionIterator()
{
@Override
protected RowIterator computeNext()
{
return mergeIterator.hasNext() ? mergeIterator.next() : endOfData();
}
};
}
private static final Comparator<RowIterator> rowIteratorComparator = Comparator.comparing(BaseRowIterator::partitionKey);
/**
* Consumes all rows in the next partition of the provided partition iterator.
*/
public static void consumeNext(PartitionIterator iterator)
{
if (iterator.hasNext())
{
try (RowIterator partition = iterator.next())
{
while (partition.hasNext())
partition.next();
}
}
}
/**
* Wraps the provided iterator so it logs the returned rows for debugging purposes.
* <p>
@ -209,5 +273,5 @@ public abstract class PartitionIterators
}
};
}
};
}
}

228
src/java/org/apache/cassandra/service/reads/tracked/AbstractPartialTrackedRead.java
View File

@ -18,8 +18,6 @@
package org.apache.cassandra.service.reads.tracked;
import java.util.List;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
@ -30,34 +28,176 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.transform.RTBoundValidator;
import org.apache.cassandra.index.Index;
import static org.apache.cassandra.db.partitions.UnfilteredPartitionIterators.MergeListener.NOOP;
public abstract class AbstractPartialTrackedRead implements PartialTrackedRead
{
private static final Logger logger = LoggerFactory.getLogger(AbstractPartialTrackedRead.class);
private enum State
protected interface Augmentable
{
INITIALIZED,
PREPARED,
READING,
FINISHED
State augment(PartitionUpdate update);
}
protected static abstract class State
{
protected static final State CLOSED = new State()
{
@Override
String name()
{
return "closed";
}
@Override
boolean isClosed()
{
return true;
}
};
abstract String name();
boolean isInitialized()
{
return false;
}
Initialized asInitialized()
{
throw new IllegalStateException("State is " + name() + ", not " + Initialized.NAME);
}
boolean isPrepared()
{
return false;
}
Prepared asPrepared()
{
throw new IllegalStateException("State is " + name() + ", not " + Prepared.NAME);
}
boolean isCompleted()
{
return false;
}
Completed asCompleted()
{
throw new IllegalStateException("State is " + name() + ", not " + Completed.NAME);
}
boolean isAugmentable()
{
return isPrepared() || isInitialized();
}
Augmentable asAugmentable()
{
if (isPrepared()) return asPrepared();
throw new IllegalStateException("State is " + name() + ", not augmentable");
}
boolean isClosed()
{
return false;
}
void close()
{
}
}
// TODO (expected): this is a redundant state, never exposed
protected final class Initialized extends State
{
static final String NAME = "initialized";
@Override
String name()
{
return NAME;
}
@Override
boolean isInitialized()
{
return true;
}
@Override
Initialized asInitialized()
{
return this;
}
Prepared prepare(UnfilteredPartitionIterator initialData)
{
return prepareInternal(initialData);
}
}
protected abstract Prepared prepareInternal(UnfilteredPartitionIterator initialData);
protected abstract class Prepared extends State implements Augmentable
{
private static final String NAME = "prepared";
@Override
String name()
{
return NAME;
}
@Override
boolean isPrepared()
{
return true;
}
@Override
Prepared asPrepared()
{
return this;
}
abstract Completed complete();
}
protected abstract class Completed extends State
{
private static final String NAME = "completed";
@Override
String name()
{
return NAME;
}
protected abstract UnfilteredPartitionIterator iterator();
protected abstract CompletedRead createResult(UnfilteredPartitionIterator iterator);
protected CompletedRead getResult()
{
UnfilteredPartitionIterator result = command().completeTrackedRead(iterator(), AbstractPartialTrackedRead.this);
// validate that the sequence of RT markers is correct: open is followed by close, deletion times for both
// ends equal, and there are no dangling RT bound in any partition.
result = RTBoundValidator.validate(result, RTBoundValidator.Stage.PROCESSED, true);
return createResult(result);
}
}
final ReadExecutionController executionController;
final Index.Searcher searcher;
final ColumnFamilyStore cfs;
final long startTimeNanos;
volatile State state = State.INITIALIZED;
private State state = new Initialized();
public AbstractPartialTrackedRead(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos)
public AbstractPartialTrackedRead(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos)
{
this.executionController = executionController;
this.searcher = searcher;
this.cfs = cfs;
this.startTimeNanos = startTimeNanos;
}
@ -68,12 +208,6 @@ public abstract class AbstractPartialTrackedRead implements PartialTrackedRead
return executionController;
}
@Override
public Index.Searcher searcher()
{
return searcher;
}
@Override
public ColumnFamilyStore cfs()
{
@ -86,69 +220,47 @@ public abstract class AbstractPartialTrackedRead implements PartialTrackedRead
return startTimeNanos;
}
abstract void freezeInitialData();
abstract UnfilteredPartitionIterator initialData();
abstract UnfilteredPartitionIterator augmentedData();
abstract void augmentResponse(PartitionUpdate update);
protected synchronized State state()
{
return state;
}
/**
* Implementors need to call this before returning this from createInProgressRead
* TODO (expected): this is a redundant transition from a redundant state (INITIALIZED)
*/
synchronized void prepare()
synchronized void prepare(UnfilteredPartitionIterator initialData)
{
logger.trace("Preparing read {}", this);
Preconditions.checkState(state == State.INITIALIZED);
freezeInitialData();
state = State.PREPARED;
state = state.asInitialized().prepare(initialData);
}
@Override
public void augment(Mutation mutation)
public synchronized void augment(Mutation mutation)
{
Preconditions.checkState(state == State.PREPARED);
PartitionUpdate update = mutation.getPartitionUpdate(command().metadata());
if (update != null)
augmentResponse(update);
state = state.asAugmentable().augment(update);
}
private UnfilteredPartitionIterator complete(UnfilteredPartitionIterator iterator)
{
return command().completeTrackedRead(iterator, this);
}
abstract CompletedRead createResult(UnfilteredPartitionIterator iterator);
@Override
public synchronized CompletedRead complete()
{
Preconditions.checkState(state == State.PREPARED);
state = State.READING;
UnfilteredPartitionIterator initial = initialData();
UnfilteredPartitionIterator augmented = augmentedData();
UnfilteredPartitionIterator result = augmented != null ?
UnfilteredPartitionIterators.merge(List.of(initial, augmented), NOOP) :
initial;
result = complete(result);
// validate that the sequence of RT markers is correct: open is followed by close, deletion times for both
// ends equal, and there are no dangling RT bound in any partition.
result = RTBoundValidator.validate(result, RTBoundValidator.Stage.PROCESSED, true);
return createResult(complete(result));
Preconditions.checkState(state.isPrepared());
Completed completed = state.asPrepared().complete();
state = completed;
return completed.getResult();
}
@Override
public synchronized void close()
{
if (state == State.FINISHED)
if (state.isClosed())
return;
logger.trace("Closing read {}", this);
state.close();
executionController.close();
state = State.FINISHED;
state = State.CLOSED;
}
}

186
src/java/org/apache/cassandra/service/reads/tracked/ExtendingCompletedRead.java Normal file
View File

@ -0,0 +1,186 @@
/*
* 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.reads.tracked;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.PartitionRangeReadCommand;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.transform.EmptyPartitionsDiscarder;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
class ExtendingCompletedRead implements PartialTrackedRead.CompletedRead
{
private static final Logger logger = LoggerFactory.getLogger(ExtendingCompletedRead.class);
final PartitionRangeReadCommand command;
final UnfilteredPartitionIterator iterator;
// merged end-result counter
final DataLimits.Counter mergedResultCounter;
private final boolean partitionsFetched;
private final boolean initialIteratorExhausted;
protected final AbstractBounds<PartitionPosition> followUpBounds;
public ExtendingCompletedRead(PartitionRangeReadCommand command,
UnfilteredPartitionIterator iterator,
boolean partitionsFetched,
boolean initialIteratorExhausted,
AbstractBounds<PartitionPosition> followUpBounds)
{
this.command = command;
this.iterator = iterator;
mergedResultCounter = command.limits().newCounter(command.nowInSec(),
true,
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness());
this.partitionsFetched = partitionsFetched;
this.initialIteratorExhausted = initialIteratorExhausted;
this.followUpBounds = followUpBounds;
}
@Override
public TrackedDataResponse response()
{
PartitionIterator filtered = UnfilteredPartitionIterators.filter(iterator, command.nowInSec());
PartitionIterator counted = Transformation.apply(filtered, mergedResultCounter);
PartitionIterator result = Transformation.apply(counted, new EmptyPartitionsDiscarder());
return TrackedDataResponse.create(result, command.columnFilter());
}
static boolean followUpReadRequired(ReadCommand command, DataLimits.Counter mergedResultCounter, boolean initialIteratorExhausted, boolean partitionsFetched)
{
// never try to request additional partitions from replicas if our reconciled partitions are already filled to the limit
if (mergedResultCounter.isDone())
return false;
// we do not apply short read protection when we have no limits at all
if (command.limits().isUnlimited())
return false;
/*
* If this is a single partition read command or an (indexed) partition range read command with
* a partition key specified, then we can't and shouldn't try fetch more partitions.
*/
if (command.isLimitedToOnePartition())
return false;
/*
* If the returned result doesn't have enough rows/partitions to satisfy even the original limit, don't ask for more.
*
* Can only take the short cut if there is no per partition limit set. Otherwise it's possible to hit false
* positives due to some rows being uncounted for in certain scenarios (see CASSANDRA-13911).
*/
if (initialIteratorExhausted && command.limits().perPartitionCount() == DataLimits.NO_LIMIT)
return false;
/*
* Either we had an empty iterator as the initial response, or our moreContents() call got us an empty iterator.
* There is no point to ask the replica for more rows - it has no more in the requested range.
*/
if (!partitionsFetched)
return false;
return true;
}
protected boolean followUpRequired()
{
return followUpReadRequired(command, mergedResultCounter, initialIteratorExhausted, partitionsFetched);
}
static int toQuery(ReadCommand command, DataLimits.Counter mergedResultCounter)
{
/*
* We are going to fetch one partition at a time for thrift and potentially more for CQL.
* The row limit will either be set to the per partition limit - if the command has no total row limit set, or
* the total # of rows remaining - if it has some. If we don't grab enough rows in some of the partitions,
* then future ShortReadRowsProtection.moreContents() calls will fetch the missing ones.
*/
return command.limits().count() != DataLimits.NO_LIMIT
? command.limits().count() - mergedResultCounter.rowsCounted()
: command.limits().perPartitionCount();
}
@Override
public Future<TrackedDataResponse> followupRead(TrackedDataResponse initialResponse, ConsistencyLevel consistencyLevel, long expiresAtNanos)
{
if (!followUpRequired())
return null;
/*
* We are going to fetch one partition at a time for thrift and potentially more for CQL.
* The row limit will either be set to the per partition limit - if the command has no total row limit set, or
* the total # of rows remaining - if it has some. If we don't grab enough rows in some of the partitions,
* then future ShortReadRowsProtection.moreContents() calls will fetch the missing ones.
*/
int toQuery = toQuery(command, mergedResultCounter);
ColumnFamilyStore.metricsFor(command.metadata().id).shortReadProtectionRequests.mark();
Tracing.trace("Requesting {} extra rows from {} for short read protection", toQuery, FBUtilities.getBroadcastAddressAndPort());
logger.info("Requesting {} extra rows from {} for short read protection", toQuery, FBUtilities.getBroadcastAddressAndPort());
return makeFollowupRead(initialResponse, toQuery, consistencyLevel, expiresAtNanos);
}
protected Future<TrackedDataResponse> makeFollowupRead(TrackedDataResponse initialResponse, int toQuery, ConsistencyLevel consistencyLevel, long expiresAtNanos)
{
TrackedRead.Range followUpRead = PartialTrackedRangeRead.makeFollowUpRead(command, followUpBounds, toQuery, consistencyLevel, expiresAtNanos);
followUpRead.start(expiresAtNanos);
AsyncPromise<TrackedDataResponse> combinedRead = new AsyncPromise<>();
followUpRead.future().addCallback((result, failure) -> {
if (failure != null)
{
combinedRead.tryFailure(failure);
return;
}
try
{
combinedRead.trySuccess(TrackedDataResponse.merge(initialResponse, result));
}
catch (Throwable t)
{
combinedRead.tryFailure(t);
}
});
return combinedRead;
}
@Override
public void close()
{
iterator.close();
}
}

177
src/java/org/apache/cassandra/service/reads/tracked/FilteredFollowupRead.java Normal file
View File

@ -0,0 +1,177 @@
/*
* 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.reads.tracked;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.SortedMap;
import java.util.concurrent.atomic.AtomicReference;
import com.google.common.collect.Iterators;
import com.google.common.collect.PeekingIterator;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.PartitionRangeReadCommand;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.service.reads.tracked.PartialTrackedRangeRead.FollowUpReadInfo;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
import static org.apache.cassandra.service.reads.tracked.ExtendingCompletedRead.followUpReadRequired;
import static org.apache.cassandra.service.reads.tracked.ExtendingCompletedRead.toQuery;
import static org.apache.cassandra.service.reads.tracked.PartialTrackedRangeRead.makeFollowUpRead;
class FilteredFollowupRead extends AsyncPromise<TrackedDataResponse>
{
private final TrackedDataResponse initialResponse;
private final int toQuery;
private final ConsistencyLevel consistencyLevel;
private final long expiresAtNanos;
private final SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo;
private final PartitionRangeReadCommand command;
private final AbstractBounds<PartitionPosition> followUpBounds;
private final DecoratedKey finalKey;
public FilteredFollowupRead(TrackedDataResponse initialResponse,
int toQuery,
ConsistencyLevel consistencyLevel,
long expiresAtNanos,
SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo,
PartitionRangeReadCommand command,
AbstractBounds<PartitionPosition> followUpBounds,
DecoratedKey finalKey)
{
this.initialResponse = initialResponse;
this.toQuery = toQuery;
this.consistencyLevel = consistencyLevel;
this.expiresAtNanos = expiresAtNanos;
this.followUpReadInfo = followUpReadInfo;
this.command = command;
this.followUpBounds = followUpBounds;
this.finalKey = finalKey;
}
private boolean interleavesWithOriginal(DecoratedKey key)
{
if (finalKey == null)
return false;
return key.compareTo(finalKey) < 0;
}
public void start()
{
ClusterMetadata metadata = ClusterMetadata.current();
List<Future<TrackedDataResponse>> futures = new ArrayList<>();
int remaining = toQuery;
PeekingIterator<DecoratedKey> followUpKeys = Iterators.peekingIterator(followUpReadInfo.keySet().iterator());
// query all keys that interleave with the range of keys from the original range read
while (followUpKeys.hasNext() && (remaining > 0 || interleavesWithOriginal(followUpKeys.peek())))
{
DecoratedKey key = followUpKeys.next();
FollowUpReadInfo info = followUpReadInfo.get(key);
remaining -= info.potentialMatches;
SinglePartitionReadCommand cmd = SinglePartitionReadCommand.fromRangeRead(key, command, command.limits().forShortReadRetry(toQuery));
TrackedRead.Partition read = TrackedRead.Partition.create(metadata, cmd, consistencyLevel);
read.start(expiresAtNanos);
futures.add(read.future());
}
SortedMap<DecoratedKey, FollowUpReadInfo> nextKeys = followUpKeys.hasNext() ? followUpReadInfo.tailMap(followUpKeys.next()) : Collections.emptySortedMap();
AtomicReference<PartialTrackedRead> partialRead;
if (remaining > 0)
{
partialRead = new AtomicReference<>();
TrackedRead.Range rangeRead = makeFollowUpRead(command, followUpBounds, remaining, consistencyLevel, expiresAtNanos);
rangeRead.startLocal(expiresAtNanos, partialRead::set);
futures.add(rangeRead.future());
}
else
{
partialRead = null;
}
FutureCombiner.allOf(futures).addCallback((responses, error) -> {
if (error != null)
{
tryFailure(error);
return;
}
try
{
List<TrackedDataResponse> allResponses = new ArrayList<>(responses);
allResponses.add(initialResponse);
TrackedDataResponse merged = TrackedDataResponse.merge(allResponses);
DataLimits.Counter mergedResultCounter = command.limits().newCounter(command.nowInSec(),
true,
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness());
boolean partitionsFetched;
boolean initialIteratorExhausted;
TrackedDataResponse response;
try (PartitionIterator iterator = merged.makeIteratorUnlimited(command))
{
partitionsFetched = iterator.hasNext();
response = TrackedDataResponse.create(mergedResultCounter.applyTo(iterator), command.columnFilter());
initialIteratorExhausted = iterator.hasNext();
}
// although we check for interleaved keys in the initial read, we always query for them in the follow up, so
// we just use normal short read protection checks here
if (followUpReadRequired(command, mergedResultCounter, initialIteratorExhausted, partitionsFetched))
{
AbstractBounds<PartitionPosition> nextBounds = this.followUpBounds;
if (partialRead != null)
{
PartialTrackedRangeRead followUpRangeRead = (PartialTrackedRangeRead) partialRead.get();
nextBounds = followUpRangeRead.followUpBounds();
}
FilteredFollowupRead followUp = new FilteredFollowupRead(response, toQuery(command, mergedResultCounter), consistencyLevel, expiresAtNanos, nextKeys, command, nextBounds, null);
followUp.start();
followUp.addCallback((result, failure) -> {
if (failure != null)
tryFailure(failure);
else
trySuccess(result);
});
}
else
{
trySuccess(response);
}
}
catch (Throwable t)
{
tryFailure(t);
}
});
}
}

682
src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRangeRead.java
View File

@ -18,7 +18,9 @@
package org.apache.cassandra.service.reads.tracked;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
@ -36,15 +38,13 @@ import org.apache.cassandra.db.PartitionRangeReadCommand;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.partitions.AbstractBTreePartition;
import org.apache.cassandra.db.partitions.AbstractUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.SimpleBTreePartition;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.transform.EmptyPartitionsDiscarder;
import org.apache.cassandra.db.transform.Transformation;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.ExcludingBounds;
@ -54,39 +54,45 @@ import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.locator.ReplicaPlans;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Future;
public class PartialTrackedRangeRead extends AbstractPartialTrackedRead
public abstract class PartialTrackedRangeRead extends AbstractPartialTrackedRead
{
private static final Logger logger = LoggerFactory.getLogger(PartialTrackedRangeRead.class);
private final PartitionRangeReadCommand command;
private final SortedMap<DecoratedKey, SimpleBTreePartition> data = new TreeMap<>();
private final UnfilteredPartitionIterator initialData;
private final boolean enforceStrictLiveness;
// short read support
private DecoratedKey lastPartitionKey; // key of the last observed partition
private boolean partitionsFetched; // whether we've seen any new partitions since iteration start or last moreContents() call
private boolean initialIteratorExhausted;
private boolean wasAugmented;
AbstractBounds<PartitionPosition> followUpBounds;
public PartialTrackedRangeRead(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command, UnfilteredPartitionIterator initialData)
protected static class FollowUpReadInfo
{
super(executionController, searcher, cfs, startTimeNanos);
int potentialMatches = 0;
}
protected final PartitionRangeReadCommand command;
private PartialTrackedRangeRead(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command)
{
super(executionController, cfs, startTimeNanos);
this.command = command;
this.initialData = initialData;
this.enforceStrictLiveness = command.metadata().enforceStrictLiveness();
}
public static PartialTrackedRangeRead create(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command, UnfilteredPartitionIterator initialData)
{
PartialTrackedRangeRead read = new PartialTrackedRangeRead(executionController, searcher, cfs, startTimeNanos, command, initialData);
RowFilter rowFilter = command.rowFilter();
PartialTrackedRangeRead read;
if (searcher != null)
{
throw new UnsupportedOperationException("TODO: CASSANDRA-20374");
}
else if (!rowFilter.isEmpty())
{
read = new PartialTrackedRangeRead.Filtered(executionController, cfs, startTimeNanos, command);
}
else
{
read = new PartialTrackedRangeRead.Simple(executionController, cfs, startTimeNanos, command);
}
try
{
read.prepare();
read.prepare(initialData);
return read;
}
catch (Throwable e)
@ -102,6 +108,227 @@ public class PartialTrackedRangeRead extends AbstractPartialTrackedRead
return command;
}
protected static class ShortReadSupport
{
final DecoratedKey lastPartitionKey; // key of the last observed partition
final boolean partitionsFetched; // whether we've seen any new partitions since iteration start or last moreContents() call
final boolean initialIteratorExhausted;
final AbstractBounds<PartitionPosition> followUpBounds;
boolean wasAugmented;
ShortReadSupport(Builder builder, boolean initialIteratorExhausted, AbstractBounds<PartitionPosition> followUpBounds)
{
this.lastPartitionKey = builder.lastPartitionKey;
this.partitionsFetched = builder.partitionsFetched;
this.initialIteratorExhausted = initialIteratorExhausted;
this.followUpBounds = followUpBounds;
this.wasAugmented = false;
}
protected static class Builder
{
final ReadCommand command;
final DataLimits.Counter counter;
DecoratedKey lastPartitionKey; // key of the last observed partition
boolean partitionsFetched; // whether we've seen any new partitions since iteration start or last moreContents() call
protected Builder(ReadCommand command)
{
this.command = command;
counter = command.limits().newCounter(command.nowInSec(),
false,
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness());
}
ShortReadSupport build()
{
boolean initialIteratorExhausted = command.limits().isExhausted(counter);
AbstractBounds<PartitionPosition> followUpBounds = null;
if (partitionsFetched)
{
AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
followUpBounds = bounds.inclusiveRight()
? new Range<>(lastPartitionKey, bounds.right)
: new ExcludingBounds<>(lastPartitionKey, bounds.right);
Preconditions.checkState(!followUpBounds.contains(lastPartitionKey));
}
return new ShortReadSupport(this, initialIteratorExhausted, followUpBounds);
}
}
}
private abstract class Materializer extends Transformation<UnfilteredRowIterator>
{
final SortedMap<DecoratedKey, SimpleBTreePartition> data = new TreeMap<>();
final ShortReadSupport.Builder shortReadSupport;
private Materializer(ReadCommand command)
{
this.shortReadSupport = new ShortReadSupport.Builder(command);
}
abstract UnfilteredPartitionIterator filter(UnfilteredPartitionIterator iterator);
abstract RangePrepared createRangePrepared();
RangePrepared materialize(UnfilteredPartitionIterator inputIterator)
{
try
{
UnfilteredPartitionIterator materialized = Transformation.apply(inputIterator, new Transformation<UnfilteredRowIterator>()
{
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
SimpleBTreePartition materialized = data.computeIfAbsent(partition.partitionKey(), key -> new SimpleBTreePartition(key, partition.metadata(), UpdateTransaction.NO_OP));
materialized.update(PartitionUpdate.fromIterator(partition, command.columnFilter()));
shortReadSupport.lastPartitionKey = partition.partitionKey();
shortReadSupport.partitionsFetched = true;
return queryPartition(materialized);
}
});
UnfilteredPartitionIterator filtered = filter(materialized);
try (UnfilteredPartitionIterator iterator = shortReadSupport.counter.applyTo(filtered))
{
consume(iterator);
}
return createRangePrepared();
}
finally
{
inputIterator.close();
}
}
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
SimpleBTreePartition materialized = data.computeIfAbsent(partition.partitionKey(), key -> new SimpleBTreePartition(key, partition.metadata(), UpdateTransaction.NO_OP));
materialized.update(PartitionUpdate.fromIterator(partition, command.columnFilter()));
shortReadSupport.lastPartitionKey = partition.partitionKey();
shortReadSupport.partitionsFetched = true;
return queryPartition(materialized);
}
}
protected abstract class RangePrepared extends Prepared
{
protected final SortedMap<DecoratedKey, SimpleBTreePartition> data;
protected final ShortReadSupport shortReadSupport;
protected boolean wasAugmented;
public RangePrepared(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport)
{
this.data = data;
this.shortReadSupport = shortReadSupport;
}
protected boolean canAcceptUpdate(PartitionUpdate update)
{
return shortReadSupport.initialIteratorExhausted || !shortReadSupport.followUpBounds.contains(update.partitionKey());
}
private SimpleBTreePartition augmentResponseInternal(PartitionUpdate update)
{
SimpleBTreePartition partition = data.computeIfAbsent(update.partitionKey(), key -> new SimpleBTreePartition(key, update.metadata(), UpdateTransaction.NO_OP));
partition.update(update);
return partition;
}
@Override
public State augment(PartitionUpdate update)
{
// if the input iterator reached the row limit, then we can't apply any augmenting mutations that are past
// the last materialized key. Since we wouldn't have materialized the local data for that key, applying an
// update would cause us to return incomplete data for it.
if (canAcceptUpdate(update))
{
logger.trace("Augmented partition {} for read {}", update.partitionKey(), PartialTrackedRangeRead.this);
augmentResponseInternal(update);
}
else
{
logger.trace("Ignoring unacceptable update from key {} on read {}", update.partitionKey(), PartialTrackedRangeRead.this);
}
wasAugmented = true;
return this;
}
}
protected abstract class RangeCompleted extends Completed
{
protected final SortedMap<DecoratedKey, SimpleBTreePartition> data;
protected final ShortReadSupport shortReadSupport;
protected final boolean wasAugmented;
public RangeCompleted(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport, boolean wasAugmented)
{
this.data = data;
this.shortReadSupport = shortReadSupport;
this.wasAugmented = wasAugmented;
}
@Override
protected UnfilteredPartitionIterator iterator()
{
Iterator<SimpleBTreePartition> iterator = data.values().iterator();
return new AbstractUnfilteredPartitionIterator()
{
@Override
public TableMetadata metadata()
{
return command.metadata();
}
@Override
public boolean hasNext()
{
return iterator.hasNext();
}
@Override
public UnfilteredRowIterator next()
{
return queryPartition(iterator.next());
}
};
}
protected abstract CompletedRead extendRead(UnfilteredPartitionIterator iterator);
@Override
protected CompletedRead createResult(UnfilteredPartitionIterator iterator)
{
if (wasAugmented)
return extendRead(iterator);
return CompletedRead.simple(iterator, command);
}
AbstractBounds<PartitionPosition> followUpBounds()
{
return shortReadSupport.followUpBounds;
}
}
abstract Materializer createMaterializer();
@Override
protected Prepared prepareInternal(UnfilteredPartitionIterator initialData)
{
// memtable contents are frozen at read completion time, when the iterator is evaluated, not at the beginning
// of the read, when references to memtables and sstables are collected. Because of this, replica coordinated
// reads can cause read monotonicity to be broken by returning data that hasn't been replicated to at least
// CL other nodes via reconciliation. To prevent this, the contents of the initial iterator are materialized
// onto heap at partition granularity until the limits of the read are reached.
Materializer materializer = createMaterializer();
return materializer.materialize(initialData);
}
UnfilteredRowIterator queryPartition(AbstractBTreePartition partition)
{
return partition.unfilteredIterator(command.columnFilter(),
@ -121,227 +348,264 @@ public class PartialTrackedRangeRead extends AbstractPartialTrackedRead
}
}
@Override
void freezeInitialData()
public AbstractBounds<PartitionPosition> followUpBounds()
{
// memtable contents are frozen at read completion time, when the iterator is evaluated, not at the beginning
// of the read, when references to memtables and sstables are collected. Because of this, replica coordinated
// reads can cause read monotonicity to be broken by returning data that hasn't been replicated to at least
// CL other nodes via reconciliation. To prevent this, the contents of the initial iterator are materialized
// onto heap at partition granularity until the limits of the read are reached.
RangeCompleted completed = (RangeCompleted) state().asCompleted();
return completed.followUpBounds();
}
UnfilteredPartitionIterator materializer = new AbstractUnfilteredPartitionIterator()
protected static TrackedRead.Range makeFollowUpRead(PartitionRangeReadCommand command, AbstractBounds<PartitionPosition> followUpBounds, int toQuery, ConsistencyLevel consistencyLevel, long expiresAtNanos)
{
DataLimits newLimits = command.limits().forShortReadRetry(toQuery);
DataRange newDataRange = command.dataRange().forSubRange(followUpBounds);
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
PartitionRangeReadCommand followUpCmd = command.withUpdatedLimitsAndDataRange(newLimits, newDataRange);
ReplicaPlan.ForRangeRead replicaPlan = ReplicaPlans.forRangeRead(keyspace,
followUpCmd.indexQueryPlan(),
consistencyLevel,
followUpCmd.dataRange().keyRange(),
1);
TrackedRead.Range read = TrackedRead.Range.create(followUpCmd, replicaPlan);
logger.trace("Short read detected, starting followup read {}", read);
return read;
}
static class Simple extends PartialTrackedRangeRead
{
private class SimplePrepared extends RangePrepared
{
@Override
public TableMetadata metadata()
public SimplePrepared(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport)
{
return initialData.metadata();
super(data, shortReadSupport);
}
@Override
public boolean hasNext()
Completed complete()
{
return initialData.hasNext();
return new SimpleCompleted(data, shortReadSupport, wasAugmented);
}
}
protected class SimpleCompleted extends RangeCompleted
{
public SimpleCompleted(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport, boolean wasAugmented)
{
super(data, shortReadSupport, wasAugmented);
}
@Override
public UnfilteredRowIterator next()
protected CompletedRead extendRead(UnfilteredPartitionIterator iterator)
{
try (UnfilteredRowIterator rowIterator = initialData.next())
return new ExtendingCompletedRead(command, iterator, shortReadSupport.partitionsFetched, shortReadSupport.initialIteratorExhausted, shortReadSupport.followUpBounds);
}
}
public Simple(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command)
{
super(executionController, cfs, startTimeNanos, command);
}
@Override
Materializer createMaterializer()
{
return new Materializer(command)
{
@Override
UnfilteredPartitionIterator filter(UnfilteredPartitionIterator iterator)
{
SimpleBTreePartition partition = augmentResponseInternal(PartitionUpdate.fromIterator(rowIterator, command.columnFilter()));
lastPartitionKey = partition.partitionKey();
partitionsFetched = true;
return queryPartition(partition);
Preconditions.checkState(command().rowFilter().isEmpty());
return iterator;
}
}
@Override
public void close()
{
super.close();
initialData.close();
}
};
// unmerged per-source counter
final DataLimits.Counter singleResultCounter = command.limits().newCounter(command.nowInSec(),
false,
command.selectsFullPartition(),
enforceStrictLiveness);
try (UnfilteredPartitionIterator iterator = singleResultCounter.applyTo(materializer))
{
consume(iterator);
}
initialIteratorExhausted = command.limits().isExhausted(singleResultCounter);
if (partitionsFetched)
{
AbstractBounds<PartitionPosition> bounds = command.dataRange().keyRange();
followUpBounds = bounds.inclusiveRight()
? new Range<>(lastPartitionKey, bounds.right)
: new ExcludingBounds<>(lastPartitionKey, bounds.right);
Preconditions.checkState(!followUpBounds.contains(lastPartitionKey));
}
wasAugmented = false;
}
@Override
UnfilteredPartitionIterator initialData()
{
Iterator<SimpleBTreePartition> iterator = data.values().iterator();
return new AbstractUnfilteredPartitionIterator()
{
@Override
public TableMetadata metadata()
{
return command.metadata();
}
@Override
public boolean hasNext()
{
return iterator.hasNext();
}
@Override
public UnfilteredRowIterator next()
{
return queryPartition(iterator.next());
}
};
}
@Override
UnfilteredPartitionIterator augmentedData()
{
return null;
}
private SimpleBTreePartition augmentResponseInternal(PartitionUpdate update)
{
SimpleBTreePartition partition = data.computeIfAbsent(update.partitionKey(), key -> new SimpleBTreePartition(key, update.metadata(), UpdateTransaction.NO_OP));
partition.update(update);
return partition;
}
@Override
void augmentResponse(PartitionUpdate update)
{
// if the input iterator reached the row limit, then we can't apply any augmenting mutations that are past
// the last materialized key. Since we wouldn't have materialized the local data for that key, applying an
// update would cause us to return incomplete data for it.
if (initialIteratorExhausted || !followUpBounds.contains(update.partitionKey()))
augmentResponseInternal(update);
wasAugmented = true;
}
private class ExtendingCompletedRead implements CompletedRead
{
final UnfilteredPartitionIterator iterator;
// merged end-result counter
final DataLimits.Counter mergedResultCounter = command.limits().newCounter(command.nowInSec(),
true,
command.selectsFullPartition(),
enforceStrictLiveness);
public ExtendingCompletedRead(UnfilteredPartitionIterator iterator)
{
this.iterator = iterator;
@Override
RangePrepared createRangePrepared()
{
return new SimplePrepared(data, shortReadSupport.build());
}
};
}
@Override
public PartitionIterator iterator()
public Index.Searcher searcher()
{
PartitionIterator filtered = UnfilteredPartitionIterators.filter(iterator, command.nowInSec());
PartitionIterator counted = Transformation.apply(filtered, mergedResultCounter);
return Transformation.apply(counted, new EmptyPartitionsDiscarder());
return null;
}
}
/**
* Since ALLOW FILTERING reads can cover a lot of partitions without returning much data, we don't want to eagerly
* materialize partitions onto the heap and keep them there. So this filters out non-matching partitions from the
* freezeInitialData phase. However, if reconciliation receives a mutation that applies to a previously discarded
* partition AND the contents of that mutation matches the row filter, we also need to retry the read against that
* partition so we don't return incomplete data. This class handles both jobs
*/
static class Filtered extends PartialTrackedRangeRead
{
protected class FilteredPrepared extends RangePrepared
{
private final Set<DecoratedKey> filteredKeys;
private final SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo = new TreeMap<>();
private final RowFilter.RowFilterTransformation filter;
public FilteredPrepared(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport, Set<DecoratedKey> filteredKeys, RowFilter.RowFilterTransformation filter)
{
super(data, shortReadSupport);
this.filteredKeys = filteredKeys;
this.filter = filter;
}
@Override
protected boolean canAcceptUpdate(PartitionUpdate update)
{
DecoratedKey key = update.partitionKey();
if (filteredKeys.contains(key))
{
int matches = filter.potentialMatches(update);
if (matches > 0)
{
FollowUpReadInfo info = followUpReadInfo.computeIfAbsent(key, k -> new FollowUpReadInfo());
info.potentialMatches += matches;
}
logger.trace("Not applying update for previously filtered partition: {}", update.partitionKey());
return false;
}
return super.canAcceptUpdate(update);
}
@Override
Completed complete()
{
return new FilteredCompleted(data, shortReadSupport, wasAugmented, followUpReadInfo);
}
}
@Override
public TrackedRead<?, ?> followupRead(ConsistencyLevel consistencyLevel, long expiresAtNanos)
protected class FilteredMaterializer extends Materializer
{
// never try to request additional partitions from replicas if our reconciled partitions are already filled to the limit
if (mergedResultCounter.isDone())
return null;
private final Set<DecoratedKey> filteredKeys = new HashSet<>();
private final RowFilter.RowFilterTransformation filter;
public FilteredMaterializer(ReadCommand command)
{
super(command);
filter = command.rowFilter().filter(command().metadata(), command().nowInSec());
}
// we do not apply short read protection when we have no limits at all
if (command.limits().isUnlimited())
return null;
@Override
UnfilteredPartitionIterator filter(UnfilteredPartitionIterator iterator)
{
return Transformation.apply(iterator, new Transformation<>()
{
@Override
protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
{
if (Transformation.apply(partition, filter).isEmpty())
{
DecoratedKey key = partition.partitionKey();
data.remove(key);
filteredKeys.add(key);
partition.close();
return null;
}
return partition;
}
});
}
/*
* If this is a single partition read command or an (indexed) partition range read command with
* a partition key specified, then we can't and shouldn't try fetch more partitions.
*/
if (command.isLimitedToOnePartition())
return null;
/*
* If the returned result doesn't have enough rows/partitions to satisfy even the original limit, don't ask for more.
*
* Can only take the short cut if there is no per partition limit set. Otherwise it's possible to hit false
* positives due to some rows being uncounted for in certain scenarios (see CASSANDRA-13911).
*/
if (initialIteratorExhausted && command.limits().perPartitionCount() == DataLimits.NO_LIMIT)
return null;
/*
* Either we had an empty iterator as the initial response, or our moreContents() call got us an empty iterator.
* There is no point to ask the replica for more rows - it has no more in the requested range.
*/
if (!partitionsFetched)
return null;
partitionsFetched = false;
/*
* We are going to fetch one partition at a time for thrift and potentially more for CQL.
* The row limit will either be set to the per partition limit - if the command has no total row limit set, or
* the total # of rows remaining - if it has some. If we don't grab enough rows in some of the partitions,
* then future ShortReadRowsProtection.moreContents() calls will fetch the missing ones.
*/
int toQuery = command.limits().count() != DataLimits.NO_LIMIT
? command.limits().count() - mergedResultCounter.rowsCounted()
: command.limits().perPartitionCount();
ColumnFamilyStore.metricsFor(command.metadata().id).shortReadProtectionRequests.mark();
Tracing.trace("Requesting {} extra rows from {} for short read protection", toQuery, FBUtilities.getBroadcastAddressAndPort());
logger.info("Requesting {} extra rows from {} for short read protection", toQuery, FBUtilities.getBroadcastAddressAndPort());
return makeFollowupRead(toQuery, consistencyLevel, expiresAtNanos);
@Override
RangePrepared createRangePrepared()
{
return new FilteredPrepared(data, shortReadSupport.build(), filteredKeys, filter);
}
}
private TrackedRead<?, ?> makeFollowupRead(int toQuery, ConsistencyLevel consistencyLevel, long expiresAtNanos)
static class FilteredCompletedRead extends ExtendingCompletedRead
{
DataLimits newLimits = command.limits().forShortReadRetry(toQuery);
private final DecoratedKey lastMatchingKey;
private final SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo;
public FilteredCompletedRead(PartitionRangeReadCommand command, UnfilteredPartitionIterator iterator, ShortReadSupport shortReadSupport, DecoratedKey lastMatchingKey, SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo)
{
super(command, iterator, shortReadSupport.partitionsFetched, shortReadSupport.initialIteratorExhausted, shortReadSupport.followUpBounds);
this.lastMatchingKey = lastMatchingKey;
this.followUpReadInfo = followUpReadInfo;
}
DataRange newDataRange = command.dataRange().forSubRange(followUpBounds);
/**
* Even if we reached the limit during materialization, if there are keys ahead of the first materialized key
* or interleaved with them, then we need to read them
* @return
*/
private boolean hasInterleavedFollowupKeys()
{
if (followUpReadInfo.isEmpty())
return false;
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
PartitionRangeReadCommand followUpCmd = command.withUpdatedLimitsAndDataRange(newLimits, newDataRange);
ReplicaPlan.ForRangeRead replicaPlan = ReplicaPlans.forRangeRead(keyspace,
followUpCmd.indexQueryPlan(),
if (lastMatchingKey == null) // null means there was no data and therefore no interleaving
return true;
return followUpReadInfo.firstKey().compareTo(lastMatchingKey) < 0;
}
@Override
protected boolean followUpRequired()
{
return hasInterleavedFollowupKeys() || super.followUpRequired();
}
@Override
protected Future<TrackedDataResponse> makeFollowupRead(TrackedDataResponse initialResponse, int toQuery, ConsistencyLevel consistencyLevel, long expiresAtNanos)
{
if (followUpReadInfo.isEmpty())
return super.makeFollowupRead(initialResponse, toQuery, consistencyLevel, expiresAtNanos);
FilteredFollowupRead followupRead = new FilteredFollowupRead(initialResponse,
toQuery,
consistencyLevel,
followUpCmd.dataRange().keyRange(),
1);
expiresAtNanos,
followUpReadInfo,
command,
followUpBounds,
lastMatchingKey);
TrackedRead.Range read = TrackedRead.Range.create(followUpCmd, replicaPlan);
logger.trace("Short read detected, starting followup read {}", read);
read.start(expiresAtNanos);
return read;
followupRead.start();
return followupRead;
}
}
private class FilteredCompleted extends RangeCompleted
{
private final SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo;
public FilteredCompleted(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport, boolean wasAugmented, SortedMap<DecoratedKey, FollowUpReadInfo> followUpReadInfo)
{
super(data, shortReadSupport, wasAugmented);
this.followUpReadInfo = followUpReadInfo;
}
@Override
protected CompletedRead extendRead(UnfilteredPartitionIterator iterator)
{
return new FilteredCompletedRead(command, iterator, shortReadSupport, data.isEmpty() ? data.lastKey() : null, followUpReadInfo);
}
}
public Filtered(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command)
{
super(executionController, cfs, startTimeNanos, command);
}
@Override
public void close()
Materializer createMaterializer()
{
iterator.close();
return new FilteredMaterializer(command);
}
}
@Override
CompletedRead createResult(UnfilteredPartitionIterator iterator)
{
if (wasAugmented)
return new ExtendingCompletedRead(iterator);
return CompletedRead.simple(iterator, command().nowInSec());
@Override
public Index.Searcher searcher()
{
return null;
}
}
}

25
src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
View File

@ -25,33 +25,46 @@ import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.utils.concurrent.Future;
public interface PartialTrackedRead
{
interface CompletedRead extends AutoCloseable
{
PartitionIterator iterator();
TrackedRead<?, ?> followupRead(ConsistencyLevel consistencyLevel, long expiresAtNanos);
TrackedDataResponse response(); // must be called from the read stage
Future<TrackedDataResponse> followupRead(TrackedDataResponse initialResponse, ConsistencyLevel consistencyLevel, long expiresAtNanos);
@Override
void close();
static CompletedRead simple(UnfilteredPartitionIterator partition, long nowInSec)
static TrackedDataResponse createResponse(UnfilteredPartitionIterator partition, ReadCommand command)
{
PartitionIterator iterator = UnfilteredPartitionIterators.filter(partition, command.nowInSec());
DataLimits.Counter counter = command.limits().newCounter(command.nowInSec(),
false,
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness()).onlyCount();
return TrackedDataResponse.create(counter.applyTo(iterator),
command.columnFilter());
}
static CompletedRead simple(UnfilteredPartitionIterator partition, ReadCommand command)
{
return new CompletedRead()
{
@Override
public PartitionIterator iterator()
public TrackedDataResponse response()
{
return UnfilteredPartitionIterators.filter(partition, nowInSec);
return createResponse(partition, command);
}
@Override
public TrackedRead<?, ?> followupRead(ConsistencyLevel consistencyLevel, long expiresAtNanos)
public Future<TrackedDataResponse> followupRead(TrackedDataResponse initialRead, ConsistencyLevel consistencyLevel, long expiresAtNanos)
{
return null;
}

130
src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedSinglePartitionRead.java
View File

@ -18,6 +18,8 @@
package org.apache.cassandra.service.reads.tracked;
import java.util.List;
import com.google.common.base.Preconditions;
import org.apache.cassandra.db.ColumnFamilyStore;
@ -29,29 +31,31 @@ import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.SimpleBTreePartition;
import org.apache.cassandra.db.partitions.SingletonUnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import static org.apache.cassandra.db.partitions.UnfilteredPartitionIterators.MergeListener.NOOP;
public class PartialTrackedSinglePartitionRead extends AbstractPartialTrackedRead
{
private final Index.Searcher searcher;
private final SinglePartitionReadCommand command;
private final UnfilteredPartitionIterator initialData;
private SimpleBTreePartition augmentedData;
public PartialTrackedSinglePartitionRead(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, SinglePartitionReadCommand command, UnfilteredPartitionIterator initialData)
public PartialTrackedSinglePartitionRead(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, SinglePartitionReadCommand command)
{
super(executionController, searcher, cfs, startTimeNanos);
super(executionController, cfs, startTimeNanos);
this.searcher = searcher;
this.command = command;
this.initialData = initialData;
}
public static PartialTrackedSinglePartitionRead create(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, SinglePartitionReadCommand command, UnfilteredPartitionIterator initialData)
{
PartialTrackedSinglePartitionRead read = new PartialTrackedSinglePartitionRead(executionController, searcher, cfs, startTimeNanos, command, initialData);
PartialTrackedSinglePartitionRead read = new PartialTrackedSinglePartitionRead(executionController, searcher, cfs, startTimeNanos, command);
try
{
read.prepare();
read.prepare(initialData);
return read;
}
catch (Throwable e)
@ -61,48 +65,88 @@ public class PartialTrackedSinglePartitionRead extends AbstractPartialTrackedRea
}
}
@Override
void freezeInitialData()
private class SinglePartitionPrepared extends Prepared
{
// the iterators from queryStorage grabs sstable references and a
// snapshot of the memtable partition so we don't need to do anything here
private final UnfilteredPartitionIterator initialData;
private SimpleBTreePartition augmentedData;
private SinglePartitionPrepared(UnfilteredPartitionIterator initialData)
{
this.initialData = initialData;
}
@Override
public State augment(PartitionUpdate update)
{
Preconditions.checkArgument(update.partitionKey().equals(command.partitionKey()));
if (augmentedData == null)
augmentedData = new SimpleBTreePartition(command.partitionKey(), command.metadata(), UpdateTransaction.NO_OP);
augmentedData.update(update);
return this;
}
@Override
Completed complete()
{
return new SinglePartitionCompleted(initialData, augmentedData);
}
}
private class SinglePartitionCompleted extends Completed
{
private final UnfilteredPartitionIterator initialData;
private final SimpleBTreePartition augmentedData;
public SinglePartitionCompleted(UnfilteredPartitionIterator initialData, SimpleBTreePartition augmentedData)
{
this.initialData = initialData;
this.augmentedData = augmentedData;
}
private UnfilteredPartitionIterator augmentedIterator()
{
if (augmentedData == null)
return null;
Slices slices = command.clusteringIndexFilter().getSlices(command.metadata());
UnfilteredRowIterator augmentedPartition = augmentedData.unfilteredIterator(command.columnFilter(), slices, command.clusteringIndexFilter().isReversed());
return new SingletonUnfilteredPartitionIterator(augmentedPartition);
}
@Override
protected UnfilteredPartitionIterator iterator()
{
UnfilteredPartitionIterator augmentedIterator = augmentedIterator();
if (augmentedIterator == null)
return initialData;
return UnfilteredPartitionIterators.merge(List.of(initialData, augmentedIterator), NOOP);
}
@Override
protected CompletedRead createResult(UnfilteredPartitionIterator iterator)
{
return CompletedRead.simple(iterator, command);
}
}
@Override
protected Prepared prepareInternal(UnfilteredPartitionIterator initialData)
{
return new SinglePartitionPrepared(initialData);
}
@Override
public Index.Searcher searcher()
{
return searcher;
}
// TODO: delete (almost?) ever
@Override
public ReadCommand command()
{
return command;
}
@Override
UnfilteredPartitionIterator initialData()
{
return initialData;
}
@Override
UnfilteredPartitionIterator augmentedData()
{
if (augmentedData == null)
return null;
Slices slices = command.clusteringIndexFilter().getSlices(command.metadata());
UnfilteredRowIterator augmented = augmentedData.unfilteredIterator(command.columnFilter(), slices, command.clusteringIndexFilter().isReversed());
return new SingletonUnfilteredPartitionIterator(augmented);
}
@Override
void augmentResponse(PartitionUpdate update)
{
Preconditions.checkArgument(update.partitionKey().equals(command.partitionKey()));
if (augmentedData == null)
augmentedData = new SimpleBTreePartition(command.partitionKey(), command.metadata(), UpdateTransaction.NO_OP);
augmentedData.update(update);
}
@Override
CompletedRead createResult(UnfilteredPartitionIterator iterator)
{
return CompletedRead.simple(iterator, command().nowInSec());
}
}

87
src/java/org/apache/cassandra/service/reads/tracked/TrackedDataResponse.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.service.reads.tracked;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterators;
import org.apache.cassandra.io.IVersionedSerializer;
@ -33,18 +34,63 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import com.google.common.base.Preconditions;
public class TrackedDataResponse
{
private final int serializationVersion;
private final ByteBuffer data;
private final List<ByteBuffer> data;
public TrackedDataResponse(int serializationVersion, ByteBuffer data)
{
this(serializationVersion, Collections.singletonList(data));
}
private TrackedDataResponse(int serializationVersion, List<ByteBuffer> data)
{
Preconditions.checkArgument(!data.isEmpty());
this.serializationVersion = serializationVersion;
this.data = data;
}
public TrackedDataResponse merge(TrackedDataResponse that)
{
return merge(this, that);
}
public static TrackedDataResponse merge(TrackedDataResponse l, TrackedDataResponse r)
{
Preconditions.checkArgument(l.serializationVersion == r.serializationVersion);
List<ByteBuffer> newData = new ArrayList<>(l.data.size() + r.data.size());
newData.addAll(l.data);
newData.addAll(r.data);
return new TrackedDataResponse(l.serializationVersion, newData);
}
public static TrackedDataResponse merge(List<TrackedDataResponse> responses)
{
Preconditions.checkArgument(!responses.isEmpty());
int version = responses.get(0).serializationVersion;
int size = responses.get(0).data.size();
for (int i=1,mi=responses.size(); i<mi; i++)
{
Preconditions.checkState(responses.get(i).serializationVersion == version);
size += responses.get(i).data.size();
}
List<ByteBuffer> newData = new ArrayList<>(size);
for (int i=0,mi=responses.size(); i<mi; i++)
newData.addAll(responses.get(i).data);
return new TrackedDataResponse(version, newData);
}
public static TrackedDataResponse create(PartitionIterator iter, ColumnFilter selection)
{
try (DataOutputBuffer buffer = new DataOutputBuffer())
@ -59,7 +105,7 @@ public class TrackedDataResponse
}
}
public PartitionIterator makeIterator(ReadCommand command)
private static PartitionIterator makeIterator(int serializationVersion, ByteBuffer data, ReadCommand command)
{
try (DataInputBuffer in = new DataInputBuffer(data, true))
{
@ -72,28 +118,55 @@ public class TrackedDataResponse
}
}
public static final IVersionedSerializer<TrackedDataResponse> serializer = new IVersionedSerializer<TrackedDataResponse>()
public PartitionIterator makeIteratorUnlimited(ReadCommand command)
{
if (data.size() == 1)
return makeIterator(serializationVersion, data.get(0), command);
List<PartitionIterator> iterators = new ArrayList<>(data.size());
for (ByteBuffer buffer : data)
iterators.add(makeIterator(serializationVersion, buffer, command));
return PartitionIterators.mergeNonOverlapping(iterators);
}
public PartitionIterator makeIterator(ReadCommand command)
{
DataLimits.Counter counter = command.limits().newCounter(command.nowInSec(),
true,
command.selectsFullPartition(),
command.metadata().enforceStrictLiveness());
return counter.applyTo(makeIteratorUnlimited(command));
}
public static final IVersionedSerializer<TrackedDataResponse> serializer = new IVersionedSerializer<>()
{
@Override
public void serialize(TrackedDataResponse response, DataOutputPlus out, int version) throws IOException
{
out.writeInt(response.serializationVersion);
ByteBufferUtil.writeWithVIntLength(response.data, out);
out.writeInt(response.data.size());
for (ByteBuffer buffer : response.data)
ByteBufferUtil.writeWithVIntLength(buffer, out);
}
@Override
public TrackedDataResponse deserialize(DataInputPlus in, int version) throws IOException
{
int serializationVersion = in.readInt();
ByteBuffer data = ByteBufferUtil.readWithVIntLength(in);
int size = in.readInt();
List<ByteBuffer> data = new ArrayList<>(size);
for (int i = 0; i < size; i++)
data.add(ByteBufferUtil.readWithVIntLength(in));
return new TrackedDataResponse(serializationVersion, data);
}
@Override
public long serializedSize(TrackedDataResponse response, int version)
{
return TypeSizes.sizeof(response.serializationVersion)
+ ByteBufferUtil.serializedSizeWithVIntLength(response.data);
long size = TypeSizes.sizeof(response.serializationVersion) + TypeSizes.sizeof(response.data.size());
for (ByteBuffer buffer : response.data)
size += ByteBufferUtil.serializedSizeWithVIntLength(buffer);
return size;
}
};
}

18
src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReadCoordinator.java
View File

@ -27,8 +27,6 @@ import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionIterators;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.ReplicaPlan;
import org.apache.cassandra.metrics.ReadRepairMetrics;
@ -45,6 +43,8 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.AbstractFuture;
import org.apache.cassandra.utils.concurrent.Accumulator;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -52,6 +52,7 @@ import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
public class TrackedLocalReadCoordinator
{
@ -512,7 +513,7 @@ public class TrackedLocalReadCoordinator
});
}
public void startLocalRead(TrackedRead.Id readId, ReadCommand command, ReplicaPlan.AbstractForRead<?, ?> replicaPlan, int[] summaryNodes, long expiresAtNanos)
public void startLocalRead(TrackedRead.Id readId, ReadCommand command, ReplicaPlan.AbstractForRead<?, ?> replicaPlan, int[] summaryNodes, long expiresAtNanos, Consumer<PartialTrackedRead> partialReadConsumer)
{
synchronized (this)
{
@ -528,6 +529,8 @@ public class TrackedLocalReadCoordinator
try
{
read = command.beginTrackedRead(controller);
if (partialReadConsumer != null)
partialReadConsumer.accept(read);
// Create another summary once initial data has been read fully. We do this to catch
// any mutations that may have arrived during initial read execution.
secondarySummary = command.createMutationSummary(true);
@ -555,20 +558,17 @@ public class TrackedLocalReadCoordinator
{
try (PartialTrackedRead.CompletedRead completedRead = read.complete())
{
TrackedDataResponse response = TrackedDataResponse.create(completedRead.iterator(), selection);
TrackedRead<?, ?> followUp = completedRead.followupRead(consistencyLevel, expiresAtNanos);
TrackedDataResponse response = completedRead.response();
Future<TrackedDataResponse> followUp = completedRead.followupRead(response, consistencyLevel, expiresAtNanos);
if (followUp != null)
{
ReadCommand command = read.command();
followUp.future().addCallback((iterator, error) -> {
followUp.addCallback((newResponse, error) -> {
if (error != null)
{
promise.tryFailure(error);
return;
}
PartitionIterator previous = response.makeIterator(command);
TrackedDataResponse newResponse = TrackedDataResponse.create(PartitionIterators.concat(List.of(previous, iterator)), selection);
promise.trySuccess(newResponse);
});
}

5
src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
View File

@ -36,6 +36,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@ -89,7 +90,7 @@ public class TrackedLocalReads implements Shutdownable
getOrCreate(summary.readId()).receiveSummary(from, summary.summary());
}
public TrackedLocalReadCoordinator beginRead(TrackedRead.Id readId, ClusterMetadata metadata, ReadCommand command, ConsistencyLevel consistencyLevel, int[] summaryNodes, long expiresAtNanos)
public TrackedLocalReadCoordinator beginRead(TrackedRead.Id readId, ClusterMetadata metadata, ReadCommand command, ConsistencyLevel consistencyLevel, int[] summaryNodes, long expiresAtNanos, Consumer<PartialTrackedRead> partialReadConsumer)
{
Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id);
@ -116,7 +117,7 @@ public class TrackedLocalReads implements Shutdownable
// TODO: confirm all summaryNodes are present in the replica plan
TrackedLocalReadCoordinator coordinator = getOrCreate(readId);
coordinator.startLocalRead(readId, command, replicaPlan, summaryNodes, expiresAtNanos);
coordinator.startLocalRead(readId, command, replicaPlan, summaryNodes, expiresAtNanos, partialReadConsumer);
return coordinator;
}

35
src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
View File

@ -53,6 +53,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -133,7 +134,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
}
}
private final AsyncPromise<PartitionIterator> future = new AsyncPromise<>();
private final AsyncPromise<TrackedDataResponse> future = new AsyncPromise<>();
private final Id readId = Id.nextId();
private final ReadCommand command;
@ -172,6 +173,11 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
protected abstract Verb verb();
public boolean intersects(DecoratedKey key)
{
return command.dataRange().contains(key);
}
public static class Partition extends TrackedRead<EndpointsForToken, ReplicaPlan.ForTokenRead>
{
private Partition(SinglePartitionReadCommand command, ReplicaPlan.AbstractForRead<EndpointsForToken, ReplicaPlan.ForTokenRead> replicaPlan, ConsistencyLevel consistencyLevel)
@ -233,7 +239,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
return hostids;
}
public void start(long expiresAt)
private void start(long expiresAt, Consumer<PartialTrackedRead> partialReadConsumer)
{
// TODO: skip local coordination if this node knows its recovering from an outage
// TODO: read speculation
@ -256,8 +262,9 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
if (dataNode == localReplica)
{
logger.trace("Locally coordinating {}", readId);
Stage.READ.submit(() -> {
TrackedLocalReadCoordinator coordinator = MutationTrackingService.instance.localReads().beginRead(readId, ClusterMetadata.current(), command, consistencyLevel, summaryHostIds, expiresAt);
TrackedLocalReadCoordinator coordinator = MutationTrackingService.instance.localReads().beginRead(readId, ClusterMetadata.current(), command, consistencyLevel, summaryHostIds, expiresAt, partialReadConsumer);
coordinator.addCallback((response, error) -> {
if (error != null)
{
@ -272,6 +279,8 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
}
else
{
logger.trace("Sending data request for {} to {}", readId, dataNode.endpoint());
Preconditions.checkArgument(partialReadConsumer == null, "Cannot supply read consumer for nonlocal reads");
DataRequest dataRequest = new DataRequest(readId, command, consistencyLevel, summaryHostIds);
Message<DataRequest> dataMessage = Message.outWithFlag(verb(), dataRequest, MessageFlag.CALL_BACK_ON_FAILURE);
MessagingService.instance().sendWithCallback(dataMessage, dataNode.endpoint(), this);
@ -289,15 +298,27 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
{
if (localReplica == replica)
{
logger.trace("Locally processing summary request for {}", readId);
Stage.READ.submit(() -> summaryRequest.executeLocally(summaryMessage, ClusterMetadata.current()));
}
else
{
logger.trace("Sending summary request for {} to {}", readId, replica.endpoint());
MessagingService.instance().send(summaryMessage, replica.endpoint());
}
}
}
public void start(long expiresAt)
{
start(expiresAt, null);
}
public void startLocal(long expiresAt, Consumer<PartialTrackedRead> partialReadConsumer)
{
start(expiresAt, partialReadConsumer);
}
public void start(Dispatcher.RequestTime requestTime)
{
start(requestTime.computeDeadline(verb().expiresAfterNanos()));
@ -305,7 +326,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
private void onResponse(TrackedDataResponse response)
{
future.trySuccess(response.makeIterator(command));
future.trySuccess(response);
}
@Override
@ -326,7 +347,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
return true;
}
public Future<PartitionIterator> future()
public Future<TrackedDataResponse> future()
{
return future;
}
@ -335,7 +356,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
{
try
{
return future.get(command.getTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
return future.get(command.getTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS).makeIterator(command);
}
catch (InterruptedException e)
{
@ -413,7 +434,7 @@ public abstract class TrackedRead<E extends Endpoints<E>, P extends ReplicaPlan.
@Override
public void executeLocally(Message<? extends Request> message, ClusterMetadata metadata)
{
TrackedLocalReadCoordinator coordinator = MutationTrackingService.instance.localReads().beginRead(readId, metadata, command, consistencyLevel, summaryNodes, message.expiresAtNanos());
TrackedLocalReadCoordinator coordinator = MutationTrackingService.instance.localReads().beginRead(readId, metadata, command, consistencyLevel, summaryNodes, message.expiresAtNanos(), null);
coordinator.addCallback((response, error) -> {
if (error != null)
{

8
test/conf/logback-dtest.xml
View File

@ -45,9 +45,9 @@
<encoder>
<pattern>%-5level [%thread] ${instance_id} %date{"yyyy-MM-dd'T'HH:mm:ss,SSS", UTC} %F:%L - %msg%n</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>DEBUG</level>
</filter>
<!-- <filter class="ch.qos.logback.classic.filter.ThresholdFilter">-->
<!-- <level>DEBUG</level>-->
<!-- </filter>-->
</appender>
<logger name="io.netty.*" level="WARN"/>
@ -63,4 +63,6 @@
<appender-ref ref="INSTANCESTDERR" />
<appender-ref ref="INSTANCESTDOUT" />
</root>
<logger name="org.apache.cassandra.replication" level="TRACE"/>
<logger name="org.apache.cassandra.service.reads.tracked" level="TRACE"/>
</configuration>

123
test/distributed/org/apache/cassandra/distributed/test/ReadRepairRangeQueriesTest.java
View File

@ -18,9 +18,15 @@
package org.apache.cassandra.distributed.test;
import java.util.Arrays;
import java.util.Comparator;
import org.junit.Assume;
import org.junit.Test;
import org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.utils.ByteBufferUtil;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
@ -156,7 +162,6 @@ public class ReadRepairRangeQueriesTest extends ReadRepairQueryTester
@Test
public void testRangeQueryWithFilterOnSelectedColumnOnSkinnyTable()
{
MutationTrackingUtils.fixmeSkipIfTracked(replicationType, "uses ALLOW FILTERING (CASSANDRA-20555)");
tester("WHERE a=2 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int PRIMARY KEY, a int, b int)")
.mutate("INSERT INTO %s (k, a, b) VALUES (1, 2, 3)",
@ -183,7 +188,6 @@ public class ReadRepairRangeQueriesTest extends ReadRepairQueryTester
@Test
public void testRangeQueryWithFilterOnSelectedColumnOnWideTable()
{
MutationTrackingUtils.fixmeSkipIfTracked(replicationType, "uses ALLOW FILTERING (CASSANDRA-20555)");
tester("WHERE a=1 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, b int, PRIMARY KEY(k, c))")
.mutate("INSERT INTO %s (k, c, a, b) VALUES (1, 1, 1, 1)",
@ -215,8 +219,6 @@ public class ReadRepairRangeQueriesTest extends ReadRepairQueryTester
@Test
public void testRangeQueryWithFilterOnUnselectedColumnOnSkinnyTable()
{
MutationTrackingUtils.fixmeSkipIfTracked(replicationType, "uses ALLOW FILTERING (CASSANDRA-20555)");
tester("WHERE b=3 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int PRIMARY KEY, a int, b int)")
.mutate("INSERT INTO %s (k, a, b) VALUES (1, 2, 3)",
@ -243,8 +245,6 @@ public class ReadRepairRangeQueriesTest extends ReadRepairQueryTester
@Test
public void testRangeQueryWithFilterOnUnselectedColumnOnWideTable()
{
if (coordinator == 2)
MutationTrackingUtils.fixmeSkipIfTracked(replicationType, "Depends on ALLOW FILTERING");
tester("WHERE b=2 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, b int, PRIMARY KEY(k, c))")
.mutate("INSERT INTO %s (k, c, a, b) VALUES (1, 1, 1, 1)",
@ -269,4 +269,113 @@ public class ReadRepairRangeQueriesTest extends ReadRepairQueryTester
rows(row(2, 1, 1, 1)),
rows());
}
/**
* Test range queries using filtering on an selected column on a table with clustering columns.
*/
@Test
public void testRangeQueryWithFilterOnSelectedColumnConflictingUpdates()
{
Assume.assumeTrue(replicationType.isTracked()); // flaky for untracked replication
tester("WHERE a=1 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, b int, PRIMARY KEY(k, c))")
.mutate(1, "INSERT INTO %s (k, c, a, b) VALUES (1, 1, 1, 1)",
"INSERT INTO %s (k, c, a, b) VALUES (1, 2, 2, 2)",
"INSERT INTO %s (k, c, a, b) VALUES (2, 1, 2, 1)",
"INSERT INTO %s (k, c, a, b) VALUES (2, 2, 2, 2)")
.mutate(2, "INSERT INTO %s (k, c, a, b) VALUES (2, 1, 1, 1)")
.mutate(1) // updates the last coordinator for mutation tracking
.queryColumns("k, c, a", 2, 2,
rows(row(1, 1, 1), row(2, 1, 1)),
rows(row(1, 1, 1, 1), row(2, 1, 1, 1)),
rows(row(1, 1, 1, null), row(2, 1, 1, 1)))
.tearDown(2,
rows(row(1, 1, 1, 1), row(1, 2, 2, 2), row(2, 1, 1, 1), row(2, 2, 2, 2)),
rows(row(1, 1, 1, 1), row(2, 1, 1, 1)));
}
/**
* Helper class for creating rows in token sorted order
*/
private static class TokenSortedKey
{
final int key;
public TokenSortedKey(int key)
{
this.key = key;
}
Token token()
{
return Murmur3Partitioner.instance.getToken(ByteBufferUtil.bytes(key));
}
public Object[] row(Object... values)
{
Object[] row = new Object[values.length + 1];
row[0] = key;
System.arraycopy(values, 0, row, 1, values.length);
return row;
}
public static TokenSortedKey[] create(int numKeys)
{
TokenSortedKey[] keys = new TokenSortedKey[numKeys];
for (int i = 0; i < numKeys; i++)
keys[i] = new TokenSortedKey(i);
Arrays.sort(keys, Comparator.comparing(TokenSortedKey::token));
return keys;
}
}
/**
* Tests the allow filtering case where the local coordinator filters out a partition that hashes to a token
* between 2 valid matches and another replica reports a mutation that causes it to be re-added to the result
*/
@Test
public void testFilteredInterleavedShortRead()
{
Assume.assumeTrue(replicationType.isTracked()); // flaky for untracked replication
TokenSortedKey[] keys = TokenSortedKey.create(4);
tester("WHERE a=1 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, PRIMARY KEY(k, c))")
.mutate(1, "INSERT INTO %s (k, c, a) VALUES (" + keys[0].key + ", 1, 1)",
"INSERT INTO %s (k, c, a) VALUES (" + keys[0].key + ", 2, 2)",
"INSERT INTO %s (k, c, a) VALUES (" + keys[1].key + ", 1, 2)",
"INSERT INTO %s (k, c, a) VALUES (" + keys[2].key + ", 1, 1)")
.mutate(2, "INSERT INTO %s (k, c, a) VALUES (" + keys[1].key + ", 1, 1)") // make keys[1] match the filter
.mutate(1) // updates the last coordinator for mutation tracking
.queryColumns("k, c, a", 3, 0,
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)))
.tearDown(1,
rows(keys[0].row(1, 1), keys[0].row(2, 2), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)));
}
@Test
public void testNonFilteredInterleavedShortRead()
{
Assume.assumeTrue(replicationType.isTracked()); // flaky for untracked replication
TokenSortedKey[] keys = TokenSortedKey.create(4);
tester("WHERE a=1 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, PRIMARY KEY(k, c))")
.mutate(1, "INSERT INTO %s (k, c, a) VALUES (" + keys[0].key + ", 1, 1)",
"INSERT INTO %s (k, c, a) VALUES (" + keys[0].key + ", 2, 2)",
"INSERT INTO %s (k, c, a) VALUES (" + keys[2].key + ", 1, 1)")
.mutate(2, "INSERT INTO %s (k, c, a) VALUES (" + keys[1].key + ", 1, 1)") // make keys[1] match the filter
.mutate(1) // updates the last coordinator for mutation tracking
.queryColumns("k, c, a", 3, 0,
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)))
.tearDown(1,
rows(keys[0].row(1, 1), keys[0].row(2, 2), keys[1].row(1, 1), keys[2].row(1, 1)),
rows(keys[0].row(1, 1), keys[1].row(1, 1), keys[2].row(1, 1)));
}
}

3
test/distributed/org/apache/cassandra/distributed/test/ReadRepairSliceQueriesTest.java
View File

@ -20,8 +20,6 @@ package org.apache.cassandra.distributed.test;
import org.junit.Test;
import org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils;
import static org.apache.cassandra.distributed.shared.AssertUtils.row;
/**
@ -64,7 +62,6 @@ public class ReadRepairSliceQueriesTest extends ReadRepairQueryTester
@Test
public void testSliceQueryWithFilter()
{
MutationTrackingUtils.fixmeSkipIfTracked(replicationType, "Depends on ALLOW FILTERING");
tester("WHERE k=0 AND a>10 AND a<40 ALLOW FILTERING")
.createTable("CREATE TABLE %s (k int, c int, a int, b int, PRIMARY KEY(k, c))")
.mutate("INSERT INTO %s (k, c, a, b) VALUES (0, 1, 10, 100)",

2
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
View File

@ -176,7 +176,7 @@ public class MutationTrackingPendingReadTest
// check that the returned data contains the unapplied mutation
try (PartialTrackedRead.CompletedRead completedRead = read.complete();
PartitionIterator partitions = completedRead.iterator())
PartitionIterator partitions = completedRead.response().makeIterator(command))
{
Assert.assertTrue(partitions.hasNext());
try (RowIterator rowIterator = partitions.next())

36
test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
View File

@ -21,6 +21,8 @@ package org.apache.cassandra.distributed.test.tracking;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.distributed.api.ICoordinator;
import org.apache.cassandra.exceptions.ReadTimeoutException;
import org.apache.cassandra.replication.CoordinatorLogId;
import org.apache.cassandra.replication.MutationSummary;
import org.junit.Assert;
@ -77,6 +79,31 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
awaitNodeLiveness(from, InetAddressAndPort.getByAddress(node.broadcastAddress()), true);
}
// TODO (expected): remove after speculation is implemented
private static Object[][] queryWithRetries(ICoordinator coordinator, String query) throws InterruptedException
{
int attempt = 0;
for (;;)
{
attempt++;
try
{
return coordinator.execute(query, ConsistencyLevel.QUORUM);
}
catch (Throwable t)
{
if (attempt < 10 && t.getClass().getSimpleName().equals(ReadTimeoutException.class.getSimpleName()))
{
Thread.sleep(2000);
}
else
{
throw t;
}
}
}
}
/**
* Test a read reconciliation where the coordinator doesn't have a read response it needs to apply
* additional mutations to
@ -146,9 +173,8 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
awaitNodeAlive(cluster.get(1), cluster.get(3));
awaitNodeDead(cluster.get(1), cluster.get(2));
Assert.assertEquals(0, numLogReconciliations(cluster.get(1)));
Object[][] result = cluster.coordinator(1).execute(format("SELECT * FROM %s.%s WHERE k=1", keyspaceName, tableName), ConsistencyLevel.QUORUM);
Object[][] result = queryWithRetries(cluster.coordinator(1), format("SELECT * FROM %s.%s WHERE k=1", keyspaceName, tableName));
Assert.assertEquals(row(row(1, 0, 0), row(1, 1, 1)), result);
// check that node3 has the new ids
@ -225,7 +251,7 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
Assert.assertEquals(0, numLogReconciliations(cluster.get(1)));
Object[][] result = cluster.coordinator(3).execute(format("SELECT * FROM %s.%s WHERE k=1", keyspaceName, tableName), ConsistencyLevel.QUORUM);
Object[][] result = queryWithRetries(cluster.coordinator(3), format("SELECT * FROM %s.%s WHERE k=1", keyspaceName, tableName));
Assert.assertEquals(row(row(1, 0, 0), row(1, 1, 1)), result);
// check that node3 has the new ids
@ -302,7 +328,7 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
// No reconciliation has happened yet
Assert.assertEquals(0, numLogReconciliations(cluster.get(1)));
Object[][] result = cluster.coordinator(1).execute(format("SELECT * FROM %s.%s", keyspaceName, tableName), ConsistencyLevel.QUORUM);
Object[][] result = queryWithRetries(cluster.coordinator(1), format("SELECT * FROM %s.%s", keyspaceName, tableName));
Assert.assertEquals(row(row(1, 0, 0), row(1, 1, 1), row(2, 2, 2)), result);
// Coordinator sends its missing mutations to 3 on read
@ -377,7 +403,7 @@ public class MutationTrackingReadReconciliationTest extends TestBaseImpl
Assert.assertEquals(0, numLogReconciliations(cluster.get(3)));
Object[][] result = cluster.coordinator(3).execute(format("SELECT * FROM %s.%s", keyspaceName, tableName), ConsistencyLevel.QUORUM);
Object[][] result = queryWithRetries(cluster.coordinator(3), format("SELECT * FROM %s.%s", keyspaceName, tableName));
Assert.assertEquals(row(row(1, 0, 0), row(1, 1, 1), row(2, 2, 2)), result);
// Coordinator sends its missing mutations to 3 on read