{"record":{"id":"fd5c367ea24e8237","repo":"quickwit-oss/quickwit","slug":"execute-merge-operation-sources-len-op","errorCode":null,"errorMessage":"execute_merge_operation: sources.len() ({}) != op.splits.len() ({})","messagePattern":"execute_merge_operation: sources\\.len\\(\\) \\((.+?)\\) != op\\.splits\\.len\\(\\) \\((.+?)\\)","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-parquet-engine/src/merge/mod.rs","lineNumber":522,"sourceCode":"/// range reads against `op.splits[i].parquet_file`. The caller (e.g.\n/// the executor wrapper that lives outside this crate) is responsible\n/// for materializing one [`RemoteByteSource`] per split based on its\n/// storage backend (S3, local FS, etc.). Splits with names that\n/// cannot be opened by the source surface as `LegacyAdapterError::Io`\n/// or `ParquetReadError`.\n///\n/// Returns the merge engine's [`MergeOutputFile`]s. Conversion to\n/// `ParquetSplitMetadata` for the metastore is the caller's\n/// responsibility — use [`metadata_aggregation::merge_parquet_split_metadata`]\n/// with `mixed_prefix_ok = op.target_prefix_len_override.is_some()`.\npub async fn execute_merge_operation(\n    op: &policy::ParquetMergeOperation,\n    sources: Vec<Arc<dyn RemoteByteSource>>,\n    output_dir: &Path,\n    config: &MergeConfig,\n) -> Result<Vec<MergeOutputFile>> {\n    if sources.len() != op.splits.len() {\n        bail!(\n            \"execute_merge_operation: sources.len() ({}) != op.splits.len() ({})\",\n            sources.len(),\n            op.splits.len(),\n        );\n    }\n\n    let mut streams: Vec<Box<dyn ColumnPageStream>> = Vec::with_capacity(op.splits.len());\n    for (split, source) in op.splits.iter().zip(sources) {\n        let path = PathBuf::from(&split.parquet_file);\n        let stream: Box<dyn ColumnPageStream> = match op.target_prefix_len_override {\n            Some(target) if split.rg_partition_prefix_len < target => {\n                // Promote this legacy input. The adapter re-encodes in\n                // memory and presents itself as a prefix_len = target\n                // single-RG stream to the merge engine.\n                let adapter = LegacyInputAdapter::try_open(source, path, target)\n                    .await\n                    .with_context(|| {\n                        format!(","sourceCodeStart":504,"sourceCodeEnd":540,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-parquet-engine/src/merge/mod.rs#L504-L540","documentation":"execute_merge_operation enforces that the number of byte sources passed in exactly matches the number of splits declared in the ParquetMergeOperation. Each split must have exactly one corresponding remote source; a mismatch means the caller built an inconsistent merge task, so it bails rather than silently merging fewer or more files than planned.","triggerScenarios":"Calling execute_merge_operation (or the merge handler that wraps it) where sources.len() != op.splits.len() — e.g. a download step skipped or duplicated a source, or splits were added to the op after sources were materialized.","commonSituations":"A merge executor whose fetch stage filters out un-fetchable files without updating the operation; unit/integration harnesses (assert_engine_parity, tests) constructing the op and source list independently.","solutions":["Ensure one Arc<dyn RemoteByteSource> is produced per split, in the same order, before calling execute_merge_operation.","Fix the download/fetch stage so failures abort the merge instead of dropping sources.","Build sources from op.splits itself rather than a separately maintained list.","If this appears in tests, update the test harness to derive sources from the op."],"exampleFix":"// before: sources collected independently of op.splits\nlet sources = fetch_all(pending_files).await?;\nexecute_merge_operation(&op, sources, &out, &cfg).await?;\n// after: derive sources from the op's splits\nlet sources: Vec<_> = op.splits.iter().map(|s| fetch_source(s)).collect::<Result<_>>()?;\nexecute_merge_operation(&op, sources, &out, &cfg).await?;","handlingStrategy":"validation","validationCode":"anyhow::ensure!(sources.len() == op.splits.len(),\n    \"sources ({}) must map 1:1 to op.splits ({})\", sources.len(), op.splits.len());","typeGuard":null,"tryCatchPattern":"match execute_merge_operation(&op, sources, &out, &cfg).await {\n    Err(e) if e.to_string().contains(\"sources.len()\") => {\n        // refetch sources from op.splits and retry once\n    }\n    other => other?,\n}","preventionTips":["Always derive the source list from op.splits, never from a separate collection.","Treat per-file fetch failures as merge failures rather than silently skipping.","Assert 1:1 mapping in merge executor tests."],"tags":["merge","invariant","argument-count"],"backgroundTag":"internal-invariant-violation","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}