test(python): cover concurrent S3 table opens (#3833)

## Summary

- add regression coverage for the reported synchronous Python workload
with 32 simultaneous `open_table` calls
- verify every independently opened S3-backed table handle can read
through the connection's shared session and object-store client

## Root cause

In Python v0.13.0, each synchronous table handle lazily constructed its
own Lance dataset. Opening many handles in parallel therefore triggered
independent S3 client construction and bucket-region resolution, which
failed under thread pressure. The current Rust-backed connection path
owns a shared Lance session and retains its object-store handle, so
table opens reuse the existing S3 client; these tests lock in that
behavior through the public Python API and a causal Session-registry
invariant.

## Validation

- `uvx --from 'ruff==0.15.20' ruff format --check
python/tests/test_s3.py`
- `uvx --from 'ruff==0.15.20' ruff check .`
- `cargo fmt --all`
- `cargo test --quiet --features remote -p lancedb
test_concurrent_open_table_reuses_connection_object_store`
- `cargo check --quiet --features remote --tests --examples`
- equivalent 32-thread `open_table(...).count_rows()` workload against a
local database
- targeted S3 test collected successfully locally; execution requires
the CI LocalStack service, which is unavailable in this runner

Fixes #1786

<!-- lance-gatekeeper-fix:v1 agent=d311f3c7151f77ae22b4997702e7b7db
generation=1 -->

---------

Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
Co-authored-by: Xuanwo <github@xuanwo.io>

python/python/tests/test_s3.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import copy
+from concurrent.futures import ThreadPoolExecutor
 from datetime import timedelta
 import threading
 
@@ -86,6 +87,25 @@ def test_s3_lifecycle(s3_bucket: str):
     asyncio.run(test())
 
 
+@pytest.mark.s3_test
+def test_concurrent_open_table(s3_bucket: str):
+    uri = f"s3://{s3_bucket}/test_concurrent_open_table"
+    db = lancedb.connect(uri, storage_options=copy.copy(CONFIG))
+    db.create_table("test", pa.table({"x": [1, 2, 3]}))
+
+    num_workers = 32
+    barrier = threading.Barrier(num_workers)
+
+    def open_and_count(_):
+        barrier.wait()
+        return db.open_table("test").count_rows()
+
+    with ThreadPoolExecutor(max_workers=num_workers) as pool:
+        row_counts = list(pool.map(open_and_count, range(num_workers)))
+
+    assert row_counts == [3] * num_workers
+
+
 @pytest.fixture()
 def kms_key():
     kms = get_boto3_client("kms", endpoint_url=CONFIG["aws_endpoint"])

rust/lancedb/src/database/listing.rs

@@ -1476,7 +1476,7 @@ mod tests {
     use crate::table::{AnyQuery, WriteOptions};
     use arrow_array::{Int32Array, RecordBatch, StringArray};
     use arrow_schema::{DataType, Field, Schema, SchemaRef};
-    use futures::{TryStreamExt, stream::once};
+    use futures::{TryStreamExt, future::try_join_all, stream::once};
     use std::path::PathBuf;
     use std::sync::Arc;
     use std::time::Duration;
@@ -1614,6 +1614,59 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_concurrent_open_table_reuses_connection_object_store() {
+        let tempdir = tempdir().unwrap();
+        let uri = tempdir.path().to_str().unwrap();
+        let session = Arc::new(lance::session::Session::default());
+        let request = ConnectRequest {
+            uri: uri.to_string(),
+            #[cfg(feature = "remote")]
+            client_config: Default::default(),
+            options: Default::default(),
+            namespace_client_properties: Default::default(),
+            manifest_enabled: false,
+            read_consistency_interval: None,
+            session: Some(session.clone()),
+        };
+        let db = ListingDatabase::connect_with_options(&request)
+            .await
+            .unwrap();
+        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));
+        db.create_table(CreateTableRequest {
+            name: "test".to_string(),
+            namespace_path: vec![],
+            data: Box::new(RecordBatch::new_empty(schema)) as Box<dyn Scannable>,
+            mode: CreateTableMode::Create,
+            write_options: Default::default(),
+            location: None,
+            namespace_client: None,
+        })
+        .await
+        .unwrap();
+
+        let before = session.store_registry().stats();
+        let opened_tables = try_join_all((0..32).map(|_| {
+            db.open_table(OpenTableRequest {
+                name: "test".to_string(),
+                namespace_path: vec![],
+                index_cache_size: None,
+                lance_read_params: None,
+                location: None,
+                namespace_client: None,
+                managed_versioning: None,
+            })
+        }))
+        .await
+        .unwrap();
+        let after = session.store_registry().stats();
+
+        assert_eq!(opened_tables.len(), 32);
+        assert_eq!(after.misses, before.misses);
+        assert_eq!(after.active_stores, before.active_stores);
+        assert!(after.hits >= before.hits + 32);
+    }
+
     #[tokio::test]
     async fn test_listing_database_root_ops_do_not_create_manifest() {
         let tempdir = tempdir().unwrap();

rust/lancedb/src/job.rs

@@ -47,6 +47,10 @@ impl TerminalResult {
         }
     }
 
+    pub(crate) fn value(&self) -> Option<&Value> {
+        self.value.as_ref()
+    }
+
     fn decode<T: DeserializeOwned>(self) -> Result<T> {
         let value = self.value.ok_or_else(|| match &self.request_id {
             Some(request_id) => Error::Http {

rust/lancedb/src/remote/table/blobs.rs

@@ -6,6 +6,7 @@
 use std::ops::Range;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::Duration;
 
 use arrow_array::{Array, LargeBinaryArray};
 use arrow_schema::DataType;
@@ -20,7 +21,7 @@ use crate::error::Result;
 use crate::remote::client::{HttpSend, RequestResultExt, RestfulLanceDbClient};
 use crate::table::BaseTable;
 
-use super::{FreshnessHeaders, RemoteTable};
+use super::{FreshnessHeaders, FreshnessState, RemoteTable, freshness_headers_snapshot};
 
 #[derive(Debug, Clone, Copy)]
 enum RangeRequestMode {
@@ -43,7 +44,10 @@ struct TableBlobRangeRequester<S: HttpSend> {
     path: String,
     version: Option<u64>,
     branch: Option<String>,
-    freshness: FreshnessHeaders,
+    freshness: Arc<std::sync::Mutex<FreshnessState>>,
+    parent_freshness: Arc<std::sync::Mutex<FreshnessState>>,
+    parent_freshness_request: FreshnessHeaders,
+    read_consistency_interval: Option<Duration>,
 }
 
 #[async_trait::async_trait]
@@ -53,8 +57,9 @@ impl<S: HttpSend> BlobRangeRequester for TableBlobRangeRequester<S> {
         range_header: &str,
         mode: RangeRequestMode,
     ) -> Result<(String, Response)> {
-        let mut request = self
-            .freshness
+        let freshness_request =
+            freshness_headers_snapshot(&self.freshness, self.read_consistency_interval);
+        let mut request = freshness_request
             .apply(self.client.get(&self.path))
             .header(header::RANGE, range_header);
         if let Some(version) = self.version {
@@ -71,6 +76,9 @@ impl<S: HttpSend> BlobRangeRequester for TableBlobRangeRequester<S> {
             return Ok((request_id, response));
         }
         let response = self.client.check_response(&request_id, response).await?;
+        freshness_request.observe_headers(&self.freshness, response.headers());
+        self.parent_freshness_request
+            .observe_headers(&self.parent_freshness, response.headers());
         Ok((request_id, response))
     }
 }
@@ -361,18 +369,21 @@ impl<S: HttpSend> RemoteTable<S> {
                 message: "fetch_blobs is not supported on this LanceDB Cloud server".into(),
             });
         }
-        let version = self.current_version().await;
+        let read_snapshot = self.snapshot_read_state().await;
         let mut body = serde_json::json!({
-            "version": version,
+            "version": read_snapshot.version,
             "column": column,
             "row_ids": row_ids,
         });
         self.apply_branch_body(&mut body);
 
         let request = self
-            .post_read(&format!("/v1/table/{}/fetch_blobs/", self.identifier))
+            .client
+            .post(&format!("/v1/table/{}/fetch_blobs/", self.identifier))
             .json(&body);
-        let (request_id, response) = self.send(request, true).await?;
+        let (request_id, response) = self
+            .send_with_freshness(request, true, read_snapshot.freshness)
+            .await?;
         let mut stream = self.read_arrow_response(&request_id, response).await?;
 
         let mut blob_chunks: Vec<Arc<dyn Array>> = Vec::new();
@@ -448,8 +459,7 @@ impl<S: HttpSend> RemoteTable<S> {
             });
         }
 
-        let version = self.current_version().await;
-        let freshness = self.snapshot_freshness_headers();
+        let read_snapshot = self.snapshot_read_state().await;
         let encoded_column = urlencoding::encode(column);
         let requesters = row_ids
             .iter()
@@ -461,9 +471,12 @@ impl<S: HttpSend> RemoteTable<S> {
                 let requester: Arc<dyn BlobRangeRequester> = Arc::new(TableBlobRangeRequester {
                     client: self.client.clone(),
                     path,
-                    version,
+                    version: read_snapshot.version,
                     branch: self.branch.clone(),
-                    freshness,
+                    freshness: Arc::new(std::sync::Mutex::new(read_snapshot.freshness_state)),
+                    parent_freshness: self.freshness.clone(),
+                    parent_freshness_request: read_snapshot.freshness,
+                    read_consistency_interval: self.client.read_consistency_interval,
                 });
                 requester
             })
@@ -685,6 +698,46 @@ mod tests {
         assert!(requests.lock().unwrap().contains(&"bytes=5-11".to_string()));
     }
 
+    #[tokio::test]
+    async fn remote_blob_file_keeps_the_open_timeline_after_parent_checkout() {
+        let range_requests = Arc::new(StdMutex::new(Vec::new()));
+        let captured = range_requests.clone();
+        let table = RemoteTable::new_mock(
+            "my_table".to_string(),
+            move |request| match request.url().path() {
+                "/v1/table/my_table/describe/" => http::Response::builder()
+                    .status(200)
+                    .body(r#"{"version":5,"schema":{"fields":[]}}"#.as_bytes().to_vec())
+                    .unwrap(),
+                "/v1/table/my_table/blob/image/10/bytes" => {
+                    captured.lock().unwrap().push((
+                        request.url().query().unwrap_or_default().to_string(),
+                        request.headers().clone(),
+                    ));
+                    range_response(&request, PAYLOAD)
+                }
+                path => panic!("unexpected path: {path}"),
+            },
+            Some(Version::new(0, 5, 0)),
+        );
+
+        table.checkout(5).await.unwrap();
+        let file = table
+            .fetch_blob_files_impl("image", &[10])
+            .await
+            .unwrap()
+            .pop()
+            .flatten()
+            .unwrap();
+        table.checkout_latest().await.unwrap();
+        file.read_range(5..12).await.unwrap();
+
+        let requests = range_requests.lock().unwrap();
+        let (query, headers) = requests.last().unwrap();
+        assert!(query.contains("version=5"));
+        assert!(!headers.contains_key("x-lancedb-min-timestamp"));
+    }
+
     #[tokio::test]
     async fn remote_blob_file_reuses_sequential_response_until_seek() {
         let requests = Arc::new(StdMutex::new(Vec::new()));

rust/lancedb/src/remote/table/insert.rs

@@ -24,7 +24,10 @@ use lance::io::exec::utils::InstrumentedRecordBatchStreamAdapter;
 use crate::Error;
 use crate::remote::ARROW_STREAM_CONTENT_TYPE;
 use crate::remote::client::{HttpSend, RestfulLanceDbClient, Sender};
-use crate::remote::table::{MergeInsertRequest, REQUEST_TIMEOUT_HEADER, RemoteTable};
+use crate::remote::table::{
+    FreshnessHeaders, FreshnessState, MergeInsertRequest, REQUEST_TIMEOUT_HEADER, RemoteTable,
+    freshness_headers_snapshot,
+};
 use crate::table::datafusion::insert::COUNT_SCHEMA;
 use crate::table::write_progress::WriteProgressTracker;
 use crate::table::{AddResult, MergeResult};
@@ -54,6 +57,38 @@ pub enum WriteResult {
     Merge(MergeResult),
 }
 
+#[derive(Debug, Clone, Default)]
+struct WriteFreshness {
+    state: Option<Arc<Mutex<FreshnessState>>>,
+    read_consistency_interval: Option<Duration>,
+}
+
+impl WriteFreshness {
+    fn prepare(
+        &self,
+        request: reqwest::RequestBuilder,
+    ) -> (reqwest::RequestBuilder, Option<FreshnessHeaders>) {
+        match &self.state {
+            Some(state) => {
+                let freshness_request =
+                    freshness_headers_snapshot(state, self.read_consistency_interval);
+                (freshness_request.apply(request), Some(freshness_request))
+            }
+            None => (request, None),
+        }
+    }
+
+    fn observe(
+        &self,
+        freshness_request: Option<FreshnessHeaders>,
+        headers: &reqwest::header::HeaderMap,
+    ) {
+        if let (Some(state), Some(freshness_request)) = (&self.state, freshness_request) {
+            freshness_request.observe_headers(state, headers);
+        }
+    }
+}
+
 /// ExecutionPlan for streaming a write (add or merge_insert) to a remote
 /// LanceDB table.
 ///
@@ -71,6 +106,7 @@ pub struct RemoteWriteExec<S: HttpSend = Sender> {
     table_name: String,
     identifier: String,
     client: RestfulLanceDbClient<S>,
+    freshness: WriteFreshness,
     input: Arc<dyn ExecutionPlan>,
     op: WriteOp,
     properties: Arc<PlanProperties>,
@@ -170,6 +206,7 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
             table_name,
             identifier,
             client,
+            freshness: WriteFreshness::default(),
             input,
             op,
             properties: Arc::new(properties),
@@ -183,6 +220,18 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
         }
     }
 
+    pub(super) fn with_freshness(
+        mut self,
+        state: Arc<Mutex<FreshnessState>>,
+        read_consistency_interval: Option<Duration>,
+    ) -> Self {
+        self.freshness = WriteFreshness {
+            state: Some(state),
+            read_consistency_interval,
+        };
+        self
+    }
+
     /// Get the add result after execution, if this exec ran an insert.
     pub fn add_result(&self) -> Option<AddResult> {
         match self
@@ -285,6 +334,7 @@ impl<S: HttpSend + 'static> RemoteWriteExec<S> {
 /// each threading the same handful of arguments.
 struct PartRequestCtx<'a, S: HttpSend> {
     client: &'a RestfulLanceDbClient<S>,
+    freshness: &'a WriteFreshness,
     identifier: &'a str,
     table_name: &'a str,
     upload_id: &'a str,
@@ -352,7 +402,11 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
     }
 
     /// Build the `/insert` request for a single multipart part.
-    fn build_part_request(&self, part_id: &str, body: reqwest::Body) -> reqwest::RequestBuilder {
+    fn build_part_request(
+        &self,
+        part_id: &str,
+        body: reqwest::Body,
+    ) -> (reqwest::RequestBuilder, Option<FreshnessHeaders>) {
         let mut request = self
             .client
             .post(&format!("/v1/table/{}/insert/", self.identifier))
@@ -368,12 +422,16 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
         if let Some(b) = self.branch {
             request = request.query(&[("branch", b)]);
         }
-        request.body(body)
+        self.freshness.prepare(request.body(body))
     }
 
     /// Send a single part's request and drain the response, mapping HTTP and
     /// table-not-found errors into `DataFusionError`.
-    async fn send_part_request(&self, request: reqwest::RequestBuilder) -> DataFusionResult<()> {
+    async fn send_part_request(
+        &self,
+        request: reqwest::RequestBuilder,
+        freshness_request: Option<FreshnessHeaders>,
+    ) -> DataFusionResult<()> {
         let (request_id, response) = self
             .client
             .send(request)
@@ -388,6 +446,8 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
             .check_response(&request_id, response)
             .await
             .map_err(|e| DataFusionError::External(Box::new(e)))?;
+        self.freshness
+            .observe(freshness_request, response.headers());
         response.bytes().await.map_err(|e| {
             DataFusionError::External(Box::new(Error::Http {
                 source: Box::new(e),
@@ -419,7 +479,7 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
         let body = reqwest::Body::wrap_stream(chunk_rx);
 
         let part_id = uuid::Uuid::new_v4().to_string();
-        let request = self.build_part_request(&part_id, body);
+        let (request, freshness_request) = self.build_part_request(&part_id, body);
 
         // Measured from just before the request is sent, matching the window the
         // client read timeout applies to the upload.
@@ -495,7 +555,7 @@ impl<S: HttpSend + 'static> PartRequestCtx<'_, S> {
             Ok::<bool, DataFusionError>(input_ended)
         };
 
-        let send = self.send_part_request(request);
+        let send = self.send_part_request(request, freshness_request);
 
         // `join!` rather than `tokio::spawn`: the producer borrows `input` (and
         // `schema`), so it cannot satisfy the `'static` bound a spawned task
@@ -569,7 +629,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
         // Building a fresh exec (with a new, empty `result`) is what makes the
         // outer rescannable retry loop work: `reset_state()` clears the captured
         // result so a re-execution starts clean.
-        Ok(Arc::new(Self::new_inner(
+        let mut exec = Self::new_inner(
             self.table_name.clone(),
             self.identifier.clone(),
             self.client.clone(),
@@ -580,7 +640,9 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
             self.branch.clone(),
             self.max_bytes_per_request,
             self.max_request_duration,
-        )))
+        );
+        exec.freshness = self.freshness.clone();
+        Ok(Arc::new(exec))
     }
 
     fn execute(
@@ -613,6 +675,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                 &self.metrics,
             ));
         let client = self.client.clone();
+        let freshness = self.freshness.clone();
         let identifier = self.identifier.clone();
         let op = self.op.clone();
         let result_slot = self.result.clone();
@@ -634,6 +697,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                 let overwrite = matches!(op, WriteOp::Insert { overwrite: true });
                 let ctx = PartRequestCtx {
                     client: &client,
+                    freshness: &freshness,
                     identifier: &identifier,
                     table_name: &table_name,
                     upload_id,
@@ -688,7 +752,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
 
             let (error_tx, mut error_rx) = tokio::sync::oneshot::channel();
             let body = Self::stream_as_http_body(input_stream, error_tx, tracker)?;
-            let request = request.body(body);
+            let (request, freshness_request) = freshness.prepare(request.body(body));
 
             let result: DataFusionResult<(String, _)> = async {
                 let (request_id, response) = client
@@ -708,6 +772,7 @@ impl<S: HttpSend + 'static> ExecutionPlan for RemoteWriteExec<S> {
                     .check_response(&request_id, response)
                     .await
                     .map_err(|e| DataFusionError::External(Box::new(e)))?;
+                freshness.observe(freshness_request, response.headers());
 
                 Ok((request_id, response))
             }