{"record":{"id":"b73ef1e0255c8f13","repo":"risingwavelabs/risingwave","slug":"dataset-guard-acquisition-error","errorCode":null,"errorMessage":"(dataset guard acquisition error)","messagePattern":"\\(dataset guard acquisition error\\)","errorType":"exception","errorClass":"SinkError::LanceDb","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/lancedb.rs","lineNumber":376,"sourceCode":"}\n\nimpl LanceDbSinkWriter {\n    pub async fn new(config: LanceDbConfig, schema: Schema) -> Result<Self> {\n        let arrow_schema = LanceDbConvert\n            .rw_schema_to_arrow_schema(&schema)\n            .map_err(|e| SinkError::LanceDb(anyhow!(e)))?;\n\n        let conn = config.common.create_connection().await?;\n        let table = config.common.open_table(&conn).await?;\n        let dataset_wrapper = table.dataset().ok_or_else(|| {\n            SinkError::LanceDb(anyhow!(\n                \"failed to get underlying lance Dataset (table may be remote)\"\n            ))\n        })?;\n        let dataset_guard = dataset_wrapper\n            .get()\n            .await\n            .map_err(|e| SinkError::LanceDb(anyhow!(e)))?;\n        let data_storage_version = dataset_guard\n            .manifest\n            .data_storage_format\n            .lance_file_version()\n            .context(\"failed to get LanceDB table storage version\")\n            .map_err(SinkError::LanceDb)?;\n        let store_params =\n            dataset_guard\n                .storage_options_accessor()\n                .map(|storage_options_accessor| ObjectStoreParams {\n                    storage_options_accessor: Some(storage_options_accessor),\n                    ..Default::default()\n                });\n        drop(dataset_guard);\n        let dataset_uri = config.common.dataset_uri(&table).await?;\n\n        Ok(Self {\n            config,","sourceCodeStart":358,"sourceCodeEnd":394,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/lancedb.rs#L358-L394","documentation":"After obtaining the shared Lance dataset wrapper in `LanceDbSinkWriter::new`, the writer awaits `dataset_wrapper.get()` to acquire a guard on the underlying dataset (which may require loading/refreshing it from storage). If that acquisition fails — e.g. I/O error reading the manifest, credentials rejected by the object store, or internal lancedb error — it is wrapped in `SinkError::LanceDb` with no added context message, so it surfaces as the raw inner error.","triggerScenarios":"`dataset_wrapper.get().await` returns Err while initializing the sink writer: object-store auth failure (bad/missing AWS/GCS/Azure credentials), network partition to storage, corrupted or unreadable _manifest/_versions files, or rate limiting from the object store during dataset load.","commonSituations":"Missing or expired cloud storage credentials in the sink config; bucket region mismatch; IAM policy lacking s3:GetObject; firewall/proxy blocking storage endpoint; dataset files deleted or corrupted by an external process.","solutions":["Check and fix the storage credentials/options in the LanceDB sink config (access keys, tokens, region) and retest.","Verify network reachability from the RisingWave node to the object store endpoint (DNS, firewall, proxy).","Inspect the wrapped error message in the SinkError chain to identify whether it is auth, I/O, or manifest corruption, and address that specific cause.","Validate the dataset is readable independently (e.g. open it with a small lancedb script or `lance` CLI) to rule out data corruption."],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// Pre-flight: verify the dataset is reachable with current credentials\nlet table = config.common.open_table(&conn).await?;\nlet wrapper = table.dataset().ok_or_else(|| anyhow!(\"no native dataset\"))?;\nwrapper.get().await.map_err(|e| {\n    anyhow!(\"pre-flight dataset load failed, check storage credentials/network: {e}\")\n})?;","typeGuard":null,"tryCatchPattern":"// Acquire the dataset guard with bounded retries for transient storage errors\nfor attempt in 0..3 {\n    match dataset_wrapper.get().await {\n        Ok(guard) => break guard,\n        Err(e) if attempt < 2 && is_transient(&e) => {\n            tokio::time::sleep(Duration::from_millis(250 * (1 << attempt))).await;\n        }\n        Err(e) => return Err(SinkError::LanceDb(anyhow!(e).context(\"dataset guard acquisition\"))),\n    }\n}","preventionTips":["Validate cloud storage credentials and region before starting the sink","Grant the sink's IAM role object-level read access (e.g. s3:GetObject) on the dataset prefix","Monitor object-store error rates; alert on auth failures","Keep dataset directories untouched by external cleanup tools while the sink runs"],"tags":["lancedb","rust","storage","object-store","io"],"backgroundTag":"network-request-failed","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}