//! Expression evaluation on GPU.
//!
//! Walks the `AExpr` arena and evaluates each node, producing a GPU column.

use cudf::Column as GpuColumn;
use cudf::reduction::ReduceOp;
use cudf::types::{DataType as GpuDataType, TypeId as GpuTypeId};
use cudf::unary::UnaryOp;
use polars_error::{PolarsResult, polars_bail, polars_err};
use polars_plan::dsl::function_expr::BooleanFunction;
use polars_plan::dsl::function_expr::FunctionExpr;
use polars_plan::plans::LiteralValue;
use polars_plan::plans::{AExpr, IRAggExpr};
use polars_utils::arena::{Arena, Node};

use crate::error::gpu_result;
use crate::gpu_frame::GpuDataFrame;
use crate::types::{is_comparison, map_dtype, map_operator};

/// Compute the output type for arithmetic operations (supertype promotion).
fn arithmetic_output_type(left_type: &GpuDataType, right_type: &GpuDataType) -> GpuDataType {
    let l = left_type.id();
    let r = right_type.id();

    if l == GpuTypeId::Float64 || r == GpuTypeId::Float64 {
        GpuDataType::new(GpuTypeId::Float64)
    } else if l == GpuTypeId::Float32 || r == GpuTypeId::Float32 {
        GpuDataType::new(GpuTypeId::Float32)
    } else if l == GpuTypeId::Int64 || r == GpuTypeId::Int64 {
        GpuDataType::new(GpuTypeId::Int64)
    } else {
        *left_type
    }
}

/// Evaluate an expression node, producing a GPU column.
pub fn eval_expr(
    node: Node,
    expr_arena: &Arena<AExpr>,
    df: &GpuDataFrame,
) -> PolarsResult<GpuColumn> {
    match expr_arena.get(node) {
        AExpr::Column(name) => df.column_by_name(name.as_str()),

        AExpr::Literal(lit) => literal_to_gpu_column(lit, df.height()),

        AExpr::BinaryExpr { left, op, right } => {
            let left_node = *left;
            let right_node = *right;
            let op = *op;
            let gpu_op = map_operator(op)?;

            // Scalar optimization: avoid broadcasting a literal to a full column.
            // Use binary_op_scalar (or binaryop::binary_op for scalar-column order)
            // to let libcudf handle the broadcast internally on GPU.
            // Falls back to column-column path for unsupported scalar types (Null, String, etc.).
            let left_is_literal = matches!(expr_arena.get(left_node), AExpr::Literal(_));
            let right_is_literal = matches!(expr_arena.get(right_node), AExpr::Literal(_));

            let try_scalar = || -> PolarsResult<GpuColumn> {
                if left_is_literal && !right_is_literal {
                    let scalar = literal_to_scalar(expr_arena.get(left_node))?;
                    let right_col = eval_expr(right_node, expr_arena, df)?;
                    let output_type = if is_comparison(op) {
                        GpuDataType::new(GpuTypeId::Bool8)
                    } else {
                        arithmetic_output_type(&scalar.data_type(), &right_col.data_type())
                    };
                    return gpu_result(cudf::binaryop::binary_op(
                        &scalar,
                        &right_col,
                        gpu_op,
                        output_type,
                    ));
                }
                if right_is_literal && !left_is_literal {
                    let left_col = eval_expr(left_node, expr_arena, df)?;
                    let scalar = literal_to_scalar(expr_arena.get(right_node))?;
                    let output_type = if is_comparison(op) {
                        GpuDataType::new(GpuTypeId::Bool8)
                    } else {
                        arithmetic_output_type(&left_col.data_type(), &scalar.data_type())
                    };
                    return gpu_result(left_col.binary_op_scalar(&scalar, gpu_op, output_type));
                }
                polars_bail!(ComputeError: "not a scalar case")
            };

            if let Ok(result) = try_scalar() {
                return Ok(result);
            }

            // Both are columns (or unsupported scalar type) — fall back to column-column op
            let left_col = eval_expr(left_node, expr_arena, df)?;
            let right_col = eval_expr(right_node, expr_arena, df)?;
            let output_type = if is_comparison(op) {
                GpuDataType::new(GpuTypeId::Bool8)
            } else {
                arithmetic_output_type(&left_col.data_type(), &right_col.data_type())
            };
            gpu_result(left_col.binary_op(&right_col, gpu_op, output_type))
        }

        AExpr::Cast { expr, dtype, .. } => {
            let expr_node = *expr;
            let dtype = dtype.clone();
            let col = eval_expr(expr_node, expr_arena, df)?;
            let gpu_dtype = map_dtype(&dtype)?;
            gpu_result(col.cast(gpu_dtype))
        }

        AExpr::Len => {
            let height = df.height();
            let values: Vec<u32> = vec![height as u32; height];
            gpu_result(GpuColumn::from_slice(&values))
        }

        AExpr::Alias(inner, _name) => eval_expr(*inner, expr_arena, df),

        AExpr::Agg(agg_expr) => eval_agg_expr(agg_expr, expr_arena, df),

        AExpr::Ternary {
            predicate,
            truthy,
            falsy,
        } => {
            let pred_node = *predicate;
            let true_node = *truthy;
            let false_node = *falsy;
            let cond = eval_expr(pred_node, expr_arena, df)?;
            let t = eval_expr(true_node, expr_arena, df)?;
            let f = eval_expr(false_node, expr_arena, df)?;
            // copy_if_else: where mask is true use self (truthy), else use other (falsy)
            gpu_result(t.copy_if_else(&f, &cond))
        }

        AExpr::Function {
            input, function, ..
        } => {
            let input = input.clone();
            let function = function.clone();
            eval_function(&input, &function, expr_arena, df)
        }

        other => polars_bail!(ComputeError: "GPU engine: unsupported expression: {:?}", other),
    }
}

/// Convert a Polars `LiteralValue` to a GPU column of the given height.
fn literal_to_gpu_column(lit: &LiteralValue, height: usize) -> PolarsResult<GpuColumn> {
    match lit {
        // BUG(type-erasure): `Null` literal always materialises as nullable i32.
        // In a context like `col("f64_col") + lit(NULL)` this will produce an
        // i32 column where an f64 null column is expected, causing a downstream
        // type-mismatch error from cuDF's binary-op kernel.
        //
        // Proper fix: propagate the expected `DataType` from the parent
        // `BinaryExpr` / `Ternary` / `Cast` node into this function so we can
        // create the null column with the correct physical type.  Until then,
        // callers that hit a type-mismatch on null columns should cast
        // explicitly.
        //
        // TODO(HIGH): accept a `dtype: &DataType` parameter and construct the
        // appropriately-typed null column.
        LiteralValue::Null => {
            let opts: Vec<Option<i32>> = vec![None; height];
            gpu_result(GpuColumn::from_optional_i32(&opts))
        }
        LiteralValue::Boolean(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::Int32(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::Int64(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::UInt32(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::UInt64(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::Float32(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::Float64(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::String(s) => {
            let strings: Vec<&str> = vec![s.as_str(); height];
            gpu_result(GpuColumn::from_strings(&strings))
        }
        LiteralValue::Int(v) => {
            let val = i64::try_from(*v).map_err(
                |_| polars_err!(ComputeError: "integer literal {} exceeds i64 range", v),
            )?;
            let data = vec![val; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        LiteralValue::Float(v) => {
            let data = vec![*v; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        other => polars_bail!(ComputeError: "GPU engine: unsupported literal type: {:?}", other),
    }
}

/// Convert a Polars `LiteralValue` to a cudf `Scalar` (no GPU column allocation).
///
/// Used by the BinaryExpr scalar optimization path to avoid broadcasting
/// a literal value into a full GPU column.
fn literal_to_scalar(expr: &AExpr) -> PolarsResult<cudf::Scalar> {
    match expr {
        AExpr::Literal(lit) => match lit {
            LiteralValue::Boolean(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Int32(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Int64(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::UInt32(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::UInt64(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Float32(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Float64(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Int(v) => {
                let val = i64::try_from(*v).map_err(
                    |_| polars_err!(ComputeError: "integer literal {} exceeds i64 range", v),
                )?;
                gpu_result(cudf::Scalar::new(val))
            }
            LiteralValue::Float(v) => gpu_result(cudf::Scalar::new(*v)),
            LiteralValue::Null => {
                polars_bail!(ComputeError: "GPU engine: null literal not supported in scalar path")
            }
            LiteralValue::String(_) => {
                polars_bail!(ComputeError: "GPU engine: string literal not supported in scalar path")
            }
            _ => {
                polars_bail!(ComputeError: "GPU engine: literal type {:?} not supported for scalar optimization", lit)
            }
        },
        _ => polars_bail!(ComputeError: "GPU engine: expected literal expression"),
    }
}

/// Evaluate a standalone aggregation expression (reduce column to scalar, broadcast to height).
fn eval_agg_expr(
    agg: &IRAggExpr,
    expr_arena: &Arena<AExpr>,
    df: &GpuDataFrame,
) -> PolarsResult<GpuColumn> {
    let height = df.height();
    match agg {
        IRAggExpr::Sum(input) => {
            reduce_and_broadcast(*input, ReduceOp::Sum, expr_arena, df, height)
        }
        IRAggExpr::Min { input, .. } => {
            reduce_and_broadcast(*input, ReduceOp::Min, expr_arena, df, height)
        }
        IRAggExpr::Max { input, .. } => {
            reduce_and_broadcast(*input, ReduceOp::Max, expr_arena, df, height)
        }
        IRAggExpr::Mean(input) => {
            reduce_and_broadcast(*input, ReduceOp::Mean, expr_arena, df, height)
        }
        IRAggExpr::Median(input) => {
            reduce_and_broadcast(*input, ReduceOp::Median, expr_arena, df, height)
        }
        IRAggExpr::Std(input, ddof) => {
            let input_node = *input;
            let ddof_val = *ddof;
            let col = eval_expr(input_node, expr_arena, df)?;
            let output_type = GpuDataType::new(GpuTypeId::Float64);
            let scalar = gpu_result(col.reduce_std_with_ddof(ddof_val as i32, output_type))?;
            broadcast_scalar(&scalar, height)
        }
        IRAggExpr::Var(input, ddof) => {
            let input_node = *input;
            let ddof_val = *ddof;
            let col = eval_expr(input_node, expr_arena, df)?;
            let output_type = GpuDataType::new(GpuTypeId::Float64);
            let scalar = gpu_result(col.reduce_var_with_ddof(ddof_val as i32, output_type))?;
            broadcast_scalar(&scalar, height)
        }
        IRAggExpr::Count(input, include_nulls) => {
            if *include_nulls {
                // Count all rows including nulls
                let count = height as u32;
                let data = vec![count; height];
                gpu_result(GpuColumn::from_slice(&data))
            } else {
                // Count only non-null rows: height - null_count
                let col = eval_expr(*input, expr_arena, df)?;
                let valid_count = (height - col.null_count()) as u32;
                let data = vec![valid_count; height];
                gpu_result(GpuColumn::from_slice(&data))
            }
        }
        IRAggExpr::NUnique(input) => {
            let col = eval_expr(*input, expr_arena, df)?;
            let n = gpu_result(col.distinct_count())? as u32;
            let data = vec![n; height];
            gpu_result(GpuColumn::from_slice(&data))
        }
        IRAggExpr::First(input) | IRAggExpr::Last(input) => {
            // In standalone context, just evaluate the input
            eval_expr(*input, expr_arena, df)
        }
        other => {
            polars_bail!(ComputeError: "GPU engine: unsupported standalone aggregation: {:?}", other)
        }
    }
}

/// Reduce a column and broadcast the scalar result to `height` rows.
fn reduce_and_broadcast(
    input_node: Node,
    op: ReduceOp,
    expr_arena: &Arena<AExpr>,
    df: &GpuDataFrame,
    height: usize,
) -> PolarsResult<GpuColumn> {
    let col = eval_expr(input_node, expr_arena, df)?;
    let dtype = col.data_type();

    // For mean/median/std/var, output is always float64
    let output_type = match op {
        ReduceOp::Mean | ReduceOp::Median | ReduceOp::Std | ReduceOp::Variance => {
            GpuDataType::new(GpuTypeId::Float64)
        }
        _ => dtype,
    };

    let scalar = gpu_result(col.reduce(op, output_type))?;

    // Extract scalar value and broadcast to all rows
    broadcast_scalar(&scalar, height)
}

/// Broadcast a scalar value to a column of the given height.
fn broadcast_scalar(scalar: &cudf::Scalar, height: usize) -> PolarsResult<GpuColumn> {
    let dtype = scalar.data_type();
    let tid = dtype.id();

    // If the scalar is not valid (null), create a null column of the correct type
    if !scalar.is_valid() {
        return match tid {
            GpuTypeId::Bool8 => {
                let opts: Vec<Option<bool>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_bool(&opts))
            }
            GpuTypeId::Int8 => {
                let opts: Vec<Option<i8>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_i8(&opts))
            }
            GpuTypeId::Int16 => {
                let opts: Vec<Option<i16>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_i16(&opts))
            }
            GpuTypeId::Int32 => {
                let opts: Vec<Option<i32>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_i32(&opts))
            }
            GpuTypeId::Int64 => {
                let opts: Vec<Option<i64>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_i64(&opts))
            }
            GpuTypeId::Uint8 => {
                let opts: Vec<Option<u8>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_u8(&opts))
            }
            GpuTypeId::Uint16 => {
                let opts: Vec<Option<u16>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_u16(&opts))
            }
            GpuTypeId::Uint32 => {
                let opts: Vec<Option<u32>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_u32(&opts))
            }
            GpuTypeId::Uint64 => {
                let opts: Vec<Option<u64>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_u64(&opts))
            }
            GpuTypeId::Float32 => {
                let opts: Vec<Option<f32>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_f32(&opts))
            }
            GpuTypeId::Float64 => {
                let opts: Vec<Option<f64>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_f64(&opts))
            }
            _ => {
                let opts: Vec<Option<i32>> = vec![None; height];
                gpu_result(GpuColumn::from_optional_i32(&opts))
            }
        };
    }

    match tid {
        GpuTypeId::Bool8 => {
            let v: bool = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Int8 => {
            let v: i8 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Int16 => {
            let v: i16 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Int32 => {
            let v: i32 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Int64 => {
            let v: i64 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Uint8 => {
            let v: u8 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Uint16 => {
            let v: u16 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Uint32 => {
            let v: u32 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Uint64 => {
            let v: u64 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Float32 => {
            let v: f32 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        GpuTypeId::Float64 => {
            let v: f64 = gpu_result(scalar.value())?;
            gpu_result(GpuColumn::from_slice(&vec![v; height]))
        }
        _ => polars_bail!(ComputeError: "GPU engine: cannot broadcast scalar of type {:?}", tid),
    }
}

/// Evaluate a built-in function expression.
fn eval_function(
    inputs: &[polars_plan::prelude::expr_ir::ExprIR],
    function: &FunctionExpr,
    expr_arena: &Arena<AExpr>,
    df: &GpuDataFrame,
) -> PolarsResult<GpuColumn> {
    match function {
        FunctionExpr::Abs => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.unary_op(UnaryOp::Abs))
        }
        FunctionExpr::Negate => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            // Negate: multiply by scalar -1 (avoids allocating a full-length column)
            let dtype = col.data_type();
            let tid = dtype.id();
            let neg_scalar = match tid {
                GpuTypeId::Int8 => gpu_result(cudf::Scalar::new(-1i8))?,
                GpuTypeId::Int16 => gpu_result(cudf::Scalar::new(-1i16))?,
                GpuTypeId::Int32 => gpu_result(cudf::Scalar::new(-1i32))?,
                GpuTypeId::Int64 => gpu_result(cudf::Scalar::new(-1i64))?,
                GpuTypeId::Float32 => gpu_result(cudf::Scalar::new(-1.0f32))?,
                GpuTypeId::Float64 => gpu_result(cudf::Scalar::new(-1.0f64))?,
                _ => {
                    polars_bail!(ComputeError: "GPU engine: Negate not supported for type {:?}", tid)
                }
            };
            gpu_result(col.binary_op_scalar(&neg_scalar, cudf::BinaryOp::Mul, dtype))
        }
        FunctionExpr::Boolean(bf) => eval_boolean_function(bf, inputs, expr_arena, df),
        FunctionExpr::FillNull => {
            // FillNull: inputs[0] is the column, inputs[1] is the fill value
            if inputs.len() != 2 {
                polars_bail!(ComputeError: "GPU engine: FillNull expects 2 inputs");
            }
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            let fill = eval_expr(inputs[1].node(), expr_arena, df)?;
            // Use copy_if_else: where NOT null, use original; where null, use fill
            let is_valid = gpu_result(col.is_valid())?;
            gpu_result(col.copy_if_else(&fill, &is_valid))
        }
        FunctionExpr::DropNulls => {
            polars_bail!(ComputeError: "GPU engine: DropNulls not supported in expression context (use Filter)")
        }
        FunctionExpr::NullCount => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            let nc = col.null_count() as u32;
            let height = df.height();
            gpu_result(GpuColumn::from_slice(&vec![nc; height]))
        }
        _ => {
            polars_bail!(ComputeError: "GPU engine: unsupported function expression {:?}", function)
        }
    }
}

/// Evaluate a boolean function.
fn eval_boolean_function(
    bf: &BooleanFunction,
    inputs: &[polars_plan::prelude::expr_ir::ExprIR],
    expr_arena: &Arena<AExpr>,
    df: &GpuDataFrame,
) -> PolarsResult<GpuColumn> {
    match bf {
        BooleanFunction::IsNull => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.is_null())
        }
        BooleanFunction::IsNotNull => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.is_valid())
        }
        BooleanFunction::IsNan => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.is_nan())
        }
        BooleanFunction::IsNotNan => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.is_not_nan())
        }
        BooleanFunction::IsFinite => {
            // is_finite = is_valid AND NOT(is_nan) AND NOT(is_inf)
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            let is_valid = gpu_result(col.is_valid())?;
            let not_nan = gpu_result(col.is_not_nan())?;
            let not_inf = gpu_result(col.is_not_inf())?;
            let valid_and_not_nan = gpu_result(is_valid.binary_op(
                &not_nan,
                cudf::BinaryOp::LogicalAnd,
                GpuDataType::new(GpuTypeId::Bool8),
            ))?;
            gpu_result(valid_and_not_nan.binary_op(
                &not_inf,
                cudf::BinaryOp::LogicalAnd,
                GpuDataType::new(GpuTypeId::Bool8),
            ))
        }
        BooleanFunction::IsInfinite => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            gpu_result(col.is_inf())
        }
        BooleanFunction::Any { ignore_nulls } => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            let height = df.height();
            if !ignore_nulls && col.null_count() > 0 {
                // Propagate null: if there are nulls and we are not ignoring them,
                // check if any non-null value is true first
                let scalar =
                    gpu_result(col.reduce(ReduceOp::Any, GpuDataType::new(GpuTypeId::Bool8)))?;
                if scalar.is_valid() {
                    let val: bool = gpu_result(scalar.value())?;
                    if val {
                        // At least one true found -> result is true
                        broadcast_scalar(&scalar, height)
                    } else {
                        // All non-null are false but there are nulls -> result is null
                        let opts: Vec<Option<bool>> = vec![None; height];
                        gpu_result(cudf::Column::from_optional_bool(&opts))
                    }
                } else {
                    // All null -> result is null
                    let opts: Vec<Option<bool>> = vec![None; height];
                    gpu_result(cudf::Column::from_optional_bool(&opts))
                }
            } else {
                let scalar =
                    gpu_result(col.reduce(ReduceOp::Any, GpuDataType::new(GpuTypeId::Bool8)))?;
                broadcast_scalar(&scalar, height)
            }
        }
        BooleanFunction::All { ignore_nulls } => {
            let col = eval_expr(inputs[0].node(), expr_arena, df)?;
            let height = df.height();
            if !ignore_nulls && col.null_count() > 0 {
                // Propagate null: if there are nulls and we are not ignoring them,
                // check if any non-null value is false first
                let scalar =
                    gpu_result(col.reduce(ReduceOp::All, GpuDataType::new(GpuTypeId::Bool8)))?;
                if scalar.is_valid() {
                    let val: bool = gpu_result(scalar.value())?;
                    if !val {
                        // At least one false found -> result is false
                        broadcast_scalar(&scalar, height)
                    } else {
                        // All non-null are true but there are nulls -> result is null
                        let opts: Vec<Option<bool>> = vec![None; height];
                        gpu_result(cudf::Column::from_optional_bool(&opts))
                    }
                } else {
                    // All null -> result is null
                    let opts: Vec<Option<bool>> = vec![None; height];
                    gpu_result(cudf::Column::from_optional_bool(&opts))
                }
            } else {
                let scalar =
                    gpu_result(col.reduce(ReduceOp::All, GpuDataType::new(GpuTypeId::Bool8)))?;
                broadcast_scalar(&scalar, height)
            }
        }
        _ => polars_bail!(ComputeError: "GPU engine: unsupported boolean function {:?}", bf),
    }
}
