enhance: add delta log stream new format reader and writer

internal/storage/event_data.go

@@ -36,6 +36,11 @@ const (
 	nullableKey     = "nullable"
 )
 
+const version = "version"
+
+// mark useMultiFieldFormat if there are multi fields in a log file
+const MULTI_FIELD = "MULTI_FIELD"
+
 type descriptorEventData struct {
 	DescriptorEventDataFixPart
 	ExtraLength       int32

internal/storage/event_writer.go

@@ -212,6 +212,16 @@ func newDescriptorEvent() *descriptorEvent {
 	}
 }
 
+func NewBaseDescriptorEvent(collectionID int64, partitionID int64, segmentID int64) *descriptorEvent {
+	de := newDescriptorEvent()
+	de.CollectionID = collectionID
+	de.PartitionID = partitionID
+	de.SegmentID = segmentID
+	de.StartTimestamp = 0
+	de.EndTimestamp = 0
+	return de
+}
+
 func newInsertEventWriter(dataType schemapb.DataType, nullable bool, dim ...int) (*insertEventWriter, error) {
 	var payloadWriter PayloadWriterInterface
 	var err error

internal/storage/serde.go

@@ -35,6 +35,7 @@
 
 type Record interface {
 	Schema() map[FieldID]schemapb.DataType
+	ArrowSchema() *arrow.Schema
 	Column(i FieldID) arrow.Array
 	Len() int
 	Release()
@@ -83,6 +84,14 @@
 	return r.schema
 }
 
+func (r *compositeRecord) ArrowSchema() *arrow.Schema {
+	var fields []arrow.Field
+	for _, rec := range r.recs {
+		fields = append(fields, rec.Schema().Field(0))
+	}
+	return arrow.NewSchema(fields, nil)
+}
+
 type serdeEntry struct {
 	// arrowType returns the arrow type for the given dimension
 	arrowType func(int) arrow.DataType
@@ -575,6 +584,10 @@
 	return r.schema
 }
 
+func (r *selectiveRecord) ArrowSchema() *arrow.Schema {
+	return r.r.ArrowSchema()
+}
+
 func (r *selectiveRecord) Column(i FieldID) arrow.Array {
 	if i == r.selectedFieldId {
 		return r.r.Column(i)
@@ -663,9 +676,10 @@
 
 func newSingleFieldRecordWriter(fieldId FieldID, field arrow.Field, writer io.Writer) (*singleFieldRecordWriter, error) {
 	schema := arrow.NewSchema([]arrow.Field{field}, nil)
+
+	// use writer properties as same as payload writer's for now
 	fw, err := pqarrow.NewFileWriter(schema, writer,
 		parquet.NewWriterProperties(
-			parquet.WithMaxRowGroupLength(math.MaxInt64), // No additional grouping for now.
 			parquet.WithCompression(compress.Codecs.Zstd),
 			parquet.WithCompressionLevel(3)),
 		pqarrow.DefaultWriterProps())
@@ -679,6 +693,46 @@
 	}, nil
 }
 
+var _ RecordWriter = (*multiFieldRecordWriter)(nil)
+
+type multiFieldRecordWriter struct {
+	fw       *pqarrow.FileWriter
+	fieldIds []FieldID
+	schema   *arrow.Schema
+
+	numRows int
+}
+
+func (mfw *multiFieldRecordWriter) Write(r Record) error {
+	mfw.numRows += r.Len()
+	columns := make([]arrow.Array, len(mfw.fieldIds))
+	for i, fieldId := range mfw.fieldIds {
+		columns[i] = r.Column(fieldId)
+	}
+	rec := array.NewRecord(mfw.schema, columns, int64(r.Len()))
+	defer rec.Release()
+	return mfw.fw.WriteBuffered(rec)
+}
+
+func (mfw *multiFieldRecordWriter) Close() {
+	mfw.fw.Close()
+}
+
+func newMultiFieldRecordWriter(fieldIds []FieldID, fields []arrow.Field, writer io.Writer) (*multiFieldRecordWriter, error) {
+	schema := arrow.NewSchema(fields, nil)
+	fw, err := pqarrow.NewFileWriter(schema, writer,
+		parquet.NewWriterProperties(parquet.WithMaxRowGroupLength(math.MaxInt64)), // No additional grouping for now.
+		pqarrow.DefaultWriterProps())
+	if err != nil {
+		return nil, err
+	}
+	return &multiFieldRecordWriter{
+		fw:       fw,
+		fieldIds: fieldIds,
+		schema:   schema,
+	}, nil
+}
+
 type SerializeWriter[T any] struct {
 	rw         RecordWriter
 	serializer Serializer[T]
@@ -773,6 +827,10 @@
 	sr.r.Release()
 }
 
+func (sr *simpleArrowRecord) ArrowSchema() *arrow.Schema {
+	return sr.r.Schema()
+}
+
 func newSimpleArrowRecord(r arrow.Record, schema map[FieldID]schemapb.DataType, field2Col map[FieldID]int) *simpleArrowRecord {
 	return &simpleArrowRecord{
 		r:         r,