GH-34807: [Go] Handle io.EOF when reading parquet footer size and magic bytes

go/parquet/file/file_reader.go

@@ -163,7 +163,7 @@ func (f *Reader) parseMetaData() error {
 	buf := make([]byte, footerSize)
 	// backup 8 bytes to read the footer size (first four bytes) and the magic bytes (last 4 bytes)
 	n, err := f.r.ReadAt(buf, f.footerOffset-int64(footerSize))
-	if err != nil {
+	if err != nil && err != io.EOF {
 		return fmt.Errorf("parquet: could not read footer: %w", err)
 	}
 	if n != len(buf) {

go/parquet/file/file_reader_test.go

@@ -19,6 +19,8 @@ package file_test
 import (
 	"bytes"
 	"encoding/binary"
+	"github.com/apache/arrow/go/v12/parquet"
+	"github.com/apache/arrow/go/v12/parquet/schema"
 	"io"
 	"math/rand"
 	"testing"
@@ -65,6 +67,19 @@ func checkStatistics(t *testing.T, stats format.Statistics, actual metadata.Enco
 	}
 }
 
+type testReader struct {
+	*bytes.Reader
+}
+
+// ReadAt for testReader returns io.EOF when off + len(b) is exactly the length of the underlying input source.
+func (tr testReader) ReadAt(b []byte, off int64) (int, error) {
+	n, err := tr.Reader.ReadAt(b, off)
+	if err == nil && (int64(n)+off == tr.Size()) {
+		return n, io.EOF
+	}
+	return n, err
+}
+
 type PageSerdeSuite struct {
 	suite.Suite
 
@@ -271,6 +286,31 @@ func (p *PageSerdeSuite) TestCompression() {
 	}
 }
 
+func TestWithEOFReader(t *testing.T) {
+	sink := encoding.NewBufferWriter(0, memory.DefaultAllocator)
+	serializer := thrift.NewThriftSerializer()
+	fields := schema.FieldList{
+		schema.NewInt32Node("int_col", parquet.Repetitions.Required, -1),
+	}
+	root, _ := schema.NewGroupNode("schema", parquet.Repetitions.Repeated, fields, -1)
+	schema := schema.NewSchema(root)
+	props := parquet.NewWriterProperties(parquet.WithVersion(parquet.V2_LATEST))
+	builder := metadata.NewFileMetadataBuilder(schema, props, nil)
+	fileMetaData, _ := builder.Finish()
+
+	size, _ := serializer.Serialize(fileMetaData, sink, nil)
+
+	binary.Write(sink, binary.LittleEndian, uint32(size))
+	magic := []byte("PAR1")
+	sink.Write(magic)
+	buf := sink.Finish()
+	defer buf.Release()
+
+	r := bytes.NewReader(buf.Bytes())
+	_, err := file.NewParquetReader(testReader{Reader: r})
+	assert.NoError(t, err)
+}
+
 func TestInvalidHeaders(t *testing.T) {
 	badHeader := []byte("PAR2")
 	_, err := file.NewParquetReader(bytes.NewReader(badHeader))