{"record":{"id":"cc118b2bc9a9e138","repo":"apache/beam","slug":"joining-unbounded-pcollections-is-currently-only-supported","errorCode":null,"errorMessage":"Joining unbounded PCollections is currently only supported for non-global windows with triggers that are known to produce output once per window,such as the default trigger with zero allowed lateness. In these cases Beam can guarantee it joins all input elements once per window. ${windowingStrategy} is not supported","messagePattern":"Joining unbounded PCollections is currently only supported for non-global windows with triggers that are known to produce output once per window,such as the default trigger with zero allowed lateness\\. In these cases Beam can guarantee it joins all input elements once per window\\. (.+?) is not supported","errorType":"exception","errorClass":"java.lang.UnsupportedOperationException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCoGBKJoinRel.java","lineNumber":143,"sourceCode":"      try {\n        leftWinFn.verifyCompatibility(rightWinFn);\n      } catch (IncompatibleWindowException e) {\n        throw new IllegalArgumentException(\n            \"WindowFns must match for a bounded-vs-bounded/unbounded-vs-unbounded join.\", e);\n      }\n\n      verifySupportedTrigger(leftRows);\n      verifySupportedTrigger(rightRows);\n\n      return standardJoin(leftRows, rightRows, leftKeyFields, rightKeyFields);\n    }\n  }\n\n  private <T> void verifySupportedTrigger(PCollection<T> pCollection) {\n    WindowingStrategy windowingStrategy = pCollection.getWindowingStrategy();\n\n    if (UNBOUNDED.equals(pCollection.isBounded()) && !triggersOncePerWindow(windowingStrategy)) {\n      throw new UnsupportedOperationException(\n          \"Joining unbounded PCollections is currently only supported for \"\n              + \"non-global windows with triggers that are known to produce output once per window,\"\n              + \"such as the default trigger with zero allowed lateness. \"\n              + \"In these cases Beam can guarantee it joins all input elements once per window. \"\n              + windowingStrategy\n              + \" is not supported\");\n    }\n  }\n\n  private boolean triggersOncePerWindow(WindowingStrategy windowingStrategy) {\n    Trigger trigger = windowingStrategy.getTrigger();\n\n    return !(windowingStrategy.getWindowFn() instanceof GlobalWindows)\n        && trigger instanceof DefaultTrigger\n        && ZERO.equals(windowingStrategy.getAllowedLateness());\n  }\n\n  private PCollection<Row> standardJoin(","sourceCodeStart":125,"sourceCodeEnd":161,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCoGBKJoinRel.java#L125-L161","documentation":"For unbounded (streaming) PCollections, Beam's CoGBK-based SQL join can only guarantee that all elements of a window are seen together when windows are non-global and the trigger produces output exactly once per window (e.g. the default trigger with zero allowed lateness). verifySupportedTrigger checks each join input's WindowingStrategy and throws UnsupportedOperationException if streaming input does not meet this guarantee, since otherwise the join could emit partial/incorrect results.","triggerScenarios":"Running a SQL JOIN where at least one input is an unbounded PCollection whose WindowingStrategy fails triggersOncePerWindow: global window on a stream, custom triggers (e.g. Repeatedly/Count/AfterEach), non-zero allowed lateness, or accumulating/retracting modes that fire multiple times per window.","commonSituations":"Joining two streaming tables where one has a custom early/late trigger configured; joining a stream in the global window (no Window.into applied) with another stream; streams with allowed lateness > 0 or discarded/accumulating fire-on-update triggers set for side inputs.","solutions":["Use the default trigger with allowed lateness zero and a non-global window on both streaming inputs before the join","Replace custom triggers (Repeatedly.forever, AfterFirst/AfterEach early firings) with the default trigger, or accumulate to fire once per window, on the join inputs","If you need firing-on-update joins, implement the join manually with CoGroupByKey plus your own trigger semantics instead of the SQL join","Switch to a temporal join pattern (e.g. window both sides with the same fixed/sliding windows and default trigger) so each window fires exactly once"],"exampleFix":"// before\nstream.apply(Window.<Row>into(GlobalWindows()).triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(100))).discardingFiredPanes());\n// after\nstream.apply(Window.<Row>into(FixedWindows.of(Duration.standardMinutes(1)))); // default trigger, zero allowed lateness","handlingStrategy":"validation","validationCode":"static boolean joinableStreaming(PCollection<?> pc) {\n  WindowingStrategy s = pc.getWindowingStrategy();\n  return !GlobalWindows.INSTANCE.equals(s.getWindowFn())\n      && s.getAllowedLateness().isZero()\n      && DefaultTrigger.of().equals(s.getTrigger());\n}","typeGuard":null,"tryCatchPattern":"try {\n  result = joinTables(streamA, streamB);\n} catch (UnsupportedOperationException e) {\n  if (e.getMessage().contains(\"Joining unbounded PCollections\")) {\n    // re-apply default trigger / non-global windows and retry\n  } else { throw e; }\n}","preventionTips":["For streaming joins, use non-global windows with the default trigger and zero allowed lateness","Avoid custom or repeated triggers on PCollections fed into SQL joins","If you need early results, do the join manually with CoGroupByKey and your own trigger design","Assert on each input's WindowingStrategy (window fn, trigger, allowed lateness) before building the join"],"tags":["java","beam-sql","streaming","trigger","windowing","join"],"backgroundTag":"unsupported-operation","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"}