{"record":{"id":"602f3f03ae091c28","repo":"apache/beam","slug":"should-be-either-a-single-pcollection-or-a-dict-of-named","errorCode":null,"errorMessage":"{} should be either a single PCollection or a dict of named PCollections.","messagePattern":"(.+?) should be either a single PCollection or a dict of named PCollections\\.","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/sql/sql_chain.py","lineNumber":174,"sourceCode":"\n  def expand(self, source):\n    \"\"\"Applies the SQL transform. If a PCollection uses a schema defined in\n    the main session, use the additional DoFn to restore it on the worker.\"\"\"\n    if isinstance(source, dict):\n      schema_loaded = {\n          tag: pcoll | 'load_schemas_{}_tag_{}_{}'.format(\n              self.output_name, tag, self.execution_count) >> beam.ParDo(\n                  self._SqlTransformDoFn(self.schemas, self.schema_annotations))\n          if pcoll.element_type in self.schemas else pcoll\n          for tag, pcoll in source.items()\n      }\n    elif isinstance(source, beam.pvalue.PCollection):\n      schema_loaded = source | 'load_schemas_{}_{}'.format(\n          self.output_name, self.execution_count) >> beam.ParDo(\n              self._SqlTransformDoFn(self.schemas, self.schema_annotations)\n          ) if source.element_type in self.schemas else source\n    else:\n      raise ValueError(\n          '{} should be either a single PCollection or a dict of named '\n          'PCollections.'.format(source))\n    return schema_loaded | 'beam_sql_{}_{}'.format(\n        self.output_name, self.execution_count) >> SqlTransform(self.query)\n\n\n@dataclass\nclass SqlChain:\n  \"\"\"A chain of SqlNodes.\n\n  Attributes:\n    nodes: all nodes by their output_names.\n    root: the first SqlNode applied chronologically.\n    current: the last node applied.\n    user_pipeline: the user defined pipeline this chain originates from. If\n      None, the whole chain just computes from raw values in queries.\n      Otherwise, at least some of the nodes in chain has queried against\n      PCollections.","sourceCodeStart":156,"sourceCodeEnd":192,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/sql/sql_chain.py#L156-L192","documentation":"SqlChain.expand accepts either a single PCollection or a dict of named PCollections as the SQL source; any other type (list, None, string, etc.) raises ValueError telling the caller the source shape is invalid.","triggerScenarios":"Constructing a SqlChain (or calling find_sql_source_based_inputs / expand) with a source that is neither beam.PCollection nor dict, e.g. passing the result of a lookup that returned None or a list of PCollections.","commonSituations":"Programmatic use of SqlChain outside the %%beam_sql magic; passing multiple PCollections as a list instead of a dict keyed by name; a variable holding a non-PCollection after a failed lookup.","solutions":["Wrap multiple PCollections in a dict: {'name1': pcoll1, 'name2': pcoll2}","Pass a single PCollection directly if the query uses one input","Check that the variable you pass is a beam.PCollection (print its type)","Resolve PCollections by name via pcoll_by_name() and confirm each lookup succeeded"],"exampleFix":"// before\nchain = SqlChain(query, source=[pcoll1, pcoll2])\n// after\nchain = SqlChain(query, source={'pcoll1': pcoll1, 'pcoll2': pcoll2})","handlingStrategy":"type-guard","validationCode":"def valid_sql_source(source):\n    import beam as _b\n    return isinstance(source, _b.pvalue.PCollection) or (\n        isinstance(source, dict) and all(isinstance(v, _b.pvalue.PCollection) for v in source.values()))","typeGuard":"def is_sql_source(source):\n    return isinstance(source, beam.pvalue.PCollection) or (\n        isinstance(source, dict) and source and\n        all(isinstance(v, beam.pvalue.PCollection) for v in source.values()))","tryCatchPattern":"try:\n    chain.to_pipeline()\nexcept ValueError as e:\n    if 'should be either a single PCollection' in str(e):\n        source = {'main': source_or_lookup}","preventionTips":["Always pass dict[name, PCollection] for multi-input SQL queries","Verify lookup results are PCollection instances before building a SqlChain","Never pass lists or None as the SQL source","Test SqlChain construction with a minimal query first"],"tags":["sql","type-mismatch","interactive-beam"],"backgroundTag":"type-mismatch","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"}