BulkWriter supports Struct (#1677)

Fix a bug that options of BulkImport() interface is missed

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

examples/src/main/java/io/milvus/v2/bulkwriter/BulkWriterRemoteExample.java

@@ -167,7 +167,7 @@ public class BulkWriterRemoteExample {
 
     private static void exampleSimpleCollection(List<BulkFileType> fileTypes) throws Exception {
         CreateCollectionReq.CollectionSchema collectionSchema = buildSimpleSchema();
-        createCollection(SIMPLE_COLLECTION_NAME, collectionSchema, false);
+        createCollection(SIMPLE_COLLECTION_NAME, collectionSchema);
 
         for (BulkFileType fileType : fileTypes) {
             remoteWriter(collectionSchema, fileType);
@@ -182,7 +182,7 @@ public class BulkWriterRemoteExample {
         for (BulkFileType fileType : fileTypes) {
             CreateCollectionReq.CollectionSchema collectionSchema = buildAllTypesSchema();
             List<List<String>> batchFiles = allTypesRemoteWriter(collectionSchema, fileType, rows);
-            createCollection(ALL_TYPES_COLLECTION_NAME, collectionSchema, false);
+            createCollection(ALL_TYPES_COLLECTION_NAME, collectionSchema);
             callBulkInsert(batchFiles);
             verifyImportData(collectionSchema, originalData);
         }
@@ -192,7 +192,7 @@ public class BulkWriterRemoteExample {
 //        for (BulkFileType fileType : fileTypes) {
 //            CreateCollectionReq.CollectionSchema collectionSchema = buildAllTypesSchema();
 //            List<List<String>> batchFiles = allTypesRemoteWriter(collectionSchema, fileType, rows);
-//            createCollection(ALL_TYPES_COLLECTION_NAME, collectionSchema, false);
+//            createCollection(ALL_TYPES_COLLECTION_NAME, collectionSchema);
 //            callCloudImport(batchFiles, ALL_TYPES_COLLECTION_NAME, "");
 //            verifyImportData(collectionSchema, originalData);
 //        }
@@ -227,6 +227,20 @@ public class BulkWriterRemoteExample {
         }
     }
 
+    private static Map<String, Object> genOriginStruct(int seed) {
+        Map<String, Object> st = new HashMap<>();
+        st.put("st_bool", seed % 3 == 0);
+        st.put("st_int8", seed % 128);
+        st.put("st_int16", seed % 16384);
+        st.put("st_int32", seed % 65536);
+        st.put("st_int64", seed);
+        st.put("st_float", (float) seed / 4);
+        st.put("st_double", seed / 3);
+        st.put("st_string", String.format("dummy_%d", seed));
+        st.put("st_float_vector", CommonUtils.generateFloatVector(DIM));
+        return st;
+    }
+
     private static List<Map<String, Object>> genOriginalData(int count) {
         List<Map<String, Object>> data = new ArrayList<>();
         for (int i = 0; i < count; ++i) {
@@ -241,11 +255,12 @@ public class BulkWriterRemoteExample {
             row.put("double", (double) i / 7);
             row.put("varchar", "varchar_" + i);
             row.put("json", String.format("{\"dummy\": %s, \"ok\": \"name_%s\"}", i, i));
+            row.put("geometry", String.format("POINT (%d %d)", i, i));
 
             // vector field
             row.put("float_vector", CommonUtils.generateFloatVector(DIM));
             row.put("binary_vector", CommonUtils.generateBinaryVector(DIM).array());
-            row.put("int8_vector", CommonUtils.generateInt8Vector(DIM).array());
+//            row.put("int8_vector", CommonUtils.generateInt8Vector(DIM).array());
             row.put("sparse_vector", CommonUtils.generateSparseVector());
 
             // array field
@@ -258,6 +273,13 @@ public class BulkWriterRemoteExample {
             row.put("array_float", GeneratorUtils.generatorFloatValue(9));
             row.put("array_double", GeneratorUtils.generatorDoubleValue(10));
 
+            // struct field
+            List<Map<String, Object>> structList = new ArrayList<>();
+            for (int k = 0; k < i % 4 + 1; k++) {
+                structList.add(genOriginStruct(i + k));
+            }
+            row.put("struct_field", structList);
+
             data.add(row);
         }
         // a special record with null/default values
@@ -273,11 +295,12 @@ public class BulkWriterRemoteExample {
             row.put("double", null);
             row.put("varchar", null);
             row.put("json", null);
+            row.put("geometry", null);
 
             // vector field
             row.put("float_vector", CommonUtils.generateFloatVector(DIM));
             row.put("binary_vector", CommonUtils.generateBinaryVector(DIM).array());
-            row.put("int8_vector", CommonUtils.generateInt8Vector(DIM).array());
+//            row.put("int8_vector", CommonUtils.generateInt8Vector(DIM).array());
             row.put("sparse_vector", CommonUtils.generateSparseVector());
 
             // array field
@@ -290,6 +313,9 @@ public class BulkWriterRemoteExample {
             row.put("array_float", GeneratorUtils.generatorFloatValue(4));
             row.put("array_double", null);
 
+            // struct field
+            row.put("struct_field", Collections.singletonList(genOriginStruct(0)));
+
             data.add(row);
         }
         return data;
@@ -313,6 +339,7 @@ public class BulkWriterRemoteExample {
                 rowObject.addProperty("double", (Number) row.get("double"));
             }
             rowObject.addProperty("varchar", row.get("varchar") == null ? null : (String) row.get("varchar"));
+            rowObject.addProperty("geometry", row.get("geometry") == null ? null : (String) row.get("geometry"));
 
             // Note: for JSON field, use gson.fromJson() to construct a real JsonObject
             // don't use rowObject.addProperty("json", jsonContent) since the value is treated as a string, not a JsonObject
@@ -322,7 +349,7 @@ public class BulkWriterRemoteExample {
             // vector field
             rowObject.add("float_vector", GSON_INSTANCE.toJsonTree(row.get("float_vector")));
             rowObject.add("binary_vector", GSON_INSTANCE.toJsonTree(row.get("binary_vector")));
-            rowObject.add("int8_vector", GSON_INSTANCE.toJsonTree(row.get("int8_vector")));
+//            rowObject.add("int8_vector", GSON_INSTANCE.toJsonTree(row.get("int8_vector")));
             rowObject.add("sparse_vector", GSON_INSTANCE.toJsonTree(row.get("sparse_vector")));
 
             // array field
@@ -335,6 +362,9 @@ public class BulkWriterRemoteExample {
             rowObject.add("array_float", GSON_INSTANCE.toJsonTree(row.get("array_float")));
             rowObject.add("array_double", GSON_INSTANCE.toJsonTree(row.get("array_double")));
 
+            // struct field
+            rowObject.add("struct_field", GSON_INSTANCE.toJsonTree(row.get("struct_field")));
+
             // dynamic fields
             if (isEnableDynamicField) {
                 rowObject.addProperty("dynamic", "dynamic_" + row.get("id"));
@@ -462,11 +492,10 @@ public class BulkWriterRemoteExample {
             JsonObject getImportProgressObject = convertJsonObject(getImportProgressResult);
             String state = getImportProgressObject.getAsJsonObject("data").get("state").getAsString();
             String progress = getImportProgressObject.getAsJsonObject("data").get("progress").getAsString();
-            if ("Failed".equals(state)) {
+            if ("Failed" .equals(state)) {
                 String reason = getImportProgressObject.getAsJsonObject("data").get("reason").getAsString();
-                System.out.printf("The job %s failed, reason: %s%n", jobId, reason);
-                break;
-            } else if ("Completed".equals(state)) {
+                throw new RuntimeException(String.format("The job %s failed, reason: %s", jobId, reason));
+            } else if ("Completed" .equals(state)) {
                 System.out.printf("The job %s completed%n", jobId);
                 break;
             } else {
@@ -475,11 +504,57 @@ public class BulkWriterRemoteExample {
         }
     }
 
+//    private static void callCloudImport(List<List<String>> batchFiles, String collectionName, String partitionName) throws InterruptedException {
+//        String objectUrl = StorageConsts.cloudStorage == CloudStorage.AZURE
+//                ? StorageConsts.cloudStorage.getAzureObjectUrl(StorageConsts.AZURE_ACCOUNT_NAME, StorageConsts.AZURE_CONTAINER_NAME, ImportUtils.getCommonPrefix(batchFiles))
+//                : StorageConsts.cloudStorage.getS3ObjectUrl(StorageConsts.STORAGE_BUCKET, ImportUtils.getCommonPrefix(batchFiles), StorageConsts.STORAGE_REGION);
+//        String accessKey = StorageConsts.cloudStorage == CloudStorage.AZURE ? StorageConsts.AZURE_ACCOUNT_NAME : StorageConsts.STORAGE_ACCESS_KEY;
+//        String secretKey = StorageConsts.cloudStorage == CloudStorage.AZURE ? StorageConsts.AZURE_ACCOUNT_KEY : StorageConsts.STORAGE_SECRET_KEY;
+//
+//        System.out.println("\n===================== call cloudImport ====================");
+//        CloudImportRequest bulkImportRequest = CloudImportRequest.builder()
+//                .objectUrl(objectUrl).accessKey(accessKey).secretKey(secretKey)
+//                .clusterId(CloudImportConsts.CLUSTER_ID).collectionName(collectionName).partitionName(partitionName)
+//                .apiKey(CloudImportConsts.API_KEY)
+//                .build();
+//        String bulkImportResult = BulkImportUtils.bulkImport(CloudImportConsts.CLOUD_ENDPOINT, bulkImportRequest);
+//        JsonObject bulkImportObject = convertJsonObject(bulkImportResult);
+//
+//        String jobId = bulkImportObject.getAsJsonObject("data").get("jobId").getAsString();
+//        System.out.println("Create a cloudImport job, job id: " + jobId);
+//
+//        System.out.println("\n===================== call cloudListImportJobs ====================");
+//        CloudListImportJobsRequest listImportJobsRequest = CloudListImportJobsRequest.builder().clusterId(CloudImportConsts.CLUSTER_ID).currentPage(1).pageSize(10).apiKey(CloudImportConsts.API_KEY).build();
+//        String listImportJobsResult = BulkImportUtils.listImportJobs(CloudImportConsts.CLOUD_ENDPOINT, listImportJobsRequest);
+//        System.out.println(listImportJobsResult);
+//        while (true) {
+//            System.out.println("Wait 5 second to check bulkInsert job state...");
+//            TimeUnit.SECONDS.sleep(5);
+//
+//            System.out.println("\n===================== call cloudGetProgress ====================");
+//            CloudDescribeImportRequest request = CloudDescribeImportRequest.builder().clusterId(CloudImportConsts.CLUSTER_ID).jobId(jobId).apiKey(CloudImportConsts.API_KEY).build();
+//            String getImportProgressResult = BulkImportUtils.getImportProgress(CloudImportConsts.CLOUD_ENDPOINT, request);
+//            JsonObject getImportProgressObject = convertJsonObject(getImportProgressResult);
+//            String importProgressState = getImportProgressObject.getAsJsonObject("data").get("state").getAsString();
+//            String progress = getImportProgressObject.getAsJsonObject("data").get("progress").getAsString();
+//
+//            if ("Failed" .equals(importProgressState)) {
+//                String reason = getImportProgressObject.getAsJsonObject("data").get("reason").getAsString();
+//                System.out.printf("The job %s failed, reason: %s%n", jobId, reason);
+//                break;
+//            } else if ("Completed" .equals(importProgressState)) {
+//                System.out.printf("The job %s completed%n", jobId);
+//                break;
+//            } else {
+//                System.out.printf("The job %s is running, state:%s progress:%s%n", jobId, importProgressState, progress);
+//            }
+//        }
+//    }
+
     /**
      * @param collectionSchema collection info
-     * @param dropIfExist      if collection already exist, will drop firstly and then create again
      */
-    private static void createCollection(String collectionName, CreateCollectionReq.CollectionSchema collectionSchema, boolean dropIfExist) {
+    private static void createCollection(String collectionName, CreateCollectionReq.CollectionSchema collectionSchema) {
         System.out.println("\n===================== create collection ====================");
         checkMilvusClientIfExist();
 
@@ -489,15 +564,8 @@ public class BulkWriterRemoteExample {
                 .consistencyLevel(ConsistencyLevel.BOUNDED)
                 .build();
 
-        Boolean has = milvusClient.hasCollection(HasCollectionReq.builder().collectionName(collectionName).build());
-        if (has) {
-            if (dropIfExist) {
-                milvusClient.dropCollection(DropCollectionReq.builder().collectionName(collectionName).build());
-                milvusClient.createCollection(requestCreate);
-            }
-        } else {
-            milvusClient.createCollection(requestCreate);
-        }
+        milvusClient.dropCollection(DropCollectionReq.builder().collectionName(collectionName).build());
+        milvusClient.createCollection(requestCreate);
 
         System.out.printf("Collection %s created%n", collectionName);
     }
@@ -558,13 +626,40 @@ public class BulkWriterRemoteExample {
         }
     }
 
-    private static void verifyImportData(CreateCollectionReq.CollectionSchema collectionSchema, List<Map<String, Object>> rows) {
-        createIndex();
+    private static void compareStruct(CreateCollectionReq.CollectionSchema collectionSchema,
+                                      Map<String, Object> expectedData, Map<String, Object> fetchedData,
+                                      String fieldName) {
+        CreateCollectionReq.StructFieldSchema field = collectionSchema.getStructField(fieldName);
+        Object expectedValue = expectedData.get(fieldName);
+        Object fetchedValue = fetchedData.get(fieldName);
+        if (fetchedValue == null) {
+            throw new RuntimeException(String.format("Struct field '%s' missed in fetched data", fieldName));
+        }
+
+        List<Map<String, Object>> expectedList = (List<Map<String, Object>>) expectedValue;
+        if (!(fetchedValue instanceof List<?>)) {
+            throw new RuntimeException(String.format("Struct field '%s' value should be a list", fieldName));
+        }
+
+        List<Map<String, Object>> fetchedList = (List<Map<String, Object>>) fetchedValue;
+        if (expectedList.size() != fetchedList.size()) {
+            throw new RuntimeException(String.format("Struct field '%s' list count unmatched", fieldName));
+        }
+
+        for (int i = 0; i < expectedList.size(); i++) {
+            Map<String, Object> expectedStruct = expectedList.get(i);
+            Map<String, Object> fetchedStruct = fetchedList.get(i);
+            if (expectedStruct.equals(fetchedStruct)) {
+                throw new RuntimeException(String.format("Struct field '%s' value unmatched", fieldName));
+            }
+        }
+    }
 
+    private static void verifyImportData(CreateCollectionReq.CollectionSchema collectionSchema, List<Map<String, Object>> rows) {
         List<Long> QUERY_IDS = Lists.newArrayList(1L, (long) rows.get(rows.size() - 1).get("id"));
         System.out.printf("Load collection and query items %s%n", QUERY_IDS);
+        createIndex(collectionSchema);
         loadCollection();
-
         String expr = String.format("id in %s", QUERY_IDS);
         System.out.println(expr);
 
@@ -597,40 +692,74 @@ public class BulkWriterRemoteExample {
 
             comparePrint(collectionSchema, originalEntity, fetchedEntity, "float_vector");
             comparePrint(collectionSchema, originalEntity, fetchedEntity, "binary_vector");
-            comparePrint(collectionSchema, originalEntity, fetchedEntity, "int8_vector");
+//            comparePrint(collectionSchema, originalEntity, fetchedEntity, "int8_vector");
             comparePrint(collectionSchema, originalEntity, fetchedEntity, "sparse_vector");
 
+            compareStruct(collectionSchema, originalEntity, fetchedEntity, "struct_field");
+
             System.out.println(fetchedEntity);
         }
         System.out.println("Result is correct!");
     }
 
-    private static void createIndex() {
+    private static void createIndex(CreateCollectionReq.CollectionSchema collectionSchema) {
         System.out.println("Create index...");
         checkMilvusClientIfExist();
 
         List<IndexParam> indexes = new ArrayList<>();
-        indexes.add(IndexParam.builder()
-                .fieldName("float_vector")
-                .indexType(IndexParam.IndexType.FLAT)
-                .metricType(IndexParam.MetricType.L2)
-                .build());
-        indexes.add(IndexParam.builder()
-                .fieldName("binary_vector")
-                .indexType(IndexParam.IndexType.BIN_FLAT)
-                .metricType(IndexParam.MetricType.HAMMING)
-                .build());
-        indexes.add(IndexParam.builder()
-                .fieldName("int8_vector")
-                .indexType(IndexParam.IndexType.AUTOINDEX)
-                .metricType(IndexParam.MetricType.L2)
-                .build());
-        indexes.add(IndexParam.builder()
-                .fieldName("sparse_vector")
-                .indexType(IndexParam.IndexType.SPARSE_WAND)
-                .metricType(IndexParam.MetricType.IP)
-                .build());
+        for (CreateCollectionReq.FieldSchema field : collectionSchema.getFieldSchemaList()) {
+            IndexParam.IndexType indexType;
+            IndexParam.MetricType metricType;
+            switch (field.getDataType()) {
+                case FloatVector:
+                case Float16Vector:
+                case BFloat16Vector:
+                    indexType = IndexParam.IndexType.IVF_FLAT;
+                    metricType = IndexParam.MetricType.L2;
+                    break;
+                case BinaryVector:
+                    indexType = IndexParam.IndexType.BIN_FLAT;
+                    metricType = IndexParam.MetricType.HAMMING;
+                    break;
+                case Int8Vector:
+                    indexType = IndexParam.IndexType.AUTOINDEX;
+                    metricType = IndexParam.MetricType.L2;
+                    break;
+                case SparseFloatVector:
+                    indexType = IndexParam.IndexType.SPARSE_WAND;
+                    metricType = IndexParam.MetricType.IP;
+                    break;
+                default:
+                    continue;
+            }
+            indexes.add(IndexParam.builder()
+                    .fieldName(field.getName())
+                    .indexName(String.format("index_%s", field.getName()))
+                    .indexType(indexType)
+                    .metricType(metricType)
+                    .build());
+        }
 
+        for (CreateCollectionReq.StructFieldSchema struct : collectionSchema.getStructFields()) {
+            for (CreateCollectionReq.FieldSchema subField : struct.getFields()) {
+                IndexParam.IndexType indexType;
+                IndexParam.MetricType metricType;
+                switch (subField.getDataType()) {
+                    case FloatVector:
+                        indexType = IndexParam.IndexType.HNSW;
+                        metricType = IndexParam.MetricType.MAX_SIM_COSINE;
+                        break;
+                    default:
+                        continue;
+                }
+                indexes.add(IndexParam.builder()
+                        .fieldName(String.format("%s[%s]", struct.getName(), subField.getName()))
+                        .indexName(String.format("index_%s", subField.getName()))
+                        .indexType(indexType)
+                        .metricType(metricType)
+                        .build());
+            }
+        }
         milvusClient.createIndex(CreateIndexReq.builder()
                 .collectionName(ALL_TYPES_COLLECTION_NAME)
                 .indexParams(indexes)
@@ -660,6 +789,7 @@ public class BulkWriterRemoteExample {
                 .collectionName(ALL_TYPES_COLLECTION_NAME)
                 .filter(expr)
                 .outputFields(outputFields)
+                .consistencyLevel(ConsistencyLevel.STRONG)
                 .build();
         QueryResp response = milvusClient.query(test);
         return response.getQueryResults();
@@ -786,6 +916,11 @@ public class BulkWriterRemoteExample {
                 .dataType(DataType.JSON)
                 .isNullable(true)
                 .build());
+        schemaV2.addField(AddFieldReq.builder()
+                .fieldName("geometry")
+                .dataType(DataType.Geometry)
+                .isNullable(true)
+                .build());
 
         // vector fields
         schemaV2.addField(AddFieldReq.builder()
@@ -798,11 +933,11 @@ public class BulkWriterRemoteExample {
                 .dataType(DataType.BinaryVector)
                 .dimension(DIM)
                 .build());
-        schemaV2.addField(AddFieldReq.builder()
-                .fieldName("int8_vector")
-                .dataType(DataType.Int8Vector)
-                .dimension(DIM)
-                .build());
+//        schemaV2.addField(AddFieldReq.builder()
+//                .fieldName("int8_vector")
+//                .dataType(DataType.Int8Vector)
+//                .dimension(DIM)
+//                .build());
         schemaV2.addField(AddFieldReq.builder()
                 .fieldName("sparse_vector")
                 .dataType(DataType.SparseFloatVector)
@@ -860,6 +995,50 @@ public class BulkWriterRemoteExample {
                 .elementType(DataType.Double)
                 .isNullable(true)
                 .build());
+        schemaV2.addField(AddFieldReq.builder()
+                .fieldName("struct_field")
+                .dataType(DataType.Array)
+                .elementType(DataType.Struct)
+                .maxCapacity(100)
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_bool")
+                        .dataType(DataType.Bool)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int8")
+                        .dataType(DataType.Int8)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int16")
+                        .dataType(DataType.Int16)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int32")
+                        .dataType(DataType.Int32)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int64")
+                        .dataType(DataType.Int64)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_float")
+                        .dataType(DataType.Float)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_double")
+                        .dataType(DataType.Double)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_string")
+                        .dataType(DataType.VarChar)
+                        .maxLength(100)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_float_vector")
+                        .dataType(DataType.FloatVector)
+                        .dimension(DIM)
+                        .build())
+                .build());
 
         return schemaV2;
     }

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/BulkWriter.java

@@ -20,10 +20,7 @@
 package io.milvus.bulkwriter;
 
 import com.google.common.collect.Lists;
-import com.google.gson.JsonElement;
-import com.google.gson.JsonNull;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
+import com.google.gson.*;
 import io.milvus.bulkwriter.common.clientenum.BulkFileType;
 import io.milvus.bulkwriter.common.clientenum.TypeSize;
 import io.milvus.bulkwriter.common.utils.V2AdapterUtils;
@@ -281,54 +278,54 @@ public abstract class BulkWriter implements AutoCloseable {
                 }
             }
 
-            DataType dataType = field.getDataType();
-            switch (dataType) {
-                case BinaryVector:
-                case FloatVector:
-                case Float16Vector:
-                case BFloat16Vector:
-                case SparseFloatVector:
-                case Int8Vector: {
-                    Pair<Object, Integer> objectAndSize = verifyVector(obj, field);
-                    rowValues.put(fieldName, objectAndSize.getLeft());
-                    rowSize += objectAndSize.getRight();
-                    break;
-                }
-                case VarChar:
-                case Geometry:
-                case Timestamptz: {
-                    Pair<Object, Integer> objectAndSize = verifyVarchar(obj, field);
-                    rowValues.put(fieldName, objectAndSize.getLeft());
-                    rowSize += objectAndSize.getRight();
-                    break;
-                }
-                case JSON: {
-                    Pair<Object, Integer> objectAndSize = verifyJSON(obj, field);
-                    rowValues.put(fieldName, objectAndSize.getLeft());
-                    rowSize += objectAndSize.getRight();
-                    break;
+            Pair<Object, Integer> objectAndSize = verifyByDatatype(field, obj);
+            if (objectAndSize != null) {
+                rowValues.put(fieldName, objectAndSize.getLeft());
+                rowSize += objectAndSize.getRight();
+            }
+        }
+
+        for (CreateCollectionReq.StructFieldSchema struct : collectionSchema.getStructFields()) {
+            String structName = struct.getName();
+            JsonArray structList = row.getAsJsonArray(structName);
+            if (structList == null) {
+                String msg = String.format("Value of struct field '%s' is not provided.", structName);
+                ExceptionUtils.throwUnExpectedException(msg);
+            }
+
+            List<Map<String, Object>> validList = new ArrayList<>();
+            for (JsonElement st : structList.asList()) {
+                if (!st.isJsonObject()) {
+                    String msg = String.format("Element of struct field '%s' must be JSON dict.", structName);
+                    ExceptionUtils.throwUnExpectedException(msg);
                 }
-                case Array: {
-                    Pair<Object, Integer> objectAndSize = verifyArray(obj, field);
-                    rowValues.put(fieldName, objectAndSize.getLeft());
-                    rowSize += objectAndSize.getRight();
-                    break;
+
+                JsonObject dict = st.getAsJsonObject();
+                Map<String, Object> validStruct = new HashMap<>();
+                for (CreateCollectionReq.FieldSchema subField : struct.getFields()) {
+                    String subFieldName = subField.getName();
+                    String combineName = String.format("%s[%s]", structName, subFieldName);
+                    boolean provided = dict.has(subFieldName);
+                    boolean isFunctionOutput = outputFieldNames.contains(combineName);
+                    if (provided && isFunctionOutput) {
+                        String msg = String.format("The field '%s'  is function output, no need to provide", combineName);
+                        ExceptionUtils.throwUnExpectedException(msg);
+                    }
+                    if (!isFunctionOutput && !provided) {
+                        String msg = String.format("Value of field '%s' is not provided.", combineName);
+                        ExceptionUtils.throwUnExpectedException(msg);
+                    }
+
+                    Pair<Object, Integer> objectAndSize = verifyByDatatype(subField, dict.get(subFieldName));
+                    if (objectAndSize != null) {
+                        validStruct.put(subFieldName, objectAndSize.getLeft());
+                        rowSize += objectAndSize.getRight();
+                    }
                 }
-                case Bool:
-                case Int8:
-                case Int16:
-                case Int32:
-                case Int64:
-                case Float:
-                case Double:
-                    Pair<Object, Integer> objectAndSize = verifyScalar(obj, field);
-                    rowValues.put(fieldName, objectAndSize.getLeft());
-                    rowSize += objectAndSize.getRight();
-                    break;
-                default:
-                    String msg = String.format("Unsupported data type of field '%s', not implemented in BulkWriter.", fieldName);
-                    ExceptionUtils.throwUnExpectedException(msg);
+                validList.add(validStruct);
             }
+
+            rowValues.put(structName, validList);
         }
 
         // process dynamic values
@@ -361,6 +358,44 @@ public abstract class BulkWriter implements AutoCloseable {
         return rowValues;
     }
 
+    private Pair<Object, Integer> verifyByDatatype(CreateCollectionReq.FieldSchema field, JsonElement obj) {
+        DataType dataType = field.getDataType();
+        String fieldName = field.getName();
+        switch (dataType) {
+            case BinaryVector:
+            case FloatVector:
+            case Float16Vector:
+            case BFloat16Vector:
+            case SparseFloatVector:
+            case Int8Vector: {
+                return verifyVector(obj, field);
+            }
+            case VarChar:
+            case Geometry:
+            case Timestamptz: {
+                return verifyVarchar(obj, field);
+            }
+            case JSON: {
+                return verifyJSON(obj, field);
+            }
+            case Array: {
+                return verifyArray(obj, field);
+            }
+            case Bool:
+            case Int8:
+            case Int16:
+            case Int32:
+            case Int64:
+            case Float:
+            case Double:
+                return verifyScalar(obj, field);
+            default:
+                String msg = String.format("Unsupported data type of field '%s', not implemented in BulkWriter.", fieldName);
+                ExceptionUtils.throwUnExpectedException(msg);
+        }
+        return null;
+    }
+
     private Pair<Object, Integer> verifyVector(JsonElement object, CreateCollectionReq.FieldSchema field) {
         Object vector = DataUtils.checkFieldValue(field, object);
         io.milvus.v2.common.DataType dataType = field.getDataType();

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/common/utils/ParquetUtils.java

@@ -31,7 +31,7 @@ import java.util.List;
 import static io.milvus.param.Constant.DYNAMIC_FIELD_NAME;
 
 public class ParquetUtils {
-    private static void setMessageType(Types.MessageTypeBuilder builder,
+    private static void setMessageType(Types.BaseGroupBuilder<?, ?> builder,
                                        PrimitiveType.PrimitiveTypeName primitiveName,
                                        LogicalTypeAnnotation logicType,
                                        CreateCollectionReq.FieldSchema field,
@@ -75,6 +75,29 @@ public class ParquetUtils {
             }
 
             switch (field.getDataType()) {
+                case Bool:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BOOLEAN, null, field, false);
+                    break;
+                case Int8:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32,
+                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(8, true), field, false);
+                    break;
+                case Int16:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32,
+                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(16, true), field, false);
+                    break;
+                case Int32:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32, null, field, false);
+                    break;
+                case Int64:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT64, null, field, false);
+                    break;
+                case Float:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, field, false);
+                    break;
+                case Double:
+                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.DOUBLE, null, field, false);
+                    break;
                 case FloatVector:
                     setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, field, true);
                     break;
@@ -89,10 +112,6 @@ public class ParquetUtils {
                 case Array:
                     fillArrayType(messageTypeBuilder, field);
                     break;
-
-                case Int64:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT64, null, field, false);
-                    break;
                 case VarChar:
                 case Geometry:
                 case Timestamptz:
@@ -101,30 +120,14 @@ public class ParquetUtils {
                     setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BINARY,
                             LogicalTypeAnnotation.stringType(), field, false);
                     break;
-                case Int8:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32,
-                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(8, true), field, false);
-                    break;
-                case Int16:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32,
-                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(16, true), field, false);
-                    break;
-                case Int32:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32, null, field, false);
-                    break;
-                case Float:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, field, false);
-                    break;
-                case Double:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.DOUBLE, null, field, false);
-                    break;
-                case Bool:
-                    setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BOOLEAN, null, field, false);
-                    break;
-
             }
         }
 
+        List<CreateCollectionReq.StructFieldSchema> structFields = collectionSchema.getStructFields();
+        for (CreateCollectionReq.StructFieldSchema struct : structFields) {
+            fillStructType(messageTypeBuilder, struct);
+        }
+
         if (collectionSchema.isEnableDynamicField()) {
             messageTypeBuilder.optional(PrimitiveType.PrimitiveTypeName.BINARY).as(LogicalTypeAnnotation.stringType())
                     .named(DYNAMIC_FIELD_NAME);
@@ -134,12 +137,8 @@ public class ParquetUtils {
 
     private static void fillArrayType(Types.MessageTypeBuilder messageTypeBuilder, CreateCollectionReq.FieldSchema field) {
         switch (field.getElementType()) {
-            case Int64:
-                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT64, null, field, true);
-                break;
-            case VarChar:
-                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BINARY,
-                        LogicalTypeAnnotation.stringType(), field, true);
+            case Bool:
+                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BOOLEAN, null, field, true);
                 break;
             case Int8:
                 setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32,
@@ -152,18 +151,63 @@ public class ParquetUtils {
             case Int32:
                 setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT32, null, field, true);
                 break;
+            case Int64:
+                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.INT64, null, field, true);
+                break;
             case Float:
                 setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, field, true);
                 break;
             case Double:
                 setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.DOUBLE, null, field, true);
                 break;
-            case Bool:
-                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BOOLEAN, null, field, true);
+            case VarChar:
+                setMessageType(messageTypeBuilder, PrimitiveType.PrimitiveTypeName.BINARY,
+                        LogicalTypeAnnotation.stringType(), field, true);
                 break;
         }
     }
 
+    private static void fillStructType(Types.MessageTypeBuilder messageTypeBuilder, CreateCollectionReq.StructFieldSchema struct) {
+        Types.BaseListBuilder.GroupElementBuilder<?, ?> groupBuilder = messageTypeBuilder.optionalList().optionalGroupElement();
+        for (CreateCollectionReq.FieldSchema subField : struct.getFields()) {
+            switch (subField.getDataType()) {
+                case Bool:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.BOOLEAN, null, subField, false);
+                    break;
+                case Int8:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.INT32,
+                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(8, true), subField, false);
+                    break;
+                case Int16:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.INT32,
+                            LogicalTypeAnnotation.IntLogicalTypeAnnotation.intType(16, true), subField, false);
+                    break;
+                case Int32:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.INT32, null, subField, false);
+                    break;
+                case Int64:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.INT64, null, subField, false);
+                    break;
+                case Float:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, subField, false);
+                    break;
+                case Double:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.DOUBLE, null, subField, false);
+                    break;
+                case VarChar:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.BINARY,
+                            LogicalTypeAnnotation.stringType(), subField, false);
+                    break;
+                case FloatVector:
+                    setMessageType(groupBuilder, PrimitiveType.PrimitiveTypeName.FLOAT, null, subField, true);
+                    break;
+                default:
+                    break;
+            }
+        }
+        groupBuilder.named(struct.getName());
+    }
+
     public static Configuration getParquetConfiguration() {
         // set fs.file.impl.disable.cache to true for this issue: https://github.com/milvus-io/milvus-sdk-java/issues/1381
         Configuration configuration = new Configuration();

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/restful/BaseRestful.java

@@ -108,6 +108,9 @@ public class BaseRestful {
     private static void setDefaultOptionsIfCallCloud(Map<String, Object> params, String apiKey) {
         if (StringUtils.isNotEmpty(apiKey)) {
             Map<String, Object> options = new HashMap<>();
+            if (params.containsKey("options")) {
+                options = (Map<String, Object>) params.get("options");
+            }
             options.put("sdk", "java");
             options.put("scene", "bulkWriter");
 

sdk-bulkwriter/src/main/java/io/milvus/bulkwriter/writer/ParquetFileWriter.java

@@ -31,6 +31,7 @@ public class ParquetFileWriter implements FormatFileWriter {
     private String filePath;
     private MessageType messageType;
     private Map<String, CreateCollectionReq.FieldSchema> nameFieldType;
+    private Map<String, CreateCollectionReq.StructFieldSchema> nameStructFieldType;
 
     public ParquetFileWriter(CreateCollectionReq.CollectionSchema collectionSchema, String filePathPrefix) throws IOException {
         this.collectionSchema = collectionSchema;
@@ -79,6 +80,9 @@ public class ParquetFileWriter implements FormatFileWriter {
                     .build());
         }
         this.nameFieldType = nameFieldType;
+
+        this.nameStructFieldType = collectionSchema.getStructFields().stream()
+                .collect(Collectors.toMap(CreateCollectionReq.StructFieldSchema::getName, e -> e));
     }
 
     @Override
@@ -92,7 +96,11 @@ public class ParquetFileWriter implements FormatFileWriter {
                 if (value == null) {
                     continue;
                 }
-                appendGroup(group, fieldName, value, nameFieldType.get(fieldName));
+                if (nameFieldType.containsKey(fieldName)) {
+                    appendGroup(group, value, nameFieldType.get(fieldName));
+                } else if (nameStructFieldType.containsKey(fieldName)) {
+                    appendStructGroup(group, value, nameStructFieldType.get(fieldName));
+                }
             }
             writer.write(group);
         } catch (IOException e) {
@@ -111,46 +119,47 @@ public class ParquetFileWriter implements FormatFileWriter {
         this.writer.close();
     }
 
-    private void appendGroup(Group group, String paramName, Object value, CreateCollectionReq.FieldSchema field) {
+    private void appendGroup(Group group, Object value, CreateCollectionReq.FieldSchema field) {
         io.milvus.v2.common.DataType dataType = field.getDataType();
+        String fieldName = field.getName();
         switch (dataType) {
             case Int8:
             case Int16:
-                group.append(paramName, (Short) value);
+                group.append(fieldName, (Short) value);
                 break;
             case Int32:
-                group.append(paramName, (Integer) value);
+                group.append(fieldName, (Integer) value);
                 break;
             case Int64:
-                group.append(paramName, (Long) value);
+                group.append(fieldName, (Long) value);
                 break;
             case Float:
-                group.append(paramName, (Float) value);
+                group.append(fieldName, (Float) value);
                 break;
             case Double:
-                group.append(paramName, (Double) value);
+                group.append(fieldName, (Double) value);
                 break;
             case Bool:
-                group.append(paramName, (Boolean) value);
+                group.append(fieldName, (Boolean) value);
                 break;
             case VarChar:
             case String:
             case Geometry:
             case Timestamptz:
             case JSON:
-                group.append(paramName, (String) value);
+                group.append(fieldName, (String) value);
                 break;
             case FloatVector:
-                addFloatArray(group, paramName, (List<Float>) value);
+                addFloatArray(group, fieldName, (List<Float>) value);
                 break;
             case BinaryVector:
             case Float16Vector:
             case BFloat16Vector:
             case Int8Vector:
-                addBinaryVector(group, paramName, (ByteBuffer) value);
+                addBinaryVector(group, fieldName, (ByteBuffer) value);
                 break;
             case SparseFloatVector:
-                addSparseVector(group, paramName, (SortedMap<Long, Float>) value);
+                addSparseVector(group, fieldName, (SortedMap<Long, Float>) value);
                 break;
             case Array:
                 io.milvus.v2.common.DataType elementType = field.getElementType();
@@ -158,28 +167,43 @@ public class ParquetFileWriter implements FormatFileWriter {
                     case Int8:
                     case Int16:
                     case Int32:
-                        addIntArray(group, paramName, (List<Integer>) value);
+                        addIntArray(group, fieldName, (List<Integer>) value);
                         break;
                     case Int64:
-                        addLongArray(group, paramName, (List<Long>) value);
+                        addLongArray(group, fieldName, (List<Long>) value);
                         break;
                     case Float:
-                        addFloatArray(group, paramName, (List<Float>) value);
+                        addFloatArray(group, fieldName, (List<Float>) value);
                         break;
                     case Double:
-                        addDoubleArray(group, paramName, (List<Double>) value);
+                        addDoubleArray(group, fieldName, (List<Double>) value);
                         break;
                     case String:
                     case VarChar:
-                        addStringArray(group, paramName, (List<String>) value);
+                        addStringArray(group, fieldName, (List<String>) value);
                         break;
                     case Bool:
-                        addBooleanArray(group, paramName, (List<Boolean>) value);
+                        addBooleanArray(group, fieldName, (List<Boolean>) value);
                         break;
                 }
         }
     }
 
+    private void appendStructGroup(Group group, Object value, CreateCollectionReq.StructFieldSchema field) {
+        Group arrayGroup = group.addGroup(field.getName());
+        Group listGroup = arrayGroup.addGroup(0);
+        List<Map<String, Object>> structs = (List<Map<String, Object>>) value;
+        for (Map<String, Object> struct : structs) {
+            Group dict = listGroup.addGroup(0);
+            for (CreateCollectionReq.FieldSchema subField : field.getFields()) {
+                if (struct.containsKey(subField.getName())) {
+                    Object val = struct.get(subField.getName());
+                    appendGroup(dict, val, subField);
+                }
+            }
+        }
+    }
+
     private static void addLongArray(Group group, String fieldName, List<Long> values) {
         Group arrayGroup = group.addGroup(fieldName);
         for (long value : values) {

sdk-bulkwriter/src/test/java/io/milvus/bulkwriter/BulkWriterTest.java

@@ -19,16 +19,15 @@
 
 package io.milvus.bulkwriter;
 
-import com.google.gson.JsonElement;
-import com.google.gson.JsonNull;
-import com.google.gson.JsonObject;
-import com.google.gson.JsonPrimitive;
+import com.google.gson.*;
+import com.google.gson.reflect.TypeToken;
 import io.milvus.bulkwriter.common.clientenum.BulkFileType;
 import io.milvus.bulkwriter.common.utils.GeneratorUtils;
 import io.milvus.bulkwriter.common.utils.ParquetReaderUtils;
 import io.milvus.bulkwriter.common.utils.V2AdapterUtils;
 import io.milvus.common.utils.JsonUtils;
 import io.milvus.exception.MilvusException;
+import io.milvus.param.Constant;
 import io.milvus.param.collection.CollectionSchemaParam;
 import io.milvus.param.collection.FieldType;
 import io.milvus.v2.common.DataType;
@@ -39,7 +38,13 @@ import org.apache.avro.util.Utf8;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.io.BufferedReader;
+import java.io.FileReader;
 import java.io.IOException;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.*;
 
 public class BulkWriterTest {
@@ -127,7 +132,7 @@ public class BulkWriterTest {
         schemaV2.addField(AddFieldReq.builder()
                 .fieldName("int8_field")
                 .dataType(DataType.Int8)
-                .defaultValue((short)8)
+                .defaultValue((short) 8)
                 .build());
         schemaV2.addField(AddFieldReq.builder()
                 .fieldName("int16_field")
@@ -207,6 +212,50 @@ public class BulkWriterTest {
                 .dataType(DataType.Int8Vector)
                 .dimension(DIMENSION)
                 .build());
+        schemaV2.addField(AddFieldReq.builder()
+                .fieldName("struct_field")
+                .dataType(DataType.Array)
+                .elementType(DataType.Struct)
+                .maxCapacity(100)
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_bool")
+                        .dataType(DataType.Bool)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int8")
+                        .dataType(DataType.Int8)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int16")
+                        .dataType(DataType.Int16)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int32")
+                        .dataType(DataType.Int32)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_int64")
+                        .dataType(DataType.Int64)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_float")
+                        .dataType(DataType.Float)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_double")
+                        .dataType(DataType.Double)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_string")
+                        .dataType(DataType.VarChar)
+                        .maxLength(100)
+                        .build())
+                .addStructField(AddFieldReq.builder()
+                        .fieldName("st_float_vector")
+                        .dataType(DataType.FloatVector)
+                        .dimension(DIMENSION)
+                        .build())
+                .build());
         return schemaV2;
     }
 
@@ -220,7 +269,7 @@ public class BulkWriterTest {
             }
 
             // some rows contains null values
-            if (i%5 == 0) {
+            if (i % 5 == 0) {
                 // scalar field
 //                rowObject.addProperty("bool_field", true); // nullable, no need to provide
 //                rowObject.add("int8_field", null); // has default value, no need to provide
@@ -235,8 +284,8 @@ public class BulkWriterTest {
                 // array field
                 rowObject.add("arr_int32_field", JsonUtils.toJsonTree(GeneratorUtils.generatorInt32Value(random.nextInt(4)))); // not nullable, must provide
 //                rowObject.add("arr_float_field", null); // nullable, no need to provide
-//                rowObject.add("arr_varchar_field", null); // nullable, no need to provide
-            } else if (i%3 == 0) {
+                rowObject.add("arr_varchar_field", JsonUtils.toJsonTree(new ArrayList<>())); //empty array
+            } else if (i % 3 == 0) {
                 // scalar field
                 rowObject.add("bool_field", null); // nullable, set null is ok
                 rowObject.add("int8_field", null); // has default value, set null to get default
@@ -281,6 +330,23 @@ public class BulkWriterTest {
             rowObject.add("sparse_vector_field", JsonUtils.toJsonTree(utils.generateSparseVector()));
             rowObject.add("int8_vector_field", JsonUtils.toJsonTree(utils.generateInt8Vector().array()));
 
+            // struct field
+            List<JsonObject> structList = new ArrayList<>();
+            for (int k = 0; k < i % 4 + 1; k++) {
+                JsonObject st = new JsonObject();
+                st.addProperty("st_bool", (i + k) % 3 == 0);
+                st.addProperty("st_int8", (i + k) % 128);
+                st.addProperty("st_int16", (i + k) % 16384);
+                st.addProperty("st_int32", (i + k) % 65536);
+                st.addProperty("st_int64", i + k);
+                st.addProperty("st_float", (float) (i + k) / 4);
+                st.addProperty("st_double", (i + k) / 3);
+                st.addProperty("st_string", String.format("dummy_%d", i + k));
+                st.add("st_float_vector", JsonUtils.toJsonTree(utils.generateFloatVector()));
+                structList.add(st);
+            }
+            rowObject.add("struct_field", JsonUtils.toJsonTree(structList));
+
             rows.add(rowObject);
         }
         return rows;
@@ -293,15 +359,15 @@ public class BulkWriterTest {
     }
 
     private static JsonElement constructJsonPrimitive(Object obj) {
-        if (obj == null ) {
+        if (obj == null) {
             return JsonNull.INSTANCE;
         } else if (obj instanceof Boolean) {
-            return new JsonPrimitive((Boolean)obj);
+            return new JsonPrimitive((Boolean) obj);
         } else if (obj instanceof Short || obj instanceof Integer || obj instanceof Long ||
                 obj instanceof Float || obj instanceof Double) {
             return new JsonPrimitive((Number) obj);
         } else if (obj instanceof String) {
-            return new JsonPrimitive((String)obj);
+            return new JsonPrimitive((String) obj);
         }
 
         Assertions.fail("Default value is illegal");
@@ -334,7 +400,7 @@ public class BulkWriterTest {
             if (fieldSchemaV2.getDimension() != null) {
                 Assertions.assertEquals(fieldSchemaV2.getDimension(), fieldSchemaV1.getDimension());
             }
-            if(fieldSchemaV2.getDataType() == DataType.VarChar) {
+            if (fieldSchemaV2.getDataType() == DataType.VarChar) {
                 Assertions.assertEquals(fieldSchemaV2.getMaxLength(), fieldSchemaV1.getMaxLength());
             }
 
@@ -360,7 +426,7 @@ public class BulkWriterTest {
                     .withLocalPath("/tmp/bulk_writer")
                     .withFileType(fileType)
                     .build();
-            try(LocalBulkWriter localBulkWriter = new LocalBulkWriter(bulkWriterParam)) {
+            try (LocalBulkWriter localBulkWriter = new LocalBulkWriter(bulkWriterParam)) {
                 JsonObject rowObject = new JsonObject();
                 rowObject.addProperty("bool_field", true);
                 rowObject.addProperty("int8_field", 1);
@@ -378,22 +444,23 @@ public class BulkWriterTest {
                 rowObject.add("arr_int32_field", JsonUtils.toJsonTree(GeneratorUtils.generatorInt32Value(2)));
                 rowObject.add("arr_float_field", JsonUtils.toJsonTree(GeneratorUtils.generatorFloatValue(3)));
                 rowObject.add("arr_varchar_field", JsonUtils.toJsonTree(GeneratorUtils.generatorVarcharValue(4, 5)));
+                rowObject.add("struct_field", JsonUtils.toJsonTree(new ArrayList<>()));
 
                 // a field missed, expect throwing an exception
 //            localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // id is auto_id, no need to input, expect throwing an exception
                 rowObject.addProperty("id", 1);
                 rowObject.addProperty("int16_field", 2);
 //            localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set null value for non-nullable field, expect throwing an exception
                 rowObject.remove("id");
                 rowObject.add("int16_field", null);
 //            localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set valid value for dynamic field
                 rowObject.addProperty("int16_field", 16);
@@ -405,48 +472,84 @@ public class BulkWriterTest {
                 // set invalid value for dynamic field, expect throwing an exception
                 rowObject.addProperty("$meta", 6);
 //            localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set incorrect dimension vector, expect throwing an exception
                 rowObject.remove("$meta");
-                rowObject.add("float_vector_field", JsonUtils.toJsonTree(utils.generateFloatVector(DIMENSION-1)));
+                rowObject.add("float_vector_field", JsonUtils.toJsonTree(utils.generateFloatVector(DIMENSION - 1)));
 //                localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set incorrect sparse vector, expect throwing an exception
                 rowObject.add("float_vector_field", JsonUtils.toJsonTree(utils.generateFloatVector()));
                 rowObject.add("sparse_vector_field", JsonUtils.toJsonTree(utils.generateFloatVector()));
 //                localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set incorrect value type for scalar field, expect throwing an exception
                 rowObject.add("sparse_vector_field", JsonUtils.toJsonTree(utils.generateSparseVector()));
                 rowObject.addProperty("float_field", Boolean.TRUE);
 //                localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set incorrect type for varchar field, expect throwing an exception
                 rowObject.addProperty("float_field", 2.5);
                 rowObject.addProperty("varchar_field", 2.5);
 //                localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
 
                 // set incorrect value type for int8 vector field, expect throwing an exception
                 rowObject.addProperty("varchar_field", "dummy");
                 rowObject.addProperty("int8_vector_field", Boolean.TRUE);
 //                localBulkWriter.appendRow(rowObject);
-                Assertions.assertThrows(MilvusException.class, ()->localBulkWriter.appendRow(rowObject));
+                Assertions.assertThrows(MilvusException.class, () -> localBulkWriter.appendRow(rowObject));
             } catch (Exception e) {
                 Assertions.fail(e.getMessage());
             }
         }
     }
 
+    private static void compareJsonArray(JsonElement j1, JsonElement j2, DataType dt) {
+        if (j1.isJsonNull()) {
+            Assertions.assertTrue(j2.isJsonNull());
+            return;
+        }
+        Assertions.assertTrue(j1.isJsonArray());
+        Assertions.assertTrue(j2.isJsonArray());
+        List<JsonElement> a1 = j1.getAsJsonArray().asList();
+        List<JsonElement> a2 = j2.getAsJsonArray().asList();
+        Assertions.assertEquals(a1.size(), a2.size());
+        for (int i = 0; i < a1.size(); i++) {
+            switch (dt) {
+                case Bool:
+                    Assertions.assertEquals(a1.get(i).getAsBoolean(), a2.get(i).getAsBoolean());
+                    break;
+                case Int8:
+                case Int16:
+                case Int32:
+                    Assertions.assertEquals(a1.get(i).getAsInt(), a2.get(i).getAsInt());
+                    break;
+                case Float:
+                    Assertions.assertEquals(a1.get(i).getAsFloat(), a2.get(i).getAsFloat());
+                    break;
+                case Double:
+                    Assertions.assertEquals(a1.get(i).getAsDouble(), a2.get(i).getAsDouble());
+                    break;
+                case VarChar:
+                    Assertions.assertEquals(a1.get(i).getAsString(), a2.get(i).getAsString());
+                default:
+                    Assertions.assertEquals(a1.get(i), a2.get(i));
+                    break;
+            }
+        }
+    }
+
     @Test
     void testWriteJson() {
         try {
-            boolean autoID = true;
-            boolean enableDynamicField = true;
+            final boolean autoID = true;
+            final boolean enableDynamicField = true;
+            final int row_count = 10;
             CreateCollectionReq.CollectionSchema schemaV2 = buildV2Schema(enableDynamicField, autoID);
             LocalBulkWriterParam bulkWriterParam = LocalBulkWriterParam.newBuilder()
                     .withCollectionSchema(schemaV2)
@@ -454,7 +557,7 @@ public class BulkWriterTest {
                     .withFileType(BulkFileType.JSON)
                     .build();
             LocalBulkWriter localBulkWriter = new LocalBulkWriter(bulkWriterParam);
-            List<JsonObject> rows = buildData(10, enableDynamicField, autoID);
+            List<JsonObject> rows = buildData(row_count, enableDynamicField, autoID);
             writeData(localBulkWriter, rows);
 
             System.out.printf("%s rows appends%n", localBulkWriter.getTotalRowCount());
@@ -463,6 +566,66 @@ public class BulkWriterTest {
             System.out.println(filePaths);
             Assertions.assertEquals(1, filePaths.size());
             Assertions.assertEquals(1, filePaths.get(0).size());
+
+            Gson gson = new Gson();
+            Reader reader = Files.newBufferedReader(Paths.get(filePaths.get(0).get(0)));
+            List<JsonObject> lines = gson.fromJson(reader, new TypeToken<List<JsonObject>>() {
+            }.getType());
+            Assertions.assertEquals(row_count, lines.size());
+            for (int i = 0; i < lines.size(); i++) {
+                JsonObject expectedDict = rows.get(i);
+                JsonObject readDict = lines.get(i);
+                Assertions.assertTrue(readDict.has(Constant.DYNAMIC_FIELD_NAME));
+                Assertions.assertTrue(expectedDict.keySet().size() == readDict.keySet().size() ||
+                        expectedDict.keySet().size() + 1 == readDict.keySet().size());
+                for (String key : readDict.keySet()) {
+                    if (key.equals(Constant.DYNAMIC_FIELD_NAME)) {
+                        continue;
+                    }
+
+                    JsonElement expectVal = expectedDict.get(key);
+                    JsonElement readVal = readDict.get(key);
+                    if (key.equals("sparse_vector_field")) {
+                        JsonObject expectSparse = (JsonObject) expectVal;
+                        JsonObject readSparse = (JsonObject) readVal;
+                        Assertions.assertEquals(expectSparse.keySet().size(), readSparse.keySet().size());
+                        for (String id : expectSparse.keySet()) {
+                            Assertions.assertTrue(readSparse.has(id));
+                            Assertions.assertEquals(expectSparse.get(id).getAsFloat(), readSparse.get(id).getAsFloat());
+                        }
+                    } else if (key.equals("float_vector_field") || key.equals("arr_float_field")) {
+                        compareJsonArray(expectVal, readVal, DataType.Float);
+                    } else if (key.equals("json_field")) {
+                        if (expectVal.isJsonNull()) {
+                            Assertions.assertTrue(readVal.isJsonNull());
+                        } else {
+                            String str = expectVal.toString();
+                            Assertions.assertEquals(str, readVal.getAsString());
+                        }
+                    } else if (key.equals("arr_varchar_field")) {
+                        compareJsonArray(expectVal, readVal, DataType.VarChar);
+                    } else if (key.equals("struct_field")) {
+                        JsonArray expectStructs = (JsonArray) expectVal;
+                        JsonArray readStructs = (JsonArray) readVal;
+                        Assertions.assertEquals(expectStructs.size(), readStructs.size());
+                        for (int k = 0; k < expectStructs.size(); k++) {
+                            JsonObject expectStruct = (JsonObject) expectStructs.get(k);
+                            JsonObject readStruct = (JsonObject) readStructs.get(k);
+                            Assertions.assertEquals(expectStruct.keySet().size(), readStruct.keySet().size());
+                            for (String id : expectStruct.keySet()) {
+                                Assertions.assertTrue(readStruct.has(id));
+                                if (id.equals("st_float_vector")) {
+                                    compareJsonArray(expectStruct.get(id), readStruct.get(id), DataType.Float);
+                                } else {
+                                    Assertions.assertEquals(expectStruct.get(id), readStruct.get(id));
+                                }
+                            }
+                        }
+                    } else {
+                        Assertions.assertEquals(expectVal, readVal);
+                    }
+                }
+            }
         } catch (Exception e) {
             Assertions.fail(e.getMessage());
         }
@@ -471,18 +634,20 @@ public class BulkWriterTest {
     @Test
     void testWriteCSV() {
         try {
-            boolean autoID = true;
-            boolean enableDynamicField = true;
+            final boolean autoID = true;
+            final boolean enableDynamicField = true;
+            final int row_count = 10;
+            final String nullKey = "XXX";
             CreateCollectionReq.CollectionSchema schemaV2 = buildV2Schema(enableDynamicField, autoID);
             LocalBulkWriterParam bulkWriterParam = LocalBulkWriterParam.newBuilder()
                     .withCollectionSchema(schemaV2)
                     .withLocalPath("/tmp/bulk_writer")
                     .withFileType(BulkFileType.CSV)
                     .withConfig("sep", "|")
-                    .withConfig("nullkey", "XXX")
+                    .withConfig("nullkey", nullKey)
                     .build();
             LocalBulkWriter localBulkWriter = new LocalBulkWriter(bulkWriterParam);
-            List<JsonObject> rows = buildData(10, enableDynamicField, autoID);
+            List<JsonObject> rows = buildData(row_count, enableDynamicField, autoID);
             writeData(localBulkWriter, rows);
 
             System.out.printf("%s rows appends%n", localBulkWriter.getTotalRowCount());
@@ -491,6 +656,75 @@ public class BulkWriterTest {
             System.out.println(filePaths);
             Assertions.assertEquals(1, filePaths.size());
             Assertions.assertEquals(1, filePaths.get(0).size());
+
+            try (BufferedReader br = new BufferedReader(new FileReader(filePaths.get(0).get(0)))) {
+                String line;
+                List<String> header = new ArrayList<>();
+                int num_line = 0;
+                while ((line = br.readLine()) != null) {
+                    if (header.isEmpty()) {
+                        header = Arrays.asList(line.split("\\|"));
+                        Assertions.assertTrue(header.contains(Constant.DYNAMIC_FIELD_NAME));
+                        continue;
+                    }
+                    JsonObject expectedRow = rows.get(num_line++);
+                    String[] values = line.split("\\|");
+                    Assertions.assertTrue(values.length == expectedRow.size() || values.length == expectedRow.size() + 1);
+                    for (int i = 0; i < header.size(); i++) {
+                        String field = header.get(i);
+                        if (field.equals(Constant.DYNAMIC_FIELD_NAME)) {
+                            continue;
+                        }
+
+                        Assertions.assertTrue(expectedRow.has(field));
+                        JsonElement expectEle = expectedRow.get(field);
+                        String readStr = values[i];
+                        if (expectEle.isJsonNull()) {
+                            Assertions.assertEquals(String.format("\"%s\"", nullKey), readStr);
+                        } else if (expectEle.isJsonArray()) {
+                            if (field.equals("struct_field")) {
+                                Gson gson = new Gson();
+                                readStr = readStr.substring(1, readStr.length() - 1);
+                                readStr = readStr.replace("\"\"", "\"");
+                                List<JsonObject> readStructs = gson.fromJson(readStr, new TypeToken<List<JsonObject>>() {
+                                }.getType());
+                                List<JsonElement> expectStructs = expectEle.getAsJsonArray().asList();
+                                Assertions.assertEquals(expectStructs.size(), readStructs.size());
+                                for (int k = 0; k < expectStructs.size(); k++) {
+                                    JsonObject expectStruct = expectStructs.get(k).getAsJsonObject();
+                                    JsonObject readStruct = readStructs.get(k);
+                                    for (String key : expectStruct.keySet()) {
+                                        Assertions.assertTrue(readStruct.has(key));
+                                        if (expectStruct.get(key).isJsonArray()) {
+                                            Assertions.assertTrue(readStruct.get(key).isJsonArray());
+                                            List<JsonElement> expectVals = expectStruct.get(key).getAsJsonArray().asList();
+                                            List<JsonElement> readVals = readStruct.get(key).getAsJsonArray().asList();
+                                            Assertions.assertEquals(expectVals.size(), readVals.size());
+                                            for (int j = 0; j < expectVals.size(); j++) {
+                                                Assertions.assertEquals(expectVals.get(j).getAsFloat(), readVals.get(j).getAsFloat());
+                                            }
+                                        } else {
+                                            Assertions.assertEquals(expectStruct.get(key), readStruct.get(key));
+                                        }
+                                    }
+                                }
+                            } else {
+                                String expectStr = String.format("\"%s\"", expectEle.getAsJsonArray().toString());
+                                readStr = readStr.replace(" ", "");
+                                readStr = readStr.replace("\"\"", "\"");
+                                Assertions.assertEquals(expectStr, readStr);
+                            }
+                        } else if (expectEle.isJsonObject()) {
+                            String expectStr = String.format("\"%s\"", expectEle.getAsJsonObject().toString());
+                            expectStr = expectStr.replace("\"", "\"\"");
+                            readStr = String.format("\"%s\"", readStr);
+                            Assertions.assertEquals(expectStr, readStr);
+                        }
+                    }
+                }
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
         } catch (Exception e) {
             Assertions.fail(e.getMessage());
         }
@@ -501,6 +735,7 @@ public class BulkWriterTest {
         String ss2 = s2.replace("\\\"", "\"").replaceAll("^\"|\"$", "");
         Assertions.assertEquals(ss1, ss2);
     }
+
     private static void verifyElement(DataType dtype, JsonElement element, Object obj) {
         switch (dtype) {
             case Bool:
@@ -524,19 +759,20 @@ public class BulkWriterTest {
             case Geometry:
             case Timestamptz:
             case JSON:
-                verifyJsonString(element.getAsString(), ((Utf8)obj).toString());
+                verifyJsonString(element.getAsString(), ((Utf8) obj).toString());
                 break;
             case SparseFloatVector:
-                verifyJsonString(element.toString(), ((Utf8)obj).toString());
+                verifyJsonString(element.toString(), ((Utf8) obj).toString());
                 break;
             default:
                 break;
         }
     }
 
-    private static void verifyRow(List<CreateCollectionReq.FieldSchema> fieldsList, List<JsonObject> originalData, GenericData.Record readRow) {
-        long id = (long)readRow.get("id");
-        JsonObject expectedRow = originalData.get((int)id);
+    private static void verifyParquetRow(CreateCollectionReq.CollectionSchema schema, List<JsonObject> originalData, GenericData.Record readRow) {
+        long id = (long) readRow.get("id");
+        JsonObject expectedRow = originalData.get((int) id);
+        List<CreateCollectionReq.FieldSchema> fieldsList = schema.getFieldSchemaList();
         for (CreateCollectionReq.FieldSchema field : fieldsList) {
             String fieldName = field.getName();
             Object readValue = readRow.get(fieldName);
@@ -563,7 +799,7 @@ public class BulkWriterTest {
                         Assertions.fail("Array field type unmatched");
                     }
                     List<JsonElement> jsonArr = expectedEle.getAsJsonArray().asList();
-                    List<GenericData.Record> objArr = (List<GenericData.Record>)readValue;
+                    List<GenericData.Record> objArr = (List<GenericData.Record>) readValue;
                     if (jsonArr.size() != objArr.size()) {
                         Assertions.fail("Array field length unmatched");
                     }
@@ -590,14 +826,80 @@ public class BulkWriterTest {
                     break;
             }
         }
+
+        List<CreateCollectionReq.StructFieldSchema> structList = schema.getStructFields();
+        for (CreateCollectionReq.StructFieldSchema struct : structList) {
+            String fieldName = struct.getName();
+            Object readValue = readRow.get(fieldName);
+            Assertions.assertInstanceOf(List.class, readValue);
+            List<?> readStructs = (List<?>) readValue;
+            JsonElement expectedEle = expectedRow.get(fieldName);
+            Assertions.assertNotNull(expectedEle);
+            JsonArray jsonArr = expectedEle.getAsJsonArray();
+            Assertions.assertNotNull(jsonArr);
+            List<JsonElement> jsonList = jsonArr.asList();
+            Assertions.assertEquals(jsonList.size(), readStructs.size());
+
+            for (int i = 0; i < jsonList.size(); i++) {
+                JsonObject expectedDict = (JsonObject) jsonList.get(i);
+                GenericData.Record readDict = (GenericData.Record) readStructs.get(i);
+                readDict = (GenericData.Record) readDict.get("element");
+                for (String key : expectedDict.keySet()) {
+                    Assertions.assertTrue(readDict.hasField(key));
+                }
+
+                Object expectedVal = expectedDict.get("st_bool").getAsBoolean();
+                Object readVal = readDict.get("st_bool");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_int8").getAsInt();
+                readVal = readDict.get("st_int8");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_int16").getAsInt();
+                readVal = readDict.get("st_int16");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_int32").getAsInt();
+                readVal = readDict.get("st_int32");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_int64").getAsLong();
+                readVal = readDict.get("st_int64");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_float").getAsFloat();
+                readVal = readDict.get("st_float");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_double").getAsDouble();
+                readVal = readDict.get("st_double");
+                Assertions.assertEquals(expectedVal, readVal);
+
+                expectedVal = expectedDict.get("st_string").getAsString();
+                Utf8 utf = (Utf8) readDict.get("st_string");
+                String decodedString = new String(utf.getBytes(), StandardCharsets.UTF_8);
+                Assertions.assertEquals(expectedVal, decodedString);
+
+                List<JsonElement> expectedArr = expectedDict.get("st_float_vector").getAsJsonArray().asList();
+                List<?> readArr = (List<?>) readDict.get("st_float_vector");
+                Assertions.assertEquals(expectedArr.size(), readArr.size());
+                for (int k = 0; k < readArr.size(); k++) {
+                    expectedVal = expectedArr.get(k).getAsFloat();
+                    readVal = ((GenericData.Record) readArr.get(k)).get("element");
+                    Assertions.assertEquals(expectedVal, readVal);
+                }
+            }
+        }
         System.out.printf("The row of id=%d is correct%n", id);
     }
 
     @Test
     public void testWriteParquet() {
         // collection schema
-        boolean autoID = false;
-        boolean enableDynamicField = false;
+        final boolean autoID = false;
+        final boolean enableDynamicField = false;
+        final int rowCount = 10;
         CreateCollectionReq.CollectionSchema schemaV2 = buildV2Schema(enableDynamicField, autoID);
 
         // local bulkwriter
@@ -608,18 +910,17 @@ public class BulkWriterTest {
                 .withChunkSize(100 * 1024)
                 .build();
 
-        int rowCount = 10;
         List<JsonObject> originalData = new ArrayList<>();
         List<List<String>> batchFiles = new ArrayList<>();
         try (LocalBulkWriter bulkWriter = new LocalBulkWriter(writerParam)) {
-            originalData = buildData(10, enableDynamicField, autoID);
+            originalData = buildData(rowCount, enableDynamicField, autoID);
             writeData(bulkWriter, originalData);
 
             bulkWriter.commit(false);
             List<List<String>> files = bulkWriter.getBatchFiles();
             System.out.printf("LocalBulkWriter done! output local files: %s%n", files);
             Assertions.assertEquals(1, files.size());
-            Assertions.assertEquals(files.get(0).size(), 1);
+            Assertions.assertEquals(1, files.get(0).size());
             batchFiles.addAll(files);
         } catch (Exception e) {
             System.out.println("LocalBulkWriter catch exception: " + e);
@@ -632,12 +933,11 @@ public class BulkWriterTest {
             final int[] counter = {0};
             for (List<String> files : batchFiles) {
                 List<JsonObject> finalOriginalData = originalData;
-                List<CreateCollectionReq.FieldSchema> fieldsList = schemaV2.getFieldSchemaList();
                 new ParquetReaderUtils() {
                     @Override
                     public void readRecord(GenericData.Record record) {
                         counter[0]++;
-                        verifyRow(fieldsList, finalOriginalData, record);
+                        verifyParquetRow(schemaV2, finalOriginalData, record);
                     }
                 }.readParquet(files.get(0));
             }

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

@@ -446,6 +446,15 @@ public class CreateCollectionReq {
             this.fieldSchemaList = fieldSchemaList;
         }
 
+        public CreateCollectionReq.StructFieldSchema getStructField(String fieldName) {
+            for (CreateCollectionReq.StructFieldSchema field : structFields) {
+                if (field.getName().equals(fieldName)) {
+                    return field;
+                }
+            }
+            return null;
+        }
+
         public List<CreateCollectionReq.StructFieldSchema> getStructFields() {
             return structFields;
         }