Skip to content

Commit 21878f7

Browse files
yhmoyhmo
authored
Fix a regression caused by timestamp refine (#1472)
Signed-off-by: yhmo <yihua.mo@zilliz.com>
1 parent d8b6234 commit 21878f7

File tree

16 files changed

+370
-69
lines changed

16 files changed

+370
-69
lines changed

‎CHANGELOG.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
# Changelog
22

3-
## milvus-sdk-java 2.6.1 (2025-07-14)
3+
## milvus-sdk-java 2.6.1 (2025-07-15)
44
### Bug
55
- Fix a bug of SearchResultsWrapper.getRowRecords() that returns wrong data for output fields
66

@@ -10,7 +10,7 @@
1010
- Avoid exception when search result is empty
1111
- BulkWriter supports Int8Vector
1212

13-
## milvus-sdk-java 2.5.11 (2025-07-14)
13+
## milvus-sdk-java 2.5.11 (2025-07-15)
1414
### Bug
1515
- Fix a bug of SearchResultsWrapper.getRowRecords() that returns wrong data for output fields
1616
- Fix a bug of flush that timestamp is not correctly passed

‎sdk-core/src/main/java/io/milvus/client/AbstractMilvusGrpcClient.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -464,8 +464,6 @@ public R<RpcStatus> createDatabase(CreateDatabaseParam requestParam) {
464464
.addAllProperties(propertiesList)
465465
.build();
466466

467-
System.out.println(requestParam.getProperties());
468-
469467
Status response = blockingStub().createDatabase(createDatabaseRequest);
470468
handleResponse(title, response);
471469
return R.success(new RpcStatus(RpcStatus.SUCCESS_MSG));
@@ -1870,6 +1868,8 @@ public R<SearchResults> search(@NonNull SearchParam requestParam) {
18701868
String title = String.format("SearchRequest collectionName:%s", requestParam.getCollectionName());
18711869

18721870
try {
1871+
// reset the db name so that the timestamp cache can set correct key for this collection
1872+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
18731873
SearchRequest searchRequest = ParamUtils.convertSearchParam(requestParam);
18741874
SearchResults response = this.blockingStub().search(searchRequest);
18751875

@@ -1897,6 +1897,8 @@ public ListenableFuture<R<SearchResults>> searchAsync(SearchParam requestParam)
18971897
logDebug(requestParam.toString());
18981898
String title = String.format("SearchAsyncRequest collectionName:%s", requestParam.getCollectionName());
18991899

1900+
// reset the db name so that the timestamp cache can set correct key for this collection
1901+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
19001902
SearchRequest searchRequest = ParamUtils.convertSearchParam(requestParam);
19011903
ListenableFuture<SearchResults> response = this.futureStub().search(searchRequest);
19021904

@@ -1942,6 +1944,8 @@ public R<SearchResults> hybridSearch(HybridSearchParam requestParam) {
19421944
String title = String.format("HybridSearchRequest collectionName:%s", requestParam.getCollectionName());
19431945

19441946
try {
1947+
// reset the db name so that the timestamp cache can set correct key for this collection
1948+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
19451949
HybridSearchRequest searchRequest = ParamUtils.convertHybridSearchParam(requestParam);
19461950
SearchResults response = this.blockingStub().hybridSearch(searchRequest);
19471951
handleResponse(title, response.getStatus());
@@ -1965,6 +1969,8 @@ public ListenableFuture<R<SearchResults>> hybridSearchAsync(HybridSearchParam re
19651969
logDebug(requestParam.toString());
19661970
String title = String.format("HybridSearchAsyncRequest collectionName:%s", requestParam.getCollectionName());
19671971

1972+
// reset the db name so that the timestamp cache can set correct key for this collection
1973+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
19681974
HybridSearchRequest searchRequest = ParamUtils.convertHybridSearchParam(requestParam);
19691975
ListenableFuture<SearchResults> response = this.futureStub().hybridSearch(searchRequest);
19701976

@@ -2011,6 +2017,8 @@ public R<QueryResults> query(@NonNull QueryParam requestParam) {
20112017
requestParam.getCollectionName(), requestParam.getExpr());
20122018

20132019
try {
2020+
// reset the db name so that the timestamp cache can set correct key for this collection
2021+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
20142022
QueryRequest queryRequest = ParamUtils.convertQueryParam(requestParam);
20152023
QueryResults response = this.blockingStub().query(queryRequest);
20162024

@@ -2046,6 +2054,8 @@ public ListenableFuture<R<QueryResults>> queryAsync(QueryParam requestParam) {
20462054
String title = String.format("QueryAsyncRequest collectionName:%s, expr:%s",
20472055
requestParam.getCollectionName(), requestParam.getExpr());
20482056

2057+
// reset the db name so that the timestamp cache can set correct key for this collection
2058+
requestParam.setDatabaseName(actualDbName(requestParam.getDatabaseName()));
20492059
QueryRequest queryRequest = ParamUtils.convertQueryParam(requestParam);
20502060
ListenableFuture<QueryResults> response = this.futureStub().query(queryRequest);
20512061

‎sdk-core/src/main/java/io/milvus/common/utils/GTsDict.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -52,19 +52,19 @@ public static String CombineCollectionName(String databaseName, String collectio
5252

5353
private ConcurrentMap<String, Long> tsDict = new ConcurrentHashMap<>();
5454

55-
public void updateCollectionTs(String collectionName, long ts) {
55+
public void updateCollectionTs(String name, long ts) {
5656
// If the collection name exists, use its value to compare to the input ts,
5757
// only when the input ts is larger than the existing value, replace it with the input ts.
5858
// If the collection name doesn't exist, directly set the input value.
59-
tsDict.compute(collectionName, (key, value) -> (value == null) ? ts : ((ts > value) ? ts : value));
59+
tsDict.compute(name, (key, value) -> (value == null) ? ts : ((ts > value) ? ts : value));
6060
}
6161

62-
public Long getCollectionTs(String collectionName) {
63-
return tsDict.get(collectionName);
62+
public Long getCollectionTs(String name) {
63+
return tsDict.get(name);
6464
}
6565

66-
public void removeCollectionTs(String collectionName) {
67-
tsDict.remove(collectionName);
66+
public void removeCollectionTs(String name) {
67+
tsDict.remove(name);
6868
}
6969

7070
public void cleanAllCollectionTs() {

‎sdk-core/src/main/java/io/milvus/param/ParamUtils.java‎

Lines changed: 25 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -849,14 +849,16 @@ public static void compatibleSearchParams(Map<String, Object> searchParams, Sear
849849

850850
@SuppressWarnings("unchecked")
851851
public static SearchRequest convertSearchParam(@NonNull SearchParam requestParam) throws ParamException {
852+
String dbName = requestParam.getDatabaseName();
853+
String collectionName = requestParam.getCollectionName();
852854
SearchRequest.Builder builder = SearchRequest.newBuilder()
853-
.setCollectionName(requestParam.getCollectionName());
855+
.setCollectionName(collectionName);
854856

855857
if (!requestParam.getPartitionNames().isEmpty()) {
856858
requestParam.getPartitionNames().forEach(builder::addPartitionNames);
857859
}
858-
if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
859-
builder.setDbName(requestParam.getDatabaseName());
860+
if (StringUtils.isNotEmpty(dbName)) {
861+
builder.setDbName(dbName);
860862
}
861863

862864
// prepare target vectors
@@ -946,7 +948,7 @@ public static SearchRequest convertSearchParam(@NonNull SearchParam requestParam
946948
builder.setDsl(requestParam.getExpr());
947949
}
948950

949-
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
951+
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
950952
builder.setTravelTimestamp(requestParam.getTravelTimestamp()); // deprecated
951953
builder.setGuaranteeTimestamp(guaranteeTimestamp);
952954

@@ -1010,14 +1012,16 @@ public static SearchRequest convertAnnSearchParam(@NonNull AnnSearchParam annSea
10101012
}
10111013

10121014
public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearchParam requestParam) throws ParamException {
1015+
String dbName = requestParam.getDatabaseName();
1016+
String collectionName = requestParam.getCollectionName();
10131017
HybridSearchRequest.Builder builder = HybridSearchRequest.newBuilder()
1014-
.setCollectionName(requestParam.getCollectionName());
1018+
.setCollectionName(collectionName);
10151019

10161020
if (!requestParam.getPartitionNames().isEmpty()) {
10171021
requestParam.getPartitionNames().forEach(builder::addPartitionNames);
10181022
}
1019-
if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
1020-
builder.setDbName(requestParam.getDatabaseName());
1023+
if (StringUtils.isNotEmpty(dbName)) {
1024+
builder.setDbName(dbName);
10211025
}
10221026

10231027
for (AnnSearchParam req : requestParam.getSearchRequests()) {
@@ -1063,7 +1067,7 @@ public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearch
10631067
requestParam.getOutFields().forEach(builder::addOutputFields);
10641068
}
10651069

1066-
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
1070+
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
10671071
builder.setGuaranteeTimestamp(guaranteeTimestamp);
10681072

10691073
if (requestParam.getConsistencyLevel() == null) {
@@ -1076,18 +1080,20 @@ public static HybridSearchRequest convertHybridSearchParam(@NonNull HybridSearch
10761080
}
10771081

10781082
public static QueryRequest convertQueryParam(@NonNull QueryParam requestParam) {
1083+
String dbName = requestParam.getDatabaseName();
1084+
String collectionName = requestParam.getCollectionName();
10791085
boolean useDefaultConsistency = (requestParam.getConsistencyLevel() == null);
1080-
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
1086+
long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), dbName, collectionName);
10811087
QueryRequest.Builder builder = QueryRequest.newBuilder()
1082-
.setCollectionName(requestParam.getCollectionName())
1088+
.setCollectionName(collectionName)
10831089
.addAllPartitionNames(requestParam.getPartitionNames())
10841090
.addAllOutputFields(requestParam.getOutFields())
10851091
.setExpr(requestParam.getExpr())
10861092
.setTravelTimestamp(requestParam.getTravelTimestamp())
10871093
.setGuaranteeTimestamp(guaranteeTimestamp);
10881094

1089-
if (StringUtils.isNotEmpty(requestParam.getDatabaseName())) {
1090-
builder.setDbName(requestParam.getDatabaseName());
1095+
if (StringUtils.isNotEmpty(dbName)) {
1096+
builder.setDbName(dbName);
10911097
}
10921098

10931099
// a new parameter from v2.2.9, if user didn't specify consistency level, set this parameter to true
@@ -1124,17 +1130,20 @@ public static QueryRequest convertQueryParam(@NonNull QueryParam requestParam) {
11241130
return builder.build();
11251131
}
11261132

1127-
private static long getGuaranteeTimestamp(ConsistencyLevelEnum consistencyLevel, String collectionName){
1133+
private static long getGuaranteeTimestamp(ConsistencyLevelEnum consistencyLevel, String dbName, String collectionName){
11281134
if(consistencyLevel == null){
1129-
Long ts = GTsDict.getInstance().getCollectionTs(collectionName);
1135+
String key = GTsDict.CombineCollectionName(dbName, collectionName);
1136+
Long ts = GTsDict.getInstance().getCollectionTs(key);
11301137
return (ts == null) ? 1L : ts;
11311138
}
11321139
switch (consistencyLevel){
11331140
case STRONG:
11341141
return 0L;
1135-
case SESSION:
1136-
Long ts = GTsDict.getInstance().getCollectionTs(collectionName);
1142+
case SESSION: {
1143+
String key = GTsDict.CombineCollectionName(dbName, collectionName);
1144+
Long ts = GTsDict.getInstance().getCollectionTs(key);
11371145
return (ts == null) ? 1L : ts;
1146+
}
11381147
case BOUNDED:
11391148
return 2L; // let server side to determine the bounded time
11401149
default:

‎sdk-core/src/main/java/io/milvus/param/dml/HybridSearchParam.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import io.milvus.param.dml.ranker.BaseRanker;
2828
import lombok.Getter;
2929
import lombok.NonNull;
30+
import lombok.Setter;
3031
import lombok.ToString;
3132

3233
import java.util.List;
@@ -37,7 +38,8 @@
3738
@Getter
3839
@ToString
3940
public class HybridSearchParam {
40-
private final String databaseName;
41+
@Setter
42+
private String databaseName;
4143
private final String collectionName;
4244
private final List<String> partitionNames;
4345
private final List<AnnSearchParam> searchRequests;

‎sdk-core/src/main/java/io/milvus/param/dml/QueryParam.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import io.milvus.param.ParamUtils;
2727
import lombok.Getter;
2828
import lombok.NonNull;
29+
import lombok.Setter;
2930
import lombok.ToString;
3031

3132
import java.util.ArrayList;
@@ -37,7 +38,8 @@
3738
@Getter
3839
@ToString
3940
public class QueryParam {
40-
private final String databaseName;
41+
@Setter
42+
private String databaseName;
4143
private final String collectionName;
4244
private final List<String> partitionNames;
4345
private final List<String> outFields;

‎sdk-core/src/main/java/io/milvus/param/dml/SearchParam.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import io.milvus.param.ParamUtils;
2929
import lombok.Getter;
3030
import lombok.NonNull;
31+
import lombok.Setter;
3132

3233
import java.nio.ByteBuffer;
3334
import java.util.List;
@@ -38,7 +39,8 @@
3839
*/
3940
@Getter
4041
public class SearchParam {
41-
private final String databaseName;
42+
@Setter
43+
private String databaseName;
4244
private final String collectionName;
4345
private final List<String> partitionNames;
4446
private final String metricType;

‎sdk-core/src/main/java/io/milvus/v2/service/collection/CollectionService.java‎

Lines changed: 22 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -97,17 +97,18 @@ public Void createCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockin
9797
.fieldName(request.getVectorFieldName())
9898
.build();
9999
CreateIndexReq createIndexReq = CreateIndexReq.builder()
100+
.databaseName(request.getDatabaseName())
101+
.collectionName(request.getCollectionName())
100102
.indexParams(Collections.singletonList(indexParam))
101-
.collectionName(request.getCollectionName())
102103
.sync(false)
103104
.build();
104105
indexService.createIndex(blockingStub, createIndexReq);
105-
//load collection, set async to true since no need to wait loading progress
106+
//load collection, set sync to false since no need to wait loading progress
106107
try {
107-
//TimeUnit.MILLISECONDS.sleep(1000);
108108
loadCollection(blockingStub, LoadCollectionReq.builder()
109-
.sync(false)
109+
.databaseName(request.getDatabaseName())
110110
.collectionName(request.getCollectionName())
111+
.sync(false)
111112
.build());
112113
} catch (Exception e) {
113114
throw new MilvusClientException(ErrorCode.SERVER_ERROR, "Load collection failed: " + e);
@@ -160,16 +161,18 @@ public Void createCollectionWithSchema(MilvusServiceGrpc.MilvusServiceBlockingSt
160161
if(request.getIndexParams() != null && !request.getIndexParams().isEmpty()) {
161162
for(IndexParam indexParam : request.getIndexParams()) {
162163
CreateIndexReq createIndexReq = CreateIndexReq.builder()
163-
.indexParams(Collections.singletonList(indexParam))
164+
.databaseName(request.getDatabaseName())
164165
.collectionName(request.getCollectionName())
166+
.indexParams(Collections.singletonList(indexParam))
165167
.sync(false)
166168
.build();
167169
indexService.createIndex(blockingStub, createIndexReq);
168170
}
169-
//load collection, set async to true since no need to wait loading progress
171+
//load collection, set sync to true since no need to wait loading progress
170172
loadCollection(blockingStub, LoadCollectionReq.builder()
171-
.sync(false)
173+
.databaseName(request.getDatabaseName())
172174
.collectionName(request.getCollectionName())
175+
.sync(false)
173176
.build());
174177
}
175178

@@ -329,15 +332,17 @@ public Void renameCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockin
329332

330333
public Void loadCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, LoadCollectionReq request) {
331334
String title = String.format("LoadCollectionRequest collectionName:%s", request.getCollectionName());
332-
LoadCollectionRequest loadCollectionRequest = LoadCollectionRequest.newBuilder()
335+
LoadCollectionRequest.Builder builder = LoadCollectionRequest.newBuilder()
333336
.setCollectionName(request.getCollectionName())
334337
.setReplicaNumber(request.getNumReplicas())
335338
.setRefresh(request.getRefresh())
336339
.addAllLoadFields(request.getLoadFields())
337340
.setSkipLoadDynamicField(request.getSkipLoadDynamicField())
338-
.addAllResourceGroups(request.getResourceGroups())
339-
.build();
340-
Status status = blockingStub.loadCollection(loadCollectionRequest);
341+
.addAllResourceGroups(request.getResourceGroups());
342+
if (StringUtils.isNotEmpty(request.getDatabaseName())) {
343+
builder.setDbName(request.getDatabaseName());
344+
}
345+
Status status = blockingStub.loadCollection(builder.build());
341346
rpcUtils.handleResponse(title, status);
342347
if (request.getSync()) {
343348
WaitForLoadCollection(blockingStub, request.getCollectionName(), request.getTimeout());
@@ -348,11 +353,13 @@ public Void loadCollection(MilvusServiceGrpc.MilvusServiceBlockingStub blockingS
348353

349354
public Void refreshLoad(MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub, RefreshLoadReq request) {
350355
String title = String.format("RefreshLoadRequest collectionName:%s", request.getCollectionName());
351-
LoadCollectionRequest loadCollectionRequest = LoadCollectionRequest.newBuilder()
356+
LoadCollectionRequest.Builder builder = LoadCollectionRequest.newBuilder()
352357
.setCollectionName(request.getCollectionName())
353-
.setRefresh(true)
354-
.build();
355-
Status status = blockingStub.loadCollection(loadCollectionRequest);
358+
.setRefresh(true);
359+
if (StringUtils.isNotEmpty(request.getDatabaseName())) {
360+
builder.setDbName(request.getDatabaseName());
361+
}
362+
Status status = blockingStub.loadCollection(builder.build());
356363
rpcUtils.handleResponse(title, status);
357364
if (request.getSync()) {
358365
WaitForLoadCollection(blockingStub, request.getCollectionName(), request.getTimeout());

‎sdk-core/src/main/java/io/milvus/v2/service/collection/request/CreateCollectionReq.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,9 @@ public class CreateCollectionReq {
8484

8585
public static abstract class CreateCollectionReqBuilder<C extends CreateCollectionReq, B extends CreateCollectionReq.CreateCollectionReqBuilder<C, B>> {
8686
public B indexParam(IndexParam indexParam) {
87+
if(null == this.indexParams$value ){
88+
this.indexParams$value = new ArrayList<>();
89+
}
8790
try {
8891
this.indexParams$value.add(indexParam);
8992
}catch (UnsupportedOperationException _e){

‎sdk-core/src/main/java/io/milvus/v2/service/collection/request/LoadCollectionReq.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
@Data
3030
@SuperBuilder
3131
public class LoadCollectionReq {
32+
private String databaseName;
3233
private String collectionName;
3334
@Builder.Default
3435
private Integer numReplicas = 1;

0 commit comments

Comments
 (0)