Improve UNNEST streaming with chunked batch emission and disable chunking for recursive/staged paths

datafusion/common/src/lib.rs

@@ -94,7 +94,10 @@ pub use schema_reference::SchemaReference;
 pub use spans::{Location, Span, Spans};
 pub use stats::{ColumnStatistics, Statistics};
 pub use table_reference::{ResolvedTableReference, TableReference};
-pub use unnest::{RecursionUnnestOption, UnnestOptions};
+pub use unnest::{
+    RecursionUnnestOption, UNNEST_PLACEHOLDER_METADATA_KEY, UnnestOptions,
+    is_unnest_placeholder_field, unnest_placeholder_field_metadata,
+};
 pub use utils::project_schema;
 
 // These are hidden from docs purely to avoid polluting the public view of what this crate exports.

datafusion/common/src/unnest.rs

@@ -17,7 +17,33 @@
 
 //! [`UnnestOptions`] for unnesting structured types
 
+use std::collections::BTreeMap;
+
+use arrow::datatypes::Field;
+
 use crate::Column;
+use crate::metadata::FieldMetadata;
+
+/// Field metadata key used to mark planner-internal unnest placeholder columns.
+pub const UNNEST_PLACEHOLDER_METADATA_KEY: &str = "datafusion:unnest_placeholder";
+
+const UNNEST_PLACEHOLDER_METADATA_VALUE: &str = "true";
+
+/// Returns metadata that marks a field as a planner-internal unnest placeholder.
+pub fn unnest_placeholder_field_metadata() -> FieldMetadata {
+    FieldMetadata::from(BTreeMap::from([(
+        UNNEST_PLACEHOLDER_METADATA_KEY.to_string(),
+        UNNEST_PLACEHOLDER_METADATA_VALUE.to_string(),
+    )]))
+}
+
+/// Returns true if the field is a planner-internal unnest placeholder.
+pub fn is_unnest_placeholder_field(field: &Field) -> bool {
+    field
+        .metadata()
+        .get(UNNEST_PLACEHOLDER_METADATA_KEY)
+        .is_some_and(|value| value == UNNEST_PLACEHOLDER_METADATA_VALUE)
+}
 
 /// Options for unnesting a column that contains a list type,
 /// replicating values in the other, non nested rows.

datafusion/core/tests/dataframe/mod.rs

@@ -18,6 +18,7 @@
 // Include tests in dataframe_functions
 mod dataframe_functions;
 mod describe;
+mod unnest_chunks;
 
 use arrow::array::{
     Array, ArrayRef, BooleanArray, DictionaryArray, FixedSizeListArray,