You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: logical_expr.rs
    path of file: ./repos/lance/rust/lance/src/datafusion
    
    the code of the file until where you have to start completion: // Copyright 2023 Lance Developers.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Extends logical expression.

use arrow_schema::DataType;

use datafusion::logical_expr::ScalarFunctionDefinition;
use datafusion::logical_expr::{
    expr::ScalarFunction, BinaryExpr, BuiltinScalarFunction, GetFieldAccess, GetIndexedField,
    Operator,
};
use datafusion::prelude::*;
use datafusion::scalar::ScalarValue;
use lance_arrow::DataTypeExt;
use lance_datafusion::expr::safe_coerce_scalar;

use crate::datatypes::Schema;
use crate::{Error, Result};
use snafu::{location, Location};
/// Resolve a Value
fn resolve_value(expr: &Expr, data_type: &DataType) -> Result<Expr> {
    match expr {
        Expr::Literal(scalar_value) => {
            Ok(Expr::Literal(safe_coerce_scalar(scalar_value, data_type).ok_or_else(|| Error::IO {
                message: format!("Received literal {expr} and could not convert to literal of type '{data_type:?}'"),
                location: location!(),
            })?))
        }
        _ => Err(Error::IO {
            message: format!("Expected a literal of type '{data_type:?}' but received: {expr}"),
            location: location!(),
        }),
    }
}

/// Given a Expr::Column or Expr::GetIndexedField, get the data type of referenced
/// field in the schema.
///
/// If the column is not found in the schema, return None. If the expression is
/// not a field reference, also returns None.
pub fn resolve_column_type(expr: &Expr, schema: &Schema) -> Option<DataType> {
    let mut field_path = Vec::new();
    let mut current_expr = expr;
    // We are looping from outer-most reference to inner-most.
    loop {
        match current_expr {
            Expr::Column(c) => {
                field_path.push(c.name.as_str());
                break;
            }
            Expr::GetIndexedField(GetIndexedField { expr, field }) => {
                if let GetFieldAccess::NamedStructField {
                    name: ScalarValue::Utf8(Some(name)),
                } = field
                {
                    field_path.push(name);
                } else {
                    // We don't support other kinds of access right now.
                    return None;
                }
                current_expr = expr.as_ref();
            }
            _ => return None,
        }
    }

    let mut path_iter = field_path.iter().rev();
    let mut field = schema.field(path_iter.next()?)?;
    for name in path_iter {
        if field.data_type().is_struct() {
            field = field.children.iter().find(|f| &f.name == name)?;
        } else {
            return None;
        }
    }
    Some(field.data_type())
}

/// Resolve logical expression `expr`.
///
/// Parameters
///
/// - *expr*: a datafusion logical expression
/// - *schema*: lance schema.
pub fn resolve_expr(expr: &Expr, schema: &Schema) -> Result<Expr> {
    match expr {
     
    
pub fn resolve_expr(expr: &Expr, schema: &Schema) -> Result<Expr> {
    match expr {
        Expr::BinaryExpr(BinaryExpr { left, op, right }) => {
            if matches!(op, Operator::And | Operator::Or) {
                return Ok(Expr::BinaryExpr(BinaryExpr {
                    left: Box::new(resolve_expr(left.as_ref(), schema)?),
                    op: *op,
                    right: Box::new(resolve_expr(right.as_ref(), schema)?),
                }));
            }
            match (left.as_ref(), right.as_ref()) {
                (Expr::Column(_) | Expr::GetIndexedField(_), Expr::Literal(_)) => {
                    if let Some(resolved_type) = resolve_column_type(left.as_ref(), schema) {
                        Ok(Expr::BinaryExpr(BinaryExpr {
                            left: left.clone(),
                            op: *op,
                            right: Box::new(resolve_value(right.as_ref(), &resolved_type)?),
                        }))
                    } else {
                        Ok(expr.clone())
                    }
                }
                (Expr::Literal(_), Expr::Column(_) | Expr::GetIndexedField(_)) => {
                    if let Some(resolved_type) = resolve_column_type(right.as_ref(), schema) {
                        Ok(Expr::BinaryExpr(BinaryExpr {
                            left: Box::new(resolve_value(left.as_ref(), &resolved_type)?),
                            op: *op,
                            right: right.clone(),
                        }))
                    } else {
                        Ok(expr.clone())
                    }
                }
                // For cases complex expressions (not just literals) on right hand side like x = 1 + 1 + -2*2
                (Expr::Column(_) | Expr::GetIndexedField(_), Expr::BinaryExpr(r)) => {
                    if let Some(resolved_type) = resolve_column_type(left.as_ref(), schema) {
                        Ok(Expr::BinaryExpr(BinaryExpr {
                            left: left.clone(),
                            op: *op,
                            right: Box::new(Expr::BinaryExpr(BinaryExpr {
                                left: coerce_expr(&r.left, &resolved_type).map(Box::new)?,
                                op: r.op,
                                right: coerce_expr(&r.right, &resolved_type).map(Box::new)?,
                            })),
                        }))
                    } else {
                        Ok(expr.clone())
                    }
                }
                _ => Ok(expr.clone()),
            }
        }
        Expr::InList(in_list) => {
            if matches!(
                in_list.expr.as_ref(),
                Expr::Column(_) | Expr::GetIndexedField(_)
            ) {
                if let Some(resolved_type) = resolve_column_type(in_list.expr.as_ref(), schema) {
                    let resolved_values = in_list
                        .list
                        .iter()
                        .map(|val| coerce_expr(val, &resolved_type))
                        .collect::<Result<Vec<_>>>()?;
                    Ok(Expr::in_list(
                        in_list.expr.as_ref().clone(),
                        resolved_values,
                        in_list.negated,
                    ))
                } else {
                    Ok(expr.clone())
                }
            } else {
                Ok(expr.clone())
            }
        }
        _ => {
            // Passthrough
            Ok(expr.clone())
        }
    }
}