{"record":{"id":"af310f81c2b9f2b6","repo":"risingwavelabs/risingwave","slug":"lancedb-pre-commit-sink-id-does-not-match-coord","errorCode":null,"errorMessage":"LanceDB pre-commit sink id {} does not match coordinator sink id {}","messagePattern":"LanceDB pre-commit sink id (.+?) does not match coordinator sink id (.+?)","errorType":"exception","errorClass":"SinkError::LanceDb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/lancedb.rs","lineNumber":757,"sourceCode":"    }\n\n    async fn commit_data(&mut self, epoch: u64, commit_metadata: Vec<u8>) -> Result<()> {\n        tracing::debug!(\"Starting LanceDB two-phase commit in epoch {epoch}.\");\n\n        if commit_metadata.is_empty() {\n            return Ok(());\n        }\n\n        let pre_commit_metadata = LanceDbPreCommitMetadata::try_from_bytes(&commit_metadata)?;\n        if pre_commit_metadata.epoch != epoch {\n            return Err(SinkError::LanceDb(anyhow!(\n                \"LanceDB pre-commit epoch {} does not match commit epoch {}\",\n                pre_commit_metadata.epoch,\n                epoch\n            )));\n        }\n        if pre_commit_metadata.sink_id != self.sink_id {\n            return Err(SinkError::LanceDb(anyhow!(\n                \"LanceDB pre-commit sink id {} does not match coordinator sink id {}\",\n                pre_commit_metadata.sink_id,\n                self.sink_id\n            )));\n        }\n\n        let transaction_properties = pre_commit_metadata.transaction_properties();\n        self.commit_fragments(\n            epoch,\n            pre_commit_metadata.fragments,\n            Some(transaction_properties),\n        )\n        .await\n    }\n\n    async fn abort(&mut self, epoch: u64, _commit_metadata: Vec<u8>) {\n        // Unreferenced files are reclaimed by Lance's old-version/orphan cleanup. This is\n        // intentionally not an eager delete: the commit may have succeeded even if its result","sourceCodeStart":739,"sourceCodeEnd":775,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/lancedb.rs#L739-L775","documentation":"Same commit_data path as the epoch check: after deserializing LanceDbPreCommitMetadata, the sink verifies the sink_id recorded at pre_commit matches this coordinator instance's sink_id. A mismatch means the prepared state belongs to a different sink instance — stale metadata from a dropped/recreated sink or two sinks sharing the same state slot.","triggerScenarios":"A sink with the same underlying LanceDB storage (same pre-commit bytes) but a different sink_id calls commit_data — typically after dropping and recreating a sink that reuses the same table/state key, or two sinks pointing at the same state.","commonSituations":"Recreating a LanceDB sink without clearing its old pre-commit metadata; misconfigured sinks sharing the same table URI/state key.","solutions":["Drop and recreate the sink with clean state (remove stale pre-commit metadata or use a fresh sink state/table)","Ensure the recreated sink uses a distinct table or clear the old metadata so it does not inherit the previous sink_id","Verify no two active sinks are configured to write to the same LanceDB table/state"],"exampleFix":"// before\nCREATE SINK s2 AS SELECT ... INTO same_lancedb_table ...; // stale pre-commit sink_id mismatch\n// after\nDROP SINK s2;\n-- clear old state / use a fresh table\nCREATE SINK s2 AS SELECT ... INTO fresh_lancedb_table ...;","handlingStrategy":"validation","validationCode":"let meta = LanceDbPreCommitMetadata::try_from_bytes(&commit_metadata)?;\nif meta.sink_id != self.sink_id {\n    // stale metadata from a previous sink instance; treat as invalid prepared state\n}","typeGuard":"fn is_sink_id_mismatch(err: &SinkError) -> bool {\n    matches!(err, SinkError::LanceDb(e) if e.to_string().contains(\"does not match coordinator sink id\"))\n}","tryCatchPattern":"match commit_result {\n    Err(e) if e.to_string().contains(\"sink id\") => {\n        // drop stale prepared state and re-prepare at the current epoch\n    }\n    other => other,\n}","preventionTips":["Never let two sink instances share the same LanceDB table + state key","After dropping a sink, clear its persisted pre-commit metadata before recreating with the same storage","Use fresh sink identifiers/tables when recreating a LanceDB sink"],"tags":["lancedb","sink","sink-id-mismatch","two-phase-commit","stale-metadata"],"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"}