Properly call from_pandas in pyarrow (vega#560)

* Properly call from_pandas in pyarrow

* bump rustc

* Fix warnings

---------

Co-authored-by: Jon Mease <[email protected]>

Author: johnwmillr

pixi.lock

@@ -131,7 +131,7 @@ jupytext = "1.15.0.*"
 openjdk = "20.0.0.*"
 minio-server = "2023.9.23.3.47.50.*"
 minio = "7.1.17.*"
-rust = "1.80.*"
+rust = "1.84.*"
 taplo = ">=0.9.3,<0.10"
 ruff = ">=0.6.9,<0.7"
 mypy = ">=1.11.2,<2"

pixi.toml

@@ -5,6 +5,9 @@ edition = "2021"
 version = "2.0.1"
 description = "Core components required by multiple VegaFusion crates, with WASM compatibility"
 
+[lints.clippy]
+module_inception = "allow"
+
 [features]
 tonic_support = ["tonic", "tonic-build"]
 py = ["pyo3", "vegafusion-common/py"]

vegafusion-core/Cargo.toml

@@ -133,46 +133,53 @@ impl ChartState {
         runtime: &dyn VegaFusionRuntimeTrait,
         updates: Vec<ExportUpdateJSON>,
     ) -> Result<Vec<ExportUpdateJSON>> {
-        let mut task_graph = self.task_graph.lock().map_err(|err| {
-            VegaFusionError::internal(format!("Failed to acquire task graph lock: {:?}", err))
-        })?;
-        let server_to_client = self.server_to_client_value_indices.clone();
-        let mut indices: Vec<NodeValueIndex> = Vec::new();
-        for export_update in &updates {
-            let var = match export_update.namespace {
-                ExportUpdateNamespace::Signal => Variable::new_signal(&export_update.name),
-                ExportUpdateNamespace::Data => Variable::new_data(&export_update.name),
-            };
-            let scoped_var: ScopedVariable = (var, export_update.scope.clone());
-            let node_value_index = *self
-                .task_graph_mapping
-                .get(&scoped_var)
-                .with_context(|| format!("No task graph node found for {scoped_var:?}"))?;
-
-            let value = match export_update.namespace {
-                ExportUpdateNamespace::Signal => {
-                    TaskValue::Scalar(ScalarValue::from_json(&export_update.value)?)
-                }
-                ExportUpdateNamespace::Data => {
-                    TaskValue::Table(VegaFusionTable::from_json(&export_update.value)?)
-                }
-            };
-
-            indices.extend(task_graph.update_value(node_value_index.node_index as usize, value)?);
-        }
-
-        // Filter to update nodes in the comm plan
-        let indices: Vec<_> = indices
-            .iter()
-            .filter(|&node| server_to_client.contains(node))
-            .cloned()
-            .collect();
+        // Scope the mutex guard to ensure it's dropped before the async call
+        let (indices, cloned_task_graph) = {
+            let mut task_graph = self.task_graph.lock().map_err(|err| {
+                VegaFusionError::internal(format!("Failed to acquire task graph lock: {:?}", err))
+            })?;
+            let server_to_client = self.server_to_client_value_indices.clone();
+            let mut indices: Vec<NodeValueIndex> = Vec::new();
+
+            for export_update in &updates {
+                let var = match export_update.namespace {
+                    ExportUpdateNamespace::Signal => Variable::new_signal(&export_update.name),
+                    ExportUpdateNamespace::Data => Variable::new_data(&export_update.name),
+                };
+                let scoped_var: ScopedVariable = (var, export_update.scope.clone());
+                let node_value_index = *self
+                    .task_graph_mapping
+                    .get(&scoped_var)
+                    .with_context(|| format!("No task graph node found for {scoped_var:?}"))?;
+
+                let value = match export_update.namespace {
+                    ExportUpdateNamespace::Signal => {
+                        TaskValue::Scalar(ScalarValue::from_json(&export_update.value)?)
+                    }
+                    ExportUpdateNamespace::Data => {
+                        TaskValue::Table(VegaFusionTable::from_json(&export_update.value)?)
+                    }
+                };
+
+                indices
+                    .extend(task_graph.update_value(node_value_index.node_index as usize, value)?);
+            }
+
+            // Filter to update nodes in the comm plan
+            let indices: Vec<_> = indices
+                .iter()
+                .filter(|&node| server_to_client.contains(node))
+                .cloned()
+                .collect();
 
-        let cloned_task_graph = task_graph.clone();
+            // Clone the task graph while we still have the lock
+            let cloned_task_graph = task_graph.clone();
 
-        // Drop the MutexGuard before await call to avoid warning
-        drop(task_graph);
+            // Return both values we need
+            (indices, cloned_task_graph)
+        }; // MutexGuard is dropped here
 
+        // Now we can safely make the async call
         let response_task_values = runtime
             .query_request(
                 Arc::new(cloned_task_graph),
@@ -185,7 +192,6 @@ impl ChartState {
             .into_iter()
             .map(|response_value| {
                 let variable = response_value.variable;
-
                 let scope = response_value.scope;
                 let value = response_value.value;
 

vegafusion-core/src/chart_state.rs

@@ -92,7 +92,7 @@ struct Tokenizer<'a> {
     full_text: &'a str,
 }
 
-impl<'a> Tokenizer<'a> {
+impl Tokenizer<'_> {
     fn new(src: &str) -> Tokenizer {
         Tokenizer {
             current_index: 0,

vegafusion-core/src/expression/lexer.rs

@@ -312,7 +312,7 @@ impl<'a> DatasetsColumnUsageVisitor<'a> {
     }
 }
 
-impl<'a> ExpressionVisitor for DatasetsColumnUsageVisitor<'a> {
+impl ExpressionVisitor for DatasetsColumnUsageVisitor<'_> {
     fn visit_member(&mut self, node: &MemberExpression) {
         if let (Some(datum_var), Some(object), Some(property)) =
             (&self.datum_var, &node.object, &node.property)