Resolve Config optimization conflicts for improved read/write concurrency

- Released Config read guard before converting values to Python objects in get and get_all
- Ensures locks are held only while collecting scalar entries, not during expensive Python object conversion
- Added regression test that runs Config.get_all and Config.set concurrently to guard against read/write contention regressions
- Improves overall performance by reducing lock contention in multi-threaded scenarios

Author: kosiew

python/tests/test_concurrency.py

@@ -65,6 +65,30 @@ def worker(index: int) -> None:
     _run_in_threads(worker, count=12)
 
 
+def test_config_set_during_get_all() -> None:
+    """Ensure config writes proceed while another thread reads all entries."""
+
+    config = Config()
+    key = "datafusion.execution.batch_size"
+
+    def reader() -> None:
+        for _ in range(200):
+            # get_all should not hold the lock while converting to Python objects
+            config.get_all()
+
+    def writer() -> None:
+        for index in range(200):
+            config.set(key, str(1024 + index))
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        reader_future = executor.submit(reader)
+        writer_future = executor.submit(writer)
+        reader_future.result(timeout=10)
+        writer_future.result(timeout=10)
+
+    assert config.get(key) is not None
+
+
 def test_case_builder_reuse_from_multiple_threads() -> None:
     """Ensure the case builder can be safely reused across threads."""
 

src/config.rs

@@ -50,17 +50,20 @@ impl PyConfig {
 
     /// Get a configuration option
     pub fn get<'py>(&self, key: &str, py: Python<'py>) -> PyResult<Bound<'py, PyAny>> {
-        let entries = {
+        let value = {
             let options = self.config.read();
-            options.entries()
+            options
+                .entries()
+                .iter()
+                .find(|entry| entry.key == key)
+                .map(|entry| entry.value.clone())
         };
 
-        for entry in entries {
-            if entry.key == key {
-                return Ok(entry.value.into_pyobject(py)?);
-            }
+        if let Some(value) = value {
+            Ok(value.into_pyobject(py)?)
+        } else {
+            Ok(None::<String>.into_pyobject(py)?)
         }
-        Ok(None::<String>.into_pyobject(py)?)
     }
 
     /// Set a configuration option
@@ -75,12 +78,16 @@ impl PyConfig {
     pub fn get_all(&self, py: Python) -> PyResult<PyObject> {
         let entries = {
             let options = self.config.read();
-            options.entries()
+            options
+                .entries()
+                .into_iter()
+                .map(|entry| (entry.key.to_string(), entry.value.clone()))
+                .collect::<Vec<_>>()
         };
 
         let dict = PyDict::new(py);
-        for entry in entries {
-            dict.set_item(entry.key, entry.value.into_pyobject(py)?)?;
+        for (key, value) in entries {
+            dict.set_item(key, value.into_pyobject(py)?)?;
         }
         Ok(dict.into())
     }