alamb opened a new issue, #24469: URL: https://github.com/apache/datafusion/issues/24469
### Describe the bug - Found while reviewing https://github.com/apache/datafusion/pull/24384 **This is not a bug on current `main`** — it only surfaces once https://github.com/apache/datafusion/pull/24384 is merged. On `main` today, `UnnestExec` emits its entire expansion of each input batch as a single batch (#24383), so `output_batches` trivially matches. With #24384 applied, `UnnestExec` bounds its output batches by `datafusion.execution.batch_size`, and when the expansion spans more than one output batch its `output_batches` metric reports fewer batches than the operator actually emits. For example, unnesting a 25-element list at `datafusion.execution.batch_size = 10` emits three batches (`[10, 10, 5]` rows) but reports `output_batches=1`. ### To Reproduce The following standalone test (e.g. dropped into `datafusion/core/tests/`) fails on the #24384 branch: ```rust use std::sync::Arc; use datafusion::common::Result; use datafusion::physical_plan::metrics::MetricValue; use datafusion::physical_plan::unnest::UnnestExec; use datafusion::physical_plan::{ExecutionPlan, collect}; use datafusion::prelude::*; /// The `output_batches` metric of `UnnestExec` should equal the number of /// batches the operator emits to its consumer. #[tokio::test] async fn unnest_output_batches_metric_matches_emitted_batches() -> Result<()> { // A single input row whose list expands to 25 rows, at batch_size 10, so // the unnest must emit its output as multiple batches. let config = SessionConfig::new() .with_batch_size(10) .with_target_partitions(1); let ctx = SessionContext::new_with_config(config); let df = ctx.sql("SELECT unnest(range(0, 25)) AS x").await?; let plan = df.create_physical_plan().await?; let batches = collect(Arc::clone(&plan), ctx.task_ctx()).await?; let emitted_sizes: Vec<usize> = batches.iter().map(|b| b.num_rows()).collect(); let unnest = find_unnest(&plan).expect("plan should contain UnnestExec"); let metrics = unnest.metrics().expect("UnnestExec should have metrics"); let output_batches = metrics .sum(|m| matches!(m.value(), MetricValue::OutputBatches(_))) .expect("output_batches metric should be present") .as_usize(); // The 25 output rows arrive as three batches of at most batch_size rows assert_eq!(emitted_sizes, vec![10, 10, 5]); // ... so the metric should report three output batches assert_eq!( output_batches, emitted_sizes.len(), "output_batches metric disagrees with the number of emitted batches" ); Ok(()) } fn find_unnest(plan: &Arc<dyn ExecutionPlan>) -> Option<Arc<dyn ExecutionPlan>> { if plan.downcast_ref::<UnnestExec>().is_some() { return Some(Arc::clone(plan)); } plan.children().into_iter().find_map(find_unnest) } ``` Output: ``` assertion `left == right` failed: output_batches metric disagrees with the number of emitted batches left: 1 right: 3 ``` ### Expected behavior `output_batches` matches the number of batches the operator emits to its consumer (`3` in the reproducer above). ### Additional context - Related to https://github.com/apache/datafusion/issues/24383 and https://github.com/apache/datafusion/issues/24468, which is the same symptom in the TopK path of `SortExec` (that one does reproduce on `main`). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
