[BEAM-12257] Can't infer accumulator coder for LazyAggregateCombineFn.

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.31.0
• Component/s: dsl-sql-zetasql
• Labels: zetasql-java-udf
• Assignee: Kyle Weaver
• Reporter: Kyle Weaver
• Created: 30/Apr/21 00:42
• Updated: 07/May/21 01:20
• Resolved: 30/Apr/21 18:45

Description

This happens when using DataflowRunner, but not DirectRunner.

 java.lang.RuntimeException: java.io.IOException: Could not obtain a Coder for the accumulator java.lang.RuntimeException: java.io.IOException: Could not obtain a Coder for the accumulator  at org.apache.beam.runners.core.construction.PipelineTranslation$1.leaveCompositeTransform(PipelineTranslation.java:78)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:584)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:579)  at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$500(TransformHierarchy.java:239)  at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:213)  at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:468)  at org.apache.beam.runners.core.construction.PipelineTranslation.toProto(PipelineTranslation.java:60)  at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:988)  at org.apache.beam.runners.dataflow.DataflowRunner.run(DataflowRunner.java:196)  at org.apache.beam.sdk.Pipeline.run(Pipeline.java:322)  at org.apache.beam.sdk.Pipeline.run(Pipeline.java:308)  at com.google.cloud.dataflow.sqllauncher.UdfIT.udaf(UdfIT.java:74)  ... 34 trimmed Caused by: java.io.IOException: Could not obtain a Coder for the accumulator  at org.apache.beam.runners.core.construction.CombineTranslation$CombineGroupedValuesPayloadTranslator.extractAccumulatorCoder(CombineTranslation.java:207)  at org.apache.beam.runners.core.construction.CombineTranslation$CombineGroupedValuesPayloadTranslator.translate(CombineTranslation.java:179)  at org.apache.beam.runners.core.construction.PTransformTranslation$KnownTransformPayloadTranslator.translate(PTransformTranslation.java:438)  at org.apache.beam.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:248)  at org.apache.beam.runners.core.construction.SdkComponents.registerPTransform(SdkComponents.java:177)  at org.apache.beam.runners.core.construction.PipelineTranslation$1.leaveCompositeTransform(PipelineTranslation.java:75)  ... 49 more Caused by: org.apache.beam.sdk.coders.CannotProvideCoderException: Cannot infer coder for type parameter AccumT  at org.apache.beam.sdk.coders.CoderRegistry.getCoder(CoderRegistry.java:328)  at org.apache.beam.sdk.transforms.CombineFnBase$AbstractGlobalCombineFn.getAccumulatorCoder(CombineFnBase.java:119)  at org.apache.beam.sdk.transforms.Combine$CombineFn.getAccumulatorCoder(Combine.java:391)  at org.apache.beam.sdk.extensions.sql.impl.transform.agg.AggregationCombineFnAdapter$WrappedCombinerBase.getAccumulatorCoder(AggregationCombineFnAdapter.java:75)  at org.apache.beam.sdk.transforms.CombineFns$ComposedCombineFn.getAccumulatorCoder(CombineFns.java:430)  at org.apache.beam.sdk.schemas.transforms.SchemaAggregateFn$Inner.getAccumulatorCoder(SchemaAggregateFn.java:335)  at org.apache.beam.runners.core.construction.CombineTranslation$CombineGroupedValuesPayloadTranslator.extractAccumulatorCoder(CombineTranslation.java:204)  ... 54 more

Issue Links

• links to: GitHub Pull Request #14693