Repository: segmentio/parquet-go (go)

Issue:
Reading of complex structs with []byte fields is incorrect
Hi, we are seeing an issue in our application where sometimes a row contains unexpected data when reading it back with `parquet.Read(<struct>)`.  [Our struct](https://github.com/grafana/tempo/blob/parquet/tempodb/encoding/vparquet/schema.go) contains child structs with several []byte, and it seems like the handling of these is incorrect near page boundaries.

Was able to create a minimal reproduction test below.   It writes 2 very small objects with 1 child byte slice.  When reading row 0 it contains the child from row 1.  An observation:  This only occurs when using `writer.Write(<struct>)` and seems ok when using `parquet.Buffer`.  Unsure if there is a fundamental difference or just needs different conditions.

```go
func TestRoundtripNestedRepeatedBytes(t *testing.T) {
	type B struct {
		C []byte
	}
	type A struct {
		A string
		B []B
	}

	var objs []A
	for i := 0; i < 2; i++ {
		o := A{
			"test" + strconv.Itoa(i),
			[]B{
				{[]byte{byte(i)}},
			},
		}
		objs = append(objs, o)
	}

	buf := new(bytes.Buffer)
	w := parquet.NewWriter(buf, parquet.PageBufferSize(100))
	for _, o := range objs {
		w.Write(&o)
	}
	w.Close()

	file := bytes.NewReader(buf.Bytes())

	r := parquet.NewReader(file)
	for i := 0; ; i++ {
		o := new(A)
		err := r.Read(o)
		if errors.Is(err, io.EOF) {
			if i < len(objs) {
				t.Errorf("too few rows were read: %d<%d", i, len(objs))
			}
			break
		}
		if !reflect.DeepEqual(*o, objs[i]) {
			t.Errorf("points mismatch at row index %d: want=%v got=%v", i, objs[i], o)
		}
	}
}
```

```
parquet-go $ go test -v -run=TestRoundtripNestedRepeatedBytes
=== RUN   TestRoundtripNestedRepeatedBytes
    buffer_test.go:526: points mismatch at row index 0: want={test0 [{[0]}]} got=&{test0 [{[1]}]}
--- FAIL: TestRoundtripNestedRepeatedBytes (0.00s)
```

Output from `parquet-tools dump <file>`
```
row group 0
--------------------------------------------------------------------------------
A:   BINARY UNCOMPRESSED DO:0 FPO:4 SZ:60/60/1.00 VC:2 ENC:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]
B:
.C:  BINARY UNCOMPRESSED DO:0 FPO:64 SZ:76/76/1.00 VC:2 ENC:RLE,DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column]

    A TV=2 RL=0 DL=0
    ----------------------------------------------------------------------------
    page 0:  DLE:RLE RLE:RLE VLE:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column] SZ:32 VC:2

    B.C TV=2 RL=1 DL=1
    ----------------------------------------------------------------------------
    page 0:  DLE:RLE RLE:RLE VLE:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column] SZ:10 VC:1
    page 1:  DLE:RLE RLE:RLE VLE:DELTA_LENGTH_BYTE_ARRAY ST:[no stats for this column] SZ:10 VC:1

BINARY A
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 2 ***
value 1: R:0 D:0 V:test0
value 2: R:0 D:0 V:test1

BINARY B.C
--------------------------------------------------------------------------------
*** row group 1 of 1, values 1 to 2 ***
value 1: R:0 D:1 V:0x00
value 2: R:0 D:1 V:0x01
```



Pull request:
fix issue 254
Fixes #254 

When a page boundary was crossed while loading column values, the page buffer being recycled via a sync.Pool was causing values loaded for other columns to be invalidated even tho the whole row had not yet been returned to the application.

This PR may introduce a performance regression on the read paths when rows contain values of type BYTE_ARRAY or FIXED_LEN_BYTE_ARRAY as we will now be making a copy of these instead of taking references to the underlying page buffer. This will be addressed in a follow up.


Patch:
diff --git a/file.go b/file.go
index 05d6dca..dc470d2 100644
--- a/file.go
+++ b/file.go
@@ -91,7 +91,7 @@ func OpenFile(r io.ReaderAt, size int64, options ...FileOption) (*File, error) {
 	}
 
 	schema := NewSchema(f.root.Name(), f.root)
-	columns := make([]*Column, 0, MaxColumnIndex+1)
+	columns := make([]*Column, 0, numLeafColumnsOf(f.root))
 	f.schema = schema
 	f.root.forEachLeaf(func(c *Column) { columns = append(columns, c) })
 
@@ -264,6 +264,9 @@ func (f *File) Root() *Column { return f.root }
 // Schema returns the schema of f.
 func (f *File) Schema() *Schema { return f.schema }
 
+// Metadata returns the metadata of f.
+func (f *File) Metadata() *format.FileMetaData { return &f.metadata }
+
 // Size returns the size of f (in bytes).
 func (f *File) Size() int64 { return f.size }
 
@@ -310,9 +313,7 @@ func (f *File) hasIndexes() bool {
 	return f.columnIndexes != nil && f.offsetIndexes != nil
 }
 
-var (
-	_ io.ReaderAt = (*File)(nil)
-)
+var _ io.ReaderAt = (*File)(nil)
 
 func sortKeyValueMetadata(keyValueMetadata []format.KeyValue) {
 	sort.Slice(keyValueMetadata, func(i, j int) bool {
diff --git a/page_values.go b/page_values.go
index c25f6da..65b6026 100644
--- a/page_values.go
+++ b/page_values.go
@@ -331,7 +331,7 @@ func (r *byteArrayPageValues) readByteArrays(values []byte) (c, n int, err error
 func (r *byteArrayPageValues) ReadValues(values []Value) (n int, err error) {
 	for n < len(values) && r.offset < len(r.page.values) {
 		value := r.page.valueAt(uint32(r.offset))
-		values[n] = r.page.makeValueBytes(value)
+		values[n] = r.page.makeValueBytes(copyBytes(value))
 		r.offset += plain.ByteArrayLengthSize
 		r.offset += len(value)
 		n++
@@ -369,7 +369,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(r.page.data[r.offset : r.offset+r.page.size])
+		values[n] = r.page.makeValueBytes(copyBytes(r.page.data[r.offset : r.offset+r.page.size]))
 		r.offset += r.page.size
 		n++
 	}


Test patch:
diff --git a/buffer_test.go b/buffer_test.go
index 924b278..8e674bf 100644
--- a/buffer_test.go
+++ b/buffer_test.go
@@ -522,6 +522,51 @@ func TestBufferRoundtripNestedRepeatedPointer(t *testing.T) {
 	}
 }
 
+func TestRoundtripNestedRepeatedBytes(t *testing.T) {
+	type B struct {
+		C []byte
+	}
+	type A struct {
+		A string
+		B []B
+	}
+
+	var objs []A
+	for i := 0; i < 2; i++ {
+		o := A{
+			"test" + strconv.Itoa(i),
+			[]B{
+				{[]byte{byte(i)}},
+			},
+		}
+		objs = append(objs, o)
+	}
+
+	buf := new(bytes.Buffer)
+	w := parquet.NewWriter(buf, parquet.PageBufferSize(100))
+	for _, o := range objs {
+		w.Write(&o)
+	}
+	w.Close()
+
+	file := bytes.NewReader(buf.Bytes())
+
+	r := parquet.NewReader(file)
+	for i := 0; ; i++ {
+		o := new(A)
+		err := r.Read(o)
+		if errors.Is(err, io.EOF) {
+			if i < len(objs) {
+				t.Errorf("too few rows were read: %d<%d", i, len(objs))
+			}
+			break
+		}
+		if !reflect.DeepEqual(*o, objs[i]) {
+			t.Errorf("points mismatch at row index %d: want=%v got=%v", i, objs[i], o)
+		}
+	}
+}
+
 func TestBufferSeekToRow(t *testing.T) {
 	type B struct {
 		I int