Repository: risingwavelabs/risingwave
File: src/connector/src/sink/iceberg/mod.rs (Rust)

Code before:
                table.metadata().default_partition_spec_id()
            )));
        }
        Ok(table)
    }
}

#[async_trait]
impl SinglePhaseCommitCoordinator for IcebergSinkCommitter {
    async fn init(&mut self) -> Result<()> {
        tracing::info!(
            "Sink id = {}: iceberg sink coordinator initing.",
            self.param.sink_id
        );

        Ok(())
    }

    async fn commit(
        &mut self,
        epoch: u64,
        metadata: Vec<SinkMetadata>,
        add_columns: Option<Vec<Field>>,
    ) -> Result<()> {
        tracing::info!("Starting iceberg direct commit in epoch {epoch}");
        // Commit data if present
        let (write_results, snapshot_id) = match self.pre_commit_inner(epoch, metadata, None)? {
            Some((write_results, snapshot_id)) => (write_results, snapshot_id),
            None => {
                // No data to commit, but may have schema change
                if let Some(add_columns) = add_columns {
                    tracing::info!(?epoch, "No data to commit, but committing schema change");
                    self.commit_schema_change(add_columns).await?;
                } else {
                    tracing::debug!(?epoch, "no data to commit");
                }
                return Ok(());
            }
        };

        self.commit_datafile(epoch, write_results, snapshot_id)
            .await?;

        // Commit schema change if present
        if let Some(add_columns) = add_columns {
            tracing::info!(?epoch, "Committing schema change after data commit");
            self.commit_schema_change(add_columns).await?;
        }

        Ok(())
    }

Review comment (suggestion):
May temporarily add a check in 2pc coordinator when we try to add columns in `pre_commit`.

Code after:
                "Partition evolution not supported, expect partition spec id {}, but got {}",
                partition_spec_id,
                table.metadata().default_partition_spec_id()
            )));
        }
        Ok(table)
    }
}

#[async_trait]
impl SinglePhaseCommitCoordinator for IcebergSinkCommitter {
    async fn init(&mut self) -> Result<()> {
        tracing::info!(
            "Sink id = {}: iceberg sink coordinator initing.",
            self.param.sink_id
        );

        Ok(())
    }

    async fn commit(
        &mut self,
        epoch: u64,
        metadata: Vec<SinkMetadata>,
        add_columns: Option<Vec<Field>>,
    ) -> Result<()> {
        tracing::info!("Starting iceberg direct commit in epoch {epoch}");

        // Commit data if present
        if let Some((write_results, snapshot_id)) = self.pre_commit_inner(epoch, metadata, None)? {
            self.commit_datafile(epoch, write_results, snapshot_id)
                .await?;
        }

        // Commit schema change if present
        if let Some(add_columns) = add_columns {
            tracing::info!(?epoch, "Committing schema change");
            self.commit_schema_change(add_columns).await?;
        }

        Ok(())
    }
}

#[async_trait]
impl TwoPhaseCommitCoordinator for IcebergSinkCommitter {
    async fn init(&mut self) -> Result<()> {
        tracing::info!(
            "Sink id = {}: iceberg sink coordinator initing.",
            self.param.sink_id
        );