fn drop_batches(&mut self) {
        if let Some(logger) = &self.logger {
            for batch in self.merging.drain(..) {
                match batch {
                    MergeState::Single(Some(batch)) => {
                        logger.log(crate::logging::DropEvent {
                            operator: self.operator.global_id,
                            length: batch.len(),
                        });
                    },
                    MergeState::Double(MergeVariant::InProgress(batch1, batch2, _)) => {
                        logger.log(crate::logging::DropEvent {
                            operator: self.operator.global_id,
                            length: batch1.len(),
                        });
                        logger.log(crate::logging::DropEvent {
                            operator: self.operator.global_id,
                            length: batch2.len(),
                        });
                    },
                    MergeState::Double(MergeVariant::Complete(Some((batch, _)))) => {
                        logger.log(crate::logging::DropEvent {
                            operator: self.operator.global_id,
                            length: batch.len(),
                        });
                    }
                    _ => { },
                }
            }
            for batch in self.pending.drain(..) {
                logger.log(crate::logging::DropEvent {
                    operator: self.operator.global_id,
                    length: batch.len(),
                });
            }
        }
    }
Drops and logs batches. Used in `set_logical_compaction` and drop.
differential-dataflow/src/trace/implementations/spine_fueled.rs