Fix a regression caused by timestamp refine (#1465)

yhmo · web-flow · commit bea199a9d1bc · 2025-07-15T15:38:47.000+08:00
Signed-off-by: yhmo &lt;yihua.mo@zilliz.com&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,6 @@
 # Changelog
 
-## milvus-sdk-java 2.5.11 (2025-07-14)
+## milvus-sdk-java 2.5.11 (2025-07-15)
 ### Bug
 - Fix a bug of SearchResultsWrapper.getRowRecords() that returns wrong data for output fields
 - Fix a bug of flush that timestamp is not correctly passed
diff --git a/pom.xml b/pom.xml
@@ -68,7 +68,7 @@
     </distributionManagement>
 
     <properties>
-        <revision>2.5.10</revision>
+        <revision>2.5.11</revision>
         <maven.compiler.source>8</maven.compiler.source>
         <maven.compiler.target>8</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
diff --git a/sdk-core/src/main/java/io/milvus/client/AbstractMilvusGrpcClient.java b/sdk-core/src/main/java/io/milvus/client/AbstractMilvusGrpcClient.java
@@ -464,8 +464,6 @@ public R<RpcStatus> createDatabase(CreateDatabaseParam requestParam) {
                     .addAllProperties(propertiesList)
                     .build();
 
-            System.out.println(requestParam.getProperties());
-
             Status response = blockingStub().createDatabase(createDatabaseRequest);
             handleResponse(title, response);
             return R.success(new RpcStatus(RpcStatus.SUCCESS_MSG));
@@ -1870,6 +1868,8 @@ public R<SearchResults> search(@NonNull SearchParam requestParam) {
         String title = String.format("SearchRequest collectionName:%s", requestParam.getCollectionName());
 
         try {
+            // reset the db name so that the timestamp cache can set correct key for this collection
+            requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
             SearchRequest searchRequest = ParamUtils.convertSearchParam(requestParam);
             SearchResults response = this.blockingStub().search(searchRequest);
 
@@ -1897,6 +1897,8 @@ public ListenableFuture<R<SearchResults>> searchAsync(SearchParam requestParam)
         logDebug(requestParam.toString());
         String title = String.format("SearchAsyncRequest collectionName:%s", requestParam.getCollectionName());
 
+        // reset the db name so that the timestamp cache can set correct key for this collection
+        requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
         SearchRequest searchRequest = ParamUtils.convertSearchParam(requestParam);
         ListenableFuture<SearchResults> response = this.futureStub().search(searchRequest);
 
@@ -1942,6 +1944,8 @@ public R<SearchResults> hybridSearch(HybridSearchParam requestParam) {
         String title = String.format("HybridSearchRequest collectionName:%s", requestParam.getCollectionName());
 
         try {
+            // reset the db name so that the timestamp cache can set correct key for this collection
+            requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
             HybridSearchRequest searchRequest = ParamUtils.convertHybridSearchParam(requestParam);
             SearchResults response = this.blockingStub().hybridSearch(searchRequest);
             handleResponse(title, response.getStatus());
@@ -1965,6 +1969,8 @@ public ListenableFuture<R<SearchResults>> hybridSearchAsync(HybridSearchParam re
         logDebug(requestParam.toString());
         String title = String.format("HybridSearchAsyncRequest collectionName:%s", requestParam.getCollectionName());
 
+        // reset the db name so that the timestamp cache can set correct key for this collection
+        requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
         HybridSearchRequest searchRequest = ParamUtils.convertHybridSearchParam(requestParam);
         ListenableFuture<SearchResults> response = this.futureStub().hybridSearch(searchRequest);
 
@@ -2011,6 +2017,8 @@ public R<QueryResults> query(@NonNull QueryParam requestParam) {
                 requestParam.getCollectionName(), requestParam.getExpr());
 
         try {
+            // reset the db name so that the timestamp cache can set correct key for this collection
+            requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
             QueryRequest queryRequest = ParamUtils.convertQueryParam(requestParam);
             QueryResults response = this.blockingStub().query(queryRequest);
 
@@ -2046,6 +2054,8 @@ public ListenableFuture<R<QueryResults>> queryAsync(QueryParam requestParam) {
         String title = String.format("QueryAsyncRequest collectionName:%s, expr:%s",
                 requestParam.getCollectionName(), requestParam.getExpr());
 
+        // reset the db name so that the timestamp cache can set correct key for this collection
+        requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
         QueryRequest queryRequest = ParamUtils.convertQueryParam(requestParam);
         ListenableFuture<QueryResults> response = this.futureStub().query(queryRequest);
 
diff --git a/sdk-core/src/main/java/io/milvus/common/utils/GTsDict.java b/sdk-core/src/main/java/io/milvus/common/utils/GTsDict.java
@@ -52,19 +52,19 @@ public static String CombineCollectionName(String databaseName, String collectio
 
     private ConcurrentMap<String, Long> tsDict = new ConcurrentHashMap<>();
 
-    public void updateCollectionTs(String collectionName, long ts) {
+    public void updateCollectionTs(String name, long ts) {
         // If the collection name exists, use its value to compare to the input ts,
         // only when the input ts is larger than the existing value, replace it with the input ts.
         // If the collection name doesn't exist, directly set the input value.
-        tsDict.compute(collectionName, (key, value) -> (value == null) ? ts : ((ts > value) ? ts : value));
+        tsDict.compute(name, (key, value) -> (value == null) ? ts : ((ts > value) ? ts : value));
     }
 
-    public Long getCollectionTs(String collectionName) {
-        return tsDict.get(collectionName);
+    public Long getCollectionTs(String name) {
+        return tsDict.get(name);
     }
 
-    public void removeCollectionTs(String collectionName) {
-        tsDict.remove(collectionName);
+    public void removeCollectionTs(String name) {
+        tsDict.remove(name);
     }
 
     public void cleanAllCollectionTs() {
diff --git a/sdk-core/src/main/java/io/milvus/param/ParamUtils.java b/sdk-core/src/main/java/io/milvus/param/ParamUtils.java
@@ -845,14 +845,16 @@ public static void compatibleSearchParams(Map<String, Object> searchParams, Sear
 
     @SuppressWarnings("unchecked")
     public static SearchRequest convertSearchParam(@NonNull SearchParam requestParam) throws ParamException {
+        String dbName = requestParam.getDatabaseName();
+        String collectionName = requestParam.getCollectionName();
         SearchRequest.Builder builder = SearchRequest.newBuilder()
-                .setCollectionName(requestParam.getCollectionName());
+                .setCollectionName(collectionName);
 
         if (!requestParam.getPartitionNames().isEmpty()) {
             requestParam.getPartitionNames().forEach(builder::addPartitionNames);
         }
-        if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
-            builder.setDbName(requestParam.getDatabaseName());
+        if (StringUtils.isNotEmpty(dbName)) {
+            builder.setDbName(dbName);
         }
 
         // prepare target vectors
@@ -942,7 +944,7 @@ public static SearchRequest convertSearchParam(@NonNull SearchParam requestParam
             builder.setDsl(requestParam.getExpr());
         }
 
-        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
+        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
         builder.setTravelTimestamp(requestParam.getTravelTimestamp()); // deprecated
         builder.setGuaranteeTimestamp(guaranteeTimestamp);
 
@@ -1006,14 +1008,16 @@ public static SearchRequest convertAnnSearchParam(@NonNull AnnSearchParam annSea
     }
 
     public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearchParam requestParam) throws ParamException {
+        String dbName = requestParam.getDatabaseName();
+        String collectionName = requestParam.getCollectionName();
         HybridSearchRequest.Builder builder = HybridSearchRequest.newBuilder()
-                .setCollectionName(requestParam.getCollectionName());
+                .setCollectionName(collectionName);
 
         if (!requestParam.getPartitionNames().isEmpty()) {
             requestParam.getPartitionNames().forEach(builder::addPartitionNames);
         }
-        if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
-            builder.setDbName(requestParam.getDatabaseName());
+        if (StringUtils.isNotEmpty(dbName)) {
+            builder.setDbName(dbName);
         }
 
         for (AnnSearchParam req : requestParam.getSearchRequests()) {
@@ -1059,7 +1063,7 @@ public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearch
             requestParam.getOutFields().forEach(builder::addOutputFields);
         }
 
-        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
+        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
         builder.setGuaranteeTimestamp(guaranteeTimestamp);
 
         if (requestParam.getConsistencyLevel() == null) {
@@ -1072,18 +1076,20 @@ public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearch
     }
 
     public static QueryRequest convertQueryParam(@NonNull QueryParam requestParam) {
+        String dbName = requestParam.getDatabaseName();
+        String collectionName = requestParam.getCollectionName();
         boolean useDefaultConsistency = (requestParam.getConsistencyLevel() == null);
-        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
+        long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
         QueryRequest.Builder builder = QueryRequest.newBuilder()
-                .setCollectionName(requestParam.getCollectionName())
+                .setCollectionName(collectionName)
                 .addAllPartitionNames(requestParam.getPartitionNames())
                 .addAllOutputFields(requestParam.getOutFields())
                 .setExpr(requestParam.getExpr())
                 .setTravelTimestamp(requestParam.getTravelTimestamp())
                 .setGuaranteeTimestamp(guaranteeTimestamp);
 
-        if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
-            builder.setDbName(requestParam.getDatabaseName());
+        if (StringUtils.isNotEmpty(dbName)) {
+            builder.setDbName(dbName);
         }
 
         // a new parameter from v2.2.9, if user didn't specify consistency level, set this parameter to true
@@ -1120,17 +1126,20 @@ public static QueryRequest convertQueryParam(@NonNull QueryParam requestParam) {
         return builder.build();
     }
 
-    private static long getGuaranteeTimestamp(ConsistencyLevelEnum consistencyLevel, String collectionName){
+    private static long getGuaranteeTimestamp(ConsistencyLevelEnum consistencyLevel, String dbName, String collectionName){
         if(consistencyLevel == null){
-            Long ts = GTsDict.getInstance().getCollectionTs(collectionName);
+            String key = GTsDict.CombineCollectionName(dbName, collectionName);
+            Long ts = GTsDict.getInstance().getCollectionTs(key);
             return  (ts == null) ? 1L : ts;
         }
         switch (consistencyLevel){
             case STRONG:
                 return 0L;
-            case SESSION:
-                Long ts = GTsDict.getInstance().getCollectionTs(collectionName);
+            case SESSION: {
+                String key = GTsDict.CombineCollectionName(dbName, collectionName);
+                Long ts = GTsDict.getInstance().getCollectionTs(key);
                 return (ts == null) ? 1L : ts;
+            }
             case BOUNDED:
                 return 2L; // let server side to determine the bounded time
             default:
diff --git a/sdk-core/src/main/java/io/milvus/param/dml/HybridSearchParam.java b/sdk-core/src/main/java/io/milvus/param/dml/HybridSearchParam.java
@@ -27,6 +27,7 @@
 import io.milvus.param.dml.ranker.BaseRanker;
 import lombok.Getter;
 import lombok.NonNull;
+import lombok.Setter;
 import lombok.ToString;
 
 import java.util.List;
@@ -37,7 +38,8 @@
 @Getter
 @ToString
 public class HybridSearchParam {
-    private final String databaseName;
+    @Setter
+    private String databaseName;
     private final String collectionName;
     private final List<String> partitionNames;
     private final List<AnnSearchParam> searchRequests;
diff --git a/sdk-core/src/main/java/io/milvus/param/dml/QueryParam.java b/sdk-core/src/main/java/io/milvus/param/dml/QueryParam.java
@@ -26,6 +26,7 @@
 import io.milvus.param.ParamUtils;
 import lombok.Getter;
 import lombok.NonNull;
+import lombok.Setter;
 import lombok.ToString;
 
 import java.util.ArrayList;
@@ -37,7 +38,8 @@
 @Getter
 @ToString
 public class QueryParam {
-    private final String databaseName;
+    @Setter
+    private String databaseName;
     private final String collectionName;
     private final List<String> partitionNames;
     private final List<String> outFields;
diff --git a/sdk-core/src/main/java/io/milvus/param/dml/SearchParam.java b/sdk-core/src/main/java/io/milvus/param/dml/SearchParam.java
@@ -28,6 +28,7 @@
 import io.milvus.param.ParamUtils;
 import lombok.Getter;
 import lombok.NonNull;
+import lombok.Setter;
 
 import java.nio.ByteBuffer;
 import java.util.List;
@@ -38,7 +39,8 @@
  */
 @Getter
 public class SearchParam {
-    private final String databaseName;
+    @Setter
+    private String databaseName;
     private final String collectionName;
     private final List<String> partitionNames;
     private final String metricType;
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/collection/CollectionService.java b/sdk-core/src/main/java/io/milvus/v2/service/collection/CollectionService.java
@@ -97,17 +97,18 @@ public Void createCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockin
                         .fieldName(request.getVectorFieldName())
                         .build();
         CreateIndexReq createIndexReq = CreateIndexReq.builder()
+                        .databaseName(request.getDatabaseName())
+                .collectionName(request.getCollectionName())
                         .indexParams(Collections.singletonList(indexParam))
-                        .collectionName(request.getCollectionName())
                         .sync(false)
                         .build();
         indexService.createIndex(blockingStub, createIndexReq);
-        //load collection, set async to true since no need to wait loading progress
+        //load collection, set sync to false since no need to wait loading progress
         try {
-            //TimeUnit.MILLISECONDS.sleep(1000);
             loadCollection(blockingStub, LoadCollectionReq.builder()
-                    .sync(false)
+                    .databaseName(request.getDatabaseName())
                     .collectionName(request.getCollectionName())
+                    .sync(false)
                     .build());
         } catch (Exception e) {
             throw new MilvusClientException(ErrorCode.SERVER_ERROR, "Load collection failed: " + e);
@@ -160,16 +161,18 @@ public Void createCollectionWithSchema(MilvusServiceGrpc.MilvusServiceBlockingSt
         if(request.getIndexParams() != null && !request.getIndexParams().isEmpty()) {
             for(IndexParam indexParam : request.getIndexParams()) {
                 CreateIndexReq createIndexReq = CreateIndexReq.builder()
-                        .indexParams(Collections.singletonList(indexParam))
+                        .databaseName(request.getDatabaseName())
                         .collectionName(request.getCollectionName())
+                        .indexParams(Collections.singletonList(indexParam))
                         .sync(false)
                         .build();
                 indexService.createIndex(blockingStub, createIndexReq);
             }
-            //load collection, set async to true since no need to wait loading progress
+            //load collection, set sync to true since no need to wait loading progress
             loadCollection(blockingStub, LoadCollectionReq.builder()
-                    .sync(false)
+                    .databaseName(request.getDatabaseName())
                     .collectionName(request.getCollectionName())
+                    .sync(false)
                     .build());
         }
 
@@ -311,15 +314,17 @@ public Void renameCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockin
 
     public Void loadCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, LoadCollectionReq request) {
         String title = String.format("LoadCollectionRequest collectionName:%s", request.getCollectionName());
-        LoadCollectionRequest loadCollectionRequest = LoadCollectionRequest.newBuilder()
+        LoadCollectionRequest.Builder builder = LoadCollectionRequest.newBuilder()
                 .setCollectionName(request.getCollectionName())
                 .setReplicaNumber(request.getNumReplicas())
                 .setRefresh(request.getRefresh())
                 .addAllLoadFields(request.getLoadFields())
                 .setSkipLoadDynamicField(request.getSkipLoadDynamicField())
-                .addAllResourceGroups(request.getResourceGroups())
-                .build();
-        Status status = blockingStub.loadCollection(loadCollectionRequest);
+                .addAllResourceGroups(request.getResourceGroups());
+        if (StringUtils.isNotEmpty(request.getDatabaseName())) {
+            builder.setDbName(request.getDatabaseName());
+        }
+        Status status = blockingStub.loadCollection(builder.build());
         rpcUtils.handleResponse(title, status);
         if (request.getSync()) {
             WaitForLoadCollection(blockingStub, request.getCollectionName(), request.getTimeout());
@@ -330,11 +335,13 @@ public Void loadCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockingS
 
     public Void refreshLoad(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, RefreshLoadReq request) {
         String title = String.format("RefreshLoadRequest collectionName:%s", request.getCollectionName());
-        LoadCollectionRequest loadCollectionRequest = LoadCollectionRequest.newBuilder()
+        LoadCollectionRequest.Builder builder = LoadCollectionRequest.newBuilder()
                 .setCollectionName(request.getCollectionName())
-                .setRefresh(true)
-                .build();
-        Status status = blockingStub.loadCollection(loadCollectionRequest);
+                .setRefresh(true);
+        if (StringUtils.isNotEmpty(request.getDatabaseName())) {
+            builder.setDbName(request.getDatabaseName());
+        }
+        Status status = blockingStub.loadCollection(builder.build());
         rpcUtils.handleResponse(title, status);
         if (request.getSync()) {
             WaitForLoadCollection(blockingStub, request.getCollectionName(), request.getTimeout());
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/collection/request/CreateCollectionReq.java b/sdk-core/src/main/java/io/milvus/v2/service/collection/request/CreateCollectionReq.java
@@ -84,6 +84,9 @@ public class CreateCollectionReq {
 
     public static abstract class CreateCollectionReqBuilder<C extends CreateCollectionReq, B extends CreateCollectionReq.CreateCollectionReqBuilder<C, B>> {
         public B indexParam(IndexParam indexParam) {
+            if(null == this.indexParams$value ){
+                this.indexParams$value = new ArrayList<>();
+            }
             try {
                 this.indexParams$value.add(indexParam);
             }catch (UnsupportedOperationException _e){
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/collection/request/LoadCollectionReq.java b/sdk-core/src/main/java/io/milvus/v2/service/collection/request/LoadCollectionReq.java
@@ -29,6 +29,7 @@
 @Data
 @SuperBuilder
 public class LoadCollectionReq {
+    private String databaseName;
     private String collectionName;
     @Builder.Default
     private Integer numReplicas = 1;
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/collection/request/RefreshLoadReq.java b/sdk-core/src/main/java/io/milvus/v2/service/collection/request/RefreshLoadReq.java
@@ -26,6 +26,7 @@
 @Data
 @SuperBuilder
 public class RefreshLoadReq {
+    private String databaseName;
     private String collectionName;
     @Builder.Default
     private Boolean async = Boolean.TRUE;
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/vector/VectorService.java b/sdk-core/src/main/java/io/milvus/v2/service/vector/VectorService.java
diff --git a/sdk-core/src/main/java/io/milvus/v2/service/vector/request/AnnSearchReq.java b/sdk-core/src/main/java/io/milvus/v2/service/vector/request/AnnSearchReq.java
diff --git a/sdk-core/src/main/java/io/milvus/v2/utils/VectorUtils.java b/sdk-core/src/main/java/io/milvus/v2/utils/VectorUtils.java
diff --git a/sdk-core/src/test/java/io/milvus/client/MilvusClientDockerTest.java b/sdk-core/src/test/java/io/milvus/client/MilvusClientDockerTest.java
diff --git a/sdk-core/src/test/java/io/milvus/v2/client/MilvusClientV2DockerTest.java b/sdk-core/src/test/java/io/milvus/v2/client/MilvusClientV2DockerTest.java
