Repository: segmentio/parquet-go
Pull request #328 (2022-08-31 18:40:51)

Close pages in the doc example.
I had to look into the code if pages needs to be close. I think adding it to the doc makes it clear.


Patch:
diff --git a/README.md b/README.md
index 9113d35..ad1b53b 100644
--- a/README.md
+++ b/README.md
@@ -402,6 +402,7 @@ mechanisms to read column values:
 
 ```go
 pages := column.Pages()
+defer checkErr(pages.Close())
 
 for {
     p, err := pages.ReadPage()
@@ -410,11 +411,11 @@ for {
     }
 
     switch page := p.Values().(type) {
-    case parquet.Int32Page:
+    case parquet.Int32Reader:
         values := make([]int32, page.NumValues())
         _, err := page.ReadInt32s(values)
         ...
-    case parquet.Int64Page:
+    case parquet.Int64Reader:
         values := make([]int64, page.NumValues())
         _, err := page.ReadInt64s(values)
         ...
diff --git a/buffer.go b/buffer.go
index 0d46e11..ad43e02 100644
--- a/buffer.go
+++ b/buffer.go
@@ -334,60 +334,6 @@ func (b *buffer) clone() (clone *buffer) {
 	return clone
 }
 
-type bufferRef struct {
-	buf *buffer
-	off int
-	len int
-}
-
-func makeBufferRef(buf *buffer) (ref bufferRef) {
-	if buf != nil {
-		buf.ref()
-		ref.buf = buf
-		ref.len = len(buf.data)
-	}
-	return ref
-}
-
-func (r *bufferRef) data() []byte {
-	if r.buf != nil {
-		i := r.off
-		j := r.off + r.len
-		return r.buf.data[i:j:j]
-	}
-	return nil
-}
-
-func (r *bufferRef) ref() bufferRef {
-	if r.buf != nil {
-		r.buf.ref()
-	}
-	return *r
-}
-
-func (r *bufferRef) unref() {
-	if r.buf != nil {
-		r.buf.unref()
-		r.buf = nil
-	}
-}
-
-func (r *bufferRef) slice(i, j int) bufferRef {
-	if r.buf != nil {
-		r.buf.ref()
-	}
-	return bufferRef{buf: r.buf, off: r.off + i, len: j - i}
-}
-
-func (r *bufferRef) clone() (clone bufferRef) {
-	clone.off = r.off
-	clone.len = r.len
-	if r.buf != nil {
-		clone.buf = r.buf.clone()
-	}
-	return clone
-}
-
 type bufferPool struct {
 	pool sync.Pool
 }
@@ -420,21 +366,49 @@ var (
 
 type bufferedPage struct {
 	Page
-	values  bufferRef
-	offsets bufferRef
+	values           *buffer
+	offsets          *buffer
+	repetitionLevels *buffer
+	definitionLevels *buffer
 }
 
 func (p *bufferedPage) Slice(i, j int64) Page {
+	bufferRef(p.values)
+	bufferRef(p.offsets)
+	bufferRef(p.definitionLevels)
+	bufferRef(p.repetitionLevels)
+
 	return &bufferedPage{
-		Page:    p.Page.Slice(i, j),
-		values:  p.values.ref(),
-		offsets: p.offsets.ref(),
+		values:           p.values,
+		offsets:          p.offsets,
+		definitionLevels: p.definitionLevels,
+		repetitionLevels: p.repetitionLevels,
+		Page:             p.Page.Slice(i, j),
 	}
 }
 
 func unref(page Page) {
 	if p, _ := page.(*bufferedPage); p != nil {
-		p.values.unref()
-		p.offsets.unref()
+		bufferUnref(p.values)
+		bufferUnref(p.offsets)
+		bufferUnref(p.definitionLevels)
+		bufferUnref(p.repetitionLevels)
+
+		p.values = nil
+		p.offsets = nil
+		p.definitionLevels = nil
+		p.repetitionLevels = nil
+	}
+}
+
+func bufferRef(buf *buffer) {
+	if buf != nil {
+		buf.ref()
+	}
+}
+
+func bufferUnref(buf *buffer) {
+	if buf != nil {
+		buf.unref()
 	}
 }
diff --git a/column.go b/column.go
index 5f3f43a..8bb840a 100644
--- a/column.go
+++ b/column.go
@@ -646,21 +646,30 @@ func (c *Column) decodeDataPage(header DataPageHeader, numValues int, repetition
 			newPage,
 			c.maxRepetitionLevel,
 			c.maxDefinitionLevel,
-			makeBufferRef(repetitionLevels),
-			makeBufferRef(definitionLevels),
+			repetitionLevels.data,
+			definitionLevels.data,
 		)
 	case c.maxDefinitionLevel > 0:
 		newPage = newOptionalPage(
 			newPage,
 			c.maxDefinitionLevel,
-			makeBufferRef(definitionLevels),
+			definitionLevels.data,
 		)
 	}
+
+	bufferRef(vbuf)
+	bufferRef(obuf)
+	bufferRef(repetitionLevels)
+	bufferRef(definitionLevels)
+
 	newPage = &bufferedPage{
-		Page:    newPage,
-		values:  makeBufferRef(vbuf),
-		offsets: makeBufferRef(obuf),
+		Page:             newPage,
+		values:           vbuf,
+		offsets:          obuf,
+		repetitionLevels: repetitionLevels,
+		definitionLevels: definitionLevels,
 	}
+
 	return newPage, nil
 }
 
diff --git a/column_buffer.go b/column_buffer.go
index 2c68d88..d144b79 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -248,8 +248,7 @@ func (col *optionalColumnBuffer) Page() Page {
 		col.reordered = false
 	}
 
-	definitionLevels := makeBufferRef(&buffer{data: col.definitionLevels})
-	return newOptionalPage(col.base.Page(), col.maxDefinitionLevel, definitionLevels)
+	return newOptionalPage(col.base.Page(), col.maxDefinitionLevel, col.definitionLevels)
 }
 
 func (col *optionalColumnBuffer) Reset() {
@@ -551,14 +550,12 @@ func (col *repeatedColumnBuffer) Page() Page {
 		col.reordered = false
 	}
 
-	repetitionLevels := makeBufferRef(&buffer{data: col.repetitionLevels})
-	definitionLevels := makeBufferRef(&buffer{data: col.definitionLevels})
 	return newRepeatedPage(
 		col.base.Page(),
 		col.maxRepetitionLevel,
 		col.maxDefinitionLevel,
-		repetitionLevels,
-		definitionLevels,
+		col.repetitionLevels,
+		col.definitionLevels,
 	)
 }
 
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/config.go b/config.go
index 25c8e09..6b48f8e 100644
--- a/config.go
+++ b/config.go
@@ -31,6 +31,7 @@ const (
 type FileConfig struct {
 	SkipPageIndex    bool
 	SkipBloomFilters bool
+	ReadBufferSize   int
 }
 
 // DefaultFileConfig returns a new FileConfig value initialized with the
@@ -39,6 +40,7 @@ func DefaultFileConfig() *FileConfig {
 	return &FileConfig{
 		SkipPageIndex:    DefaultSkipPageIndex,
 		SkipBloomFilters: DefaultSkipBloomFilters,
+		ReadBufferSize:   defaultReadBufferSize,
 	}
 }
 
@@ -313,6 +315,16 @@ func SkipBloomFilters(skip bool) FileOption {
 	return fileOption(func(config *FileConfig) { config.SkipBloomFilters = skip })
 }
 
+// ReadBufferSize is a file configuration option which controls the default
+// buffer sizes for reads made to the provided io.Reader. The default of 4096
+// is appropriate for disk based access but if your reader is backed by something
+// like network storage it can be advantageous to increase this value.
+//
+// Defaults to 4096.
+func ReadBufferSize(sz int) FileOption {
+	return fileOption(func(config *FileConfig) { config.ReadBufferSize = sz })
+}
+
 // PageBufferSize configures the size of column page buffers on parquet writers.
 //
 // Note that the page buffer size refers to the in-memory buffers where pages
diff --git a/file.go b/file.go
index e9802b3..78832f5 100644
--- a/file.go
+++ b/file.go
@@ -14,9 +14,8 @@ import (
 )
 
 const (
-	defaultDictBufferSize  = 8192
-	defaultReadBufferSize  = 4096
-	defaultLevelBufferSize = 1024
+	defaultDictBufferSize = 8192
+	defaultReadBufferSize = 4096
 )
 
 // File represents a parquet file. The layout of a Parquet file can be found
@@ -31,6 +30,7 @@ type File struct {
 	columnIndexes []format.ColumnIndex
 	offsetIndexes []format.OffsetIndex
 	rowGroups     []RowGroup
+	config        *FileConfig
 }
 
 // OpenFile opens a parquet file and reads the content between offset 0 and the given
@@ -41,11 +41,11 @@ type File struct {
 // a file does not validate that the pages have valid checksums.
 func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 	b := make([]byte, 8)
-	f := &File{reader: r, size: size}
 	c, err := NewFileConfig(options...)
 	if err != nil {
 		return nil, err
 	}
+	f := &File{reader: r, size: size, config: c}
 
 	if _, err := r.ReadAt(b[:4], 0); err != nil {
 		return nil, fmt.Errorf("reading magic header of parquet file: %w", err)
@@ -452,9 +452,10 @@ func (c *fileColumnChunk) NumValues() int64 {
 }
 
 type filePages struct {
-	chunk   *fileColumnChunk
-	rbuf    *bufio.Reader
-	section io.SectionReader
+	chunk    *fileColumnChunk
+	rbuf     *bufio.Reader
+	rbufpool *sync.Pool
+	section  io.SectionReader
 
 	protocol thrift.CompactProtocol
 	decoder  thrift.Decoder
@@ -465,12 +466,15 @@ type filePages struct {
 	index      int
 	skip       int64
 	dictionary Dictionary
+
+	bufferSize int
 }
 
 func (f *filePages) init(c *fileColumnChunk) {
 	f.chunk = c
 	f.baseOffset = c.chunk.MetaData.DataPageOffset
 	f.dataOffset = f.baseOffset
+	f.bufferSize = c.file.config.ReadBufferSize
 
 	if c.chunk.MetaData.DictionaryPageOffset != 0 {
 		f.baseOffset = c.chunk.MetaData.DictionaryPageOffset
@@ -478,7 +482,7 @@ func (f *filePages) init(c *fileColumnChunk) {
 	}
 
 	f.section = *io.NewSectionReader(c.file, f.baseOffset, c.chunk.MetaData.TotalCompressedSize)
-	f.rbuf = getBufioReader(&f.section)
+	f.rbuf, f.rbufpool = getBufioReader(&f.section, f.bufferSize)
 	f.decoder.Reset(f.protocol.NewReader(f.rbuf))
 }
 
@@ -545,8 +549,8 @@ func (f *filePages) ReadPage() (Page, error) {
 
 func (f *filePages) readDictionary() error {
 	chunk := io.NewSectionReader(f.chunk.file, f.baseOffset, f.chunk.chunk.MetaData.TotalCompressedSize)
-	rbuf := getBufioReader(chunk)
-	defer putBufioReader(rbuf)
+	rbuf, pool := getBufioReader(chunk, f.bufferSize)
+	defer putBufioReader(rbuf, pool)
 
 	decoder := thrift.NewDecoder(f.protocol.NewReader(rbuf))
 	header := new(format.PageHeader)
@@ -680,7 +684,7 @@ func (f *filePages) SeekToRow(rowIndex int64) (err error) {
 }
 
 func (f *filePages) Close() error {
-	putBufioReader(f.rbuf)
+	putBufioReader(f.rbuf, f.rbufpool)
 	f.chunk = nil
 	f.section = io.SectionReader{}
 	f.rbuf = nil
@@ -697,23 +701,40 @@ func (f *filePages) columnPath() columnPath {
 	return columnPath(f.chunk.column.Path())
 }
 
+type putBufioReaderFunc func()
+
 var (
-	bufioReaderPool sync.Pool
+	bufioReaderPoolLock sync.Mutex
+	bufioReaderPool     = map[int]*sync.Pool{}
 )
 
-func getBufioReader(r io.Reader) *bufio.Reader {
-	rbuf, _ := bufioReaderPool.Get().(*bufio.Reader)
+func getBufioReader(r io.Reader, bufferSize int) (*bufio.Reader, *sync.Pool) {
+	pool := getBufioReaderPool(bufferSize)
+	rbuf, _ := pool.Get().(*bufio.Reader)
 	if rbuf == nil {
-		rbuf = bufio.NewReaderSize(r, defaultReadBufferSize)
+		rbuf = bufio.NewReaderSize(r, bufferSize)
 	} else {
 		rbuf.Reset(r)
 	}
-	return rbuf
+	return rbuf, pool
 }
 
-func putBufioReader(rbuf *bufio.Reader) {
-	if rbuf != nil {
+func putBufioReader(rbuf *bufio.Reader, pool *sync.Pool) {
+	if rbuf != nil && pool != nil {
 		rbuf.Reset(nil)
-		bufioReaderPool.Put(rbuf)
+		pool.Put(rbuf)
 	}
 }
+
+func getBufioReaderPool(size int) *sync.Pool {
+	bufioReaderPoolLock.Lock()
+	defer bufioReaderPoolLock.Unlock()
+
+	if pool := bufioReaderPool[size]; pool != nil {
+		return pool
+	}
+
+	pool := &sync.Pool{}
+	bufioReaderPool[size] = pool
+	return pool
+}
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.go b/page.go
index 1852bc0..9c991ac 100644
--- a/page.go
+++ b/page.go
@@ -376,10 +376,10 @@ func errPageBoundsOutOfRange(i, j, n int64) error {
 type optionalPage struct {
 	base               Page
 	maxDefinitionLevel byte
-	definitionLevels   bufferRef
+	definitionLevels   []byte
 }
 
-func newOptionalPage(base Page, maxDefinitionLevel byte, definitionLevels bufferRef) *optionalPage {
+func newOptionalPage(base Page, maxDefinitionLevel byte, definitionLevels []byte) *optionalPage {
 	return &optionalPage{
 		base:               base,
 		maxDefinitionLevel: maxDefinitionLevel,
@@ -393,21 +393,21 @@ func (page *optionalPage) Column() int { return page.base.Column() }
 
 func (page *optionalPage) Dictionary() Dictionary { return page.base.Dictionary() }
 
-func (page *optionalPage) NumRows() int64 { return int64(page.definitionLevels.len) }
+func (page *optionalPage) NumRows() int64 { return int64(len(page.definitionLevels)) }
 
-func (page *optionalPage) NumValues() int64 { return int64(page.definitionLevels.len) }
+func (page *optionalPage) NumValues() int64 { return int64(len(page.definitionLevels)) }
 
 func (page *optionalPage) NumNulls() int64 {
-	return int64(countLevelsNotEqual(page.definitionLevels.data(), page.maxDefinitionLevel))
+	return int64(countLevelsNotEqual(page.definitionLevels, page.maxDefinitionLevel))
 }
 
 func (page *optionalPage) Bounds() (min, max Value, ok bool) { return page.base.Bounds() }
 
-func (page *optionalPage) Size() int64 { return page.base.Size() + int64(page.definitionLevels.len) }
+func (page *optionalPage) Size() int64 { return page.base.Size() + int64(len(page.definitionLevels)) }
 
 func (page *optionalPage) RepetitionLevels() []byte { return nil }
 
-func (page *optionalPage) DefinitionLevels() []byte { return page.definitionLevels.data() }
+func (page *optionalPage) DefinitionLevels() []byte { return page.definitionLevels }
 
 func (page *optionalPage) Data() encoding.Values { return page.base.Data() }
 
@@ -422,18 +422,19 @@ func (page *optionalPage) Clone() Page {
 	return newOptionalPage(
 		page.base.Clone(),
 		page.maxDefinitionLevel,
-		page.definitionLevels.clone(),
+		copyBytes(page.definitionLevels),
 	)
 }
 
 func (page *optionalPage) Slice(i, j int64) Page {
-	definitionLevels := page.definitionLevels.data()
-	numNulls1 := int64(countLevelsNotEqual(definitionLevels[:i], page.maxDefinitionLevel))
-	numNulls2 := int64(countLevelsNotEqual(definitionLevels[i:j], page.maxDefinitionLevel))
+	maxDefinitionLevel := page.maxDefinitionLevel
+	definitionLevels := page.definitionLevels
+	numNulls1 := int64(countLevelsNotEqual(definitionLevels[:i], maxDefinitionLevel))
+	numNulls2 := int64(countLevelsNotEqual(definitionLevels[i:j], maxDefinitionLevel))
 	return newOptionalPage(
 		page.base.Slice(i-numNulls1, j-(numNulls1+numNulls2)),
-		page.maxDefinitionLevel,
-		page.definitionLevels.slice(int(i), int(j)),
+		maxDefinitionLevel,
+		definitionLevels[i:j:j],
 	)
 }
 
@@ -441,11 +442,11 @@ type repeatedPage struct {
 	base               Page
 	maxRepetitionLevel byte
 	maxDefinitionLevel byte
-	definitionLevels   bufferRef
-	repetitionLevels   bufferRef
+	definitionLevels   []byte
+	repetitionLevels   []byte
 }
 
-func newRepeatedPage(base Page, maxRepetitionLevel, maxDefinitionLevel byte, repetitionLevels, definitionLevels bufferRef) *repeatedPage {
+func newRepeatedPage(base Page, maxRepetitionLevel, maxDefinitionLevel byte, repetitionLevels, definitionLevels []byte) *repeatedPage {
 	return &repeatedPage{
 		base:               base,
 		maxRepetitionLevel: maxRepetitionLevel,
@@ -461,27 +462,23 @@ func (page *repeatedPage) Column() int { return page.base.Column() }
 
 func (page *repeatedPage) Dictionary() Dictionary { return page.base.Dictionary() }
 
-func (page *repeatedPage) NumRows() int64 {
-	return int64(countLevelsEqual(page.repetitionLevels.data(), 0))
-}
+func (page *repeatedPage) NumRows() int64 { return int64(countLevelsEqual(page.repetitionLevels, 0)) }
 
-func (page *repeatedPage) NumValues() int64 {
-	return int64(page.definitionLevels.len)
-}
+func (page *repeatedPage) NumValues() int64 { return int64(len(page.definitionLevels)) }
 
 func (page *repeatedPage) NumNulls() int64 {
-	return int64(countLevelsNotEqual(page.definitionLevels.data(), page.maxDefinitionLevel))
+	return int64(countLevelsNotEqual(page.definitionLevels, page.maxDefinitionLevel))
 }
 
 func (page *repeatedPage) Bounds() (min, max Value, ok bool) { return page.base.Bounds() }
 
 func (page *repeatedPage) Size() int64 {
-	return int64(page.repetitionLevels.len) + int64(page.definitionLevels.len) + page.base.Size()
+	return int64(len(page.repetitionLevels)) + int64(len(page.definitionLevels)) + page.base.Size()
 }
 
-func (page *repeatedPage) RepetitionLevels() []byte { return page.repetitionLevels.data() }
+func (page *repeatedPage) RepetitionLevels() []byte { return page.repetitionLevels }
 
-func (page *repeatedPage) DefinitionLevels() []byte { return page.definitionLevels.data() }
+func (page *repeatedPage) DefinitionLevels() []byte { return page.definitionLevels }
 
 func (page *repeatedPage) Data() encoding.Values { return page.base.Data() }
 
@@ -497,8 +494,8 @@ func (page *repeatedPage) Clone() Page {
 		page.base.Clone(),
 		page.maxRepetitionLevel,
 		page.maxDefinitionLevel,
-		page.repetitionLevels.clone(),
-		page.definitionLevels.clone(),
+		copyBytes(page.repetitionLevels),
+		copyBytes(page.definitionLevels),
 	)
 }
 
@@ -514,7 +511,11 @@ func (page *repeatedPage) Slice(i, j int64) Page {
 		panic(errPageBoundsOutOfRange(i, j, numRows))
 	}
 
-	repetitionLevels := page.repetitionLevels.data()
+	maxRepetitionLevel := page.maxRepetitionLevel
+	maxDefinitionLevel := page.maxDefinitionLevel
+	repetitionLevels := page.repetitionLevels
+	definitionLevels := page.definitionLevels
+
 	rowIndex0 := 0
 	rowIndex1 := len(repetitionLevels)
 	rowIndex2 := len(repetitionLevels)
@@ -539,19 +540,18 @@ func (page *repeatedPage) Slice(i, j int64) Page {
 		}
 	}
 
-	definitionLevels := page.definitionLevels.data()
-	numNulls1 := countLevelsNotEqual(definitionLevels[:rowIndex1], page.maxDefinitionLevel)
-	numNulls2 := countLevelsNotEqual(definitionLevels[rowIndex1:rowIndex2], page.maxDefinitionLevel)
+	numNulls1 := countLevelsNotEqual(definitionLevels[:rowIndex1], maxDefinitionLevel)
+	numNulls2 := countLevelsNotEqual(definitionLevels[rowIndex1:rowIndex2], maxDefinitionLevel)
 
 	i = int64(rowIndex1 - numNulls1)
 	j = int64(rowIndex2 - (numNulls1 + numNulls2))
 
 	return newRepeatedPage(
 		page.base.Slice(i, j),
-		page.maxRepetitionLevel,
-		page.maxDefinitionLevel,
-		page.repetitionLevels.slice(int(rowIndex1), int(rowIndex2)),
-		page.definitionLevels.slice(int(rowIndex1), int(rowIndex2)),
+		maxRepetitionLevel,
+		maxDefinitionLevel,
+		repetitionLevels[rowIndex1:rowIndex2:rowIndex2],
+		definitionLevels[rowIndex1:rowIndex2:rowIndex2],
 	)
 }
 
diff --git a/page_values.go b/page_values.go
index 719af93..85bca61 100644
--- a/page_values.go
+++ b/page_values.go
@@ -16,7 +16,7 @@ type optionalPageValues struct {
 
 func (r *optionalPageValues) ReadValues(values []Value) (n int, err error) {
 	maxDefinitionLevel := r.page.maxDefinitionLevel
-	definitionLevels := r.page.definitionLevels.data()
+	definitionLevels := r.page.definitionLevels
 	columnIndex := ^int16(r.page.Column())
 
 	for n < len(values) && r.offset < len(definitionLevels) {
@@ -64,8 +64,8 @@ type repeatedPageValues struct {
 
 func (r *repeatedPageValues) ReadValues(values []Value) (n int, err error) {
 	maxDefinitionLevel := r.page.maxDefinitionLevel
-	definitionLevels := r.page.definitionLevels.data()
-	repetitionLevels := r.page.repetitionLevels.data()
+	definitionLevels := r.page.definitionLevels
+	repetitionLevels := r.page.repetitionLevels
 	columnIndex := ^int16(r.page.Column())
 
 	for n < len(values) && r.offset < len(definitionLevels) {
@@ -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)
+	}
+}