Pass travel timestamp and guarantee timestamp for query/search interface (#249)

Signed-off-by: yhmo <yihua.mo@zilliz.com>

CHANGELOG.md

@@ -1,4 +1,16 @@
-# Changelog   
+# Changelog
+
+## milvus-sdk-java 2.0.1 (2021-01-18)
+
+### Improvement
+
+- \#248 - Pass travel timestamp and guarantee timestamp for query/search interface
+
+## milvus-sdk-java 2.0.0 (2021-12-31)
+
+### Feature
+
+- \#183 - java sdk for milvus 2.0
 
 ## milvus-sdk-java 0.8.5 (2020-08-26)
 

README.md

@@ -15,7 +15,7 @@ The following table shows compatibilities between Milvus and Java SDK.
 
 | Milvus version | Java SDK version |
 | :------------: | :--------------: |
-|     2.0      |    2.0.0         |
+|     2.0      |    2.0.1         |
 
 ### Install Java SDK
 
@@ -27,14 +27,14 @@ You can use **Apache Maven** or **Gradle**/**Grails** to download the SDK.
         <dependency>
             <groupId>io.milvus</groupId>
             <artifactId>milvus-sdk-java</artifactId>
-            <version>2.0.0</version>
+            <version>2.0.1</version>
         </dependency>
        ```
 
    - Gradle/Grails
 
         ```gradle
-        compile 'io.milvus:milvus-sdk-java:2.0.0'
+        compile 'io.milvus:milvus-sdk-java:2.0.1'
         ```
 
 ### Examples

examples/main/io/milvus/GeneralExample.java

@@ -70,7 +70,7 @@ public class GeneralExample {
         }
     }
 
-    private R<RpcStatus> createCollection(long timeoutMiliseconds) {
+    private R<RpcStatus> createCollection(long timeoutMilliseconds) {
         System.out.println("========== createCollection() ==========");
         FieldType fieldType1 = FieldType.newBuilder()
                 .withName(ID_FIELD)
@@ -109,7 +109,7 @@ public class GeneralExample {
                 .addFieldType(fieldType3)
 //                .addFieldType(fieldType4)
                 .build();
-        R<RpcStatus> response = milvusClient.withTimeout(timeoutMiliseconds, TimeUnit.MILLISECONDS)
+        R<RpcStatus> response = milvusClient.withTimeout(timeoutMilliseconds, TimeUnit.MILLISECONDS)
                                             .createCollection(createCollectionReq);
         handleResponseStatus(response);
         System.out.println(response);
@@ -315,6 +315,7 @@ public class GeneralExample {
 
     private R<SearchResults> searchFace(String expr) {
         System.out.println("========== searchFace() ==========");
+        long begin = System.currentTimeMillis();
 
         List<String> outFields = Collections.singletonList(AGE_FIELD);
         List<List<Float>> vectors = generateFloatVectors(5);
@@ -328,9 +329,14 @@ public class GeneralExample {
                 .withVectorFieldName(VECTOR_FIELD)
                 .withExpr(expr)
                 .withParams(SEARCH_PARAM)
+                .withGuaranteeTimestamp(Constant.GUARANTEE_EVENTUALLY_TS)
                 .build();
 
         R<SearchResults> response = milvusClient.search(searchParam);
+        long end = System.currentTimeMillis();
+        long cost = (end - begin);
+        System.out.println("Search time cost: " + cost + "ms");
+
         handleResponseStatus(response);
         SearchResultsWrapper wrapper = new SearchResultsWrapper(response.getData().getResults());
         for (int i = 0; i < vectors.size(); ++i) {
@@ -346,6 +352,7 @@ public class GeneralExample {
 
 //    private R<SearchResults> searchProfile(String expr) {
 //        System.out.println("========== searchProfile() ==========");
+//        long begin = System.currentTimeMillis();
 //
 //        List<String> outFields = Collections.singletonList(AGE_FIELD);
 //        List<ByteBuffer> vectors = generateBinaryVectors(5);
@@ -363,6 +370,10 @@ public class GeneralExample {
 //
 //
 //        R<SearchResults> response = milvusClient.search(searchParam);
+//        long end = System.currentTimeMillis();
+//        long cost = (end - begin);
+//        System.out.println("Search time cost: " + cost + "ms");
+//
 //        handleResponseStatus(response);
 //        SearchResultsWrapper wrapper = new SearchResultsWrapper(response.getData().getResults());
 //        for (int i = 0; i < vectors.size(); ++i) {

examples/pom.xml

@@ -25,7 +25,7 @@
 
     <groupId>io.milvus</groupId>
     <artifactId>milvus-sdk-java-examples</artifactId>
-    <version>2.0.0</version>
+    <version>2.0.1</version>
     <build>
         <plugins>
             <plugin>
@@ -63,7 +63,7 @@
         <dependency>
             <groupId>io.milvus</groupId>
             <artifactId>milvus-sdk-java</artifactId>
-            <version>2.0.0</version>
+            <version>2.0.1</version>
         </dependency>
         <dependency>
             <groupId>com.google.code.gson</groupId>

pom.xml

@@ -25,7 +25,7 @@
 
     <groupId>io.milvus</groupId>
     <artifactId>milvus-sdk-java</artifactId>
-    <version>2.0.0</version>
+    <version>2.0.1</version>
     <packaging>jar</packaging>
 
     <name>io.milvus:milvus-sdk-java</name>

src/main/java/io/milvus/Response/MutationResultWrapper.java

@@ -63,4 +63,16 @@ public class MutationResultWrapper {
     public long getDeleteCount() {
         return result.getDeleteCnt();
     }
+
+    /**
+     * Get timestamp of the operation marked by server. You can use this timestamp as for guarantee timestamp of query/search api.
+     *
+     * Note: the timestamp is not an absolute timestamp, it is a hybrid value combined by UTC time and internal flags.
+     *  We call it TSO, for more information please refer to: https://github.com/milvus-io/milvus/blob/master/docs/design_docs/milvus_hybrid_ts_en.md
+     *
+     * @return <code>int</code> row count of the deleted entities
+     */
+    public long getOperationTs() {
+        return result.getTimestamp();
+    }
 }

src/main/java/io/milvus/client/AbstractMilvusGrpcClient.java

@@ -1462,6 +1462,9 @@ public abstract class AbstractMilvusGrpcClient implements MilvusClient {
                 builder.setDsl(requestParam.getExpr());
             }
 
+            builder.setTravelTimestamp(requestParam.getTravelTimestamp());
+            builder.setGuaranteeTimestamp(requestParam.getGuaranteeTimestamp());
+
             SearchRequest searchRequest = builder.build();
             SearchResults response = this.blockingStub().search(searchRequest);
 
@@ -1499,6 +1502,8 @@ public abstract class AbstractMilvusGrpcClient implements MilvusClient {
                     .addAllPartitionNames(requestParam.getPartitionNames())
                     .addAllOutputFields(requestParam.getOutFields())
                     .setExpr(requestParam.getExpr())
+                    .setTravelTimestamp(requestParam.getTravelTimestamp())
+                    .setGuaranteeTimestamp(requestParam.getGuaranteeTimestamp())
                     .build();
 
             QueryResults response = this.blockingStub().query(queryRequest);

src/main/java/io/milvus/param/Constant.java

@@ -48,4 +48,13 @@ public class Constant {
 
     // max value for waiting create index interval, unit: millisecond
     public static final Long MAX_WAITING_INDEX_INTERVAL = 2000L;
+
+
+    // set this value for "withGuaranteeTimestamp" of QueryParam/SearchParam
+    // to instruct server execute query/search immediately.
+    public static final Long GUARANTEE_EVENTUALLY_TS = 1L;
+
+    // set this value for "withGuaranteeTimestamp" of QueryParam/SearchParam
+    // to instruct server execute query/search after all DML operations finished.
+    public static final Long GUARANTEE_STRONG_TS = 0L;
 }

src/main/java/io/milvus/param/dml/QueryParam.java

@@ -21,6 +21,7 @@ package io.milvus.param.dml;
 
 import com.google.common.collect.Lists;
 import io.milvus.exception.ParamException;
+import io.milvus.param.Constant;
 import io.milvus.param.ParamUtils;
 import lombok.Getter;
 import lombok.NonNull;
@@ -37,12 +38,16 @@ public class QueryParam {
     private final List<String> partitionNames;
     private final List<String> outFields;
     private final String expr;
+    private final long travelTimestamp;
+    private final long guaranteeTimestamp;
 
     private QueryParam(@NonNull Builder builder) {
         this.collectionName = builder.collectionName;
         this.partitionNames = builder.partitionNames;
         this.outFields = builder.outFields;
         this.expr = builder.expr;
+        this.travelTimestamp = builder.travelTimestamp;
+        this.guaranteeTimestamp = builder.guaranteeTimestamp;
     }
 
     public static Builder newBuilder() {
@@ -57,6 +62,8 @@ public class QueryParam {
         private final List<String> partitionNames = Lists.newArrayList();
         private final List<String> outFields = new ArrayList<>();
         private String expr = "";
+        private Long travelTimestamp = 0L;
+        private Long guaranteeTimestamp = Constant.GUARANTEE_EVENTUALLY_TS;
 
         private Builder() {
         }
@@ -132,6 +139,37 @@ public class QueryParam {
             return this;
         }
 
+        /**
+         * Specify an absolute timestamp in a query to get results based on a data view at a specified point in time.
+         * Default value is 0, server executes query on a full data view.
+         *
+         * @param ts a timestamp value
+         * @return <code>Builder</code>
+         */
+        public Builder withTravelTimestamp(@NonNull Long ts) {
+            this.travelTimestamp = ts;
+            return this;
+        }
+
+        /**
+         * Instructs server to see insert/delete operations performed before a provided timestamp.
+         * If no such timestamp is specified, the server will wait for the latest operation to finish and query.
+         *
+         * Note: The timestamp is not an absolute timestamp, it is a hybrid value combined by UTC time and internal flags.
+         *  We call it TSO, for more information please refer to: https://github.com/milvus-io/milvus/blob/master/docs/design_docs/milvus_hybrid_ts_en.md
+         *  You can get a TSO from insert/delete operations, see the <code>MutationResultWrapper</code> class.
+         *  Use an operation's TSO to set this parameter, the server will execute query after this operation is finished.
+         *
+         * Default value is GUARANTEE_EVENTUALLY_TS, query executes query immediately.
+         *
+         * @param ts a timestamp value
+         * @return <code>Builder</code>
+         */
+        public Builder withGuaranteeTimestamp(@NonNull Long ts) {
+            this.guaranteeTimestamp = ts;
+            return this;
+        }
+
         /**
          * Verifies parameters and creates a new <code>QueryParam</code> instance.
          *
@@ -141,6 +179,14 @@ public class QueryParam {
             ParamUtils.CheckNullEmptyString(collectionName, "Collection name");
             ParamUtils.CheckNullEmptyString(expr, "Expression");
 
+            if (travelTimestamp < 0) {
+                throw new ParamException("The travel timestamp must be greater than 0");
+            }
+
+            if (guaranteeTimestamp < 0) {
+                throw new ParamException("The guarantee timestamp must be greater than 0");
+            }
+
             return new QueryParam(this);
         }
     }

src/main/java/io/milvus/param/dml/SearchParam.java

@@ -21,13 +21,13 @@ package io.milvus.param.dml;
 
 import com.google.common.collect.Lists;
 import io.milvus.exception.ParamException;
+import io.milvus.param.Constant;
 import io.milvus.param.MetricType;
 import io.milvus.param.ParamUtils;
 
 import lombok.Getter;
 import lombok.NonNull;
 import java.nio.ByteBuffer;
-import java.util.ArrayList;
 import java.util.List;
 
 /**
@@ -45,6 +45,8 @@ public class SearchParam {
     private final List<?> vectors;
     private final int roundDecimal;
     private final String params;
+    private final long travelTimestamp;
+    private final long guaranteeTimestamp;
 
     private SearchParam(@NonNull Builder builder) {
         this.collectionName = builder.collectionName;
@@ -57,6 +59,8 @@ public class SearchParam {
         this.vectors = builder.vectors;
         this.roundDecimal = builder.roundDecimal;
         this.params = builder.params;
+        this.travelTimestamp = builder.travelTimestamp;
+        this.guaranteeTimestamp = builder.guaranteeTimestamp;
     }
 
     public static Builder newBuilder() {
@@ -73,10 +77,12 @@ public class SearchParam {
         private String vectorFieldName;
         private Integer topK;
         private String expr = "";
-        private List<String> outFields = new ArrayList<>();
+        private final List<String> outFields = Lists.newArrayList();
         private List<?> vectors;
         private Integer roundDecimal = -1;
         private String params = "{}";
+        private Long travelTimestamp = 0L;
+        private Long guaranteeTimestamp = Constant.GUARANTEE_EVENTUALLY_TS;
 
        Builder() {
         }
@@ -168,7 +174,7 @@ public class SearchParam {
          * @return <code>Builder</code>
          */
         public Builder withOutFields(@NonNull List<String> outFields) {
-            this.outFields = outFields;
+            outFields.forEach(this::addOutField);
             return this;
         }
 
@@ -223,6 +229,37 @@ public class SearchParam {
             return this;
         }
 
+        /**
+         * Specify an absolute timestamp in a search to get results based on a data view at a specified point in time.
+         * Default value is 0, server executes search on a full data view.
+         *
+         * @param ts a timestamp value
+         * @return <code>Builder</code>
+         */
+        public Builder withTravelTimestamp(@NonNull Long ts) {
+            this.travelTimestamp = ts;
+            return this;
+        }
+
+        /**
+         * Instructs server to see insert/delete operations performed before a provided timestamp.
+         * If no such timestamp is specified, the server will wait for the latest operation to finish and search.
+         *
+         * Note: The timestamp is not an absolute timestamp, it is a hybrid value combined by UTC time and internal flags.
+         *  We call it TSO, for more information please refer to: https://github.com/milvus-io/milvus/blob/master/docs/design_docs/milvus_hybrid_ts_en.md
+         *  You can get a TSO from insert/delete operations, see the <code>MutationResultWrapper</code> class.
+         *  Use an operation's TSO to set this parameter, the server will execute search after this operation is finished.
+         *
+         * Default value is GUARANTEE_EVENTUALLY_TS, server executes search immediately.
+         *
+         * @param ts a timestamp value
+         * @return <code>Builder</code>
+         */
+        public Builder withGuaranteeTimestamp(@NonNull Long ts) {
+            this.guaranteeTimestamp = ts;
+            return this;
+        }
+
         /**
          * Verifies parameters and creates a new <code>SearchParam</code> instance.
          *
@@ -236,6 +273,14 @@ public class SearchParam {
                 throw new ParamException("TopK value is illegal");
             }
 
+            if (travelTimestamp < 0) {
+                throw new ParamException("The travel timestamp must be greater than 0");
+            }
+
+            if (guaranteeTimestamp < 0) {
+                throw new ParamException("The guarantee timestamp must be greater than 0");
+            }
+
             if (metricType == MetricType.INVALID) {
                 throw new ParamException("Metric type is illegal");
             }

src/main/java/io/milvus/param/index/CreateIndexParam.java

@@ -212,7 +212,7 @@ public class CreateIndexParam {
                 }
             }
 
-            ParamUtils.CheckNullEmptyString(extraParam, "Index extra param");
+//            ParamUtils.CheckNullEmptyString(extraParam, "Index extra param");
 
             return new CreateIndexParam(this);
         }

src/test/java/io/milvus/client/MilvusServiceClientTest.java

@@ -44,7 +44,6 @@ import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.*;
-import static org.junit.jupiter.api.Assertions.assertFalse;
 
 class MilvusServiceClientTest {
     private final int testPort = 53019;
@@ -1092,7 +1091,28 @@ class MilvusServiceClientTest {
                 .withFieldName("field1")
                 .withIndexType(IndexType.IVF_FLAT)
                 .withMetricType(MetricType.L2)
-                .withExtraParam("")
+                .withSyncMode(Boolean.TRUE)
+                .withSyncWaitingInterval(-1L)
+                .build()
+        );
+
+        assertThrows(ParamException.class, () -> CreateIndexParam.newBuilder()
+                .withCollectionName("collection1")
+                .withFieldName("field1")
+                .withIndexType(IndexType.IVF_FLAT)
+                .withMetricType(MetricType.L2)
+                .withSyncMode(Boolean.TRUE)
+                .withSyncWaitingInterval(Constant.MAX_WAITING_INDEX_INTERVAL + 1L)
+                .build()
+        );
+
+        assertThrows(ParamException.class, () -> CreateIndexParam.newBuilder()
+                .withCollectionName("collection1")
+                .withFieldName("field1")
+                .withIndexType(IndexType.IVF_FLAT)
+                .withMetricType(MetricType.L2)
+                .withSyncMode(Boolean.TRUE)
+                .withSyncWaitingTimeout(0L)
                 .build()
         );
     }
@@ -1461,6 +1481,36 @@ class MilvusServiceClientTest {
                 .build()
         );
 
+        assertThrows(ParamException.class, () -> SearchParam.newBuilder()
+                .withCollectionName("collection1")
+                .withPartitionNames(partitions)
+                .addPartitionName("p2")
+                .withParams("dummy")
+                .withOutFields(outputFields)
+                .withVectorFieldName("field1")
+                .withMetricType(MetricType.IP)
+                .withTopK(5)
+                .withVectors(vectors)
+                .withExpr("dummy")
+                .withTravelTimestamp(-1L)
+                .build()
+        );
+
+        assertThrows(ParamException.class, () -> SearchParam.newBuilder()
+                .withCollectionName("collection1")
+                .withPartitionNames(partitions)
+                .addPartitionName("p2")
+                .withParams("dummy")
+                .withOutFields(outputFields)
+                .withVectorFieldName("field1")
+                .withMetricType(MetricType.IP)
+                .withTopK(5)
+                .withVectors(vectors)
+                .withExpr("dummy")
+                .withGuaranteeTimestamp(-1L)
+                .build()
+        );
+
         List<Float> vector1 = Collections.singletonList(0.1F);
         vectors.add(vector1);
         assertThrows(ParamException.class, () -> SearchParam.newBuilder()
@@ -1598,12 +1648,15 @@ class MilvusServiceClientTest {
                 .withPartitionNames(partitions)
                 .withParams("dummy")
                 .withOutFields(outputFields)
+                .addOutField("f2")
                 .withVectorFieldName("field1")
                 .withMetricType(MetricType.IP)
                 .withTopK(5)
                 .withVectors(vectors)
                 .withExpr("dummy")
                 .withRoundDecimal(5)
+                .withTravelTimestamp(1L)
+                .withGuaranteeTimestamp(1L)
                 .build();
         R<SearchResults> resp = client.search(param);
         assertEquals(R.Status.Success.getCode(), resp.getStatus());
@@ -1663,6 +1716,24 @@ class MilvusServiceClientTest {
                 .withExpr("")
                 .build()
         );
+
+        assertThrows(ParamException.class, () -> QueryParam.newBuilder()
+                .withCollectionName("collection1")
+                .withPartitionNames(partitions)
+                .withOutFields(outputFields)
+                .withExpr("dummy")
+                .withTravelTimestamp(-1L)
+                .build()
+        );
+
+        assertThrows(ParamException.class, () -> QueryParam.newBuilder()
+                .withCollectionName("collection1")
+                .withPartitionNames(partitions)
+                .withOutFields(outputFields)
+                .withExpr("dummy")
+                .withGuaranteeTimestamp(-1L)
+                .build()
+        );
     }
 
     @Test
@@ -1673,7 +1744,10 @@ class MilvusServiceClientTest {
                 .withCollectionName("collection1")
                 .withPartitionNames(partitions)
                 .withOutFields(outputFields)
+                .addOutField("d1")
                 .withExpr("dummy")
+                .withTravelTimestamp(0L)
+                .withGuaranteeTimestamp(1L)
                 .build();
 
         testFuncByName("query", param);
@@ -2157,8 +2231,10 @@ class MilvusServiceClientTest {
                         .setIntId(LongArray.newBuilder()
                                 .addAllData(nID)
                                 .build()))
+                .setTimestamp(1000)
                 .build();
         MutationResultWrapper longWrapper = new MutationResultWrapper(results);
+        assertEquals(1000, longWrapper.getOperationTs());
         assertEquals(nID.size(), longWrapper.getInsertCount());
         assertEquals(nID.size(), longWrapper.getDeleteCount());
         assertThrows(ParamException.class, longWrapper::getStringIDs);