{"record":{"id":"a562de1152b0bbbe","repo":"apache/beam","slug":"combinefn-spec-missing-type-fn-spec","errorCode":null,"errorMessage":"CombineFn spec missing type: {fn_spec}","messagePattern":"CombineFn spec missing type: (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/yaml/yaml_combine.py","lineNumber":122,"sourceCode":"  def __init__(\n      self,\n      group_by: Iterable[str],\n      combine: Mapping[str, Mapping[str, Any]],\n      language: Optional[str] = None):\n    self._group_by = group_by\n    self._combine = combine\n    self._language = language\n\n  def expand(self, pcoll):\n    input_types = dict(named_fields_from_element_type(pcoll.element_type))\n    all_fields = list(input_types.keys())\n    unknown_keys = set(self._group_by) - set(all_fields)\n    if unknown_keys:\n      raise ValueError(f'Unknown grouping columns: {list(unknown_keys)}')\n\n    def create_combine_fn(fn_spec):\n      if 'type' not in fn_spec:\n        raise ValueError(f'CombineFn spec missing type: {fn_spec}')\n      elif fn_spec['type'] in BUILTIN_COMBINE_FNS:\n        return BUILTIN_COMBINE_FNS[fn_spec['type']]\n      elif self._language == 'python':\n        # TODO(yaml): Support output_type here as well.\n        fn = python_callable.PythonCallableWithSource.load_from_source(\n            fn_spec['type'])\n        if 'config' in fn_spec:\n          fn = fn(**fn_spec['config'])\n        return fn\n      else:\n        raise TypeError('Unknown CombineFn: {fn_spec}')\n\n    def extract_return_type(expr):\n      if isinstance(expr, str) and expr in input_types:\n        return input_types[expr]\n      expr_hints = get_type_hints(expr)\n      if (expr_hints and expr_hints.has_simple_output_type() and\n          expr_hints.simple_output_type(None) != typehints.Any):","sourceCodeStart":104,"sourceCodeEnd":140,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/yaml/yaml_combine.py#L104-L140","documentation":"create_combine_fn inside Combine.expand raises ValueError when an aggregation's fn spec mapping has no 'type' key. Each combine fn spec must name either a builtin combine function or a Python callable source.","triggerScenarios":"YAML combine config entry like {value: amount, output: total} without fn/type, or a fn mapping like fn: {config: ...} missing type.","commonSituations":"Omitting 'fn'/'type' while only specifying 'value'; malformed copy-paste where the fn block was truncated.","solutions":["Add fn: <name> (shorthand) or fn: {type: <name>} to each combine entry.","Use a builtin name (sum, count, mean, min, max, etc.) or supply a Python callable via source.","Validate each entry has both a destination field and a fn/type."],"exampleFix":"// before\ncombine:\n  total: {value: amount}\n// after\ncombine:\n  total: {fn: sum, value: amount}","handlingStrategy":"validation","validationCode":"def check_agg_specs(combine_cfg: dict):\n    for dest, agg in combine_cfg.get('combine', {}).items():\n        fn = agg.get('fn')\n        if isinstance(fn, dict) and 'type' not in fn:\n            raise SystemExit(f'combine.{dest}: fn spec missing type')\n        if fn is None and 'fn' not in agg:\n            raise SystemExit(f'combine.{dest}: missing fn')","typeGuard":"def has_fn_type(agg: dict) -> bool:\n    fn = agg.get('fn')\n    return isinstance(fn, str) or (isinstance(fn, dict) and isinstance(fn.get('type'), str))","tryCatchPattern":"try:\n    expand(pcoll)\nexcept ValueError as e:\n    if 'CombineFn spec missing type' in str(e):\n        raise SystemExit('Add fn: <name> or fn: {type: <name>} to each combine entry') from e\n    raise","preventionTips":["Always specify fn (string shorthand or mapping with type)","Lint combine entries for both destination and fn","Copy aggregation examples from Beam docs"],"tags":["yaml","config","beam-yaml"],"backgroundTag":"missing-required-config-field","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}