{"record":{"id":"5c4c42f674a8ad8e","repo":"risingwavelabs/risingwave","slug":"should-not-get-split-info-from-unpartitioned-sourc","errorCode":null,"errorMessage":"Should not get split info from unpartitioned source scan info","messagePattern":"Should not get split info from unpartitioned source scan info","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/frontend/src/scheduler/plan_fragmenter.rs","lineNumber":376,"sourceCode":"        Self::Incomplete(fetch_info)\n    }\n\n    pub async fn complete(self, batch_parallelism: usize) -> SchedulerResult<Self> {\n        match self {\n            SourceScanInfo::Incomplete(fetch_info) => fetch_info.complete(batch_parallelism).await,\n            SourceScanInfo::Unpartitioned(data) => data.complete(batch_parallelism),\n            SourceScanInfo::Complete(_) => {\n                unreachable!(\"Never call complete when SourceScanInfo is already complete\")\n            }\n        }\n    }\n\n    pub fn split_info(&self) -> SchedulerResult<&Vec<SplitImpl>> {\n        match self {\n            Self::Incomplete(_) => Err(SchedulerError::Internal(anyhow!(\n                \"Should not get split info from incomplete source scan info\"\n            ))),\n            Self::Unpartitioned(_) => Err(SchedulerError::Internal(anyhow!(\n                \"Should not get split info from unpartitioned source scan info\"\n            ))),\n            Self::Complete(split_info) => Ok(split_info),\n        }\n    }\n}\n\nimpl UnpartitionedData {\n    fn complete(self, batch_parallelism: usize) -> SchedulerResult<SourceScanInfo> {\n        let splits = match self {\n            UnpartitionedData::Iceberg { task, limit } => {\n                IcebergScanTaskPlanner::plan_splits(task, batch_parallelism, limit)?\n                    .into_iter()\n                    .map(SplitImpl::Iceberg)\n                    .collect()\n            }\n        };\n        Ok(SourceScanInfo::Complete(splits))","sourceCodeStart":358,"sourceCodeEnd":394,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/frontend/src/scheduler/plan_fragmenter.rs#L358-L394","documentation":"`SourceScanInfo::split_info` (src/frontend/src/scheduler/plan_fragmenter.rs:376) errors when called on the `Unpartitioned` variant. Unpartitioned sources have no per-split parallelism, so there is no `Vec<SplitImpl>` to return; only `Complete` sources expose split info.","triggerScenarios":"Calling `split_info()` on a source scan info produced for an unpartitioned source (e.g. Datagen, or single-node sources) instead of a partitioned source like Kafka or an Iceberg/File scan.","commonSituations":"Scheduler code paths assuming all source scans are partitionable, executed against sources like datagen or system sources where parallel split scheduling does not apply.","solutions":["Check `SourceScanInfo` variant before calling `split_info()` and handle `Unpartitioned` with a parallelism-1 schedule.","Verify the source kind supports partitioned batch reads; use a table/MV over the source if you need queryable splits.","Fix the caller (fragmenter/scheduler) to branch on the enum state instead of unconditionally requesting splits."],"exampleFix":"// before\nlet splits = source_scan_info.split_info()?;\n// after\nlet splits = match &source_scan_info {\n    SourceScanInfo::Complete(_) => source_scan_info.split_info()?,\n    SourceScanInfo::Unpartitioned(_) | SourceScanInfo::Incomplete(_) => {\n        Default::default() // no parallel splits\n    }\n};","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"fn splits_available(info: &SourceScanInfo) -> bool {\n    matches!(info, SourceScanInfo::Complete(_))\n}","tryCatchPattern":"let splits = match &source_scan_info {\n    SourceScanInfo::Complete(_) => source_scan_info.split_info()?,\n    SourceScanInfo::Unpartitioned(_) => return Ok(parallelism_one_schedule()),\n    SourceScanInfo::Incomplete(_) => return Err(anyhow!(\"incomplete scan info\")),\n};","preventionTips":["Branch on SourceScanInfo variant when scheduling source scans.","Know which connector kinds are unpartitioned (e.g. datagen) and schedule them with parallelism 1.","Add exhaustiveness tests over the enum in scheduler code."],"tags":["scheduler","plan-fragmenter","source"],"backgroundTag":"invalid-state-transition","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}