{"record":{"id":"47dadc3d7811aed5","repo":"apache/druid","slug":"toomanyrowswithsamekey","errorCode":"TooManyRowsWithSameKey","errorMessage":"TooManyRowsWithSameKeyFault: too many rows with same key %s (bytes %s > max %s)","messagePattern":"TooManyRowsWithSameKeyFault: too many rows with same key (.+?) \\(bytes (.+?) > max (.+?)\\)","errorType":"error_code","errorClass":"MSQException","httpStatus":null,"severity":"error","filePath":"multi-stage-query/src/main/java/org/apache/druid/msq/querykit/common/SortMergeJoinFrameProcessor.java","lineNumber":421,"sourceCode":"    if (awaitSet.isEmpty()) {\n      // No tracker reported that it \"needsMoreData\" to read the current cursor. However, we may still need to read\n      // more data to have a complete set for the current mark.\n      for (int i = 0; i < inputChannels.size(); i++) {\n        final Tracker tracker = trackers.get(i);\n        if (!tracker.hasCompleteSetForMark()) {\n          if (tracker.canBufferMoreFrames()) {\n            awaitSet.add(i);\n          } else if (trackerAtLimit < 0) {\n            trackerAtLimit = i;\n          }\n        }\n      }\n    }\n\n    if (awaitSet.isEmpty() && trackerAtLimit >= 0) {\n      // All trackers that need more data are at their max buffered bytes limit. Generate a nice exception.\n      final Tracker tracker = trackers.get(trackerAtLimit);\n      throw new MSQException(\n          new TooManyRowsWithSameKeyFault(\n              tracker.readMarkKey(),\n              tracker.totalBytesBuffered(),\n              maxBufferedBytes\n          )\n      );\n    }\n\n    return ReturnOrAwait.awaitAll(awaitSet);\n  }\n\n  /**\n   * Whether all trackers return true from {@link Tracker#isAtEnd()}.\n   */\n  private boolean allTrackersAreAtEnd()\n  {\n    for (Tracker tracker : trackers) {\n      if (!tracker.isAtEnd()) {","sourceCodeStart":403,"sourceCodeEnd":439,"githubUrl":"https://github.com/apache/druid/blob/9b90983fd291f26935af934383ce360473179e4d/multi-stage-query/src/main/java/org/apache/druid/msq/querykit/common/SortMergeJoinFrameProcessor.java#L403-L439","documentation":"The sort-merge join buffers all rows sharing the current join key from both inputs. When every tracker needing more data has hit maxBufferedBytes, nextAwait throws TooManyRowsWithSameKeyFault, since it cannot buffer the key's full row group within the byte limit and still produce a correct merge. This protects workers from unbounded memory use on skewed keys.","triggerScenarios":"A join key with more matching rows than fit in maxBufferedBytes (context setting, default ~100MiB total across trackers); detected in nextAwait when awaitSet is empty and trackerAtLimit >= 0 during runIncrementally.","commonSituations":"Joining on a low-cardinality key (e.g. country='US' matching millions of rows); data skew where one key dominates; queries copied with a small maxRowsWithSameKey/maxBufferedBytes context value.","solutions":["Increase the context parameter maxRowsWithSameKey / maxBufferedBytes for the query (memory permitting)","Choose a higher-cardinality join key or add columns to the join condition to reduce per-key row counts","Pre-aggregate one or both sides before the join to shrink per-key groups","Filter rows before joining so fewer duplicates share each key"],"exampleFix":"// before\n\"context\": {\"maxRowsWithSameKey\": 100000}\n// after\n\"context\": {\"maxRowsWithSameKey\": 1000000}","handlingStrategy":"validation","validationCode":"// Estimate max rows per join-key group vs maxRowsWithSameKey budget\nlong maxGroupRows = estimateMaxRowsPerKeyGroup(leftTable, rightTable, joinKey);\nlong maxRowsWithSameKey = context.get(\"maxRowsWithSameKey\", 100000);\nif (maxGroupRows > maxRowsWithSameKey) {\n  throw new IllegalStateException(\"Join key skew: \" + maxGroupRows + \" rows per key > limit \" + maxRowsWithSameKey);\n}","typeGuard":null,"tryCatchPattern":"try {\n  runMsqQuery(query);\n} catch (MSQException e) {\n  if (e.getFault() instanceof TooManyRowsWithSameKey) {\n    TooManyRowsWithSameKeyFault f = (TooManyRowsWithSameKeyFault) e.getFault();\n    // raise maxBufferedBytes/maxRowsWithSameKey above f.getBytes() or de-skew the key\n  }\n}","preventionTips":["Profile join key cardinality/distribution before big joins","Avoid joining on low-cardinality columns","Pre-aggregate the many-side of skewed joins"],"tags":["msq","join","data-skew","memory-limits"],"backgroundTag":"value-out-of-range","analyzedSha":"9b90983fd291f26935af934383ce360473179e4d","analyzedAt":"2026-09-07T13:32:30.957Z","contentChangedAt":"2026-09-07T13:32:30.957Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}