You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "belliottsmith (via GitHub)" <gi...@apache.org> on 2024/11/13 16:45:20 UTC
[PR] Key recovery sees ranges [cassandra]
belliottsmith opened a new pull request, #3679:
URL: https://github.com/apache/cassandra/pull/3679
Thanks for sending a pull request! Here are some tips if you're new here:
* Ensure you have added or run the [appropriate tests](https://cassandra.apache.org/_/development/testing.html) for your PR.
* Be sure to keep the PR description updated to reflect all changes.
* Write your PR title to summarize what this PR proposes.
* If possible, provide a concise example to reproduce the issue for a faster review.
* Read our [contributor guidelines](https://cassandra.apache.org/_/development/index.html)
* If you're making a documentation change, see our [guide to documentation contribution](https://cassandra.apache.org/_/development/documentation.html)
Commit messages should follow the following format:
```
<One sentence description, usually Jira title or CHANGES.txt summary>
<Optional lengthier description (context on patch)>
patch by <Authors>; reviewed by <Reviewers> for CASSANDRA-#####
Co-authored-by: Name1 <email1>
Co-authored-by: Name2 <email2>
```
The [Cassandra Jira](https://issues.apache.org/jira/projects/CASSANDRA/issues/)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1847080355
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
wasn't able to get the stack trace I was looking for, got distracted by a weird seed; I sent you in slack.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1843356274
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
it isn't more correct to use the wrong ranges that were provided by load, if indeed they were wrong. Usually the logic asking to perform the calculation has a better idea of what needs to be calculated. I will make sure there is validation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#issuecomment-2481248348
> thats fine by me. That was the original plan when I wrote that but we wanted to keep it in C*, so I don't have issues pushing an abstraction to accord-core to make it easier to maintain both sides
So, I've had to look at some related logic as part of my burn test run cleanup, and I have there also made steps to better unify the `SafeCommandStore` implementations, so I'll probably do the work to unify there instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1847107577
##########
test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java:
##########
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.TreeMap;
-
-import org.junit.Test;
-
-import accord.api.RoutingKey;
-import accord.impl.IntKey;
-import accord.primitives.SaveStatus;
-import accord.primitives.Range;
-import accord.primitives.Ranges;
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import accord.utils.AccordGens;
-import accord.utils.Gen;
-import accord.utils.Gens;
-import org.apache.cassandra.service.accord.CommandsForRangesLoader.Summary;
-
-import static accord.utils.Property.qt;
-import static org.assertj.core.api.Assertions.assertThat;
-
-public class CommandsForRangesTest
Review Comment:
believe the reason is part of this comment https://github.com/apache/cassandra/pull/3679#discussion_r1847105488
we used to do `slice` on the owned ranges, now we are slicing at a higher level (after we did the scan) based off `touches` which can be larger than `owns`...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1847081580
##########
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java:
##########
@@ -156,12 +158,12 @@ public PreLoadContext canExecute(PreLoadContext context)
}
if (unavailable == null)
- return context;
+ return with;
if (unavailable.size() == keys.size())
return null;
- return PreLoadContext.contextFor(context.primaryTxnId(), context.additionalTxnId(), keys.without(RoutingKeys.ofSortedUnique(unavailable)), keyHistory);
+ return PreLoadContext.contextFor(with.primaryTxnId(), with.additionalTxnId(), keys.without(RoutingKeys.ofSortedUnique(unavailable)), keyHistory);
Review Comment:
this is going to conflict with your other patch, but looks fine from this patches point of view.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#issuecomment-2478490739
I also noticed we weren't using any of the historic range transactions here. So I've had them create summary entries. This would be another thing that's good to test, as this was likely a long standing correctness bug. Hopefully the simulator will be up and running properly soon and begin exercising these bugs properly, but perhaps also Alex's semi-integrated burn test will include the integration `CommandsForRanges` and spot problems here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1843453176
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -161,32 +162,276 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
+ if (summary.ranges == null)
Review Comment:
I'm not sure it really is nullable, looking at it more closely. The annotation is wrong. I'll revert and remove the annotation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1853821458
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -160,33 +158,279 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
Invariants.checkState(testTxnId.equals(summary.findAsDep));
}
- // TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final @Nonnull TxnId txnId;
+ public final @Nonnull Timestamp executeAt;
+ public final @Nonnull SaveStatus saveStatus;
+ public final @Nonnull Ranges ranges;
+
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SaveStatus saveStatus, @Nonnull Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
+
+ public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
+ {
+ RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
+ TxnId minTxnId = redundantBefore.min(keysOrRanges, e -> e.gcBefore);
+ Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
+ return new CommandsForRanges.Loader(this, keysOrRanges, redundantBefore, minTxnId, maxTxnId, findAsDep);
+ }
+
+ public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
+ {
+ transitive.merge(txnId, ranges, remappingFunction);
+ }
+
+ public void gcBefore(TxnId gcBefore, Ranges ranges)
+ {
+ Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
+ while (iterator.hasNext())
+ {
+ Map.Entry<TxnId, Ranges> e = iterator.next();
+ Ranges newRanges = e.getValue().without(ranges);
+ if (newRanges.isEmpty())
+ iterator.remove();
+ e.setValue(newRanges);
+ }
+ }
+ }
+
+ public static class Loader
{
- Ranges ranges = this.ranges.slice(slice, Minimal);
- NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
- for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
+ private final Manager manager;
+ final Unseekables<?> searchKeysOrRanges;
+ final RedundantBefore redundantBefore;
+ final TxnId minTxnId;
+ final Timestamp maxTxnId;
+ @Nullable final TxnId findAsDep;
+
+ public Loader(Manager manager, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ {
+ this.manager = manager;
+ this.searchKeysOrRanges = searchKeysOrRanges;
+ this.redundantBefore = redundantBefore;
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.findAsDep = findAsDep;
+ }
+
+ public void intersects(Consumer<TxnId> forEach)
{
- if (!e.getValue().ranges.intersects(slice)) continue;
- copy.put(e.getKey(), e.getValue().slice(slice));
+ switch (searchKeysOrRanges.domain())
+ {
+ case Range:
+ for (Unseekable range : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
+ break;
+ case Key:
+ for (Unseekable key : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
+ }
+
+ if (!manager.transitive.isEmpty())
+ {
+ for (var e : manager.transitive.tailMap(minTxnId, true).entrySet())
+ {
+ if (e.getValue().intersects(searchKeysOrRanges))
+ forEach.accept(e.getKey());
+ }
+ }
+ }
+
+ public void forEachInCache(Consumer<Summary> forEach, AccordCommandStore.Caches caches)
+ {
+ for (TxnId txnId : manager.cachedRangeTxns)
+ {
+ AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
+ Summary summary = from(state);
+ if (summary != null)
+ forEach.accept(summary);
+ }
+ }
+
+ public Summary load(TxnId txnId)
+ {
+ if (findAsDep == null)
+ {
+ SavedCommand.MinimalCommand cmd = manager.commandStore.loadMinimal(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+ else
+ {
+ Command cmd = manager.commandStore.loadCommand(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+
+ Ranges ranges = manager.transitive.get(txnId);
+ if (ranges == null)
+ return null;
+
+ ranges = ranges.intersecting(searchKeysOrRanges);
+ if (ranges.isEmpty())
+ return null;
+
+ return new Summary(txnId, txnId, SaveStatus.NotDefined, ranges, null, false);
+ }
+
+ public Summary from(AccordCacheEntry<TxnId, Command> state)
+ {
+ if (state.key().domain() != Routable.Domain.Range)
+ return null;
+
+ switch (state.status())
+ {
+ default: throw new AssertionError("Unhandled status: " + state.status());
+ case LOADING:
+ case WAITING_TO_LOAD:
+ case UNINITIALIZED:
+ return null;
+
+ case LOADED:
+ case MODIFIED:
+ case SAVING:
+ case FAILED_TO_SAVE:
+ }
+
+ TxnId txnId = state.key();
+ if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
+ return null;
+
+ Command command = state.getExclusive();
+ if (command == null)
+ return null;
+ return from(command);
+ }
+
+ public Summary from(Command cmd)
+ {
+ return from(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
+ }
+
+ public Summary from(SavedCommand.MinimalCommand cmd)
+ {
+ Invariants.checkState(findAsDep == null);
+ return from(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
+ }
+
+ private Summary from(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
+ {
+ if (participants == null)
+ return null;
+
+ Ranges keysOrRanges = participants.touches().toRanges();
Review Comment:
because we rely on quorum intersections across topology changes. "Touches" means we have processed a phase (e.g. a PreAccept, Accept, Commit etc) that intersected with ranges we own _at some point_, and we did this because some future epoch broadcast backwards to us, to ensure commands we process intersect these later command's quorums. This means for both recovery and for dependency calculation, anyone asking for information about a command intersecting those ranges needs to encounter this command.
Ownership means we're responsible _for that command_
Touches means we're responsible for some _interactions_ with that command
Touches is always bigger than owns, but still represents ranges we're responsible for.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1847105488
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -160,33 +158,279 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
Invariants.checkState(testTxnId.equals(summary.findAsDep));
}
- // TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final @Nonnull TxnId txnId;
+ public final @Nonnull Timestamp executeAt;
+ public final @Nonnull SaveStatus saveStatus;
+ public final @Nonnull Ranges ranges;
+
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SaveStatus saveStatus, @Nonnull Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
+
+ public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
+ {
+ RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
+ TxnId minTxnId = redundantBefore.min(keysOrRanges, e -> e.gcBefore);
+ Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
+ return new CommandsForRanges.Loader(this, keysOrRanges, redundantBefore, minTxnId, maxTxnId, findAsDep);
+ }
+
+ public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
+ {
+ transitive.merge(txnId, ranges, remappingFunction);
+ }
+
+ public void gcBefore(TxnId gcBefore, Ranges ranges)
+ {
+ Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
+ while (iterator.hasNext())
+ {
+ Map.Entry<TxnId, Ranges> e = iterator.next();
+ Ranges newRanges = e.getValue().without(ranges);
+ if (newRanges.isEmpty())
+ iterator.remove();
+ e.setValue(newRanges);
+ }
+ }
+ }
+
+ public static class Loader
{
- Ranges ranges = this.ranges.slice(slice, Minimal);
- NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
- for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
+ private final Manager manager;
+ final Unseekables<?> searchKeysOrRanges;
+ final RedundantBefore redundantBefore;
+ final TxnId minTxnId;
+ final Timestamp maxTxnId;
+ @Nullable final TxnId findAsDep;
+
+ public Loader(Manager manager, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ {
+ this.manager = manager;
+ this.searchKeysOrRanges = searchKeysOrRanges;
+ this.redundantBefore = redundantBefore;
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.findAsDep = findAsDep;
+ }
+
+ public void intersects(Consumer<TxnId> forEach)
{
- if (!e.getValue().ranges.intersects(slice)) continue;
- copy.put(e.getKey(), e.getValue().slice(slice));
+ switch (searchKeysOrRanges.domain())
+ {
+ case Range:
+ for (Unseekable range : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
+ break;
+ case Key:
+ for (Unseekable key : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
+ }
+
+ if (!manager.transitive.isEmpty())
+ {
+ for (var e : manager.transitive.tailMap(minTxnId, true).entrySet())
+ {
+ if (e.getValue().intersects(searchKeysOrRanges))
+ forEach.accept(e.getKey());
+ }
+ }
+ }
+
+ public void forEachInCache(Consumer<Summary> forEach, AccordCommandStore.Caches caches)
+ {
+ for (TxnId txnId : manager.cachedRangeTxns)
+ {
+ AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
+ Summary summary = from(state);
+ if (summary != null)
+ forEach.accept(summary);
+ }
+ }
+
+ public Summary load(TxnId txnId)
+ {
+ if (findAsDep == null)
+ {
+ SavedCommand.MinimalCommand cmd = manager.commandStore.loadMinimal(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+ else
+ {
+ Command cmd = manager.commandStore.loadCommand(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+
+ Ranges ranges = manager.transitive.get(txnId);
+ if (ranges == null)
+ return null;
+
+ ranges = ranges.intersecting(searchKeysOrRanges);
+ if (ranges.isEmpty())
+ return null;
+
+ return new Summary(txnId, txnId, SaveStatus.NotDefined, ranges, null, false);
+ }
+
+ public Summary from(AccordCacheEntry<TxnId, Command> state)
+ {
+ if (state.key().domain() != Routable.Domain.Range)
+ return null;
+
+ switch (state.status())
+ {
+ default: throw new AssertionError("Unhandled status: " + state.status());
+ case LOADING:
+ case WAITING_TO_LOAD:
+ case UNINITIALIZED:
+ return null;
+
+ case LOADED:
+ case MODIFIED:
+ case SAVING:
+ case FAILED_TO_SAVE:
+ }
+
+ TxnId txnId = state.key();
+ if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
+ return null;
+
+ Command command = state.getExclusive();
+ if (command == null)
+ return null;
+ return from(command);
+ }
+
+ public Summary from(Command cmd)
+ {
+ return from(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
+ }
+
+ public Summary from(SavedCommand.MinimalCommand cmd)
+ {
+ Invariants.checkState(findAsDep == null);
+ return from(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
+ }
+
+ private Summary from(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
+ {
+ if (participants == null)
+ return null;
+
+ Ranges keysOrRanges = participants.touches().toRanges();
Review Comment:
previously I have seen `touches` have more ranges than `owns`, so wondering why we do this rather than `owns`? previously we would load then `slice` to what is logically `owns`, so wondering why this change? we added slice to CFR as we were including ranges outside of owns and you previously said that was a bug, so feels like we are allowing that to come back?
##########
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java:
##########
@@ -341,6 +343,7 @@ public RangesForEpoch ranges()
private <O> O mapReduce(Routables<?> keysOrRanges, BiFunction<CommandsSummary, O, O> map, O accumulate)
{
+ Invariants.checkState(context.keys().containsAll(keysOrRanges));
Review Comment:
```suggestion
Invariants.checkState(context.keys().containsAll(keysOrRanges), "Attempted to access keysOrRanges outside of what was asked for; asked for %s, accessed %s", context.keys(), keysOrRanges);
```
previous logic would tell you what ranges you were missing, we shouldn't loose that detail
##########
src/java/org/apache/cassandra/service/accord/AccordTask.java:
##########
@@ -941,7 +944,35 @@ void revert()
commandsForKey.forEach((k, v) -> v.revert());
}
- public class RangeScanner implements Runnable
+ protected void addToQueue(TaskQueue queue)
+ {
+ Invariants.checkState(queue.kind == state || (queue.kind == State.WAITING_TO_LOAD && state == WAITING_TO_SCAN_RANGES), "Invalid queue type: %s vs %s", queue.kind, this, AccordTask::toDescription);
+ Invariants.checkState(this.queued == null, "Already queued with state: %s", this, AccordTask::toDescription);
+ queued = queue;
+ queue.append(this);
+ }
+
+ TaskQueue<?> queued()
Review Comment:
```suggestion
@Nullable
TaskQueue<?> queued()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1844657235
##########
test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java:
##########
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.TreeMap;
-
-import org.junit.Test;
-
-import accord.api.RoutingKey;
-import accord.impl.IntKey;
-import accord.primitives.SaveStatus;
-import accord.primitives.Range;
-import accord.primitives.Ranges;
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import accord.utils.AccordGens;
-import accord.utils.Gen;
-import accord.utils.Gens;
-import org.apache.cassandra.service.accord.CommandsForRangesLoader.Summary;
-
-import static accord.utils.Property.qt;
-import static org.assertj.core.api.Assertions.assertThat;
-
-public class CommandsForRangesTest
Review Comment:
why was this test removed?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1843358952
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -161,32 +162,276 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
+ if (summary.ranges == null)
+ return;
+
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final TxnId txnId;
+ @Nullable public final Timestamp executeAt;
+ @Nullable public final SaveStatus saveStatus;
+ @Nullable public final Ranges ranges;
+
+ // TODO (required): this logic is still broken (was already): needs to consider exact range matches
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges == null ? null : ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
Review Comment:
This is a cache listener, so it is only invoked by the cache (which must itself be invoked while holding its lock). This code hasn't changed in this patch, just been renamed perhaps. I'm surprised to see + against it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1842735120
##########
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java:
##########
@@ -350,25 +352,6 @@ private <O> O mapReduceForRange(Routables<?> keysOrRanges, BiFunction<CommandsSu
if (commandsForRanges == null)
return accumulate;
- switch (keysOrRanges.domain())
Review Comment:
https://github.com/apache/cassandra-accord/pull/137 does `Invariants.paranoid(keysOrRanges.contains(key));` in CFK, so I think this is to push the check into CFK/CFR right?
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
https://github.com/apache/cassandra-accord/pull/137 does `Invariants.paranoid(keysOrRanges.contains(key));`, and this patch removed the checks for non-intersecting input... but CFR doesn't add any similar checks? Why is that?
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
high level question... `ranges` before were the *known* ranges asked for in the `PreLoadContext`, so we are dropping that in favor of ranges the user is defining... why is that? If you ask for `[0, 10)` but actually work with `[20, 42)` is this not a logical bug? We can't find those ranges because you didn't ask to load them... so isn't the load time more accurate?
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -161,32 +162,276 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
+ if (summary.ranges == null)
Review Comment:
moments like this I always regret not documenting why something is nullable... I don't see a code path that reaches null, but the field is flagged as nullable so this change is the correct behavior...
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -161,32 +162,276 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
+ if (summary.ranges == null)
+ return;
+
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final TxnId txnId;
+ @Nullable public final Timestamp executeAt;
+ @Nullable public final SaveStatus saveStatus;
+ @Nullable public final Ranges ranges;
+
+ // TODO (required): this logic is still broken (was already): needs to consider exact range matches
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges == null ? null : ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
Review Comment:
what threads can this be called from? looks like we are allowing more access in different places, is this callback called while holding the `commandStore.lockCaches()`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1844660507
##########
src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java:
##########
@@ -158,14 +160,28 @@ public synchronized NavigableSet<ByteBuffer> search(int storeId, TableId tableId
return pks;
}
- private TreeMap<Range, Set<DecoratedKey>> search(RangeTree<byte[], Range, DecoratedKey> tokensToPks, byte[] start, byte[] end)
+ public TreeMap<Range, Set<DecoratedKey>> search(RangeTree<byte[], Range, DecoratedKey> tokensToPks, byte[] start, byte[] end)
{
TreeMap<Range, Set<DecoratedKey>> matches = new TreeMap<>();
tokensToPks.search(new Range(start, end), e -> matches.computeIfAbsent(e.getKey(), ignore -> new HashSet<>()).add(e.getValue()));
return matches;
}
+ public synchronized NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key)
+ {
+ var rangesToPks = map.get(new Group(storeId, tableId));
Review Comment:
```suggestion
RangeTree<byte[], Range, DecoratedKey> rangesToPks = map.get(new Group(storeId, tableId));
```
since there is a desire to ban `var` this should be `RangeTree<byte[], Range, DecoratedKey>`
##########
src/java/org/apache/cassandra/index/accord/RoutesSearcher.java:
##########
@@ -97,22 +98,80 @@ public Entry next()
}
}
- public void intersects(int store, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ private CloseableIterator<Entry> searchKey(int store, AccordRoutingKey key)
{
- intersects(store, range.start(), range.end(), minTxnId, maxTxnId, forEach);
+ RowFilter rowFilter = RowFilter.create(false);
+ rowFilter.add(participants, Operator.GTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(participants, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(store_id, Operator.EQ, Int32Type.instance.decompose(store));
+
+ var cmd = PartitionRangeReadCommand.create(cfs.metadata(),
+ FBUtilities.nowInSeconds(),
+ columnFilter,
+ rowFilter,
+ limits,
+ dataRange);
+ Index.Searcher s = index.searcherFor(cmd);
+ try (var controller = cmd.executionController())
Review Comment:
```suggestion
try (ReadExecutionController controller = cmd.executionController())
```
since we want to ban `var`
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -160,33 +158,279 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
Invariants.checkState(testTxnId.equals(summary.findAsDep));
}
- // TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final @Nonnull TxnId txnId;
+ public final @Nonnull Timestamp executeAt;
+ public final @Nonnull SaveStatus saveStatus;
+ public final @Nonnull Ranges ranges;
+
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SaveStatus saveStatus, @Nonnull Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
+
+ public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
+ {
+ RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
+ TxnId minTxnId = redundantBefore.min(keysOrRanges, e -> e.gcBefore);
+ Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
+ return new CommandsForRanges.Loader(this, keysOrRanges, redundantBefore, minTxnId, maxTxnId, findAsDep);
+ }
+
+ public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
+ {
+ transitive.merge(txnId, ranges, remappingFunction);
+ }
+
+ public void gcBefore(TxnId gcBefore, Ranges ranges)
+ {
+ Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
+ while (iterator.hasNext())
+ {
+ Map.Entry<TxnId, Ranges> e = iterator.next();
+ Ranges newRanges = e.getValue().without(ranges);
+ if (newRanges.isEmpty())
+ iterator.remove();
+ e.setValue(newRanges);
+ }
+ }
+ }
+
+ public static class Loader
{
- Ranges ranges = this.ranges.slice(slice, Minimal);
- NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
- for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
+ private final Manager manager;
+ final Unseekables<?> searchKeysOrRanges;
+ final RedundantBefore redundantBefore;
+ final TxnId minTxnId;
+ final Timestamp maxTxnId;
+ @Nullable final TxnId findAsDep;
+
+ public Loader(Manager manager, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ {
+ this.manager = manager;
+ this.searchKeysOrRanges = searchKeysOrRanges;
+ this.redundantBefore = redundantBefore;
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.findAsDep = findAsDep;
+ }
+
+ public void intersects(Consumer<TxnId> forEach)
{
- if (!e.getValue().ranges.intersects(slice)) continue;
- copy.put(e.getKey(), e.getValue().slice(slice));
+ switch (searchKeysOrRanges.domain())
+ {
+ case Range:
+ for (Unseekable range : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
+ break;
+ case Key:
+ for (Unseekable key : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
+ }
+
+ if (!manager.transitive.isEmpty())
+ {
+ for (var e : manager.transitive.tailMap(minTxnId, true).entrySet())
+ {
+ if (e.getValue().intersects(searchKeysOrRanges))
+ forEach.accept(e.getKey());
+ }
+ }
+ }
+
+ public void forEachInCache(Consumer<Summary> forEach, AccordCommandStore.Caches caches)
+ {
+ for (TxnId txnId : manager.cachedRangeTxns)
+ {
+ AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
+ Summary summary = from(state);
+ if (summary != null)
+ forEach.accept(summary);
+ }
+ }
+
+ public Summary load(TxnId txnId)
+ {
+ if (findAsDep == null)
+ {
+ SavedCommand.MinimalCommand cmd = manager.commandStore.loadMinimal(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+ else
+ {
+ Command cmd = manager.commandStore.loadCommand(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+
+ Ranges ranges = manager.transitive.get(txnId);
+ if (ranges == null)
+ return null;
+
+ ranges = ranges.intersecting(searchKeysOrRanges);
+ if (ranges.isEmpty())
+ return null;
+
+ return new Summary(txnId, txnId, SaveStatus.NotDefined, ranges, null, false);
Review Comment:
good catch
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -160,33 +158,279 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
Invariants.checkState(testTxnId.equals(summary.findAsDep));
}
- // TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final @Nonnull TxnId txnId;
+ public final @Nonnull Timestamp executeAt;
+ public final @Nonnull SaveStatus saveStatus;
+ public final @Nonnull Ranges ranges;
+
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SaveStatus saveStatus, @Nonnull Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
+ {
+ private final AccordCommandStore commandStore;
+ private final RoutesSearcher searcher = new RoutesSearcher();
+ private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
+ private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
+
+ public Manager(AccordCommandStore commandStore)
+ {
+ this.commandStore = commandStore;
+ try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
+ {
+ caches.commands().register(this);
+ }
+ }
+
+ @Override
+ public void onAdd(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.add(txnId);
+ }
+
+ @Override
+ public void onEvict(AccordCacheEntry<TxnId, Command> state)
+ {
+ TxnId txnId = state.key();
+ if (txnId.is(Routable.Domain.Range))
+ cachedRangeTxns.remove(txnId);
+ }
+
+ public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
+ {
+ RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
+ TxnId minTxnId = redundantBefore.min(keysOrRanges, e -> e.gcBefore);
+ Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
+ TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
+ return new CommandsForRanges.Loader(this, keysOrRanges, redundantBefore, minTxnId, maxTxnId, findAsDep);
+ }
+
+ public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
+ {
+ transitive.merge(txnId, ranges, remappingFunction);
+ }
+
+ public void gcBefore(TxnId gcBefore, Ranges ranges)
+ {
+ Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
+ while (iterator.hasNext())
+ {
+ Map.Entry<TxnId, Ranges> e = iterator.next();
+ Ranges newRanges = e.getValue().without(ranges);
+ if (newRanges.isEmpty())
+ iterator.remove();
+ e.setValue(newRanges);
+ }
+ }
+ }
+
+ public static class Loader
{
- Ranges ranges = this.ranges.slice(slice, Minimal);
- NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
- for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
+ private final Manager manager;
+ final Unseekables<?> searchKeysOrRanges;
+ final RedundantBefore redundantBefore;
+ final TxnId minTxnId;
+ final Timestamp maxTxnId;
+ @Nullable final TxnId findAsDep;
+
+ public Loader(Manager manager, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
+ {
+ this.manager = manager;
+ this.searchKeysOrRanges = searchKeysOrRanges;
+ this.redundantBefore = redundantBefore;
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.findAsDep = findAsDep;
+ }
+
+ public void intersects(Consumer<TxnId> forEach)
{
- if (!e.getValue().ranges.intersects(slice)) continue;
- copy.put(e.getKey(), e.getValue().slice(slice));
+ switch (searchKeysOrRanges.domain())
+ {
+ case Range:
+ for (Unseekable range : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
+ break;
+ case Key:
+ for (Unseekable key : searchKeysOrRanges)
+ manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
+ }
+
+ if (!manager.transitive.isEmpty())
+ {
+ for (var e : manager.transitive.tailMap(minTxnId, true).entrySet())
+ {
+ if (e.getValue().intersects(searchKeysOrRanges))
+ forEach.accept(e.getKey());
+ }
+ }
+ }
+
+ public void forEachInCache(Consumer<Summary> forEach, AccordCommandStore.Caches caches)
+ {
+ for (TxnId txnId : manager.cachedRangeTxns)
+ {
+ AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
+ Summary summary = from(state);
+ if (summary != null)
+ forEach.accept(summary);
+ }
+ }
+
+ public Summary load(TxnId txnId)
+ {
+ if (findAsDep == null)
+ {
+ SavedCommand.MinimalCommand cmd = manager.commandStore.loadMinimal(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+ else
+ {
+ Command cmd = manager.commandStore.loadCommand(txnId);
+ if (cmd != null)
+ return from(cmd);
+ }
+
+ Ranges ranges = manager.transitive.get(txnId);
+ if (ranges == null)
+ return null;
+
+ ranges = ranges.intersecting(searchKeysOrRanges);
+ if (ranges.isEmpty())
+ return null;
+
+ return new Summary(txnId, txnId, SaveStatus.NotDefined, ranges, null, false);
+ }
+
+ public Summary from(AccordCacheEntry<TxnId, Command> state)
+ {
+ if (state.key().domain() != Routable.Domain.Range)
+ return null;
+
+ switch (state.status())
+ {
+ default: throw new AssertionError("Unhandled status: " + state.status());
+ case LOADING:
+ case WAITING_TO_LOAD:
+ case UNINITIALIZED:
+ return null;
+
+ case LOADED:
+ case MODIFIED:
+ case SAVING:
+ case FAILED_TO_SAVE:
+ }
+
+ TxnId txnId = state.key();
+ if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
+ return null;
+
+ Command command = state.getExclusive();
+ if (command == null)
+ return null;
+ return from(command);
+ }
+
+ public Summary from(Command cmd)
+ {
+ return from(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
+ }
+
+ public Summary from(SavedCommand.MinimalCommand cmd)
+ {
+ Invariants.checkState(findAsDep == null);
+ return from(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
+ }
+
+ private Summary from(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
+ {
+ if (participants == null)
+ return null;
+
+ Ranges keysOrRanges = participants.touches().toRanges();
+ if (keysOrRanges.domain() != Routable.Domain.Range)
Review Comment:
is this needed? the type is `Ranges`
##########
src/java/org/apache/cassandra/index/accord/RoutesSearcher.java:
##########
@@ -97,22 +98,80 @@ public Entry next()
}
}
- public void intersects(int store, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
+ private CloseableIterator<Entry> searchKey(int store, AccordRoutingKey key)
{
- intersects(store, range.start(), range.end(), minTxnId, maxTxnId, forEach);
+ RowFilter rowFilter = RowFilter.create(false);
+ rowFilter.add(participants, Operator.GTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(participants, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(store_id, Operator.EQ, Int32Type.instance.decompose(store));
+
+ var cmd = PartitionRangeReadCommand.create(cfs.metadata(),
+ FBUtilities.nowInSeconds(),
+ columnFilter,
+ rowFilter,
+ limits,
+ dataRange);
+ Index.Searcher s = index.searcherFor(cmd);
+ try (var controller = cmd.executionController())
+ {
+ UnfilteredPartitionIterator partitionIterator = s.search(controller);
+ return new CloseableIterator<>()
+ {
+ private final Entry entry = new Entry();
+ @Override
+ public void close()
+ {
+ partitionIterator.close();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return partitionIterator.hasNext();
+ }
+
+ @Override
+ public Entry next()
+ {
+ UnfilteredRowIterator next = partitionIterator.next();
+ var partitionKeyComponents = AccordKeyspace.CommandRows.splitPartitionKey(next.partitionKey());
Review Comment:
```suggestion
ByteBuffer[] partitionKeyComponents = AccordKeyspace.CommandRows.splitPartitionKey(next.partitionKey());
```
since we want to ban `var`
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
ill try to get you an example of what I am seeing. the topology mixup tests have been hitting cases where we load different ranges than what we are mapReducing, so we can't actually say the ranges don't exist as we never tried to load them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1847453430
##########
src/java/org/apache/cassandra/service/accord/AccordTask.java:
##########
@@ -789,14 +793,14 @@ protected void cleanupExclusive()
state(FAILED);
}
- public RangeScanner rangeScanner()
+ public RangeTxnScanner rangeScanner()
Review Comment:
```suggestion
@Nullable
public RangeTxnScanner rangeScanner()
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#issuecomment-2478487601
If you're OK with it, I might actually refactor some of this code into accord-core so we can share with `InMemoryCommandStore`? The `CommandsForRanges` abstraction (and `Summary`) could go there, so we can test that it makes correct recovery decisions.
Want to mention it before I do it, to minimise redundant review work given will create a lot of (effectively no-op) deltas again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1843424015
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -161,32 +162,276 @@ private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId te
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
+ if (summary.ranges == null)
+ return;
+
for (Range range : summary.ranges)
{
- if (!this.ranges.intersects(range))
- continue;
- collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
+ if (keysOrRanges.intersects(range))
+ collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
- for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
+ for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
- for (CommandsForRangesLoader.Summary command : e.getValue())
+ for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
- public CommandsForRanges slice(Ranges slice)
+ public static class Summary
+ {
+ public final TxnId txnId;
+ @Nullable public final Timestamp executeAt;
+ @Nullable public final SaveStatus saveStatus;
+ @Nullable public final Ranges ranges;
+
+ // TODO (required): this logic is still broken (was already): needs to consider exact range matches
+ public final TxnId findAsDep;
+ public final boolean hasAsDep;
+
+ @VisibleForTesting
+ Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, TxnId findAsDep, boolean hasAsDep)
+ {
+ this.txnId = txnId;
+ this.executeAt = executeAt;
+ this.saveStatus = saveStatus;
+ this.ranges = ranges;
+ this.findAsDep = findAsDep;
+ this.hasAsDep = hasAsDep;
+ }
+
+ public Summary slice(Ranges slice)
+ {
+ return new Summary(txnId, executeAt, saveStatus, ranges == null ? null : ranges.slice(slice, Minimal), findAsDep, hasAsDep);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "Summary{" +
+ "txnId=" + txnId +
+ ", executeAt=" + executeAt +
+ ", saveStatus=" + saveStatus +
+ ", ranges=" + ranges +
+ ", findAsDep=" + findAsDep +
+ ", hasAsDep=" + hasAsDep +
+ '}';
+ }
+ }
+
+ public static class Manager implements AccordCache.Listener<TxnId, Command>
Review Comment:
This is just renamed and moved from CommandsForRangesLoader, to centralise the logic. Very little has changed, only the addition of your new key methods.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#issuecomment-2480121105
Can you re-add the logic from `test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java` in `poc/key-recovery-sees-ranges`? The current stateful test for range index just gets extended to do key search as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith merged PR #3679:
URL: https://github.com/apache/cassandra/pull/3679
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "dcapwell (via GitHub)" <gi...@apache.org>.
dcapwell commented on PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#issuecomment-2480116623
> The CommandsForRanges abstraction (and Summary) could go there, so we can test that it makes correct recovery decisions.
thats fine by me. That was the original plan when I wrote that but we wanted to keep it in C*, so I don't have issues pushing an abstraction to accord-core to make it easier to maintain both sides
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1844973906
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
I have added validation to `AccordSafeCommandStore.mapReduce`. Also, I think this bug is fixed in the investigate mixup test PR, unless there's another cause. I forget what the cause was, but I am fairly sure I have addressed one cause either in this PR or that one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1853813738
##########
test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java:
##########
@@ -1,131 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.cassandra.service.accord;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.TreeMap;
-
-import org.junit.Test;
-
-import accord.api.RoutingKey;
-import accord.impl.IntKey;
-import accord.primitives.SaveStatus;
-import accord.primitives.Range;
-import accord.primitives.Ranges;
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import accord.utils.AccordGens;
-import accord.utils.Gen;
-import accord.utils.Gens;
-import org.apache.cassandra.service.accord.CommandsForRangesLoader.Summary;
-
-import static accord.utils.Property.qt;
-import static org.assertj.core.api.Assertions.assertThat;
-
-public class CommandsForRangesTest
Review Comment:
this wasn't testing anything anymore, and I think wasn't even before this patch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] Key recovery sees ranges [cassandra]
Posted by "belliottsmith (via GitHub)" <gi...@apache.org>.
belliottsmith commented on code in PR #3679:
URL: https://github.com/apache/cassandra/pull/3679#discussion_r1853812899
##########
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java:
##########
@@ -48,56 +64,41 @@
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
+import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
-public class CommandsForRanges implements CommandsSummary
+public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
- public final Ranges ranges;
- private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
-
- private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- this.ranges = ranges;
- this.map = map;
- }
-
- public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
- {
- return new CommandsForRanges(ranges, map);
- }
-
- @VisibleForTesting
- public int size()
+ public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
- return map.size();
+ super(m);
}
@Override
- public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
+ public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
Review Comment:
I think this is covered by the new checks I added to `AccordCommandStore.mapReduce`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org