This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 1996fda0 fix(arrow/util): avoid protobuf map schema goroutines (#1082)
1996fda0 is described below
commit 1996fda09eb60eb6e1c62726c67451b53ffcf80d
Author: Minh Vu <[email protected]>
AuthorDate: Wed Aug 5 20:21:35 2026 +0200
fix(arrow/util): avoid protobuf map schema goroutines (#1082)
## What changed
Build protobuf map Arrow types directly from descriptors and replace
channel-backed map entry generation with a synchronous `iter.Seq`.
## Why
Schema discovery previously started an unbuffered producer and returned
after its first result. Maps with multiple entries left that producer
blocked. Conversion failures could also stop consumption early and
strand the producer. Synchronous iteration makes early termination safe
and avoids runtime-value-dependent schema construction.
## Testing
- `go test ./arrow/util`
- Added coverage showing map schemas are independent of runtime entries
---
arrow/util/protobuf_reflect.go | 28 +++++++++++-----------------
arrow/util/protobuf_reflect_test.go | 11 +++++++++++
2 files changed, 22 insertions(+), 17 deletions(-)
diff --git a/arrow/util/protobuf_reflect.go b/arrow/util/protobuf_reflect.go
index 1e826c3f..17fa9344 100644
--- a/arrow/util/protobuf_reflect.go
+++ b/arrow/util/protobuf_reflect.go
@@ -18,6 +18,7 @@ package util
import (
"fmt"
+ "iter"
"reflect"
"github.com/apache/arrow-go/v18/arrow"
@@ -374,12 +375,6 @@ func (pfr *ProtobufFieldReflection) asMap()
protobufMapReflection {
}
func (pmr protobufMapReflection) getDataType() arrow.DataType {
- for kvp := range pmr.generateKeyValuePairs() {
- if kvp.err != nil {
- continue
- }
- return kvp.kvp.getDataType()
- }
return protobufMapKeyValuePairReflection{
k: ProtobufFieldReflection{
parent: pmr.parent,
@@ -408,11 +403,8 @@ func (pmr protobufMapKeyValuePairReflection) getDataType()
arrow.DataType {
return arrow.MapOf(pmr.k.getDataType(), pmr.v.getDataType())
}
-func (pmr protobufMapReflection) generateKeyValuePairs() chan
protobufMapKeyValuePairResult {
- out := make(chan protobufMapKeyValuePairResult)
-
- go func() {
- defer close(out)
+func (pmr protobufMapReflection) generateKeyValuePairs()
iter.Seq[protobufMapKeyValuePairResult] {
+ return func(yield func(protobufMapKeyValuePairResult) bool) {
if !pmr.rValue.IsValid() {
kvp := protobufMapKeyValuePairReflection{
k: ProtobufFieldReflection{
@@ -426,13 +418,15 @@ func (pmr protobufMapReflection) generateKeyValuePairs()
chan protobufMapKeyValu
schemaOptions: pmr.schemaOptions,
},
}
- out <- protobufMapKeyValuePairResult{kvp: kvp}
+ yield(protobufMapKeyValuePairResult{kvp: kvp})
return
}
for _, k := range pmr.rValue.MapKeys() {
mapKey, err := getMapKey(k)
if err != nil {
- out <- protobufMapKeyValuePairResult{err: err}
+ if !yield(protobufMapKeyValuePairResult{err:
err}) {
+ return
+ }
continue
}
kvp := protobufMapKeyValuePairReflection{
@@ -451,11 +445,11 @@ func (pmr protobufMapReflection) generateKeyValuePairs()
chan protobufMapKeyValu
schemaOptions: pmr.schemaOptions,
},
}
- out <- protobufMapKeyValuePairResult{kvp: kvp}
+ if !yield(protobufMapKeyValuePairResult{kvp: kvp}) {
+ return
+ }
}
- }()
-
- return out
+ }
}
func getMapKey(v reflect.Value) (protoreflect.Value, error) {
diff --git a/arrow/util/protobuf_reflect_test.go
b/arrow/util/protobuf_reflect_test.go
index e27ba7e4..70a0e390 100644
--- a/arrow/util/protobuf_reflect_test.go
+++ b/arrow/util/protobuf_reflect_test.go
@@ -337,6 +337,17 @@ func TestGetSchema(t *testing.T) {
CheckSchema(t, pmr, want)
}
+func TestMapSchemaDoesNotDependOnValues(t *testing.T) {
+ withValues := util_message.AllTheTypesNoAny{
+ SimpleMap: map[int32]string{1: "one", 2: "two"},
+ }
+ empty := util_message.AllTheTypesNoAny{}
+
+ withValuesSchema := NewProtobufMessageReflection(&withValues).Schema()
+ emptySchema := NewProtobufMessageReflection(&empty).Schema()
+ require.True(t, withValuesSchema.Equal(emptySchema))
+}
+
func TestRecordFromProtobuf(t *testing.T) {
f := AllTheTypesFixture()
