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: pipeline.rs
    path of file: ./repos/arroyo/arroyo-sql/src
    
    the code of the file until where you have to start completion: #![allow(clippy::comparison_chain)]
use std::collections::HashMap;
use std::iter::once;

use std::time::Duration;
use std::unreachable;

use anyhow::{anyhow, bail};
use anyhow::{Ok, Result};
use arrow_schema::DataType;
use arroyo_datastream::{Operator, WindowType};
use datafusion_common::{DFField, ScalarValue};
use datafusion_expr::expr::ScalarUDF;
use datafusion_expr::{
    BinaryExpr, BuiltInWindowFunction, Expr, JoinConstraint, LogicalPlan, Window, WriteOp,
};

use quote::{quote, ToTokens};

use crate::code_gen::{CodeGenerator, ValuePointerContext, VecAggregationContext};
use crate::expressions::{
    AggregateComputation, AggregateResultExtraction, ExpressionContext, RustUdfExpression,
};
use crate::external::{ProcessingMode, SqlSink, SqlSource};
use crate::operators::{AsyncUdfProjection, UnnestFieldType, UnnestProjection};
use crate::schemas::window_type_def;
use crate::tables::{Insert, Table};
use crate::{
    expressions::{Column, ColumnExpression, Expression, SortExpression},
    operators::{AggregateProjection, Projection},
    types::{interval_month_day_nanos_to_duration, StructDef, StructField, TypeDef},
    ArroyoSchemaProvider,
};

#[derive(Debug, Clone)]
pub enum SqlOperator {
    Source(SourceOperator),
    Aggregator(Box<SqlOperator>, AggregateOperator),
    JoinOperator(Box<SqlOperator>, Box<SqlOperator>, JoinOperator),
    Window(Box<SqlOperator>, SqlWindowOperator),
    RecordTransform(Box<SqlOperator>, RecordTransform),
    Union(Vec<SqlOperator>),
    Sink(String, SqlSink, Box<SqlOperator>),
    NamedTable(String, Box<SqlOperator>),
}

#[derive(Debug, Clone)]
pub enum
    
    pub fn insert_sql_plan(&mut self, plan: &LogicalPlan) -> Result<SqlOperator> {
        // Check that output types are supported by Arroyo to avoid compile-time errors
        let _output_struct_def: StructDef = plan.schema().clone().try_into()?;
        match plan {
            LogicalPlan::Projection(projection) => self.insert_projection(projection),
            LogicalPlan::Filter(filter) => self.insert_filter(filter),
            LogicalPlan::Aggregate(aggregate) => self.insert_aggregation(aggregate),
            LogicalPlan::Sort(_) => bail!("sorting is not currently supported"),
            LogicalPlan::Join(join) => self.insert_join(join),
            LogicalPlan::CrossJoin(_) => bail!("cross joins are not currently supported"),
            LogicalPlan::Repartition(_) => bail!("repartitions are not currently supported"),
            LogicalPlan::Union(union) => self.insert_union(union),
            LogicalPlan::TableScan(table_scan) => self.insert_table_scan(table_scan),
            LogicalPlan::EmptyRelation(_) => bail!("empty relations not currently supported"),
            LogicalPlan::Subquery(subquery) => self.insert_sql_plan(&subquery.subquery),
            LogicalPlan::SubqueryAlias(subquery_alias) => {
                self.insert_subquery_alias(subquery_alias)
            }
            LogicalPlan::Limit(_) => bail!("limit not currently supported"),
            LogicalPlan::Ddl(ddl_statement) => match ddl_statement {
                datafusion_expr::DdlStatement::CreateExternalTable(_) => {
                    bail!("creating external tables is not currently supported")
                }
                datafusion_expr::DdlStatement::CreateMemoryTable(create_memory_table) => {
                    bail!(
                        "creating memory tables is not currently supported: {:?}",
                        create_memory_table.input,
                    )
                }
                datafusion_expr::DdlStatement::CreateView(_) => {
                    bail!("creating views is not currently supported")
                }
                datafusion_expr::DdlStatement::CreateCatalogSchema(_) => {
                    bail!("creating catalog schemas is not currently supported")
                }
                datafusion_expr::DdlStatement::CreateCatalog(_) => {
                    bail!("creating catalogs is not currently supported")
                }
                datafusion_expr::DdlStatement::DropTable(_) => {
                    bail!("dropping tables is not currently supported")
                }
                datafusion_expr::DdlStatement::DropView(_) => {
                    bail!("dropping views is not currently supported")
                }
                datafusion_expr::DdlStatement::DropCatalogSchema(_) => {
                    bail!("dropping catalog schemas is not currently supported")
                }
            },
            LogicalPlan::Values(_) => bail!("values are not currently supported"),
            LogicalPlan::Explain(_) => bail!("explain is not currently supported"),
            LogicalPlan::Analyze(_) => bail!("analyze is not currently supported"),
            LogicalPlan::Extension(_) => bail!("extensions are not currently supported"),
            LogicalPlan::Distinct(_) => bail!("distinct is not currently supported"),
            LogicalPlan::Window(window) => self.insert_window(window),
            LogicalPlan::Prepare(_) => bail!("prepare commands are not currently supported"),
            LogicalPlan::Dml(dml) => self.insert_dml(dml),
            LogicalPlan::DescribeTable(_) => bail!("describe table not currently supported"),
            LogicalPlan::Unnest(_) => bail!("unnest not currently supported"),
            LogicalPlan::Statement(_) => bail!("statements not currently supported"),
            LogicalPlan::Copy(_) => bail!("copy not currently supported"),
        }
    }