Function write_data_files 

fn write_data_files<'scope, H: EnvelopeHandler + 'static>(
    name: String,
    input: StreamVec<'scope, Timestamp, <TraceAgent<OrdValSpine<Option<Row>, Row, Timestamp, Diff>> as TraceReader>::Batch>,
    batch_desc_input: StreamVec<'scope, Timestamp, (Antichain<Timestamp>, Antichain<Timestamp>)>,
    table_ready_stream: StreamVec<'scope, Timestamp, Infallible>,
    sink_id: GlobalId,
    from_id: GlobalId,
    key_is_synthetic: bool,
    as_of: Antichain<Timestamp>,
    connection: IcebergSinkConnection,
    storage_configuration: StorageConfiguration,
    materialize_arrow_schema: Arc<Schema>,
    metrics: Arc<IcebergSinkMetrics>,
    statistics: SinkStatistics,
) -> (StreamVec<'scope, Timestamp, BoundedDataFile>, StreamVec<'scope, Timestamp, HealthStatusMessage>, PressOnDropButton)

Construct the envelope-specific closures that write_data_files needs.

Write rows into Parquet data files bounded by batch descriptions. Rows are matched to batches by timestamp; if a batch description hasn’t arrived yet, rows are stashed until it does. This allows batches to be minted ahead of data arrival.