Synchronize Cayenne partition commits across partitions (spiceai#10125) (spiceai#10819)

* Add CayenneStagedAppend two-phase commit lifecycle (spiceai#10125)

* Add commit_compaction_in_txn catalog primitive (spiceai#10125)

* Tighten CayenneTableProvider scan() lock against barrier (spiceai#10125)

* Extract PreparedOverwrite lifecycle from write_all_overwrite (spiceai#10125)

* Add CayennePartitionedInsertStrategy for atomic overwrite (spiceai#10125)

* Add top-level partitioned-commit WAL format (spiceai#10125)

* Extend CayennePartitionedInsertStrategy to atomic append (spiceai#10125)

* Address Copilot review comments — atomic WAL write, S3 fallback (spiceai#10125)

- partitioned_wal: write WAL via tempfile + fsync + atomic rename + parent
  dir fsync so a crash mid-write can never leave an unparseable file that
  read_all_in would skip.
- partitioned_insert_strategy: skip the top-level partitioned WAL for
  S3-backed table roots (tokio::fs would attempt local paths like
  `s3:/bucket/...` and fail). Per-partition staging WAL still anchors
  single-partition recovery on S3; the cross-partition set anchor is
  deferred until the WAL grows object-store IO.
- partitioned_insert_strategy: module docs now describe the actual append
  coordination flow (CayennePartitionedAppendSink) instead of the stale
  "falls through to DefaultInsertStrategy" wording.
- staging_wal / overwrite: rollback paths hold the per-table write guard
  for the full cleanup so a concurrent writer cannot observe leftover
  staging state mid-rollback.

* Drop stringified-column shortcut in partition-expr compilation (spiceai#10125)

create_partition_physical_exprs previously tried an unqualified column
lookup using `other.to_string()` for any non-Column partition expression
before falling back to compiling the real expression. If the input schema
contained a column whose name happened to match the debug form of a
derived partition expression, the shortcut would silently succeed and
route partitioning through the wrong physical expression — incorrect
partition keys, incorrect commit groupings.

Compile bare `Expr::Column` partitions via unqualified-name lookup as
before; all other expressions are compiled as-is. No string-form fallback.
Addresses Copilot comment on PR spiceai#10819.

Also restages fmt-only fixes left from the trunk merge.

* Add backticks around DataFusion in builder.rs test doc

clippy::doc_markdown caught this on the test-clippy step (the lib step
clears with the same rule, but the test build picks up a different cfg
set). Pre-existing on trunk; fixing here so the PR branch's lint is
green.

* Update stats edge-case test to assert post-merge aggregation

persist_table_stats now merges the current write's accumulator with the
existing metastore aggregate (current trunk behavior, see spiceai#10818). The
test expectations were locked to the pre-merge "latest-write-only" path
with a TBD note for when aggregation lands — refresh them to the merged
shape: num_rows = 3 + 2 = 5, min/max span both batches. Renamed the test
+ doc to match.

Author: phillipleblanc

Cargo.lock

@@ -70,7 +70,7 @@ turso = ["dep:turso"]
 partition-table-provider = ["dep:runtime-table-partition"]
 
 [dev-dependencies]
-criterion = { version = "0.7", features = ["html_reports"] }
+criterion = { version = "0.7", features = ["html_reports", "async_tokio"] }
 datafusion-federation.workspace = true
 datafusion-functions = { workspace = true }
 insta = { workspace = true, features = ["json"] }
@@ -80,6 +80,7 @@ tempfile = { workspace = true }
 test-framework = { path = "../test-framework" }
 test-log = { version = "0.2", features = ["trace"] }
 tokio = { workspace = true, features = ["test-util", "rt-multi-thread", "fs"] }
+tokio-stream = { workspace = true }
 
 [[bench]]
 harness = false
@@ -96,3 +97,7 @@ name = "deletion_index_probe"
 [[bench]]
 harness = false
 name = "mutation_writer"
+
+[[bench]]
+harness = false
+name = "listing_fence_overhead"

crates/cayenne/Cargo.toml

@@ -0,0 +1,154 @@
+// Copyright 2026 The Spice.ai OSS Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      https://www.apache.org/licenses/LICENSE-2.0
+
+//! Microbenchmark for the `listing_table` synchronization pattern in
+//! `CayenneTableProvider::scan()`.
+//!
+//! Step 3 of issue #10125 replaced `Arc<std::sync::RwLock<Arc<ListingTable>>>`
+//! with `Arc<ArcSwap<ListingTable>>` + a separate
+//! `Arc<tokio::sync::RwLock<()>>` fence. The previous "grab Arc under brief
+//! sync guard, drop guard, then `.scan().await`" pattern is replaced with a
+//! fence read held across the `.scan().await` so that concurrent writer
+//! barriers cannot interleave with the listing operation.
+//!
+//! This bench isolates the per-scan synchronization cost of each pattern
+//! (no DataFusion listing, no I/O, no filesystem state — just the lock
+//! primitives) and runs both inside the same tokio runtime, so the `.await`
+//! that the new pattern requires does not double-count `Runtime::block_on`
+//! overhead.
+
+#![allow(clippy::expect_used)]
+
+use std::hint::black_box;
+use std::sync::Arc;
+
+use arc_swap::ArcSwap;
+use criterion::{Criterion, criterion_group, criterion_main};
+use tokio::runtime::Runtime;
+
+/// Stand-in for `Arc<ListingTable>` so the comparison focuses on lock cost
+/// rather than the size of the inner type.
+type Inner = String;
+
+fn make_inner() -> Arc<Inner> {
+    Arc::new("listing_table_placeholder".to_string())
+}
+
+// ----------------------------------------------------------------------------
+// Uncontended single-task access.
+// ----------------------------------------------------------------------------
+//
+// Steady-state read path: one task calling scan() with no concurrent reader
+// or writer. Measures the per-call overhead of acquiring the synchronization
+// primitive + loading the Arc. Both patterns run inside the same tokio
+// runtime via `to_async` so the async pattern doesn't pay block_on overhead.
+
+fn bench_uncontended_old_pattern(c: &mut Criterion) {
+    let rt = Runtime::new().expect("runtime");
+    let lock: Arc<std::sync::RwLock<Arc<Inner>>> = Arc::new(std::sync::RwLock::new(make_inner()));
+
+    c.bench_function("uncontended/old_sync_rwlock_then_arc_clone", |b| {
+        b.to_async(&rt).iter(|| async {
+            let guard = lock.read().expect("read");
+            let snapshot = Arc::clone(&guard);
+            drop(guard);
+            black_box(snapshot);
+        });
+    });
+}
+
+fn bench_uncontended_new_pattern(c: &mut Criterion) {
+    let rt = Runtime::new().expect("runtime");
+    let arc_swap: Arc<ArcSwap<Inner>> = Arc::new(ArcSwap::new(make_inner()));
+    let fence: Arc<tokio::sync::RwLock<()>> = Arc::new(tokio::sync::RwLock::new(()));
+
+    c.bench_function("uncontended/new_fence_read_then_arcswap_load", |b| {
+        b.to_async(&rt).iter(|| async {
+            let _fence_guard = fence.read().await;
+            let snapshot = arc_swap.load_full();
+            black_box(snapshot);
+        });
+    });
+}
+
+// ----------------------------------------------------------------------------
+// Concurrent-reader access, no writer.
+// ----------------------------------------------------------------------------
+//
+// Multi-tenant steady state: several scans share the same partition. Both
+// std::sync::RwLock and tokio::sync::RwLock allow parallel readers, but each
+// adds different atomic-counter overhead per acquisition.
+
+fn bench_concurrent_readers_old_pattern(c: &mut Criterion) {
+    let rt = Runtime::new().expect("runtime");
+    let lock: Arc<std::sync::RwLock<Arc<Inner>>> = Arc::new(std::sync::RwLock::new(make_inner()));
+
+    // Background reader that keeps a read guard outstanding most of the time.
+    let bg_lock = Arc::clone(&lock);
+    let bg = std::thread::spawn(move || {
+        loop {
+            let _guard = bg_lock.read().expect("bg read");
+            std::thread::yield_now();
+            if Arc::strong_count(&bg_lock) == 1 {
+                break;
+            }
+        }
+    });
+
+    c.bench_function("concurrent_readers/old_sync_rwlock", |b| {
+        b.to_async(&rt).iter(|| async {
+            let guard = lock.read().expect("read");
+            let snapshot = Arc::clone(&guard);
+            drop(guard);
+            black_box(snapshot);
+        });
+    });
+
+    drop(lock);
+    let _ = bg.join();
+}
+
+fn bench_concurrent_readers_new_pattern(c: &mut Criterion) {
+    let rt = Runtime::new().expect("runtime");
+    let arc_swap: Arc<ArcSwap<Inner>> = Arc::new(ArcSwap::new(make_inner()));
+    let fence: Arc<tokio::sync::RwLock<()>> = Arc::new(tokio::sync::RwLock::new(()));
+
+    let bg_fence = Arc::clone(&fence);
+    let bg = std::thread::spawn(move || {
+        let rt = Runtime::new().expect("bg runtime");
+        rt.block_on(async {
+            loop {
+                let _fence_guard = bg_fence.read().await;
+                tokio::task::yield_now().await;
+                if Arc::strong_count(&bg_fence) == 1 {
+                    break;
+                }
+            }
+        });
+    });
+
+    c.bench_function("concurrent_readers/new_fence_arcswap", |b| {
+        b.to_async(&rt).iter(|| async {
+            let _fence_guard = fence.read().await;
+            let snapshot = arc_swap.load_full();
+            black_box(snapshot);
+        });
+    });
+
+    drop(fence);
+    let _ = bg.join();
+}
+
+criterion_group!(
+    benches,
+    bench_uncontended_old_pattern,
+    bench_uncontended_new_pattern,
+    bench_concurrent_readers_old_pattern,
+    bench_concurrent_readers_new_pattern,
+);
+criterion_main!(benches);