Repository: segmentio/parquet-go
Pull request #320 (2022-08-26 19:55:46)

Configurable read buffer
The default read buffer size is standard for disk access, but we are using this library with object storage backing the parquet.File. With object storage hundreds of tiny reads of 4096 is obviously a non-starter so we built a [simple buffer](https://github.com/grafana/tempo/blob/main/pkg/io/buffered.go). Unfortunately this buffer often overreads by quite a bit and for performance sensitive accesses we have found that simply increasing `defaultBufferSize` in this library beats our buffer in nearly every case.

This PR makes the `defaultBufferSize` configurable to allow those who are backing their `parquet.File`s with unconventional storage some control over how it is accessed. It also removes `defaultLevelBufferSize` because it was unused. No issue with re-adding if it's there for a reason I don't understand.

Finally, I was unable to write a convincing test for this configuration option. Any advice on how to proceed would be appreciated.



Patch:
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index 0afee02..3f06928 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -315,9 +315,17 @@ func writeRowsFuncOfStruct(t reflect.Type, schema *Schema, path columnPath) writ
 	}
 
 	return func(buffers []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
-		for _, column := range columns {
-			if err := column.writeRows(buffers, rows.Offset(column.offset), levels); err != nil {
-				return err
+		if rows.Len() == 0 {
+			for _, column := range columns {
+				if err := column.writeRows(buffers, rows, levels); err != nil {
+					return err
+				}
+			}
+		} else {
+			for _, column := range columns {
+				if err := column.writeRows(buffers, rows.Offset(column.offset), levels); err != nil {
+					return err
+				}
 			}
 		}
 		return 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/convert.go b/convert.go
index 8a4934b..7d40a8f 100644
--- a/convert.go
+++ b/convert.go
@@ -53,6 +53,7 @@ type conversion struct {
 	targetColumnKinds   []Kind
 	targetToSourceIndex []int16
 	sourceToTargetIndex []int16
+	convertFunc         convertFunc
 	schema              *Schema
 	buffers             sync.Pool
 }
@@ -82,10 +83,103 @@ func (c *conversion) putBuffer(b *conversionBuffer) {
 	c.buffers.Put(b)
 }
 
+// convertFunc takes a target and source row, then copies data
+// from the source to the target according to the heuristics determined
+// by the target schema given in makeConvertFunc
+type convertFunc func(Row, levels, *conversionBuffer) (Row, error)
+
+func (c *conversion) makeConvertFunc(node Node) (convert convertFunc) {
+	if !node.Leaf() {
+		_, convert = c.convertFuncOf(0, node)
+	}
+	return convert
+}
+
+func (c *conversion) convertFuncOf(tgtIdx int16, node Node) (int16, convertFunc) {
+	switch {
+	case node.Repeated():
+		return c.convertFuncOfRepeated(tgtIdx, node)
+	default:
+		return c.convertFuncOfRequired(tgtIdx, node)
+	}
+}
+
+func (c *conversion) convertFuncOfRequired(tgtIdx int16, node Node) (int16, convertFunc) {
+	switch {
+	case node.Leaf():
+		return c.convertFuncOfLeaf(tgtIdx, node)
+	default:
+		return c.convertFuncOfGroup(tgtIdx, node)
+	}
+}
+
+// convertFuncOfLeaf is the base case to our schema-tree traversal, doing the
+// actual copy of the value from the old Row (via conversionBuffer) to the new Row
+func (c *conversion) convertFuncOfLeaf(tgtIdx int16, node Node) (int16, convertFunc) {
+	return tgtIdx + 1, func(tgt Row, _ levels, src *conversionBuffer) (Row, error) {
+		value := Value{}
+		if tgtIdx >= 0 && len(src.columns[tgtIdx]) > 0 {
+			// Pop the top value and remove from the buffer.
+			value = src.columns[tgtIdx][0]
+			src.columns[tgtIdx] = src.columns[tgtIdx][1:]
+		}
+		value.kind = ^int8(c.targetColumnKinds[tgtIdx])
+		value.columnIndex = ^tgtIdx
+		tgt = append(tgt, value)
+		return tgt, nil
+	}
+}
+
+func (c *conversion) convertFuncOfGroup(tgtIdx int16, node Node) (int16, convertFunc) {
+	fields := node.Fields()
+	funcs := make([]convertFunc, len(fields))
+
+	for i, field := range fields {
+		tgtIdx, funcs[i] = c.convertFuncOf(tgtIdx, field)
+	}
+
+	return tgtIdx, func(tgt Row, levels levels, src *conversionBuffer) (Row, error) {
+		var err error
+		for i, convFunc := range funcs {
+			if tgt, err = convFunc(tgt, levels, src); err != nil {
+				err = fmt.Errorf("%s → %w", fields[i].Name(), err)
+				break
+			}
+		}
+		return tgt, err
+	}
+}
+
+func (c *conversion) convertFuncOfRepeated(tgtIdx int16, node Node) (int16, convertFunc) {
+	nextIdx, convFunc := c.convertFuncOf(tgtIdx, Required(node))
+	return nextIdx, func(tgt Row, levels levels, src *conversionBuffer) (Row, error) {
+		var err error
+
+		levels.repetitionDepth++
+
+		for _, elem := range src.columns[tgtIdx] {
+			if elem.repetitionLevel != levels.repetitionLevel {
+				break
+			}
+			tgt, err = convFunc(tgt, levels, src)
+			levels.repetitionLevel = levels.repetitionDepth
+		}
+
+		return tgt, err
+	}
+}
+
+// Convert here satisfies the Conversion interface, and does the actual work to
+// convert betweeen the source and target Rows.
 func (c *conversion) Convert(target, source Row) (Row, error) {
+	if c.convertFunc == nil {
+		c.convertFunc = c.makeConvertFunc(c.schema)
+	}
+
 	buffer := c.getBuffer()
 	defer c.putBuffer(buffer)
 
+	// Build conversion buffer
 	for _, value := range source {
 		sourceIndex := value.Column()
 		targetIndex := c.sourceToTargetIndex[sourceIndex]
@@ -96,17 +190,18 @@ func (c *conversion) Convert(target, source Row) (Row, error) {
 		}
 	}
 
+	// Fill empty columns
 	for i, values := range buffer.columns {
 		if len(values) == 0 {
-			values = append(values, Value{
+			buffer.columns[i] = append(buffer.columns[i], Value{
 				kind:        ^int8(c.targetColumnKinds[i]),
 				columnIndex: ^int16(i),
 			})
 		}
-		target = append(target, values...)
 	}
 
-	return target, nil
+	// Construct row from buffer
+	return c.convertFunc(target, levels{}, buffer)
 }
 
 func (c *conversion) Column(i int) int {
@@ -178,12 +273,13 @@ func Convert(to, from Node) (conv Conversion, err error) {
 		sourceToTargetIndex[i] = targetColumn.columnIndex
 	}
 
-	return &conversion{
+	c := &conversion{
 		targetColumnKinds:   targetColumnKinds,
 		targetToSourceIndex: targetToSourceIndex,
 		sourceToTargetIndex: sourceToTargetIndex,
 		schema:              schema,
-	}, nil
+	}
+	return c, nil
 }
 
 // ConvertRowGroup constructs a wrapper of the given row group which applies
diff --git a/dictionary.go b/dictionary.go
index 40a6802..a3aa1a5 100644
--- a/dictionary.go
+++ b/dictionary.go
@@ -1371,6 +1371,8 @@ func (col *indexedColumnBuffer) Clone() ColumnBuffer {
 	}
 }
 
+func (col *indexedColumnBuffer) Type() Type { return col.typ.Type }
+
 func (col *indexedColumnBuffer) ColumnIndex() ColumnIndex { return indexedColumnIndex{col} }
 
 func (col *indexedColumnBuffer) OffsetIndex() OffsetIndex { return indexedOffsetIndex{col} }
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/row.go b/row.go
index f87df25..def7723 100644
--- a/row.go
+++ b/row.go
@@ -167,7 +167,12 @@ func (r *forwardRowSeeker) SeekToRow(rowIndex int64) error {
 		r.seek = rowIndex
 		return nil
 	}
-	return fmt.Errorf("SeekToRow: %T does not implement parquet.RowSeeker: cannot seek backward from row %d to %d", r.rows, r.index, rowIndex)
+	return fmt.Errorf(
+		"SeekToRow: %T does not implement parquet.RowSeeker: cannot seek backward from row %d to %d",
+		r.rows,
+		r.index,
+		rowIndex,
+	)
 }
 
 // CopyRows copies rows from src to dst.
@@ -361,6 +366,10 @@ type levels struct {
 	definitionLevel byte
 }
 
+// deconstructFunc accepts a row, the current levels, the value to deserialize
+// the current column onto, and returns the row minus the deserialied value(s)
+// It recurses until it hits a leaf node, then deserializes that value
+// individually as the base case.
 type deconstructFunc func(Row, levels, reflect.Value) Row
 
 func deconstructFuncOf(columnIndex int16, node Node) (int16, deconstructFunc) {
@@ -521,14 +530,24 @@ func reconstructFuncOf(columnIndex int16, node Node) (int16, reconstructFunc) {
 
 //go:noinline
 func reconstructFuncOfOptional(columnIndex int16, node Node) (int16, reconstructFunc) {
+	// We convert the optional func to required so that we eventually reach the
+	// leaf base-case.  We're still using the heuristics of optional in the
+	// returned closure (see levels.definitionLevel++), but we don't actually do
+	// deserialization here, that happens in the leaf function, hence this line.
 	nextColumnIndex, reconstruct := reconstructFuncOf(columnIndex, Required(node))
 	rowLength := nextColumnIndex - columnIndex
+
 	return nextColumnIndex, func(value reflect.Value, levels levels, row Row) (Row, error) {
 		if !row.startsWith(columnIndex) {
 			return row, fmt.Errorf("row is missing optional column %d", columnIndex)
 		}
 		if len(row) < int(rowLength) {
-			return row, fmt.Errorf("expected optional column %d to have at least %d values but got %d", columnIndex, rowLength, len(row))
+			return row, fmt.Errorf(
+				"expected optional column %d to have at least %d values but got %d",
+				columnIndex,
+				rowLength,
+				len(row),
+			)
 		}
 
 		levels.definitionLevel++
@@ -587,7 +606,12 @@ func reconstructRepeated(columnIndex, rowLength int16, levels levels, row Row, d
 		return row, fmt.Errorf("row is missing repeated column %d: %+v", columnIndex, row)
 	}
 	if len(row) < int(rowLength) {
-		return row, fmt.Errorf("expected repeated column %d to have at least %d values but got %d", columnIndex, rowLength, len(row))
+		return row, fmt.Errorf(
+			"expected repeated column %d to have at least %d values but got %d",
+			columnIndex,
+			rowLength,
+			len(row),
+		)
 	}
 
 	levels.repetitionDepth++
@@ -599,6 +623,7 @@ func reconstructRepeated(columnIndex, rowLength int16, levels levels, row Row, d
 
 	var err error
 	for row.startsWith(columnIndex) && row[0].repetitionLevel == levels.repetitionLevel {
+		// TODO: check specific error here.
 		if row, err = do(levels, row); err != nil {
 			break
 		}
diff --git a/schema.go b/schema.go
index 0dcf7cb..894f044 100644
--- a/schema.go
+++ b/schema.go
@@ -259,11 +259,18 @@ func (s *Schema) Reconstruct(value interface{}, row Row) error {
 		v = v.Elem()
 	}
 	var err error
+
 	if s.reconstruct != nil {
 		row, err = s.reconstruct(v, levels{}, row)
 		if len(row) > 0 && err == nil {
-			err = fmt.Errorf("%d values remain unused after reconstructing go value of type %s from parquet row", len(row), v.Type())
+			err = fmt.Errorf(
+				"%d values remain unused after reconstructing go value of type %s from parquet row",
+				len(row),
+				v.Type(),
+			)
 		}
+	} else {
+		panic(fmt.Sprintf("Reconstruct called when undefined on schema: %v", s))
 	}
 	return err
 }
diff --git a/sparse/array.go b/sparse/array.go
index 401837a..9848589 100644
--- a/sparse/array.go
+++ b/sparse/array.go
@@ -55,6 +55,9 @@ func (a array) slice(i, j int) array {
 }
 
 func (a array) offset(off uintptr) array {
+	if a.ptr == nil {
+		panic("offset of nil array")
+	}
 	return array{
 		ptr: unsafe.Add(a.ptr, off),
 		len: a.len,


Test patch:
diff --git a/convert_test.go b/convert_test.go
index 1f8e6c2..d02041a 100644
--- a/convert_test.go
+++ b/convert_test.go
@@ -7,6 +7,49 @@ import (
 	"github.com/segmentio/parquet-go"
 )
 
+type AddressBook2 struct {
+	Owner             string    `parquet:"owner,zstd"`
+	OwnerPhoneNumbers []string  `parquet:"ownerPhoneNumbers,gzip"`
+	Contacts          []Contact `parquet:"contacts"`
+	Extra             string    `parquet:"extra"`
+}
+
+type AddressBook3 struct {
+	Owner    string     `parquet:"owner,zstd"`
+	Contacts []Contact2 `parquet:"contacts"`
+}
+
+type Contact2 struct {
+	Name         string   `parquet:"name"`
+	PhoneNumbers []string `parquet:"phoneNumbers,zstd"`
+	Addresses    []string `parquet:"addresses,zstd"`
+}
+
+type AddressBook4 struct {
+	Owner    string     `parquet:"owner,zstd"`
+	Contacts []Contact2 `parquet:"contacts"`
+	Extra    string     `parquet:"extra"`
+}
+
+type SimpleNumber struct {
+	Number *int64 `parquet:"number,optional"`
+}
+
+type SimpleContact struct {
+	Numbers []SimpleNumber `parquet:"numbers"`
+}
+
+type SimpleAddressBook struct {
+	Name    string
+	Contact SimpleContact
+}
+
+type SimpleAddressBook2 struct {
+	Name    string
+	Contact SimpleContact
+	Extra   string
+}
+
 var conversionTests = [...]struct {
 	scenario string
 	from     interface{}
@@ -14,7 +57,6 @@ var conversionTests = [...]struct {
 }{
 	{
 		scenario: "convert between rows which have the same schema",
-
 		from: AddressBook{
 			Owner: "Julien Le Dem",
 			OwnerPhoneNumbers: []string{
@@ -31,7 +73,6 @@ var conversionTests = [...]struct {
 				},
 			},
 		},
-
 		to: AddressBook{
 			Owner: "Julien Le Dem",
 			OwnerPhoneNumbers: []string{
@@ -97,6 +138,136 @@ var conversionTests = [...]struct {
 			Names []string
 		}{ID: 1, Names: []string{}},
 	},
+
+	{
+		scenario: "extra column on complex struct",
+		from: AddressBook{
+			Owner:             "Julien Le Dem",
+			OwnerPhoneNumbers: []string{},
+			Contacts: []Contact{
+				{
+					Name:        "Dmitriy Ryaboy",
+					PhoneNumber: "555 987 6543",
+				},
+				{
+					Name: "Chris Aniszczyk",
+				},
+			},
+		},
+		to: AddressBook2{
+			Owner:             "Julien Le Dem",
+			OwnerPhoneNumbers: []string{},
+			Contacts: []Contact{
+				{
+					Name:        "Dmitriy Ryaboy",
+					PhoneNumber: "555 987 6543",
+				},
+				{
+					Name: "Chris Aniszczyk",
+				},
+			},
+		},
+	},
+
+	{
+		scenario: "handle nested repeated elements during conversion",
+		from: AddressBook3{
+			Owner: "Julien Le Dem",
+			Contacts: []Contact2{
+				{
+					Name: "Dmitriy Ryaboy",
+					PhoneNumbers: []string{
+						"555 987 6543",
+						"555 123 4567",
+					},
+					Addresses: []string{},
+				},
+				{
+					Name: "Chris Aniszczyk",
+					PhoneNumbers: []string{
+						"555 345 8129",
+					},
+					Addresses: []string{
+						"42 Wallaby Way Sydney",
+						"1 White House Way",
+					},
+				},
+				{
+					Name: "Bob Ross",
+					PhoneNumbers: []string{
+						"555 198 3628",
+					},
+					Addresses: []string{
+						"::1",
+					},
+				},
+			},
+		},
+		to: AddressBook4{
+			Owner: "Julien Le Dem",
+			Contacts: []Contact2{
+				{
+					Name: "Dmitriy Ryaboy",
+					PhoneNumbers: []string{
+						"555 987 6543",
+						"555 123 4567",
+					},
+					Addresses: []string{},
+				},
+				{
+					Name: "Chris Aniszczyk",
+					PhoneNumbers: []string{
+						"555 345 8129",
+					},
+					Addresses: []string{
+						"42 Wallaby Way Sydney",
+						"1 White House Way",
+					},
+				},
+				{
+					Name: "Bob Ross",
+					PhoneNumbers: []string{
+						"555 198 3628",
+					},
+					Addresses: []string{
+						"::1",
+					},
+				},
+			},
+			Extra: "",
+		},
+	},
+
+	{
+		scenario: "handle nested repeated elements during conversion",
+		from: SimpleAddressBook{
+			Name: "New Contact",
+			Contact: SimpleContact{
+				Numbers: []SimpleNumber{
+					SimpleNumber{
+						Number: nil,
+					},
+					SimpleNumber{
+						Number: newInt64(1329),
+					},
+				},
+			},
+		},
+		to: SimpleAddressBook2{
+			Name: "New Contact",
+			Contact: SimpleContact{
+				Numbers: []SimpleNumber{
+					SimpleNumber{
+						Number: nil,
+					},
+					SimpleNumber{
+						Number: newInt64(1329),
+					},
+				},
+			},
+			Extra: "",
+		},
+	},
 }
 
 func TestConvert(t *testing.T) {
@@ -110,12 +281,19 @@ func TestConvert(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			row := from.Deconstruct(nil, test.from)
-			row, err = conv.Convert(nil, row)
+			oldRow := from.Deconstruct(nil, test.from)
+			row, err := conv.Convert(nil, oldRow)
 			if err != nil {
 				t.Fatal(err)
 			}
 
+			// Helpful debugging info
+			// newRow := to.Deconstruct(nil, test.to)
+			// fmt.Printf("conv: %+v\n", conv)
+			// fmt.Printf("old row: %+v\n", oldRow)
+			// fmt.Printf("new row (desired state): %+v\n", newRow)
+			// fmt.Printf("new row (converted from old): %+v\n", row)
+
 			value := reflect.New(reflect.TypeOf(test.to))
 			if err := to.Reconstruct(value.Interface(), row); err != nil {
 				t.Fatal(err)
@@ -129,4 +307,5 @@ func TestConvert(t *testing.T) {
 	}
 }
 
+func newInt64(i int64) *int64    { return &i }
 func newString(s string) *string { return &s }
diff --git a/dictionary_test.go b/dictionary_test.go
index ccf5b38..e95a49d 100644
--- a/dictionary_test.go
+++ b/dictionary_test.go
@@ -1,6 +1,7 @@
 package parquet_test
 
 import (
+	"bytes"
 	"fmt"
 	"math/rand"
 	"testing"
@@ -198,3 +199,76 @@ func BenchmarkDictionary(b *testing.B) {
 		})
 	}
 }
+
+func TestIssue312(t *testing.T) {
+	node := parquet.String()
+	node = parquet.Encoded(node, &parquet.RLEDictionary)
+	g := parquet.Group{}
+	g["mystring"] = node
+	schema := parquet.NewSchema("test", g)
+
+	rows := []parquet.Row{[]parquet.Value{parquet.ValueOf("hello").Level(0, 0, 0)}}
+
+	var storage bytes.Buffer
+
+	tests := []struct {
+		name        string
+		getRowGroup func(t *testing.T) parquet.RowGroup
+	}{
+		{
+			name: "Writer",
+			getRowGroup: func(t *testing.T) parquet.RowGroup {
+				t.Helper()
+
+				w := parquet.NewWriter(&storage, schema)
+				_, err := w.WriteRows(rows)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if err := w.Close(); err != nil {
+					t.Fatal(err)
+				}
+
+				r := bytes.NewReader(storage.Bytes())
+				f, err := parquet.OpenFile(r, int64(storage.Len()))
+				if err != nil {
+					t.Fatal(err)
+				}
+				return f.RowGroups()[0]
+			},
+		},
+		{
+			name: "Buffer",
+			getRowGroup: func(t *testing.T) parquet.RowGroup {
+				t.Helper()
+
+				b := parquet.NewBuffer(schema)
+				_, err := b.WriteRows(rows)
+				if err != nil {
+					t.Fatal(err)
+				}
+				return b
+			},
+		},
+	}
+
+	for _, testCase := range tests {
+		t.Run(testCase.name, func(t *testing.T) {
+			rowGroup := testCase.getRowGroup(t)
+
+			chunk := rowGroup.ColumnChunks()[0]
+			idx := chunk.ColumnIndex()
+			val := idx.MinValue(0)
+			columnType := chunk.Type()
+			values := columnType.NewValues(val.Bytes(), []uint32{0, uint32(len(val.Bytes()))})
+
+			// This test ensures that the dictionary type created by column
+			// chunks of parquet readers and buffers are the same. We want the
+			// column chunk type to be the actual value type, even when the
+			// schema uses a dictionary encoding.
+			//
+			// https://github.com/segmentio/parquet-go/issues/312
+			_ = columnType.NewDictionary(0, 1, values)
+		})
+	}
+}