Repository: segmentio/parquet-go
Pull request #329 (2022-08-31 18:58:54)

don't copy byte-array values
This PR is a follow up to https://github.com/segmentio/parquet-go/pull/265 which addresses the performance regression that I had introduced to address the original bug.

I left documentation in the code, at a high level the fix is to never read more than one page per column in each call to `ReadRows`, so we don't invalidate pointers written to the output `[]Row` values.

Contributes to https://github.com/segmentio/parquet-go/issues/226
```
name                                                 old time/op  new time/op  delta
MergeFiles/BYTE_ARRAY/groups=2,rows=20000            51.4µs ± 4%  11.8µs ± 2%   -77.04%  (p=0.000 n=10+10)
MergeFiles/FIXED_LEN_BYTE_ARRAY/groups=2,rows=20000  32.9µs ± 2%   7.6µs ± 1%   -76.92%  (p=0.000 n=10+10)

name                                                 old row/s    new row/s    delta
MergeFiles/BYTE_ARRAY/groups=2,rows=20000             18.8M ± 4%   82.0M ± 2%  +335.29%  (p=0.000 n=10+10)
MergeFiles/FIXED_LEN_BYTE_ARRAY/groups=2,rows=20000   29.4M ± 2%  127.6M ± 1%  +333.30%  (p=0.000 n=10+10)
```


Patch:
diff --git a/column_chunk.go b/column_chunk.go
index 17567d2..6d45b79 100644
--- a/column_chunk.go
+++ b/column_chunk.go
@@ -1,7 +1,6 @@
 package parquet
 
 import (
-	"errors"
 	"io"
 )
 
@@ -40,169 +39,7 @@ type pageAndValueWriter interface {
 	ValueWriter
 }
 
-type columnChunkReader struct {
-	// These two fields must be configured to initialize the reader.
-	buffer []Value     // buffer holding values read from the pages
-	offset int         // offset of the next value in the buffer
-	reader Pages       // reader of column pages
-	values ValueReader // reader for values from the current page
-	page   Page        // current page
-}
-
-func (r *columnChunkReader) buffered() int {
-	return len(r.buffer) - r.offset
-}
-
-func (r *columnChunkReader) reset() {
-	clearValues(r.buffer)
-	r.buffer = r.buffer[:0]
-	r.offset = 0
-	r.values = nil
-	unref(r.page)
-	r.page = nil
-}
-
-func (r *columnChunkReader) close() (err error) {
-	r.reset()
-	return r.reader.Close()
-}
-
-func (r *columnChunkReader) seekToRow(rowIndex int64) error {
-	// TODO: there are a few optimizations we can make here:
-	// * is the row buffered already? => advance the offset
-	// * is the row in the current page? => seek in values
-	r.reset()
-	return r.reader.SeekToRow(rowIndex)
-}
-
-func (r *columnChunkReader) readValues() error {
-	if r.offset < len(r.buffer) {
-		return nil
-	}
-	if r.values == nil {
-		for {
-			unref(r.page)
-			r.page = nil
-			p, err := r.reader.ReadPage()
-			if err != nil {
-				return err
-			}
-			r.page = p
-			if p.NumValues() > 0 {
-				r.values = p.Values()
-				break
-			}
-		}
-	}
-	n, err := r.values.ReadValues(r.buffer[:cap(r.buffer)])
-	if errors.Is(err, io.EOF) {
-		r.values = nil
-	}
-	if n > 0 {
-		err = nil
-	}
-	r.buffer = r.buffer[:n]
-	r.offset = 0
-	return err
-}
-
-/*
-func (r *columnChunkReader) writeBufferedRowsTo(w pageAndValueWriter, rowCount int64) (numRows int64, err error) {
-	if rowCount == 0 {
-		return 0, nil
-	}
-
-	for {
-		for r.offset < len(r.buffer) {
-			values := r.buffer[r.offset:]
-			// We can only determine that the full row has been consumed if we
-			// have more values in the buffer, and the next value is the start
-			// of a new row. Otherwise, we have to load more values from the
-			// page, which may yield EOF if all values have been consumed, in
-			// which case we know that we have read the full row, and otherwise
-			// we will enter this check again on the next loop iteration.
-			if numRows == rowCount {
-				if values[0].repetitionLevel == 0 {
-					return numRows, nil
-				}
-				values, _ = splitRowValues(values)
-			} else {
-				values = limitRowValues(values, int(rowCount-numRows))
-			}
-
-			n, err := w.WriteValues(values)
-			numRows += int64(countRowsOf(values[:n]))
-			r.offset += n
-			if err != nil {
-				return numRows, err
-			}
-		}
-
-		if err := r.readValuesFromCurrentPage(); err != nil {
-			if err == io.EOF {
-				err = nil
-			}
-			return numRows, err
-		}
-	}
-}
-
-func (r *columnChunkReader) writeRowsTo(w pageAndValueWriter, limit int64) (numRows int64, err error) {
-	for numRows < limit {
-		if r.values != nil {
-			n, err := r.writeBufferedRowsTo(w, numRows-limit)
-			numRows += n
-			if err != nil || numRows == limit {
-				return numRows, err
-			}
-		}
-
-		r.buffer = r.buffer[:0]
-		r.offset = 0
-
-		for numRows < limit {
-			p, err := r.reader.ReadPage()
-			if err != nil {
-				return numRows, err
-			}
-
-			pageRows := int64(p.NumRows())
-			// When the page is fully contained in the remaining range of rows
-			// that we intend to copy, we can use an optimized page copy rather
-			// than writing rows one at a time.
-			//
-			// Data pages v1 do not expose the number of rows available, which
-			// means we cannot take the optimized page copy path in those cases.
-			if pageRows == 0 || int64(pageRows) > limit {
-				r.values = p.Values()
-				err := r.readValuesFromCurrentPage()
-				if err == nil {
-					// More values have been buffered, break out of the inner loop
-					// to go back to the beginning of the outer loop and write
-					// buffered values to the output.
-					break
-				}
-				if errors.Is(err, io.EOF) {
-					// The page contained no values? Unclear if this is valid but
-					// we can handle it by reading the next page.
-					r.values = nil
-					continue
-				}
-				return numRows, err
-			}
-
-			if _, err := w.WritePage(p); err != nil {
-				return numRows, err
-			}
-
-			numRows += pageRows
-		}
-	}
-	return numRows, nil
-}
-*/
-
-type readRowsFunc func([]Row, byte, []columnChunkReader) (int, error)
+type readRowsFunc func(*rowGroupRows, []Row, byte) (int, error)
 
 func readRowsFuncOf(node Node, columnIndex int, repetitionDepth byte) (int, readRowsFunc) {
 	var read readRowsFunc
@@ -226,7 +63,7 @@ func readRowsFuncOf(node Node, columnIndex int, repetitionDepth byte) (int, read
 
 //go:noinline
 func readRowsFuncOfRepeated(read readRowsFunc, repetitionDepth byte) readRowsFunc {
-	return func(rows []Row, repetitionLevel byte, columns []columnChunkReader) (int, error) {
+	return func(r *rowGroupRows, rows []Row, repetitionLevel byte) (int, error) {
 		for i := range rows {
 			// Repeated columns have variable number of values, we must process
 			// them one row at a time because we cannot predict how many values
@@ -235,7 +72,7 @@ func readRowsFuncOfRepeated(read readRowsFunc, repetitionDepth byte) readRowsFun
 
 			// The first pass looks for values marking the beginning of a row by
 			// having a repetition level equal to the current level.
-			n, err := read(row, repetitionLevel, columns)
+			n, err := read(r, row, repetitionLevel)
 			if err != nil {
 				// The error here may likely be io.EOF, the read function may
 				// also have successfully read a row, which is indicated by a
@@ -262,7 +99,7 @@ func readRowsFuncOfRepeated(read readRowsFunc, repetitionDepth byte) readRowsFun
 			// repeated values until we get the indication that we consumed
 			// them all (the read function returns zero and no errors).
 			for {
-				n, err := read(row, repetitionDepth, columns)
+				n, err := read(r, row, repetitionDepth)
 				if err != nil {
 					return i + 1, err
 				}
@@ -280,7 +117,7 @@ func readRowsFuncOfGroup(node Node, columnIndex int, repetitionDepth byte) (int,
 	fields := node.Fields()
 
 	if len(fields) == 0 {
-		return columnIndex, func(_ []Row, _ byte, _ []columnChunkReader) (int, error) {
+		return columnIndex, func(*rowGroupRows, []Row, byte) (int, error) {
 			return 0, io.EOF
 		}
 	}
@@ -298,10 +135,10 @@ func readRowsFuncOfGroup(node Node, columnIndex int, repetitionDepth byte) (int,
 		columnIndex, group[i] = readRowsFuncOf(fields[i], columnIndex, repetitionDepth)
 	}
 
-	return columnIndex, func(rows []Row, repetitionLevel byte, columns []columnChunkReader) (int, error) {
+	return columnIndex, func(r *rowGroupRows, rows []Row, repetitionLevel byte) (int, error) {
 		// When reading a group, we use the first column as an indicator of how
 		// may rows can be read during this call.
-		n, err := group[0](rows, repetitionLevel, columns)
+		n, err := group[0](r, rows, repetitionLevel)
 
 		if n > 0 {
 			// Read values for all rows that the group is able to consume.
@@ -310,8 +147,8 @@ func readRowsFuncOfGroup(node Node, columnIndex int, repetitionDepth byte) (int,
 			// be more to read in other columns, therefore we must always read
 			// all columns and cannot stop on the first error.
 			for _, read := range group[1:] {
-				_, err2 := read(rows[:n], repetitionLevel, columns)
-				if err2 != nil && !errors.Is(err2, io.EOF) {
+				_, err2 := read(r, rows[:n], repetitionLevel)
+				if err2 != nil && err2 != io.EOF {
 					return 0, err2
 				}
 			}
@@ -326,46 +163,53 @@ func readRowsFuncOfLeaf(columnIndex int, repetitionDepth byte) (int, readRowsFun
 	var read readRowsFunc
 
 	if repetitionDepth == 0 {
-		read = func(rows []Row, _ byte, columns []columnChunkReader) (int, error) {
+		read = func(r *rowGroupRows, rows []Row, _ byte) (int, error) {
 			// When the repetition depth is zero, we know that there is exactly
 			// one value per row for this column, and therefore we can consume
 			// as many values as there are rows to fill.
-			col := &columns[columnIndex]
-
-			for n := 0; n < len(rows); {
-				if col.offset < len(col.buffer) {
-					rows[n] = append(rows[n], col.buffer[col.offset])
-					n++
-					col.offset++
-					continue
-				}
-				if err := col.readValues(); err != nil {
-					return n, err
+			col := &r.columns[columnIndex]
+			buf := r.buffer(columnIndex)
+
+			for i := range rows {
+				if col.offset == col.length {
+					n, err := col.values.ReadValues(buf)
+					col.offset = 0
+					col.length = int32(n)
+					if n == 0 && err != nil {
+						return 0, err
+					}
 				}
+
+				rows[i] = append(rows[i], buf[col.offset])
+				col.offset++
 			}
 
 			return len(rows), nil
 		}
 	} else {
-		read = func(rows []Row, repetitionLevel byte, columns []columnChunkReader) (int, error) {
+		read = func(r *rowGroupRows, rows []Row, repetitionLevel byte) (int, error) {
 			// When the repetition depth is not zero, we know that we will be
 			// called with a single row as input. We attempt to read at most one
 			// value of a single row and return to the caller.
-			col := &columns[columnIndex]
-
-			for {
-				if col.offset < len(col.buffer) {
-					if col.buffer[col.offset].repetitionLevel != repetitionLevel {
-						return 0, nil
-					}
-					rows[0] = append(rows[0], col.buffer[col.offset])
-					col.offset++
-					return 1, nil
-				}
-				if err := col.readValues(); err != nil {
+			col := &r.columns[columnIndex]
+			buf := r.buffer(columnIndex)
+
+			if col.offset == col.length {
+				n, err := col.values.ReadValues(buf)
+				col.offset = 0
+				col.length = int32(n)
+				if n == 0 && err != nil {
 					return 0, err
 				}
 			}
+
+			if buf[col.offset].repetitionLevel != repetitionLevel {
+				return 0, nil
+			}
+
+			rows[0] = append(rows[0], buf[col.offset])
+			col.offset++
+			return 1, nil
 		}
 	}
 
diff --git a/internal/unsafecast/unsafecast_go18.go b/internal/unsafecast/unsafecast_go18.go
index fc31b68..a30eb6a 100644
--- a/internal/unsafecast/unsafecast_go18.go
+++ b/internal/unsafecast/unsafecast_go18.go
@@ -24,7 +24,7 @@ func AddressOf[T any](data []T) *T {
 
 // AddressOfBytes returns the address of the first byte in data.
 func AddressOfBytes(data []byte) *byte {
-	return AddressOf(data)
+	return *(**byte)(unsafe.Pointer(&data))
 }
 
 // AddressOfString returns the address of the first byte in data.
diff --git a/null.go b/null.go
index 1a0edf2..ab0f752 100644
--- a/null.go
+++ b/null.go
@@ -77,6 +77,9 @@ func nullIndexFuncOf(t reflect.Type) nullIndexFunc {
 	case reflect.Slice:
 		return nullIndexSlice
 
+	case reflect.Map:
+		return nullIndexPointer
+
 	case reflect.Array:
 		if t.Elem().Kind() == reflect.Uint8 {
 			switch size := t.Len(); size {
diff --git a/page_values.go b/page_values.go
index 719af93..c3b011e 100644
--- a/page_values.go
+++ b/page_values.go
@@ -334,8 +334,7 @@ func (r *byteArrayPageValues) readByteArrays(values []byte) (c, n int, err error
 func (r *byteArrayPageValues) ReadValues(values []Value) (n int, err error) {
 	numValues := r.page.len()
 	for n < len(values) && r.offset < numValues {
-		value := r.page.index(r.offset)
-		values[n] = r.page.makeValueBytes(copyBytes(value))
+		values[n] = r.page.makeValueBytes(r.page.index(r.offset))
 		r.offset++
 		n++
 	}
@@ -372,7 +371,7 @@ func (r *fixedLenByteArrayPageValues) ReadFixedLenByteArrays(values []byte) (n i
 
 func (r *fixedLenByteArrayPageValues) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.data) {
-		values[n] = r.page.makeValueBytes(copyBytes(r.page.data[r.offset : r.offset+r.page.size]))
+		values[n] = r.page.makeValueBytes(r.page.data[r.offset : r.offset+r.page.size])
 		r.offset += r.page.size
 		n++
 	}
diff --git a/row.go b/row.go
index def7723..ab95ca7 100644
--- a/row.go
+++ b/row.go
@@ -84,6 +84,10 @@ type RowReader interface {
 	// each row read from the reader. If the rows are not nil, the backing array
 	// of the slices will be used as an optimization to avoid re-allocating new
 	// arrays.
+	//
+	// The application is expected to handle the case where ReadRows returns
+	// less rows than requested and no error, by looking at the first returned
+	// value from ReadRows, which is the number of rows that were read.
 	ReadRows([]Row) (int, error)
 }
 
diff --git a/row_group.go b/row_group.go
index a351325..a026ad8 100644
--- a/row_group.go
+++ b/row_group.go
@@ -252,30 +252,42 @@ func NewRowGroupRowReader(rowGroup RowGroup) Rows {
 
 type rowGroupRows struct {
 	rowGroup RowGroup
-	columns  []columnChunkReader
-	seek     int64
+	buffers  []Value
+	readers  []asyncPages
+	columns  []columnChunkRows
 	inited   bool
 	closed   bool
 	done     chan<- struct{}
 }
 
+type columnChunkRows struct {
+	rows   int64
+	offset int32
+	length int32
+	page   Page
+	values ValueReader
+}
+
+const columnBufferSize = defaultValueBufferSize
+
+func (r *rowGroupRows) buffer(i int) []Value {
+	j := (i + 0) * columnBufferSize
+	k := (i + 1) * columnBufferSize
+	return r.buffers[j:k:k]
+}
+
 func (r *rowGroupRows) init() {
-	const columnBufferSize = defaultValueBufferSize
 	columns := r.rowGroup.ColumnChunks()
-	readers := make([]asyncPages, len(columns))
-	buffer := make([]Value, columnBufferSize*len(columns))
-	r.columns = make([]columnChunkReader, len(columns))
+
+	r.buffers = make([]Value, len(columns)*columnBufferSize)
+	r.readers = make([]asyncPages, len(columns))
+	r.columns = make([]columnChunkRows, len(columns))
 
 	done := make(chan struct{})
 	r.done = done
 
 	for i, column := range columns {
-		reader := &readers[i]
-		reader.init(column.Pages(), done)
-
-		r.columns[i].buffer = buffer[:0:columnBufferSize]
-		r.columns[i].reader = reader
-		buffer = buffer[columnBufferSize:]
+		r.readers[i].init(column.Pages(), done)
 	}
 
 	r.inited = true
@@ -285,14 +297,28 @@ func (r *rowGroupRows) init() {
 	runtime.SetFinalizer(r, func(r *rowGroupRows) { r.Close() })
 }
 
-func (r *rowGroupRows) Reset() {
+func (r *rowGroupRows) clear() {
 	for i := range r.columns {
+		unref(r.columns[i].page)
+	}
+
+	for i := range r.columns {
+		r.columns[i] = columnChunkRows{}
+	}
+
+	for i := range r.buffers {
+		r.buffers[i] = Value{}
+	}
+}
+
+func (r *rowGroupRows) Reset() {
+	for i := range r.readers {
 		// Ignore errors because we are resetting the reader, if the error
 		// persists we will see it on the next read, and otherwise we can
 		// read back from the beginning.
-		r.columns[i].seekToRow(0)
+		r.readers[i].SeekToRow(0)
 	}
-	r.seek = 0
+	r.clear()
 }
 
 func (r *rowGroupRows) Close() error {
@@ -303,112 +329,110 @@ func (r *rowGroupRows) Close() error {
 		r.done = nil
 	}
 
-	for i := range r.columns {
-		if err := r.columns[i].close(); err != nil {
+	for i := range r.readers {
+		if err := r.readers[i].Close(); err != nil {
 			lastErr = err
 		}
 	}
 
+	r.clear()
 	r.inited = true
 	r.closed = true
 	return lastErr
 }
 
-func (r *rowGroupRows) Schema() *Schema {
-	return r.rowGroup.Schema()
-}
-
 func (r *rowGroupRows) SeekToRow(rowIndex int64) error {
+	var lastErr error
+
 	if r.closed {
 		return io.ErrClosedPipe
 	}
 
-	for i := range r.columns {
-		if err := r.columns[i].seekToRow(rowIndex); err != nil {
-			return err
+	if !r.inited {
+		r.init()
+	}
+
+	for i := range r.readers {
+		if err := r.readers[i].SeekToRow(rowIndex); err != nil {
+			lastErr = err
 		}
 	}
 
-	r.seek = rowIndex
-	return nil
+	r.clear()
+	return lastErr
 }
 
 func (r *rowGroupRows) ReadRows(rows []Row) (int, error) {
+	if r.closed {
+		return 0, io.EOF
+	}
+
 	if !r.inited {
 		r.init()
-		if r.seek > 0 {
-			if err := r.SeekToRow(r.seek); err != nil {
-				return 0, err
+	}
+
+	// Limit the number of rows that we read to the smallest number of rows
+	// remaining in the current page of each column. This is necessary because
+	// the pointers exposed to the returned rows need to remain valid until the
+	// next call to ReadRows, SeekToRow, Reset, or Close. If we release one of
+	// the columns' page, the rows that were already read during the ReadRows
+	// call would be invalidated, and might reference memory locations that have
+	// been reused due to pooling of page buffers.
+	numRows := int64(len(rows))
+
+	for i := range r.columns {
+		c := &r.columns[i]
+		// When all rows of the current page of a column have been consumed we
+		// have to read the next page. This will effectively invalidate all
+		// pointers of values previously held in the page, which is valid if
+		// the application respects the RowReader interface and does not retain
+		// parquet values without cloning them first.
+		for c.rows == 0 {
+			var err error
+			clearValues(r.buffer(i))
+
+			c.offset = 0
+			c.length = 0
+			c.values = nil
+			unref(c.page)
+
+			c.page, err = r.readers[i].ReadPage()
+			if err != nil {
+				if err != io.EOF {
+					return 0, err
+				}
+				break
 			}
+
+			c.rows = c.page.NumRows()
+			c.values = c.page.Values()
 		}
-	}
 
-	if r.closed {
-		return 0, io.EOF
+		if c.rows < numRows {
+			numRows = c.rows
+		}
 	}
 
 	for i := range rows {
 		rows[i] = rows[i][:0]
 	}
 
-	return r.rowGroup.Schema().readRows(rows, 0, r.columns)
-}
-
-/*
-func (r *rowGroupRows) WriteRowsTo(w RowWriter) (int64, error) {
-	if r.rowGroup == nil {
-		return CopyRows(w, struct{ RowReaderWithSchema }{r})
-	}
-	defer func() { r.rowGroup, r.seek = nil, 0 }()
-	rowGroup := r.rowGroup
-	if r.seek > 0 {
-		columns := rowGroup.ColumnChunks()
-		seekRowGroup := &seekRowGroup{
-			base:    rowGroup,
-			seek:    r.seek,
-			columns: make([]ColumnChunk, len(columns)),
-		}
-		seekColumnChunks := make([]seekColumnChunk, len(columns))
-		for i := range seekColumnChunks {
-			seekColumnChunks[i].base = columns[i]
-			seekColumnChunks[i].seek = r.seek
-			seekRowGroup.columns[i] = &seekColumnChunks[i]
-		}
-		rowGroup = seekRowGroup
+	if numRows == 0 {
+		return 0, io.EOF
 	}
 
-	switch dst := w.(type) {
-	case RowGroupWriter:
-		return dst.WriteRowGroup(rowGroup)
+	n, err := r.Schema().readRows(r, rows[:numRows], 0)
 
-	case PageWriter:
-		for _, column := range rowGroup.ColumnChunks() {
-			_, err := copyPagesAndClose(dst, column.Pages())
-			if err != nil {
-				return 0, err
-			}
-		}
-		return rowGroup.NumRows(), nil
+	for i := range r.columns {
+		r.columns[i].rows -= int64(n)
 	}
 
-	return CopyRows(w, struct{ RowReaderWithSchema }{r})
+	return n, err
 }
 
-func (r *rowGroupRows) writeRowsTo(w pageAndValueWriter, limit int64) (numRows int64, err error) {
-	for i := range r.columns {
-		n, err := r.columns[i].writeRowsTo(w, limit)
-		if err != nil {
-			return numRows, err
-		}
-		if i == 0 {
-			numRows = n
-		} else if numRows != n {
-			return numRows, fmt.Errorf("column %d wrote %d rows but the previous column(s) wrote %d rows", i, n, numRows)
-		}
-	}
-	return numRows, nil
+func (r *rowGroupRows) Schema() *Schema {
+	return r.rowGroup.Schema()
 }
-*/
 
 type seekRowGroup struct {
 	base    RowGroup


Test patch:
diff --git a/page_test.go b/page_test.go
index b199314..980627e 100644
--- a/page_test.go
+++ b/page_test.go
@@ -437,8 +437,9 @@ func TestOptionalPagePreserveIndex(t *testing.T) {
 	defer rows.Close()
 
 	rowbuf := make([]parquet.Row, 2)
+
 	n, err := rows.ReadRows(rowbuf)
-	if err != io.EOF {
+	if err != nil && err != io.EOF {
 		t.Fatal("reading rows:", err)
 	}
 	if n != 1 {
@@ -447,6 +448,14 @@ func TestOptionalPagePreserveIndex(t *testing.T) {
 	if rowbuf[0][0].Column() != 0 {
 		t.Errorf("wrong index: got=%d want=%d", rowbuf[0][0].Column(), 0)
 	}
+
+	n, err = rows.ReadRows(rowbuf)
+	if err != io.EOF {
+		t.Fatal("reading EOF:", err)
+	}
+	if n != 0 {
+		t.Fatal("expected no more rows after EOF:", n)
+	}
 }
 
 func TestRepeatedPageTrailingNulls(t *testing.T) {
diff --git a/writer_go18_test.go b/writer_go18_test.go
index 69224fc..ca08275 100644
--- a/writer_go18_test.go
+++ b/writer_go18_test.go
@@ -206,3 +206,90 @@ func TestIssue279(t *testing.T) {
 		}
 	}
 }
+
+func TestIssue302(t *testing.T) {
+	tests := []struct {
+		name string
+		fn   func(t *testing.T)
+	}{
+		{
+			name: "SimpleMap",
+			fn: func(t *testing.T) {
+				type M map[string]int
+
+				type T struct {
+					M M `parquet:","`
+				}
+
+				b := new(bytes.Buffer)
+				_ = parquet.NewGenericWriter[T](b)
+
+			},
+		},
+
+		{
+			name: "MapWithValueTag",
+			fn: func(t *testing.T) {
+				type M map[string]int
+
+				type T struct {
+					M M `parquet:"," parquet-value:",zstd"`
+				}
+
+				b := new(bytes.Buffer)
+				_ = parquet.NewGenericWriter[T](b)
+
+			},
+		},
+
+		{
+			name: "MapWithOptionalTag",
+			fn: func(t *testing.T) {
+				type M map[string]int
+
+				type T struct {
+					M M `parquet:",optional"`
+				}
+
+				b := new(bytes.Buffer)
+				w := parquet.NewGenericWriter[T](b)
+				expect := []T{
+					T{
+						M: M{
+							"Holden": 1,
+							"Naomi":  2,
+						},
+					},
+					T{
+						M: nil,
+					},
+					T{
+						M: M{
+							"Naomi":  1,
+							"Holden": 2,
+						},
+					},
+				}
+				_, err := w.Write(expect)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if err = w.Close(); err != nil {
+					t.Fatal(err)
+				}
+
+				bufReader := bytes.NewReader(b.Bytes())
+				r := parquet.NewGenericReader[T](bufReader)
+				values := make([]T, 3)
+				_, err = r.Read(values)
+				if !reflect.DeepEqual(expect, values) {
+					t.Fatalf("values do not match.\n\texpect: %v\n\tactual: %v", expect, values)
+				}
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, test.fn)
+	}
+}