You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2021/05/25 12:58:46 UTC
[ignite-3] branch main updated: IGNITE-14667 Optimization of key
criterion inference for aggregated watch. Fixes #121
This is an automated email from the ASF dual-hosted git repository.
sk0x50 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 50d97de IGNITE-14667 Optimization of key criterion inference for aggregated watch. Fixes #121
50d97de is described below
commit 50d97de45ccb517502c7be93e7ef4089b6feaccb
Author: Kirill Gusakov <kg...@gmail.com>
AuthorDate: Tue May 25 15:57:11 2021 +0300
IGNITE-14667 Optimization of key criterion inference for aggregated watch. Fixes #121
Signed-off-by: Slava Koptilin <sl...@gmail.com>
---
.../internal/metastorage/MetaStorageManager.java | 59 +++-
.../internal/metastorage/watch/KeyCriterion.java | 305 +++++++++++++++++----
.../metastorage/watch/WatchAggregator.java | 13 +-
.../internal/metastorage/WatchAggregatorTest.java | 189 ++++++++++++-
4 files changed, 483 insertions(+), 83 deletions(-)
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
index d838d67..f0c90f9 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/MetaStorageManager.java
@@ -32,6 +32,8 @@ import org.apache.ignite.configuration.schemas.runner.NodeConfiguration;
import org.apache.ignite.internal.metastorage.client.MetaStorageServiceImpl;
import org.apache.ignite.internal.metastorage.server.SimpleInMemoryKeyValueStorage;
import org.apache.ignite.internal.metastorage.server.raft.MetaStorageCommandListener;
+import org.apache.ignite.internal.metastorage.watch.AggregatedWatch;
+import org.apache.ignite.internal.metastorage.watch.KeyCriterion;
import org.apache.ignite.internal.metastorage.watch.WatchAggregator;
import org.apache.ignite.internal.raft.Loza;
import org.apache.ignite.internal.util.Cursor;
@@ -41,10 +43,10 @@ import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteInternalCheckedException;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.metastorage.client.MetaStorageService;
import org.apache.ignite.metastorage.client.CompactedException;
import org.apache.ignite.metastorage.client.Condition;
import org.apache.ignite.metastorage.client.Entry;
+import org.apache.ignite.metastorage.client.MetaStorageService;
import org.apache.ignite.metastorage.client.Operation;
import org.apache.ignite.metastorage.client.OperationTimeoutException;
import org.apache.ignite.metastorage.client.WatchListener;
@@ -165,14 +167,11 @@ public class MetaStorageManager {
vaultMgr.appliedRevision() + 1,
this::storeEntries
);
- if (watch.isEmpty())
- deployFut.complete(Optional.empty());
- else
- metaStorageSvcFut.thenApply(svc -> svc.watch(
- watch.get().keyCriterion().toRange().getKey(),
- watch.get().keyCriterion().toRange().getValue(),
- watch.get().revision(),
- watch.get().listener()).thenAccept(id -> deployFut.complete(Optional.of(id))).join());
+
+ if (watch.isEmpty())
+ deployFut.complete(Optional.empty());
+ else
+ dispatchAppropriateMetaStorageWatch(watch.get()).thenAccept(id -> deployFut.complete(Optional.of(id))).join();
}
catch (IgniteInternalCheckedException e) {
throw new IgniteInternalException("Couldn't receive applied revision during deploy watches", e);
@@ -443,11 +442,7 @@ public class MetaStorageManager {
if (watch.isEmpty())
return CompletableFuture.completedFuture(Optional.empty());
else
- return metaStorageSvcFut.thenCompose(svc -> svc.watch(
- watch.get().keyCriterion().toRange().get1(),
- watch.get().keyCriterion().toRange().get2(),
- watch.get().revision(),
- watch.get().listener()).thenApply(Optional::of));
+ return dispatchAppropriateMetaStorageWatch(watch.get()).thenApply(Optional::of);
}));
return deployFut;
@@ -601,4 +596,40 @@ public class MetaStorageManager {
}
}
}
+
+ /**
+ * Dispatches appropriate metastorage watch method according to inferred watch criterion.
+ *
+ * @param aggregatedWatch Aggregated watch.
+ * @return Future, which will be completed after new watch registration finished.
+ */
+ private CompletableFuture<IgniteUuid> dispatchAppropriateMetaStorageWatch(AggregatedWatch aggregatedWatch) {
+ if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.CollectionCriterion) {
+ var criterion = (KeyCriterion.CollectionCriterion) aggregatedWatch.keyCriterion();
+
+ return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
+ criterion.keys(),
+ aggregatedWatch.revision(),
+ aggregatedWatch.listener()));
+ }
+ else if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.ExactCriterion) {
+ var criterion = (KeyCriterion.ExactCriterion) aggregatedWatch.keyCriterion();
+
+ return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
+ criterion.key(),
+ aggregatedWatch.revision(),
+ aggregatedWatch.listener()));
+ }
+ else if (aggregatedWatch.keyCriterion() instanceof KeyCriterion.RangeCriterion) {
+ var criterion = (KeyCriterion.RangeCriterion) aggregatedWatch.keyCriterion();
+
+ return metaStorageSvcFut.thenCompose(metaStorageSvc -> metaStorageSvc.watch(
+ criterion.from(),
+ criterion.to(),
+ aggregatedWatch.revision(),
+ aggregatedWatch.listener()));
+ }
+ else
+ throw new UnsupportedOperationException("Unsupported type of criterion");
+ }
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
index 622589b..a0a57cc 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/KeyCriterion.java
@@ -20,59 +20,69 @@ package org.apache.ignite.internal.metastorage.watch;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.Objects;
+import java.util.Set;
import org.apache.ignite.lang.ByteArray;
-import org.apache.ignite.lang.IgniteBiTuple;
/**
* Filter for listen key's changes on metastore.
*/
-public interface KeyCriterion {
+public abstract class KeyCriterion {
/**
- * Translates any type of key criterion to range of keys.
+ * Checks if this key criterion contains the key.
*
- * @return Ignite tuple with first key as start of range and second as the end.
+ * @return {@code true} if criterion contains the key, {@code false} otherwise.
*/
- public IgniteBiTuple<ByteArray, ByteArray> toRange();
+ public abstract boolean contains(ByteArray key);
/**
- * Check if this key criterion contains the key.
+ * Union current key criterion with another one.
*
- * @return true if criterion contains the key, false otherwise.
+ * @param keyCriterion Criterion to calculate the union with.
+ * @param swapTry Set to {@code true} if current criterion can't calculate the union
+ * and trying to calculate it from the opposite side.
+ * @return Result key criterion.
*/
- public boolean contains(ByteArray key);
+ protected abstract KeyCriterion union(KeyCriterion keyCriterion, boolean swapTry);
/**
- * Simple criterion which contains exactly one key.
+ * Union two key criteria and produce the new one.
+ *
+ * Rules for the union of different types of criteria:
+ * <pre>
+ * exact + exact = collection|exact
+ * collection + exact = collection
+ * collection + collection = collection
+ * range + exact = range
+ * range + collection = range
+ * range + range = range
+ * </pre>
+ *
+ * @param keyCriterion Criterion to calculate the union with.
+ * @return Result of criteria union.
*/
- static class ExactCriterion implements KeyCriterion {
- /** The key of criterion. */
- private final ByteArray key;
-
- /**
- * Creates the instance of exact criterion.
- *
- * @param key Instance of the reference key.
- */
- public ExactCriterion(ByteArray key) {
- this.key = key;
- }
-
- /** {@inheritDoc} */
- @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
- return new IgniteBiTuple<>(key, key);
- }
-
- /** {@inheritDoc} */
- @Override public boolean contains(ByteArray key) {
- return this.key.equals(key);
- }
+ public KeyCriterion union(KeyCriterion keyCriterion) {
+ return union(keyCriterion, false);
+ }
+ /**
+ * Creates a common exception that indicates the given key criterions cannot be combined.
+ *
+ * @param keyCriterion1 Criterion.
+ * @param keyCriterion2 Criterion.
+ *
+ * @return Common exception that indicates the given key criterions cannot be combined.
+ */
+ private static RuntimeException unsupportedUnionException(KeyCriterion keyCriterion1, KeyCriterion keyCriterion2) {
+ return new UnsupportedOperationException("Can't calculate the union between " + keyCriterion1.getClass() +
+ "and " + keyCriterion2.getClass() + " key criteria.");
}
/**
* Criterion which contains the range of keys.
*/
- static class RangeCriterion implements KeyCriterion {
+ public static class RangeCriterion extends KeyCriterion {
/** Start of the range. */
private final ByteArray from;
@@ -91,22 +101,135 @@ public interface KeyCriterion {
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
- return new IgniteBiTuple<>(from, to);
+ @Override public boolean contains(ByteArray key) {
+ return key.compareTo(from) >= 0 && key.compareTo(to) < 0;
+ }
+
+ /**
+ * Calculates range representation for prefix criterion
+ * as {@code (prefixKey, nextKey(prefixKey)) }.
+ *
+ * @param prefixKey Prefix criterion.
+ * @return Calculated range
+ */
+ public static RangeCriterion fromPrefixKey(ByteArray prefixKey) {
+ return new RangeCriterion(prefixKey, nextKey(prefixKey));
}
/** {@inheritDoc} */
- @Override public boolean contains(ByteArray key) {
- return key.compareTo(from) >= 0 && key.compareTo(to) < 0;
+ @Override protected KeyCriterion union(KeyCriterion keyCriterion, boolean swapTry) {
+ ByteArray from;
+ ByteArray to;
+
+ if (keyCriterion instanceof ExactCriterion) {
+ from = ((ExactCriterion)keyCriterion).key;
+ to = nextKey(from);
+ }
+ else if (keyCriterion instanceof CollectionCriterion) {
+ from = Collections.min(((CollectionCriterion)keyCriterion).keys);
+ to = nextKey(Collections.max(((CollectionCriterion)keyCriterion).keys));
+ }
+ else if (keyCriterion instanceof RangeCriterion) {
+ from = ((RangeCriterion)keyCriterion).from;
+ to = ((RangeCriterion)keyCriterion).to;
+ }
+ else if (!swapTry)
+ return keyCriterion.union(this, true);
+ else
+ throw KeyCriterion.unsupportedUnionException(this, keyCriterion);
+
+ return new RangeCriterion(
+ minFromNullables(this.from, from),
+ maxFromNullables(this.to, to)
+ );
+
+ }
+
+ /**
+ * Calculates the maximum of two keys in the scope of keys' range.
+ * According to the logic of range keys - null is an equivalent to +Inf in the range end position.
+ *
+ * @param key1 The first key to compare.
+ * @param key2 The second key to compare.
+ * @return Maximum key.
+ */
+ private static ByteArray maxFromNullables(ByteArray key1, ByteArray key2) {
+ if (key1 != null && key2 != null)
+ return (key1.compareTo(key2) >= 0) ? key1 : key2;
+ else
+ return null;
+ }
+
+ /**
+ * Calculates the minimum of two keys in the scope of keys' range.
+ * According to the logic of range keys - null is an equivalent to -Inf in the range start position.
+ *
+ * @param key1 The first key to compare.
+ * @param key2 The second key to compare.
+ * @return Minimum key.
+ */
+ private static ByteArray minFromNullables(ByteArray key1, ByteArray key2) {
+ if (key1 != null && key2 != null)
+ return (key1.compareTo(key2) < 0) ? key1 : key2;
+ else
+ return null;
+ }
+
+ /**
+ * Calculates the next key for received key.
+ *
+ * @param key Input key.
+ * @return Next key.
+ */
+ private static ByteArray nextKey(ByteArray key) {
+ var bytes = Arrays.copyOf(key.bytes(), key.bytes().length);
+
+ if (bytes[bytes.length - 1] != Byte.MAX_VALUE)
+ bytes[bytes.length - 1]++;
+ else
+ bytes = Arrays.copyOf(bytes, bytes.length + 1);
+
+ return new ByteArray(bytes);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ RangeCriterion criterion = (RangeCriterion)o;
+ return Objects.equals(from, criterion.from) && Objects.equals(to, criterion.to);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(from, to);
+ }
+
+ /**
+ * @return Start of the range.
+ */
+ public ByteArray from() {
+ return from;
+ }
+
+ /**
+ * @return End of the range (exclusive).
+ */
+ public ByteArray to() {
+ return to;
}
}
/**
* Criterion which consists collection of keys.
*/
- static class CollectionCriterion implements KeyCriterion {
+ public static class CollectionCriterion extends KeyCriterion {
/** Collection of keys. */
- private final Collection<ByteArray> keys;
+ private final Set<ByteArray> keys;
/**
* Creates the instance of collection criterion.
@@ -114,51 +237,115 @@ public interface KeyCriterion {
* @param keys Collection of keys.
*/
public CollectionCriterion(Collection<ByteArray> keys) {
- this.keys = keys;
+ this.keys = new HashSet<>(keys);
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
- return new IgniteBiTuple<>(Collections.min(keys), Collections.max(keys));
+ @Override public boolean contains(ByteArray key) {
+ return keys.contains(key);
}
/** {@inheritDoc} */
- @Override public boolean contains(ByteArray key) {
- return keys.contains(key);
+ @Override protected KeyCriterion union(KeyCriterion keyCriterion, boolean swapTry) {
+ var newKeys = new HashSet<>(keys);
+
+ if (keyCriterion instanceof ExactCriterion) {
+ newKeys.add(((ExactCriterion)keyCriterion).key);
+
+ return new CollectionCriterion(newKeys);
+ }
+ else if (keyCriterion instanceof CollectionCriterion) {
+ newKeys.addAll(((CollectionCriterion)keyCriterion).keys);
+
+ return new CollectionCriterion(newKeys);
+ } else if (!swapTry)
+ return keyCriterion.union(keyCriterion, true);
+ else
+ throw KeyCriterion.unsupportedUnionException(this, keyCriterion);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ CollectionCriterion criterion = (CollectionCriterion)o;
+ return Objects.equals(keys, criterion.keys);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return Objects.hash(keys);
+ }
+
+ /**
+ * @return Collection of keys.
+ */
+ public Set<ByteArray> keys() {
+ return keys;
}
}
/**
- * Criterion which consists of all keys with defined prefix.
+ * Simple criterion which contains exactly one key.
*/
- static class PrefixCriterion implements KeyCriterion {
- /** Prefix of the key. */
- private final ByteArray prefixKey;
+ public static class ExactCriterion extends KeyCriterion {
+ /** The key of criterion. */
+ private final ByteArray key;
/**
- * Creates the instance of prefix key criterion.
+ * Creates the instance of exact criterion.
*
- * @param prefixKey Prefix of the key.
+ * @param key Instance of the reference key.
*/
- public PrefixCriterion(ByteArray prefixKey) {
- this.prefixKey = prefixKey;
+ public ExactCriterion(ByteArray key) {
+ this.key = key;
}
/** {@inheritDoc} */
- @Override public IgniteBiTuple<ByteArray, ByteArray> toRange() {
- var bytes = Arrays.copyOf(prefixKey.bytes(), prefixKey.bytes().length);
+ @Override public boolean contains(ByteArray key) {
+ return this.key.equals(key);
+ }
- if (bytes[bytes.length - 1] != Byte.MAX_VALUE)
- bytes[bytes.length - 1]++;
+ /** {@inheritDoc} */
+ @Override protected KeyCriterion union(KeyCriterion keyCriterion, boolean swapTry) {
+ if (keyCriterion instanceof ExactCriterion) {
+ if (equals(keyCriterion))
+ return this;
+ else
+ return new CollectionCriterion(Arrays.asList(key, ((ExactCriterion)keyCriterion).key));
+ }
+ else if (!swapTry)
+ return keyCriterion.union(this, true);
else
- bytes = Arrays.copyOf(bytes, bytes.length + 1);
+ throw KeyCriterion.unsupportedUnionException(this, keyCriterion);
+ }
- return new IgniteBiTuple<>(prefixKey, new ByteArray(bytes));
+ /** {@inheritDoc} */
+ @Override public boolean equals(Object o) {
+ if (this == o)
+ return true;
+
+ if (o == null || getClass() != o.getClass())
+ return false;
+
+ ExactCriterion criterion = (ExactCriterion)o;
+ return Objects.equals(key, criterion.key);
}
/** {@inheritDoc} */
- @Override public boolean contains(ByteArray key) {
- return key.compareTo(prefixKey) >= 0 && key.compareTo(toRange().getValue()) < 0;
+ @Override public int hashCode() {
+ return Objects.hash(key);
+ }
+
+ /**
+ * @return The key of criterion.
+ */
+ public ByteArray key() {
+ return key;
}
}
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
index 6fb8023..6c59290 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/watch/WatchAggregator.java
@@ -25,7 +25,6 @@ import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
-
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.metastorage.client.EntryEvent;
@@ -70,7 +69,7 @@ public class WatchAggregator {
* @return id of registered watch. Can be used for remove watch from later.
*/
public long addPrefix(ByteArray key, WatchListener lsnr) {
- var watch = new Watch(new KeyCriterion.PrefixCriterion(key), lsnr);
+ var watch = new Watch(KeyCriterion.RangeCriterion.fromPrefixKey(key), lsnr);
var id = idCntr.incrementAndGet();
watches.put(id, watch);
return id;
@@ -147,12 +146,12 @@ public class WatchAggregator {
*
* @return aggregated criterion.
*/
- // TODO: IGNITE-14667 We can do it better than infer range always
private KeyCriterion inferGeneralCriteria() {
- return new KeyCriterion.RangeCriterion(
- watches.values().stream().map(w -> w.keyCriterion().toRange().getKey()).min(ByteArray::compareTo).get(),
- watches.values().stream().map(w -> w.keyCriterion().toRange().getValue()).max(ByteArray::compareTo).get()
- );
+ return
+ watches.values().stream()
+ .map(Watch::keyCriterion)
+ .reduce(KeyCriterion::union)
+ .get();
}
/**
diff --git a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
index 2c48c78..461a1bd 100644
--- a/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
+++ b/modules/metastorage/src/test/java/org/apache/ignite/internal/metastorage/WatchAggregatorTest.java
@@ -18,7 +18,10 @@
package org.apache.ignite.internal.metastorage;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
+import org.apache.ignite.internal.metastorage.watch.KeyCriterion;
import org.apache.ignite.internal.metastorage.watch.WatchAggregator;
import org.apache.ignite.lang.ByteArray;
import org.apache.ignite.metastorage.client.Entry;
@@ -29,6 +32,7 @@ import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
+
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -36,13 +40,19 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
+/**
+ *
+ */
public class WatchAggregatorTest {
-
+ /**
+ *
+ */
@Test
public void testEventsRouting() {
var watchAggregator = new WatchAggregator();
var lsnr1 = mock(WatchListener.class);
var lsnr2 = mock(WatchListener.class);
+
watchAggregator.add(new ByteArray("1"), lsnr1);
watchAggregator.add(new ByteArray("2"), lsnr2);
@@ -67,6 +77,9 @@ public class WatchAggregatorTest {
assertEquals(List.of(entryEvt2), watchEvt2Res.getValue().entryEvents());
}
+ /**
+ *
+ */
@Test
public void testCancel() {
var watchAggregator = new WatchAggregator();
@@ -99,6 +112,9 @@ public class WatchAggregatorTest {
verify(lsnr2, times(2)).onUpdate(any());
}
+ /**
+ *
+ */
@Test
public void testCancelByFalseFromListener() {
var watchAggregator = new WatchAggregator();
@@ -131,28 +147,195 @@ public class WatchAggregatorTest {
}
- private Entry entry(String key, String value, long revision, long updateCntr) {
+ /**
+ *
+ */
+ @Test
+ public void testOneCriterionInference() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ assertEquals(new KeyCriterion.ExactCriterion(new ByteArray("key")), keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testTwoExactCriteriaUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key1"), null);
+ watchAggregator.add(new ByteArray("key2"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.CollectionCriterion(
+ new HashSet<>(Arrays.asList(new ByteArray("key1"), new ByteArray("key2")))
+ );
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testTwoEqualExactCriteriaUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key1"), null);
+ watchAggregator.add(new ByteArray("key1"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ assertEquals(keyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testThatKeyCriteriaUnionAssociative() {
+ var data = Arrays.asList(
+ new KeyCriterion.RangeCriterion(new ByteArray("0"), new ByteArray("5")),
+ new KeyCriterion.CollectionCriterion(Arrays.asList(new ByteArray("1"), new ByteArray("2"))),
+ new KeyCriterion.ExactCriterion(new ByteArray("3")));
+
+ for (int i = 0; i < data.size() - 1; i++) {
+ for (int j = i + 1; j < data.size(); j++)
+ assertEquals(data.get(i).union(data.get(j)), data.get(j).union(data.get(i)));
+ }
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testTwoEqualCollectionCriteriaUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(Arrays.asList(new ByteArray("key1"), new ByteArray("key2")), null);
+ watchAggregator.add(Arrays.asList(new ByteArray("key1"), new ByteArray("key2")), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.CollectionCriterion(
+ new HashSet<>(Arrays.asList(new ByteArray("key1"), new ByteArray("key2")))
+ );
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testExactInTheMiddleAndRangeCriteriaOnTheEdgesUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key1"), null);
+ watchAggregator.add(new ByteArray("key0"), new ByteArray("key2"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.RangeCriterion(
+ new ByteArray("key0"), new ByteArray("key2"));
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testHighExactAndLowerRangeCriteriaUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key3"), null);
+ watchAggregator.add(new ByteArray("key0"), new ByteArray("key2"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.RangeCriterion(
+ new ByteArray("key0"), new ByteArray("key4"));
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testNullKeyAsStartOfRangeCriterion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key0"), null);
+ watchAggregator.add(null, new ByteArray("key2"), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.RangeCriterion(
+ null, new ByteArray("key2"));
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testNullKeyAsEndOfRangeCriterion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key3"), null);
+ watchAggregator.add(new ByteArray("key1"), null, null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.RangeCriterion(
+ new ByteArray("key1"), null);
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ @Test
+ public void testAllTypesOfCriteriaUnion() {
+ var watchAggregator = new WatchAggregator();
+
+ watchAggregator.add(new ByteArray("key0"), null);
+ watchAggregator.add(new ByteArray("key1"), new ByteArray("key2"), null);
+ watchAggregator.add(Arrays.asList(new ByteArray("key0"), new ByteArray("key3")), null);
+
+ var keyCriterion = watchAggregator.watch(0, null).get().keyCriterion();
+ var expKeyCriterion = new KeyCriterion.RangeCriterion(
+ new ByteArray("key0"), new ByteArray("key4"));
+ assertEquals(expKeyCriterion, keyCriterion);
+ }
+
+ /**
+ *
+ */
+ private Entry entry(String key, String val, long revision, long updateCntr) {
return new Entry() {
+ /** {@inheritDoc} */
@Override public @NotNull ByteArray key() {
return new ByteArray(key);
}
+ /** {@inheritDoc} */
@Override public @Nullable byte[] value() {
- return value.getBytes(StandardCharsets.UTF_8);
+ return val.getBytes(StandardCharsets.UTF_8);
}
+ /** {@inheritDoc} */
@Override public long revision() {
return revision;
}
+ /** {@inheritDoc} */
@Override public long updateCounter() {
return updateCntr;
}
+ /** {@inheritDoc} */
@Override public boolean empty() {
return false;
}
+ /** {@inheritDoc} */
@Override public boolean tombstone() {
return false;
}