{"record":{"id":"456cb85b909227f4","repo":"apache/beam","slug":"the-beam-sql-magic-tries-to-query-pcollections-from-multiple","errorCode":null,"errorMessage":"The beam_sql magic tries to query PCollections from multiple pipelines: %s and %s","messagePattern":"The beam_sql magic tries to query PCollections from multiple pipelines: (.+?) and (.+?)","errorType":"validation","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/interactive/interactive_environment.py","lineNumber":715,"sourceCode":"    the body of the document is always changing due to cell [re-]execution,\n    deletion and re-ordering. Thus, HTML imports shouldn't be put in the body\n    especially the output areas of notebook cells.\n    \"\"\"\n    try:\n      from IPython.display import Javascript\n      from IPython.display import display_javascript\n      display_javascript(\n          Javascript(_HTML_IMPORT_TEMPLATE.format(hrefs=html_hrefs)))\n    except ImportError:\n      pass  # NOOP if dependencies are not available.\n\n  def get_sql_chain(self, pipeline, set_user_pipeline=False):\n    if pipeline not in self.sql_chain:\n      self.sql_chain[pipeline] = SqlChain()\n    chain = self.sql_chain[pipeline]\n    if set_user_pipeline:\n      if chain.user_pipeline and chain.user_pipeline is not pipeline:\n        raise ValueError(\n            'The beam_sql magic tries to query PCollections from multiple '\n            'pipelines: %s and %s',\n            chain.user_pipeline,\n            pipeline)\n      chain.user_pipeline = pipeline\n    return chain\n\n  def _get_gcs_cache_dir(self, pipeline, cache_dir):\n    cache_dir_path = PurePath(cache_dir)\n    if len(cache_dir_path.parts) < 2:\n      _LOGGER.error(\n          'GCS bucket cache path \"%s\" is too short to be valid. See '\n          'https://cloud.google.com/storage/docs/naming-buckets for '\n          'the expected format.',\n          cache_dir)\n      raise ValueError('cache_root GCS bucket path is invalid.')\n    bucket_name = cache_dir_path.parts[1]\n    assert_bucket_exists(bucket_name)","sourceCodeStart":697,"sourceCodeEnd":733,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/interactive/interactive_environment.py#L697-L733","documentation":"InteractiveEnvironment tracks the SQL 'chain' (sequence of beam_sql magics) per pipeline. get_sql_chain(pipeline, set_user_pipeline=True) raises this ValueError when the chain already has a user pipeline and a different pipeline object is being registered — i.e. the %%beam_sql magic is being used to query PCollections from more than one distinct pipeline, which the SQL chain cannot represent. Note the raise passes printf-style args to ValueError without a % placeholder, so the printed message shows a raw tuple; the two pipelines are chain.user_pipeline and the new pipeline.","triggerScenarios":"Applying %%beam_sql to a PCollection from pipeline B when previous beam_sql queries in the session were chained on pipeline A; mixing SQL queries over pcolls from a re-created pipeline object; calling get_sql_chain directly with set_user_pipeline=True after another pipeline was registered.","commonSituations":"Notebooks: re-running the pipeline-definition cell creates a new Pipeline object while earlier beam_sql chains still point at the old one; combining SQL results from an import-time pipeline with the notebook pipeline; switching runners mid-session.","solutions":["Use one beam.Pipeline() for all PCollections involved in beam_sql queries; re-run the beam_sql queries after rebuilding the pipeline so the chain re-binds.","Clear stale interactive environment state (restart the kernel) if old pipeline references persist after re-running cells.","Route all inputs of a SQL query through the same pipeline (e.g. convert other sources with to_pcollection on that pipeline).","Check the two pipeline objects in the raw tuple args to identify which stale pipeline is registered.","Materialize one side with ib.collect and feed it back as a Create source instead of mixing chains."],"exampleFix":"// before: pa = beam.Pipeline() used for first beam_sql, pb = beam.Pipeline() for the second | // after: p = beam.Pipeline(); r1 = p | 'R1' >> ...; r2 = p | 'R2' >> ...; run both %%beam_sql queries against pcolls from p","handlingStrategy":"validation","validationCode":"def chain_belongs_to(pipeline): ie = ib.current_env(); return all(c.user_pipeline is None or c.user_pipeline is pipeline for c in getattr(ie, 'sql_chain', {}).values()); assert chain_belongs_to(pipeline)","typeGuard":"def chain_belongs_to(pipeline): ie = ib.current_env(); return all(c.user_pipeline is None or c.user_pipeline is pipeline for c in getattr(ie, 'sql_chain', {}).values())","tryCatchPattern":"try: result = run_beam_sql(pcoll) | except ValueError as e: (print('Rebuild pcolls on one pipeline and re-run all beam_sql cells') if 'multiple' in str(e) and 'pipelines' in str(e) else None); raise","preventionTips":["Keep all beam_sql inputs on one pipeline object.","After rebuilding a pipeline, re-run every beam_sql cell so chains re-register.","Avoid mixing SQL queries across cells backed by different pipeline objects.","Restart the kernel if stale chain state persists."],"tags":["python","apache-beam","interactive","sql","value-error"],"backgroundTag":"conflicting-config-options","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"}