{"record":{"id":"74bc0046f9ffd87b","repo":"risingwavelabs/risingwave","slug":"udf-aggregate-what-has-rows-but-expected-exa","errorCode":null,"errorMessage":"UDF aggregate {what} has {} rows, but expected exactly 1","messagePattern":"UDF aggregate (.+?) has (.+?) rows, but expected exactly 1","errorType":"validation","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/expr/core/src/aggregate/user_defined.rs","lineNumber":128,"sourceCode":"    }\n\n    /// Decode the state from a datum in state table.\n    fn decode_state(&self, datum: Datum) -> Result<AggregateState> {\n        let array = {\n            let mut builder = DataType::Bytea.create_array_builder(1);\n            builder.append(datum);\n            builder.finish()\n        };\n        let state = UdfArrowConvert::default().to_array(self.state_field.data_type(), &array)?;\n        Ok(AggregateState::Any(Box::new(State(state))))\n    }\n}\n\n/// The runtime is external input: reject an array of the wrong length so that `datum_at(0)` at\n/// the call sites is in bounds.\nfn ensure_single_row(array: &ArrayRef, what: &str) -> Result<()> {\n    if array.len() != 1 {\n        return Err(anyhow::anyhow!(\n            \"UDF aggregate {what} has {} rows, but expected exactly 1\",\n            array.len()\n        )\n        .into());\n    }\n    Ok(())\n}\n\n// In arrow-udf, aggregate state is represented as an `ArrayRef`.\n// To avoid unnecessary conversion between `ArrayRef` and `Datum`,\n// we store `ArrayRef` directly in our `AggregateState`.\n#[derive(Debug)]\nstruct State(ArrayRef);\n\nimpl EstimateSize for State {\n    fn estimated_heap_size(&self) -> usize {\n        self.0.get_array_memory_size()\n    }","sourceCodeStart":110,"sourceCodeEnd":146,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/expr/core/src/aggregate/user_defined.rs#L110-L146","documentation":"`ensure_single_row` guards UDF aggregate state/result arrays: both must contain exactly one row because call sites read via `datum_at(0)`. If the external UDF runtime returns an array with 0 or >1 rows, this error prevents an out-of-bounds read.","triggerScenarios":"`get_result` or `encode_state` on a UserDefinedAggregate receives an ArrayRef whose len() != 1 — typically a UDF server returning an empty batch or multiple rows for the merged state/result.","commonSituations":"UDF implementation of `merge`/`state` accumulates multiple rows; UDF server misbehaves on empty input; protocol drift where the server streams partial batches.","solutions":["Fix the UDF implementation to always return exactly one row for state and result","Check UDF server behavior for empty/degenerate inputs and return a single null row instead of zero rows","Inspect the UDF server logs to see why the batch length is not 1","Upgrade/sync RisingWave and UDF service versions"],"exampleFix":"// before (UDF returns empty array for empty input)\nInt64Array::from(Vec::<i64>::new())\n// after\nInt64Array::from(vec![None]) // single null row preserves arity","handlingStrategy":"validation","validationCode":"if result_array.len() != 1 {\n    // reject before calling encode_state/get_result-dependent code\n}","typeGuard":"fn is_single_row(a: &arrow::array::ArrayRef) -> bool { a.len() == 1 }","tryCatchPattern":"match agg.get_result().await {\n    Ok(v) => v,\n    Err(e) if e.to_string().contains(\"expected exactly 1\") => {\n        // treat as UDF server contract violation; surface to user\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Ensure UDF state/merge functions always return exactly one row","Test UDF servers against empty and multi-batch inputs","Document the single-row contract in UDF SDKs"],"tags":["udf","arrow","shape-mismatch","rust"],"backgroundTag":"unexpected-response-shape","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"}