You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datasketches.apache.org by jm...@apache.org on 2021/07/22 22:09:26 UTC
[datasketches-server] branch testing updated: move synchronized to
the specific sketches rather than processing a single call at a time
This is an automated email from the ASF dual-hosted git repository.
jmalkin pushed a commit to branch testing
in repository https://gitbox.apache.org/repos/asf/datasketches-server.git
The following commit(s) were added to refs/heads/testing by this push:
new ce385c0 move synchronized to the specific sketches rather than processing a single call at a time
ce385c0 is described below
commit ce385c023d2c4f59c13a688aa189f755739233d8
Author: Jon Malkin <jm...@users.noreply.github.com>
AuthorDate: Thu Jul 22 14:59:05 2021 -0700
move synchronized to the specific sketches rather than processing a single call at a time
---
.../server/BaseSketchesQueryHandler.java | 2 +-
.../datasketches/server/DataQueryHandler.java | 48 +++++-----
.../apache/datasketches/server/MergeHandler.java | 79 ++++++++++------
.../apache/datasketches/server/ResetHandler.java | 48 +++++-----
.../datasketches/server/SerializationHandler.java | 63 +++++++------
.../apache/datasketches/server/SketchStorage.java | 63 +++++++------
.../apache/datasketches/server/UpdateHandler.java | 102 +++++++++++----------
.../datasketches/server/SketchServerTest.java | 2 +-
.../datasketches/server/SketchStorageTest.java | 4 +-
.../datasketches/server/UpdateHandlerTest.java | 4 +-
10 files changed, 232 insertions(+), 183 deletions(-)
diff --git a/src/main/java/org/apache/datasketches/server/BaseSketchesQueryHandler.java b/src/main/java/org/apache/datasketches/server/BaseSketchesQueryHandler.java
index a3b1b8e..651c527 100644
--- a/src/main/java/org/apache/datasketches/server/BaseSketchesQueryHandler.java
+++ b/src/main/java/org/apache/datasketches/server/BaseSketchesQueryHandler.java
@@ -108,7 +108,7 @@ public abstract class BaseSketchesQueryHandler extends AbstractHandler {
* @param query A JSON query to process
* @return A JSON response
*/
- final synchronized JsonObject callProcessQuery(final JsonObject query) {
+ final JsonObject callProcessQuery(final JsonObject query) {
return processQuery(query);
}
diff --git a/src/main/java/org/apache/datasketches/server/DataQueryHandler.java b/src/main/java/org/apache/datasketches/server/DataQueryHandler.java
index 42f9aca..cac8f58 100644
--- a/src/main/java/org/apache/datasketches/server/DataQueryHandler.java
+++ b/src/main/java/org/apache/datasketches/server/DataQueryHandler.java
@@ -59,35 +59,39 @@ public class DataQueryHandler extends BaseSketchesQueryHandler {
throw new IllegalArgumentException("Invalid sketch name: " + key);
}
- final SketchStorage.SketchEntry se = sketches.getSketch(key);
JsonObject result = new JsonObject();
- // pre-populate with the sketch name, but may be overwritten with
- // null depending on the queyr
- result.addProperty(QUERY_NAME_FIELD, key);
+ // we do need to lock the sketch even for query processing
+ synchronized (key.intern()) {
+ final SketchStorage.SketchEntry se = sketches.getSketch(key);
- switch (se.family) {
- case UNION:
- case HLL:
- case CPC:
- result = processDistinctQuery(result, query, se.family, se.sketch);
- break;
+ // pre-populate with the sketch name, but may be overwritten with
+ // null depending on the query
+ result.addProperty(QUERY_NAME_FIELD, key);
- case KLL:
- result = processQuantilesQuery(result, query, se.family, se.sketch);
- break;
+ switch (se.family_) {
+ case UNION:
+ case HLL:
+ case CPC:
+ result = processDistinctQuery(result, query, se.family_, se.sketch_);
+ break;
- case FREQUENCY:
- result = processFrequencyQuery(result, query, se.family, se.sketch);
- break;
+ case KLL:
+ result = processQuantilesQuery(result, query, se.family_, se.sketch_);
+ break;
- case RESERVOIR:
- case VAROPT:
- result = processSamplingQuery(result, query, se.family, se.sketch);
- break;
+ case FREQUENCY:
+ result = processFrequencyQuery(result, query, se.family_, se.sketch_);
+ break;
- default:
- throw new IllegalStateException("Unexpected sketch family: " + se.family);
+ case RESERVOIR:
+ case VAROPT:
+ result = processSamplingQuery(result, query, se.family_, se.sketch_);
+ break;
+
+ default:
+ throw new IllegalStateException("Unexpected sketch family: " + se.family_);
+ }
}
return result;
diff --git a/src/main/java/org/apache/datasketches/server/MergeHandler.java b/src/main/java/org/apache/datasketches/server/MergeHandler.java
index f3c7fcf..d75fad1 100644
--- a/src/main/java/org/apache/datasketches/server/MergeHandler.java
+++ b/src/main/java/org/apache/datasketches/server/MergeHandler.java
@@ -64,6 +64,20 @@ public class MergeHandler extends BaseSketchesQueryHandler {
super(sketches);
}
+ /**
+ * Holds an entry in the list of sketches to merge.
+ * <tt>name.intern()</tt> can be used to synchronize the sketch to avoid concurrency issues
+ */
+ static class MergeEntry {
+ final String name_;
+ final Object sketch_;
+
+ MergeEntry(final String name, final Object sketch) {
+ name_ = name;
+ sketch_ = sketch;
+ }
+ }
+
@Override
protected JsonObject processQuery(final JsonObject query) {
// optional targets:
@@ -95,15 +109,23 @@ public class MergeHandler extends BaseSketchesQueryHandler {
Family dstFamily = null;
if (dst != null) {
se = sketches.getSketch(dst);
- dstFamily = se.family;
+ dstFamily = se.family_;
}
// we'll process (and dedup) any stored sketches before we handle encoded inputs
// but we'll run through all of them before doing anything
- final ArrayList<Object> srcSketches = new ArrayList<>(srcList.size());
+ final ArrayList<MergeEntry> srcSketches = new ArrayList<>(srcList.size());
dstFamily = prepareSketches(srcList, dstFamily, dst, srcSketches);
- final byte[] skBytes = mergeSketches(dstFamily, k, se, srcSketches);
+ final byte[] skBytes;
+ // need to synchronize if we have a named sketch
+ if (se == null) {
+ skBytes = mergeSketches(dstFamily, k, null, srcSketches);
+ } else {
+ synchronized (se.name_.intern()) {
+ skBytes = mergeSketches(dstFamily, k, se, srcSketches);
+ }
+ }
// skBytes == null if merging into another sketch; only non-null if returning a serialized image
if (skBytes != null) {
@@ -115,7 +137,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
}
}
- private Family prepareSketches(final JsonArray sources, Family family, final String dst, final ArrayList<Object> sketchList) {
+ private Family prepareSketches(final JsonArray sources, Family family, final String dst, final ArrayList<MergeEntry> sketchList) {
final HashSet<String> namedSet = new HashSet<>();
// TODO: Check for sketch value types with distinct counting?
@@ -131,7 +153,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
// check family
final String key = elmt.getAsString();
final SketchStorage.SketchEntry entry = sketches.getSketch(key);
- if (entry == null || (family != null && family != entry.family)) {
+ if (entry == null || (family != null && family != entry.family_)) {
throw new SketchesException("Input sketches must exist and be of the same family as the target");
}
@@ -139,14 +161,14 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (!namedSet.contains(key)) {
namedSet.add(key);
if (family == null) {
- family = entry.family;
+ family = entry.family_;
}
// if we have a theta Union we need to get the result first
- if (entry.family == Family.UNION) {
- sketchList.add(((Union) entry.sketch).getResult());
+ if (entry.family_ == Family.UNION) {
+ sketchList.add(new MergeEntry(key, ((Union) entry.sketch_).getResult()));
} else {
- sketchList.add(entry.sketch);
+ sketchList.add(new MergeEntry(key, entry.sketch_));
}
}
} else { // is JsonObject
@@ -166,8 +188,9 @@ public class MergeHandler extends BaseSketchesQueryHandler {
}
// add to list, save family if we didn't have one yet
+ // use hashcode of sketch as name -- we'll later create a needless lock but cleaner than conditional locking
final Object sketch = deserializeSketch(skFamily, skString);
- sketchList.add(sketch);
+ sketchList.add(new MergeEntry(Integer.toString(sketch.hashCode()), sketch));
if (family == null) {
family = skFamily;
}
@@ -213,7 +236,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
@SuppressWarnings("unchecked")
private static byte[] mergeSketches(final Family family, final int k,
- final SketchStorage.SketchEntry dstEntry, final ArrayList<Object> sketchList) {
+ final SketchStorage.SketchEntry dstEntry, final ArrayList<MergeEntry> sketchList) {
if (family == null || sketchList.size() == 0) {
return null;
}
@@ -223,15 +246,17 @@ public class MergeHandler extends BaseSketchesQueryHandler {
case QUICKSELECT: {
// for theta, the destination is already a union so no need to add explicitly
final Union dst = dstEntry == null ? new SetOperationBuilder().setNominalEntries(1 << k).buildUnion()
- : (Union) dstEntry.sketch;
- for (final Object obj : sketchList) {
- dst.union((CompactSketch) obj);
+ : (Union) dstEntry.sketch_;
+ for (final MergeEntry me : sketchList) {
+ synchronized (me.name_.intern()) {
+ dst.union((CompactSketch) me.sketch_);
+ }
}
if (dstEntry == null) {
return dst.getResult().toByteArray();
} else {
- dstEntry.sketch = dst;
+ dstEntry.sketch_ = dst;
return null;
}
}
@@ -239,7 +264,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
case HLL: {
final org.apache.datasketches.hll.Union union = new org.apache.datasketches.hll.Union(k);
if (dstEntry != null) {
- union.update((HllSketch) dstEntry.sketch);
+ union.update((HllSketch) dstEntry.sketch_);
}
for (final Object obj : sketchList) {
union.update((HllSketch) obj);
@@ -248,7 +273,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.getResult().toCompactByteArray();
} else {
- dstEntry.sketch = union.getResult();
+ dstEntry.sketch_ = union.getResult();
return null;
}
}
@@ -256,7 +281,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
case CPC: {
final CpcUnion union = new CpcUnion(k);
if (dstEntry != null) {
- union.update((CpcSketch) dstEntry.sketch);
+ union.update((CpcSketch) dstEntry.sketch_);
}
for (final Object obj : sketchList) {
union.update((CpcSketch) obj);
@@ -265,14 +290,14 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.getResult().toByteArray();
} else {
- dstEntry.sketch = union.getResult();
+ dstEntry.sketch_ = union.getResult();
return null;
}
}
case KLL: {
// Only merge(), no separate union. Slightly abusing terminology to call it union
- final KllFloatsSketch union = dstEntry == null ? new KllFloatsSketch(k) : (KllFloatsSketch) dstEntry.sketch;
+ final KllFloatsSketch union = dstEntry == null ? new KllFloatsSketch(k) : (KllFloatsSketch) dstEntry.sketch_;
for (final Object obj : sketchList) {
union.merge((KllFloatsSketch) obj);
@@ -281,14 +306,14 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.toByteArray();
} else {
- dstEntry.sketch = union;
+ dstEntry.sketch_ = union;
return null;
}
}
case FREQUENCY: {
// Only merge(), no separate union. Slightly abusing terminology to call it union
- final ItemsSketch<String> union = dstEntry == null ? new ItemsSketch<>(k) : (ItemsSketch<String>) dstEntry.sketch;
+ final ItemsSketch<String> union = dstEntry == null ? new ItemsSketch<>(k) : (ItemsSketch<String>) dstEntry.sketch_;
for (final Object obj : sketchList) {
union.merge((ItemsSketch<String>) obj);
@@ -297,7 +322,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.toByteArray(new ArrayOfStringsSerDe());
} else {
- dstEntry.sketch = union;
+ dstEntry.sketch_ = union;
return null;
}
}
@@ -305,7 +330,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
case RESERVOIR: {
final ReservoirItemsUnion<String> union = ReservoirItemsUnion.newInstance(k);
if (dstEntry != null) {
- union.update((ReservoirItemsSketch<String>) dstEntry.sketch);
+ union.update((ReservoirItemsSketch<String>) dstEntry.sketch_);
}
for (final Object obj : sketchList) {
@@ -315,7 +340,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.getResult().toByteArray(new ArrayOfStringsSerDe());
} else {
- dstEntry.sketch = union.getResult();
+ dstEntry.sketch_ = union.getResult();
return null;
}
}
@@ -323,7 +348,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
case VAROPT: {
final VarOptItemsUnion<String> union = VarOptItemsUnion.newInstance(k);
if (dstEntry != null) {
- union.update((VarOptItemsSketch<String>) dstEntry.sketch);
+ union.update((VarOptItemsSketch<String>) dstEntry.sketch_);
}
for (final Object obj : sketchList) {
@@ -333,7 +358,7 @@ public class MergeHandler extends BaseSketchesQueryHandler {
if (dstEntry == null) {
return union.getResult().toByteArray(new ArrayOfStringsSerDe());
} else {
- dstEntry.sketch = union.getResult();
+ dstEntry.sketch_ = union.getResult();
return null;
}
}
diff --git a/src/main/java/org/apache/datasketches/server/ResetHandler.java b/src/main/java/org/apache/datasketches/server/ResetHandler.java
index ec22a47..493d9ab 100644
--- a/src/main/java/org/apache/datasketches/server/ResetHandler.java
+++ b/src/main/java/org/apache/datasketches/server/ResetHandler.java
@@ -50,30 +50,32 @@ public class ResetHandler extends BaseSketchesQueryHandler {
throw new IllegalArgumentException("Invalid sketch name: " + key);
}
- final SketchStorage.SketchEntry se = sketches.getSketch(key);
+ synchronized (key.intern()) {
+ final SketchStorage.SketchEntry se = sketches.getSketch(key);
- switch (se.family) {
- case UNION:
- ((Union) se.sketch).reset();
- break;
- case KLL:
- se.sketch = new KllFloatsSketch(se.configK);
- break;
- case FREQUENCY:
- ((ItemsSketch<String>) se.sketch).reset();
- break;
- case HLL:
- ((HllSketch) se.sketch).reset();
- break;
- case CPC:
- ((CpcSketch) se.sketch).reset();
- break;
- case RESERVOIR:
- ((ReservoirItemsSketch<String>) se.sketch).reset();
- break;
- case VAROPT:
- ((VarOptItemsSketch<String>) se.sketch).reset();
- break;
+ switch (se.family_) {
+ case UNION:
+ ((Union) se.sketch_).reset();
+ break;
+ case KLL:
+ se.sketch_ = new KllFloatsSketch(se.configK_);
+ break;
+ case FREQUENCY:
+ ((ItemsSketch<String>) se.sketch_).reset();
+ break;
+ case HLL:
+ ((HllSketch) se.sketch_).reset();
+ break;
+ case CPC:
+ ((CpcSketch) se.sketch_).reset();
+ break;
+ case RESERVOIR:
+ ((ReservoirItemsSketch<String>) se.sketch_).reset();
+ break;
+ case VAROPT:
+ ((VarOptItemsSketch<String>) se.sketch_).reset();
+ break;
+ }
}
// nothing to return from reset calls
diff --git a/src/main/java/org/apache/datasketches/server/SerializationHandler.java b/src/main/java/org/apache/datasketches/server/SerializationHandler.java
index 8c285db..464667b 100644
--- a/src/main/java/org/apache/datasketches/server/SerializationHandler.java
+++ b/src/main/java/org/apache/datasketches/server/SerializationHandler.java
@@ -60,42 +60,47 @@ public class SerializationHandler extends BaseSketchesQueryHandler {
throw new IllegalArgumentException("Invalid sketch name: " + name);
}
- final SketchStorage.SketchEntry se = sketches.getSketch(name);
-
final byte[] bytes;
- switch (se.family) {
- case UNION:
- bytes = ((Union) se.sketch).getResult().toByteArray();
- break;
- case KLL:
- bytes = ((KllFloatsSketch) se.sketch).toByteArray();
- break;
- case FREQUENCY:
- bytes = ((ItemsSketch<String>) se.sketch).toByteArray(new ArrayOfStringsSerDe());
- break;
- case HLL:
- bytes = ((HllSketch) se.sketch).toCompactByteArray();
- break;
- case CPC:
- bytes = ((CpcSketch) se.sketch).toByteArray();
- break;
- case RESERVOIR:
- bytes = ((ReservoirItemsSketch<String>) se.sketch).toByteArray(new ArrayOfStringsSerDe());
- break;
- case VAROPT:
- bytes = ((VarOptItemsSketch<String>) se.sketch).toByteArray(new ArrayOfStringsSerDe());
- break;
- default:
- throw new IllegalStateException("Unexpected value: " + se.family);
+ final SketchStorage.SketchEntry se;
+
+ // need to lock the sketch even when just reading
+ synchronized (name.intern()) {
+ se = sketches.getSketch(name);
+
+ switch (se.family_) {
+ case UNION:
+ bytes = ((Union) se.sketch_).getResult().toByteArray();
+ break;
+ case KLL:
+ bytes = ((KllFloatsSketch) se.sketch_).toByteArray();
+ break;
+ case FREQUENCY:
+ bytes = ((ItemsSketch<String>) se.sketch_).toByteArray(new ArrayOfStringsSerDe());
+ break;
+ case HLL:
+ bytes = ((HllSketch) se.sketch_).toCompactByteArray();
+ break;
+ case CPC:
+ bytes = ((CpcSketch) se.sketch_).toByteArray();
+ break;
+ case RESERVOIR:
+ bytes = ((ReservoirItemsSketch<String>) se.sketch_).toByteArray(new ArrayOfStringsSerDe());
+ break;
+ case VAROPT:
+ bytes = ((VarOptItemsSketch<String>) se.sketch_).toByteArray(new ArrayOfStringsSerDe());
+ break;
+ default:
+ throw new IllegalStateException("Unexpected value: " + se.family_);
+ }
}
final String b64Sketch = Base64.getUrlEncoder().encodeToString(bytes);
final JsonObject result = new JsonObject();
result.addProperty(QUERY_NAME_FIELD, name);
- result.addProperty(CONFIG_FAMILY_FIELD, se.family.getFamilyName());
- if (se.type != null)
- result.addProperty(CONFIG_TYPE_FIELD, se.type.getTypeName());
+ result.addProperty(CONFIG_FAMILY_FIELD, se.family_.getFamilyName());
+ if (se.type_ != null)
+ result.addProperty(CONFIG_TYPE_FIELD, se.type_.getTypeName());
result.addProperty(QUERY_SKETCH_FIELD, b64Sketch);
return result;
diff --git a/src/main/java/org/apache/datasketches/server/SketchStorage.java b/src/main/java/org/apache/datasketches/server/SketchStorage.java
index b6af0c5..e8bc491 100644
--- a/src/main/java/org/apache/datasketches/server/SketchStorage.java
+++ b/src/main/java/org/apache/datasketches/server/SketchStorage.java
@@ -58,29 +58,39 @@ public class SketchStorage {
}
static class SketchEntry {
- public final Family family;
- public final ValueType type;
- public Object sketch;
- public int configK;
-
- SketchEntry(final Family family, final ValueType type, final Object sketch, final int configK) throws IllegalArgumentException {
+ public final Family family_;
+ public final ValueType type_;
+ public Object sketch_;
+ public final String name_;
+ public int configK_;
+
+ SketchEntry(@NonNull final Family family,
+ final ValueType type,
+ @NonNull final Object sketch,
+ @NonNull final String name,
+ final int configK) throws IllegalArgumentException {
if (isDistinctCounting(family) && type == null)
throw new IllegalArgumentException("Must specify a value type for distinct counting sketches");
- this.family = family;
- this.type = type;
- this.sketch = sketch;
- this.configK = configK;
+ family_ = family;
+ type_ = type;
+ sketch_ = sketch;
+ name_ = name;
+ configK_ = configK;
}
- SketchEntry(final Family family, final Object sketch, final int configK) throws IllegalArgumentException {
+ SketchEntry(@NonNull final Family family,
+ @NonNull final Object sketch,
+ @NonNull final String name,
+ final int configK) throws IllegalArgumentException {
if (isDistinctCounting(family))
throw new IllegalArgumentException("Must specify a value type for distinct counting sketches");
- this.family = family;
- this.type = null;
- this.sketch = sketch;
- this.configK = configK;
+ family_ = family;
+ type_ = null;
+ sketch_ = sketch;
+ name_ = name;
+ configK_ = configK;
}
}
@@ -95,17 +105,17 @@ public class SketchStorage {
for (final Map.Entry<String, SketchEntry> e : sketchMap.entrySet()) {
final JsonObject item = new JsonObject();
item.addProperty(CONFIG_SKETCH_NAME_FIELD, e.getKey());
- switch (e.getValue().family) {
+ switch (e.getValue().family_) {
case UNION:
- item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type.getTypeName());
+ item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type_.getTypeName());
item.addProperty(CONFIG_FAMILY_FIELD, SKETCH_FAMILY_THETA);
break;
case CPC:
- item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type.getTypeName());
+ item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type_.getTypeName());
item.addProperty(CONFIG_FAMILY_FIELD, SKETCH_FAMILY_CPC);
break;
case HLL:
- item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type.getTypeName());
+ item.addProperty(CONFIG_TYPE_FIELD, e.getValue().type_.getTypeName());
item.addProperty(CONFIG_FAMILY_FIELD, SKETCH_FAMILY_HLL);
break;
case FREQUENCY:
@@ -149,38 +159,39 @@ public class SketchStorage {
SketchEntry sketchEntry = null;
final Family family = BaseSketchesQueryHandler.familyFromString(info.family);
+ final int k = info.k; // to reduce derferences in code later
switch (family) {
case QUICKSELECT:
// make a Union so we can handle merges later
sketchEntry = new SketchEntry(Family.UNION, ValueType.stringToType(info.type),
- new SetOperationBuilder().setNominalEntries(1 << info.k).buildUnion(), info.k);
+ new SetOperationBuilder().setNominalEntries(1 << k).buildUnion(), info.name, k);
break;
case HLL:
sketchEntry = new SketchEntry(Family.HLL, ValueType.stringToType(info.type),
- new HllSketch(info.k), info.k);
+ new HllSketch(k), info.name, k);
break;
case CPC:
sketchEntry = new SketchEntry(Family.CPC, ValueType.stringToType(info.type),
- new CpcSketch(info.k), info.k);
+ new CpcSketch(k), info.name, k);
break;
case KLL:
- sketchEntry = new SketchEntry(Family.KLL, new KllFloatsSketch(info.k), info.k);
+ sketchEntry = new SketchEntry(Family.KLL, new KllFloatsSketch(k), info.name, k);
break;
case FREQUENCY:
- sketchEntry = new SketchEntry(Family.FREQUENCY, new ItemsSketch<String>(info.k), info.k);
+ sketchEntry = new SketchEntry(Family.FREQUENCY, new ItemsSketch<String>(k), info.name, k);
break;
case RESERVOIR:
- sketchEntry = new SketchEntry(Family.RESERVOIR, ReservoirItemsSketch.<String>newInstance(info.k), info.k);
+ sketchEntry = new SketchEntry(Family.RESERVOIR, ReservoirItemsSketch.<String>newInstance(k), info.name, k);
break;
case VAROPT:
- sketchEntry = new SketchEntry(Family.VAROPT, VarOptItemsSketch.<String>newInstance(info.k), info.k);
+ sketchEntry = new SketchEntry(Family.VAROPT, VarOptItemsSketch.<String>newInstance(k), info.name, k);
break;
}
diff --git a/src/main/java/org/apache/datasketches/server/UpdateHandler.java b/src/main/java/org/apache/datasketches/server/UpdateHandler.java
index eb8c97a..84dc15a 100644
--- a/src/main/java/org/apache/datasketches/server/UpdateHandler.java
+++ b/src/main/java/org/apache/datasketches/server/UpdateHandler.java
@@ -80,10 +80,12 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
throw new IllegalArgumentException("Attempt to call update with missing name or sketch not found");
}
- if (data.isJsonArray()) {
- processBatchUpdate(se, data.getAsJsonArray());
- } else {
- processSingleUpdate(se, data);
+ synchronized (name.intern()) {
+ if (data.isJsonArray()) {
+ processBatchUpdate(se, data.getAsJsonArray());
+ } else {
+ processSingleUpdate(se, data);
+ }
}
}
@@ -94,54 +96,54 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
@SuppressWarnings("unchecked")
private static void processBatchUpdate(final SketchStorage.SketchEntry entry,
final JsonArray data) {
- switch (entry.family) {
+ switch (entry.family_) {
case UNION: // theta
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- for (final JsonElement e : data) { ((Union) entry.sketch).update(e.getAsDouble()); }
+ for (final JsonElement e : data) { ((Union) entry.sketch_).update(e.getAsDouble()); }
break;
case INT: case LONG:
- for (final JsonElement e : data) { ((Union) entry.sketch).update(e.getAsLong()); }
+ for (final JsonElement e : data) { ((Union) entry.sketch_).update(e.getAsLong()); }
break;
case STRING: default:
- for (final JsonElement e : data) { ((Union) entry.sketch).update(e.getAsString()); }
+ for (final JsonElement e : data) { ((Union) entry.sketch_).update(e.getAsString()); }
break;
}
break;
case CPC:
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- for (final JsonElement e : data) { ((CpcSketch) entry.sketch).update(e.getAsDouble()); }
+ for (final JsonElement e : data) { ((CpcSketch) entry.sketch_).update(e.getAsDouble()); }
break;
case INT: case LONG:
- for (final JsonElement e : data) { ((CpcSketch) entry.sketch).update(e.getAsLong()); }
+ for (final JsonElement e : data) { ((CpcSketch) entry.sketch_).update(e.getAsLong()); }
break;
case STRING: default:
- for (final JsonElement e : data) { ((CpcSketch) entry.sketch).update(e.getAsString()); }
+ for (final JsonElement e : data) { ((CpcSketch) entry.sketch_).update(e.getAsString()); }
break;
}
break;
case HLL:
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- for (final JsonElement e : data) { ((HllSketch) entry.sketch).update(e.getAsDouble()); }
+ for (final JsonElement e : data) { ((HllSketch) entry.sketch_).update(e.getAsDouble()); }
break;
case INT: case LONG:
- for (final JsonElement e : data) { ((CpcSketch) entry.sketch).update(e.getAsLong()); }
+ for (final JsonElement e : data) { ((CpcSketch) entry.sketch_).update(e.getAsLong()); }
break;
case STRING: default:
- for (final JsonElement e : data) { ((CpcSketch) entry.sketch).update(e.getAsString()); }
+ for (final JsonElement e : data) { ((CpcSketch) entry.sketch_).update(e.getAsString()); }
break;
}
break;
case KLL:
- for (final JsonElement e : data) { ((KllFloatsSketch) entry.sketch).update(e.getAsFloat()); }
+ for (final JsonElement e : data) { ((KllFloatsSketch) entry.sketch_).update(e.getAsFloat()); }
break;
case FREQUENCY:
@@ -154,15 +156,15 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
}
final String item = inputPair.get(QUERY_PAIR_ITEM_FIELD).getAsString();
final int weight = inputPair.get(QUERY_PAIR_WEIGHT_FIELD).getAsInt();
- ((ItemsSketch<String>) entry.sketch).update(item, weight);
+ ((ItemsSketch<String>) entry.sketch_).update(item, weight);
} else {
- ((ItemsSketch<String>) entry.sketch).update(e.getAsString());
+ ((ItemsSketch<String>) entry.sketch_).update(e.getAsString());
}
}
break;
case RESERVOIR:
- for (final JsonElement e : data) { ((ReservoirItemsSketch<String>) entry.sketch).update(e.getAsString()); }
+ for (final JsonElement e : data) { ((ReservoirItemsSketch<String>) entry.sketch_).update(e.getAsString()); }
break;
case VAROPT:
@@ -175,69 +177,69 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
}
final String item = inputPair.get(QUERY_PAIR_ITEM_FIELD).getAsString();
final double weight = inputPair.get(QUERY_PAIR_WEIGHT_FIELD).getAsDouble();
- ((VarOptItemsSketch<String>) entry.sketch).update(item, weight);
+ ((VarOptItemsSketch<String>) entry.sketch_).update(item, weight);
} else {
- ((VarOptItemsSketch<String>) entry.sketch).update(e.getAsString(), 1.0);
+ ((VarOptItemsSketch<String>) entry.sketch_).update(e.getAsString(), 1.0);
}
}
break;
default:
- throw new IllegalArgumentException("Unsupported sketch type: " + entry.family);
+ throw new IllegalArgumentException("Unsupported sketch type: " + entry.family_);
}
}
@SuppressWarnings("unchecked")
private static void processSingleUpdate(final SketchStorage.SketchEntry entry,
final JsonElement data) {
- switch (entry.family) {
+ switch (entry.family_) {
case UNION:
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- ((Union) entry.sketch).update(data.getAsDouble());
+ ((Union) entry.sketch_).update(data.getAsDouble());
break;
case INT: case LONG:
- ((Union) entry.sketch).update(data.getAsLong());
+ ((Union) entry.sketch_).update(data.getAsLong());
break;
case STRING: default:
- ((Union) entry.sketch).update(data.getAsString());
+ ((Union) entry.sketch_).update(data.getAsString());
break;
}
break;
case CPC:
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- ((CpcSketch) entry.sketch).update(data.getAsDouble());
+ ((CpcSketch) entry.sketch_).update(data.getAsDouble());
break;
case INT: case LONG:
- ((CpcSketch) entry.sketch).update(data.getAsLong());
+ ((CpcSketch) entry.sketch_).update(data.getAsLong());
break;
case STRING: default:
- ((CpcSketch) entry.sketch).update(data.getAsString());
+ ((CpcSketch) entry.sketch_).update(data.getAsString());
break;
}
break;
case HLL:
- assert(entry.type != null);
- switch (entry.type) {
+ assert(entry.type_ != null);
+ switch (entry.type_) {
case FLOAT: case DOUBLE:
- ((HllSketch) entry.sketch).update(data.getAsDouble());
+ ((HllSketch) entry.sketch_).update(data.getAsDouble());
break;
case INT: case LONG:
- ((HllSketch) entry.sketch).update(data.getAsLong());
+ ((HllSketch) entry.sketch_).update(data.getAsLong());
break;
case STRING: default:
- ((HllSketch) entry.sketch).update(data.getAsString());
+ ((HllSketch) entry.sketch_).update(data.getAsString());
break;
}
break;
case KLL:
- ((KllFloatsSketch) entry.sketch).update(data.getAsFloat());
+ ((KllFloatsSketch) entry.sketch_).update(data.getAsFloat());
break;
case FREQUENCY:
@@ -249,14 +251,14 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
}
final String item = inputPair.get(QUERY_PAIR_ITEM_FIELD).getAsString();
final int weight = inputPair.get(QUERY_PAIR_WEIGHT_FIELD).getAsInt();
- ((ItemsSketch<String>) entry.sketch).update(item, weight);
+ ((ItemsSketch<String>) entry.sketch_).update(item, weight);
} else {
- ((ItemsSketch<String>) entry.sketch).update(data.getAsString());
+ ((ItemsSketch<String>) entry.sketch_).update(data.getAsString());
}
break;
case RESERVOIR:
- ((ReservoirItemsSketch<String>) entry.sketch).update(data.getAsString());
+ ((ReservoirItemsSketch<String>) entry.sketch_).update(data.getAsString());
break;
case VAROPT:
@@ -268,14 +270,14 @@ public class UpdateHandler extends BaseSketchesQueryHandler {
}
final String item = inputPair.get(QUERY_PAIR_ITEM_FIELD).getAsString();
final double weight = inputPair.get(QUERY_PAIR_WEIGHT_FIELD).getAsDouble();
- ((VarOptItemsSketch<String>) entry.sketch).update(item, weight);
+ ((VarOptItemsSketch<String>) entry.sketch_).update(item, weight);
} else {
- ((VarOptItemsSketch<String>) entry.sketch).update(data.getAsString(), 1.0);
+ ((VarOptItemsSketch<String>) entry.sketch_).update(data.getAsString(), 1.0);
}
break;
default:
- throw new IllegalArgumentException("Unsupported sketch type: " + entry.family);
+ throw new IllegalArgumentException("Unsupported sketch type: " + entry.family_);
}
}
diff --git a/src/test/java/org/apache/datasketches/server/SketchServerTest.java b/src/test/java/org/apache/datasketches/server/SketchServerTest.java
index 61a6d04..4a2a15d 100644
--- a/src/test/java/org/apache/datasketches/server/SketchServerTest.java
+++ b/src/test/java/org/apache/datasketches/server/SketchServerTest.java
@@ -37,7 +37,7 @@ public class SketchServerTest {
final ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
SketchServer server = null;
try {
- server = new SketchServer(Objects.requireNonNull(classLoader.getResource("test_config.json")).getFile());
+ server = new SketchServer(Objects.requireNonNull(classLoader.getResource("config_with_port.json")).getFile());
} catch (final IOException e) {
fail();
}
diff --git a/src/test/java/org/apache/datasketches/server/SketchStorageTest.java b/src/test/java/org/apache/datasketches/server/SketchStorageTest.java
index 39076b2..03f4b5c 100644
--- a/src/test/java/org/apache/datasketches/server/SketchStorageTest.java
+++ b/src/test/java/org/apache/datasketches/server/SketchStorageTest.java
@@ -41,14 +41,14 @@ public class SketchStorageTest {
@Test
public void invalidSketchEntry() {
try {
- new SketchStorage.SketchEntry(Family.CPC, null, new CpcSketch(12), 12);
+ new SketchStorage.SketchEntry(Family.CPC, null, new CpcSketch(12), "cpcSketch", 12);
fail();
} catch (IllegalArgumentException e) {
// expected
}
try {
- new SketchStorage.SketchEntry(Family.HLL, new HllSketch(10), 10);
+ new SketchStorage.SketchEntry(Family.HLL, new HllSketch(10), "hllSketch", 10);
fail();
} catch (IllegalArgumentException e) {
// expected
diff --git a/src/test/java/org/apache/datasketches/server/UpdateHandlerTest.java b/src/test/java/org/apache/datasketches/server/UpdateHandlerTest.java
index 49969bb..a77369b 100644
--- a/src/test/java/org/apache/datasketches/server/UpdateHandlerTest.java
+++ b/src/test/java/org/apache/datasketches/server/UpdateHandlerTest.java
@@ -70,7 +70,7 @@ public class UpdateHandlerTest extends ServerTestBase {
assertTrue(element.isJsonNull());
final SketchStorage.SketchEntry entry = server_.getSketch(sketchName);
- final CpcSketch sk = (CpcSketch) entry.sketch;
+ final CpcSketch sk = (CpcSketch) entry.sketch_;
assertEquals(sk.getEstimate(), 2 * nPoints, 2 * nPoints * 1e-2);
}
@@ -108,7 +108,7 @@ public class UpdateHandlerTest extends ServerTestBase {
assertTrue(element.isJsonNull());
final SketchStorage.SketchEntry entry = server_.getSketch(sketchName);
- final org.apache.datasketches.frequencies.ItemsSketch<String> sk = (org.apache.datasketches.frequencies.ItemsSketch<String>) entry.sketch;
+ final org.apache.datasketches.frequencies.ItemsSketch<String> sk = (org.apache.datasketches.frequencies.ItemsSketch<String>) entry.sketch_;
assertEquals(sk.getEstimate("item1"), 2);
assertEquals(sk.getEstimate("item2"), 5);
assertEquals(sk.getEstimate("item3"), 10);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datasketches.apache.org
For additional commands, e-mail: commits-help@datasketches.apache.org