clean up

Signed-off-by: shaoting-huang <shaoting.huang@zilliz.com>

internal/storage/serde_events.go

@@ -430,25 +430,24 @@ func (dsw *DeltalogStreamWriter) GetRecordWriter() (RecordWriter, error) {
 	if dsw.rw != nil {
 		return dsw.rw, nil
 	}
-	var dataType arrow.DataType
 	switch dsw.fieldSchema.DataType {
-	case schemapb.DataType_Int64:
-		dataType = &arrow.Int64Type{}
-	case schemapb.DataType_String:
-		dataType = arrow.BinaryTypes.String
+	case schemapb.DataType_Int64, schemapb.DataType_VarChar:
+		dim, _ := typeutil.GetDim(dsw.fieldSchema)
+		rw, err := newSingleFieldRecordWriter(dsw.fieldSchema.FieldID, arrow.Field{
+			Name:     dsw.fieldSchema.Name,
+			Type:     serdeMap[dsw.fieldSchema.DataType].arrowType(int(dim)),
+			Nullable: false, // No nullable check here.
+		}, &dsw.buf)
+		if err != nil {
+			return nil, err
+		}
+		dsw.rw = rw
+		return rw, nil
 	default:
-		return nil, fmt.Errorf("does not support data type")
+		return nil, merr.WrapErrServiceInternal(fmt.Sprintf(
+			"does not support delta log primary key data type %s",
+			dsw.fieldSchema.DataType.String()))
 	}
-	rw, err := newSingleFieldRecordWriter(dsw.fieldSchema.FieldID, arrow.Field{
-		Name:     dsw.fieldSchema.Name,
-		Type:     dataType,
-		Nullable: false,
-	}, &dsw.buf)
-	if err != nil {
-		return nil, err
-	}
-	dsw.rw = rw
-	return rw, nil
 }
 
 func (dsw *DeltalogStreamWriter) Finalize() (*Blob, error) {