MinAccumulator over List<T> drops inner-Field metadata for non-null row groups, causing concat / comparison failures #21981

DataType::List(field) => {
    let list_array = array.as_list::<i32>();
    let nested_array = list_array.value(index);
    SingleRowListArrayBuilder::new(nested_array)
        .with_field(field)
        .build_list_scalar()
}

pub fn with_field(self, field: &Field) -> Self {
    self.with_field_name(Some(field.name().to_owned()))
        .with_nullable(field.is_nullable())
}

DataType::List(field_ref) => ScalarValue::List(Arc::new(
    GenericListArray::new_null(Arc::clone(field_ref), 1),
)),

use std::collections::HashMap;
use std::sync::Arc;

use datafusion::arrow::array::{Array, ArrayRef, ListArray, RecordBatch, StringArray};
use datafusion::arrow::buffer::OffsetBuffer;
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::datasource::MemTable;
use datafusion::functions_aggregate::min_max::MinAccumulator;
use datafusion::logical_expr::Accumulator;
use datafusion::prelude::SessionContext;
use datafusion::scalar::ScalarValue;

fn inner_with_metadata() -> Arc<Field> {
    let mut md = HashMap::new();
    md.insert("ARROW:extension:name".to_string(), "arrow.json".to_string());
    md.insert("ARROW:extension:metadata".to_string(), "{}".to_string());
    Arc::new(Field::new("item", DataType::Utf8, true).with_metadata(md))
}

fn dump(label: &str, sv: &ScalarValue) {
    if let ScalarValue::List(arr) = sv {
        if let DataType::List(inner) = arr.data_type() {
            println!("  {label}: inner_field metadata={:?}", inner.metadata());
            return;
        }
    }
    println!("  {label}: {sv:?}");
}

fn unit_repro() {
    println!("== Direct MinAccumulator reproducer ==");
    let inner = inner_with_metadata();
    let list_dt = DataType::List(Arc::clone(&inner));

    // Row group 1: all NULL → metadata preserved
    let mut acc1 = MinAccumulator::try_new(&list_dt).unwrap();
    let null_arr: ArrayRef = Arc::new(ListArray::new_null(Arc::clone(&inner), 2));
    acc1.update_batch(&[null_arr]).unwrap();
    let sv1 = acc1.evaluate().unwrap();
    dump("row group 1 (all-null)", &sv1);

    // Row group 2: has values → metadata stripped
    let mut acc2 = MinAccumulator::try_new(&list_dt).unwrap();
    let values: ArrayRef = Arc::new(StringArray::from(vec!["alpha", "beta"]));
    let offsets = OffsetBuffer::new(vec![0_i32, 1, 2].into());
    let arr_vals: ArrayRef = Arc::new(ListArray::new(Arc::clone(&inner), offsets, values, None));
    acc2.update_batch(&[arr_vals]).unwrap();
    let sv2 = acc2.evaluate().unwrap();
    dump("row group 2 (values)", &sv2);

    println!(
        "  iter_to_array: {:?}",
        ScalarValue::iter_to_array(vec![sv1, sv2]).map(|a| a.data_type().clone())
    );
}

async fn sql_repro() -> datafusion::error::Result<()> {
    println!();
    println!("== SQL MIN() reproducer ==");
    let inner = inner_with_metadata();
    let list_dt = DataType::List(Arc::clone(&inner));
    let schema = Arc::new(Schema::new(vec![Field::new("col", list_dt, true)]));

    // Two partitions: one all-null row, one non-null row.
    let null_arr: ArrayRef = Arc::new(ListArray::new_null(Arc::clone(&inner), 1));
    let batch_a = RecordBatch::try_new(schema.clone(), vec![null_arr])?;

    let values: ArrayRef = Arc::new(StringArray::from(vec!["alpha"]));
    let offsets = OffsetBuffer::new(vec![0_i32, 1].into());
    let list_arr: ArrayRef = Arc::new(ListArray::new(Arc::clone(&inner), offsets, values, None));
    let batch_b = RecordBatch::try_new(schema.clone(), vec![list_arr])?;

    let table = MemTable::try_new(schema, vec![vec![batch_a], vec![batch_b]])?;
    let ctx = SessionContext::new();
    ctx.register_table("t", Arc::new(table))?;

    let df = ctx.sql("SELECT MIN(col) AS m FROM t").await?;
    match df.collect().await {
        Ok(batches) => println!("  result: {batches:?}"),
        Err(e) => println!("  SQL FAILED: {e}"),
    }
    Ok(())
}

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    unit_repro();
    sql_repro().await
}