test: add coverage for parallel json reads

Author: andrei-ionescu

default-engine/src/json.rs

@@ -1129,6 +1129,50 @@ mod tests {
         }
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_read_json_files_parallel_empty_files() {
+        let store = Arc::new(InMemory::new());
+        let handler = DefaultJsonHandler::new(
+            store,
+            Arc::new(TokioMultiThreadExecutor::new(
+                tokio::runtime::Handle::current(),
+            )),
+        )
+        .with_parallel_chunks(NonZero::new(4));
+        let physical_schema = schema_ref! { nullable "val": INTEGER };
+        let result: Vec<_> = handler
+            .read_json_files(&[], physical_schema, None)
+            .unwrap()
+            .try_collect()
+            .unwrap();
+        assert!(result.is_empty(), "empty file list must yield no batches");
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_read_json_files_parallel_missing_file_errors() {
+        let store = Arc::new(InMemory::new());
+        let missing_path = Path::from("test/missing");
+        let url = Url::parse(&format!("memory:/{missing_path}")).unwrap();
+        let files = vec![FileMeta {
+            location: url,
+            last_modified: 0,
+            size: 100,
+        }];
+        let handler = DefaultJsonHandler::new(
+            store,
+            Arc::new(TokioMultiThreadExecutor::new(
+                tokio::runtime::Handle::current(),
+            )),
+        )
+        .with_parallel_chunks(NonZero::new(4));
+        let physical_schema = schema_ref! { nullable "val": INTEGER };
+        let result: DeltaResult<Vec<_>> = handler
+            .read_json_files(&files, physical_schema, None)
+            .unwrap()
+            .try_collect();
+        assert!(result.is_err(), "missing file must produce an error");
+    }
+
     // Helper function to create test data
     fn create_test_data(values: Vec<&str>) -> DeltaResult<Box<dyn EngineData>> {
         let schema = Arc::new(ArrowSchema::new(vec![Field::new(

default-engine/src/lib.rs

@@ -537,6 +537,7 @@ mod tests {
             .with_task_executor(executor)
             .with_buffer_size(NonZero::new(4).unwrap())
             .with_batch_size(NonZero::new(8).unwrap())
+            .with_parallel_chunks(NonZero::new(4))
             .build();
         test_arrow_engine(&engine, &url);
     }