{"record":{"id":"c15726d221a77dff","repo":"risingwavelabs/risingwave","slug":"iceberg-metadata-scan-received-a-non-iceberg-conne","errorCode":null,"errorMessage":"Iceberg metadata scan received a non-Iceberg connector","messagePattern":"Iceberg metadata scan received a non-Iceberg connector","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/batch/executors/src/executor/iceberg_metadata_scan.rs","lineNumber":109,"sourceCode":"            MetadataType::Unspecified => {\n                return Err(anyhow!(\"Iceberg metadata type is unspecified\").into());\n            }\n        };\n        let time_travel_info = node\n            .time_travel\n            .as_ref()\n            .map(|time_travel| match time_travel {\n                TimeTravel::SnapshotId(snapshot_id) => IcebergTimeTravelInfo::Version(*snapshot_id),\n                TimeTravel::TimestampMs(timestamp_ms) => {\n                    IcebergTimeTravelInfo::TimestampMs(*timestamp_ms)\n                }\n            });\n        let config = ConnectorProperties::extract(\n            WithOptionsSecResolved::new(node.with_properties.clone(), node.secret_refs.clone()),\n            false,\n        )?;\n        let ConnectorProperties::Iceberg(properties) = config else {\n            return Err(anyhow!(\"Iceberg metadata scan received a non-Iceberg connector\").into());\n        };\n\n        Ok(Box::new(IcebergMetadataScanExecutor {\n            schema: metadata_type.schema(),\n            properties: *properties,\n            metadata_type,\n            time_travel_info,\n            identity: source.plan_node().get_identity().clone(),\n            chunk_size: source.context().get_config().developer.chunk_size,\n        }))\n    }\n}\n","sourceCodeStart":91,"sourceCodeEnd":122,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/batch/executors/src/executor/iceberg_metadata_scan.rs#L91-L122","documentation":"`deserialize_stream_chunk` is only meant to decode data rows; a stored `LogStoreOp::Barrier` inside the requested `[start_seq_id, end_seq_id)` seq-id range is not decodable into a stream chunk, so the decoder fails fast. Barriers must be handled by the barrier-specific decode path instead.","triggerScenarios":"Calling `deserialize_stream_chunk` on a seq-id range that includes a written `LogStoreOp::Barrier` entry (matched by the `(_, LogStoreOp::Barrier { .. })` arm). Thrown at serde.rs:509.","commonSituations":"Reader range-selection logic including barrier seq ids when fetching data rows; races where a barrier is written between computing start_seq_id and end_seq_id; callers using the chunk decoder on a mixed range during recovery replay.","solutions":["Exclude barrier entries from the read range (stop end_seq_id before the barrier's seq id).","Route barrier ops to the barrier-decoding path (`next_op` / barrier alignment) instead of `deserialize_stream_chunk`.","Check seq-id bookkeeping so data reads never overlap barrier writes.","If this happens during recovery, ensure replay alternates correctly between data ranges and barriers."],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// rust\n// stop the data range before the first barrier seq id\nlet end = end_seq_id.min(next_barrier_seq_id - 1);\nanyhow::ensure!(end > start_seq_id, \"no data rows between {} and barrier at {}\", start_seq_id, next_barrier_seq_id);","typeGuard":null,"tryCatchPattern":"// rust\nmatch reader.deserialize_stream_chunk(start, end, epoch).await {\n    Ok(chunk) => process(chunk),\n    Err(e) if e.to_string().contains(\"should not get barrier\") => {\n        // recompute the range excluding barrier seq ids and retry once\n    }\n    Err(e) => return Err(e),\n}","preventionTips":["Exclude barrier seq ids from data-read ranges.","Handle barriers on the dedicated barrier decode path.","Re-read the barrier position if writes may land between range selection and read."],"tags":["log-store","deserialization","barrier","api-misuse"],"backgroundTag":"unsupported-operation","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"}