Consider the following simple Apache Beam Go program:
package main
import (
"context"
"log"
"github.com/apache/beam/sdks/v2/go/pkg/beam"
"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
_ "github.com/apache/beam/sdks/v2/go/pkg/beam/runners/direct"
"github.com/apache/beam/sdks/v2/go/pkg/beam/transforms/stats"
"github.com/apache/beam/sdks/v2/go/pkg/beam/x/beamx"
)
func init() {
register.Function2x0(fn)
}
func main() {
beam.Init()
p := beam.NewPipeline()
s := p.Root()
rows := beam.Create(s, "1", "1", "2")
counts := stats.Count(s, rows)
beam.ParDo0(s, fn, rows, beam.SideInput{Input: counts})
if err := beamx.Run(context.Background(), p); err != nil {
log.Fatal(err)
}
}
func fn(in string, lookup func(string) func(*int) bool) {
var count int
lookup("1")(&count)
}
Running with the direct runner results in the following error:
2022/09/08 22:09:36 Executing pipeline with the direct runner.
2022/09/08 22:09:36 Pipeline:
2022/09/08 22:09:36 Nodes: {1: []uint8/bytes GLO}
{2: string/string GLO}
{3: KV<string,int>/KV<string,int[varintz]> GLO}
{4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}
{5: KV<string,int>/KV<string,int[varintz]> GLO}
Edges: 1: Impulse [] -> [Out: []uint8 -> {1: []uint8/bytes GLO}]
2: ParDo [In(Main): []uint8 <- {1: []uint8/bytes GLO}] -> [Out: T -> {2: string/string GLO}]
3: ParDo [In(Main): T <- {2: string/string GLO}] -> [Out: KV<T,int> -> {3: KV<string,int>/KV<string,int[varintz]> GLO}]
4: CoGBK [In(Main): KV<string,int> <- {3: KV<string,int>/KV<string,int[varintz]> GLO}] -> [Out: CoGBK<string,int> -> {4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}]
5: Combine [In(Main): int <- {4: CoGBK<string,int>/CoGBK<string,int[varintz]> GLO}] -> [Out: KV<string,int> -> {5: KV<string,int>/KV<string,int[varintz]> GLO}]
6: ParDo [In(Main): string <- {2: string/string GLO} In(MultiMap): KV<string,int> <- {5: KV<string,int>/KV<string,int[varintz]> GLO}] -> []
2022/09/08 22:09:36 Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
2022/09/08 22:09:36 wait[2] unblocked w/ 3 [false]
2022/09/08 22:09:36 panic: interface conversion: interface {} is string, not int
Full error:
while executing FinishBundle for Plan[plan]:
10: Impulse[0]
1: ParDo[main.fn] Out:[]
2: wait[1] Out:1
3: buffer[3]. wait:2 Out:1
4: Combine[stats.sumIntFn] Keyed:false Out:3
5: CoGBK. Out:4
6: Inject[0]. Out:5
7: ParDo[stats.keyedCountFn] Out:[6]
8: Multiplex. Out:[7 2]
9: ParDo[beam.createFn] Out:[8]
caused by:
panic: interface conversion: interface {} is string, not int
How come the type is inferred as string when in fact it is successfully registered as a side input function with a KV<string,int> signature?
A very similar approach can be seen here:
Update: Seems like this is a bug in the direct runner since running it on Dataflow works.