fn normalize_candidate_batch( batch: RecordBatch, filter: Option<&mut CachedPrimaryKeyFilter>, last_primary_key: &mut Vec<u8>, has_last: &mut bool, ) -> Result<Option<RecordBatch>>