fix(remote): bound concurrent blob reads across handles (#4311)

python/python/lancedb/remote/__init__.py

@@ -170,6 +170,10 @@ class ClientConfig:
         Configuration for retrying failed requests.
     timeout_config: Optional[TimeoutConfig]
         Configuration for request timeouts.
+    blob_request_concurrency: Optional[int]
+        Maximum concurrent blob HTTP requests across all handles on this
+        connection. Defaults to 8. Set a positive integer here or via
+        `LANCE_CLIENT_BLOB_REQUEST_CONCURRENCY`.
     extra_headers: Optional[dict]
         Additional headers to include in requests.
     id_delimiter: Optional[str]
@@ -193,6 +197,7 @@ class ClientConfig:
     user_agent: str = f"LanceDB-Python-Client/{__version__}"
     retry_config: RetryConfig = field(default_factory=RetryConfig)
     timeout_config: Optional[TimeoutConfig] = field(default_factory=TimeoutConfig)
+    blob_request_concurrency: Optional[int] = None
     extra_headers: Optional[dict] = None
     id_delimiter: Optional[str] = None
     tls_config: Optional[TlsConfig] = None

python/python/lancedb/remote/db.py

@@ -112,6 +112,7 @@ def _client_config_to_dict(config: ClientConfig) -> dict[str, Any]:
         "user_agent": config.user_agent,
         "retry_config": _retry_config_to_dict(config.retry_config),
         "timeout_config": _timeout_config_to_dict(config.timeout_config),
+        "blob_request_concurrency": config.blob_request_concurrency,
         "extra_headers": config.extra_headers,
         "id_delimiter": config.id_delimiter,
         "tls_config": _tls_config_to_dict(config.tls_config),

python/python/tests/test_remote_db.py

@@ -2728,6 +2728,11 @@ def test_create_client():
     assert isinstance(db.client_config, ClientConfig)
     assert db.client_config.retry_config.retries == 42
 
+    db = lancedb.connect(
+        **mandatory_args, client_config={"blob_request_concurrency": 3}
+    )
+    assert db.client_config.blob_request_concurrency == 3
+
     with pytest.warns(DeprecationWarning):
         db = lancedb.connect(**mandatory_args, connection_timeout=42)
         assert db.client_config.timeout_config.connect_timeout == timedelta(seconds=42)

python/src/connection.rs

@@ -1220,6 +1220,7 @@ pub struct PyClientConfig {
     user_agent: String,
     retry_config: Option<PyClientRetryConfig>,
     timeout_config: Option<PyClientTimeoutConfig>,
+    blob_request_concurrency: Option<usize>,
     extra_headers: Option<HashMap<String, String>>,
     id_delimiter: Option<String>,
     tls_config: Option<PyClientTlsConfig>,
@@ -1305,6 +1306,7 @@ impl From<PyClientConfig> for lancedb::remote::ClientConfig {
             user_agent: value.user_agent,
             retry_config: value.retry_config.map(Into::into).unwrap_or_default(),
             timeout_config: value.timeout_config.map(Into::into).unwrap_or_default(),
+            blob_request_concurrency: value.blob_request_concurrency,
             extra_headers: value.extra_headers.unwrap_or_default(),
             id_delimiter: value.id_delimiter,
             tls_config: value.tls_config.map(Into::into),