apache/beam · error · ValueError

The beam_sql magic tries to query PCollections from multiple

Error message

The beam_sql magic tries to query PCollections from multiple pipelines: %s and %s

What it means

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.

Source

Thrown at sdks/python/apache_beam/runners/interactive/interactive_environment.py:715

    the body of the document is always changing due to cell [re-]execution,
    deletion and re-ordering. Thus, HTML imports shouldn't be put in the body
    especially the output areas of notebook cells.
    """
    try:
      from IPython.display import Javascript
      from IPython.display import display_javascript
      display_javascript(
          Javascript(_HTML_IMPORT_TEMPLATE.format(hrefs=html_hrefs)))
    except ImportError:
      pass  # NOOP if dependencies are not available.

  def get_sql_chain(self, pipeline, set_user_pipeline=False):
    if pipeline not in self.sql_chain:
      self.sql_chain[pipeline] = SqlChain()
    chain = self.sql_chain[pipeline]
    if set_user_pipeline:
      if chain.user_pipeline and chain.user_pipeline is not pipeline:
        raise ValueError(
            'The beam_sql magic tries to query PCollections from multiple '
            'pipelines: %s and %s',
            chain.user_pipeline,
            pipeline)
      chain.user_pipeline = pipeline
    return chain

  def _get_gcs_cache_dir(self, pipeline, cache_dir):
    cache_dir_path = PurePath(cache_dir)
    if len(cache_dir_path.parts) < 2:
      _LOGGER.error(
          'GCS bucket cache path "%s" is too short to be valid. See '
          'https://cloud.google.com/storage/docs/naming-buckets for '
          'the expected format.',
          cache_dir)
      raise ValueError('cache_root GCS bucket path is invalid.')
    bucket_name = cache_dir_path.parts[1]
    assert_bucket_exists(bucket_name)

View on GitHub (pinned to 12126d8942)

Solutions

1. 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.
2. Clear stale interactive environment state (restart the kernel) if old pipeline references persist after re-running cells.
3. Route all inputs of a SQL query through the same pipeline (e.g. convert other sources with to_pcollection on that pipeline).
4. Check the two pipeline objects in the raw tuple args to identify which stale pipeline is registered.
5. Materialize one side with ib.collect and feed it back as a Create source instead of mixing chains.

Example fix

// 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

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)

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())

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Conflicting config options: "cannot be used together" — configuration validation errors across open-source libraries — this error's family across 162 libraries.

Related errors

• The given pcoll %s is not a dict, an iterable or a PCollecti
• All PCollections must belong to the same pipeline.
• The given pcoll {pcoll_container} is not a dict, an iterable
• cache_root GCS bucket path is invalid.
• PCollection not available, please run the pipeline.

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/456cb85b909227f4. Report an issue: GitHub.