You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2026/06/18 09:53:38 UTC
(cassandra-accord) 01/02: (BTree)?ReducingRangeMap fixes and improvements Fix: - (BTree)?ReducingRangeMapTest - BTree: bad invariant check, latent foldl bug Improve: - Implement efficient unconstrained fold
This is an automated email from the ASF dual-hosted git repository.
belliottsmith pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra-accord.git
commit 2cdda7f976e02aa5c3dee603237baeaa32545db6
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Fri Jun 5 10:43:21 2026 +0100
(BTree)?ReducingRangeMap fixes and improvements
Fix:
- (BTree)?ReducingRangeMapTest
- BTree: bad invariant check, latent foldl bug
Improve:
- Implement efficient unconstrained fold
patch by Benedict; reviewed by Alan Wang for CASSANDRA-21439
---
.../java/accord/impl/AbstractFetchCoordinator.java | 2 +-
.../java/accord/utils/BTreeReducingRangeMap.java | 25 ++------------
.../src/main/java/accord/utils/btree/BTree.java | 28 +++++++++++++++
.../java/accord/utils/btree/ReducingBTree.java | 17 +++++++--
.../accord/utils/BTreeReducingRangeMapTest.java | 40 +++++++++++++++++-----
.../java/accord/utils/ReducingRangeMapTest.java | 14 ++++----
6 files changed, 84 insertions(+), 42 deletions(-)
diff --git a/accord-core/src/main/java/accord/impl/AbstractFetchCoordinator.java b/accord-core/src/main/java/accord/impl/AbstractFetchCoordinator.java
index 4f3ccf1d..a418ac2c 100644
--- a/accord-core/src/main/java/accord/impl/AbstractFetchCoordinator.java
+++ b/accord-core/src/main/java/accord/impl/AbstractFetchCoordinator.java
@@ -267,7 +267,7 @@ public abstract class AbstractFetchCoordinator extends FetchCoordinator
// must be invoked by implementations some time after the read has started OR must override safeToReadAt()
protected void readStarted(SafeCommandStore safeStore)
{
- safeToReadAfter = Timestamp.nonNullOrMax(Timestamp.NONE, Timestamp.nonNullOrMax(safeToReadAfter, safeStore.commandStore().unsafeGetMaxConflicts().foldl(MaxConflicts.Entry::get, Timestamp.NONE, TxnId.NONE)));
+ safeToReadAfter = Timestamp.nonNullOrMax(Timestamp.NONE, Timestamp.nonNullOrMax(safeToReadAfter, safeStore.commandStore().unsafeGetMaxConflicts().foldl(TxnId.NONE, (id, e, min) -> e.get(min, id), Timestamp.NONE)));
}
protected Timestamp safeToReadAfter()
diff --git a/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java b/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
index 33c0f74d..6e5eb4dd 100644
--- a/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
+++ b/accord-core/src/main/java/accord/utils/BTreeReducingRangeMap.java
@@ -72,33 +72,14 @@ public class BTreeReducingRangeMap<E extends Entry<E>> implements Iterable<E>
return BTree.find(tree, (RoutingKey k, Entry<?> e) -> Entry.compare(k, e), key);
}
- public E foldl(BiFunction<E, E, E> reduce)
- {
- // TODO (expected): use BTree fold methods
- require(!isEmpty());
- Iterator<E> iter = iterator();
- E result = iter.next();
- while (iter.hasNext())
- result = reduce.apply(result, iter.next());
- return result;
- }
-
public <V2> V2 foldl(BiFunction<E, V2, V2> reduce, V2 accumulator)
{
- // TODO (expected): use BTree fold methods
- require(!isEmpty());
- for (E e : this)
- accumulator = reduce.apply(e, accumulator);
- return accumulator;
+ return BTree.foldl(tree, reduce, accumulator);
}
- public <V2, P1> V2 foldl(TriFunction<E, V2, P1, V2> reduce, V2 accumulator, P1 p1)
+ public <V2, P1> V2 foldl(P1 p1, TriFunction<P1, E, V2, V2> reduce, V2 accumulator)
{
- // TODO (expected): use BTree fold methods
- require(!isEmpty());
- for (E e : this)
- accumulator = reduce.apply(e, accumulator, p1);
- return accumulator;
+ return BTree.foldl(tree, p1, reduce, accumulator);
}
@Override
diff --git a/accord-core/src/main/java/accord/utils/btree/BTree.java b/accord-core/src/main/java/accord/utils/btree/BTree.java
index 26c3b25a..f13bc59d 100644
--- a/accord-core/src/main/java/accord/utils/btree/BTree.java
+++ b/accord-core/src/main/java/accord/utils/btree/BTree.java
@@ -40,6 +40,7 @@ import com.google.common.collect.Ordering;
import accord.utils.AsymmetricComparator;
import accord.utils.Invariants;
import accord.utils.SortedArrays;
+import accord.utils.TriFunction;
import accord.utils.btree.IntervalBTree.IntervalMaxIndex;
import accord.utils.btree.UpdateFunction.NoOp;
@@ -1848,6 +1849,33 @@ public class BTree
}
}
+ public static <I, O> O foldl(Object[] btree, BiFunction<I, O, O> function, O accumulator)
+ {
+ return BTree.<I, BiFunction<I, O, O>, O>foldl(btree, function, BiFunction::apply, accumulator);
+ }
+
+ public static <I, P, O> O foldl(Object[] btree, P param, TriFunction<P, I, O, O> function, O accumulator)
+ {
+ if (isLeaf(btree))
+ return foldlLeaf(btree, function, param, accumulator);
+
+ int keys = getBranchKeyEnd(btree);
+ for (int i = 0; i < keys; i++)
+ {
+ accumulator = foldl((Object[]) btree[keys + i], param, function, accumulator);
+ accumulator = function.apply(param, (I) btree[i], accumulator);
+ }
+ return foldl((Object[]) btree[2 * keys], param, function, accumulator);
+ }
+
+ private static <I, P, O> O foldlLeaf(Object[] btree, TriFunction<P, I, O, O> function, P param, O accumulator)
+ {
+ int limit = sizeOfLeaf(btree);
+ for (int i = 0; i < limit; i++)
+ accumulator = function.apply(param, (I) btree[i], accumulator);
+ return accumulator;
+ }
+
/**
* Simple method to walk the btree forwards and apply a function till a stop condition is reached
* <p>
diff --git a/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java b/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
index 3127724f..555c6677 100644
--- a/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
+++ b/accord-core/src/main/java/accord/utils/btree/ReducingBTree.java
@@ -573,7 +573,14 @@ public class ReducingBTree
{
ri = ranges.findNext(ri, to, ev.start(), ReducingBTree::compareWithEnd, FAST);
if (ri < 0) ri = childTo = -1 - ri;
- else childTo = ri + 1;
+ else
+ {
+ childTo = ri + 1;
+
+ // we intersect another range; refresh rv to ensure we advance correctly
+ rv = ranges.get(ri);
+ ces = rv.end().compareTo(ev.start());
+ }
}
if (childTo > childFrom)
{
@@ -740,7 +747,13 @@ public class ReducingBTree
{
ri = ranges.findNext(ri, to, ev.start(), ReducingBTree::compareWithEnd, FAST);
if (ri < 0) ri = childTo = -1 - ri;
- else childTo = ri + 1;
+ else
+ {
+ childTo = ri + 1;
+ // we intersect another range but same tree child range; refresh rv to ensure we advance correctly
+ rv = ranges.get(ri);
+ ces = rv.end().compareTo(ev.start());
+ }
}
if (childTo > childFrom)
{
diff --git a/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java b/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
index 7cc42c2f..cd2955ce 100644
--- a/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
+++ b/accord-core/src/test/java/accord/utils/BTreeReducingRangeMapTest.java
@@ -41,6 +41,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ThreadLocalRandom;
import java.util.stream.Collectors;
+import static accord.api.ProtocolModifiers.isRangeEndInclusive;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Integer.MIN_VALUE;
@@ -85,7 +86,10 @@ public class BTreeReducingRangeMapTest
static final BTreeReducingRangeMap<Entry> EMPTY = new BTreeReducingRangeMap<>();
static final RoutingKey MINIMUM_EXCL = new IntKey.Routing(MIN_VALUE);
static final RoutingKey MAXIMUM_EXCL = new IntKey.Routing(MAX_VALUE);
- static boolean END_INCLUSIVE = false;
+ // Must match the production setting: BTreeReducingRangeMap.get/foldl use isRangeEndInclusive(),
+ // so the canonical TreeMap (which models intervals as (prev, K] via ceilingEntry) only agrees
+ // with the map under test when END_INCLUSIVE has the same value.
+ static boolean END_INCLUSIVE = isRangeEndInclusive();
private static IntKey.Routing rk(int t)
{
@@ -169,8 +173,8 @@ public class BTreeReducingRangeMapTest
{
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
List<ListenableFuture<Void>> results = new ArrayList<>();
- int count = 100000;
- for (int numberOfAdditions : new int[] { 1, 10, 100 })
+ int count = 1000;
+ for (int numberOfAdditions : new int[] { 100, 10, 1 })
{
for (float maxCoveragePerRange : new float[] { 0.01f, 0.1f, 0.5f })
{
@@ -302,6 +306,11 @@ public class BTreeReducingRangeMapTest
return canonical.ceilingEntry(rk).getValue();
}
+ static Timestamp tsOrNull(Entry e)
+ {
+ return e == null ? null : e.timestamp;
+ }
+
RandomWithCanonical merge(Random random, RandomWithCanonical other)
{
RandomWithCanonical result = new RandomWithCanonical();
@@ -352,9 +361,9 @@ public class BTreeReducingRangeMapTest
{
for (RoutingKey rk : canonical.keySet())
{
- Assertions.assertEquals(get(decr(rk)), test.get(decr(rk)), id);
- Assertions.assertEquals(get(rk), test.get(rk), id);
- Assertions.assertEquals(get(incr(rk)), test.get(incr(rk)), id);
+ Assertions.assertEquals(get(decr(rk)), tsOrNull(test.get(decr(rk))), id);
+ Assertions.assertEquals(get(rk), tsOrNull(test.get(rk)), id);
+ Assertions.assertEquals(get(incr(rk)), tsOrNull(test.get(incr(rk))), id);
}
// check some random
@@ -363,7 +372,7 @@ public class BTreeReducingRangeMapTest
while (remaining-- > 0)
{
RoutingKey routingKey = rk(random);
- Assertions.assertEquals(get(routingKey), test.get(routingKey), id);
+ Assertions.assertEquals(get(routingKey), tsOrNull(test.get(routingKey)), id);
}
}
@@ -395,7 +404,7 @@ public class BTreeReducingRangeMapTest
}
List<Timestamp> foldl = test.foldl(keys, (e, timestamps) -> {
- if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e))
+ if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e.timestamp))
timestamps.add(e.timestamp);
return timestamps;
}, new ArrayList<>());
@@ -412,7 +421,7 @@ public class BTreeReducingRangeMapTest
Assertions.assertEquals(canonFoldl, foldl, id);
foldl = test.foldl(ranges, (e, timestamps) -> {
- if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e))
+ if (timestamps.isEmpty() || !timestamps.get(timestamps.size() - 1).equals(e.timestamp))
timestamps.add(e.timestamp);
return timestamps;
}, new ArrayList<>());
@@ -432,6 +441,19 @@ public class BTreeReducingRangeMapTest
}
}
Assertions.assertEquals(canonFoldl, foldl, id);
+
+ foldl = test.foldl((e, timestamps) -> {
+ timestamps.add(e.timestamp);
+ return timestamps;
+ }, new ArrayList<>());
+
+ canonFoldl.clear();
+ for (Timestamp v : canonical.values())
+ {
+ if (v != null)
+ canonFoldl.add(v);
+ }
+ Assertions.assertEquals(canonFoldl, foldl, id);
}
}
}
diff --git a/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java b/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
index 9d4fea55..759e2ff2 100644
--- a/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
+++ b/accord-core/src/test/java/accord/utils/ReducingRangeMapTest.java
@@ -45,6 +45,7 @@ import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import org.opentest4j.AssertionFailedError;
+import static accord.api.ProtocolModifiers.isRangeEndInclusive;
import static java.lang.Integer.MAX_VALUE;
import static java.lang.Integer.MIN_VALUE;
@@ -54,7 +55,10 @@ public class ReducingRangeMapTest
static final ReducingRangeMap<Timestamp> EMPTY = new ReducingRangeMap<>();
static final RoutingKey MINIMUM_EXCL = new IntKey.Routing(MIN_VALUE);
static final RoutingKey MAXIMUM_EXCL = new IntKey.Routing(MAX_VALUE);
- static boolean END_INCLUSIVE = false;
+ // Must match the production setting: ReducingRangeMap.get/foldl use isRangeEndInclusive(),
+ // so the canonical TreeMap (which models intervals as (prev, K] via ceilingEntry) only agrees
+ // with the map under test when END_INCLUSIVE has the same value.
+ static boolean END_INCLUSIVE = isRangeEndInclusive();
private static RoutingKey rk(int t)
{
@@ -168,7 +172,7 @@ public class ReducingRangeMapTest
{
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
List<ListenableFuture<Void>> results = new ArrayList<>();
- int count = 100000;
+ int count = 1000;
for (int numberOfAdditions : new int[] { 1, 10, 100 })
{
for (float maxCoveragePerRange : new float[] { 0.01f, 0.1f, 0.5f })
@@ -336,12 +340,6 @@ public class ReducingRangeMapTest
return Timestamp.max(a, b);
}
- @Override
- protected Timestamp tryMergeEqual(Timestamp a, Timestamp b)
- {
- return a;
- }
-
@Override
protected ReducingRangeMap<Timestamp> buildInternal()
{
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org