Describe the bug
This is a regression introduced by #24384
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:
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
Describe the bug
This is a regression introduced by #24384
UnnestExecbounds its output batches bydatafusion.execution.batch_size, and when the expansion spans more than one output batch itsoutput_batchesmetric reports fewer batches than the operator actually emits.For example, unnesting a 25-element list at
datafusion.execution.batch_size = 10emits three batches ([10, 10, 5]rows) but reportsoutput_batches=1.To Reproduce
The following standalone test (e.g. dropped into
datafusion/core/tests/) fails on the #24384 branch:Output:
Expected behavior
output_batchesmatches the number of batches the operator emits to its consumer (3in the reproducer above).Additional context
output_batchesmetric does not match the number of emitted batches #24468, which is the same symptom in the TopK path ofSortExec(that one does reproduce onmain).