{"record":{"id":"884f4118583db7c6","repo":"apache/beam","slug":"windowfns-must-match-for-a-bounded-vs-bounded-unbounded-vs","errorCode":null,"errorMessage":"WindowFns must match for a bounded-vs-bounded/unbounded-vs-unbounded join.","messagePattern":"WindowFns must match for a bounded-vs-bounded/unbounded-vs-unbounded join\\.","errorType":"exception","errorClass":"java.lang.IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamCoGBKJoinRel.java","lineNumber":128,"sourceCode":"                  \"right_TimestampCombiner\",\n                  Window.<Row>configure().withTimestampCombiner(TimestampCombiner.EARLIEST));\n\n      // extract the join fields\n      List<Pair<RexNode, RexNode>> pairs = extractJoinRexNodes(condition);\n      int leftRowColumnCount = BeamSqlRelUtils.getBeamRelInput(left).getRowType().getFieldCount();\n\n      FieldAccessDescriptor leftKeyFields =\n          BeamJoinTransforms.getJoinColumns(true, pairs, 0, leftSchema);\n      FieldAccessDescriptor rightKeyFields =\n          BeamJoinTransforms.getJoinColumns(false, pairs, leftRowColumnCount, rightSchema);\n\n      WindowFn leftWinFn = leftRows.getWindowingStrategy().getWindowFn();\n      WindowFn rightWinFn = rightRows.getWindowingStrategy().getWindowFn();\n\n      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. \"","sourceCodeStart":110,"sourceCodeEnd":146,"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#L110-L146","documentation":"BeamCoGBKJoinRel.expand joins two PCollections via CoGroupByKey, which requires both inputs to use compatible windowing. It calls WindowFn.verifyCompatibility on the left and right window functions and, if they are incompatible, throws IllegalArgumentException explaining that window functions must match for a join. This guarantees elements land in the same windows on both sides so the join is well-defined.","triggerScenarios":"Executing a Beam SQL JOIN where one side of the join uses a different WindowFn than the other — e.g. fixed windows of 1 minute vs 5 minutes, or sliding vs fixed windows, or global window vs fixed windows — on bounded or unbounded PCollections.","commonSituations":"Joining a streaming table (windowed with WithTimestamps/Window.into) with a bounded side table that stayed in the GlobalWindows; two streams windowed with different durations or window types being joined in SQL; changing one input's windowing after building the pipeline.","solutions":["Apply the same WindowFn to both join inputs before the join (e.g. Window.into(FixedWindows.of(Duration.standardMinutes(1))) on both sides)","If one side is a bounded lookup table, window it with the same WindowFn as the streaming side (or convert the join to use global windows only if both are bounded and it's semantically safe)","Reorder the pipeline so windowing is applied before the SQL JOIN transform","Check WindowingStrategy of both PCollections (getWindowingStrategy().getWindowFn()) and make verifyCompatibility pass before invoking the SQL join"],"exampleFix":"// before\nPCollection<Row> right = rows.apply(Window.into(FixedWindows.of(Duration.standardMinutes(5))));\n// after (match the left side)\nPCollection<Row> right = rows.apply(Window.into(FixedWindows.of(Duration.standardMinutes(1))));","handlingStrategy":"validation","validationCode":"try {\n  left.getWindowingStrategy().getWindowFn().verifyCompatibility(right.getWindowingStrategy().getWindowFn());\n} catch (IncompatibleWindowException e) {\n  throw new IllegalStateException(\"Join inputs use incompatible WindowFns: \" + e.getMessage());\n}","typeGuard":null,"tryCatchPattern":"try {\n  return sqlEnv.executeJoin(left, right);\n} catch (IllegalArgumentException e) {\n  if (e.getMessage().contains(\"WindowFns must match\")) {\n    // re-window both sides with the same WindowFn and retry\n  } else { throw e; }\n}","preventionTips":["Always apply the same Window.into(...) to both sides of a SQL join on windowed data","Check getWindowingStrategy().getWindowFn() equality/compatibility of both PCollections before joining","Avoid mixing bounded (global window) and windowed streaming inputs in joins","Centralize window configuration so all joins derive their WindowFn from one constant"],"tags":["java","beam-sql","windowing","join","streaming"],"backgroundTag":"incompatible-window-fn","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}