You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by bd...@apache.org on 2026/04/14 21:46:56 UTC
(cassandra) 07/36: CEP-45: Fix ALLOW FILTERING queries for mutation tracking
This is an automated email from the ASF dual-hosted git repository.
bdeggleston pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
commit 14a743f8f73023b81c951651f19f7d11a28c8b72
Author: Blake Eggleston <bl...@ultrablake.com>
AuthorDate: Wed Apr 30 10:52:30 2025 -0700
CEP-45: Fix ALLOW FILTERING queries for mutation tracking
Patch by Blake Eggleston; Reviewed by Aleksey Yeschenko for CASSANDRA-20555
---
src/java/org/apache/cassandra/db/ReadCommand.java | 31 +-
.../cassandra/db/SinglePartitionReadCommand.java | 17 +
.../org/apache/cassandra/db/filter/RowFilter.java | 135 +++--
.../db/partitions/PartitionIterators.java | 66 ++-
.../reads/tracked/AbstractPartialTrackedRead.java | 228 ++++++--
.../reads/tracked/ExtendingCompletedRead.java | 186 ++++++
.../reads/tracked/FilteredFollowupRead.java | 177 ++++++
.../reads/tracked/PartialTrackedRangeRead.java | 638 +++++++++++++++------
.../service/reads/tracked/PartialTrackedRead.java | 25 +-
.../tracked/PartialTrackedSinglePartitionRead.java | 110 ++--
.../service/reads/tracked/TrackedDataResponse.java | 87 ++-
.../reads/tracked/TrackedLocalReadCoordinator.java | 18 +-
.../service/reads/tracked/TrackedLocalReads.java | 5 +-
.../service/reads/tracked/TrackedRead.java | 35 +-
test/conf/logback-dtest.xml | 8 +-
.../test/ReadRepairRangeQueriesTest.java | 123 +++-
.../test/ReadRepairSliceQueriesTest.java | 3 -
.../tracking/MutationTrackingPendingReadTest.java | 2 +-
.../MutationTrackingReadReconciliationTest.java | 36 +-
19 files changed, 1546 insertions(+), 384 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index edd5aad08e..7f70a35eb4 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -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.
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index 667c8349f9..1b40d679eb 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -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.
*
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index ef9f1a7e2c..c3c5fc3a80 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -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,76 +218,123 @@ public class RowFilter implements Iterable<RowFilter.Expression>
return false;
}
- /**
- * 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.
- * The filter will always be evaluated strictly in conjunction with replica filtering protection at the
- * coordinator, however, even after CASSANDRA-19007 is addressed.
- *
- * @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-190007">CASSANDRA-19007</a>
- */
- protected Transformation<BaseRowIterator<?>> filter(TableMetadata metadata, long nowInSec)
+ public static class RowFilterTransformation extends Transformation<BaseRowIterator<?>>
{
- List<Expression> partitionLevelExpressions = new ArrayList<>();
- List<Expression> rowLevelExpressions = new ArrayList<>();
- for (Expression e: expressions)
+ 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)
{
- if (e.column.isStatic() || e.column.isPartitionKey())
- partitionLevelExpressions.add(e);
- else
- rowLevelExpressions.add(e);
+ 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();
}
- long numberOfRegularColumnExpressions = rowLevelExpressions.size();
- final boolean filterNonStaticColumns = numberOfRegularColumnExpressions > 0;
-
- return new Transformation<>()
+ public int potentialMatches(PartitionUpdate update)
{
- DecoratedKey pk;
-
- @Override
- protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
+ try (UnfilteredRowIterator partition = update.unfilteredIterator())
{
- pk = partition.partitionKey();
-
- // Short-circuit all partitions that won't match based on static and partition keys
+ int matches = 0;
for (Expression e : partitionLevelExpressions)
- if (!e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
+ {
+ if (e.isSatisfiedBy(metadata, partition.partitionKey(), partition.staticRow(), nowInSec))
+ {
+ matches++;
+ break;
+ }
+ }
+
+ while (partition.hasNext())
+ {
+ Unfiltered unfiltered = partition.next();
+
+ if (unfiltered instanceof Row)
{
- partition.close();
- return null;
+ Row row = (Row) unfiltered;
+
+ for (Expression e : rowLevelExpressions)
+ {
+ if (e.isSatisfiedBy(metadata, pk, row, nowInSec))
+ {
+ matches++;
+ break;
+ }
+ }
}
+ }
+ return matches;
+ }
+ }
- BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
- ? Transformation.apply((UnfilteredRowIterator) partition, this)
- : Transformation.apply((RowIterator) partition, this);
+ @Override
+ protected BaseRowIterator<?> applyToPartition(BaseRowIterator<?> partition)
+ {
+ pk = partition.partitionKey();
- if (filterNonStaticColumns && !iterator.hasNext())
+ // 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))
{
- iterator.close();
+ partition.close();
return null;
}
+ }
+
+ BaseRowIterator<?> iterator = partition instanceof UnfilteredRowIterator
+ ? Transformation.apply((UnfilteredRowIterator) partition, this)
+ : Transformation.apply((RowIterator) partition, this);
- return iterator;
+ 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());
+ 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;
+ for (Expression e : rowLevelExpressions)
+ if (!e.isSatisfiedBy(metadata, pk, purged, nowInSec))
+ return null;
- return row;
- }
- };
+ 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.
+ * The filter will always be evaluated strictly in conjunction with replica filtering protection at the
+ * coordinator, however, even after CASSANDRA-19007 is addressed.
+ *
+ * @see <a href="https://issues.apache.org/jira/browse/CASSANDRA-190007">CASSANDRA-19007</a>
+ */
+ public RowFilterTransformation filter(TableMetadata metadata, long nowInSec)
+ {
+ return new RowFilterTransformation( this, metadata, nowInSec);
}
/**
diff --git a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java
index 569cee34d4..866f5c0e51 100644
--- a/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java
+++ b/src/java/org/apache/cassandra/db/partitions/PartitionIterators.java
@@ -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
}
};
}
- };
+ }
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/AbstractPartialTrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/AbstractPartialTrackedRead.java
index 0cbb3be38d..2819068a43 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/AbstractPartialTrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/AbstractPartialTrackedRead.java
@@ -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
+ {
+ 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
{
- INITIALIZED,
- PREPARED,
- READING,
- FINISHED
+ 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;
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/ExtendingCompletedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/ExtendingCompletedRead.java
new file mode 100644
index 0000000000..f5fa4dc25a
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/reads/tracked/ExtendingCompletedRead.java
@@ -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();
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/FilteredFollowupRead.java b/src/java/org/apache/cassandra/service/reads/tracked/FilteredFollowupRead.java
new file mode 100644
index 0000000000..121ce1a581
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/reads/tracked/FilteredFollowupRead.java
@@ -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);
+ }
+
+ });
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRangeRead.java b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRangeRead.java
index 4f50442950..1a6c91db4f 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRangeRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRangeRead.java
@@ -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;
+ protected static class FollowUpReadInfo
+ {
+ int potentialMatches = 0;
+ }
- // 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;
+ protected final PartitionRangeReadCommand command;
- public PartialTrackedRangeRead(ReadExecutionController executionController, Index.Searcher searcher, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command, UnfilteredPartitionIterator initialData)
+ private PartialTrackedRangeRead(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command)
{
- super(executionController, searcher, cfs, startTimeNanos);
+ 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();
+ }
+
+ 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);
- UnfilteredPartitionIterator materializer = new AbstractUnfilteredPartitionIterator()
+ 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);
}
+ }
- @Override
- public UnfilteredRowIterator next()
+ protected class SimpleCompleted extends RangeCompleted
+ {
+ public SimpleCompleted(SortedMap<DecoratedKey, SimpleBTreePartition> data, ShortReadSupport shortReadSupport, boolean wasAugmented)
{
- try (UnfilteredRowIterator rowIterator = initialData.next())
- {
- SimpleBTreePartition partition = augmentResponseInternal(PartitionUpdate.fromIterator(rowIterator, command.columnFilter()));
- lastPartitionKey = partition.partitionKey();
- partitionsFetched = true;
- return queryPartition(partition);
- }
+ super(data, shortReadSupport, wasAugmented);
}
@Override
- public void close()
+ protected CompletedRead extendRead(UnfilteredPartitionIterator iterator)
{
- super.close();
- initialData.close();
+ return new ExtendingCompletedRead(command, iterator, shortReadSupport.partitionsFetched, shortReadSupport.initialIteratorExhausted, shortReadSupport.followUpBounds);
}
- };
-
- // unmerged per-source counter
- final DataLimits.Counter singleResultCounter = command.limits().newCounter(command.nowInSec(),
- false,
- command.selectsFullPartition(),
- enforceStrictLiveness);
- try (UnfilteredPartitionIterator iterator = singleResultCounter.applyTo(materializer))
+ }
+
+ public Simple(ReadExecutionController executionController, ColumnFamilyStore cfs, long startTimeNanos, PartitionRangeReadCommand command)
+ {
+ super(executionController, cfs, startTimeNanos, command);
+ }
+
+ @Override
+ Materializer createMaterializer()
{
- consume(iterator);
+ return new Materializer(command)
+ {
+ @Override
+ UnfilteredPartitionIterator filter(UnfilteredPartitionIterator iterator)
+ {
+ Preconditions.checkState(command().rowFilter().isEmpty());
+ return iterator;
+ }
+
+ @Override
+ RangePrepared createRangePrepared()
+ {
+ return new SimplePrepared(data, shortReadSupport.build());
+ }
+ };
}
- initialIteratorExhausted = command.limits().isExhausted(singleResultCounter);
- if (partitionsFetched)
+
+ @Override
+ public Index.Searcher searcher()
{
- 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 null;
}
- wasAugmented = false;
}
- @Override
- UnfilteredPartitionIterator initialData()
+
+ /**
+ * 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
{
- Iterator<SimpleBTreePartition> iterator = data.values().iterator();
- return new AbstractUnfilteredPartitionIterator()
+
+ protected class FilteredPrepared extends RangePrepared
{
- @Override
- public TableMetadata metadata()
+ 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)
{
- return command.metadata();
+ super(data, shortReadSupport);
+ this.filteredKeys = filteredKeys;
+ this.filter = filter;
}
@Override
- public boolean hasNext()
+ protected boolean canAcceptUpdate(PartitionUpdate update)
{
- return iterator.hasNext();
+ 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
- public UnfilteredRowIterator next()
+ Completed complete()
{
- return queryPartition(iterator.next());
+ return new FilteredCompleted(data, shortReadSupport, wasAugmented, followUpReadInfo);
}
- };
- }
-
- @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
- {
+ protected class FilteredMaterializer extends Materializer
+ {
+ 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());
+ }
- final UnfilteredPartitionIterator iterator;
- // merged end-result counter
- final DataLimits.Counter mergedResultCounter = command.limits().newCounter(command.nowInSec(),
- true,
- command.selectsFullPartition(),
- enforceStrictLiveness);
+ @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;
+ }
+ });
+ }
- public ExtendingCompletedRead(UnfilteredPartitionIterator iterator)
- {
- this.iterator = iterator;
+ @Override
+ RangePrepared createRangePrepared()
+ {
+ return new FilteredPrepared(data, shortReadSupport.build(), filteredKeys, filter);
+ }
}
- @Override
- public PartitionIterator iterator()
+ static class FilteredCompletedRead extends ExtendingCompletedRead
{
- PartitionIterator filtered = UnfilteredPartitionIterators.filter(iterator, command.nowInSec());
- PartitionIterator counted = Transformation.apply(filtered, mergedResultCounter);
- return Transformation.apply(counted, new EmptyPartitionsDiscarder());
- }
+ 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;
+ }
- @Override
- public TrackedRead<?, ?> followupRead(ConsistencyLevel consistencyLevel, long expiresAtNanos)
- {
- // never try to request additional partitions from replicas if our reconciled partitions are already filled to the limit
- if (mergedResultCounter.isDone())
- return null;
+ /**
+ * 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;
- // we do not apply short read protection when we have no limits at all
- if (command.limits().isUnlimited())
- return null;
+ if (lastMatchingKey == null) // null means there was no data and therefore no interleaving
+ return true;
- /*
- * 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;
+ return followUpReadInfo.firstKey().compareTo(lastMatchingKey) < 0;
+ }
- /*
- * 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();
+ @Override
+ protected boolean followUpRequired()
+ {
+ return hasInterleavedFollowupKeys() || super.followUpRequired();
+ }
- 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());
+ @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,
+ expiresAtNanos,
+ followUpReadInfo,
+ command,
+ followUpBounds,
+ lastMatchingKey);
+
+ followupRead.start();
+ return followupRead;
+ }
- return makeFollowupRead(toQuery, consistencyLevel, expiresAtNanos);
}
- private TrackedRead<?, ?> makeFollowupRead(int toQuery, ConsistencyLevel consistencyLevel, long expiresAtNanos)
+ private class FilteredCompleted extends RangeCompleted
{
- DataLimits newLimits = command.limits().forShortReadRetry(toQuery);
-
- DataRange newDataRange = command.dataRange().forSubRange(followUpBounds);
+ 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;
+ }
- 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);
+ @Override
+ protected CompletedRead extendRead(UnfilteredPartitionIterator iterator)
+ {
+ return new FilteredCompletedRead(command, iterator, shortReadSupport, data.isEmpty() ? data.lastKey() : null, followUpReadInfo);
+ }
+ }
- TrackedRead.Range read = TrackedRead.Range.create(followUpCmd, replicaPlan);
- logger.trace("Short read detected, starting followup read {}", read);
- read.start(expiresAtNanos);
- return read;
+ 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;
+ }
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
index daf222239c..99e509ee7f 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedRead.java
@@ -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;
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedSinglePartitionRead.java b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedSinglePartitionRead.java
index 6aa64a2985..e15dba8a5c 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedSinglePartitionRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedSinglePartitionRead.java
@@ -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;
- @Override
- public ReadCommand command()
- {
- return command;
+ 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);
+ }
}
- @Override
- UnfilteredPartitionIterator initialData()
+ private class SinglePartitionCompleted extends Completed
{
- return initialData;
+ 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
- UnfilteredPartitionIterator augmentedData()
+ protected Prepared prepareInternal(UnfilteredPartitionIterator initialData)
{
- 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);
+ return new SinglePartitionPrepared(initialData);
}
@Override
- void augmentResponse(PartitionUpdate update)
+ public Index.Searcher searcher()
{
- Preconditions.checkArgument(update.partitionKey().equals(command.partitionKey()));
- if (augmentedData == null)
- augmentedData = new SimpleBTreePartition(command.partitionKey(), command.metadata(), UpdateTransaction.NO_OP);
-
- augmentedData.update(update);
+ return searcher;
}
+ // TODO: delete (almost?) ever
+
@Override
- CompletedRead createResult(UnfilteredPartitionIterator iterator)
+ public ReadCommand command()
{
- return CompletedRead.simple(iterator, command().nowInSec());
+ return command;
}
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedDataResponse.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedDataResponse.java
index b55095bf02..01fad19fb5 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedDataResponse.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedDataResponse.java
@@ -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;
}
};
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReadCoordinator.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReadCoordinator.java
index 6732adca4b..c62531240a 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReadCoordinator.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReadCoordinator.java
@@ -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);
});
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
index 5c2417e6b5..d356ce2efa 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
@@ -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;
}
diff --git a/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java b/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
index 9109b69c01..0d029a554c 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
@@ -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)
{
diff --git a/test/conf/logback-dtest.xml b/test/conf/logback-dtest.xml
index 22d2e9faa4..ee79d68c87 100644
--- a/test/conf/logback-dtest.xml
+++ b/test/conf/logback-dtest.xml
@@ -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>
diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReadRepairRangeQueriesTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReadRepairRangeQueriesTest.java
index b2af1f938e..3ad81cdc9e 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/ReadRepairRangeQueriesTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ReadRepairRangeQueriesTest.java
@@ -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)));
+ }
}
diff --git a/test/distributed/org/apache/cassandra/distributed/test/ReadRepairSliceQueriesTest.java b/test/distributed/org/apache/cassandra/distributed/test/ReadRepairSliceQueriesTest.java
index 58e9ecca11..62f3702d50 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/ReadRepairSliceQueriesTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/ReadRepairSliceQueriesTest.java
@@ -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)",
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
index 7e17c9c557..7d859ce110 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
@@ -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())
diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
index 306d6e5092..37f6c6a46f 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingReadReconciliationTest.java
@@ -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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org