Address martin-g review comments on JSON arrays PR apache#19924

1. Forward stream read errors to result_tx instead of only logging them
2. Switch byte channel from std::sync::mpsc to tokio::sync::mpsc to
   avoid blocking a tokio worker thread when the buffer is full
3. Add validate_complete() after parsing to detect malformed JSON
4. Use workspace = true for serde_json dep; remove unused log dep
5. Use PathBuf::join() instead of hardcoded "/" in test paths
6. Detect leading non-whitespace content before '[' in JSON array parser

https://claude.ai/code/session_01UGW59c5Fu8isLiNhU191uy

Author: claude

Cargo.lock

@@ -59,9 +59,9 @@ mod tests {
     /// Create a temporary JSON file and return (TempDir, path)
     fn create_temp_json(content: &str) -> (tempfile::TempDir, String) {
         let tmp_dir = tempfile::TempDir::new().unwrap();
-        let path = format!("{}/test.json", tmp_dir.path().to_string_lossy());
+        let path = tmp_dir.path().join("test.json");
         std::fs::write(&path, content).unwrap();
-        (tmp_dir, path)
+        (tmp_dir, path.to_string_lossy().to_string())
     }
 
     /// Infer schema from JSON array format file
@@ -395,7 +395,8 @@ mod tests {
     async fn test_write_empty_json_from_sql() -> Result<()> {
         let ctx = SessionContext::new();
         let tmp_dir = tempfile::TempDir::new()?;
-        let path = format!("{}/empty_sql.json", tmp_dir.path().to_string_lossy());
+        let path = tmp_dir.path().join("empty_sql.json");
+        let path = path.to_string_lossy().to_string();
         let df = ctx.sql("SELECT CAST(1 AS BIGINT) AS id LIMIT 0").await?;
         df.write_json(&path, crate::dataframe::DataFrameWriteOptions::new(), None)
             .await?;
@@ -421,7 +422,8 @@ mod tests {
         )?;
 
         let tmp_dir = tempfile::TempDir::new()?;
-        let path = format!("{}/empty_batch.json", tmp_dir.path().to_string_lossy());
+        let path = tmp_dir.path().join("empty_batch.json");
+        let path = path.to_string_lossy().to_string();
         let df = ctx.read_batch(empty_batch.clone())?;
         df.write_json(&path, crate::dataframe::DataFrameWriteOptions::new(), None)
             .await?;
@@ -586,7 +588,8 @@ mod tests {
         use std::io::Write;
 
         let tmp_dir = tempfile::TempDir::new()?;
-        let path = format!("{}/array.json.gz", tmp_dir.path().to_string_lossy());
+        let path = tmp_dir.path().join("array.json.gz");
+        let path = path.to_string_lossy().to_string();
 
         let file = std::fs::File::create(&path)?;
         let mut encoder = GzEncoder::new(file, Compression::default());

datafusion/core/src/datasource/file_format/json.rs

@@ -43,9 +43,8 @@ datafusion-physical-expr-common = { workspace = true }
 datafusion-physical-plan = { workspace = true }
 datafusion-session = { workspace = true }
 futures = { workspace = true }
-log = "0.4.29"
 object_store = { workspace = true }
-serde_json = "1.0.149"
+serde_json = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true, features = ["sync"] }
 

datafusion/datasource-json/Cargo.toml

@@ -358,14 +358,18 @@ impl FileOpener for JsonOpener {
                         let decompressed_stream =
                             file_compression_type.convert_stream(s.boxed())?;
 
-                        // Channel for bytes: async producer -> sync consumer
+                        // Channel for bytes: async producer -> blocking consumer
+                        // Uses tokio::sync::mpsc so the async send never blocks a
+                        // tokio worker thread; the consumer calls blocking_recv()
+                        // inside spawn_blocking.
                         let (byte_tx, byte_rx) =
-                            std::sync::mpsc::sync_channel::<bytes::Bytes>(
+                            tokio::sync::mpsc::channel::<bytes::Bytes>(
                                 CHANNEL_BUFFER_SIZE,
                             );
 
                         // Channel for results: sync producer -> async consumer
                         let (result_tx, result_rx) = tokio::sync::mpsc::channel(2);
+                        let error_tx = result_tx.clone();
 
                         // Async task: read from object store stream and send bytes to channel
                         // Store the SpawnedTask to keep it alive until stream is dropped
@@ -374,12 +378,16 @@ impl FileOpener for JsonOpener {
                             while let Some(chunk) = decompressed_stream.next().await {
                                 match chunk {
                                     Ok(bytes) => {
-                                        if byte_tx.send(bytes).is_err() {
+                                        if byte_tx.send(bytes).await.is_err() {
                                             break; // Consumer dropped
                                         }
                                     }
                                     Err(e) => {
-                                        log::error!("Error reading JSON stream: {e}");
+                                        let _ = error_tx
+                                            .send(Err(arrow::error::ArrowError::ExternalError(
+                                                Box::new(e),
+                                            )))
+                                            .await;
                                         break;
                                     }
                                 }
@@ -391,14 +399,14 @@ impl FileOpener for JsonOpener {
                         // Store the SpawnedTask to keep it alive until stream is dropped
                         let parse_task = SpawnedTask::spawn_blocking(move || {
                             let channel_reader = ChannelReader::new(byte_rx);
-                            let ndjson_reader = JsonArrayToNdjsonReader::with_capacity(
+                            let mut ndjson_reader = JsonArrayToNdjsonReader::with_capacity(
                                 channel_reader,
                                 JSON_CONVERTER_BUFFER_SIZE,
                             );
 
                             match ReaderBuilder::new(schema)
                                 .with_batch_size(batch_size)
-                                .build(ndjson_reader)
+                                .build(&mut ndjson_reader)
                             {
                                 Ok(arrow_reader) => {
                                     for batch_result in arrow_reader {
@@ -412,6 +420,13 @@ impl FileOpener for JsonOpener {
                                     let _ = result_tx.blocking_send(Err(e));
                                 }
                             }
+
+                            // Validate the JSON array was properly formed
+                            if let Err(e) = ndjson_reader.validate_complete() {
+                                let _ = result_tx.blocking_send(Err(
+                                    arrow::error::ArrowError::JsonError(e.to_string()),
+                                ));
+                            }
                             // result_tx dropped here, closes the stream
                         });
 

datafusion/datasource-json/src/source.rs

@@ -18,7 +18,6 @@
 //! Utility types for JSON processing
 
 use std::io::{BufRead, Read};
-use std::sync::mpsc::Receiver;
 
 use bytes::Bytes;
 
@@ -130,6 +129,8 @@ pub struct JsonArrayToNdjsonReader<R: Read> {
     output_filled: usize,
     /// Whether trailing non-whitespace content was detected after ']'
     has_trailing_content: bool,
+    /// Whether leading non-whitespace content was detected before '['
+    has_leading_content: bool,
 }
 
 impl<R: Read> JsonArrayToNdjsonReader<R> {
@@ -156,6 +157,7 @@ impl<R: Read> JsonArrayToNdjsonReader<R> {
             output_pos: 0,
             output_filled: 0,
             has_trailing_content: false,
+            has_leading_content: false,
         }
     }
 
@@ -169,6 +171,12 @@ impl<R: Read> JsonArrayToNdjsonReader<R> {
     /// - Missing closing `]`
     /// - Unexpected trailing content after `]`
     pub fn validate_complete(&self) -> std::io::Result<()> {
+        if self.has_leading_content {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "Malformed JSON: unexpected leading content before '['",
+            ));
+        }
         if self.depth != 0 {
             return Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidData,
@@ -204,8 +212,9 @@ impl<R: Read> JsonArrayToNdjsonReader<R> {
                 // Looking for the opening '[', skip whitespace
                 if byte == b'[' {
                     self.state = JsonArrayState::InArray;
+                } else if !byte.is_ascii_whitespace() {
+                    self.has_leading_content = true;
                 }
-                // Skip whitespace and the '[' itself
                 None
             }
             JsonArrayState::InArray => {
@@ -376,7 +385,7 @@ impl<R: Read> BufRead for JsonArrayToNdjsonReader<R> {
 // │                     byte_tx.send(chunk)                                 │
 // └─────────────────────────────────────────────────────────────────────────┘
 //                                 │
-//                                 ▼ std::sync::mpsc::sync_channel<Bytes>
+//                                 ▼ tokio::sync::mpsc::channel<Bytes>
 //                                 │   (bounded, ~32MB buffer)
 //                                 ▼
 // ┌─────────────────────────────────────────────────────────────────────────┐
@@ -401,19 +410,21 @@ impl<R: Read> BufRead for JsonArrayToNdjsonReader<R> {
 // - Miscellaneous: ~4MB
 //
 
-/// A synchronous `Read` implementation that receives bytes from a channel.
+/// A synchronous `Read` implementation that receives bytes from an async channel.
 ///
 /// This enables true streaming between async and sync contexts without
-/// loading the entire file into memory.
+/// loading the entire file into memory. Uses `tokio::sync::mpsc::Receiver`
+/// with `blocking_recv()` so the async producer never blocks a tokio worker
+/// thread, while the sync consumer (running in `spawn_blocking`) safely blocks.
 pub struct ChannelReader {
-    rx: Receiver<Bytes>,
+    rx: tokio::sync::mpsc::Receiver<Bytes>,
     current: Option<Bytes>,
     pos: usize,
 }
 
 impl ChannelReader {
-    /// Create a new ChannelReader from a receiver.
-    pub fn new(rx: Receiver<Bytes>) -> Self {
+    /// Create a new ChannelReader from a tokio mpsc receiver.
+    pub fn new(rx: tokio::sync::mpsc::Receiver<Bytes>) -> Self {
         Self {
             rx,
             current: None,
@@ -437,13 +448,13 @@ impl Read for ChannelReader {
             }
 
             // Current chunk exhausted, get next from channel
-            match self.rx.recv() {
-                Ok(bytes) => {
+            match self.rx.blocking_recv() {
+                Some(bytes) => {
                     self.current = Some(bytes);
                     self.pos = 0;
                     // Loop back to read from new chunk
                 }
-                Err(_) => return Ok(0), // Channel closed = EOF
+                None => return Ok(0), // Channel closed = EOF
             }
         }
     }
@@ -596,6 +607,39 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_json_array_with_leading_junk() {
+        let input = r#"junk[{"a":1}, {"a":2}]"#;
+        let mut reader = JsonArrayToNdjsonReader::new(input.as_bytes());
+        let mut output = String::new();
+        reader.read_to_string(&mut output).unwrap();
+
+        // Should still extract the valid array content
+        assert_eq!(output, "{\"a\":1}\n{\"a\":2}");
+
+        // But validation should catch the leading junk
+        let result = reader.validate_complete();
+        assert!(result.is_err());
+        let err_msg = result.unwrap_err().to_string();
+        assert!(
+            err_msg.contains("leading content"),
+            "Expected leading content error, got: {err_msg}"
+        );
+    }
+
+    #[test]
+    fn test_json_array_with_leading_whitespace_ok() {
+        let input = r#"
+  [{"a":1}, {"a":2}]"#;
+        let mut reader = JsonArrayToNdjsonReader::new(input.as_bytes());
+        let mut output = String::new();
+        reader.read_to_string(&mut output).unwrap();
+        assert_eq!(output, "{\"a\":1}\n{\"a\":2}");
+
+        // Leading whitespace should be fine
+        reader.validate_complete().unwrap();
+    }
+
     #[test]
     fn test_validate_complete_valid_with_trailing_whitespace() {
         let input = r#"[{"a":1},{"a":2}]
@@ -690,11 +734,11 @@ mod tests {
     /// Test ChannelReader
     #[test]
     fn test_channel_reader() {
-        let (tx, rx) = std::sync::mpsc::sync_channel(4);
+        let (tx, rx) = tokio::sync::mpsc::channel(4);
 
-        // Send some chunks
-        tx.send(Bytes::from("Hello, ")).unwrap();
-        tx.send(Bytes::from("World!")).unwrap();
+        // Send some chunks (try_send is non-async)
+        tx.try_send(Bytes::from("Hello, ")).unwrap();
+        tx.try_send(Bytes::from("World!")).unwrap();
         drop(tx); // Close channel
 
         let mut reader = ChannelReader::new(rx);
@@ -707,9 +751,9 @@ mod tests {
     /// Test ChannelReader with small reads
     #[test]
     fn test_channel_reader_small_reads() {
-        let (tx, rx) = std::sync::mpsc::sync_channel(4);
+        let (tx, rx) = tokio::sync::mpsc::channel(4);
 
-        tx.send(Bytes::from("ABCDEFGHIJ")).unwrap();
+        tx.try_send(Bytes::from("ABCDEFGHIJ")).unwrap();
         drop(tx);
 
         let mut reader = ChannelReader::new(rx);