[Bug]: LocalBulkWriter error

[luminous123]

Is there an existing issue for this?

• I have searched the existing issues

Environment

- Milvus version:2.4.x
- Deployment mode:standalone 
- SDK version: milvus-sdk-java v2.4.1
- OS:Ubuntu 
- CPU/Memory: 
- GPU: 
- Others:

Current Behavior

when I use LocalBulkWriter to write PARQUET file and I see the error message attached below :

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/software/maven/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.17.1/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/software/maven/repository/org/slf4j/slf4j-reload4j/1.7.36/slf4j-reload4j-1.7.36.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
2024-06-22 17:50:06 io.milvus.bulkwriter.LocalBulkWriter createDirIfNotExist
INFO: Data path created: /opt/software/milvus/data
2024-06-22 17:50:06 io.milvus.bulkwriter.LocalBulkWriter createDirIfNotExist
INFO: Data path created: /opt/software/milvus/data/1f2e92f2-f75b-4d70-988c-6ae31a601118
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Prepare to flush buffer, row_count: 10000, size: 31080000
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Flush thread begin, name: Thread-1
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Wait flush to finish
2024-06-22 17:50:12 org.apache.hadoop.util.NativeCodeLoader
WARN: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Exception in thread "Thread-1" java.lang.NullPointerException
at io.milvus.bulkwriter.Buffer.appendGroup(Buffer.java:197)
at io.milvus.bulkwriter.Buffer.persistParquet(Buffer.java:182)
at io.milvus.bulkwriter.Buffer.persist(Buffer.java:132)
at io.milvus.bulkwriter.LocalBulkWriter.flush(LocalBulkWriter.java:120)
at io.milvus.bulkwriter.LocalBulkWriter.lambda$commit$0(LocalBulkWriter.java:97)
at java.lang.Thread.run(Thread.java:748)
2024-06-22 17:50:14 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Commit done with async=false

Process finished with exit code 0

Expected Behavior

No response

Steps To Reproduce

package com.fiberhome.milvus;

import com.alibaba.fastjson.JSONObject;
import io.milvus.bulkwriter.LocalBulkWriter;
import io.milvus.bulkwriter.LocalBulkWriterParam;
import io.milvus.bulkwriter.common.clientenum.BulkFileType;
import io.milvus.grpc.DataType;

import io.milvus.param.collection.CollectionSchemaParam;
import io.milvus.param.collection.FieldType;

public class BulkWriter {
    public void writer() throws Exception{
        FieldType id = FieldType.newBuilder()
                .withName("id")
                .withDataType(DataType.Int64)
                .withPrimaryKey(true)
                .withAutoID(false)
                .build();

        FieldType vector = FieldType.newBuilder()
                .withName("vector")
                .withDataType(DataType.FloatVector)
                .withDimension(768)
                .build();

        FieldType scalar1 = FieldType.newBuilder()
                .withName("scalar_1")
                .withDataType(DataType.VarChar)
                .withMaxLength(512)
                .build();

        FieldType scalar2 = FieldType.newBuilder()
                .withName("scalar_2")
                .withDataType(DataType.Int64)
                .build();

        CollectionSchemaParam schema = CollectionSchemaParam.newBuilder()
                .withEnableDynamicField(true)
                .addFieldType(id)
                .addFieldType(vector)
                .addFieldType(scalar1)
                .addFieldType(scalar2)
                .build();

        LocalBulkWriterParam localBulkWriterParam = LocalBulkWriterParam.newBuilder()
                .withCollectionSchema(schema)
                .withLocalPath("/opt/software/milvus/data/")
                .withChunkSize(512 * 1024 * 1024)
                //.withChunkSize(1024 * 1024)
                .withFileType(BulkFileType.PARQUET)
                .build();

        LocalBulkWriter localBulkWriter = new LocalBulkWriter(localBulkWriterParam);

        for (int i = 0; i < 10000; i++) {
            JSONObject json = new JSONObject();
            json.put("id", i);
            json.put("vector", Util.generateFloatVectors(768));
            json.put("scalar_1", Util.getRandomString(20));
            json.put("scalar_2", (long) (Math.random() * 100));

            localBulkWriter.appendRow(json);
            //remoteBulkWriter.appendRow(json);
        }

        localBulkWriter.commit(false);

    }

    public static void main(String[] args) throws Exception{
        BulkWriter bulkWriter = new BulkWriter();
        bulkWriter.writer();
    }
}






package com.fiberhome.milvus;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

public class Util {

    public static String getRandomString(int length) {
        Random random = new Random();
        StringBuilder stringBuilder = new StringBuilder(length);

        for (int i = 0; i < length; i++) {
            int index = random.nextInt(Contants.CHARACTERS.length());
            stringBuilder.append(Contants.CHARACTERS.charAt(index));
        }
        return stringBuilder.toString();
    }

    public static List<Float> generateFloatVectors(int dimension) {
        List<Float> vector = new ArrayList();

        for (int i=0; i< dimension; i++) {
            Random rand = new Random();
            vector.add(rand.nextFloat());
        }

        return vector;
    }
}

Milvus Log

SLF4J: Class path contains multiple SLF4J bindings.
SLF4J: Found binding in [jar:file:/opt/software/maven/repository/org/apache/logging/log4j/log4j-slf4j-impl/2.17.1/log4j-slf4j-impl-2.17.1.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: Found binding in [jar:file:/opt/software/maven/repository/org/slf4j/slf4j-reload4j/1.7.36/slf4j-reload4j-1.7.36.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See http://www.slf4j.org/codes.html#multiple_bindings for an explanation.
SLF4J: Actual binding is of type [org.apache.logging.slf4j.Log4jLoggerFactory]
2024-06-22 17:50:06 io.milvus.bulkwriter.LocalBulkWriter createDirIfNotExist
INFO: Data path created: /opt/software/milvus/data
2024-06-22 17:50:06 io.milvus.bulkwriter.LocalBulkWriter createDirIfNotExist
INFO: Data path created: /opt/software/milvus/data/1f2e92f2-f75b-4d70-988c-6ae31a601118
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Prepare to flush buffer, row_count: 10000, size: 31080000
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Flush thread begin, name: Thread-1
2024-06-22 17:50:08 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Wait flush to finish
2024-06-22 17:50:12 org.apache.hadoop.util.NativeCodeLoader
WARN: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Exception in thread "Thread-1" java.lang.NullPointerException
at io.milvus.bulkwriter.Buffer.appendGroup(Buffer.java:197)
at io.milvus.bulkwriter.Buffer.persistParquet(Buffer.java:182)
at io.milvus.bulkwriter.Buffer.persist(Buffer.java:132)
at io.milvus.bulkwriter.LocalBulkWriter.flush(LocalBulkWriter.java:120)
at io.milvus.bulkwriter.LocalBulkWriter.lambda$commit$0(LocalBulkWriter.java:97)
at java.lang.Thread.run(Thread.java:748)
2024-06-22 17:50:14 io.milvus.bulkwriter.LocalBulkWriter commit
INFO: Commit done with async=false

Process finished with exit code 0

Anything else?

No response

[xiaofan-luan]

/assign @yhmo
please help on it

[yhmo]

This is a bug of BulkWriter in Java SDK v2.4.1. The BulklWriter cannot correctly handle collections with dynamic schema.
This bug has been fixed by this pr: milvus-io/milvus-sdk-java#899, which will be released in v2.4.2.

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.
Rotten issues close after 30d of inactivity. Reopen the issue with /reopen.