{"record":{"id":"0410b28d9943936d","repo":"apache/druid","slug":"broadcasttablestoolarge","errorCode":"BroadcastTablesTooLarge","errorMessage":"BroadcastTablesTooLargeFault: broadcast tables exceed reserved memory %s","messagePattern":"BroadcastTablesTooLargeFault: broadcast tables exceed reserved memory (.+?)","errorType":"error_code","errorClass":"MSQException","httpStatus":null,"severity":"error","filePath":"multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BroadcastJoinSegmentMapFnProcessor.java","lineNumber":252,"sourceCode":"   * Reads up to one frame from each readable side channel, and uses them to incrementally build up joinable\n   * broadcast tables.\n   *\n   * @param readableInputs all readable input channel numbers, including non-side-channels\n   * @return whether side channels have been fully read\n   */\n  boolean buildBroadcastTablesIncrementally(final IntSet readableInputs)\n  {\n    final IntIterator inputChannelIterator = readableInputs.iterator();\n\n    while (inputChannelIterator.hasNext()) {\n      final int channelNumber = inputChannelIterator.nextInt();\n      if (sideChannelNumbers.contains(channelNumber) && channels.get(channelNumber).canRead()) {\n        final Frame frame = channels.get(channelNumber).readFrame();\n\n        memoryUsed += frame.numBytes();\n\n        if (memoryUsed > memoryReservedForBroadcastJoin) {\n          throw new MSQException(\n              new BroadcastTablesTooLargeFault(\n                  memoryReservedForBroadcastJoin,\n                  Optional.ofNullable(query)\n                          .map(q -> q.context().getString(PlannerContext.CTX_SQL_JOIN_ALGORITHM))\n                          .map(JoinAlgorithm::fromString)\n                          .orElse(null)\n              )\n          );\n        }\n\n        addFrame(channelNumber, frame);\n      }\n    }\n\n    for (int channelNumber : sideChannelNumbers) {\n      if (!channels.get(channelNumber).isFinished()) {\n        return false;\n      }","sourceCodeStart":234,"sourceCodeEnd":270,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/BroadcastJoinSegmentMapFnProcessor.java#L234-L270","documentation":"Raised when the total byte size of broadcast tables read into memory during a broadcast join exceeds the memory reserved for the broadcast join (BroadcastTablesTooLargeFault wrapped in MSQException). Broadcast joins materialize the entire right-hand tables on every worker, so Druid enforces a reserved-memory cap and aborts the query rather than risk heap exhaustion.","triggerScenarios":"Running a broadcast (hash) join in MSQ where the sum of broadcast frame sizes read in BroadcastTablesTooLargeFault.buildBroadcastTablesIncrementally exceeds memoryReservedForBroadcastJoin (from memory reserve config / cluster capacity).","commonSituations":"Joining large dimension tables that were assumed small; context join algorithm set to broadcast inadvertently (default for equi-joins with small tables); cluster with low memory reserve for broadcast joins.","solutions":["Reduce the size of the broadcast side of the join (filter rows/columns before joining).","Force a sort-merge (partitioned) join instead: set \"sqlJoinAlgorithm\":\"sortMerge\" in the query context.","Increase memory reserved for broadcast joins (druid.msq.memory.reservedForBroadcastJoin / stage memory tuning) or add more workers so per-worker load drops.","Pre-materialize and downcast the small table, or filter the large table upstream."],"exampleFix":"// before\nSELECT * FROM big_events t JOIN dims d ON t.dim_id = d.id; // broadcast join\n// after\nSELECT * FROM big_events t\nJOIN dims d ON t.dim_id = d.id\n-- context: {\"sqlJoinAlgorithm\": \"sortMerge\"}","handlingStrategy":"try-catch","validationCode":"// estimate broadcast size before query: sum of bytes of small-side tables\nlong broadcastBytes = estimateTableBytes(smallSideTables);\nif (broadcastBytes > reservedBroadcastMemory) { /* switch to sortMerge */ }","typeGuard":null,"tryCatchPattern":"try {\n  result = runMsqQuery(query);\n} catch (MSQException e) {\n  if (e.getFault() instanceof BroadcastTablesTooLargeFault) {\n    query.context().put(\"sqlJoinAlgorithm\", \"sortMerge\");\n    result = runMsqQuery(query); // retry with partitioned join\n  } else throw e;\n}","preventionTips":["Keep broadcast-side tables small (filter rows and columns).","Set sqlJoinAlgorithm=sortMerge when join inputs may be large.","Size memoryReservedForBroadcastJoin to your largest expected dimension table."],"tags":["druid","msq","join","broadcast","memory-limit"],"backgroundTag":"payload-too-large","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}