DataFrame::fill_null and fill_nan fail on column names with uppercase letters or dots

[timsaucer]

Describe the bug

DataFrame::fill_null and DataFrame::fill_nan fail with a schema error on any DataFrame that has a column whose name is not a plain lowercase identifier, such as Name or a.b. This happens even when that column is not in the columns list being filled.

fill_columns rebuilds every column in the projection with col(field.name()). col parses its argument as a SQL identifier, so Name is normalized to name, and a.b becomes column b qualified by table a. Neither resolves against the schema.

https://github.com/apache/datafusion/blob/main/datafusion/core/src/dataframe/mod.rs (fill_columns, the three col(field.name()) calls in the projection)

To Reproduce

use std::sync::Arc;

use datafusion::arrow::array::{Float64Array, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::arrow::record_batch::RecordBatch;
use datafusion::prelude::*;
use datafusion::scalar::ScalarValue;

#[tokio::main]
async fn main() -> datafusion::error::Result<()> {
    let ctx = SessionContext::new();
    let schema = Arc::new(Schema::new(vec![
        Field::new("Name", DataType::Utf8, true),
        Field::new("v", DataType::Float64, true),
    ]));
    let batch = RecordBatch::try_new(
        schema,
        vec![
            Arc::new(StringArray::from(vec![Some("a"), None])),
            Arc::new(Float64Array::from(vec![Some(f64::NAN), None])),
        ],
    )?;
    let df = ctx.read_batch(batch)?;
    let zero = ScalarValue::Float64(Some(0.0));

    // Only "v" is being filled, but both calls fail on "Name".
    df.clone().fill_null(&zero, &["v"])?;
    df.clone().fill_nan(&zero, &["v"])?;
    Ok(())
}

On 55.1.0:

Error: SchemaError(FieldNotFound { field: Column { relation: None, name: "name" }, valid_fields: [Column { relation: Some(Bare { table: "?table?" }), name: "Name" }, Column { relation: Some(Bare { table: "?table?" }), name: "v" }] }, Some(""))

With a column named a.b instead of Name, the error is FieldNotFound { field: Column { relation: Some(Bare { table: "a" }), name: "b" }, ... }.

Expected behavior

Both calls succeed, fill v, and pass Name (or a.b) through unchanged.

Additional context

columns entries are matched by exact name (field_with_name), so there is no way to quote around this from the caller's side: "Name" is reported as not found, and even columns = ["v"] fails because of the unrelated Name column.

Building each column reference from the schema entry, rather than from its name, fixes both cases. It also keeps the qualifier, so it would handle a DataFrame with the same column name under two qualifiers, for example after a join:

self.logical_plan()
    .schema()
    .iter()
    .map(|(qualifier, field)| {
        let column = Expr::Column(Column::from((qualifier, field)));
        // ... use `column` wherever `col(field.name())` is used today
    })

I checked this projection against both the Name and a.b schemas above, and both run. ident(field.name()) would also fix the parsing, but it drops the qualifier.

Found through datafusion-python, where DataFrame.fill_null and the new DataFrame.fill_nan wrap these methods: apache/datafusion-python#1763 (comment)