Concurrent metadata commit throughput benchmark snippet for Lance, Delta Lake, and Iceberg

concurrent-commit-benchmark-snippet.rs

// Abridged from the Rust harness used for the Lance / Delta Lake / Iceberg benchmark.
// Each writer repeatedly appends until it reaches the append budget or wall-clock deadline.
// Successful and failed attempts are counted by the runner.
// `delta_commit_append` and `iceberg_commit_append` are the same commit helpers shown
// in the metadata commit latency snippet.

async fn lance_run_writer(
    &self,
    writer_id: usize,
    appends: usize,
    rows_per_append: usize,
    deadline: Option<Instant>,
    per_attempt_timeout: Option<Duration>,
) -> WriterStats {
    let mut stats = WriterStats {
        latencies: Vec::with_capacity(appends),
        ..Default::default()
    };
    let mut dataset = match load_lance_dataset(&self.config).await {
        Ok(dataset) => Arc::new(dataset),
        Err(err) => {
            eprintln!("lance writer {writer_id} initial load failed: {err:#}");
            stats.failures = appends;
            return stats;
        }
    };
    let id_base = 1_000_000_000 + writer_id * appends * rows_per_append;

    for append_idx in 0..appends {
        if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
            break;
        }

        let batch = arrow_batch(id_base + append_idx * rows_per_append, rows_per_append);
        let base_dataset = dataset.clone();
        let attempt = async move {
            let transaction = InsertBuilder::new(base_dataset.clone())
                .execute_uncommitted(vec![batch])
                .await?;
            let commit_start = Instant::now();
            let new_dataset = CommitBuilder::new(base_dataset)
                .with_skip_auto_cleanup(true)
                .with_inline_transaction(true)
                .execute(transaction)
                .await?;
            Ok::<_, anyhow::Error>((commit_start.elapsed(), Arc::new(new_dataset)))
        };

        let result = match per_attempt_timeout {
            Some(timeout) => tokio::time::timeout(timeout, attempt)
                .await
                .unwrap_or_else(|_| Err(anyhow::anyhow!("per-attempt timeout"))),
            None => attempt.await,
        };

        match result {
            Ok((commit_latency, new_dataset)) => {
                stats.successes += 1;
                stats.latencies.push(commit_latency);
                dataset = new_dataset;
            }
            Err(err) => {
                stats.failures += 1;
                eprintln!("lance writer {writer_id} append {append_idx} failed: {err:#}");
                if let Ok(reloaded) = load_lance_dataset(&self.config).await {
                    dataset = Arc::new(reloaded);
                }
            }
        }
    }

    stats
}

async fn delta_run_writer(
    &self,
    writer_id: usize,
    appends: usize,
    rows_per_append: usize,
    deadline: Option<Instant>,
    per_attempt_timeout: Option<Duration>,
) -> WriterStats {
    let mut stats = WriterStats {
        latencies: Vec::with_capacity(appends),
        ..Default::default()
    };
    let mut table = match delta_load_table(
        &self.config,
        &self.storage_options,
        self.storage_backend.as_ref(),
    )
    .await
    {
        Ok(table) => table,
        Err(err) => {
            eprintln!("delta writer {writer_id} initial load failed: {err:#}");
            stats.failures = appends;
            return stats;
        }
    };
    let id_base = 2_000_000_000 + writer_id * appends * rows_per_append;

    for append_idx in 0..appends {
        if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
            break;
        }

        let attempt = delta_commit_append(
            &self.config,
            &mut table,
            rows_per_append,
            id_base + append_idx * rows_per_append,
        );
        let result = match per_attempt_timeout {
            Some(timeout) => tokio::time::timeout(timeout, attempt)
                .await
                .unwrap_or_else(|_| Err(anyhow::anyhow!("per-attempt timeout")))
                .map(|(commit_latency, _)| commit_latency),
            None => attempt.await.map(|(commit_latency, _)| commit_latency),
        };

        match result {
            Ok(commit_latency) => {
                stats.successes += 1;
                stats.latencies.push(commit_latency);
            }
            Err(err) => {
                stats.failures += 1;
                eprintln!("delta writer {writer_id} append {append_idx} failed: {err:#}");
                if let Ok(reloaded) = delta_load_table(
                    &self.config,
                    &self.storage_options,
                    self.storage_backend.as_ref(),
                )
                .await
                {
                    table = reloaded;
                }
            }
        }
    }

    stats
}

async fn iceberg_run_writer(
    &self,
    writer_id: usize,
    appends: usize,
    rows_per_append: usize,
    deadline: Option<Instant>,
    per_attempt_timeout: Option<Duration>,
) -> WriterStats {
    let mut stats = WriterStats {
        latencies: Vec::with_capacity(appends),
        ..Default::default()
    };
    let mut table = match self.catalog.load_table(&self.ident).await {
        Ok(table) => table,
        Err(err) => {
            eprintln!("iceberg writer {writer_id} initial load failed: {err:#}");
            stats.failures = appends;
            return stats;
        }
    };
    let id_base = 3_000_000_000 + writer_id * appends * rows_per_append;

    for append_idx in 0..appends {
        if deadline.is_some_and(|deadline| Instant::now() >= deadline) {
            break;
        }

        let start_id = id_base + append_idx * rows_per_append;
        let attempt = async {
            let data_file =
                prepare_iceberg_data_file(&table, &self.config, start_id, rows_per_append).await?;
            let data_path = data_file.file_path().to_string();
            match iceberg_commit_append(self.catalog.as_ref(), &table, data_file).await {
                Ok(result) => Ok(result),
                Err(err) => {
                    let _ = table.file_io().delete(&data_path).await;
                    Err(err)
                }
            }
        };
        let result = match per_attempt_timeout {
            Some(timeout) => tokio::time::timeout(timeout, attempt)
                .await
                .unwrap_or_else(|_| Err(anyhow::anyhow!("per-attempt timeout"))),
            None => attempt.await,
        };

        match result {
            Ok((commit_latency, new_table)) => {
                stats.successes += 1;
                stats.latencies.push(commit_latency);
                table = new_table;
            }
            Err(err) => {
                stats.failures += 1;
                eprintln!("iceberg writer {writer_id} append {append_idx} failed: {err:#}");
                if let Ok(reloaded) = self.catalog.load_table(&self.ident).await {
                    table = reloaded;
                }
            }
        }
    }

    stats
}