Function commit_to_iceberg

fn commit_to_iceberg<G>(
    name: String,
    sink_id: GlobalId,
    sink_version: u64,
    batch_input: &Stream<G, BoundedDataFile>,
    batch_desc_input: &Stream<G, (Antichain<Timestamp>, Antichain<Timestamp>)>,
    write_frontier: Rc<RefCell<Antichain<Timestamp>>>,
    connection: IcebergSinkConnection,
    storage_configuration: StorageConfiguration,
    write_handle: impl Future<Output = Result<WriteHandle<SourceData, (), Timestamp, StorageDiff>>> + 'static,
) -> (Stream<G, HealthStatusMessage>, PressOnDropButton)
where
    G: Scope<Timestamp = Timestamp>,

Commit completed batches to Iceberg as snapshots. Batches are committed in timestamp order to ensure strong consistency guarantees downstream. Each snapshot includes the Materialize frontier in its metadata for resume support.