ManishSharma1609
diff --git a/‎Cargo.lock‎
Lines changed: 165 additions & 165 deletions b/‎Cargo.lock‎
Lines changed: 165 additions & 165 deletions
diff --git a/‎Cargo.toml‎
Lines changed: 12 additions & 12 deletions b/‎Cargo.toml‎
Lines changed: 12 additions & 12 deletions
diff --git a/‎crates/cayenne/Cargo.toml‎
Lines changed: 56 additions & 0 deletions b/‎crates/cayenne/Cargo.toml‎
Lines changed: 56 additions & 0 deletions
diff --git a/‎crates/cayenne/README.md‎
Lines changed: 13 additions & 10 deletions b/‎crates/cayenne/README.md‎
Lines changed: 13 additions & 10 deletions
diff --git a/‎crates/cayenne/benches/apply_on_conflict_rpc_ceiling.rs‎
Lines changed: 179 additions & 0 deletions b/‎crates/cayenne/benches/apply_on_conflict_rpc_ceiling.rs‎
Lines changed: 179 additions & 0 deletions
@@ -347,12 +347,12 @@ flatbuffers = "25.2.10"
 # Vortex tagged releases publish version 0.1.0 in their Cargo.toml, not the actual release version,
 # so we specify git deps here directly. The [patch.crates-io] entries below exist to override
 # transitive vortex dependencies pulled in by other patched crates.
-vortex = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-array = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-datafusion = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-scan = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-session = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-utils = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
+vortex = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-array = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-datafusion = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-scan = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-session = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-utils = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
 
 x509-certificate = "0.25.0"
 spiceai = { git = "https://github.com/spiceai/spice-rs.git", rev = "4c63970cf1d78c71eb69e2f4346d5d4795631b47" } # branch: trunk
@@ -470,9 +470,9 @@ parquet = { git = "https://github.com/spiceai/arrow-rs.git", rev = "96190c47f7a1
 
 object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "f0a772cd49d2ebb1f19f487ccd93d705f48dc891" }
 
-vortex = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-array = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-datafusion = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-scan = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-session = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
-vortex-utils = { git = "https://github.com/spiceai/vortex.git", rev = "c765c90de6e1adccb822d0bddfe67d6738cc0d39" } # spiceai-52.5
+vortex = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-array = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-datafusion = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-scan = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-session = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
+vortex-utils = { git = "https://github.com/spiceai/vortex.git", rev = "86cc337cb133fc54ec45bd665e348453a84785a6" } # spiceai-52.5
@@ -129,6 +129,18 @@ name = "checkpoint_fence_stall"
 harness = false
 name = "metastore_connection_contention"
 
+[[bench]]
+harness = false
+name = "stats_persistence_rpc_ceiling"
+
+[[bench]]
+harness = false
+name = "apply_partial_deletion_filter_per_scan"
+
+[[bench]]
+harness = false
+name = "apply_on_conflict_rpc_ceiling"
+
 [[bench]]
 harness = false
 name = "validate_on_conflict_buffering"
@@ -165,6 +177,10 @@ name = "deletion_vector_bitmap_to_treemap"
 harness = false
 name = "wide_table_key_probe_scan"
 
+[[bench]]
+harness = false
+name = "load_existing_keyset_cap_disabled"
+
 [[bench]]
 harness = false
 name = "vs_duckdb_in_list_delete"
@@ -200,6 +216,11 @@ harness = false
 name = "vs_duckdb_upsert"
 required-features = ["duckdb-bench"]
 
+[[bench]]
+harness = false
+name = "vs_duckdb_upsert_scaling"
+required-features = ["duckdb-bench"]
+
 [[bench]]
 harness = false
 name = "vs_duckdb_groupby"
@@ -218,3 +239,38 @@ required-features = ["duckdb-bench"]
 [[bench]]
 harness = false
 name = "compaction_picker"
+
+[[bench]]
+harness = false
+name = "scan_listing_cache_invalidation"
+
+[[bench]]
+harness = false
+name = "int64_pk_filter_keep_mask_alloc"
+
+[[bench]]
+harness = false
+name = "position_delete_redundant_walks"
+
+[[bench]]
+harness = false
+name = "apply_partial_filter_empty_alloc"
+
+[[bench]]
+harness = false
+name = "pk_lookup_file_group_fanout"
+required-features = ["duckdb-bench"]
+
+[[bench]]
+harness = false
+name = "pk_lookup_session_cache_warmup"
+required-features = ["duckdb-bench"]
+
+[[bench]]
+harness = false
+name = "pk_in_list_vs_range_rewrite"
+required-features = ["duckdb-bench"]
+
+[[bench]]
+harness = false
+name = "get_max_delete_sequence_walk"
@@ -122,9 +122,10 @@ pub trait MetadataCatalog: Send + Sync {
     async fn get_table(&self, table_name: &str) -> CatalogResult<TableMetadata>;
     async fn drop_table(&self, table_name: &str) -> CatalogResult<bool>;
 
-    // Sequence numbers
+    // Sequence numbers (reserve reduces round-trips on serialized backends)
     async fn increment_sequence_number(&self, table_id: &str) -> CatalogResult<i64>;
     async fn get_sequence_number(&self, table_id: &str) -> CatalogResult<i64>;
+    async fn reserve_sequence_numbers(&self, table_id: &str, count: u32) -> CatalogResult<i64>;
 
     // Delete files (position- and key-based)
     async fn add_delete_file(&self, delete_file: DeleteFile) -> CatalogResult<String>;
@@ -186,7 +187,7 @@ pub trait MetadataCatalog: Send + Sync {
 - **`InlinedDataStats`** — `{ total_rows, segment_count, total_bytes }` aggregated from `cayenne_inlined_data` for memtable-pressure decisions.
 - **`PartitionMetadata`** — composite partition key, partition path, record/byte counts.
 - **`TableStatistics`** — serialized `FileStatistics` blob plus `num_rows`; populated from Vortex file footers and read by the DataFusion planner.
-- **`VortexConfig`** — Vortex-side tuning. All fields configurable per dataset via `cayenne_*` runtime parameters:
+- **`VortexConfig`** — Vortex-side tuning. All fields configurable per dataset via `cayenne_*` runtime parameters. The runtime applies refresh-mode defaults before parsing explicit params: `refresh_mode: caching`, `changes`, and `append` with `refresh_check_interval <= 5m` favor small incremental writes, while manual/cron/long-interval append plus `refresh_mode: full`, `snapshot`, `disabled`, and unspecified refresh modes favor large Vortex writes by default. Append workloads can be small or large depending on caller batch size, so tune the inline and compaction parameters explicitly if refresh cadence does not reflect write size.
 
 ```rust
 pub struct VortexConfig {
@@ -204,20 +205,22 @@ pub struct VortexConfig {
     pub write_concurrency: Option<usize>,     // None = session target_partitions; forced to 1 if sort_columns set
 
     // Compaction
-    pub compaction_trigger_files: usize,      // default 8
+    pub compaction_trigger_files: usize,      // default caching/changes/short-append=4, otherwise=8
+    pub compaction_trigger_protected_snapshots: usize, // default caching/changes/short-append=4, otherwise=8
+    pub compaction_trigger_snapshot_age_ms: u64,  // default caching/changes/short-append=60_000, otherwise=300_000; 0 disables age trigger
     pub compaction_max_levels: usize,         // default 3
     pub compaction_max_files_per_pick: usize, // default 32
-    pub compaction_background_interval_ms: u64,  // default 30_000, 0 disables background loop
+    pub compaction_background_interval_ms: u64,  // default caching/changes/short-append=10_000, otherwise=30_000; 0 disables background loop
 
     // Inline-write admission (per-call gate)
-    pub inline_max_rows: usize,               // default 1_024
-    pub inline_max_bytes: usize,              // default 1_048_576 (1 MiB serialized IPC)
-    pub inline_max_buffer_bytes: usize,       // default 4_194_304 (4 MiB pre-decode buffer)
+    pub inline_max_rows: usize,               // default caching/changes/short-append=1_024, otherwise=0
+    pub inline_max_bytes: usize,              // default caching/changes/short-append=1_048_576, otherwise=0
+    pub inline_max_buffer_bytes: usize,       // default caching/changes/short-append=4_194_304, otherwise=0
 
     // Inline-memtable flush triggers (cumulative gate)
-    pub inline_flush_max_rows: i64,           // default 10_000
-    pub inline_flush_max_segments: i64,       // default 64
-    pub inline_flush_max_bytes: i64,          // default 8_388_608 (8 MiB total IPC)
+    pub inline_flush_max_rows: i64,           // default caching/changes/short-append=2_048, otherwise=10_000
+    pub inline_flush_max_segments: i64,       // default caching/changes/short-append=16, otherwise=64
+    pub inline_flush_max_bytes: i64,          // default caching/changes/short-append=2_097_152, otherwise=8_388_608
 
     // PK conflict detection
     pub pk_conflict_detection: PkConflictDetection,  // default Auto; None opts into blind append for CDC
 
@@ -0,0 +1,179 @@
+// 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
+
+//! Regression bench: per-upsert metastore call count in
+//! [`crate::provider::table::CayenneTableProvider::apply_on_conflict_deletions`].
+//!
+//! Older versions of the on-conflict path ran a non-atomic 3+ metastore-call
+//! sequence on every upsert that produces deletion vectors:
+//!
+//! 1. `catalog.increment_sequence_number(table_id)` — 1 call
+//! 2. `DeletionVectorWriter::write(specs)` — writes deletion-vector
+//!    files to disk (NOT counted here; we measure metastore calls only)
+//! 3. For each `DeletionVectorWriteResult`:
+//!    `catalog.add_delete_file(result.delete_file)` — 1 call per file
+//! 4. `catalog.add_insert_records_batch(...)` — 1 call per insert-record chunk
+//!
+//! For a typical PK-mode upsert that produces 1-2 delete files the cumulative
+//! cost was 3-4 metastore calls per upsert. None of those calls were wrapped
+//! in a transaction, so a crash between step 3 and step 4 could leave the
+//! catalog with delete-file records at `delete_sequence` but no insert-record
+//! at `insert_sequence`, permanently hiding the new row on restart.
+//!
+//! The production path now calls `commit_on_conflict_deletions`
+//! ([`crate::catalog::MetadataCatalog::commit_on_conflict_deletions`], wired in
+//! at `provider/table.rs:4506-4526`) which opens one transaction, INSERTs every
+//! `cayenne_delete_file` row, INSERTs every insert-record row (chunked under
+//! SQLite's 32 K-param cap, as `add_insert_records_batch_in_chunks` already
+//! does internally), and commits. Crash anywhere before commit → catalog state
+//! unchanged → the upsert is fully re-driveable from the calling write path.
+//!
+//! Counted metastore-call totals for the one-insert-record-chunk case:
+//!
+//! | path                                            | calls per upsert     | atomic? |
+//! |-------------------------------------------------|----------------------|---------|
+//! | older (`apply_on_conflict_deletions`, no txn)   | `2 + delete_files`   | no      |
+//! | current (`commit_on_conflict_deletions` in txn) | `4 + delete_files`   | yes     |
+//!
+//! ## What this bench measures
+//!
+//! Pure shape — same `tokio::sync::Mutex<()>` + `tokio::time::sleep(call_latency)`
+//! pattern as `stats_persistence_rpc_ceiling.rs`. No real SQLite, no Cayenne
+//! setup. Two lanes per `(delete_files_per_upsert, upsert_count, call_latency)`:
+//!
+//! - `no_txn_baseline` — each upsert: 1 call (increment) + N calls
+//!   (add_delete_file × N) + 1 call (add_insert_records_batch). Total =
+//!   `(N + 2)` calls per upsert. Mirrors the older non-atomic path.
+//! - `atomic_txn_calls` — current behavior. Each upsert: 1 call (increment) +
+//!   1 call (begin transaction) + N calls (delete-file INSERTs) + 1 call
+//!   (insert-record chunk INSERT) + 1 call (commit). Total = `(N + 4)` calls
+//!   per upsert for this bench's single-chunk setup.
+//!
+//! The bench keeps the call-count tradeoff visible — atomicity costs a
+//! constant 2 extra calls per upsert in exchange for closing the crash window.
+//!
+//! `cargo bench --bench apply_on_conflict_rpc_ceiling -p cayenne`.
+
+#![expect(clippy::expect_used)]
+
+use std::hint::black_box;
+use std::sync::Arc;
+use std::time::Duration;
+
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use tokio::sync::Mutex;
+
+/// Per-call simulated metastore latency — mirrors the other RPC-ceiling benches.
+const RTTS: &[(&str, Duration)] = &[
+    ("rtt_1ms", Duration::from_millis(1)),
+    ("rtt_10ms", Duration::from_millis(10)),
+];
+
+/// Delete files per upsert. Typical PK-mode upsert produces 1-2 files
+/// (one per touched virtual file). Larger counts model partitioned or
+/// wide-fan-out upserts.
+const DELETE_FILES_PER_UPSERT: &[usize] = &[1, 2, 4];
+
+/// Upsert count per iteration. 32 keeps the worst-case shape
+/// (`N=4`, `rtt_10ms`, current) under 2.5 s per iteration so Criterion
+/// `--quick` produces a multi-sample distribution. Pattern lifted from
+/// `stats_persistence_rpc_ceiling.rs:128`.
+const UPSERTS_PER_ITERATION: usize = 32;
+
+/// One simulated metastore call. Same shape as other RPC-ceiling benches.
+async fn one_metastore_call(pool: &Mutex<()>, rtt: Duration) {
+    let _guard = pool.lock().await;
+    tokio::time::sleep(rtt).await;
+}
+
+/// Lane A: mirrors the older non-atomic `apply_on_conflict_deletions`. Each
+/// upsert pays `2 + delete_files` separate metastore calls.
+async fn run_current(pool: &Arc<Mutex<()>>, upserts: usize, delete_files: usize, rtt: Duration) {
+    for _ in 0..upserts {
+        // 1. increment_sequence_number
+        one_metastore_call(pool, rtt).await;
+        // 3. add_delete_file × N
+        for _ in 0..delete_files {
+            one_metastore_call(pool, rtt).await;
+        }
+        // 4. add_insert_records_batch
+        one_metastore_call(pool, rtt).await;
+    }
+}
+
+/// Lane B: current behavior — `commit_on_conflict_deletions` as implemented:
+/// `increment_sequence_number`, `BEGIN`, per-delete-file `INSERT`, one
+/// insert-record chunk `INSERT`, then `COMMIT`.
+async fn run_proposed(pool: &Arc<Mutex<()>>, upserts: usize, delete_files: usize, rtt: Duration) {
+    for _ in 0..upserts {
+        // 1. increment_sequence_number
+        one_metastore_call(pool, rtt).await;
+        // 2. begin_transaction
+        one_metastore_call(pool, rtt).await;
+        // 3. delete-file INSERT × N
+        for _ in 0..delete_files {
+            one_metastore_call(pool, rtt).await;
+        }
+        // 4. one insert-record chunk INSERT
+        one_metastore_call(pool, rtt).await;
+        // 5. commit
+        one_metastore_call(pool, rtt).await;
+    }
+}
+
+fn bench_apply_on_conflict_rpc_ceiling(c: &mut Criterion) {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(4)
+        .enable_all()
+        .build()
+        .expect("tokio runtime");
+
+    let mut group = c.benchmark_group("apply_on_conflict_rpc_ceiling");
+    group.sample_size(10);
+
+    for &(rtt_label, rtt) in RTTS {
+        for &delete_files in DELETE_FILES_PER_UPSERT {
+            let upserts_total = u64::try_from(UPSERTS_PER_ITERATION).unwrap_or(u64::MAX);
+            group.throughput(Throughput::Elements(upserts_total));
+
+            let id = format!("delete_files={delete_files}/{rtt_label}");
+            let pool_a = Arc::new(Mutex::new(()));
+            group.bench_with_input(
+                BenchmarkId::new("no_txn_baseline", &id),
+                &delete_files,
+                |b, &delete_files| {
+                    b.to_async(&rt).iter(|| {
+                        let pool = Arc::clone(&pool_a);
+                        async move {
+                            run_current(&pool, UPSERTS_PER_ITERATION, delete_files, rtt).await;
+                            black_box(pool);
+                        }
+                    });
+                },
+            );
+            let pool_b = Arc::new(Mutex::new(()));
+            group.bench_with_input(
+                BenchmarkId::new("atomic_txn_calls", &id),
+                &delete_files,
+                |b, &delete_files| {
+                    b.to_async(&rt).iter(|| {
+                        let pool = Arc::clone(&pool_b);
+                        async move {
+                            run_proposed(&pool, UPSERTS_PER_ITERATION, delete_files, rtt).await;
+                            black_box(pool);
+                        }
+                    });
+                },
+            );
+        }
+    }
+    group.finish();
+}
+
+criterion_group!(benches, bench_apply_on_conflict_rpc_ceiling);
+criterion_main!(benches);