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
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new f70fc7e77b CEP-45: Fix mutation tracking range read bugs #2
f70fc7e77b is described below
commit f70fc7e77b716fd656f44611a4e8ee831d59e43c
Author: Aleksey Yeshchenko <[email protected]>
AuthorDate: Mon Sep 28 13:05:14 2026 +0000
CEP-45: Fix mutation tracking range read bugs #2
- Cleanup parent read resources when followup reads fail
- Account for augmented reads when performing 2i followup read
patch by Blake Eggleston, Ariel Weisberg, Aleksey Yeschenko; reviewed by
Blake Eggleston, Ariel Weisberg, Aleksey Yeschenko for CASSANDRA-21726
Co-authored-by: Ariel Weisberg <[email protected]>
Co-authored-by: Blake Eggleston <[email protected]>
---
.../reads/tracked/PartialTrackedIndexRead.java | 32 +++-
.../service/reads/tracked/TrackedLocalReads.java | 8 +
.../service/reads/tracked/TrackedRead.java | 18 ++-
.../tracking/MutationTrackingPendingReadTest.java | 167 +++++++++++++++++++++
.../test/tracking/TrackedImportFailureTest.java | 4 +-
.../test/tracking/TrackedIndexedShortReadTest.java | 112 ++++++++++++++
6 files changed, 333 insertions(+), 8 deletions(-)
diff --git
a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedIndexRead.java
b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedIndexRead.java
index e414a29cca..93a264bcb1 100644
---
a/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedIndexRead.java
+++
b/src/java/org/apache/cassandra/service/reads/tracked/PartialTrackedIndexRead.java
@@ -169,6 +169,10 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
AsyncPromise<FollowUpRead<Match, Searcher>> followUpPromise = new
AsyncPromise<>();
TrackedRead.Partition trackedRead =
TrackedRead.Partition.create(metadata, partitionReadCommand, consistencyLevel,
requestTime);
+ trackedRead.future().addCallback((response, failure) -> {
+ if (failure != null)
+ followUpPromise.tryFailure(failure);
+ });
trackedRead.startLocal(requestTime, null, ((promise1, read,
consistencyLevel1, rt) -> {
try
{
@@ -176,6 +180,7 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
}
catch (Exception e)
{
+ read.close();
followUpPromise.tryFailure(e);
}
}));
@@ -538,6 +543,8 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
// TODO: maybe we should immediately start a follow up read if
it's likely this key will be included in the response
if (!followUpReads.containsKey(key) && indexNewKey(update))
{
+ // maxKey isn't raised here because maxKey is only
relevant to the range read, and the
+ // followup read is a separate read that's not part of the
range read
Future<FollowUpRead<Match, Searcher>> followUpRead =
FollowUpRead.start(command, update.partitionKey(), consistencyLevel,
requestTime);
followUpReads.put(key, followUpRead);
}
@@ -672,6 +679,12 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
private final PeekingIterator<Match> materializedIterator;
private final CloseablePeekingIterator<Match> additionalIterator;
private boolean followUpRequired = false;
+ /**
+ * Key of the last match returned; null if none. Reconciliation can
hand over keys past maxKey that are
+ * returned from their FollowUpRead, so this can be past maxKey. Do
not raise maxKey when returning these
+ * keys: maxKey guards against reading unscanned entries from
additionalIterator without a snapshot view.
+ */
+ private DecoratedKey lastKey = null;
public MergingStoppingMatchIterator(DecoratedKey maxKey,
Iterator<Match> materializedIterator, CloseablePeekingIterator<Match>
additionalIterator)
{
@@ -682,6 +695,14 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
@Override
protected Match computeNext()
+ {
+ Match match = nextMatch();
+ if (match != null)
+ lastKey = match.key();
+ return match;
+ }
+
+ private Match nextMatch()
{
if (materializedIterator.hasNext() && additionalIterator.hasNext())
{
@@ -772,11 +793,16 @@ public class PartialTrackedIndexRead<Match extends
IndexMatch, Searcher extends
{
Preconditions.checkState(command.isRangeRequest());
AbstractBounds<PartitionPosition> bounds =
command.dataRange().keyRange();
- if (maxKey == null)
+ // the matches returned past maxKey came from FollowUpReads, so
the follow up must start after them too
+ // For context, the index results are returned in partition order,
however it's possible that the most recent
+ // result came from a followup read. If we don't take that into
account when creating followup bounds, then
+ // we will return duplicate results (because we'll reread the
augmented keys)
+ DecoratedKey readTo = maxKey(maxKey, matchIterator.lastKey);
+ if (readTo == null)
return bounds;
return bounds.inclusiveRight()
- ? new Range<>(maxKey, bounds.right)
- : new ExcludingBounds<>(maxKey, bounds.right);
+ ? new Range<>(readTo, bounds.right)
+ : new ExcludingBounds<>(readTo, bounds.right);
}
private class UnfilteredResultIterator extends
AbstractIterator<UnfilteredRowIterator> implements UnfilteredPartitionIterator
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 f9942baa4a..8ea461cf43 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedLocalReads.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.service.reads.tracked;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
+import java.util.concurrent.TimeoutException;
import java.util.function.Consumer;
import javax.annotation.Nullable;
@@ -86,6 +87,12 @@ public class TrackedLocalReads implements
ExpiredStatePurger.Expireable
return beginRead(readId, metadata, command, consistencyLevel,
summaryNodes, requestTime, null, completer);
}
+ /**
+ * @param partialReadConsumer if non null, handed the read as soon as it
has been begun, so a caller coordinating
+ * this read locally can interrogate it once it
has completed. Only reads this node
+ * coordinates itself can supply one.
+ */
+
public AsyncPromise<TrackedDataResponse> beginRead(
TrackedRead.Id readId,
ClusterMetadata metadata,
@@ -273,6 +280,7 @@ public class TrackedLocalReads implements
ExpiredStatePurger.Expireable
void abort()
{
read.close();
+ promise.tryFailure(new TimeoutException("Tracked read " + readId +
" expired before reconciling"));
}
void acknowledgeReconcile(Log2OffsetsMap<?> augmentingOffsets)
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 3362198d48..dd3447812e 100644
--- a/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
+++ b/src/java/org/apache/cassandra/service/reads/tracked/TrackedRead.java
@@ -316,13 +316,21 @@ public abstract class TrackedRead<E extends Endpoints<E>,
P extends ReplicaPlan.
{
logger.trace("Locally coordinating {}", readId);
Stage.READ.submit(() -> {
- AsyncPromise<TrackedDataResponse> promise =
-
MutationTrackingService.instance().localReads().beginRead(readId,
ClusterMetadata.current(), command, consistencyLevel, summaryNodes,
requestTime, partialReadConsumer, completer);
+ AsyncPromise<TrackedDataResponse> promise;
+ try
+ {
+ promise =
MutationTrackingService.instance().localReads().beginRead(readId,
ClusterMetadata.current(), command, consistencyLevel, summaryNodes,
requestTime, partialReadConsumer, completer);
+ }
+ catch (Throwable t)
+ {
+ future.tryFailure(t);
+ throw t;
+ }
promise.addCallback((response, error) -> {
if (error != null)
{
- // TODO: notify coordinator that read has failed
logger.error("Error while processing read", error);
+ future.tryFailure(error);
return;
}
logger.trace("Finished locally coordinating {}", this);
@@ -424,6 +432,10 @@ public abstract class TrackedRead<E extends Endpoints<E>,
P extends ReplicaPlan.
reasons = failure.reasonByEndpoint();
}
+ else if (ex instanceof TimeoutException)
+ {
+ throw new ReadTimeoutException(replicaPlan.consistencyLevel(),
0, replicaPlan.readQuorum(), false);
+ }
throw new ReadFailureException(replicaPlan.consistencyLevel(), 0,
replicaPlan.readQuorum(), false, reasons);
}
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 b36378684b..6efaf203e3 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java
@@ -19,10 +19,12 @@ package org.apache.cassandra.distributed.test.tracking;
import java.io.IOException;
import java.io.UncheckedIOException;
+import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.Collections;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
@@ -34,6 +36,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import com.google.common.collect.ImmutableSet;
@@ -93,6 +96,7 @@ import org.apache.cassandra.replication.ActiveLogReconciler;
import org.apache.cassandra.replication.CoordinatorLogId;
import org.apache.cassandra.replication.Log2OffsetsMap;
import org.apache.cassandra.replication.MutationId;
+import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationSummary;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.replication.Offsets;
@@ -619,6 +623,169 @@ public class MutationTrackingPendingReadTest
}
}
+ /**
+ * A tracked index range read must complete and close when its follow up
read fails during augmentation.
+ * When reconciliation delivers a key beyond the scanned range, the range
read starts a follow up read
+ * and waits in {@code IndexPreComplete}. If the follow up read fails, the
failure must propagate to the
+ * range read so its {@link ReadExecutionController} closes and releases
the table's read {@link OpOrder.Group}.
+ */
+ @Test
+ public void testFollowUpReadAugmentFailureCompletesItsIndexRangeRead()
throws Throwable
+ {
+ runFailedFollowUpRead("follow_up_augment_failure_test", false);
+ }
+
+ /**
+ * A tracked index range read must complete and close when its follow up
read expires.
+ * When a summary node does not answer and the purger aborts the follow up
read, the timeout
+ * must propagate to the range read so its {@link ReadExecutionController}
closes.
+ */
+ @Test
+ public void testFollowUpReadExpiryCompletesItsIndexRangeRead() throws
Throwable
+ {
+ runFailedFollowUpRead("follow_up_expiry_test", true);
+ }
+
+ private static void runFailedFollowUpRead(String keyspaceName, boolean
expireFollowUp) throws Throwable
+ {
+ try (Cluster cluster = Cluster.build(2)
+ .withConfig(cfg ->
cfg.with(Feature.NETWORK)
+
.with(Feature.GOSSIP)
+
.set("mutation_tracking.enabled", true)
+
.set("read_request_timeout", "2000ms")
+
.set("range_request_timeout", "2000ms"))
+ .start())
+ {
+ String tableName = "tbl";
+ cluster.schemaChange(format("CREATE KEYSPACE %s WITH replication =
" +
+ "{'class': 'SimpleStrategy',
'replication_factor': 2} " +
+ "AND replication_type='tracked';",
keyspaceName));
+ cluster.schemaChange(format("CREATE TABLE %s.%s (k int, c int, v
int, primary key (k, c));", keyspaceName, tableName));
+ cluster.schemaChange(format("CREATE INDEX tbl_v ON %s.%s(v) USING
'legacy_local_table';", keyspaceName, tableName));
+ cluster.coordinator(1).execute(format("INSERT INTO %s.%s (k, c, v)
VALUES (1, 0, 1)", keyspaceName, tableName), ConsistencyLevel.ALL);
+
+
cluster.setUncaughtExceptionsFilter(MutationTrackingPendingReadTest::isFailedFollowUpRead);
+
+
cluster.filters().verbs(Verb.MT_SUMMARY_REQ.id).from(1).to(2).drop();
+
+ cluster.get(1).runOnInstance(() -> {
+ TableMetadata metadata =
Schema.instance.getTableMetadata(keyspaceName, tableName);
+ ColumnFamilyStore cfs =
Keyspace.open(keyspaceName).getColumnFamilyStore(tableName);
+ awaitCondition(() -> cfs.getBuiltIndexes().contains("tbl_v"),
"Index tbl_v was never built");
+ ClusterMetadata clusterMetadata = ClusterMetadata.current();
+ NodeId summaryNode =
Iterables.getOnlyElement(Sets.difference(clusterMetadata.directory.peerIds(),
+
Collections.singleton(clusterMetadata.myNodeId())));
+ TrackedLocalReads localReads =
MutationTrackingService.instance().localReads();
+ Map<TrackedRead.Id, ?> pendingReads =
pendingLocalReads(localReads);
+
+ RowFilter rowFilter = RowFilter.create(false);
+ rowFilter.add(metadata.getColumn(bytes("v")), Operator.EQ,
bytes(1));
+ PartitionRangeReadCommand command =
PartitionRangeReadCommand.create(metadata,
+
FBUtilities.nowInSeconds(),
+
ColumnFilter.all(metadata),
+
rowFilter,
+
DataLimits.NONE,
+
DataRange.allData(metadata.partitioner));
+
+ TrackedRead.Id rangeReadId = TrackedRead.Id.nextId();
+ AtomicReference<PartialTrackedRead> rangeRead = new
AtomicReference<>();
+ AsyncPromise<TrackedDataResponse> promise =
+ localReads.beginRead(rangeReadId,
+ clusterMetadata,
+ command,
+
org.apache.cassandra.db.ConsistencyLevel.ALL,
+ new int[]{ summaryNode.id() },
+
Dispatcher.RequestTime.forImmediateExecution(),
+ rangeRead::set,
+ TrackedLocalReads.Completer.DEFAULT);
+ Assert.assertFalse("The range read reconciled before its
summary node answered", promise.isDone());
+ Assert.assertTrue("Expected an index read, got " +
rangeRead.get(), rangeRead.get() instanceof PartialTrackedIndexRead);
+
+ DecoratedKey unscanned =
metadata.partitioner.decorateKey(bytes(2));
+ MutationId written =
MutationTrackingService.instance().nextMutationId(keyspaceName,
unscanned.getToken());
+ SimpleBuilders.MutationBuilder builder = new
SimpleBuilders.MutationBuilder(written, keyspaceName, unscanned);
+ builder.update(metadata).row(bytes(1)).add("v", 1);
+ builder.build().apply();
+ Assert.assertNotNull("The write never reached the mutation
journal",
+ MutationJournal.instance().read(new
ShortMutationId(written)));
+
+ Log2OffsetsMap.Mutable rangeAugmenting = new
Log2OffsetsMap.Mutable();
+ rangeAugmenting.add(new ShortMutationId(written));
+ localReads.acknowledgeReconcile(rangeReadId, rangeAugmenting);
+
+ awaitCondition(() -> !pendingReads.isEmpty(), "The follow up
read never began");
+ TrackedRead.Id followUpId =
Iterables.getOnlyElement(pendingReads.keySet());
+ Assert.assertNotEquals(rangeReadId, followUpId);
+ Assert.assertFalse("The range read completed before its follow
up read did", promise.isDone());
+
+ if (expireFollowUp)
+ {
+ awaitCondition(() ->
!pendingReads.containsKey(followUpId), "The follow up read never expired");
+ }
+ else
+ {
+ MutationId unwritten =
MutationTrackingService.instance().nextMutationId(keyspaceName,
unscanned.getToken());
+ Log2OffsetsMap.Mutable followUpAugmenting = new
Log2OffsetsMap.Mutable();
+ followUpAugmenting.add(new ShortMutationId(unwritten));
+ localReads.acknowledgeReconcile(followUpId,
followUpAugmenting);
+ }
+
+ boolean rangeReadCompleted = await(promise::isDone, 10);
+ OpOrder.Barrier barrier = cfs.readOrdering.newBarrier();
+ barrier.issue();
+ boolean drained = await(barrier.getSyncPoint()::isFinished,
10);
+
+ boolean drainedOnceRangeReadClosed = drained;
+ if (!drained)
+ {
+ rangeRead.get().close();
+ drainedOnceRangeReadClosed =
await(barrier.getSyncPoint()::isFinished, 10);
+ }
+
+ Assert.assertTrue(format("The failed follow up read left its
range read pending: range read completed=%s, " +
+ "barrier drained=%s, barrier drained
once the range read was closed by hand=%s",
+ rangeReadCompleted, drained,
drainedOnceRangeReadClosed),
+ rangeReadCompleted && drained);
+ Assert.assertNotNull("The range read succeeded without the
follow up read it depended on", promise.cause());
+ });
+ }
+ }
+
+ private static boolean isFailedFollowUpRead(Throwable t)
+ {
+ for (Throwable cause = t; cause != null; cause = cause.getCause())
+ {
+ if (cause instanceof TimeoutException)
+ return true;
+ if (cause.getMessage() != null &&
cause.getMessage().startsWith("Missing mutation"))
+ return true;
+ }
+ return false;
+ }
+
+ private static boolean await(BooleanSupplier condition, long seconds)
+ {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(seconds);
+ while (!condition.getAsBoolean() && System.nanoTime() - deadline < 0)
+ Uninterruptibles.sleepUninterruptibly(10, TimeUnit.MILLISECONDS);
+ return condition.getAsBoolean();
+ }
+
+ @SuppressWarnings("unchecked")
+ private static Map<TrackedRead.Id, ?> pendingLocalReads(TrackedLocalReads
localReads)
+ {
+ try
+ {
+ Field coordinators =
TrackedLocalReads.class.getDeclaredField("coordinators");
+ coordinators.setAccessible(true);
+ return (Map<TrackedRead.Id, ?>) coordinators.get(localReads);
+ }
+ catch (ReflectiveOperationException e)
+ {
+ throw new AssertionError(e);
+ }
+ }
+
/**
* {@link ReadExecutionController#close()} is not idempotent, and a second
close releases the table's read
* {@link OpOrder.Group} again, so a flush barrier finishes while another
read on that group is still running.
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java
index 19fc8f9f0f..623013ce10 100644
---
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedImportFailureTest.java
@@ -40,7 +40,7 @@ import
org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.shared.AssertUtils;
import org.apache.cassandra.distributed.shared.Uninterruptibles;
-import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.exceptions.ReadFailureException;
import org.apache.cassandra.io.UnversionedSerializer;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.File;
@@ -188,7 +188,7 @@ public class TrackedImportFailureTest extends
TrackedTransferTestBase
assertCoordinatedRead(missed, rows -> {
assertRows(rows, row(1, 1));
});
- }).isInstanceOf(missed.callOnInstance(() ->
ReadTimeoutException.class)); // use instance classloader
+ }).isInstanceOf(missed.callOnInstance(() ->
ReadFailureException.class)); // use instance classloader
List<String> logs = missed.logs().grep(mark, "Missing mutation
ShortMutationId").getResult();
Assertions.assertThat(logs).isNotEmpty();
diff --git
a/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedIndexedShortReadTest.java
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedIndexedShortReadTest.java
new file mode 100644
index 0000000000..ba5cea1694
--- /dev/null
+++
b/test/distributed/org/apache/cassandra/distributed/test/tracking/TrackedIndexedShortReadTest.java
@@ -0,0 +1,112 @@
+/*
+ * 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.distributed.test.tracking;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
+
+/**
+ * An index read whose short read follows a key that reconciliation handed it,
with SAI and with a legacy index.
+ */
+@RunWith(Parameterized.class)
+public class TrackedIndexedShortReadTest extends TrackedRangeReadTestBase
+{
+ private static final String SAI_TABLE =
+ "CREATE TABLE %s.tbl (pk0 int, pk1 text, ck int, v int, w int, PRIMARY
KEY ((pk0, pk1), ck)) WITH read_repair = 'NONE';" +
+ "CREATE INDEX tbl_v ON %s.tbl(v) USING 'SAI'";
+
+ private static final String LEGACY_TABLE =
+ "CREATE TABLE %s.tbl (pk0 int, pk1 text, ck int, v int, w int, PRIMARY
KEY ((pk0, pk1), ck)) WITH read_repair = 'NONE';" +
+ "CREATE INDEX tbl_v ON %s.tbl(v) USING 'legacy_local_table'";
+
+ /**
+ * Node 1 lacks (1,'z') and gets it through reconciliation. (1,'z') sorts
between the last key node 1's index scan
+ * reached and the next index match the scan did not read, so the read
returns (1,'z') and then stops for a short
+ * read, and the short read must not return (1,'z') a second time.
+ * <p>
+ * Murmur3 orders the keys (1,'n'), (1,'u'), (1,'z'), (1,'a'); all match
the index on {@code v}. With a page size of
+ * two, node 1's index scan stops after (1,'u'), leaving (1,'a') an index
match it has not read. (1,'n') and (1,'u')
+ * fail the {@code w = 1} filter, so after (1,'z') the page is not full
and the read goes on to (1,'a').
+ */
+ protected static void
indexedRangeReadHandedAKeyBeforeTheNextUnreadMatch(String name, String table)
+ {
+ String[] writes =
+ {
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'n', 1, 100, 0)
USING TIMESTAMP 10",
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'u', 1, 100, 0)
USING TIMESTAMP 11",
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'a', 1, 100, 1)
USING TIMESTAMP 12",
+ "!1:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'z', 1, 100, 1)
USING TIMESTAMP 13"
+ };
+ String select = "SELECT pk0, pk1, ck, v, w FROM %s.tbl WHERE v = 100
AND w = 1 ALLOW FILTERING";
+ assertTrackedMatchesOracle(name, table, writes, select, 2, (keyspace,
oracle) -> {
+ assertReadTogetherFromNode1(keyspace, row(1, "n"), row(1, "u"),
row(1, "z"), row(1, "a"));
+ assertDataReplicaCannotAnswerAlone(keyspace, select, oracle);
+ });
+ }
+
+ /**
+ * The same shape as {@link
#indexedRangeReadHandedAKeyBeforeTheNextUnreadMatch} with no ALLOW FILTERING:
(1,'n')
+ * and (1,'u') are rejected because node 1 missed the updates that moved
their {@code v} off 100, so node 1's index
+ * still returns them and reconciliation delivers the updates.
+ */
+ protected static void
indexedRangeReadWithStaleEntriesHandedAKeyBeforeTheNextUnreadMatch(String name,
String table)
+ {
+ String[] writes =
+ {
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'n', 1, 100, 1)
USING TIMESTAMP 10",
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'u', 1, 100, 1)
USING TIMESTAMP 11",
+ "*:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'a', 1, 100, 1)
USING TIMESTAMP 12",
+ "!1:INSERT INTO %s.tbl (pk0, pk1, ck, v, w) VALUES (1, 'z', 1, 100, 1)
USING TIMESTAMP 13",
+ "!1:UPDATE %s.tbl USING TIMESTAMP 14 SET v = 200 WHERE pk0 = 1 AND pk1
= 'n' AND ck = 1",
+ "!1:UPDATE %s.tbl USING TIMESTAMP 15 SET v = 200 WHERE pk0 = 1 AND pk1
= 'u' AND ck = 1"
+ };
+ String select = "SELECT pk0, pk1, ck, v, w FROM %s.tbl WHERE v = 100";
+ assertTrackedMatchesOracle(name, table, writes, select, 2, (keyspace,
oracle) -> {
+ assertReadTogetherFromNode1(keyspace, row(1, "n"), row(1, "u"),
row(1, "z"), row(1, "a"));
+ assertDataReplicaCannotAnswerAlone(keyspace, select, oracle);
+ });
+ }
+
+ @Test
+ public void testIndexedRangeReadHandedAKeyBeforeTheNextUnreadMatch()
+ {
+
indexedRangeReadHandedAKeyBeforeTheNextUnreadMatch("k_indexed_key_before_unread",
SAI_TABLE);
+ }
+
+ @Test
+ public void testLegacyIndexedRangeReadHandedAKeyBeforeTheNextUnreadMatch()
+ {
+
indexedRangeReadHandedAKeyBeforeTheNextUnreadMatch("k_legacy_indexed_key_before_unread",
LEGACY_TABLE);
+ }
+
+ @Test
+ public void
testIndexedRangeReadWithStaleEntriesHandedAKeyBeforeTheNextUnreadMatch()
+ {
+
indexedRangeReadWithStaleEntriesHandedAKeyBeforeTheNextUnreadMatch("k_indexed_stale_before_unread",
SAI_TABLE);
+ }
+
+ @Test
+ public void
testLegacyIndexedRangeReadWithStaleEntriesHandedAKeyBeforeTheNextUnreadMatch()
+ {
+
indexedRangeReadWithStaleEntriesHandedAKeyBeforeTheNextUnreadMatch("k_legacy_stale_before_unread",
LEGACY_TABLE);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
