Implement deterministic pgjson serialization

Add deterministic JSON serialization that maintains consistent key
ordering regardless of map ordering. Introduce ordered key logic in
display.rs, along with a new pretty-writer. Update root pgjson
emission to use the ordered writer and add a regression test to
verify key sequence for Values plan output.

Author: kosiew

datafusion/expr/src/logical_plan/display.rs

@@ -283,6 +283,107 @@ pub struct PgJsonVisitor<'a, 'b> {
     parent_ids: Vec<u32>,
 }
 
+fn ordered_keys(map: &serde_json::Map<String, serde_json::Value>) -> Vec<&str> {
+    let mut ordered_keys = vec![];
+
+    // pgjson node objects are always emitted as:
+    // Node Type, node-specific keys, Plans, Output
+    if map.contains_key("Node Type") {
+        ordered_keys.push("Node Type");
+
+        let mut middle_keys = map
+            .keys()
+            .map(String::as_str)
+            .filter(|k| *k != "Node Type" && *k != "Plans" && *k != "Output")
+            .collect::<Vec<_>>();
+        middle_keys.sort_unstable();
+        ordered_keys.extend(middle_keys);
+
+        if map.contains_key("Plans") {
+            ordered_keys.push("Plans");
+        }
+
+        if map.contains_key("Output") {
+            ordered_keys.push("Output");
+        }
+    } else {
+        let mut keys = map.keys().map(String::as_str).collect::<Vec<_>>();
+        keys.sort_unstable();
+        ordered_keys.extend(keys);
+    }
+
+    ordered_keys
+}
+
+fn write_ordered_json(
+    value: &serde_json::Value,
+    buf: &mut String,
+    indent: usize,
+) -> datafusion_common::Result<()> {
+    match value {
+        serde_json::Value::Null
+        | serde_json::Value::Bool(_)
+        | serde_json::Value::Number(_)
+        | serde_json::Value::String(_) => {
+            let scalar = serde_json::to_string(value)
+                .map_err(|e| DataFusionError::External(Box::new(e)))?;
+            buf.push_str(&scalar);
+        }
+        serde_json::Value::Array(values) => {
+            if values.is_empty() {
+                buf.push_str("[]");
+                return Ok(());
+            }
+
+            buf.push('[');
+            buf.push('\n');
+            for (idx, value) in values.iter().enumerate() {
+                buf.push_str(&" ".repeat((indent + 1) * 2));
+                write_ordered_json(value, buf, indent + 1)?;
+                if idx + 1 != values.len() {
+                    buf.push(',');
+                }
+                buf.push('\n');
+            }
+            buf.push_str(&" ".repeat(indent * 2));
+            buf.push(']');
+        }
+        serde_json::Value::Object(map) => {
+            if map.is_empty() {
+                buf.push_str("{}");
+                return Ok(());
+            }
+
+            buf.push('{');
+            buf.push('\n');
+
+            let keys = ordered_keys(map);
+            for (idx, key) in keys.iter().enumerate() {
+                let value = map
+                    .get(*key)
+                    .ok_or_else(|| internal_datafusion_err!("Missing key in object!"))?;
+
+                buf.push_str(&" ".repeat((indent + 1) * 2));
+                let escaped_key = serde_json::to_string(key)
+                    .map_err(|e| DataFusionError::External(Box::new(e)))?;
+                buf.push_str(&escaped_key);
+                buf.push_str(": ");
+                write_ordered_json(value, buf, indent + 1)?;
+
+                if idx + 1 != keys.len() {
+                    buf.push(',');
+                }
+                buf.push('\n');
+            }
+
+            buf.push_str(&" ".repeat(indent * 2));
+            buf.push('}');
+        }
+    }
+
+    Ok(())
+}
+
 impl<'a, 'b> PgJsonVisitor<'a, 'b> {
     pub fn new(f: &'a mut fmt::Formatter<'b>) -> Self {
         Self {
@@ -692,12 +793,9 @@ impl<'n> TreeNodeVisitor<'n> for PgJsonVisitor<'_, '_> {
         } else {
             // This is the root node
             let plan = serde_json::json!([{"Plan": current_node}]);
-            write!(
-                self.f,
-                "{}",
-                serde_json::to_string_pretty(&plan)
-                    .map_err(|e| DataFusionError::External(Box::new(e)))?
-            )?;
+            let mut plan_str = String::new();
+            write_ordered_json(&plan, &mut plan_str, 0)?;
+            write!(self.f, "{plan_str}")?;
         }
 
         Ok(TreeNodeRecursion::Continue)

datafusion/expr/src/logical_plan/plan.rs

@@ -4593,6 +4593,30 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_display_pg_json_values_field_order() -> Result<()> {
+        let plan = LogicalPlanBuilder::values(vec![vec![lit(1_i64)]])?.build()?;
+
+        let output = plan.display_pg_json().to_string();
+        let node_type_pos = output
+            .find("\"Node Type\": \"Values\"")
+            .expect("expected Node Type key in pgjson output");
+        let values_pos = output
+            .find("\"Values\": \"(Int64(1))\"")
+            .expect("expected Values key in pgjson output");
+        let plans_pos = output
+            .find("\"Plans\": []")
+            .expect("expected Plans key in pgjson output");
+        let output_pos = output
+            .find("\"Output\": [")
+            .expect("expected Output key in pgjson output");
+
+        assert!(node_type_pos < values_pos);
+        assert!(values_pos < plans_pos);
+        assert!(plans_pos < output_pos);
+        Ok(())
+    }
+
     /// Tests for the Visitor trait and walking logical plan nodes
     #[derive(Debug, Default)]
     struct OkVisitor {