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: query.rs
    path of file: ./repos/LocustDB/src/engine/planning
    
    the code of the file until where you have to start completion: use crate::engine::*;
use crate::ingest::raw_val::RawVal;
use crate::mem_store::column::DataSource;
use crate::syntax::expression::*;
use crate::syntax::limit::*;
use crate::QueryError;
use std::collections::HashMap;
use std::collections::HashSet;
use std::iter::Iterator;
use std::sync::Arc;
use std::u64;

#[derive(Debug, Clone)]
pub struct ColumnInfo {
    pub expr: Expr,
    pub name: Option<String>,
}

/// NormalFormQuery observes the following invariants:
/// - none of the expressions contain aggregation functions
/// - if aggregate.len() > 0 then order_by.len() == 0 and vice versa
#[derive(Debug, Clone)]
pub struct NormalFormQuery {
    pub projection: Vec<ColumnInfo>,
    pub filter: Expr,
    pub aggregate: Vec<(Aggregator, ColumnInfo)>,
    pub order_by: Vec<(Expr, bool)>,
    pub limit: LimitClause,
}

#[derive(Debug, Clone)]
pub struct Query {
    pub select: Vec<ColumnInfo>,
    pub table: String,
    pub filter: Expr,
    pub order_by: Vec<(Expr, bool)>,
    pub limit: LimitClause,
}

impl NormalFormQuery {
    #[inline(never)] // produces more useful profiles
    pub fn run<'a>(
        &self,
        columns: &'a HashMap<String, Arc<dyn DataSource>>,
        explain: bool,
        show: bool,
        partition: usize,
        partition_len: usize,
    ) -> Result<(BatchResult<'a>, Option<String>), QueryError> {
        println!("Running {:?}", self);
        let limit = (self.limit.limit + self.limit.offset) as usize;
        println!("limit: {limit}");
        let mut planner = QueryPlanner::default();

        let (filter_plan, _) = QueryPlan::compile_expr(
            &self.filter,
            Filter::None,
            columns,
            partition_len,
            &mut planner,
        )?;
        let mut filter = match filter_plan.tag {
            EncodingType::U8 => Filter::U8(filter_plan.u8()?),
            EncodingType::NullableU8 => Filter::NullableU8(filter_plan.nullable_u8()?),
            _ => Filter::None,
        };

        // Sorting
        let mut sort_indices = None;
        for (plan, desc) in self.order_by.iter().rev() {
            let (ranking, _) = query_plan::order_preserving(
                QueryPlan::compile_expr(plan, filter, columns, partition_len, &mut planner)?,
                &mut planner,
            );

            // PERF: better criterion for using top_n
            // PERF: top_n for multiple columns?
            sort_indices = Some(if limit < partition_len / 2 && self.order_by.len() == 1 {
                planner.top_n(ranking, limit, *desc)
            } else {
                // PERF: sort directly if only single column selected
                match sort_indices {
                    None => {
                        let indices = planner.indices(ranking);
                        planner.sort_by(ranking, indices, *desc, false /* unstable sort */)
                    }
                    Some(indices) => {
                        planner.sort_by(ranking, indices, *desc,
    
    pub fn normalize(&self) -> Result<(NormalFormQuery, Option<NormalFormQuery>), QueryError> {
        let mut final_projection = Vec::<ColumnInfo>::new();
        let mut select = Vec::<ColumnInfo>::new();
        let mut aggregate = Vec::new();
        let mut aggregate_colnames = Vec::new();
        let mut select_colnames = Vec::new();
        for col_info in &self.select {
            let (full_expr, aggregates) = Query::extract_aggregators(
                &col_info.expr,
                &mut aggregate_colnames,
                col_info.name.clone(),
            )?;
            if aggregates.is_empty() {
                let column_name = format!("_cs{}", select_colnames.len());
                select_colnames.push(column_name.clone());
                select.push(ColumnInfo {
                    expr: full_expr,
                    name: col_info.name.clone(),
                });
                final_projection.push(ColumnInfo {
                    expr: Expr::ColName(column_name),
                    name: col_info.name.clone(),
                });
            } else {
                aggregate.extend(aggregates);
                final_projection.push(ColumnInfo {
                    expr: full_expr,
                    name: col_info.name.clone(),
                });
            }
        }

        let require_final_pass = (!aggregate.is_empty() && !self.order_by.is_empty())
            || final_projection
                .iter()
                .any(|col_info| !matches!(col_info.expr, Expr::ColName(_)));

        Ok(if require_final_pass {
            let mut final_order_by = Vec::new();
            for (expr, desc) in &self.order_by {
                let (full_expr, aggregates) =
                    Query::extract_aggregators(expr, &mut aggregate_colnames, None)?;
                if aggregates.is_empty() {
                    let column_name = format!("_cs{}", select_colnames.len());
                    select_colnames.push(column_name.clone());
                    select.push(ColumnInfo {
                        expr: full_expr,
                        name: None,
                    });
                    final_order_by.push((Expr::ColName(column_name), *desc));
                } else {
                    aggregate.extend(aggregates);
                    final_order_by.push((full_expr, *desc));
                }
            }
            (
                NormalFormQuery {
                    projection: select,
                    filter: self.filter.clone(),
                    aggregate,
                    order_by: vec![],
                    limit: LimitClause {
                        limit: u64::MAX,
                        offset: 0,
                    },
                },
                Some(NormalFormQuery {
                    projection: final_projection,
                    filter: Expr::Const(RawVal::Int(1)),
                    aggregate: vec![],
                    order_by: final_order_by,
                    limit: self.limit.clone(),
                }),
            )
        } else {
            (
                NormalFormQuery {
                    projection: select,
                    filter: self.filter.clone(),
                    aggregate,
                    order_by: self.order_by.clone(),
                    limit: self.limit.clone(),
                },
                None,
            )
        })
    }