Add support for tunable write consistency (#123)

Author: parkerduckworth

batch/journey.test.js

@@ -179,9 +179,10 @@ describe("batch importing", () => {
         },
       ];
 
-      it("imports them", () => {
+      it("imports them with consistency level", () => {
         client.batch
           .objectsBatcher()
+          .withConsistencyLevel(weaviate.replication.ConsistencyLevel.ONE)
           .withObjects([toImport[0], toImport[1]])
           .do()
           .then()
@@ -215,7 +216,7 @@ describe("batch importing", () => {
       })
     });
 
-    it("imports the refs with raw objects", () => {
+    it("imports the refs with raw objects and consistency level", () => {
       return client.batch
         .referencesBatcher()
         .withReference({
@@ -226,6 +227,7 @@ describe("batch importing", () => {
           from: `weaviate://localhost/${thingClassName}/${thingIds[1]}/refProp`,
           to: `weaviate://localhost/${otherThingClassName}/${otherThingIds[1]}`,
         })
+        .withConsistencyLevel(weaviate.replication.ConsistencyLevel.ALL)
         .do()
         .then((res) => {
           res.forEach((elem) => {
@@ -429,7 +431,7 @@ describe("batch deleting", () => {
       })
   )
 
-  it("batch deletes with default dryRun", () => {
+  it("batch deletes with default dryRun and consistency level", () => {
     const inAMinute = "" + (new Date().getTime() + 60 * 1000);
     return client.batch
       .objectsBatchDeleter()
@@ -440,6 +442,7 @@ describe("batch deleting", () => {
         path: ["_creationTimeUnix"]
       })
       .withOutput(weaviate.batch.DeleteOutput.VERBOSE)
+      .withConsistencyLevel(weaviate.replication.ConsistencyLevel.QUORUM)
       .do()
       .then(result => {
         expect(result.dryRun).toBe(false);

batch/objectsBatchDeleter.js

@@ -1,4 +1,5 @@
 import { isValidStringProperty } from "../validation/string";
+import { buildObjectsPath } from "./path"
 
 export default class ObjectsBatchDeleter {
   className;
@@ -31,6 +32,11 @@ export default class ObjectsBatchDeleter {
     return this;
   }
 
+  withConsistencyLevel = (cl) => {
+    this.consistencyLevel = cl;
+    return this;
+  };
+
   payload() {
     return {
       match: {
@@ -72,7 +78,11 @@ export default class ObjectsBatchDeleter {
         new Error("invalid usage: " + this.errors.join(", "))
       );
     }
-    const path = `/batch/objects`;
+    let params = new URLSearchParams()
+    if (this.consistencyLevel) {
+      params.set("consistency_level", this.consistencyLevel)
+    }
+    const path = buildObjectsPath(params);
     return this.client.delete(path, this.payload(), true);
   };
 }

batch/objectsBatcher.js

@@ -1,3 +1,5 @@
+import { buildObjectsPath } from "./path"
+
 export default class ObjectsBatcher {
   constructor(client) {
     this.client = client;
@@ -25,6 +27,11 @@ export default class ObjectsBatcher {
     return this.withObjects(object);
   };
 
+  withConsistencyLevel = (cl) => {
+    this.consistencyLevel = cl;
+    return this;
+  };
+
   payload = () => ({
     objects: this.objects,
   });
@@ -50,7 +57,11 @@ export default class ObjectsBatcher {
         new Error("invalid usage: " + this.errors.join(", "))
       );
     }
-    const path = `/batch/objects`;
+    let params = new URLSearchParams()
+    if (this.consistencyLevel) {
+      params.set("consistency_level", this.consistencyLevel)
+    }
+    const path = buildObjectsPath(params);
     return this.client.post(path, this.payload());
   };
 }

batch/path.js

@@ -0,0 +1,16 @@
+export function buildObjectsPath(queryParams) {
+  let path = "/batch/objects";
+  return buildPath(path, queryParams)
+}
+
+export function buildRefsPath(queryParams) {
+  let path = "/batch/references";
+  return buildPath(path, queryParams)
+}
+
+function buildPath(path, queryParams) {
+  if (queryParams && queryParams.toString() != "") {
+    path = `${path}?${queryParams.toString()}`
+  }
+  return path
+}

batch/path.test.js

@@ -0,0 +1,27 @@
+import { buildObjectsPath, buildRefsPath } from "./path"
+
+describe("paths", () => {
+  it("builds batch objects without params", () => {
+    let path = buildObjectsPath(new URLSearchParams())
+    expect(path).toEqual("/batch/objects")
+  })
+
+  it("builds batch objects with params", () => {
+    let path = buildObjectsPath(new URLSearchParams({
+      consistency_level: "ONE"
+    }))
+    expect(path).toEqual("/batch/objects?consistency_level=ONE")
+  })
+
+  it("builds batch references without params", () => {
+    let path = buildRefsPath(new URLSearchParams())
+    expect(path).toEqual("/batch/references")
+  })
+
+  it("builds batch object with params", () => {
+    let path = buildRefsPath(new URLSearchParams({
+      consistency_level: "ONE"
+    }))
+    expect(path).toEqual("/batch/references?consistency_level=ONE")
+  })
+})

batch/referencesBatcher.js

@@ -1,3 +1,5 @@
+import { buildRefsPath } from "./path"
+
 export default class ReferencesBatcher {
   constructor(client, beaconPath) {
     this.client = client;
@@ -26,6 +28,11 @@ export default class ReferencesBatcher {
     return this.withReferences(reference);
   }
 
+  withConsistencyLevel = (cl) => {
+    this.consistencyLevel = cl;
+    return this;
+  };
+
   payload = () => this.references;
 
   validateReferenceCount = () => {
@@ -49,7 +56,11 @@ export default class ReferencesBatcher {
         new Error("invalid usage: " + this.errors.join(", "))
       );
     }
-    const path = `/batch/references`;
+    let params = new URLSearchParams()
+    if (this.consistencyLevel) {
+      params.set("consistency_level", this.consistencyLevel)
+    }
+    const path = buildRefsPath(params);
     const payloadPromise = Promise.all(this.references.map(ref => this.rebuildReferencePromise(ref)));
 
     return payloadPromise.then(payload => this.client.post(path, payload));

ci/docker-compose-azure-cc.yml

@@ -11,7 +11,7 @@ services:
       - --scheme
       - http
       - --write-timeout=600s
-    image: semitechnologies/weaviate:preview-replace-shardingconfig-replicas-with-replication-factor-29e987d
+    image: semitechnologies/weaviate:1.18.0-alpha.0-be532d2
     ports:
       - 8081:8081
     restart: on-failure:0

ci/docker-compose-okta-cc.yml

@@ -10,7 +10,7 @@ services:
       - --scheme
       - http
       - --write-timeout=600s
-    image: semitechnologies/weaviate:1.15.4-b7811d4
+    image: semitechnologies/weaviate:1.18.0-alpha.0-be532d2
     ports:
       - 8082:8082
     restart: on-failure:0

ci/docker-compose-okta-users.yml

@@ -10,7 +10,7 @@ services:
       - --scheme
       - http
       - --write-timeout=600s
-    image: semitechnologies/weaviate:1.17.0
+    image: semitechnologies/weaviate:1.18.0-alpha.0-be532d2
     ports:
       - 8083:8083
     restart: on-failure:0

ci/docker-compose-wcs-noscope.yml

@@ -10,7 +10,7 @@ services:
       - --scheme
       - http
       - --write-timeout=600s
-    image: semitechnologies/weaviate:1.17.0
+    image: semitechnologies/weaviate:1.18.0-alpha.0-be532d2
     ports:
       - 8086:8086
     restart: on-failure:0