fix(parquet/metadata): return page index decode errors (#1032)

### Rationale for this change

`RowGroupPageIndexReader` exposes error-returning methods, but malformed
serialized column or offset indexes could panic inside the lower-level
constructors. A corrupt Parquet file could therefore terminate the
caller instead of producing a read error.

### What changes are included in this PR?

* Convert page-index deserialization panics into errors at the row-group
reader boundary.
* Wrap failures with `arrow.ErrInvalid` and identify the affected index
kind.
* Preserve the existing lower-level constructor signatures for
compatibility.

### Are these changes tested?

Yes. Regression coverage exercises malformed serialized column and
offset indexes. The full `parquet/metadata` package passes.

Author: fallintoplace

parquet/metadata/page_index.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"runtime"
 	"sync"
 
 	"github.com/apache/arrow-go/v18/arrow"
@@ -244,6 +245,32 @@ func NewOffsetIndex(serializedIndex []byte, _ *parquet.ReaderProperties, decrypt
 	return &offsetIndex
 }
 
+func deserializeColumnIndex(descr *schema.Column, serializedIndex []byte, props *parquet.ReaderProperties) (idx ColumnIndex, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			if _, ok := recovered.(runtime.Error); ok {
+				panic(recovered)
+			}
+			err = fmt.Errorf("%w: %w", arrow.ErrInvalid,
+				shared_utils.FormatRecoveredError("invalid column index", recovered))
+		}
+	}()
+	return NewColumnIndex(descr, serializedIndex, props, nil), nil
+}
+
+func deserializeOffsetIndex(serializedIndex []byte, props *parquet.ReaderProperties) (idx OffsetIndex, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			if _, ok := recovered.(runtime.Error); ok {
+				panic(recovered)
+			}
+			err = fmt.Errorf("%w: %w", arrow.ErrInvalid,
+				shared_utils.FormatRecoveredError("invalid offset index", recovered))
+		}
+	}()
+	return NewOffsetIndex(serializedIndex, props, nil), nil
+}
+
 type readRange struct {
 	Offset, Length int64
 }
@@ -358,7 +385,10 @@ func (r *RowGroupPageIndexReader) GetColumnIndex(i int) (ColumnIndex, error) {
 		serializedIndex = decrypted
 	}
 
-	idx := NewColumnIndex(descr, serializedIndex, r.props, nil)
+	idx, err := deserializeColumnIndex(descr, serializedIndex, r.props)
+	if err != nil {
+		return nil, err
+	}
 	r.colIndexes[i] = idx
 	return idx, nil
 }
@@ -418,7 +448,10 @@ func (r *RowGroupPageIndexReader) GetOffsetIndex(i int) (OffsetIndex, error) {
 		serializedIndex = decrypted
 	}
 
-	oidx := NewOffsetIndex(serializedIndex, r.props, nil)
+	oidx, err := deserializeOffsetIndex(serializedIndex, r.props)
+	if err != nil {
+		return nil, err
+	}
 	r.offsetIndices[i] = oidx
 	return oidx, nil
 }

parquet/metadata/page_index_decode_test.go

@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package metadata
+
+import (
+	"bytes"
+	"errors"
+	"testing"
+
+	"github.com/apache/arrow-go/v18/arrow"
+	"github.com/apache/arrow-go/v18/parquet"
+	format "github.com/apache/arrow-go/v18/parquet/internal/gen-go/parquet"
+	"github.com/apache/arrow-go/v18/parquet/internal/thrift"
+	"github.com/apache/arrow-go/v18/parquet/schema"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeserializePageIndexReturnsErrors(t *testing.T) {
+	descr := schema.NewColumn(schema.NewInt32Node("values", parquet.Repetitions.Required, -1), 0, 0)
+
+	_, err := deserializeColumnIndex(descr, []byte{0xff}, nil)
+	assert.Error(t, err)
+	assert.True(t, errors.Is(err, arrow.ErrInvalid))
+
+	_, err = deserializeOffsetIndex([]byte{0xff}, nil)
+	assert.Error(t, err)
+	assert.True(t, errors.Is(err, arrow.ErrInvalid))
+}
+
+func TestRowGroupPageIndexReaderReturnsDecodeErrors(t *testing.T) {
+	meta := constructFakeMetadata([]PageIndexRanges{{
+		ColIndexOffset: 0, ColIndexLen: 1,
+		OffsetIndexOffset: 1, OffsetIndexLen: 1,
+	}})
+	reader := &PageIndexReader{
+		Input:        bytes.NewReader([]byte{0xff, 0xff}),
+		FileMetadata: meta,
+		Props:        parquet.NewReaderProperties(nil),
+	}
+	rgReader, err := reader.RowGroup(0)
+	require.NoError(t, err)
+	require.NotNil(t, rgReader)
+
+	_, err = rgReader.GetColumnIndex(0)
+	assert.ErrorIs(t, err, arrow.ErrInvalid)
+	assertCausePreserved(t, err)
+
+	_, err = rgReader.GetOffsetIndex(0)
+	assert.ErrorIs(t, err, arrow.ErrInvalid)
+	assertCausePreserved(t, err)
+}
+
+func assertCausePreserved(t *testing.T, err error) {
+	t.Helper()
+	unwrapper, ok := err.(interface{ Unwrap() []error })
+	require.True(t, ok)
+	require.Len(t, unwrapper.Unwrap(), 2)
+}
+
+func TestDeserializeColumnIndexRepanicsRuntimeErrors(t *testing.T) {
+	var serialized bytes.Buffer
+	_, err := thrift.NewThriftSerializer().Serialize(&format.ColumnIndex{}, &serialized, nil)
+	require.NoError(t, err)
+
+	assert.Panics(t, func() {
+		deserializeColumnIndex(nil, serialized.Bytes(), nil)
+	})
+}