GH-41988: [Go] Add FormatRecoveredError to consistently handle recove…

…ry with wrapped errors (#41989)

### Rationale for this change

Part of #36186 (Fixes it, IMO)
Fixes #41988.

### What changes are included in this PR?

A new internal `FormatRecoveredError()` function adds consistency to
errors generated from panic recovery.

### Are these changes tested?

Yes.

### Are there any user-facing changes?

Users will be able to unwrap errors thrown by allocators, for example.

* GitHub Issue: #41988

go/arrow/array/concat.go

@@ -520,12 +520,7 @@ func concat(data []arrow.ArrayData, mem memory.Allocator) (arr arrow.ArrayData,
 	out := &Data{refCount: 1, dtype: data[0].DataType(), nulls: 0}
 	defer func() {
 		if pErr := recover(); pErr != nil {
-			switch e := pErr.(type) {
-			case error:
-				err = fmt.Errorf("arrow/concat: %w", e)
-			default:
-				err = fmt.Errorf("arrow/concat: %v", pErr)
-			}
+			err = utils.FormatRecoveredError("arrow/concat", pErr)
 		}
 		if err != nil {
 			out.Release()

go/arrow/avro/schema.go

@@ -25,6 +25,7 @@ import (
 	"github.com/apache/arrow/go/v17/arrow"
 	"github.com/apache/arrow/go/v17/arrow/decimal128"
 	"github.com/apache/arrow/go/v17/internal/types"
+	"github.com/apache/arrow/go/v17/internal/utils"
 	avro "github.com/hamba/avro/v2"
 )
 
@@ -76,14 +77,7 @@ func ArrowSchemaFromAvro(schema avro.Schema) (s *arrow.Schema, err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			s = nil
-			switch x := r.(type) {
-			case string:
-				err = fmt.Errorf("invalid avro schema: %s", x)
-			case error:
-				err = fmt.Errorf("invalid avro schema: %w", x)
-			default:
-				err = fmt.Errorf("invalid avro schema: unknown error")
-			}
+			err = utils.FormatRecoveredError("invalid avro schema", r)
 		}
 	}()
 	n := newSchemaNode()

go/arrow/flight/record_batch_reader.go

@@ -27,6 +27,7 @@ import (
 	"github.com/apache/arrow/go/v17/arrow/internal/debug"
 	"github.com/apache/arrow/go/v17/arrow/ipc"
 	"github.com/apache/arrow/go/v17/arrow/memory"
+	"github.com/apache/arrow/go/v17/internal/utils"
 )
 
 // DataStreamReader is an interface for receiving flight data messages on a stream
@@ -217,7 +218,7 @@ func StreamChunksFromReader(rdr array.RecordReader, ch chan<- StreamChunk) {
 	defer close(ch)
 	defer func() {
 		if err := recover(); err != nil {
-			ch <- StreamChunk{Err: fmt.Errorf("panic while reading: %s", err)}
+			ch <- StreamChunk{Err: utils.FormatRecoveredError("panic while reading", err)}
 		}
 	}()
 
@@ -243,7 +244,7 @@ func ConcatenateReaders(rdrs []array.RecordReader, ch chan<- StreamChunk) {
 		}
 
 		if err := recover(); err != nil {
-			ch <- StreamChunk{Err: fmt.Errorf("panic while reading: %s", err)}
+			ch <- StreamChunk{Err: utils.FormatRecoveredError("panic while reading", err)}
 		}
 	}()
 

go/arrow/internal/cdata_integration/entrypoints.go

@@ -29,6 +29,7 @@ import (
 	"github.com/apache/arrow/go/v17/arrow/cdata"
 	"github.com/apache/arrow/go/v17/arrow/internal/arrjson"
 	"github.com/apache/arrow/go/v17/arrow/memory"
+	"github.com/apache/arrow/go/v17/internal/utils"
 )
 
 // #include <stdint.h>
@@ -59,7 +60,7 @@ func ArrowGo_FreeError(cError *C.char) {
 func handlePanic(err *error) {
 	if e := recover(); e != nil {
 		// Add a prefix while wrapping the panic-error
-		*err = fmt.Errorf("panic: %w", e.(error))
+		*err = utils.FormatRecoveredError("panic", e)
 	}
 }
 

go/arrow/ipc/reader.go

@@ -30,6 +30,7 @@ import (
 	"github.com/apache/arrow/go/v17/arrow/internal/dictutils"
 	"github.com/apache/arrow/go/v17/arrow/internal/flatbuf"
 	"github.com/apache/arrow/go/v17/arrow/memory"
+	"github.com/apache/arrow/go/v17/internal/utils"
 )
 
 // Reader reads records from an io.Reader.
@@ -60,7 +61,7 @@ type Reader struct {
 func NewReaderFromMessageReader(r MessageReader, opts ...Option) (reader *Reader, err error) {
 	defer func() {
 		if pErr := recover(); pErr != nil {
-			err = fmt.Errorf("arrow/ipc: unknown error while reading: %v", pErr)
+			err = utils.FormatRecoveredError("arrow/ipc: unknown error while reading", pErr)
 		}
 	}()
 	cfg := newConfig()
@@ -213,7 +214,7 @@ func (r *Reader) getInitialDicts() bool {
 func (r *Reader) next() bool {
 	defer func() {
 		if pErr := recover(); pErr != nil {
-			r.err = fmt.Errorf("arrow/ipc: unknown error while reading: %v", pErr)
+			r.err = utils.FormatRecoveredError("arrow/ipc: unknown error while reading", pErr)
 		}
 	}()
 	if r.schema == nil {

go/arrow/ipc/writer.go

@@ -151,7 +151,7 @@ func (w *Writer) Close() error {
 func (w *Writer) Write(rec arrow.Record) (err error) {
 	defer func() {
 		if pErr := recover(); pErr != nil {
-			err = fmt.Errorf("arrow/ipc: unknown error while writing: %v", pErr)
+			err = utils.FormatRecoveredError("arrow/ipc: unknown error while writing", pErr)
 		}
 	}()
 

go/internal/utils/recovery.go

@@ -0,0 +1,31 @@
+// 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 utils
+
+import "fmt"
+
+// FormatRecoveredError is used in cases where a panic/recover receives an
+// object which is potentially an error that could be wrapped, instead of
+// formatted, so that callers can see it.  This may be useful, for example,
+// with custom Allocators which panic to signal failure; these panics will be
+// recovered as wrapped errors, letting the client distinguish them.
+func FormatRecoveredError(msg string, recovered any) error {
+	if err, ok := recovered.(error); ok {
+		return fmt.Errorf("%s: %w", msg, err)
+	}
+	return fmt.Errorf("%s: %v", msg, recovered)
+}

go/internal/utils/recovery_test.go

@@ -0,0 +1,62 @@
+// 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 utils
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type testError struct{}
+
+var _ error = testError{}
+
+func (testError) Error() string {
+	return "test error"
+}
+
+func TestFormatRecoveredError(t *testing.T) {
+	defer func() {
+		thing := recover()
+		assert.NotNil(t, thing)
+		assert.Error(t, thing.(testError))
+
+		err := FormatRecoveredError("recovered thing", thing)
+
+		assert.Equal(t, "recovered thing: test error", err.Error())
+		assert.True(t, errors.Is(err, testError{}))
+		assert.Equal(t, "test error", errors.Unwrap(err).(testError).Error())
+	}()
+
+	panic(testError{})
+}
+
+func TestFormatRecoveredNonError(t *testing.T) {
+	defer func() {
+		thing := recover()
+		assert.NotNil(t, thing)
+
+		err := FormatRecoveredError("recovered thing", thing)
+
+		assert.Equal(t, "recovered thing: just a message", err.Error())
+		assert.False(t, errors.Is(err, testError{}))
+	}()
+
+	panic("just a message")
+}