{"record":{"id":"0e4886fb19794e65","repo":"prestodb/presto","slug":"input-not-found-for-sourcefragmentid-sourcefrag","errorCode":null,"errorMessage":"Input not found for sourceFragmentId: ${sourceFragmentId}","messagePattern":"Input not found for sourceFragmentId: (.+?)","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"presto-spark-base/src/main/java/com/facebook/presto/spark/execution/task/PrestoSparkTaskExecutorFactory.java","lineNumber":768,"sourceCode":"                }\n\n                if (broadcastInput != null) {\n                    checkArgument(inMemoryInput == null, \"single remote source is not expected to accept different kind of inputs\");\n                    // TODO: Enable NullifyingIterator once migrated to one task per JVM model\n                    // NullifyingIterator removes element from the list upon return\n                    // This allows GC to gradually reclaim memory\n                    // remoteSourcePageInputs.add(getNullifyingIterator(broadcastInput.value()));\n                    broadcastInputsListBuilder.add((List<?>) broadcastInput.value());\n                    continue;\n                }\n\n                if (inMemoryInput != null) {\n                    // for in-memory inputs pages can be released incrementally to save memory\n                    remoteSourcePageInputsBuilder.add(getNullifyingIterator(inMemoryInput));\n                    continue;\n                }\n\n                throw new IllegalStateException(\"Input not found for sourceFragmentId: \" + sourceFragmentId);\n            }\n            List<PrestoSparkShuffleInput> remoteSourceRowInputs = remoteSourceRowInputsBuilder.build();\n            List<java.util.Iterator<PrestoSparkSerializedPage>> remoteSourcePageInputs = remoteSourcePageInputsBuilder.build();\n            List<List<?>> broadcastInputsList = broadcastInputsListBuilder.build();\n            if (!remoteSourceRowInputs.isEmpty()) {\n                shuffleInputs.put(remoteSource.getId(), remoteSourceRowInputs);\n            }\n            if (!remoteSourcePageInputs.isEmpty()) {\n                pageInputs.put(remoteSource.getId(), remoteSourcePageInputs);\n            }\n            if (!broadcastInputsList.isEmpty()) {\n                broadcastInputs.put(remoteSource.getId(), broadcastInputsList);\n            }\n        }\n    }\n\n    private List<TaskSource> getTaskSources(Iterator<SerializedPrestoSparkTaskSource> serializedTaskSources)\n    {","sourceCodeStart":750,"sourceCodeEnd":786,"githubUrl":"https://github.com/prestodb/presto/blob/55bb57d202de3b926896fa966c2c4a44c779634e/presto-spark-base/src/main/java/com/facebook/presto/spark/execution/task/PrestoSparkTaskExecutorFactory.java#L750-L786","documentation":"PrestoSparkTaskExecutorFactory.fillJavaExecutionTaskInputs resolves each remote source (downstream fragment input) by sourceFragmentId in the provided shuffle/broadcast/in-memory input maps. When a fragment's input has no matching entry, it throws IllegalStateException. This means the task inputs RDD does not cover all remote sources the plan fragment expects.","triggerScenarios":"doCreate -> fillJavaExecutionTaskInputs is called with PrestoSparkTaskInputs whose inputs map lacks an entry for a sourceFragmentId referenced by the fragment's RemoteSourceNodes (and the input is not an in-memory input either).","commonSituations":"Coordinator/executor version skew so fragment IDs differ; custom RDD factory or input providers missing an input; a query plan change that added a remote source not covered by cached RDDs.","solutions":["Ensure coordinator and Spark workers run the same Presto version — mismatched plans cause fragment ID mismatch","Check that PrestoSparkRddFactory produced inputs for every remote source of the fragment (inspect rddInputs construction)","Clear any cached/stale RDD inputs and re-run the query","If using custom input provisioning, verify all sourceFragmentIds from the fragment are populated in the inputs map"],"exampleFix":"// before\nthrow new IllegalStateException(\"Input not found for sourceFragmentId: \" + sourceFragmentId);\n// after (fail with diagnostic)\nthrow new IllegalStateException(\"Input not found for sourceFragmentId: \" + sourceFragmentId\n    + \", available fragment ids: \" + inputs.getSourceFragmentIds());","handlingStrategy":"validation","validationCode":"Set<String> available = inputs.getSourceFragmentIds();\nfor (RemoteSourceNode source : fragment.getRemoteSourceNodes()) {\n    checkState(available.contains(source.getId().toString()), \"missing input for \" + source.getId());\n}","typeGuard":"boolean inputExists(Map<String, ?> inputs, String sourceFragmentId) { return inputs != null && inputs.containsKey(sourceFragmentId); }","tryCatchPattern":"try { fillJavaExecutionTaskInputs(...); } catch (IllegalStateException e) { if (e.getMessage().startsWith(\"Input not found\")) { log.error(\"fragment id mismatch or missing rdd input\", e); } throw e; }","preventionTips":["Run identical Presto versions on coordinator and Spark workers","Validate rddInputs cover all remote sources before task submission","Avoid caching RDD inputs across plan-changing upgrades"],"tags":["presto-on-spark","illegal-state","missing-input","shuffle-input"],"backgroundTag":"missing-shuffle-input-for-fragment","analyzedSha":"55bb57d202de3b926896fa966c2c4a44c779634e","analyzedAt":"2026-09-04T12:50:26.162Z","contentChangedAt":"2026-09-04T12:50:26.162Z","schemaVersion":2},"datasetVersion":"2026-09-11T21:17:09.523Z"}