Iterator mvcc (#1179)

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

src/main/java/io/milvus/orm/iterator/QueryIterator.java

@@ -32,6 +32,8 @@ import io.milvus.v2.service.collection.request.CreateCollectionReq;
 import io.milvus.v2.service.vector.request.QueryIteratorReq;
 import io.milvus.v2.utils.RpcUtils;
 import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 
@@ -39,6 +41,7 @@ import static io.milvus.param.Constant.NO_CACHE_ID;
 import static io.milvus.param.Constant.UNLIMITED;
 
 public class QueryIterator {
+    protected static final Logger logger = LoggerFactory.getLogger(RpcUtils.class);
     private final IteratorCache iteratorCache;
     private final MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub;
     private final FieldType primaryField;
@@ -52,6 +55,7 @@ public class QueryIterator {
     private int cacheIdInUse;
     private long returnedCount;
     private final RpcUtils rpcUtils;
+    private long sessionTs = 0;
 
     public QueryIterator(QueryIteratorParam queryIteratorParam,
                          MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub,
@@ -67,6 +71,7 @@ public class QueryIterator {
         this.offset = queryIteratorParam.getOffset();
         this.rpcUtils = new RpcUtils();
 
+        setupTsByRequest();
         seek();
     }
 
@@ -86,9 +91,24 @@ public class QueryIterator {
         this.offset = queryIteratorParam.getOffset();
         this.rpcUtils = new RpcUtils();
 
+        setupTsByRequest();
         seek();
     }
 
+    // perform a query to get the first time stamp check point
+    // the time stamp will be input for the next query to skip something
+    private void setupTsByRequest() {
+        QueryResults response = getQueryResultsWrapper(expr, 0L, 1L, 0L);
+        if (response.getSessionTs() <= 0) {
+            logger.warn("Failed to get mvccTs from milvus server, use client-side ts instead");
+            // fall back to latest session ts by local time
+            long ts = System.currentTimeMillis() + 1000L;
+            this.sessionTs = ts << 18;
+        } else {
+            this.sessionTs = response.getSessionTs();
+        }
+    }
+
     private void seek() {
         this.cacheIdInUse = NO_CACHE_ID;
         if (offset == 0) {
@@ -96,7 +116,9 @@ public class QueryIterator {
             return;
         }
 
-        List<QueryResultsWrapper.RowRecord> res = getQueryResultsWrapper(expr, 0L, offset);
+        QueryResults response = getQueryResultsWrapper(expr, 0L, offset, this.sessionTs);
+        QueryResultsWrapper queryWrapper = new QueryResultsWrapper(response);
+        List<QueryResultsWrapper.RowRecord> res = queryWrapper.getRowRecords();
         int resultIndex = Math.min(res.size(), (int) offset);
         updateCursor(res.subList(0, resultIndex));
         offset = 0;
@@ -112,7 +134,10 @@ public class QueryIterator {
         } else {
             iteratorCache.releaseCache(cacheIdInUse);
             String currentExpr = setupNextExpr();
-            List<QueryResultsWrapper.RowRecord> res = getQueryResultsWrapper(currentExpr, offset, batchSize);
+            logger.debug("Query iterator next expression: " + currentExpr);
+            QueryResults response = getQueryResultsWrapper(currentExpr, offset, batchSize, this.sessionTs);
+            QueryResultsWrapper queryWrapper = new QueryResultsWrapper(response);
+            List<QueryResultsWrapper.RowRecord> res = queryWrapper.getRowRecords();
             maybeCache(res);
             ret = res.subList(0, Math.min(batchSize, res.size()));
         }
@@ -174,7 +199,7 @@ public class QueryIterator {
         return ret != null && ret.size() >= batchSize;
     }
 
-    private List<QueryResultsWrapper.RowRecord> getQueryResultsWrapper(String expr, long offset, long limit) {
+    private QueryResults getQueryResultsWrapper(String expr, long offset, long limit, long ts) {
         QueryParam queryParam = QueryParam.newBuilder()
                 .withDatabaseName(queryIteratorParam.getDatabaseName())
                 .withCollectionName(queryIteratorParam.getCollectionName())
@@ -190,12 +215,15 @@ public class QueryIterator {
                 .build();
 
         QueryRequest queryRequest = ParamUtils.convertQueryParam(queryParam);
+        // pass the session ts to query interface
+        if (ts > 0) {
+            queryRequest = queryRequest.toBuilder().setGuaranteeTimestamp(ts).build();
+        }
         QueryResults response = blockingStub.query(queryRequest);
 
         String title = String.format("QueryRequest collectionName:%s", queryIteratorParam.getCollectionName());
         rpcUtils.handleResponse(title, response.getStatus());
 
-        QueryResultsWrapper queryWrapper = new QueryResultsWrapper(response);
-        return queryWrapper.getRowRecords();
+        return response;
     }
 }

src/main/java/io/milvus/orm/iterator/SearchIterator.java

@@ -57,6 +57,7 @@ public class SearchIterator {
     private Float filteredDistance = null;
     private Map<String, Object> params;
     private final RpcUtils rpcUtils;
+    private long sessionTs = 0;
 
     public SearchIterator(SearchIteratorParam searchIteratorParam,
                           MilvusServiceGrpc.MilvusServiceBlockingStub blockingStub,
@@ -171,7 +172,17 @@ public class SearchIterator {
     }
 
     private void initSearchIterator() {
-        SearchResultsWrapper searchResultsWrapper = executeNextSearch(params, expr, false);
+        SearchResults response = executeNextSearch(params, expr, false, 0L);
+        if (response.getSessionTs() <= 0) {
+            logger.warn("Failed to get mvccTs from milvus server, use client-side ts instead");
+            // fall back to latest session ts by local time
+            long ts = System.currentTimeMillis() + 1000L;
+            this.sessionTs = ts << 18;
+        } else {
+            this.sessionTs = response.getSessionTs();
+        }
+
+        SearchResultsWrapper searchResultsWrapper = new SearchResultsWrapper(response.getResults());
         List<QueryResultsWrapper.RowRecord> result = searchResultsWrapper.getRowRecords(0);
         if (CollectionUtils.isNullOrEmpty(result)) {
             String msg = "Cannot init search iterator because init page contains no matched rows, " +
@@ -234,7 +245,7 @@ public class SearchIterator {
         }
     }
 
-    private SearchResultsWrapper executeNextSearch(Map<String, Object> params, String nextExpr, boolean toExtendBatch) {
+    private SearchResults executeNextSearch(Map<String, Object> params, String nextExpr, boolean toExtendBatch, long ts) {
         SearchParam.Builder searchParamBuilder = SearchParam.newBuilder()
                 .withDatabaseName(searchIteratorParam.getDatabaseName())
                 .withCollectionName(searchIteratorParam.getCollectionName())
@@ -257,12 +268,16 @@ public class SearchIterator {
         fillVectorsByPlType(searchParamBuilder);
 
         SearchRequest searchRequest = ParamUtils.convertSearchParam(searchParamBuilder.build());
+        // pass the session ts to search interface
+        if (ts > 0) {
+            searchRequest = searchRequest.toBuilder().setGuaranteeTimestamp(ts).build();
+        }
         SearchResults response = blockingStub.search(searchRequest);
 
         String title = String.format("SearchRequest collectionName:%s", searchIteratorParam.getCollectionName());
         rpcUtils.handleResponse(title, response.getStatus());
 
-        return new SearchResultsWrapper(response.getResults());
+        return response;
     }
 
     private void fillVectorsByPlType(SearchParam.Builder searchParamBuilder) {
@@ -373,8 +388,8 @@ public class SearchIterator {
         while (true) {
             Map<String, Object> nextParams = nextParams(coefficient);
             String nextExpr = filteredDuplicatedResultExpr(expr);
-            SearchResultsWrapper searchResultsWrapper = executeNextSearch(nextParams, nextExpr, true);
-
+            SearchResults response = executeNextSearch(nextParams, nextExpr, true, this.sessionTs);
+            SearchResultsWrapper searchResultsWrapper = new SearchResultsWrapper(response.getResults());
             updateFilteredIds(searchResultsWrapper);
             List<QueryResultsWrapper.RowRecord> newPage = searchResultsWrapper.getRowRecords(0);
 

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

@@ -831,11 +831,11 @@ public class ParamUtils {
                                 .setValue(requestParam.getGroupSize().toString())
                                 .build());
             }
-            if (requestParam.getGroupStrictSize() != null) {
+            if (requestParam.getStrictGroupSize() != null) {
                 builder.addSearchParams(
                         KeyValuePair.newBuilder()
                                 .setKey(Constant.GROUP_STRICT_SIZE)
-                                .setValue(requestParam.getGroupStrictSize().toString())
+                                .setValue(requestParam.getStrictGroupSize().toString())
                                 .build());
             }
         }
@@ -964,11 +964,11 @@ public class ParamUtils {
                                 .setValue(requestParam.getGroupSize().toString())
                                 .build());
             }
-            if (requestParam.getGroupStrictSize() != null) {
+            if (requestParam.getStrictGroupSize() != null) {
                 builder.addRankParams(
                         KeyValuePair.newBuilder()
                                 .setKey(Constant.GROUP_STRICT_SIZE)
-                                .setValue(requestParam.getGroupStrictSize().toString())
+                                .setValue(requestParam.getStrictGroupSize().toString())
                                 .build());
             }
         }
@@ -1001,7 +1001,14 @@ public class ParamUtils {
     }
 
     public static QueryRequest convertQueryParam(@NonNull QueryParam requestParam) {
+        boolean useDefaultConsistency = (requestParam.getConsistencyLevel() == null);
         long guaranteeTimestamp = getGuaranteeTimestamp(requestParam.getConsistencyLevel(), requestParam.getCollectionName());
+        // special logic for iterator
+        // don't pass guaranteeTimestamp for iterator, the query() interface might return empty list.
+        if (requestParam.isIterator()) {
+            useDefaultConsistency = true;
+            guaranteeTimestamp = 0L;
+        }
         QueryRequest.Builder builder = QueryRequest.newBuilder()
                 .setCollectionName(requestParam.getCollectionName())
                 .addAllPartitionNames(requestParam.getPartitionNames())
@@ -1015,7 +1022,7 @@ public class ParamUtils {
         }
 
         // a new parameter from v2.2.9, if user didn't specify consistency level, set this parameter to true
-        if (requestParam.getConsistencyLevel() == null) {
+        if (useDefaultConsistency) {
             builder.setUseDefaultConsistency(true);
         } else {
             builder.setConsistencyLevelValue(requestParam.getConsistencyLevel().getCode());

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

@@ -49,7 +49,7 @@ public class HybridSearchParam {
 
     private final String groupByFieldName;
     private final Integer groupSize;
-    private final Boolean groupStrictSize;
+    private final Boolean strictGroupSize;
 
     private HybridSearchParam(@NonNull Builder builder) {
         this.databaseName = builder.databaseName;
@@ -63,7 +63,7 @@ public class HybridSearchParam {
         this.consistencyLevel = builder.consistencyLevel;
         this.groupByFieldName = builder.groupByFieldName;
         this.groupSize = builder.groupSize;
-        this.groupStrictSize = builder.groupStrictSize;
+        this.strictGroupSize = builder.strictGroupSize;
     }
 
     public static Builder newBuilder() {
@@ -85,7 +85,7 @@ public class HybridSearchParam {
         private ConsistencyLevelEnum consistencyLevel = null;
         private String groupByFieldName = null;
         private Integer groupSize = null;
-        private Boolean groupStrictSize = null;
+        private Boolean strictGroupSize = null;
 
         Builder() {
         }
@@ -241,11 +241,11 @@ public class HybridSearchParam {
          * Whether to force the number of each group to be groupSize.
          * Set to false, milvus might return some groups with number of items less than groupSize.
          *
-         * @param groupStrictSize whether to force the number of each group to be groupSize
+         * @param strictGroupSize whether to force the number of each group to be groupSize
          * @return <code>Builder</code>
          */
-        public Builder withGroupStrictSize(@NonNull Boolean groupStrictSize) {
-            this.groupStrictSize = groupStrictSize;
+        public Builder withStrictGroupSize(@NonNull Boolean strictGroupSize) {
+            this.strictGroupSize = strictGroupSize;
             return this;
         }
 

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

@@ -57,7 +57,7 @@ public class SearchParam {
     private final boolean ignoreGrowing;
     private final String groupByFieldName;
     private final Integer groupSize;
-    private final Boolean groupStrictSize;
+    private final Boolean strictGroupSize;
     private final PlaceholderType plType;
     private final boolean iterator;
 
@@ -81,7 +81,7 @@ public class SearchParam {
         this.ignoreGrowing = builder.ignoreGrowing;
         this.groupByFieldName = builder.groupByFieldName;
         this.groupSize = builder.groupSize;
-        this.groupStrictSize = builder.groupStrictSize;
+        this.strictGroupSize = builder.strictGroupSize;
         this.plType = builder.plType;
         this.iterator = builder.iterator;
     }
@@ -113,7 +113,7 @@ public class SearchParam {
         private Boolean ignoreGrowing = Boolean.FALSE;
         private String groupByFieldName;
         private Integer groupSize = null;
-        private Boolean groupStrictSize = null;
+        private Boolean strictGroupSize = null;
         private Boolean iterator = Boolean.FALSE;
 
         // plType is used to distinct vector type
@@ -398,11 +398,11 @@ public class SearchParam {
          * Whether to force the number of each group to be groupSize.
          * Set to false, milvus might return some groups with number of items less than groupSize.
          *
-         * @param groupStrictSize whether to force the number of each group to be groupSize
+         * @param strictGroupSize whether to force the number of each group to be groupSize
          * @return <code>Builder</code>
          */
-        public Builder withGroupStrictSize(@NonNull Boolean groupStrictSize) {
-            this.groupStrictSize = groupStrictSize;
+        public Builder withStrictGroupSize(@NonNull Boolean strictGroupSize) {
+            this.strictGroupSize = strictGroupSize;
             return this;
         }
 

src/main/java/io/milvus/v2/service/vector/VectorService.java

@@ -161,6 +161,7 @@ public class VectorService extends BaseService {
 
         return QueryResp.builder()
                 .queryResults(convertUtils.getEntities(response))
+                .sessionTs(response.getSessionTs())
                 .build();
 
     }
@@ -178,6 +179,7 @@ public class VectorService extends BaseService {
 
         return SearchResp.builder()
                 .searchResults(convertUtils.getEntities(response))
+                .sessionTs(response.getSessionTs())
                 .build();
     }
 

src/main/java/io/milvus/v2/service/vector/request/HybridSearchReq.java

@@ -45,5 +45,5 @@ public class HybridSearchReq
 
     private String groupByFieldName;
     private Integer groupSize;
-    private Boolean groupStrictSize;
+    private Boolean strictGroupSize;
 }

src/main/java/io/milvus/v2/service/vector/request/SearchReq.java

@@ -60,7 +60,7 @@ public class SearchReq {
     private boolean ignoreGrowing;
     private String groupByFieldName;
     private Integer groupSize;
-    private Boolean groupStrictSize;
+    private Boolean strictGroupSize;
 
     // Expression template, to improve expression parsing performance in complicated list
     // Assume user has a filter = "pk > 3 and city in ["beijing", "shanghai", ......]

src/main/java/io/milvus/v2/service/vector/response/QueryResp.java

@@ -29,6 +29,7 @@ import java.util.Map;
 @SuperBuilder
 public class QueryResp {
     private List<QueryResult> queryResults;
+    private long sessionTs = 1L; // default eventually ts
 
     @Data
     @SuperBuilder

src/main/java/io/milvus/v2/service/vector/response/SearchResp.java

@@ -29,6 +29,7 @@ import java.util.Map;
 @SuperBuilder
 public class SearchResp {
     private List<List<SearchResult>> searchResults;
+    private long sessionTs = 1L; // default eventually ts
 
     @Data
     @SuperBuilder

src/main/java/io/milvus/v2/utils/VectorUtils.java

@@ -192,11 +192,11 @@ public class VectorUtils {
                                 .build());
             }
 
-            if (request.getGroupStrictSize() != null) {
+            if (request.getStrictGroupSize() != null) {
                 builder.addSearchParams(
                         KeyValuePair.newBuilder()
                                 .setKey(Constant.GROUP_STRICT_SIZE)
-                                .setValue(request.getGroupStrictSize().toString())
+                                .setValue(request.getStrictGroupSize().toString())
                                 .build());
             }
         }
@@ -436,11 +436,11 @@ public class VectorUtils {
                                 .build());
             }
 
-            if (request.getGroupStrictSize() != null) {
+            if (request.getStrictGroupSize() != null) {
                 builder.addRankParams(
                         KeyValuePair.newBuilder()
                                 .setKey(Constant.GROUP_STRICT_SIZE)
-                                .setValue(request.getGroupStrictSize().toString())
+                                .setValue(request.getStrictGroupSize().toString())
                                 .build());
             }
         }

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

@@ -2885,136 +2885,136 @@ class MilvusClientDockerTest {
         }
     }
 
-//    @Test
-//    public void testIterator() {
-//        String randomCollectionName = generator.generate(10);
-//
-//        CollectionSchemaParam schema = buildSchema(true, false, true,
-//                Arrays.asList(DataType.FloatVector, DataType.JSON));
-//
-//        // create collection
-//        CreateCollectionParam createParam = CreateCollectionParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .withSchema(schema)
-//                .build();
-//
-//        R<RpcStatus> createR = client.createCollection(createParam);
-//        Assertions.assertEquals(R.Status.Success.getCode(), createR.getStatus().intValue());
-//
-//        // insert data
-//        int rowCount = 1000;
-//        List<JsonObject> rows = new ArrayList<>();
-//        for (long i = 0L; i < rowCount; ++i) {
-//            JsonObject row = new JsonObject();
-//            row.addProperty("id", Long.toString(i));
-//            row.add(DataType.FloatVector.name(), JsonUtils.toJsonTree(generateFloatVectors(1).get(0)));
-//            JsonObject json = new JsonObject();
-//            if (i%2 == 0) {
-//                json.addProperty("even", true);
-//            }
-//            row.add(DataType.JSON.name(), json);
-//            row.addProperty("dynamic", i);
-//            rows.add(row);
-//        }
-//
-//        InsertParam insertParam = InsertParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .withRows(rows)
-//                .build();
-//
-//        R<MutationResult> insertR = client.insert(insertParam);
-//        Assertions.assertEquals(R.Status.Success.getCode(), insertR.getStatus().intValue());
-//
-//        // create index
-//        CreateIndexParam indexParam = CreateIndexParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .withFieldName(DataType.FloatVector.name())
-//                .withIndexType(IndexType.FLAT)
-//                .withMetricType(MetricType.L2)
-//                .build();
-//
-//        R<RpcStatus> createIndexR = client.createIndex(indexParam);
-//        Assertions.assertEquals(R.Status.Success.getCode(), createIndexR.getStatus().intValue());
-//
-//        // load collection
-//        R<RpcStatus> loadR = client.loadCollection(LoadCollectionParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .build());
-//        Assertions.assertEquals(R.Status.Success.getCode(), loadR.getStatus().intValue());
-//
-//        // query iterator
-//        QueryIteratorParam.Builder queryIteratorParamBuilder = QueryIteratorParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .withExpr("dynamic < 300")
-//                .withOutFields(Lists.newArrayList("*"))
-//                .withBatchSize(100L)
-//                .withConsistencyLevel(ConsistencyLevelEnum.BOUNDED);
-//
-//        R<QueryIterator> qResponse = client.queryIterator(queryIteratorParamBuilder.build());
-//        Assertions.assertEquals(R.Status.Success.getCode(), qResponse.getStatus().intValue());
-//
-//        QueryIterator queryIterator = qResponse.getData();
-//        int counter = 0;
-//        while (true) {
-//            List<QueryResultsWrapper.RowRecord> res = queryIterator.next();
-//            if (res.isEmpty()) {
-//                System.out.println("query iteration finished, close");
-//                queryIterator.close();
-//                break;
-//            }
-//
-//            for (QueryResultsWrapper.RowRecord record : res) {
-//                Assertions.assertInstanceOf(Long.class, record.get("dynamic"));
-//                Assertions.assertInstanceOf(String.class, record.get("id"));
-//                Object vec = record.get(DataType.FloatVector.name());
-//                Assertions.assertInstanceOf(List.class, vec);
-//                List<Float> vector = (List<Float>)vec;
-//                Assertions.assertEquals(DIMENSION, vector.size());
-//                Assertions.assertInstanceOf(JsonElement.class, record.get(DataType.JSON.name()));
-////                System.out.println(record);
-//                counter++;
-//            }
-//        }
-//        Assertions.assertEquals(300, counter);
-//
-//        // search iterator
-//        List<List<Float>> vectors = generateFloatVectors(1);
-//        SearchIteratorParam.Builder searchIteratorParamBuilder = SearchIteratorParam.newBuilder()
-//                .withCollectionName(randomCollectionName)
-//                .withOutFields(Lists.newArrayList("*"))
-//                .withBatchSize(10L)
-//                .withVectorFieldName(DataType.FloatVector.name())
-//                .withFloatVectors(vectors)
-//                .withTopK(50)
-//                .withMetricType(MetricType.L2);
-//
-//        R<SearchIterator> sResponse = client.searchIterator(searchIteratorParamBuilder.build());
-//        Assertions.assertEquals(R.Status.Success.getCode(), sResponse.getStatus().intValue());
-//
-//        SearchIterator searchIterator = sResponse.getData();
-//        counter = 0;
-//        while (true) {
-//            List<QueryResultsWrapper.RowRecord> res = searchIterator.next();
-//            if (res.isEmpty()) {
-//                System.out.println("search iteration finished, close");
-//                searchIterator.close();
-//                break;
-//            }
-//
-//            for (QueryResultsWrapper.RowRecord record : res) {
-//                Assertions.assertInstanceOf(Long.class, record.get("dynamic"));
-//                Assertions.assertInstanceOf(String.class, record.get("id"));
-//                Object vec = record.get(DataType.FloatVector.name());
-//                Assertions.assertInstanceOf(List.class, vec);
-//                List<Float> vector = (List<Float>)vec;
-//                Assertions.assertEquals(DIMENSION, vector.size());
-//                Assertions.assertInstanceOf(JsonElement.class, record.get(DataType.JSON.name()));
-////                System.out.println(record);
-//                counter++;
-//            }
-//        }
-//        Assertions.assertEquals(50, counter);
-//    }
+    @Test
+    public void testIterator() {
+        String randomCollectionName = generator.generate(10);
+
+        CollectionSchemaParam schema = buildSchema(true, false, true,
+                Arrays.asList(DataType.FloatVector, DataType.JSON));
+
+        // create collection
+        CreateCollectionParam createParam = CreateCollectionParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .withSchema(schema)
+                .build();
+
+        R<RpcStatus> createR = client.createCollection(createParam);
+        Assertions.assertEquals(R.Status.Success.getCode(), createR.getStatus().intValue());
+
+        // insert data
+        int rowCount = 1000;
+        List<JsonObject> rows = new ArrayList<>();
+        for (long i = 0L; i < rowCount; ++i) {
+            JsonObject row = new JsonObject();
+            row.addProperty("id", Long.toString(i));
+            row.add(DataType.FloatVector.name(), JsonUtils.toJsonTree(generateFloatVectors(1).get(0)));
+            JsonObject json = new JsonObject();
+            if (i%2 == 0) {
+                json.addProperty("even", true);
+            }
+            row.add(DataType.JSON.name(), json);
+            row.addProperty("dynamic", i);
+            rows.add(row);
+        }
+
+        InsertParam insertParam = InsertParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .withRows(rows)
+                .build();
+
+        R<MutationResult> insertR = client.insert(insertParam);
+        Assertions.assertEquals(R.Status.Success.getCode(), insertR.getStatus().intValue());
+
+        // create index
+        CreateIndexParam indexParam = CreateIndexParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .withFieldName(DataType.FloatVector.name())
+                .withIndexType(IndexType.FLAT)
+                .withMetricType(MetricType.L2)
+                .build();
+
+        R<RpcStatus> createIndexR = client.createIndex(indexParam);
+        Assertions.assertEquals(R.Status.Success.getCode(), createIndexR.getStatus().intValue());
+
+        // load collection
+        R<RpcStatus> loadR = client.loadCollection(LoadCollectionParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .build());
+        Assertions.assertEquals(R.Status.Success.getCode(), loadR.getStatus().intValue());
+
+        // query iterator
+        QueryIteratorParam.Builder queryIteratorParamBuilder = QueryIteratorParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .withExpr("dynamic < 300")
+                .withOutFields(Lists.newArrayList("*"))
+                .withBatchSize(100L)
+                .withConsistencyLevel(ConsistencyLevelEnum.BOUNDED);
+
+        R<QueryIterator> qResponse = client.queryIterator(queryIteratorParamBuilder.build());
+        Assertions.assertEquals(R.Status.Success.getCode(), qResponse.getStatus().intValue());
+
+        QueryIterator queryIterator = qResponse.getData();
+        int counter = 0;
+        while (true) {
+            List<QueryResultsWrapper.RowRecord> res = queryIterator.next();
+            if (res.isEmpty()) {
+                System.out.println("query iteration finished, close");
+                queryIterator.close();
+                break;
+            }
+
+            for (QueryResultsWrapper.RowRecord record : res) {
+                Assertions.assertInstanceOf(Long.class, record.get("dynamic"));
+                Assertions.assertInstanceOf(String.class, record.get("id"));
+                Object vec = record.get(DataType.FloatVector.name());
+                Assertions.assertInstanceOf(List.class, vec);
+                List<Float> vector = (List<Float>)vec;
+                Assertions.assertEquals(DIMENSION, vector.size());
+                Assertions.assertInstanceOf(JsonElement.class, record.get(DataType.JSON.name()));
+//                System.out.println(record);
+                counter++;
+            }
+        }
+        Assertions.assertEquals(300, counter);
+
+        // search iterator
+        List<List<Float>> vectors = generateFloatVectors(1);
+        SearchIteratorParam.Builder searchIteratorParamBuilder = SearchIteratorParam.newBuilder()
+                .withCollectionName(randomCollectionName)
+                .withOutFields(Lists.newArrayList("*"))
+                .withBatchSize(10L)
+                .withVectorFieldName(DataType.FloatVector.name())
+                .withFloatVectors(vectors)
+                .withTopK(50)
+                .withMetricType(MetricType.L2);
+
+        R<SearchIterator> sResponse = client.searchIterator(searchIteratorParamBuilder.build());
+        Assertions.assertEquals(R.Status.Success.getCode(), sResponse.getStatus().intValue());
+
+        SearchIterator searchIterator = sResponse.getData();
+        counter = 0;
+        while (true) {
+            List<QueryResultsWrapper.RowRecord> res = searchIterator.next();
+            if (res.isEmpty()) {
+                System.out.println("search iteration finished, close");
+                searchIterator.close();
+                break;
+            }
+
+            for (QueryResultsWrapper.RowRecord record : res) {
+                Assertions.assertInstanceOf(Long.class, record.get("dynamic"));
+                Assertions.assertInstanceOf(String.class, record.get("id"));
+                Object vec = record.get(DataType.FloatVector.name());
+                Assertions.assertInstanceOf(List.class, vec);
+                List<Float> vector = (List<Float>)vec;
+                Assertions.assertEquals(DIMENSION, vector.size());
+                Assertions.assertInstanceOf(JsonElement.class, record.get(DataType.JSON.name()));
+//                System.out.println(record);
+                counter++;
+            }
+        }
+        Assertions.assertEquals(50, counter);
+    }
 
     @Test
     void testDatabase() {

src/test/java/io/milvus/v2/client/MilvusClientV2DockerTest.java

@@ -1387,213 +1387,213 @@ class MilvusClientV2DockerTest {
         Assertions.assertEquals(1L, insertResp.getInsertCnt());
     }
 
-//    @Test
-//    public void testIterator() {
-//        String randomCollectionName = generator.generate(10);
-//        CreateCollectionReq.CollectionSchema collectionSchema = baseSchema();
-//        collectionSchema.addField(AddFieldReq.builder()
-//                .fieldName("float_vector")
-//                .dataType(DataType.FloatVector)
-//                .dimension(dimension)
-//                .build());
-//        collectionSchema.addField(AddFieldReq.builder()
-//                .fieldName("binary_vector")
-//                .dataType(DataType.BinaryVector)
-//                .dimension(dimension)
-//                .build());
-//        collectionSchema.addField(AddFieldReq.builder()
-//                .fieldName("sparse_vector")
-//                .dataType(DataType.SparseFloatVector)
-//                .dimension(dimension)
-//                .build());
-//        collectionSchema.addField(AddFieldReq.builder()
-//                .fieldName("bfloat16_vector")
-//                .dataType(DataType.BFloat16Vector)
-//                .dimension(dimension)
-//                .build());
-//
-//        List<IndexParam> indexParams = new ArrayList<>();
-//        indexParams.add(IndexParam.builder()
-//                .fieldName("float_vector")
-//                .indexType(IndexParam.IndexType.FLAT)
-//                .metricType(IndexParam.MetricType.L2)
-//                .build());
-//        indexParams.add(IndexParam.builder()
-//                .fieldName("binary_vector")
-//                .indexType(IndexParam.IndexType.BIN_FLAT)
-//                .metricType(IndexParam.MetricType.HAMMING)
-//                .build());
-//        indexParams.add(IndexParam.builder()
-//                .fieldName("sparse_vector")
-//                .indexType(IndexParam.IndexType.SPARSE_INVERTED_INDEX)
-//                .metricType(IndexParam.MetricType.IP)
-//                .extraParams(new HashMap<String,Object>(){{put("drop_ratio_build", 0.1);}})
-//                .build());
-//        indexParams.add(IndexParam.builder()
-//                .fieldName("bfloat16_vector")
-//                .indexType(IndexParam.IndexType.FLAT)
-//                .metricType(IndexParam.MetricType.COSINE)
-//                .build());
-//
-//        CreateCollectionReq requestCreate = CreateCollectionReq.builder()
-//                .collectionName(randomCollectionName)
-//                .collectionSchema(collectionSchema)
-//                .indexParams(indexParams)
-//                .build();
-//        client.createCollection(requestCreate);
-//
-//        // insert rows
-//        long count = 10000;
-//        List<JsonObject> data = generateRandomData(collectionSchema, count);
-//        InsertResp insertResp = client.insert(InsertReq.builder()
-//                .collectionName(randomCollectionName)
-//                .data(data)
-//                .build());
-//        Assertions.assertEquals(count, insertResp.getInsertCnt());
-//
-//        // get row count
-//        long rowCount = getRowCount(randomCollectionName);
-//        Assertions.assertEquals(count, rowCount);
-//
-//        // search iterator
-//        SearchIterator searchIterator = client.searchIterator(SearchIteratorReq.builder()
-//                .collectionName(randomCollectionName)
-//                .outputFields(Lists.newArrayList("*"))
-//                .batchSize(20L)
-//                .vectorFieldName("float_vector")
-//                .vectors(Collections.singletonList(new FloatVec(generateFolatVector())))
-//                .expr("int64_field > 500 && int64_field < 1000")
-//                .params("{\"range_filter\": 5.0, \"radius\": 50.0}")
-//                .topK(1000)
-//                .metricType(IndexParam.MetricType.L2)
-//                .consistencyLevel(ConsistencyLevel.EVENTUALLY)
-//                .build());
-//
-//        int counter = 0;
-//        while (true) {
-//            List<QueryResultsWrapper.RowRecord> res = searchIterator.next();
-//            if (res.isEmpty()) {
-//                System.out.println("search iteration finished, close");
-//                searchIterator.close();
-//                break;
-//            }
-//
-//            for (QueryResultsWrapper.RowRecord record : res) {
-//                Assertions.assertInstanceOf(Float.class, record.get("score"));
-//                Assertions.assertTrue((float)record.get("score") >= 5.0);
-//                Assertions.assertTrue((float)record.get("score") <= 50.0);
-//
-//                Assertions.assertInstanceOf(Boolean.class, record.get("bool_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int8_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int16_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int32_field"));
-//                Assertions.assertInstanceOf(Long.class, record.get("int64_field"));
-//                Assertions.assertInstanceOf(Float.class, record.get("float_field"));
-//                Assertions.assertInstanceOf(Double.class, record.get("double_field"));
-//                Assertions.assertInstanceOf(String.class, record.get("varchar_field"));
-//                Assertions.assertInstanceOf(JsonObject.class, record.get("json_field"));
-//                Assertions.assertInstanceOf(List.class, record.get("arr_int_field"));
-//                Assertions.assertInstanceOf(List.class, record.get("float_vector"));
-//                Assertions.assertInstanceOf(ByteBuffer.class, record.get("binary_vector"));
-//                Assertions.assertInstanceOf(ByteBuffer.class, record.get("bfloat16_vector"));
-//                Assertions.assertInstanceOf(SortedMap.class, record.get("sparse_vector"));
-//
-//                long int64Val = (long)record.get("int64_field");
-//                Assertions.assertTrue(int64Val > 500L && int64Val < 1000L);
-//
-//                String varcharVal = (String)record.get("varchar_field");
-//                Assertions.assertTrue(varcharVal.startsWith("varchar_"));
-//
-//                JsonObject jsonObj = (JsonObject)record.get("json_field");
-//                Assertions.assertTrue(jsonObj.has(String.format("JSON_%d", int64Val)));
-//
-//                List<Integer> intArr = (List<Integer>)record.get("arr_int_field");
-//                Assertions.assertTrue(intArr.size() <= 50); // max capacity 50 is defined in the baseSchema()
-//
-//                List<Float> floatVector = (List<Float>)record.get("float_vector");
-//                Assertions.assertEquals(dimension, floatVector.size());
-//
-//                ByteBuffer binaryVector = (ByteBuffer)record.get("binary_vector");
-//                Assertions.assertEquals(dimension, binaryVector.limit()*8);
-//
-//                ByteBuffer bfloat16Vector = (ByteBuffer)record.get("bfloat16_vector");
-//                Assertions.assertEquals(dimension*2, bfloat16Vector.limit());
-//
-//                SortedMap<Long, Float> sparseVector = (SortedMap<Long, Float>)record.get("sparse_vector");
-//                Assertions.assertTrue(sparseVector.size() >= 10 && sparseVector.size() <= 20); // defined in generateSparseVector()
-//
-//                counter++;
-//            }
-//        }
-//        System.out.println(String.format("There are %d items match score between [5.0, 50.0]", counter));
-//
-//        // query iterator
-//        QueryIterator queryIterator = client.queryIterator(QueryIteratorReq.builder()
-//                .collectionName(randomCollectionName)
-//                .expr("int64_field < 300")
-//                .outputFields(Lists.newArrayList("*"))
-//                .batchSize(50L)
-//                .offset(5)
-//                .limit(400)
-//                .consistencyLevel(ConsistencyLevel.EVENTUALLY)
-//                .build());
-//
-//        counter = 0;
-//        while (true) {
-//            List<QueryResultsWrapper.RowRecord> res = queryIterator.next();
-//            if (res.isEmpty()) {
-//                System.out.println("query iteration finished, close");
-//                queryIterator.close();
-//                break;
-//            }
-//
-//            for (QueryResultsWrapper.RowRecord record : res) {
-//                Assertions.assertInstanceOf(Boolean.class, record.get("bool_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int8_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int16_field"));
-//                Assertions.assertInstanceOf(Integer.class, record.get("int32_field"));
-//                Assertions.assertInstanceOf(Long.class, record.get("int64_field"));
-//                Assertions.assertInstanceOf(Float.class, record.get("float_field"));
-//                Assertions.assertInstanceOf(Double.class, record.get("double_field"));
-//                Assertions.assertInstanceOf(String.class, record.get("varchar_field"));
-//                Assertions.assertInstanceOf(JsonObject.class, record.get("json_field"));
-//                Assertions.assertInstanceOf(List.class, record.get("arr_int_field"));
-//                Assertions.assertInstanceOf(List.class, record.get("float_vector"));
-//                Assertions.assertInstanceOf(ByteBuffer.class, record.get("binary_vector"));
-//                Assertions.assertInstanceOf(ByteBuffer.class, record.get("bfloat16_vector"));
-//                Assertions.assertInstanceOf(SortedMap.class, record.get("sparse_vector"));
-//
-//                long int64Val = (long)record.get("int64_field");
-//                Assertions.assertTrue(int64Val < 300L);
-//
-//                String varcharVal = (String)record.get("varchar_field");
-//                Assertions.assertTrue(varcharVal.startsWith("varchar_"));
-//
-//                JsonObject jsonObj = (JsonObject)record.get("json_field");
-//                Assertions.assertTrue(jsonObj.has(String.format("JSON_%d", int64Val)));
-//
-//                List<Integer> intArr = (List<Integer>)record.get("arr_int_field");
-//                Assertions.assertTrue(intArr.size() <= 50); // max capacity 50 is defined in the baseSchema()
-//
-//                List<Float> floatVector = (List<Float>)record.get("float_vector");
-//                Assertions.assertEquals(dimension, floatVector.size());
-//
-//                ByteBuffer binaryVector = (ByteBuffer)record.get("binary_vector");
-//                Assertions.assertEquals(dimension, binaryVector.limit()*8);
-//
-//                ByteBuffer bfloat16Vector = (ByteBuffer)record.get("bfloat16_vector");
-//                Assertions.assertEquals(dimension*2, bfloat16Vector.limit());
-//
-//                SortedMap<Long, Float> sparseVector = (SortedMap<Long, Float>)record.get("sparse_vector");
-//                Assertions.assertTrue(sparseVector.size() >= 10 && sparseVector.size() <= 20); // defined in generateSparseVector()
-//
-//                counter++;
-//            }
-//        }
-//        Assertions.assertEquals(295, counter);
-//
-//        client.dropCollection(DropCollectionReq.builder().collectionName(randomCollectionName).build());
-//    }
+    @Test
+    public void testIterator() {
+        String randomCollectionName = generator.generate(10);
+        CreateCollectionReq.CollectionSchema collectionSchema = baseSchema();
+        collectionSchema.addField(AddFieldReq.builder()
+                .fieldName("float_vector")
+                .dataType(DataType.FloatVector)
+                .dimension(dimension)
+                .build());
+        collectionSchema.addField(AddFieldReq.builder()
+                .fieldName("binary_vector")
+                .dataType(DataType.BinaryVector)
+                .dimension(dimension)
+                .build());
+        collectionSchema.addField(AddFieldReq.builder()
+                .fieldName("sparse_vector")
+                .dataType(DataType.SparseFloatVector)
+                .dimension(dimension)
+                .build());
+        collectionSchema.addField(AddFieldReq.builder()
+                .fieldName("bfloat16_vector")
+                .dataType(DataType.BFloat16Vector)
+                .dimension(dimension)
+                .build());
+
+        List<IndexParam> indexParams = new ArrayList<>();
+        indexParams.add(IndexParam.builder()
+                .fieldName("float_vector")
+                .indexType(IndexParam.IndexType.FLAT)
+                .metricType(IndexParam.MetricType.L2)
+                .build());
+        indexParams.add(IndexParam.builder()
+                .fieldName("binary_vector")
+                .indexType(IndexParam.IndexType.BIN_FLAT)
+                .metricType(IndexParam.MetricType.HAMMING)
+                .build());
+        indexParams.add(IndexParam.builder()
+                .fieldName("sparse_vector")
+                .indexType(IndexParam.IndexType.SPARSE_INVERTED_INDEX)
+                .metricType(IndexParam.MetricType.IP)
+                .extraParams(new HashMap<String,Object>(){{put("drop_ratio_build", 0.1);}})
+                .build());
+        indexParams.add(IndexParam.builder()
+                .fieldName("bfloat16_vector")
+                .indexType(IndexParam.IndexType.FLAT)
+                .metricType(IndexParam.MetricType.COSINE)
+                .build());
+
+        CreateCollectionReq requestCreate = CreateCollectionReq.builder()
+                .collectionName(randomCollectionName)
+                .collectionSchema(collectionSchema)
+                .indexParams(indexParams)
+                .build();
+        client.createCollection(requestCreate);
+
+        // insert rows
+        long count = 10000;
+        List<JsonObject> data = generateRandomData(collectionSchema, count);
+        InsertResp insertResp = client.insert(InsertReq.builder()
+                .collectionName(randomCollectionName)
+                .data(data)
+                .build());
+        Assertions.assertEquals(count, insertResp.getInsertCnt());
+
+        // get row count
+        long rowCount = getRowCount(randomCollectionName);
+        Assertions.assertEquals(count, rowCount);
+
+        // search iterator
+        SearchIterator searchIterator = client.searchIterator(SearchIteratorReq.builder()
+                .collectionName(randomCollectionName)
+                .outputFields(Lists.newArrayList("*"))
+                .batchSize(20L)
+                .vectorFieldName("float_vector")
+                .vectors(Collections.singletonList(new FloatVec(generateFolatVector())))
+                .expr("int64_field > 500 && int64_field < 1000")
+                .params("{\"range_filter\": 5.0, \"radius\": 50.0}")
+                .topK(1000)
+                .metricType(IndexParam.MetricType.L2)
+                .consistencyLevel(ConsistencyLevel.EVENTUALLY)
+                .build());
+
+        int counter = 0;
+        while (true) {
+            List<QueryResultsWrapper.RowRecord> res = searchIterator.next();
+            if (res.isEmpty()) {
+                System.out.println("search iteration finished, close");
+                searchIterator.close();
+                break;
+            }
+
+            for (QueryResultsWrapper.RowRecord record : res) {
+                Assertions.assertInstanceOf(Float.class, record.get("score"));
+                Assertions.assertTrue((float)record.get("score") >= 5.0);
+                Assertions.assertTrue((float)record.get("score") <= 50.0);
+
+                Assertions.assertInstanceOf(Boolean.class, record.get("bool_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int8_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int16_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int32_field"));
+                Assertions.assertInstanceOf(Long.class, record.get("int64_field"));
+                Assertions.assertInstanceOf(Float.class, record.get("float_field"));
+                Assertions.assertInstanceOf(Double.class, record.get("double_field"));
+                Assertions.assertInstanceOf(String.class, record.get("varchar_field"));
+                Assertions.assertInstanceOf(JsonObject.class, record.get("json_field"));
+                Assertions.assertInstanceOf(List.class, record.get("arr_int_field"));
+                Assertions.assertInstanceOf(List.class, record.get("float_vector"));
+                Assertions.assertInstanceOf(ByteBuffer.class, record.get("binary_vector"));
+                Assertions.assertInstanceOf(ByteBuffer.class, record.get("bfloat16_vector"));
+                Assertions.assertInstanceOf(SortedMap.class, record.get("sparse_vector"));
+
+                long int64Val = (long)record.get("int64_field");
+                Assertions.assertTrue(int64Val > 500L && int64Val < 1000L);
+
+                String varcharVal = (String)record.get("varchar_field");
+                Assertions.assertTrue(varcharVal.startsWith("varchar_"));
+
+                JsonObject jsonObj = (JsonObject)record.get("json_field");
+                Assertions.assertTrue(jsonObj.has(String.format("JSON_%d", int64Val)));
+
+                List<Integer> intArr = (List<Integer>)record.get("arr_int_field");
+                Assertions.assertTrue(intArr.size() <= 50); // max capacity 50 is defined in the baseSchema()
+
+                List<Float> floatVector = (List<Float>)record.get("float_vector");
+                Assertions.assertEquals(dimension, floatVector.size());
+
+                ByteBuffer binaryVector = (ByteBuffer)record.get("binary_vector");
+                Assertions.assertEquals(dimension, binaryVector.limit()*8);
+
+                ByteBuffer bfloat16Vector = (ByteBuffer)record.get("bfloat16_vector");
+                Assertions.assertEquals(dimension*2, bfloat16Vector.limit());
+
+                SortedMap<Long, Float> sparseVector = (SortedMap<Long, Float>)record.get("sparse_vector");
+                Assertions.assertTrue(sparseVector.size() >= 10 && sparseVector.size() <= 20); // defined in generateSparseVector()
+
+                counter++;
+            }
+        }
+        System.out.println(String.format("There are %d items match score between [5.0, 50.0]", counter));
+
+        // query iterator
+        QueryIterator queryIterator = client.queryIterator(QueryIteratorReq.builder()
+                .collectionName(randomCollectionName)
+                .expr("int64_field < 300")
+                .outputFields(Lists.newArrayList("*"))
+                .batchSize(50L)
+                .offset(5)
+                .limit(400)
+                .consistencyLevel(ConsistencyLevel.EVENTUALLY)
+                .build());
+
+        counter = 0;
+        while (true) {
+            List<QueryResultsWrapper.RowRecord> res = queryIterator.next();
+            if (res.isEmpty()) {
+                System.out.println("query iteration finished, close");
+                queryIterator.close();
+                break;
+            }
+
+            for (QueryResultsWrapper.RowRecord record : res) {
+                Assertions.assertInstanceOf(Boolean.class, record.get("bool_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int8_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int16_field"));
+                Assertions.assertInstanceOf(Integer.class, record.get("int32_field"));
+                Assertions.assertInstanceOf(Long.class, record.get("int64_field"));
+                Assertions.assertInstanceOf(Float.class, record.get("float_field"));
+                Assertions.assertInstanceOf(Double.class, record.get("double_field"));
+                Assertions.assertInstanceOf(String.class, record.get("varchar_field"));
+                Assertions.assertInstanceOf(JsonObject.class, record.get("json_field"));
+                Assertions.assertInstanceOf(List.class, record.get("arr_int_field"));
+                Assertions.assertInstanceOf(List.class, record.get("float_vector"));
+                Assertions.assertInstanceOf(ByteBuffer.class, record.get("binary_vector"));
+                Assertions.assertInstanceOf(ByteBuffer.class, record.get("bfloat16_vector"));
+                Assertions.assertInstanceOf(SortedMap.class, record.get("sparse_vector"));
+
+                long int64Val = (long)record.get("int64_field");
+                Assertions.assertTrue(int64Val < 300L);
+
+                String varcharVal = (String)record.get("varchar_field");
+                Assertions.assertTrue(varcharVal.startsWith("varchar_"));
+
+                JsonObject jsonObj = (JsonObject)record.get("json_field");
+                Assertions.assertTrue(jsonObj.has(String.format("JSON_%d", int64Val)));
+
+                List<Integer> intArr = (List<Integer>)record.get("arr_int_field");
+                Assertions.assertTrue(intArr.size() <= 50); // max capacity 50 is defined in the baseSchema()
+
+                List<Float> floatVector = (List<Float>)record.get("float_vector");
+                Assertions.assertEquals(dimension, floatVector.size());
+
+                ByteBuffer binaryVector = (ByteBuffer)record.get("binary_vector");
+                Assertions.assertEquals(dimension, binaryVector.limit()*8);
+
+                ByteBuffer bfloat16Vector = (ByteBuffer)record.get("bfloat16_vector");
+                Assertions.assertEquals(dimension*2, bfloat16Vector.limit());
+
+                SortedMap<Long, Float> sparseVector = (SortedMap<Long, Float>)record.get("sparse_vector");
+                Assertions.assertTrue(sparseVector.size() >= 10 && sparseVector.size() <= 20); // defined in generateSparseVector()
+
+                counter++;
+            }
+        }
+        Assertions.assertEquals(295, counter);
+
+        client.dropCollection(DropCollectionReq.builder().collectionName(randomCollectionName).build());
+    }
 
     @Test
     void testDatabase() {