{"record":{"id":"b17a53a41b06cb81","repo":"apache/beam","slug":"approximateunique-perkey-requires-its-input-to-use-kvcoder","errorCode":null,"errorMessage":"ApproximateUnique.PerKey requires its input to use KvCoder","messagePattern":"ApproximateUnique\\.PerKey requires its input to use KvCoder","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java","lineNumber":268,"sourceCode":"    /**\n     * @see ApproximateUnique#perKey(double)\n     */\n    public PerKey(double estimationError) {\n      if (estimationError < 0.01 || estimationError > 0.5) {\n        throw new IllegalArgumentException(\n            \"ApproximateUnique.PerKey needs an \"\n                + \"estimation error between 1% (0.01) and 50% (0.5).\");\n      }\n\n      this.sampleSize = sampleSizeFromEstimationError(estimationError);\n      this.maximumEstimationError = estimationError;\n    }\n\n    @Override\n    public PCollection<KV<K, Long>> expand(PCollection<KV<K, V>> input) {\n      Coder<KV<K, V>> inputCoder = input.getCoder();\n      if (!(inputCoder instanceof KvCoder)) {\n        throw new IllegalStateException(\n            \"ApproximateUnique.PerKey requires its input to use KvCoder\");\n      }\n      @SuppressWarnings(\"unchecked\")\n      final Coder<V> coder = ((KvCoder<K, V>) inputCoder).getValueCoder();\n\n      return input.apply(Combine.perKey(new ApproximateUniqueCombineFn<>(sampleSize, coder)));\n    }\n\n    @Override\n    public void populateDisplayData(DisplayData.Builder builder) {\n      super.populateDisplayData(builder);\n      ApproximateUnique.populateDisplayData(builder, sampleSize, maximumEstimationError);\n    }\n  }\n\n  /////////////////////////////////////////////////////////////////////////////\n\n  /**","sourceCodeStart":250,"sourceCodeEnd":286,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/ApproximateUnique.java#L250-L286","documentation":"ApproximateUnique.PerKey operates per key, so it must know how to encode and decode the value part of each KV to build an ApproximateUniqueCombineFn for the values. At graph construction it checks the input PCollection's Coder; if it is not a KvCoder it throws IllegalStateException because per-value encoders cannot be extracted. Beam infers KvCoder automatically for KV outputs, so this failure usually means a coder was set or transformed explicitly and lost the KvCoder type.","triggerScenarios":"Applying ApproximateUnique.perKey() to a PCollection<KV<K,V>> whose coder was set via setCoder() to a non-KvCoder, or produced by a transform that erased the coder (e.g. raw PCollection from a custom source with a generic coder).","commonSituations":"Custom DoFns or external sources whose output coder defaults to a generic coder instead of KvCoder; manually calling setCoder on a KV collection; test pipelines that bypass coder inference.","solutions":["Do not override the coder on the KV input; let Beam infer KvCoder from the key and value coders","If you must set it explicitly, use KvCoder.of(keyCoder, valueCoder)","Ensure the upstream transform outputs a properly typed KV with coders inferable via CoderRegistry"],"exampleFix":"// before\noutput.setCoder(SerializableCoder.of(KV.class));\n// after\noutput.setCoder(KvCoder.of(StringUtf8Coder.of(), VarLongCoder.of()));","handlingStrategy":"validation","validationCode":"if (!(input.getCoder() instanceof KvCoder)) {\n  throw new IllegalStateException(\"upstream PCollection must use KvCoder before ApproximateUnique.perKey\");\n}","typeGuard":"Coder<KV<K,V>> c = input.getCoder();\nboolean isKv = c instanceof KvCoder;","tryCatchPattern":"try {\n  result = input.apply(ApproximateUnique.perKey(0.05));\n} catch (IllegalStateException e) {\n  // re-apply with explicit KvCoder on upstream\n  input.setCoder(KvCoder.of(keyCoder, valueCoder));\n  result = input.apply(ApproximateUnique.perKey(0.05));\n}","preventionTips":["Never override a KV PCollection's coder with a non-KvCoder","Let Beam infer coders; check with getCoderResultsInGlobalWindow during pipeline debugging","For custom sources, explicitly set KvCoder.of(keyCoder, valueCoder) on outputs"],"tags":["java","apache-beam","coder","pipeline-construction"],"backgroundTag":"incompatible-source-type","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}