fix: add dir_listing_to_manifest_migration_enabled flag to avoid extra object store calls (lance-format#6507)

## Summary

- Adds `dir_listing_to_manifest_migration_enabled` flag (default:
`false`) to `DirectoryNamespaceBuilder` and `DirectoryNamespace`
- When `false` and both `manifest_enabled` and `dir_listing_enabled` are
`true`, root-level table operations (`table_exists`, `describe_table`,
`list_tables`) skip the manifest check and use directory listing
directly, avoiding extra object store listing calls
- When `true`, preserves the existing hybrid behavior of checking
manifest first then falling back to directory listing
- Includes a test with a counting object store wrapper verifying only a
single `list_with_delimiter` call is made for root-level table
operations without migration mode

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jackye1995

python/python/tests/test_namespace_dir.py

@@ -815,6 +815,7 @@ def test_connect_with_properties(self, use_custom):
             "root": f"memory://test_connect_{unique_id}",
             "manifest_enabled": "true",
             "dir_listing_enabled": "true",
+            "dir_listing_to_manifest_migration_enabled": "true",
         }
 
         # Connect via lance.namespace.connect

rust/lance-core/src/utils/testing.rs

@@ -17,7 +17,11 @@ use std::collections::HashMap;
 use std::fmt::Debug;
 use std::future;
 use std::ops::Range;
-use std::sync::{Arc, Mutex};
+use std::pin::Pin;
+use std::sync::{
+    Arc, Mutex,
+    atomic::{AtomicUsize, Ordering},
+};
 
 // A policy function takes in the name of the operation (e.g. "put") and the location
 // that is being accessed / modified and returns an optional error.
@@ -121,6 +125,42 @@ impl std::fmt::Display for ProxyObjectStore {
     }
 }
 
+/// An object store wrapper that counts listing operations.
+///
+/// This increments the shared counter for both `list` and `list_with_delimiter`
+/// so tests can observe all listing-based directory and version discovery calls.
+#[derive(Debug)]
+pub struct CountingObjectStore {
+    target: Arc<dyn ObjectStore>,
+    listing_count: Arc<AtomicUsize>,
+}
+
+impl CountingObjectStore {
+    pub fn new(target: Arc<dyn ObjectStore>, listing_count: Arc<AtomicUsize>) -> Self {
+        Self {
+            target,
+            listing_count,
+        }
+    }
+
+    fn record_listing(&self) {
+        self.listing_count.fetch_add(1, Ordering::SeqCst);
+    }
+
+    fn delegate_list(
+        &self,
+        prefix: Option<&Path>,
+    ) -> Pin<Box<dyn futures::Stream<Item = OSResult<ObjectMeta>> + Send>> {
+        self.target.list(prefix)
+    }
+}
+
+impl std::fmt::Display for CountingObjectStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "CountingObjectStore({})", self.target)
+    }
+}
+
 #[async_trait]
 impl ObjectStore for ProxyObjectStore {
     async fn put_opts(
@@ -204,3 +244,65 @@ impl ObjectStore for ProxyObjectStore {
         self.target.copy_if_not_exists(from, to).await
     }
 }
+
+#[async_trait]
+impl ObjectStore for CountingObjectStore {
+    async fn put_opts(
+        &self,
+        location: &Path,
+        bytes: PutPayload,
+        opts: PutOptions,
+    ) -> OSResult<PutResult> {
+        self.target.put_opts(location, bytes, opts).await
+    }
+
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOptions,
+    ) -> OSResult<Box<dyn MultipartUpload>> {
+        self.target.put_multipart_opts(location, opts).await
+    }
+
+    async fn get_opts(&self, location: &Path, options: GetOptions) -> OSResult<GetResult> {
+        self.target.get_opts(location, options).await
+    }
+
+    async fn get_range(&self, location: &Path, range: Range<u64>) -> OSResult<Bytes> {
+        self.target.get_range(location, range).await
+    }
+
+    async fn get_ranges(&self, location: &Path, ranges: &[Range<u64>]) -> OSResult<Vec<Bytes>> {
+        self.target.get_ranges(location, ranges).await
+    }
+
+    async fn head(&self, location: &Path) -> OSResult<ObjectMeta> {
+        self.target.head(location).await
+    }
+
+    async fn delete(&self, location: &Path) -> OSResult<()> {
+        self.target.delete(location).await
+    }
+
+    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, OSResult<ObjectMeta>> {
+        self.record_listing();
+        self.delegate_list(prefix).boxed()
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> OSResult<ListResult> {
+        self.record_listing();
+        self.target.list_with_delimiter(prefix).await
+    }
+
+    async fn copy(&self, from: &Path, to: &Path) -> OSResult<()> {
+        self.target.copy(from, to).await
+    }
+
+    async fn rename(&self, from: &Path, to: &Path) -> OSResult<()> {
+        self.target.rename(from, to).await
+    }
+
+    async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> OSResult<()> {
+        self.target.copy_if_not_exists(from, to).await
+    }
+}