Metadata commit latency benchmark snippet for Lance, Delta Lake, and Iceberg

metadata-commit-benchmark-snippet.rs

// Abridged from the Rust harness used for the Lance / Delta Lake / Iceberg benchmark.
// The important detail is the timing boundary: data-file or fragment preparation
// happens before `commit_start`, so `commit_latency` measures the metadata commit path.

async fn lance_append_and_measure(&mut self, iteration: usize) -> Result<ContinuousStep> {
    let start_id = iteration * self.config.rows_per_append;
    let batch = arrow_batch(start_id, self.config.rows_per_append);
    let params = write_params(
        self.session.clone(),
        self.store_params.clone(),
        WriteMode::Append,
    );

    let transaction = InsertBuilder::new(self.dataset.clone())
        .with_params(&params)
        .execute_uncommitted(vec![batch])
        .await
        .context("write uncommitted Lance fragment")?;

    let commit_start = Instant::now();
    let new_dataset = CommitBuilder::new(self.dataset.clone())
        .with_session(self.session.clone())
        .with_skip_auto_cleanup(true)
        .with_inline_transaction(true)
        .execute(transaction)
        .await
        .context("commit Lance transaction")?;
    let commit_latency = commit_start.elapsed();

    self.dataset = Arc::new(new_dataset);
    Ok(ContinuousStep {
        version: self.dataset.manifest().version as i64,
        commit_latency,
        cold_reload_latency: Duration::ZERO,
        latest_file_count: 0,
    })
}

async fn delta_commit_append(
    config: &BenchConfig,
    table: &mut deltalake::DeltaTable,
    rows_per_append: usize,
    start_id: usize,
) -> Result<(Duration, u64)> {
    let batch = arrow_batch(start_id, rows_per_append);
    let mut writer = deltalake::writer::RecordBatchWriter::for_table(table)
        .context("create Delta record batch writer")?;
    writer.write(batch).await.context("write Delta parquet data file")?;
    let add_actions: Vec<deltalake::kernel::Action> = writer
        .flush()
        .await
        .context("flush Delta parquet data file")?
        .into_iter()
        .map(deltalake::kernel::Action::Add)
        .collect();

    let commit_start = Instant::now();
    let commit_properties = deltalake::kernel::transaction::CommitProperties::default()
        .with_create_checkpoint(config.delta_create_checkpoint)
        .with_cleanup_expired_logs(Some(false));
    let finalized = deltalake::kernel::transaction::CommitBuilder::from(commit_properties)
        .with_actions(add_actions)
        .build(
            Some(table.snapshot()? as &dyn deltalake::kernel::transaction::TableReference),
            table.log_store(),
            deltalake::protocol::DeltaOperation::Write {
                mode: deltalake::protocol::SaveMode::Append,
                partition_by: None,
                predicate: None,
            },
        )
        .await
        .context("commit Delta transaction")?;
    let version = finalized.version();
    table.state = Some(finalized.snapshot());
    Ok((commit_start.elapsed(), version))
}

async fn iceberg_commit_append(
    catalog: &dyn Catalog,
    table: &IcebergTable,
    data_file: DataFile,
) -> Result<(Duration, IcebergTable)> {
    let tx = Transaction::new(table);
    let action = tx
        .fast_append()
        .with_check_duplicate(false)
        .add_data_files(vec![data_file]);
    let tx = action
        .apply(tx)
        .context("apply Iceberg fast append action")?;

    let commit_start = Instant::now();
    let new_table = tx
        .commit(catalog)
        .await
        .context("commit Iceberg fast append transaction")?;
    Ok((commit_start.elapsed(), new_table))
}