Merge pull request #537 from weaviate/feat/ssb

Server-side batching (SSB)

Author: bevzzz

.github/workflows/test.yaml

@@ -15,7 +15,32 @@ env:
   DOCKER_IMAGES_TAR: docker-images.tar
 
 jobs:
+  maven-cache:
+    name: Cache Maven dependencies
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v4
+        with:
+          distribution: "zulu"
+          java-version: "17"
+          cache: "maven"
+      - run: mvn dependency:go-offline
+
+  unit:
+    name: Unit tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-java@v4
+        with:
+          distribution: "zulu"
+          java-version: "17"
+          cache: "maven"
+      - run: mvn test -Dtest='!*ITest'
+
   docker-cache:
+    needs: unit
     name: Cache shared Docker images
     runs-on: ubuntu-latest
     steps:
@@ -59,22 +84,10 @@ jobs:
           path: ${{ env.DOCKER_IMAGES_TAR }}
           key: ${{ env.DOCKER_CACHE_KEY }}
 
-  maven-cache:
-    name: Cache Maven dependencies
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-java@v4
-        with:
-          distribution: "zulu"
-          java-version: "17"
-          cache: "maven"
-      - run: mvn dependency:go-offline
-
-  test:
-    name: Test
+  integration:
+    name: Integration Tests
     runs-on: ubuntu-latest
-    needs: [docker-cache, maven-cache]
+    needs: [maven-cache, unit, docker-cache]
     strategy:
       fail-fast: false
       matrix:
@@ -118,4 +131,4 @@ jobs:
           AZURE_CLIENT_SECRET: ${{ secrets.AZURE_CLIENT_SECRET }}
           OPENAI_APIKEY: ${{ secrets.OPENAI_APIKEY }}
           WEAVIATE_VERSION: ${{ matrix.WEAVIATE_VERSION }}
-        run: mvn verify -Dgpg.skip
+        run: mvn verify -Dtest='*ITest' -Dgpg.skip

pom.xml

@@ -170,9 +170,8 @@
     </dependency>
     <dependency>
         <groupId>org.slf4j</groupId>
-        <artifactId>slf4j-nop</artifactId>
+        <artifactId>slf4j-api</artifactId>
         <version>${slf4j.version}</version>
-        <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.mock-server</groupId>
@@ -237,13 +236,11 @@
           <artifactId>maven-surefire-plugin</artifactId>
           <version>3.5.4</version>
           <configuration>
-            <trimStackTrace>false</trimStackTrace>
-            <parallel>classes</parallel>
-            <forkCount>4</forkCount>
-            <reuseForks>true</reuseForks>
-            <threadCount>1</threadCount>
-            <perCoreThreadCount>true</perCoreThreadCount>
-            <runOrder>balanced</runOrder>
+            <systemPropertyVariables>
+              <java.util.logging.config.file>
+                ${project.basedir}/src/test/resources/logging.properties
+              </java.util.logging.config.file>
+            </systemPropertyVariables>
           </configuration>
         </plugin>
         <plugin>
@@ -546,5 +543,5 @@
         <artifactId>central-publishing-maven-plugin</artifactId>
       </plugin>
     </plugins>
-  </build>
+    </build>
 </project>

src/it/java/io/weaviate/containers/Container.java

@@ -19,6 +19,7 @@ public class Container {
   public static final MinIo MINIO = MinIo.createDefault();
 
   public static ContainerGroup compose(Weaviate weaviate, GenericContainer<?>... containers) {
+    assert weaviate != WEAVIATE : "cannot compose with static WEAVIATE contaier";
     return new ContainerGroup(weaviate, containers);
   }
 
@@ -50,7 +51,9 @@ public void start() {
     @Override
     public void stop() {
       weaviate.stop();
-      containers.forEach(GenericContainer::stop);
+      containers.stream()
+          .dropWhile(c -> c == MODEL2VEC || c == IMG2VEC_NEURAL || c == MINIO)
+          .forEach(GenericContainer::stop);
     }
 
     private void setSharedNetwork() {

src/it/java/io/weaviate/containers/Weaviate.java

@@ -28,9 +28,11 @@ public class Weaviate extends WeaviateContainer {
   public static final String DOCKER_IMAGE = "semitechnologies/weaviate";
   public static final String LATEST_VERSION = Version.latest().semver.toString();
   public static final String VERSION;
+  private static final boolean DEBUG;
 
   static {
     VERSION = System.getenv().getOrDefault("WEAVIATE_VERSION", LATEST_VERSION);
+    DEBUG = System.getenv("DEBUG") != null;
   }
   public static String OIDC_ISSUER = "https://auth.wcs.api.weaviate.io/auth/realms/SeMI";
 
@@ -93,6 +95,9 @@ public String getContainerName() {
   public WeaviateClient getClient() {
     if (!isRunning()) {
       start();
+      if (DEBUG) {
+        followOutput(frame -> System.out.println(frame.getUtf8String()));
+      }
     }
     if (clientInstance != null) {
       return clientInstance;

src/it/java/io/weaviate/integration/BackupITest.java

@@ -7,15 +7,16 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.stream.IntStream;
 
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.AfterClass;
 import org.junit.Test;
 
-import com.sun.nio.sctp.IllegalUnbindException;
-
 import io.weaviate.ConcurrentTest;
 import io.weaviate.client6.v1.api.WeaviateClient;
 import io.weaviate.client6.v1.api.backup.Backup;
@@ -25,10 +26,16 @@
 import io.weaviate.containers.Weaviate;
 
 public class BackupITest extends ConcurrentTest {
+  private static final ExecutorService EXEC = Executors.newSingleThreadExecutor();
   private static final WeaviateClient client = Weaviate.custom()
       .withFilesystemBackup("/tmp/backups").build()
       .getClient();
 
+  @AfterClass
+  public static void tearDown() {
+    EXEC.shutdownNow();
+  }
+
   @Test
   public void test_lifecycle() throws IOException, TimeoutException {
     Weaviate.Version.V132.orSkip(); // List backups not implemented in earlier versions
@@ -258,6 +265,6 @@ private CompletableFuture<Void> spamData(String collectionName) {
         throw new CompletionException(e);
       }
       return null;
-    });
+    }, EXEC);
   }
 }

src/it/java/io/weaviate/integration/BatchITest.java

@@ -0,0 +1,41 @@
+package io.weaviate.integration;
+
+import java.io.IOException;
+
+import org.assertj.core.api.Assertions;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import io.weaviate.ConcurrentTest;
+import io.weaviate.client6.v1.api.WeaviateClient;
+import io.weaviate.client6.v1.api.collections.WeaviateObject;
+import io.weaviate.client6.v1.api.collections.batch.BatchContext;
+import io.weaviate.containers.Container;
+import io.weaviate.containers.Weaviate;
+
+public class BatchITest extends ConcurrentTest {
+  private static final WeaviateClient client = Container.WEAVIATE.getClient();
+
+  @BeforeClass
+  public static void __() {
+    Weaviate.Version.V136.orSkip();
+  }
+
+  @Test
+  public void test10_000Objects() throws IOException {
+    var nsThings = ns("Things");
+
+    var things = client.collections.create(nsThings);
+
+    // Act
+    try (BatchContext<?> batch = things.batch.start()) {
+      for (int i = 0; i < 10_000; i++) {
+        batch.add(WeaviateObject.of());
+      }
+    } catch (InterruptedException e) {
+    }
+
+    // Assert
+    Assertions.assertThat(things.size()).isEqualTo(10_000);
+  }
+}

src/it/java/io/weaviate/integration/ClusterITest.java

@@ -3,6 +3,7 @@
 import java.io.IOException;
 
 import org.assertj.core.api.Assertions;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import io.weaviate.ConcurrentTest;
@@ -17,7 +18,13 @@
 import io.weaviate.containers.Weaviate.Version;
 
 public class ClusterITest extends ConcurrentTest {
-  private static final WeaviateClient client = Weaviate.cluster(3).getClient();
+  private static final Weaviate cluster = Weaviate.cluster(3);
+  private static final WeaviateClient client = cluster.getClient();
+
+  @AfterClass
+  public static void tearDown() {
+    cluster.stop();
+  }
 
   @Test
   public void test_shardingState() throws IOException {

src/it/java/io/weaviate/integration/CollectionsITest.java

@@ -370,8 +370,8 @@ public void test_dropPropertyIndex() throws IOException {
             .returns(false, Property::indexFilterable)
             .returns(false, Property::indexSearchable)
             .returns(false, Property::indexRangeFilters));
-    }
- 
+  }
+
   @Test
   public void test_asyncReplicationConfig() throws IOException {
     Weaviate.Version.latest().orSkip();

src/it/java/io/weaviate/integration/DataITest.java

@@ -486,7 +486,7 @@ public void testDataTypes() throws IOException {
         .asInstanceOf(InstanceOfAssertFactories.map(String.class, Object.class))
         // Most of PhoneNumber fields are only present on read and are null on write.
         .usingRecursiveComparison()
-        .withComparatorForType(ORMITest::comparePhoneNumbers, PhoneNumber.class)
+        .withComparatorForType(OrmITest::comparePhoneNumbers, PhoneNumber.class)
         .isEqualTo(want);
   }
 

…aviate/integration/OIDCSupportITest.java …aviate/integration/OidcSupportITest.javasrc/it/java/io/weaviate/integration/OIDCSupportITest.java renamed to src/it/java/io/weaviate/integration/OidcSupportITest.java src/it/java/io/weaviate/integration/OIDCSupportITest.java renamed to src/it/java/io/weaviate/integration/OidcSupportITest.java

@@ -9,6 +9,7 @@
 
 import org.assertj.core.api.Assertions;
 import org.junit.Assume;
+import org.junit.Ignore;
 import org.junit.Test;
 
 import io.weaviate.ConcurrentTest;
@@ -28,7 +29,7 @@
  * Additionally, {@code WCS_DUMMY_CI_PW} and {@code OKTA_CLIENT_SECRET}
  * environment variables must be set.
  */
-public class OIDCSupportITest extends ConcurrentTest {
+public class OidcSupportITest extends ConcurrentTest {
   private static final String WCS_DUMMY_CI_USERNAME = "oidc-test-user@weaviate.io";
   private static final String WCS_DUMMY_CI_PW = System.getenv("WCS_DUMMY_CI_PW");
 
@@ -66,10 +67,7 @@ public void test_bearerToken() throws Exception {
     var auth = SpyTokenProvider.spyOn(Authentication.bearerToken(t.accessToken(), t.refreshToken(), 0));
     pingWeaviate(wcsContainer, auth);
 
-    var newT = auth.getToken();
-    Assertions.assertThat(newT.accessToken())
-        .as("expect access_token was refreshed")
-        .isNotEqualTo(t.accessToken());
+    eventually(() -> auth.getToken() != t, 100, 5, "expect access_token was refreshed");
 
     // Check that the new token authenticates requests.
     pingWeaviate(wcsContainer, auth);
@@ -81,7 +79,7 @@ public void test_resourceOwnerPassword() throws Exception {
     Assume.assumeTrue("WCS_DUMMY_CI_PW is not set", WCS_DUMMY_CI_PW != null && !WCS_DUMMY_CI_PW.isBlank());
     Assume.assumeTrue("no internet connection", hasInternetConnection());
 
-    // Check norwal resource owner password flow works.
+    // Check normal resource owner password flow works.
     var password = Authentication.resourceOwnerPassword(WCS_DUMMY_CI_USERNAME, WCS_DUMMY_CI_PW, List.of());
     var auth = SpyTokenProvider.spyOn(password);
     pingWeaviate(wcsContainer, auth);