Hook up new grpc backend, fix aggregate.hybrid bug, update CI images

Author: tsmith023

.github/workflows/main.yaml

@@ -8,10 +8,11 @@ on:
 
 env:
   WEAVIATE_124: 1.24.26
-  WEAVIATE_125: 1.25.28
-  WEAVIATE_126: 1.26.13
-  WEAVIATE_127: 1.27.9
-  WEAVIATE_128: 1.28.2
+  WEAVIATE_125: 1.25.30
+  WEAVIATE_126: 1.26.14
+  WEAVIATE_127: 1.27.11
+  WEAVIATE_128: 1.28.4
+  WEAVIATE_129: 1.29.0-rc.0
 
 jobs:
   checks:

src/collections/aggregate/index.ts

@@ -1,14 +1,16 @@
-import Connection from '../../connection/index.js';
+import Connection from '../../connection/grpc.js';
 
 import { ConsistencyLevel } from '../../data/index.js';
 import { DbVersionSupport } from '../../utils/dbVersion.js';
 
 import { FilterValue } from '../filters/index.js';
 
-import { WeaviateQueryError } from '../../errors.js';
+import { WeaviateInvalidInputError, WeaviateQueryError } from '../../errors.js';
 import { Aggregator } from '../../graphql/index.js';
 import { PrimitiveKeys, toBase64FromMedia } from '../../index.js';
-import { Bm25QueryProperty } from '../query/types.js';
+import { Deserialize } from '../deserialize/index.js';
+import { Bm25QueryProperty, NearVectorInputType } from '../query/types.js';
+import { NearVectorInputGuards } from '../query/utils.js';
 import { Serialize } from '../serialize/index.js';
 
 export type AggregateBaseOptions<M> = {
@@ -63,10 +65,10 @@ export type AggregateBoolean = {
 
 export type AggregateDate = {
   count?: number;
-  maximum?: number;
-  median?: number;
-  minimum?: number;
-  mode?: number;
+  maximum?: string;
+  median?: string;
+  minimum?: string;
+  mode?: string;
 };
 
 export type AggregateNumber = {
@@ -87,7 +89,7 @@ export type AggregateText = {
   count?: number;
   topOccurrences?: {
     occurs?: number;
-    value?: number;
+    value?: string;
   }[];
 };
 
@@ -345,6 +347,7 @@ class AggregateManager<T> implements Aggregate<T> {
   dbVersionSupport: DbVersionSupport;
   consistencyLevel?: ConsistencyLevel;
   tenant?: string;
+  grpcChecker: Promise<boolean>;
 
   private constructor(
     connection: Connection,
@@ -359,6 +362,8 @@ class AggregateManager<T> implements Aggregate<T> {
     this.consistencyLevel = consistencyLevel;
     this.tenant = tenant;
 
+    this.grpcChecker = this.dbVersionSupport.supportsAggregateGRPC().then((res) => res.supports);
+
     this.groupBy = {
       hybrid: <M extends PropertiesMetrics<T> | undefined = undefined>(
         query: string,
@@ -446,13 +451,15 @@ class AggregateManager<T> implements Aggregate<T> {
     };
   }
 
-  query() {
+  private grpc = () => this.connection.aggregate(this.name, this.consistencyLevel, this.tenant);
+
+  private gql() {
     return new Aggregator(this.connection);
   }
 
   base(metrics?: PropertiesMetrics<T>, filters?: FilterValue, groupBy?: PropertyOf<T> | GroupByAggregate<T>) {
     let fields = 'meta { count }';
-    let builder = this.query().withClassName(this.name);
+    let builder = this.gql().withClassName(this.name);
     if (metrics) {
       if (Array.isArray(metrics)) {
         fields += metrics.map((m) => this.metrics(m)).join(' ');
@@ -516,10 +523,15 @@ class AggregateManager<T> implements Aggregate<T> {
     return new AggregateManager<T>(connection, name, dbVersionSupport, consistencyLevel, tenant);
   }
 
-  hybrid<M extends PropertiesMetrics<T>>(
+  async hybrid<M extends PropertiesMetrics<T>>(
     query: string,
     opts?: AggregateHybridOptions<T, M>
   ): Promise<AggregateResult<T, M>> {
+    if (await this.grpcChecker) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withHybrid(Serialize.aggregate.hybrid(query, opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
     let builder = this.base(opts?.returnMetrics, opts?.filters).withHybrid({
       query: query,
       alpha: opts?.alpha,
@@ -538,8 +550,14 @@ class AggregateManager<T> implements Aggregate<T> {
     image: string | Buffer,
     opts?: AggregateNearOptions<M>
   ): Promise<AggregateResult<T, M>> {
+    const [b64, usesGrpc] = await Promise.all([await toBase64FromMedia(image), await this.grpcChecker]);
+    if (usesGrpc) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withNearImage(Serialize.aggregate.nearImage(b64, opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
     const builder = this.base(opts?.returnMetrics, opts?.filters).withNearImage({
-      image: await toBase64FromMedia(image),
+      image: b64,
       certainty: opts?.certainty,
       distance: opts?.distance,
       targetVectors: opts?.targetVector ? [opts.targetVector] : undefined,
@@ -550,10 +568,15 @@ class AggregateManager<T> implements Aggregate<T> {
     return this.do(builder);
   }
 
-  nearObject<M extends PropertiesMetrics<T>>(
+  async nearObject<M extends PropertiesMetrics<T>>(
     id: string,
     opts?: AggregateNearOptions<M>
   ): Promise<AggregateResult<T, M>> {
+    if (await this.grpcChecker) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withNearObject(Serialize.aggregate.nearObject(id, opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
     const builder = this.base(opts?.returnMetrics, opts?.filters).withNearObject({
       id: id,
       certainty: opts?.certainty,
@@ -566,10 +589,15 @@ class AggregateManager<T> implements Aggregate<T> {
     return this.do(builder);
   }
 
-  nearText<M extends PropertiesMetrics<T>>(
+  async nearText<M extends PropertiesMetrics<T>>(
     query: string | string[],
     opts?: AggregateNearOptions<M>
   ): Promise<AggregateResult<T, M>> {
+    if (await this.grpcChecker) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withNearText(Serialize.aggregate.nearText(query, opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
     const builder = this.base(opts?.returnMetrics, opts?.filters).withNearText({
       concepts: Array.isArray(query) ? query : [query],
       certainty: opts?.certainty,
@@ -582,10 +610,20 @@ class AggregateManager<T> implements Aggregate<T> {
     return this.do(builder);
   }
 
-  nearVector<M extends PropertiesMetrics<T>>(
-    vector: number[],
+  async nearVector<M extends PropertiesMetrics<T>>(
+    vector: NearVectorInputType,
     opts?: AggregateNearOptions<M>
   ): Promise<AggregateResult<T, M>> {
+    if (await this.grpcChecker) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withNearVector(Serialize.aggregate.nearVector(vector, opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
+    if (!NearVectorInputGuards.is1DArray(vector)) {
+      throw new WeaviateInvalidInputError(
+        'Vector can only be a 1D array of numbers when using `nearVector` with <1.29 Weaviate versions.'
+      );
+    }
     const builder = this.base(opts?.returnMetrics, opts?.filters).withNearVector({
       vector: vector,
       certainty: opts?.certainty,
@@ -598,9 +636,15 @@ class AggregateManager<T> implements Aggregate<T> {
     return this.do(builder);
   }
 
-  overAll<M extends PropertiesMetrics<T>>(opts?: AggregateOverAllOptions<M>): Promise<AggregateResult<T, M>> {
-    const builder = this.base(opts?.returnMetrics, opts?.filters);
-    return this.do(builder);
+  async overAll<M extends PropertiesMetrics<T>>(
+    opts?: AggregateOverAllOptions<M>
+  ): Promise<AggregateResult<T, M>> {
+    if (await this.grpcChecker) {
+      return this.grpc()
+        .then((aggregate) => aggregate.withFetch(Serialize.aggregate.overAll(opts)))
+        .then((reply) => Deserialize.aggregate(reply));
+    }
+    return this.do(this.base(opts?.returnMetrics, opts?.filters));
   }
 
   do = <M extends PropertiesMetrics<T> | undefined = undefined>(

src/collections/aggregate/integration.test.ts

@@ -442,7 +442,7 @@ describe('Testing of collection.aggregate search methods', () => {
     }
     const result = await collection.aggregate.hybrid('test', {
       alpha: 0.5,
-      maxVectorDistance: 0,
+      maxVectorDistance: 1,
       queryProperties: ['text'],
       returnMetrics: collection.metrics.aggregate('text').text(['count']),
     });

src/collections/deserialize/index.ts

@@ -1,11 +1,29 @@
 import { WeaviateDeserializationError } from '../../errors.js';
 import { Tenant as TenantREST } from '../../openapi/types.js';
+import {
+  AggregateReply,
+  AggregateReply_Aggregations_Aggregation,
+  AggregateReply_Aggregations_Aggregation_Boolean,
+  AggregateReply_Aggregations_Aggregation_DateMessage,
+  AggregateReply_Aggregations_Aggregation_Integer,
+  AggregateReply_Aggregations_Aggregation_Number,
+  AggregateReply_Aggregations_Aggregation_Text,
+} from '../../proto/v1/aggregate.js';
 import { BatchObject as BatchObjectGRPC, BatchObjectsReply } from '../../proto/v1/batch.js';
 import { BatchDeleteReply } from '../../proto/v1/batch_delete.js';
 import { ListValue, Properties as PropertiesGrpc, Value } from '../../proto/v1/properties.js';
 import { MetadataResult, PropertiesResult, SearchReply } from '../../proto/v1/search_get.js';
 import { TenantActivityStatus, TenantsGetReply } from '../../proto/v1/tenants.js';
 import { DbVersionSupport } from '../../utils/dbVersion.js';
+import {
+  AggregateBoolean,
+  AggregateDate,
+  AggregateNumber,
+  AggregateResult,
+  AggregateText,
+  AggregateType,
+  PropertiesMetrics,
+} from '../index.js';
 import { referenceFromObjects } from '../references/utils.js';
 import { Tenant } from '../tenants/index.js';
 import {
@@ -36,6 +54,96 @@ export class Deserialize {
     return new Deserialize(supports125ListValue);
   }
 
+  private static aggregateBoolean(
+    aggregation: AggregateReply_Aggregations_Aggregation_Boolean
+  ): AggregateBoolean {
+    return {
+      count: aggregation.count,
+      percentageFalse: aggregation.percentageFalse,
+      percentageTrue: aggregation.percentageTrue,
+      totalFalse: aggregation.totalFalse,
+      totalTrue: aggregation.totalTrue,
+    };
+  }
+
+  private static aggregateDate(
+    aggregation: AggregateReply_Aggregations_Aggregation_DateMessage
+  ): AggregateDate {
+    const parse = (date: string | undefined) => (date !== undefined ? date : undefined);
+    return {
+      count: aggregation.count,
+      maximum: parse(aggregation.maximum),
+      median: parse(aggregation.median),
+      minimum: parse(aggregation.minimum),
+      mode: parse(aggregation.mode),
+    };
+  }
+
+  private static aggregateInt(aggregation: AggregateReply_Aggregations_Aggregation_Integer): AggregateNumber {
+    return {
+      count: aggregation.count,
+      maximum: aggregation.maximum,
+      mean: aggregation.mean,
+      median: aggregation.median,
+      minimum: aggregation.minimum,
+      mode: aggregation.mode,
+      sum: aggregation.sum,
+    };
+  }
+
+  private static aggregateNumber(
+    aggregation: AggregateReply_Aggregations_Aggregation_Number
+  ): AggregateNumber {
+    return {
+      count: aggregation.count,
+      maximum: aggregation.maximum,
+      mean: aggregation.mean,
+      median: aggregation.median,
+      minimum: aggregation.minimum,
+      mode: aggregation.mode,
+      sum: aggregation.sum,
+    };
+  }
+
+  private static aggregateText(aggregation: AggregateReply_Aggregations_Aggregation_Text): AggregateText {
+    return {
+      count: aggregation.count,
+      topOccurrences: aggregation.topOccurences?.items.map((occurrence) => {
+        return {
+          occurs: occurrence.occurs,
+          value: occurrence.value,
+        };
+      }),
+    };
+  }
+
+  private static mapAggregate(aggregation: AggregateReply_Aggregations_Aggregation): AggregateType {
+    if (aggregation.boolean !== undefined) return Deserialize.aggregateBoolean(aggregation.boolean);
+    if (aggregation.date !== undefined) return Deserialize.aggregateDate(aggregation.date);
+    if (aggregation.int !== undefined) return Deserialize.aggregateInt(aggregation.int);
+    if (aggregation.number !== undefined) return Deserialize.aggregateNumber(aggregation.number);
+    // if (aggregation.reference !== undefined) return aggregation.reference;
+    if (aggregation.text !== undefined) return Deserialize.aggregateText(aggregation.text);
+    throw new WeaviateDeserializationError(`Unknown aggregation type: ${aggregation}`);
+  }
+
+  public static aggregate<T, M extends PropertiesMetrics<T>>(reply: AggregateReply): AggregateResult<T, M> {
+    if (reply.singleResult === undefined) {
+      throw new WeaviateDeserializationError('No single result in aggregate response');
+    }
+    return {
+      totalCount: reply.singleResult.objectsCount!,
+      properties: (reply.singleResult.aggregations
+        ? Object.fromEntries(
+            reply.singleResult.aggregations.aggregations.map((aggregation) => [
+              aggregation.property,
+              Deserialize.mapAggregate(aggregation),
+            ])
+          )
+        : {}) as AggregateResult<T, M>['properties'],
+    };
+  }
+
   public query<T>(reply: SearchReply): WeaviateReturn<T> {
     return {
       objects: reply.results.map((result) => {