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

Code before:
            return Err(SinkError::Iceberg(anyhow!(
                "Unsupported sink schema change op in iceberg sink: {:?}",
                schema_change.op
            )));
        };

        let add_columns = add_columns_op.fields.iter().map(Field::from).collect_vec();

        tracing::info!(
            "Starting to commit schema change with {} columns",
            add_columns.len()
        );

        // Step 1: Get current table metadata
        let metadata = self.table.metadata();
        let mut next_field_id = metadata.last_column_id() + 1;
        tracing::debug!("Starting schema change, next_field_id: {}", next_field_id);

        // Step 2: Build new fields to add
        let iceberg_create_table_arrow_convert = IcebergCreateTableArrowConvert::default();
        let mut new_fields = Vec::new();

        for field in &add_columns {
            // Convert RisingWave Field to Arrow Field using IcebergCreateTableArrowConvert
            let arrow_field = iceberg_create_table_arrow_convert
                .to_arrow_field(&field.name, &field.data_type)
                .with_context(|| format!("Failed to convert field '{}' to arrow", field.name))
                .map_err(SinkError::Iceberg)?;

            // Convert Arrow DataType to Iceberg Type
            let iceberg_type = iceberg::arrow::arrow_type_to_type(arrow_field.data_type())
                .map_err(|err| {
                    SinkError::Iceberg(
                        anyhow!(err).context("Failed to convert Arrow type to Iceberg type"),
                    )
                })?;

            // Create NestedField with the next available field ID
            let nested_field = Arc::new(NestedField::optional(
                next_field_id,
                &field.name,
                iceberg_type,
            ));

            new_fields.push(nested_field);
            tracing::info!("Prepared field '{}' with ID {}", field.name, next_field_id);
            next_field_id += 1;
        }

        // Step 3: Create Transaction with UpdateSchemaAction
        tracing::info!(

Review comment (bug):
The error message states "cannot determine whether schema change is applied" but doesn't provide enough information about what the current schema actually contains. Consider including the current schema field count in the error message to help with debugging when this error occurs.
```suggestion
            "Current iceberg schema ({} cols) does not match either original_schema ({} cols) or original_schema + add_columns ({} cols); cannot determine whether schema change is applied",
            current_arrow_schema.fields().len(),
            schema_change.original_schema.len(),
            expected_after_change.len(),
```

Code after:
            .fields
            .iter()
            .map(|pb_field| {
                let field = Field::from(pb_field);
                iceberg_arrow_convert
                    .to_arrow_field(&field.name, &field.data_type)
                    .context("Failed to convert field to arrow")
                    .map_err(SinkError::Iceberg)
            })
            .collect::<Result<_>>()?;

        let mut expected_after_change = original_arrow_fields;
        expected_after_change.extend(add_arrow_fields);

        // If current schema equals original_schema + add_columns, then schema change is applied.
        if schema_matches(&expected_after_change) {
            tracing::debug!(
                "Current iceberg schema matches original_schema + add_columns ({} columns); schema change already applied",
                expected_after_change.len()
            );
            return Ok(true);
        }

        Err(SinkError::Iceberg(anyhow!(
            "Current iceberg schema does not match either original_schema ({} cols) or original_schema + add_columns; cannot determine whether schema change is applied",
            schema_change.original_schema.len()
        )))
    }

    /// Commit schema changes (e.g., add columns) to the iceberg table.
    /// This function uses Transaction API to atomically update the table schema
    /// with optimistic locking to prevent concurrent conflicts.
    async fn commit_schema_change(&mut self, schema_change: PbSinkSchemaChange) -> Result<()> {
        use iceberg::spec::NestedField;

        let Some(risingwave_pb::stream_plan::sink_schema_change::Op::AddColumns(add_columns_op)) =
            schema_change.op.as_ref()
        else {
            return Err(SinkError::Iceberg(anyhow!(
                "Unsupported sink schema change op in iceberg sink: {:?}",
                schema_change.op
            )));
        };

        let add_columns = add_columns_op.fields.iter().map(Field::from).collect_vec();

        tracing::info!(
            "Starting to commit schema change with {} columns",
            add_columns.len()
        );