Repository: risingwavelabs/risingwave
File: src/stream/src/common/log_store_impl/kv_log_store/writer.rs (Rust)

Code before:
        if self.is_paused {
            assert_eq!(flush_info.flush_count, 0);
            assert_eq!(flush_info.flush_size, 0);
        }
        flush_info.report(&self.metrics);

        let truncate_offset = self.tx.pop_truncation(epoch);
        let post_seal_epoch = self.state.seal_current_epoch(
            next_epoch,
            truncate_offset
                .map(|(epoch, seq_id)| {
                    LogStoreVnodeProgress::Aligned(self.state.vnodes().clone(), epoch, seq_id)
                })
                .unwrap_or(LogStoreVnodeProgress::None),
        );
        self.tx.barrier(epoch, options.is_checkpoint, next_epoch);
        let update_vnode_bitmap_tx = &mut self.update_vnode_bitmap_tx;
        let tx = &mut self.tx;
        self.seq_id = FIRST_SEQ_ID;
        Ok(LogWriterPostFlushCurrentEpoch::new(move || {
            async move {
                {
                    let new_vnodes = options.new_vnode_bitmap;

                    post_seal_epoch
                        .post_yield_barrier(new_vnodes.clone())
                        .await?;
                    if let Some(new_vnodes) = new_vnodes {
                        update_vnode_bitmap_tx
                            .send((new_vnodes, next_epoch))
                            .map_err(|_| anyhow!("fail to send update vnode bitmap to reader"))?;
                        tx.clear();
                    }
                    Ok(())
                }
            }
            .boxed()
        }))
    }

    fn pause(&mut self) -> LogStoreResult<()> {
        info!("KvLogStore of {} is paused", self.identity);
        assert!(!self.is_paused);
        self.is_paused = true;
        self.paused_notifier
            .send(true)
            .map_err(|_| anyhow!("unable to set pause"))
    }

    fn resume(&mut self) -> LogStoreResult<()> {
        info!("KvLogStore of {} is resumed", self.identity);

Review comment (suggestion):
Just mark here, if the sink log store lag is too large e.g. 1 days, the schema change would potentially block the whole checkpoint procedure.

Code after:
        }
        flush_info.report(&self.metrics);

        if let Some(truncate_offset) = self.tx.pop_truncation(epoch) {
            self.last_truncate_offset = Some(truncate_offset);
        }
        let post_seal_epoch = self.state.seal_current_epoch(
            next_epoch,
            self.last_truncate_offset
                .map(|(epoch, seq_id)| {
                    LogStoreVnodeProgress::Aligned(self.state.vnodes().clone(), epoch, seq_id)
                })
                .unwrap_or(LogStoreVnodeProgress::None),
        );
        let has_schema_change = options.add_columns.is_some();
        self.tx.barrier(
            epoch,
            options.is_checkpoint,
            next_epoch,
            options.add_columns,
        );
        if has_schema_change {
            let truncate_offset = self.tx.wait_for_barrier_truncation(epoch).await?;
            assert_eq!(truncate_offset, (epoch, None));
            self.last_truncate_offset = Some(truncate_offset);
        }
        let update_vnode_bitmap_tx = &mut self.update_vnode_bitmap_tx;
        let tx = &mut self.tx;
        self.seq_id = FIRST_SEQ_ID;
        Ok(LogWriterPostFlushCurrentEpoch::new(move || {
            async move {
                {
                    let new_vnodes = options.new_vnode_bitmap;

                    post_seal_epoch
                        .post_yield_barrier(new_vnodes.clone())
                        .await?;
                    if let Some(new_vnodes) = new_vnodes {
                        update_vnode_bitmap_tx
                            .send((new_vnodes, next_epoch))
                            .map_err(|_| anyhow!("fail to send update vnode bitmap to reader"))?;
                        tx.clear();
                    }
                    Ok(())
                }
            }
            .boxed()
        }))
    }

    fn pause(&mut self) -> LogStoreResult<()> {