perf(catalog): probe the next __manifest version by name instead of listing

changelog.d/catalog-freshness-probe-heads-next-version.performance.md

@@ -0,0 +1,8 @@
+- A read on a warm graph handle checks for new commits by asking for its
+  `__manifest` version and the next one by name instead of listing the version
+  directory. On S3-compatible storage that listing returned up to 1,000 keys
+  on every read once a branch passed 1,000 commits; the check is now two small
+  HEAD requests sent together while nothing changed. See the
+  [catalog freshness probe][freshness-probe].
+
+[freshness-probe]: ../docs/dev/lance.md#current-compatibility-fences

crates/omnigraph-catalog/src/lib.rs

@@ -855,14 +855,55 @@ impl CapturedManifestProbe {
     }
 }
 
+/// The latest attached version of `dataset`'s location (main or the branch
+/// tree it is checked out on), asked for by name from the version it holds.
+///
+/// `latest_version_id` lists `_versions/`. On a store whose listing is
+/// ordered (S3, GCS, Azure) that is one request returning up to a thousand
+/// keys however few of them are new, and a warm read pays it on every query.
+/// Attached versions are contiguous: Lance writes version `v + 1` only with a
+/// put-if-absent over a read of `v`, and nothing deletes a version of a live
+/// branch's `__manifest`, so while the held version `v` exists the first
+/// absent name above it ends the history. (`cleanup` collects a retired
+/// branch's whole tree; the named-branch probe's identifier read catches
+/// that.) A reclamation of old `__manifest` versions must keep every version
+/// above one a live handle holds.
+///
+/// The probe HEADs `v` and `v + 1` together, then `v + 2` when `v + 1`
+/// exists. `v` must answer: a removed or unreadable `__manifest` answers "not
+/// found" for every name, and only the listing reports it as the failure it
+/// is. A gap of more than two versions, a detached handle, an absent `v` or
+/// any error falls back to the listing, which stays the authority. Like the
+/// listing, the probe relies on the store showing a completed write to a
+/// later request.
+async fn latest_attached_version(dataset: &Dataset) -> lance::Result<u64> {
+    let held = dataset.version().version;
+    if !lance_table::format::is_detached_version(held) {
+        let scheme = dataset.manifest_location().naming_scheme;
+        let base = dataset.branch_location().path;
+        let store = dataset.object_store(None).await?;
+        let [at, next, after] = [held, held + 1, held + 2].map(|v| scheme.manifest_path(&base, v));
+        let (held_present, next_present) = futures::join!(store.exists(&at), store.exists(&next));
+        match (held_present, next_present) {
+            (Ok(true), Ok(false)) => return Ok(held),
+            (Ok(true), Ok(true)) => {
+                if let Ok(false) = store.exists(&after).await {
+                    return Ok(held + 1);
+                }
+            }
+            _ => {}
+        }
+    }
+    dataset.latest_version_id().await
+}
+
 async fn probe_dataset_latest_incarnation(
     dataset: &Dataset,
     active_branch: Option<&str>,
 ) -> Result<ManifestIncarnation> {
     if active_branch.is_none() {
         return Ok(ManifestIncarnation {
-            version: dataset
-                .latest_version_id()
+            version: latest_attached_version(dataset)
                 .await
                 .map_err(OmniError::storage)?,
             e_tag: dataset.manifest_location().e_tag.clone(),
@@ -879,8 +920,7 @@ async fn probe_dataset_latest_incarnation(
     // yields the replacement identifier rather than a false match to the held
     // lifetime.
     let held = async {
-        let version = dataset
-            .latest_version_id()
+        let version = latest_attached_version(dataset)
             .await
             .map_err(|error| branch_ref_error(error, branch))?;
         let native = dataset.manifest().branch.as_deref().ok_or_else(|| {

crates/omnigraph-catalog/src/tests.rs

@@ -9536,15 +9536,17 @@ impl lance::io::WrappingObjectStore for SmallScanProbeMarker {
     }
 }
 
-/// Counts the requests `__history` is sent by verb, "not found" GETs included
-/// (Lance's tracker records only successful GETs); with `refuse_suffix` a
-/// suffix range is refused before it is sent, as `object_store`'s Azure client does.
+/// Counts the requests a wrapped store (`__history` through [`Self::probes`])
+/// is sent by verb, "not found" GETs and HEADs included (Lance's tracker
+/// records only successful GETs); with `refuse_suffix` a suffix range is
+/// refused before it is sent, as `object_store`'s Azure client does.
 #[derive(Debug, Default, Clone)]
 pub(super) struct HistoryGets {
     refuse_suffix: bool,
     sent: Arc<std::sync::atomic::AtomicUsize>,
     absent: Arc<std::sync::atomic::AtomicUsize>,
     heads: Arc<std::sync::atomic::AtomicUsize>,
+    lists: Arc<std::sync::atomic::AtomicUsize>,
 }
 
 impl HistoryGets {
@@ -9564,6 +9566,11 @@ impl HistoryGets {
         };
         (take(&self.sent), take(&self.absent), take(&self.heads))
     }
+
+    /// The listings since the last call.
+    pub(super) fn drain_lists(&self) -> usize {
+        self.lists.swap(0, std::sync::atomic::Ordering::Relaxed)
+    }
 }
 
 impl lance::io::WrappingObjectStore for HistoryGets {
@@ -9649,13 +9656,19 @@ impl object_store::ObjectStore for HistoryGetStore {
         &self,
         prefix: Option<&object_store::path::Path>,
     ) -> futures::stream::BoxStream<'static, object_store::Result<object_store::ObjectMeta>> {
+        self.gets
+            .lists
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         self.original.list(prefix)
     }
 
     async fn list_with_delimiter(
         &self,
         prefix: Option<&object_store::path::Path>,
     ) -> object_store::Result<object_store::ListResult> {
+        self.gets
+            .lists
+            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
         self.original.list_with_delimiter(prefix).await
     }
 
@@ -9669,6 +9682,95 @@ impl object_store::ObjectStore for HistoryGetStore {
     }
 }
 
+/// The freshness probe asks for the held version and the ones after it by
+/// name. `latest_version_id` lists `_versions/`, and on a store whose listing
+/// is ordered (S3, and Lance's in-memory stores) that one request carries up
+/// to a thousand keys on every warm read. A removed `__manifest` still fails
+/// the probe: every name is absent there, so the listing answers.
+#[tokio::test]
+async fn freshness_probe_asks_for_the_next_version_by_name() {
+    use arrow_array::ArrayRef;
+    async fn commit(uri: &str, mode: WriteMode, enable_v2_manifest_paths: bool) {
+        let batch = RecordBatch::try_from_iter([(
+            "value",
+            Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+        )])
+        .unwrap();
+        let schema = batch.schema();
+        Dataset::write(
+            RecordBatchIterator::new(vec![Ok(batch)], schema),
+            uri,
+            Some(WriteParams {
+                mode,
+                enable_v2_manifest_paths,
+                skip_auto_cleanup: true,
+                ..Default::default()
+            }),
+        )
+        .await
+        .unwrap();
+    }
+    for v2_paths in [true, false] {
+        let uri = format!("shared-memory://freshness-{}/__manifest", ulid::Ulid::new());
+        let append = || commit(&uri, WriteMode::Append, v2_paths);
+        commit(&uri, WriteMode::Create, v2_paths).await;
+        let sent = HistoryGets::default();
+        let held = DatasetBuilder::from_uri(&uri)
+            .with_store_params(lance::io::ObjectStoreParams {
+                object_store_wrapper: Some(Arc::new(sent.clone())),
+                ..Default::default()
+            })
+            .load()
+            .await
+            .unwrap();
+        assert_eq!(held.version().version, 1);
+        sent.drain();
+        sent.drain_lists();
+
+        // Unchanged: version 1 is found and version 2 is absent.
+        let latest = probe_dataset_latest_incarnation(&held, None).await.unwrap();
+        assert_eq!(latest.version, 1, "v2_paths={v2_paths}");
+        assert_eq!(sent.drain(), (0, 1, 2), "v2_paths={v2_paths}");
+        assert_eq!(sent.drain_lists(), 0, "v2_paths={v2_paths}");
+
+        // Another writer commits once: versions 1 and 2 are found, 3 is absent.
+        append().await;
+        let latest = probe_dataset_latest_incarnation(&held, None).await.unwrap();
+        assert_eq!(latest.version, 2, "v2_paths={v2_paths}");
+        assert_eq!(sent.drain(), (0, 1, 3), "v2_paths={v2_paths}");
+        assert_eq!(sent.drain_lists(), 0, "v2_paths={v2_paths}");
+
+        // A gap of more than two versions: the listing answers.
+        append().await;
+        append().await;
+        let latest = probe_dataset_latest_incarnation(&held, None).await.unwrap();
+        assert_eq!(latest.version, 4, "v2_paths={v2_paths}");
+        assert_eq!(sent.drain(), (0, 0, 3), "v2_paths={v2_paths}");
+        assert_eq!(sent.drain_lists(), 1, "v2_paths={v2_paths}");
+
+        // A removed `__manifest`: no name answers, and the listing's "not
+        // found" fails the probe instead of reporting the held version.
+        held.object_store(None)
+            .await
+            .unwrap()
+            .remove_dir_all(held.branch_location().path)
+            .await
+            .unwrap();
+        sent.drain();
+        sent.drain_lists();
+        let error = probe_dataset_latest_incarnation(&held, None)
+            .await
+            .unwrap_err();
+        assert_eq!(
+            error.storage_failure().map(|failure| failure.kind),
+            Some(StorageFailureKind::NotFound),
+            "v2_paths={v2_paths}: {error}"
+        );
+        assert_eq!(sent.drain(), (0, 2, 2), "v2_paths={v2_paths}");
+        assert_eq!(sent.drain_lists(), 1, "v2_paths={v2_paths}");
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct SmallScanIo {
     requests: u64,

crates/omnigraph-dst/cost_table.txt

@@ -1,66 +1,63 @@
-AddFriend                    l.get        calls=110
-AddFriend                    l.list       calls=16
+AddFriend                    l.get        calls=117
+AddFriend                    l.list       calls=9
 AddFriend                    l.put        calls=24
-BranchCreate                 l.get        calls=4
-BranchCreate                 l.list       calls=8
+BranchCreate                 l.get        calls=6
+BranchCreate                 l.list       calls=6
 BranchCreate                 l.put        calls=6
 BranchDelete                 l.delete     calls=1
 BranchDelete                 l.get        calls=31
 BranchDelete                 l.list       calls=6
 BranchDelete                 l.put        calls=2
 Cleanup                      l.delete     calls=20
-Cleanup                      l.get        calls=179
-Cleanup                      l.list       calls=172
-DeletePerson                 l.get        calls=3
-DeletePerson                 l.list       calls=3
-EnsureIndices                l.get        calls=4
-EnsureIndices                l.list       calls=2
-InsertLegacy                 l.get        calls=16
-InsertLegacy                 l.list       calls=11
+Cleanup                      l.get        calls=187
+Cleanup                      l.list       calls=164
+DeletePerson                 l.get        calls=6
+EnsureIndices                l.get        calls=6
+InsertLegacy                 l.get        calls=22
+InsertLegacy                 l.list       calls=5
 InsertLegacy                 l.put        calls=18
-InsertV                      l.get        calls=1
-InsertV                      l.list       calls=3
+InsertV                      l.get        calls=3
+InsertV                      l.list       calls=1
 InsertV                      l.put        calls=6
 Optimize                     a.exists     calls=2
 Optimize                     a.put        calls=1
-Optimize                     l.get        calls=54
-Optimize                     l.list       calls=17
+Optimize                     l.get        calls=63
+Optimize                     l.list       calls=8
 Optimize                     l.put        calls=50
-RemoveFriendshipsFrom        l.get        calls=13
-RemoveFriendshipsFrom        l.list       calls=8
+RemoveFriendshipsFrom        l.get        calls=20
+RemoveFriendshipsFrom        l.list       calls=1
 RemoveFriendshipsFrom        l.put        calls=5
-UpdateV                      l.get        calls=4
-UpdateV                      l.list       calls=4
+UpdateV                      l.get        calls=8
 _audit                       a.delete     calls=1
 _audit                       a.get        calls=2
 _audit                       a.list       calls=2
 _audit                       a.put_if_absent calls=1
-_audit                       l.get        calls=728
-_audit                       l.list       calls=113
-_close                       l.list       calls=3
-_history                     l.list       calls=30
+_audit                       l.get        calls=749
+_audit                       l.list       calls=92
+_close                       l.get        calls=3
+_history                     l.get        calls=30
 _setup                       a.delete     calls=3
 _setup                       a.exists     calls=2
 _setup                       a.list       calls=1
 _setup                       a.put_if_absent calls=3
-_setup                       l.get        calls=45
-_setup                       l.list       calls=18
+_setup                       l.get        calls=50
+_setup                       l.list       calls=13
 _setup                       l.put        calls=35
 _verify                      a.get        calls=10
-_verify                      l.get        calls=1425
-_verify                      l.list       calls=429
+_verify                      l.get        calls=1620
+_verify                      l.list       calls=254
 MergeTargetHistory2          l.delete     calls=3
-MergeTargetHistory2          l.get        calls=111
-MergeTargetHistory2          l.list       calls=12
+MergeTargetHistory2          l.get        calls=116
+MergeTargetHistory2          l.list       calls=7
 MergeTargetHistory2          l.put        calls=12
 MergeTargetHistory32         l.delete     calls=3
-MergeTargetHistory32         l.get        calls=111
-MergeTargetHistory32         l.list       calls=12
+MergeTargetHistory32         l.get        calls=116
+MergeTargetHistory32         l.list       calls=7
 MergeTargetHistory32         l.put        calls=12
 MergeBackAfterThreeWay       l.delete     calls=3
-MergeBackAfterThreeWay       l.get        calls=129
-MergeBackAfterThreeWay       l.list       calls=17
+MergeBackAfterThreeWay       l.get        calls=134
+MergeBackAfterThreeWay       l.list       calls=12
 MergeBackAfterThreeWay       l.put        calls=12
-BranchWrite                  l.get        calls=21
-BranchWrite                  l.list       calls=5
+BranchWrite                  l.get        calls=23
+BranchWrite                  l.list       calls=3
 BranchWrite                  l.put        calls=6

crates/omnigraph-gqt/tests/runner_dispatch.rs

@@ -1366,8 +1366,11 @@ fn measure_counts_schema_contract_requests_issue_817() {
     // first write, and Lance's detached commit loads the base's indexes. Behind
     // a strict insert's inline transaction the index section lies outside the
     // tail block the open read, and Lance opens the manifest again without the
-    // size it already learned: that HEAD is the one read a step may repeat.
-    let repeated_manifest_heads = |step: u64| -> u64 {
+    // size it already learned: that HEAD is one read a step may repeat. The
+    // other is the catalog freshness probe's HEAD of the held `__manifest`
+    // version, which a write makes when it captures its branch and again when
+    // it revalidates that capture.
+    let repeated_heads = |step: u64, class: &str| -> u64 {
         let group = measurements
             .iter()
             .find(|group| group["slot"] == "step" && group["step"] == step)
@@ -1377,7 +1380,7 @@ fn measure_counts_schema_contract_requests_issue_817() {
             .as_array()
             .unwrap()
             .iter()
-            .filter(|request| request["class"] == "table_meta" && request["verb"] == "head")
+            .filter(|request| request["class"] == class && request["verb"] == "head")
             .filter(|request| !seen.insert(request["path"].as_str().unwrap().to_string()))
             .count() as u64
     };
@@ -1389,14 +1392,20 @@ fn measure_counts_schema_contract_requests_issue_817() {
         );
         assert_eq!(
             counts["repeat_reads"],
-            repeated_manifest_heads(step),
+            repeated_heads(step, "table_meta") + repeated_heads(step, "manifest_meta"),
             "step {step}: neither the inline contract nor table data repeats a read, \
-             except the reopen of an indexed table's manifest"
+             except the reopen of an indexed table's manifest and the freshness \
+             probe's HEAD of the held `__manifest` version"
         );
     }
     assert_eq!(
-        repeated_manifest_heads(2),
+        repeated_heads(2, "table_meta"),
         1,
         "the second insert reopens the first insert's manifest for its index section"
     );
+    assert_eq!(
+        repeated_heads(3, "manifest_meta"),
+        0,
+        "a read probes its branch's freshness once"
+    );
 }