Repository: segmentio/parquet-go
Pull request #485 (2023-03-16 19:27:34)

Row group statistics
See #410. This PR adds row group statistics, which will enable row-group level predicate pushdown. 

The tests mostly pass with the changes I added, only that the column encodings seem to have a nondeterministic order. I had to build `parquet-tools` from scratch, so I'm not sure if this was just me building the wrong version or something. 


Patch:
diff --git a/Makefile b/Makefile
index b58b15c..137660e 100644
--- a/Makefile
+++ b/Makefile
@@ -9,4 +9,4 @@ format:
 	differ gofmt -w .
 
 test:
-	go test -trimpath -race ./...
+	go test -v -trimpath -race -tags= ./...
diff --git a/column_buffer.go b/column_buffer.go
index a5538eb..35d909c 100644
--- a/column_buffer.go
+++ b/column_buffer.go
@@ -828,7 +828,7 @@ func (col *booleanColumnBuffer) WriteValues(values []Value) (int, error) {
 func (col *booleanColumnBuffer) writeValues(rows sparse.Array, _ columnLevels) {
 	numBytes := bitpack.ByteCount(uint(col.numValues) + uint(rows.Len()))
 	if cap(col.bits) < numBytes {
-		col.bits = append(make([]byte, 0, 2*cap(col.bits)), col.bits...)
+		col.bits = append(make([]byte, 0, max(numBytes, 2*cap(col.bits))), col.bits...)
 	}
 	col.bits = col.bits[:numBytes]
 	i := 0
diff --git a/column_buffer_go18.go b/column_buffer_go18.go
index 539392c..1d4385c 100644
--- a/column_buffer_go18.go
+++ b/column_buffer_go18.go
@@ -3,6 +3,7 @@
 package parquet
 
 import (
+	"encoding/json"
 	"math/bits"
 	"reflect"
 	"time"
@@ -28,6 +29,10 @@ type writeRowsFunc func(columns []ColumnBuffer, rows sparse.Array, levels column
 // parquet schema. The column path indicates the column that the function is
 // being generated for in the parquet schema.
 func writeRowsFuncOf(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
+	if leaf, exists := schema.Lookup(path...); exists && leaf.Node.Type().LogicalType() != nil && leaf.Node.Type().LogicalType().Json != nil {
+		return writeRowsFuncOfJSON(t, schema, path)
+	}
+
 	switch t {
 	case reflect.TypeOf(deprecated.Int96{}):
 		return writeRowsFuncOfRequired(t, schema, path)
@@ -397,6 +402,44 @@ func writeRowsFuncOfMap(t reflect.Type, schema *Schema, path columnPath) writeRo
 	}
 }
 
+func writeRowsFuncOfJSON(t reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
+	// If this is a string or a byte array write directly.
+	switch t.Kind() {
+	case reflect.String:
+		return writeRowsFuncOfRequired(t, schema, path)
+	case reflect.Slice:
+		if t.Elem().Kind() == reflect.Uint8 {
+			return writeRowsFuncOfRequired(t, schema, path)
+		}
+	}
+
+	// Otherwise handle with a json.Marshal
+	asStrT := reflect.TypeOf(string(""))
+	writer := writeRowsFuncOfRequired(asStrT, schema, path)
+
+	return func(columns []ColumnBuffer, rows sparse.Array, levels columnLevels) error {
+		if rows.Len() == 0 {
+			return writer(columns, rows, levels)
+		}
+		for i := 0; i < rows.Len(); i++ {
+			val := reflect.NewAt(t, rows.Index(i))
+			asI := val.Interface()
+
+			b, err := json.Marshal(asI)
+			if err != nil {
+				return err
+			}
+
+			asStr := string(b)
+			a := sparse.MakeStringArray([]string{asStr})
+			if err := writer(columns, a.UnsafeArray(), levels); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
 func writeRowsFuncOfTime(_ reflect.Type, schema *Schema, path columnPath) writeRowsFunc {
 	t := reflect.TypeOf(int64(0))
 	elemSize := uintptr(t.Size())
diff --git a/go.mod b/go.mod
index 9804d55..56ebdae 100644
--- a/go.mod
+++ b/go.mod
@@ -11,6 +11,7 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.9
 	github.com/segmentio/encoding v0.3.5
 	golang.org/x/sys v0.0.0-20211110154304-99a53858aa08
+	google.golang.org/protobuf v1.30.0
 )
 
 require github.com/mattn/go-runewidth v0.0.9 // indirect
diff --git a/go.sum b/go.sum
index 5109074..69e2f68 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,8 @@
 github.com/andybalholm/brotli v1.0.3 h1:fpcw+r1N1h0Poc1F/pHbW40cUm/lMEQslZtCkBQ0UnM=
 github.com/andybalholm/brotli v1.0.3/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
 github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hexops/gotextdiff v1.0.3 h1:gitA9+qJrrTCsiCl7+kh75nPqQt1cx4ZkudSTLoUqJM=
@@ -17,3 +20,8 @@ github.com/segmentio/encoding v0.3.5 h1:UZEiaZ55nlXGDL92scoVuw00RmiRCazIEmvPSbSv
 github.com/segmentio/encoding v0.3.5/go.mod h1:n0JeuIqEQrQoPDGsjo8UNd1iA0U8d8+oHAA4E3G3OxM=
 golang.org/x/sys v0.0.0-20211110154304-99a53858aa08 h1:WecRHqgE09JBkh/584XIE6PMz5KKE/vER4izNUi30AQ=
 golang.org/x/sys v0.0.0-20211110154304-99a53858aa08/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.30.0 h1:kPPoIgf3TsEvrm0PFe15JQ+570QVxYzEvvHqChK+cng=
+google.golang.org/protobuf v1.30.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
diff --git a/reader_go18.go b/reader_go18.go
index bbbc225..f355428 100644
--- a/reader_go18.go
+++ b/reader_go18.go
@@ -103,6 +103,13 @@ func (r *GenericReader[T]) Reset() {
 	r.base.Reset()
 }
 
+// Read reads the next rows from the reader into the given rows slice up to len(rows).
+//
+// The returned values are safe to reuse across Read calls and do not share
+// memory with the reader's underlying page buffers.
+//
+// The method returns the number of rows read and io.EOF when no more rows
+// can be read from the reader.
 func (r *GenericReader[T]) Read(rows []T) (int, error) {
 	return r.read(r, rows)
 }
@@ -127,24 +134,46 @@ func (r *GenericReader[T]) Close() error {
 	return r.base.Close()
 }
 
+// readRows reads the next rows from the reader into the given rows slice up to len(rows).
+//
+// The returned values are safe to reuse across readRows calls and do not share
+// memory with the reader's underlying page buffers.
+//
+// The method returns the number of rows read and io.EOF when no more rows
+// can be read from the reader.
 func (r *GenericReader[T]) readRows(rows []T) (int, error) {
-	if cap(r.base.rowbuf) < len(rows) {
-		r.base.rowbuf = make([]Row, len(rows))
+	nRequest := len(rows)
+	if cap(r.base.rowbuf) < nRequest {
+		r.base.rowbuf = make([]Row, nRequest)
 	} else {
-		r.base.rowbuf = r.base.rowbuf[:len(rows)]
+		r.base.rowbuf = r.base.rowbuf[:nRequest]
 	}
 
-	n, err := r.base.ReadRows(r.base.rowbuf)
-	if n > 0 {
-		schema := r.base.Schema()
-
-		for i, row := range r.base.rowbuf[:n] {
-			if err := schema.Reconstruct(&rows[i], row); err != nil {
-				return i, err
+	var n, nTotal int
+	var err error
+	for {
+		// ReadRows reads the minimum remaining rows in a column page across all columns
+		// of the underlying reader, unless the length of the slice passed to it is smaller.
+		// In that case, ReadRows will read the number of rows equal to the length of the
+		// given slice argument. We limit that length to never be more than requested
+		// because sequential reads can cross page boundaries.
+		n, err = r.base.ReadRows(r.base.rowbuf[:nRequest-nTotal])
+		if n > 0 {
+			schema := r.base.Schema()
+
+			for i, row := range r.base.rowbuf[:n] {
+				if err2 := schema.Reconstruct(&rows[nTotal+i], row); err2 != nil {
+					return nTotal + i, err2
+				}
 			}
 		}
+		nTotal += n
+		if n == 0 || nTotal == nRequest || err != nil {
+			break
+		}
 	}
-	return n, err
+
+	return nTotal, err
 }
 
 var (
diff --git a/schema.go b/schema.go
index a62e43f..d6d2845 100644
--- a/schema.go
+++ b/schema.go
@@ -548,14 +548,18 @@ func nodeOf(t reflect.Type, tag []string) Node {
 			}
 		}
 
-		n = Map(
-			makeNodeOf(t.Key(), t.Name(), []string{keyTag}),
-			makeNodeOf(t.Elem(), t.Name(), []string{valueTag}),
-		)
+		if strings.Contains(mapTag, "json") {
+			n = JSON()
+		} else {
+			n = Map(
+				makeNodeOf(t.Key(), t.Name(), []string{keyTag}),
+				makeNodeOf(t.Elem(), t.Name(), []string{valueTag}),
+			)
+		}
 
 		forEachTagOption([]string{mapTag}, func(option, args string) {
 			switch option {
-			case "":
+			case "", "json":
 				return
 			case "optional":
 				n = Optional(n)
diff --git a/type.go b/type.go
index a100c67..cfa8b53 100644
--- a/type.go
+++ b/type.go
@@ -2,6 +2,7 @@ package parquet
 
 import (
 	"bytes"
+	"encoding/json"
 	"fmt"
 	"math/bits"
 	"reflect"
@@ -1579,7 +1580,25 @@ func (t *jsonType) EstimateDecodeSize(numValues int, src []byte, enc encoding.En
 }
 
 func (t *jsonType) AssignValue(dst reflect.Value, src Value) error {
-	return byteArrayType{}.AssignValue(dst, src)
+	// Assign value using ByteArrayType for BC...
+	switch dst.Kind() {
+	case reflect.String:
+		return byteArrayType{}.AssignValue(dst, src)
+	case reflect.Slice:
+		if dst.Type().Elem().Kind() == reflect.Uint8 {
+			return byteArrayType{}.AssignValue(dst, src)
+		}
+	}
+
+	// Otherwise handle with json.Unmarshal
+	b := src.byteArray()
+	val := reflect.New(dst.Type()).Elem()
+	err := json.Unmarshal(b, val.Addr().Interface())
+	if err != nil {
+		return err
+	}
+	dst.Set(val)
+	return nil
 }
 
 func (t *jsonType) ConvertValue(val Value, typ Type) (Value, error) {
diff --git a/writer.go b/writer.go
index 863b339..3acda12 100644
--- a/writer.go
+++ b/writer.go
@@ -1309,9 +1309,27 @@ func (c *writerColumn) recordPageStats(headerSize int32, header *format.PageHead
 	if page != nil {
 		numNulls := page.NumNulls()
 		numValues := page.NumValues()
-		minValue, maxValue, _ := page.Bounds()
+		minValue, maxValue, pageHasBounds := page.Bounds()
 		c.columnIndex.IndexPage(numValues, numNulls, minValue, maxValue)
 		c.columnChunk.MetaData.NumValues += numValues
+		c.columnChunk.MetaData.Statistics.NullCount += numNulls
+
+		if pageHasBounds {
+			var existingMaxValue, existingMinValue Value
+
+			if c.columnChunk.MetaData.Statistics.MaxValue != nil && c.columnChunk.MetaData.Statistics.MinValue != nil {
+				existingMaxValue = c.columnType.Kind().Value(c.columnChunk.MetaData.Statistics.MaxValue)
+				existingMinValue = c.columnType.Kind().Value(c.columnChunk.MetaData.Statistics.MinValue)
+			}
+
+			if existingMaxValue.isNull() || c.columnType.Compare(maxValue, existingMaxValue) > 0 {
+				c.columnChunk.MetaData.Statistics.MaxValue = maxValue.Bytes()
+			}
+
+			if existingMinValue.isNull() || c.columnType.Compare(minValue, existingMinValue) < 0 {
+				c.columnChunk.MetaData.Statistics.MinValue = minValue.Bytes()
+			}
+		}
 
 		c.offsetIndex.PageLocations = append(c.offsetIndex.PageLocations, format.PageLocation{
 			Offset:             c.columnChunk.MetaData.TotalCompressedSize,


Test patch:
diff --git a/page_test.go b/page_test.go
index 03b6ce5..c3e55a6 100644
--- a/page_test.go
+++ b/page_test.go
@@ -29,13 +29,16 @@ func testPageBoolean(t *testing.T) {
 	t.Run("parquet", func(t *testing.T) {
 		testPage(t, schema, pageTest{
 			write: func(w parquet.ValueWriter) (interface{}, error) {
-				values := []bool{false, true}
+				values := make([]bool, 50_000)
+				for i := range values {
+					values[i] = i%2 == 0
+				}
 				n, err := w.(parquet.BooleanWriter).WriteBooleans(values)
 				return values[:n], err
 			},
 
 			read: func(r parquet.ValueReader) (interface{}, error) {
-				values := make([]bool, 2)
+				values := make([]bool, 50_000)
 				n, err := r.(parquet.BooleanReader).ReadBooleans(values)
 				return values[:n], err
 			},
diff --git a/parquet_go18_test.go b/parquet_go18_test.go
index cebf311..71b29f7 100644
--- a/parquet_go18_test.go
+++ b/parquet_go18_test.go
@@ -12,6 +12,7 @@ import (
 	"testing"
 
 	"github.com/segmentio/parquet-go"
+	"google.golang.org/protobuf/types/known/structpb"
 )
 
 func ExampleReadFile() {
@@ -326,6 +327,152 @@ func TestIssue377(t *testing.T) {
 	assertRowsEqual(t, rows, ods)
 }
 
+func TestIssue423(t *testing.T) {
+	type Inner struct {
+		Value string `parquet:","`
+	}
+	type Outer struct {
+		Label string  `parquet:","`
+		Inner Inner   `parquet:",json"`
+		Slice []Inner `parquet:",json"`
+		// This is the only tricky situation. Because we're delegating to json Marshaler/Unmarshaler
+		// We use the json tags for optionality.
+		Ptr *Inner `json:",omitempty" parquet:",json"`
+
+		// This tests BC behavior that slices of bytes and json strings still get written/read in a BC way.
+		String        string                     `parquet:",json"`
+		Bytes         []byte                     `parquet:",json"`
+		MapOfStructPb map[string]*structpb.Value `parquet:",json"`
+		StructPB      *structpb.Value            `parquet:",json"`
+	}
+
+	writeRows := []Outer{
+		{
+			Label: "welp",
+			Inner: Inner{
+				Value: "this is a string",
+			},
+			Slice: []Inner{
+				{
+					Value: "in a slice",
+				},
+			},
+			Ptr:    nil,
+			String: `{"hello":"world"}`,
+			Bytes:  []byte(`{"goodbye":"world"}`),
+			MapOfStructPb: map[string]*structpb.Value{
+				"answer": structpb.NewNumberValue(42.00),
+			},
+			StructPB: structpb.NewBoolValue(true),
+		},
+		{
+			Label: "foxes",
+			Inner: Inner{
+				Value: "the quick brown fox jumped over the yellow lazy dog.",
+			},
+			Slice: []Inner{
+				{
+					Value: "in a slice",
+				},
+			},
+			Ptr: &Inner{
+				Value: "not nil",
+			},
+			String: `{"hello":"world"}`,
+			Bytes:  []byte(`{"goodbye":"world"}`),
+			MapOfStructPb: map[string]*structpb.Value{
+				"doubleAnswer": structpb.NewNumberValue(84.00),
+			},
+			StructPB: structpb.NewBoolValue(false),
+		},
+	}
+
+	schema := parquet.SchemaOf(new(Outer))
+	fmt.Println(schema.String())
+	buf := new(bytes.Buffer)
+	w := parquet.NewGenericWriter[Outer](buf, schema)
+	_, err := w.Write(writeRows)
+	if err != nil {
+		t.Fatal("write error: ", err)
+	}
+	w.Close()
+
+	file := bytes.NewReader(buf.Bytes())
+	readRows, err := parquet.Read[Outer](file, file.Size())
+	if err != nil {
+		t.Fatal("read error: ", err)
+	}
+
+	assertRowsEqual(t, writeRows, readRows)
+}
+
+func TestReadFileGenericMultipleRowGroupsMultiplePages(t *testing.T) {
+	type MyRow struct {
+		ID    [16]byte `parquet:"id,delta,uuid"`
+		File  string   `parquet:"file,dict,zstd"`
+		Index int64    `parquet:"index,delta,zstd"`
+	}
+
+	numRows := 20_000
+	maxPageBytes := 5000
+
+	tmp, err := os.CreateTemp("/tmp", "*.parquet")
+	if err != nil {
+		t.Fatal("os.CreateTemp: ", err)
+	}
+	path := tmp.Name()
+	defer os.Remove(path)
+	t.Log("file:", path)
+
+	// The page buffer size ensures we get multiple pages out of this example.
+	w := parquet.NewGenericWriter[MyRow](tmp, parquet.PageBufferSize(maxPageBytes))
+	// Need to write 1 row at a time here as writing many at once disregards PageBufferSize option.
+	for i := 0; i < numRows; i++ {
+		row := MyRow{
+			ID:    [16]byte{15: byte(i)},
+			File:  "hi" + fmt.Sprint(i),
+			Index: int64(i),
+		}
+		_, err := w.Write([]MyRow{row})
+		if err != nil {
+			t.Fatal("w.Write: ", err)
+		}
+		// Flush writes rows as row group. 4 total (20k/5k) in this file.
+		if (i+1)%maxPageBytes == 0 {
+			err = w.Flush()
+			if err != nil {
+				t.Fatal("w.Flush: ", err)
+			}
+		}
+	}
+	err = w.Close()
+	if err != nil {
+		t.Fatal("w.Close: ", err)
+	}
+	err = tmp.Close()
+	if err != nil {
+		t.Fatal("tmp.Close: ", err)
+	}
+
+	rows, err := parquet.ReadFile[MyRow](path)
+	if err != nil {
+		t.Fatal("parquet.ReadFile: ", err)
+	}
+
+	if len(rows) != numRows {
+		t.Fatalf("not enough values were read: want=%d got=%d", len(rows), numRows)
+	}
+	for i, row := range rows {
+		id := [16]byte{15: byte(i)}
+		file := "hi" + fmt.Sprint(i)
+		index := int64(i)
+
+		if row.ID != id || row.File != file || row.Index != index {
+			t.Fatalf("rows mismatch at index: %d got: %+v", i, row)
+		}
+	}
+}
+
 func assertRowsEqual[T any](t *testing.T, rows1, rows2 []T) {
 	if !reflect.DeepEqual(rows1, rows2) {
 		t.Error("rows mismatch")
diff --git a/parquet_test.go b/parquet_test.go
index 04d5001..5147b06 100644
--- a/parquet_test.go
+++ b/parquet_test.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+
 	"github.com/segmentio/parquet-go"
 	"github.com/segmentio/parquet-go/deprecated"
 	"github.com/segmentio/parquet-go/internal/quick"
@@ -490,6 +491,7 @@ var quickCheckConfig = quick.Config{
 	Sizes: []int{
 		0, 1, 2, 3, 4, 5, 6, 7, 8, 9,
 		10, 20, 30, 40, 50, 123,
+		4096 + 1,
 	},
 }
 
diff --git a/reader_go18_test.go b/reader_go18_test.go
index d44b4aa..7c406e4 100644
--- a/reader_go18_test.go
+++ b/reader_go18_test.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"math/rand"
+	"os"
 	"reflect"
 	"testing"
 
@@ -83,7 +84,7 @@ func testGenericReaderRows[Row any](rows []Row) error {
 		return fmt.Errorf("not enough values were read: want=%d got=%d", len(rows), n)
 	}
 	if !reflect.DeepEqual(rows, result) {
-		return fmt.Errorf("rows mismatch:\nwant: %+v\ngot:  %+v", rows, result)
+		return fmt.Errorf("rows mismatch:\nwant: %+v\ngot: %+v", rows, result)
 	}
 	return nil
 }
@@ -118,7 +119,7 @@ func TestIssue400(t *testing.T) {
 	}
 
 	r := parquet.NewGenericReader[A](bytes.NewReader(b.Bytes()))
-	values := make([]A, 2)
+	values := make([]A, 1)
 	_, err = r.Read(values)
 	if err != nil {
 		t.Fatal(err)
@@ -128,6 +129,101 @@ func TestIssue400(t *testing.T) {
 	}
 }
 
+func TestReadMinPageSize(t *testing.T) {
+	// NOTE: min page size is 307 for MyRow schema
+	t.Run("test read less than min page size", func(t *testing.T) { testReadMinPageSize(128, t) })
+	t.Run("test read equal to min page size", func(t *testing.T) { testReadMinPageSize(307, t) })
+	t.Run("test read more than min page size", func(t *testing.T) { testReadMinPageSize(384, t) })
+	// NOTE: num rows is 20,000
+	t.Run("test read equal to num rows", func(t *testing.T) { testReadMinPageSize(20_000, t) })
+	t.Run("test read more than num rows", func(t *testing.T) { testReadMinPageSize(25_000, t) })
+}
+
+func testReadMinPageSize(readSize int, t *testing.T) {
+	type MyRow struct {
+		ID    [16]byte `parquet:"id,delta,uuid"`
+		File  string   `parquet:"file,dict,zstd"`
+		Index int64    `parquet:"index,delta,zstd"`
+	}
+
+	numRows := 20_000
+	maxPageBytes := 5000
+
+	tmp, err := os.CreateTemp("/tmp", "*.parquet")
+	if err != nil {
+		t.Fatal("os.CreateTemp: ", err)
+	}
+	path := tmp.Name()
+	defer os.Remove(path)
+	t.Log("file:", path)
+
+	// The page buffer size ensures we get multiple pages out of this example.
+	w := parquet.NewGenericWriter[MyRow](tmp, parquet.PageBufferSize(maxPageBytes))
+	// Need to write 1 row at a time here as writing many at once disregards PageBufferSize option.
+	for i := 0; i < numRows; i++ {
+		row := MyRow{
+			ID:    [16]byte{15: byte(i)},
+			File:  "hi" + fmt.Sprint(i),
+			Index: int64(i),
+		}
+		_, err := w.Write([]MyRow{row})
+		if err != nil {
+			t.Fatal("w.Write: ", err)
+		}
+		// Flush writes rows as row group. 4 total (20k/5k) in this file.
+		if (i+1)%maxPageBytes == 0 {
+			err = w.Flush()
+			if err != nil {
+				t.Fatal("w.Flush: ", err)
+			}
+		}
+	}
+	err = w.Close()
+	if err != nil {
+		t.Fatal("w.Close: ", err)
+	}
+	err = tmp.Close()
+	if err != nil {
+		t.Fatal("tmp.Close: ", err)
+	}
+
+	file, err := os.Open(path)
+	if err != nil {
+		t.Fatal("os.Open", err)
+	}
+	reader := parquet.NewGenericReader[MyRow](file)
+	read := int64(0)
+	nRows := reader.NumRows()
+	rows := make([]MyRow, 0, nRows)
+	buf := make([]MyRow, readSize) // NOTE: min page size is 307 for MyRow schema
+
+	for read < nRows {
+		num, err := reader.Read(buf)
+		read += int64(num)
+		if err != nil && !errors.Is(err, io.EOF) {
+			t.Fatal("Read:", err)
+		}
+		rows = append(rows, buf...)
+	}
+
+	if err := reader.Close(); err != nil {
+		t.Fatal("Close", err)
+	}
+
+	if len(rows) < numRows {
+		t.Fatalf("not enough values were read: want=%d got=%d", len(rows), numRows)
+	}
+	for i, row := range rows[:numRows] {
+		id := [16]byte{15: byte(i)}
+		file := "hi" + fmt.Sprint(i)
+		index := int64(i)
+
+		if row.ID != id || row.File != file || row.Index != index {
+			t.Fatalf("rows mismatch at index: %d got: %+v", i, row)
+		}
+	}
+}
+
 func BenchmarkGenericReader(b *testing.B) {
 	benchmarkGenericReader[benchmarkRowType](b)
 	benchmarkGenericReader[booleanColumn](b)
diff --git a/writer_test.go b/writer_test.go
index 4b6b482..ebedd3e 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -12,6 +12,7 @@ import (
 	"github.com/hexops/gotextdiff"
 	"github.com/hexops/gotextdiff/myers"
 	"github.com/hexops/gotextdiff/span"
+
 	"github.com/segmentio/parquet-go"
 	"github.com/segmentio/parquet-go/compress"
 )
@@ -50,7 +51,7 @@ func generateParquetFile(rows rows, options ...parquet.WriterOption) ([]byte, er
 	defer tmp.Close()
 	path := tmp.Name()
 	defer os.Remove(path)
-	//fmt.Println(path)
+	// fmt.Println(path)
 
 	writerOptions := []parquet.WriterOption{parquet.PageBufferSize(20)}
 	writerOptions = append(writerOptions, options...)
@@ -101,8 +102,8 @@ var writerTests = []struct {
 		},
 		dump: `row group 0
 --------------------------------------------------------------------------------
-first_name:  BINARY ZSTD DO:4 FPO:55 SZ:90/72/0.80 VC:3 ENC:PLAIN,RLE_DICTIONARY ST:[no stats for this column]
-last_name:   BINARY ZSTD DO:0 FPO:94 SZ:127/121/0.95 VC:3 ENC:DELTA_BYTE_ARRAY ST:[no stats for this column]
+first_name:  BINARY ZSTD DO:4 FPO:55 SZ:90/72/0.80 VC:3 ENC:RLE_DICTIONARY,PLAIN ST:[min: Han, max: Luke, num_nulls not defined]
+last_name:   BINARY ZSTD DO:0 FPO:94 SZ:127/121/0.95 VC:3 ENC:DELTA_BYTE_ARRAY ST:[min: Skywalker, max: Solo, num_nulls not defined]
 
     first_name TV=3 RL=0 DL=0 DS: 3 DE:PLAIN
     ----------------------------------------------------------------------------
@@ -139,8 +140,8 @@ value 3: R:0 D:0 V:Skywalker
 		},
 		dump: `row group 0
 --------------------------------------------------------------------------------
-first_name:  BINARY ZSTD DO:4 FPO:55 SZ:86/77/0.90 VC:3 ENC:RLE_DICTIONARY,PLAIN ST:[no stats for this column]
-last_name:   BINARY ZSTD DO:0 FPO:90 SZ:137/131/0.96 VC:3 ENC:DELTA_BYTE_ARRAY ST:[no stats for this column]
+first_name:  BINARY ZSTD DO:4 FPO:55 SZ:86/77/0.90 VC:3 ENC:PLAIN,RLE_DICTIONARY ST:[min: Han, max: Luke, num_nulls not defined]
+last_name:   BINARY ZSTD DO:0 FPO:90 SZ:137/131/0.96 VC:3 ENC:DELTA_BYTE_ARRAY ST:[min: Skywalker, max: Solo, num_nulls not defined]
 
     first_name TV=3 RL=0 DL=0 DS: 3 DE:PLAIN
     ----------------------------------------------------------------------------
@@ -185,9 +186,9 @@ value 3: R:0 D:0 V:Skywalker
 		},
 		dump: `row group 0
 --------------------------------------------------------------------------------
-name:       BINARY GZIP DO:4 FPO:70 SZ:126/101/0.80 VC:10 ENC:PLAIN,RLE_DICTIONARY ST:[no stats for this column]
-timestamp:  INT64 GZIP DO:0 FPO:130 SZ:299/550/1.84 VC:10 ENC:DELTA_BINARY_PACKED ST:[no stats for this column]
-value:      DOUBLE GZIP DO:0 FPO:429 SZ:292/192/0.66 VC:10 ENC:PLAIN ST:[no stats for this column]
+name:       BINARY GZIP DO:4 FPO:70 SZ:126/101/0.80 VC:10 ENC:PLAIN,RLE_DICTIONARY ST:[min: http_request_total, max: http_request_total, num_nulls not defined]
+timestamp:  INT64 GZIP DO:0 FPO:130 SZ:299/550/1.84 VC:10 ENC:DELTA_BINARY_PACKED ST:[min: 1639444033, max: 1639444144, num_nulls not defined]
+value:      DOUBLE GZIP DO:0 FPO:429 SZ:292/192/0.66 VC:10 ENC:PLAIN ST:[min: -0.0, max: 100.0, num_nulls not defined]
 
     name TV=10 RL=0 DL=0 DS: 1 DE:PLAIN
     ----------------------------------------------------------------------------
@@ -280,11 +281,11 @@ value 10: R:0 D:0 V:10.0
 
 		dump: `row group 0
 --------------------------------------------------------------------------------
-owner:              BINARY ZSTD DO:0 FPO:4 SZ:81/73/0.90 VC:2 ENC:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
-ownerPhoneNumbers:  BINARY GZIP DO:0 FPO:85 SZ:179/129/0.72 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
+owner:              BINARY ZSTD DO:0 FPO:4 SZ:81/73/0.90 VC:2 ENC:DELTA_LENGTH_BYTE_ARRAY ST:[min: A. Nonymous, max: Julien Le Dem, num_nulls not defined]
+ownerPhoneNumbers:  BINARY GZIP DO:0 FPO:85 SZ:179/129/0.72 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: 555 123 4567, max: 555 666 1337, num_nulls: 1]
 contacts:
-.name:              BINARY UNCOMPRESSED DO:0 FPO:264 SZ:138/138/1.00 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
-.phoneNumber:       BINARY ZSTD DO:0 FPO:402 SZ:113/95/0.84 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
+.name:              BINARY UNCOMPRESSED DO:0 FPO:264 SZ:138/138/1.00 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: Chris Aniszczyk, max: Dmitriy Ryaboy, num_nulls: 1]
+.phoneNumber:       BINARY ZSTD DO:0 FPO:402 SZ:113/95/0.84 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: 555 987 6543, max: 555 987 6543, num_nulls: 2]
 
     owner TV=2 RL=0 DL=0
     ----------------------------------------------------------------------------
@@ -362,11 +363,11 @@ value 3: R:0 D:0 V:<null>
 
 		dump: `row group 0
 --------------------------------------------------------------------------------
-owner:              BINARY ZSTD DO:0 FPO:4 SZ:86/78/0.91 VC:2 ENC:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
-ownerPhoneNumbers:  BINARY GZIP DO:0 FPO:90 SZ:172/122/0.71 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
+owner:              BINARY ZSTD DO:0 FPO:4 SZ:86/78/0.91 VC:2 ENC:DELTA_LENGTH_BYTE_ARRAY ST:[min: A. Nonymous, max: Julien Le Dem, num_nulls not defined]
+ownerPhoneNumbers:  BINARY GZIP DO:0 FPO:90 SZ:172/122/0.71 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: 555 123 4567, max: 555 666 1337, num_nulls: 1]
 contacts:
-.name:              BINARY UNCOMPRESSED DO:0 FPO:262 SZ:132/132/1.00 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
-.phoneNumber:       BINARY ZSTD DO:0 FPO:394 SZ:108/90/0.83 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
+.name:              BINARY UNCOMPRESSED DO:0 FPO:262 SZ:132/132/1.00 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: Chris Aniszczyk, max: Dmitriy Ryaboy, num_nulls: 1]
+.phoneNumber:       BINARY ZSTD DO:0 FPO:394 SZ:108/90/0.83 VC:3 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[min: 555 987 6543, max: 555 987 6543, num_nulls: 2]
 
     owner TV=2 RL=0 DL=0
     ----------------------------------------------------------------------------
@@ -425,8 +426,8 @@ value 3: R:0 D:0 V:<null>
 		},
 		dump: `row group 0
 --------------------------------------------------------------------------------
-name:   BINARY UNCOMPRESSED DO:4 FPO:49 SZ:73/73/1.00 VC:2 ENC:PLAIN,RLE_DICTIONARY ST:[no stats for this column]
-value:  DOUBLE UNCOMPRESSED DO:0 FPO:77 SZ:39/39/1.00 VC:2 ENC:PLAIN ST:[no stats for this column]
+name:   BINARY UNCOMPRESSED DO:4 FPO:49 SZ:73/73/1.00 VC:2 ENC:RLE_DICTIONARY,PLAIN ST:[min: customer1, max: customer2, num_nulls not defined]
+value:  DOUBLE UNCOMPRESSED DO:0 FPO:77 SZ:39/39/1.00 VC:2 ENC:PLAIN ST:[min: 1.0, max: 42.0, num_nulls not defined]
 
     name TV=2 RL=0 DL=0 DS: 2 DE:PLAIN
     ----------------------------------------------------------------------------
@@ -453,7 +454,7 @@ value 2: R:0 D:0 V:1.0
 
 func TestWriter(t *testing.T) {
 	if !hasParquetTools() {
-		t.Skip("parquet-tools are not installed")
+		t.Skip("Skipping TestWriter writerTests because parquet-tools are not installed in Github CI. FIXME.") // TODO
 	}
 
 	for _, test := range writerTests {