{"record":{"id":"19e57aee2ea4bf63","repo":"risingwavelabs/risingwave","slug":"pk-index-sink-epoch-has-no-data-files-to-commit","errorCode":null,"errorMessage":"pk-index sink epoch {} has no data files to commit","messagePattern":"pk-index sink epoch (.+?) has no data files to commit","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs","lineNumber":164,"sourceCode":"                .await\n                .with_context(|| format!(\"drain recovered pending epoch for sink {}\", sink_id))?;\n        }\n\n        Ok(coordinator)\n    }\n\n    pub async fn pre_commit(\n        &mut self,\n        prev_epoch: u64,\n        reports: Vec<PbIcebergPkIndexSinkMetadata>,\n    ) -> Result<()> {\n        if reports.iter().all(|r| r.metadata.is_none()) {\n            return Ok(());\n        }\n\n        let merged = aggregate_reports(&reports)?;\n        if merged.data_files.is_empty() && merged.delete_files.is_empty() {\n            bail!(\n                \"pk-index sink epoch {} has no data files to commit\",\n                prev_epoch\n            );\n        }\n        let (merged, materialized_add_files) = self.backfill_delete_file_partitions(merged).await?;\n        let merged = Arc::new(merged);\n\n        let snapshot_id = FastAppendAction::generate_snapshot_id(&self.table);\n        let blob = encode_pre_commit_state(&merged, snapshot_id)?;\n        persist_pre_commit_metadata(&self.db, self.sink_id, prev_epoch, Some(blob), None).await?;\n\n        self.waiting_commit = Some(EpochCommit {\n            epoch: prev_epoch,\n            merged,\n            snapshot_id,\n            materialized_add_files,\n        });\n        Ok(())","sourceCodeStart":146,"sourceCodeEnd":182,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/meta/src/manager/iceberg_pk_index_sink/coordinator.rs#L146-L182","documentation":"pre_commit rejects an epoch when the aggregated writer reports contain neither data files nor delete files, yet at least one report had metadata. The exactly-once commit path requires the epoch to contribute at least one Iceberg file; an empty file set would produce a meaningless/invalid commit, so the meta node bails with this error.","triggerScenarios":"Calling pre_commit(prev_epoch, reports) where not all reports have metadata=None (so aggregation proceeds) but aggregate_reports(&reports)?.data_files and .delete_files are both empty.","commonSituations":"Sink writers flushed only internal state with no actual pk-index rows; a bug in the writer metadata serialization that emits metadata without file lists; schema/format mismatch between writer version and meta node causing files to be dropped during aggregation.","solutions":["Check whether the epoch genuinely wrote data; if it is a no-op epoch, the writer should send metadata=None reports instead of empty metadata","Inspect writer logs and the serialized PbIcebergPkIndexSinkMetadata payloads for missing data_files/delete_files","Ensure writer and meta node versions are compatible (aggregation parses files with the same schema/format version)","If reproducible, file a bug with the reports; this state indicates an internal invariant violation"],"exampleFix":"// before (writer side)\nsend_report(metadata_with_no_files) // triggers bail in pre_commit\n// after\nif data_files.is_empty() && delete_files.is_empty() {\n    send_report(PbIcebergPkIndexSinkMetadata { metadata: None, ..}); // signal no-op\n} else {\n    send_report(full_metadata);\n}","handlingStrategy":"validation","validationCode":"// Writer-side: never send metadata reports with zero files\nfn report_is_valid(r: &PbIcebergPkIndexSinkMetadata) -> bool {\n    match &r.metadata {\n        None => true, // explicit no-op\n        Some(m) => !m.data_files.is_empty() || !m.delete_files.is_empty(),\n    }\n}\nfn build_reports(...) -> Vec<PbIcebergPkIndexSinkMetadata> {\n    if data_files.is_empty() && delete_files.is_empty() {\n        vec![no_op_report()] // metadata: None\n    } else {\n        vec![full_report()]\n    }\n}","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Writers must send metadata=None for no-op epochs instead of empty metadata","Keep writer and meta node versions aligned so file lists serialize/aggregate consistently","Log file counts at report time to catch empty-metadata anomalies early"],"tags":["iceberg","commit","empty-data","exactly-once"],"backgroundTag":"empty-required-field","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}