Redesign json array streaming for datafusion (#31)

Author: zhuqi-lucas

Cargo.lock

@@ -181,6 +181,7 @@ tempfile = "3"
 testcontainers = { version = "0.25.2", features = ["default"] }
 testcontainers-modules = { version = "0.13" }
 tokio = { version = "1.48", features = ["macros", "rt", "sync"] }
+tokio-stream = "0.1"
 url = "2.5.7"
 
 [workspace.lints.clippy]

Cargo.toml

@@ -121,7 +121,7 @@ async fn json_opener() -> Result<()> {
         projected,
         FileCompressionType::UNCOMPRESSED,
         Arc::new(object_store),
-        false,
+        true,
     );
 
     let scan_config = FileScanConfigBuilder::new(

datafusion-examples/examples/csv_json_opener.rs

@@ -2813,23 +2813,22 @@ config_namespace! {
     pub struct JsonOptions {
         pub compression: CompressionTypeVariant, default = CompressionTypeVariant::UNCOMPRESSED
         pub schema_infer_max_rec: Option<usize>, default = None
-        pub compression_level: Option<usize>, default = None
-        /// The format of JSON input files.
-        ///
-        /// When `false` (default), expects newline-delimited JSON (NDJSON):
-        /// ```text
-        /// {"key1": 1, "key2": "val"}
-        /// {"key1": 2, "key2": "vals"}
-        /// ```
-        ///
-        /// When `true`, expects JSON array format:
-        /// ```text
-        /// [
-        ///     {"key1": 1, "key2": "val"},
-        ///     {"key1": 2, "key2": "vals"}
-        /// ]
-        /// ```
-        pub format_array: bool, default = false
+       /// The JSON format to use when reading files.
+       ///
+       /// When `true` (default), expects newline-delimited JSON (NDJSON):
+       /// ```text
+       /// {"key1": 1, "key2": "val"}
+       /// {"key1": 2, "key2": "vals"}
+       /// ```
+       ///
+       /// When `false`, expects JSON array format:
+       /// ```text
+       /// [
+       ///   {"key1": 1, "key2": "val"},
+       ///   {"key1": 2, "key2": "vals"}
+       /// ]
+       /// ```
+       pub newline_delimited: bool, default = true
     }
 }
 

datafusion/common/src/config.rs

@@ -479,7 +479,7 @@ impl DataFrame {
     /// # #[tokio::main]
     /// # async fn main() -> Result<()> {
     /// let ctx = SessionContext::new();
-    /// let df = ctx.read_json("tests/data/unnest.json", NdJsonReadOptions::default()).await?;
+    /// let df = ctx.read_json("tests/data/unnest.json", JsonReadOptions::default()).await?;
     /// // expand into multiple columns if it's json array, flatten field name if it's nested structure
     /// let df = df.unnest_columns(&["b","c","d"])?;
     /// let expected = vec![