ormer 0.2.2

use super::common::common_helpers;
use crate::abstract_layer::DbType;
use crate::abstract_layer::common::{SingleSqlStatement, SqlExecutor, SqlStatement};
use crate::hooks::{HookContext, HookOperation};
use crate::migration::{SchemaColumn, schema_column};
use crate::model::{DbBackendTypeMapper, Model, Row, Value};
use crate::query::builder::{
    FourTableSelect, GroupedSelect, InnerJoinedSelect, LeftJoinedSelect, MultiTableSelect,
    RelatedSelect, RightJoinedSelect, Select, WhereExpr,
};
use crate::query::filter::FilterExpr;
use crate::query::insert::{
    InsertAssignment, InsertConflict, IntoInsertAssignment, IntoInsertDefaultColumn,
};
use crate::query::update::UpdateAssignment;
use crate::raw_sql::IntoRawSql;
use crate::utils::{FutureTraceExt, ResultTraceExt};
use crate::{
    impl_backend_executor_methods, impl_backend_four_table_executor_methods_with_lifetime,
    impl_backend_join_executor_methods_with_lifetime,
    impl_backend_multi_table_executor_methods_with_lifetime,
    impl_backend_related_executor_methods_with_lifetime, impl_insert_conflict_methods,
};
use chrono::{Datelike, Timelike};
use mysql_async::Pool;
use mysql_async::prelude::*;
use std::collections::HashMap;
use std::marker::PhantomData;

type ModelUpdateBatch = Vec<(Vec<(String, Value)>, Vec<FilterExpr>)>;

fn table_name_for<T: Model>() -> &'static str {
    T::table_name_for_db(DbType::MySQL)
}

fn mysql_value_from_ormer_value(value: &crate::model::Value) -> mysql_async::Value {
    match value {
        crate::model::Value::Integer(v) => mysql_async::Value::Int(*v),
        crate::model::Value::Text(v) => mysql_async::Value::Bytes(v.as_bytes().to_vec()),
        crate::model::Value::TextArray(v) => {
            mysql_async::Value::Bytes(crate::model::stringify_string_vec(v).into_bytes())
        }
        crate::model::Value::Real(v) => mysql_async::Value::Double(*v),
        crate::model::Value::Boolean(v) => mysql_async::Value::Int(if *v { 1 } else { 0 }),
        crate::model::Value::Duration(v) => {
            mysql_async::Value::Int(v.as_micros().min(i64::MAX as u128) as i64)
        }
        crate::model::Value::Bytes(v) => mysql_async::Value::Bytes(v.clone()),
        crate::model::Value::DateTime(v) => mysql_async::Value::Date(
            v.year() as u16,
            v.month() as u8,
            v.day() as u8,
            v.hour() as u8,
            v.minute() as u8,
            v.second() as u8,
            v.timestamp_subsec_micros(),
        ),
        crate::model::Value::Date(v) => {
            mysql_async::Value::Date(v.year() as u16, v.month() as u8, v.day() as u8, 0, 0, 0, 0)
        }
        crate::model::Value::Time(v) => mysql_async::Value::Time(
            false,
            0,
            v.hour() as u8,
            v.minute() as u8,
            v.second() as u8,
            v.nanosecond() / 1_000,
        ),
        crate::model::Value::Json(v) => mysql_async::Value::Bytes(v.to_string().into_bytes()),
        crate::model::Value::Uuid(v) => mysql_async::Value::Bytes(v.to_string().into_bytes()),
        crate::model::Value::BigInt(v) => mysql_async::Value::Int(*v as i64),
        crate::model::Value::IntegerArray(_)
        | crate::model::Value::BigIntArray(_)
        | crate::model::Value::NullableBigIntArray(_) => {
            panic!("MySQL backend does not support PostgreSQL array values")
        }
        crate::model::Value::Null => mysql_async::Value::NULL,
    }
}

/// MySQL 类型映射器
pub struct MySQLTypeMapper;

impl DbBackendTypeMapper for MySQLTypeMapper {
    fn sql_type(
        rust_type: &str,
        is_primary: bool,
        is_auto_increment: bool,
        is_nullable: bool,
        enum_variants: Option<&[&str]>,
    ) -> String {
        // MySQL 支持 ENUM 类型
        if let Some(variants) = enum_variants {
            let variants_str = variants
                .iter()
                .map(|v| format!("'{}'", v))
                .collect::<Vec<_>>()
                .join(", ");
            return common_helpers::sql_type_with_nullability(
                &format!("ENUM({})", variants_str),
                is_nullable,
            );
        }

        // 首先处理主键类型
        if is_primary {
            let int_type = match rust_type {
                "i8" | "i16" | "u8" => "TINYINT",
                "i32" | "u16" => "INT",
                "i64" | "u32" | "u64" => "BIGINT",
                _ => "INT", // 默认使用 INT
            };
            if is_auto_increment {
                return format!("{int_type} PRIMARY KEY AUTO_INCREMENT");
            } else {
                return format!("{int_type} PRIMARY KEY");
            }
        }

        // 基础类型映射
        let base_type = match rust_type {
            // 整数类型
            "i8" => "TINYINT",
            "i16" => "SMALLINT",
            "i32" => "INT",
            "i64" => "BIGINT",
            // 无符号整数
            "u8" => "TINYINT UNSIGNED",
            "u16" => "SMALLINT UNSIGNED",
            "u32" => "INT UNSIGNED",
            "u64" => "BIGINT UNSIGNED",
            // 浮点类型
            "f32" => "FLOAT",
            "f64" => "DOUBLE",
            // 时长类型
            "Duration" | "std::time::Duration" => "BIGINT",
            // 字符串类型
            "String" => "VARCHAR(255)",
            // 布尔类型
            "bool" => "TINYINT(1)",
            // 字节数组
            "Vec<u8>" | "&[u8]" => "BLOB",
            // 日期时间类型
            "DateTime"
            | "chrono::DateTime"
            | "chrono::DateTime<chrono::Utc>"
            | "NaiveDateTime"
            | "chrono::NaiveDateTime" => "DATETIME",
            "NaiveDate" | "chrono::NaiveDate" => "DATE",
            "NaiveTime" | "chrono::NaiveTime" => "TIME",
            // JSON 类型
            "JsonValue" | "serde_json::Value" => "JSON",
            // 默认使用 TEXT
            _ => "TEXT",
        };

        common_helpers::sql_type_with_nullability(base_type, is_nullable)
    }
}

/// MySQL 数据库连接封装
pub struct Database {
    pool: Pool,
}

/// 创建表执行器
pub struct CreateTableExecutor<'a, T: Model> {
    pool: &'a Pool,
    table_name: Option<String>,
    _marker: std::marker::PhantomData<T>,
}

impl<'a, T: Model> CreateTableExecutor<'a, T> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let create_sql = crate::generate_create_table_sql_with_name::<T>(
            crate::abstract_layer::DbType::MySQL,
            self.table_name.as_deref(),
        )?;
        Ok(SqlStatement::single(DbType::MySQL, create_sql, Vec::new()))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, T: Model> SqlExecutor for CreateTableExecutor<'a, T> {
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        CreateTableExecutor::to_sql(self)
    }

    async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
        let mut conn = self.pool.get_conn().trace().await?;
        for statement in sql.statements {
            conn.query_drop(&statement.sql).trace().await?;
        }
        Ok(())
    }
}

/// 删除表执行器
pub struct DropTableExecutor<'a, T: Model> {
    pool: &'a Pool,
    _marker: std::marker::PhantomData<T>,
}

impl<'a, T: Model> DropTableExecutor<'a, T> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        Ok(SqlStatement::single(
            DbType::MySQL,
            format!(
                "DROP TABLE IF EXISTS {}",
                common_helpers::quote_table_name::<T>(DbType::MySQL)
            ),
            Vec::new(),
        ))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, T: Model> SqlExecutor for DropTableExecutor<'a, T> {
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        DropTableExecutor::to_sql(self)
    }

    async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
        let mut conn = self.pool.get_conn().trace().await?;
        for statement in sql.statements {
            conn.query_drop(&statement.sql).trace().await?;
        }
        Ok(())
    }
}

/// 插入执行器
pub struct InsertExecutor<'a, I: crate::model::Insertable> {
    pool: &'a Pool,
    models: I,
    conflict: Option<InsertConflict>,
    _marker: std::marker::PhantomData<I::Model>,
}

impl_insert_conflict_methods!(InsertExecutor, with_conflict);

impl<'a, I: crate::model::Insertable + Send + Sync> InsertExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }

        let (sql, all_values) = common_helpers::build_insert_statement_with_conflict::<I::Model>(
            DbType::MySQL,
            &refs,
            self.conflict.as_ref(),
        )?;

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<<I::Model as Model>::AutoIncrementKeyType> {
        <Self as SqlExecutor>::execute(self).await
    }

    pub async fn returning(self) -> crate::Result<Vec<I::Model>> {
        Err(crate::ormer_error!(
            "MySQL does not support RETURNING clause"
        ))
    }

    #[allow(dead_code)]
    async fn insert_impl<T: Model>(&self, models: &[&T]) -> crate::Result<T::AutoIncrementKeyType> {
        if models.is_empty() {
            return Ok(T::AutoIncrementKeyType::default());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        let (sql, all_values) = common_helpers::build_insert_statement::<T>(DbType::MySQL, models);
        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        // 获取自增ID（如果有自增主键）
        let has_auto_increment = T::COLUMN_SCHEMA.iter().any(|c| c.is_auto_increment);
        if has_auto_increment {
            let last_id = conn.last_insert_id().unwrap_or(0);
            let result =
                common_helpers::convert_auto_increment_key::<T::AutoIncrementKeyType>(last_id)?;
            return Ok(result);
        }

        Ok(T::AutoIncrementKeyType::default())
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor for InsertExecutor<'a, I> {
    type Output = <I::Model as Model>::AutoIncrementKeyType;

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        InsertExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(<I::Model as Model>::AutoIncrementKeyType::default());
        }

        let hook_ctx = HookContext::new(HookOperation::Insert);
        self.models.run_before_insert(hook_ctx).await?;

        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;
        let mut conn = self.pool.get_conn().trace().await?;
        conn.exec_drop(&statement.sql, params).trace().await?;

        let has_auto_increment = I::Model::COLUMN_SCHEMA.iter().any(|c| c.is_auto_increment);
        let result = if has_auto_increment {
            let last_id = conn.last_insert_id().unwrap_or(0);
            common_helpers::convert_auto_increment_key::<Self::Output>(last_id)
        } else {
            Ok(<I::Model as Model>::AutoIncrementKeyType::default())
        }?;

        self.models.run_after_insert(hook_ctx).await?;
        Ok(result)
    }
}

pub struct InsertPartialExecutor<'a, T: Model> {
    db: &'a Database,
    assignments: Vec<InsertAssignment>,
    source_table: Option<&'static str>,
    _marker: PhantomData<T>,
}

impl<'a, T: Model> InsertPartialExecutor<'a, T> {
    fn with_assignments(mut self, assignments: Vec<InsertAssignment>) -> Self {
        self.assignments.extend(assignments);
        self
    }

    fn with_source_table(mut self, source_table: &'static str) -> Self {
        self.source_table = Some(source_table);
        self
    }

    pub fn set<F, A>(mut self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> A,
        A: IntoInsertAssignment<T>,
    {
        self.assignments
            .push(f(T::Where::default()).into_insert_assignment());
        self
    }

    pub fn default<F, C>(mut self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> C,
        C: IntoInsertDefaultColumn<T>,
    {
        self.assignments.push(InsertAssignment::default(
            f(T::Where::default()).into_insert_default_column(),
        ));
        self
    }

    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        common_helpers::validate_insert_model_table::<T>(DbType::MySQL, self.source_table)?;
        let statement =
            common_helpers::build_partial_insert_statement::<T>(DbType::MySQL, &self.assignments)?;
        Ok(SqlStatement::single(
            DbType::MySQL,
            statement.sql,
            statement.params,
        ))
    }

    pub async fn execute(self) -> crate::Result<<T as Model>::AutoIncrementKeyType>
    where
        T: Send + Sync,
    {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, T: Model + Send + Sync> SqlExecutor for InsertPartialExecutor<'a, T> {
    type Output = <T as Model>::AutoIncrementKeyType;

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        InsertPartialExecutor::to_sql(self)
    }

    async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(<T as Model>::AutoIncrementKeyType::default());
        }

        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;
        let mut conn = self.db.pool.get_conn().trace().await?;
        conn.exec_drop(&statement.sql, params).trace().await?;

        let has_auto_increment = T::COLUMN_SCHEMA.iter().any(|c| c.is_auto_increment);
        if has_auto_increment {
            let last_id = conn.last_insert_id().unwrap_or(0);
            common_helpers::convert_auto_increment_key::<Self::Output>(last_id)
        } else {
            Ok(<T as Model>::AutoIncrementKeyType::default())
        }
    }
}

/// 插入或更新执行器
pub struct InsertOrUpdateExecutor<'a, I: crate::model::Insertable> {
    pool: &'a Pool,
    models: I,
    _marker: std::marker::PhantomData<I::Model>,
}

impl<'a, I: crate::model::Insertable + Send + Sync> InsertOrUpdateExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }

        let (mut sql, all_values) = common_helpers::build_batch_insert_statement::<I::Model>(
            DbType::MySQL,
            "INSERT INTO",
            <I::Model as Model>::table_name_for_db(DbType::MySQL),
            I::Model::COLUMNS,
            &refs,
            common_helpers::BatchInsertValuesMode::All,
        );

        sql.push_str(" ON DUPLICATE KEY UPDATE ");
        let mut first = true;
        for col_name in I::Model::COLUMNS.iter() {
            if !first {
                sql.push_str(", ");
            }
            sql.push_str(&common_helpers::quote_mysql_values_assignment(
                DbType::MySQL,
                col_name,
            ));
            first = false;
        }

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }

    #[allow(dead_code)]
    async fn insert_or_update_batch<T: Model>(&self, models: &[&T]) -> crate::Result<()> {
        if models.is_empty() {
            return Ok(());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        // 构建批量插入或更新的 SQL: INSERT INTO table (cols) VALUES (...), (...) ON DUPLICATE KEY UPDATE ...
        let (mut sql, all_values) = common_helpers::build_batch_insert_statement::<T>(
            DbType::MySQL,
            "INSERT INTO",
            T::table_name_for_db(DbType::MySQL),
            T::COLUMNS,
            models,
            common_helpers::BatchInsertValuesMode::All,
        );

        // 添加 ON DUPLICATE KEY UPDATE 子句
        sql.push_str(" ON DUPLICATE KEY UPDATE ");
        let mut first = true;
        for col_name in T::COLUMNS.iter() {
            if !first {
                sql.push_str(", ");
            }
            sql.push_str(&common_helpers::quote_mysql_values_assignment(
                DbType::MySQL,
                col_name,
            ));
            first = false;
        }

        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        Ok(())
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor for InsertOrUpdateExecutor<'a, I> {
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        InsertOrUpdateExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(());
        }
        let hook_ctx = HookContext::new(HookOperation::Insert);
        self.models.run_before_insert(hook_ctx).await?;
        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;
        let mut conn = self.pool.get_conn().trace().await?;
        conn.exec_drop(&statement.sql, params).trace().await?;
        self.models.run_after_insert(hook_ctx).await?;
        Ok(())
    }
}

/// 插入或忽略执行器
pub struct InsertOrIgnoreExecutor<'a, I: crate::model::Insertable> {
    pool: &'a Pool,
    models: I,
    _marker: std::marker::PhantomData<I::Model>,
}

impl<'a, I: crate::model::Insertable + Send + Sync> InsertOrIgnoreExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }

        let (sql, all_values) = common_helpers::build_batch_insert_statement::<I::Model>(
            DbType::MySQL,
            "INSERT IGNORE INTO",
            <I::Model as Model>::table_name_for_db(DbType::MySQL),
            I::Model::COLUMNS,
            &refs,
            common_helpers::BatchInsertValuesMode::All,
        );

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }

    #[allow(dead_code)]
    async fn insert_or_ignore_batch<T: Model>(&self, models: &[&T]) -> crate::Result<()> {
        if models.is_empty() {
            return Ok(());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        // 构建批量插入或忽略的 SQL: INSERT IGNORE INTO table (cols) VALUES (...), (...)
        let (sql, all_values) = common_helpers::build_batch_insert_statement::<T>(
            DbType::MySQL,
            "INSERT IGNORE INTO",
            T::table_name_for_db(DbType::MySQL),
            T::COLUMNS,
            models,
            common_helpers::BatchInsertValuesMode::All,
        );

        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        Ok(())
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor for InsertOrIgnoreExecutor<'a, I> {
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        InsertOrIgnoreExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(());
        }
        let hook_ctx = HookContext::new(HookOperation::Insert);
        self.models.run_before_insert(hook_ctx).await?;
        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;
        let mut conn = self.pool.get_conn().trace().await?;
        conn.exec_drop(&statement.sql, params).trace().await?;
        self.models.run_after_insert(hook_ctx).await?;
        Ok(())
    }
}

impl Database {
    /// 连接到 MySQL 数据库
    pub async fn connect(_db_type: super::DbType, connection_string: &str) -> crate::Result<Self> {
        // 解析连接字符串
        let opts = mysql_async::Opts::from_url(connection_string)
            .trace_for("mysql_async::Opts::from_url")?;

        let pool = Pool::new(opts);

        Ok(Self { pool })
    }

    /// 从已有的 mysql_async Pool 创建 Database
    ///
    /// mysql_async::Pool 本身就是连接池，内部使用 Arc 管理，clone 是轻量操作。
    pub fn from_pool(pool: Pool) -> Self {
        Self { pool }
    }

    /// 创建表 - 返回执行器
    pub fn create_table<T: Model>(&self) -> CreateTableExecutor<'_, T> {
        CreateTableExecutor {
            pool: &self.pool,
            table_name: None,
            _marker: std::marker::PhantomData,
        }
    }

    /// 验证表结构是否与模型定义匹配
    pub async fn validate_table<T: Model>(&self) -> crate::Result<()> {
        let mut conn = self.pool.get_conn().trace().await?;

        // 检查表是否存在
        let table_exists = self.check_table_exists::<T>(&mut conn).trace().await?;

        if !table_exists {
            return Err(crate::ormer_error!(
                "Schema mismatch: table {}, reason: Table does not exist",
                T::TABLE_NAME
            ));
        }

        // 表已存在，验证表结构
        self.validate_table_schema::<T>(&mut conn).await
    }

    /// 检查表是否存在
    async fn check_table_exists<T: Model>(
        &self,
        conn: &mut mysql_async::Conn,
    ) -> crate::Result<bool> {
        let sql = "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = ?";

        let result: Option<u64> = conn
            .exec_first(sql, (table_name_for::<T>(),))
            .trace()
            .await?;

        Ok(result.unwrap_or(0) > 0)
    }

    /// 验证表结构是否与模型定义匹配（内部使用）
    async fn validate_table_schema<T: Model>(
        &self,
        conn: &mut mysql_async::Conn,
    ) -> crate::Result<()> {
        // 查询表的列信息
        let sql = r#"
            SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?
            ORDER BY ORDINAL_POSITION
        "#;

        let rows: Vec<mysql_async::Row> = conn.exec(sql, (table_name_for::<T>(),)).trace().await?;

        // 收集实际的表结构
        let mut actual_columns: Vec<(String, String, bool)> = Vec::new();
        for row in rows {
            let name: String = row.get(0).unwrap_or_default();
            let col_type: String = row.get(1).unwrap_or_default();
            let is_nullable: String = row.get(2).unwrap_or_default();
            actual_columns.push((name, col_type, is_nullable == "YES"));
        }

        // 比较列数量
        if actual_columns.len() != T::COLUMNS.len() {
            return Err(crate::ormer_error!(
                "Schema mismatch: table {}, reason: Column count mismatch: expected {}, but actual is {}",
                T::TABLE_NAME,
                T::COLUMNS.len(),
                actual_columns.len()
            ));
        }

        // 比较每一列的定义
        for (i, expected_col) in T::COLUMN_SCHEMA.iter().enumerate() {
            if i >= actual_columns.len() {
                return Err(crate::ormer_error!(
                    "Schema mismatch: table {}, reason: Missing column: {}",
                    T::TABLE_NAME,
                    expected_col.name
                ));
            }

            let (actual_name, actual_type, actual_nullable) = &actual_columns[i];

            // 检查列名
            if actual_name != expected_col.name {
                return Err(crate::ormer_error!(
                    "Schema mismatch: table {}, reason: Column name mismatch at position {}: expected '{}', but actual is '{}'",
                    T::TABLE_NAME,
                    i,
                    expected_col.name,
                    actual_name
                ));
            }

            // 检查列类型（只比较基础类型，不包含约束）
            let expected_type = crate::abstract_layer::DbType::MySQL.sql_type(
                expected_col.rust_type,
                expected_col.is_primary,
                expected_col.is_auto_increment,
                expected_col.is_nullable,
                expected_col.enum_variants,
            );

            // 对于类型比较，我们需要提取基础类型（不包含 AUTO_INCREMENT, PRIMARY KEY, NOT NULL 等约束）
            let type_to_compare = if expected_col.is_primary {
                // 主键的基础类型
                match expected_col.rust_type {
                    "i8" | "i16" | "u8" => "TINYINT".to_string(),
                    "i32" | "u16" => "INT".to_string(),
                    "i64" | "u32" | "u64" => "BIGINT".to_string(),
                    _ => "INT".to_string(),
                }
            } else {
                // 非主键列，提取基础类型（去掉 NOT NULL）
                let full_type = crate::abstract_layer::DbType::MySQL.sql_type(
                    expected_col.rust_type,
                    false,
                    expected_col.is_auto_increment,
                    expected_col.is_nullable,
                    expected_col.enum_variants,
                );
                // 去掉 " NOT NULL" 后缀
                full_type.replace(" NOT NULL", "")
            };

            if !self.types_compatible(actual_type, &type_to_compare) {
                return Err(crate::ormer_error!(
                    "Schema mismatch: table {}, reason: Column type mismatch for '{}': expected '{expected_type}', but actual is '{actual_type}'",
                    T::TABLE_NAME,
                    expected_col.name
                ));
            }

            // 检查 NOT NULL 约束（主键列除外，因为主键自动 NOT NULL）
            if !expected_col.is_primary {
                let expected_nullable = expected_col.is_nullable;
                if *actual_nullable != expected_nullable {
                    return Err(crate::ormer_error!(
                        "Schema mismatch: table {}, reason: Column nullability mismatch for '{}': expected {}NULL, but actual is {}NULL",
                        T::TABLE_NAME,
                        expected_col.name,
                        if expected_nullable { "" } else { "NOT " },
                        if *actual_nullable { "" } else { "NOT " }
                    ));
                }
            }
        }

        Ok(())
    }

    /// 检查 SQL 类型是否兼容
    fn types_compatible(&self, actual: &str, expected: &str) -> bool {
        // 标准化类型名称
        fn normalize(s: &str) -> String {
            let upper = s.to_uppercase();
            // 提取基础类型名（去掉括号内的参数）
            let base_type = if let Some(pos) = upper.find('(') {
                &upper[..pos]
            } else {
                &upper[..]
            };

            match base_type {
                // 整数类型
                "TINYINT" => "TINYINT".to_string(),
                "SMALLINT" => "SMALLINT".to_string(),
                "MEDIUMINT" => "MEDIUMINT".to_string(),
                "INT" | "INTEGER" => "INT".to_string(),
                "BIGINT" => "BIGINT".to_string(),
                // 无符号整数
                t if t.ends_with(" UNSIGNED") => {
                    let unsigned_type = t.replace(" ", "");
                    match unsigned_type.as_str() {
                        "TINYINTUNSIGNED" => "TINYINT UNSIGNED".to_string(),
                        "SMALLINTUNSIGNED" => "SMALLINT UNSIGNED".to_string(),
                        "MEDIUMINTUNSIGNED" => "MEDIUMINT UNSIGNED".to_string(),
                        "INTUNSIGNED" | "INTEGERUNSIGNED" => "INT UNSIGNED".to_string(),
                        "BIGINTUNSIGNED" => "BIGINT UNSIGNED".to_string(),
                        _ => t.to_string(),
                    }
                }
                // 浮点类型
                "FLOAT" => "FLOAT".to_string(),
                "DOUBLE" | "DOUBLEPRECISION" => "DOUBLE".to_string(),
                // 字符串类型（忽略长度参数）
                "VARCHAR" | "CHAR" | "TEXT" | "TINYTEXT" | "MEDIUMTEXT" | "LONGTEXT" => {
                    "VARCHAR".to_string()
                }
                // 布尔类型（MySQL 使用 TINYINT(1) 存储布尔值）
                "BOOL" | "BOOLEAN" => "TINYINT".to_string(),
                // 字节类型
                "BLOB" | "TINYBLOB" | "MEDIUMBLOB" | "LONGBLOB" | "VARBINARY" | "BINARY" => {
                    "BLOB".to_string()
                }
                // 其他
                _ => base_type.to_string(),
            }
        }

        normalize(actual) == normalize(expected)
    }

    /// 插入记录 - 返回执行器
    pub fn insert<I: crate::model::Insertable>(&self, models: I) -> InsertExecutor<'_, I> {
        InsertExecutor {
            pool: &self.pool,
            models,
            conflict: None,
            _marker: std::marker::PhantomData,
        }
    }

    pub fn insert_partial<T: Model>(&self) -> InsertPartialExecutor<'_, T> {
        InsertPartialExecutor {
            db: self,
            assignments: Vec::new(),
            source_table: None,
            _marker: PhantomData,
        }
    }

    pub fn insert_model<T>(
        &self,
        model: impl crate::model::InsertModel<T>,
    ) -> InsertPartialExecutor<'_, T>
    where
        T: Model,
    {
        self.insert_partial::<T>()
            .with_source_table(model.insert_table_name())
            .with_assignments(model.insert_assignments())
    }

    /// 插入或更新记录 - 返回执行器
    pub fn insert_or_update<I: crate::model::Insertable>(
        &self,
        models: I,
    ) -> InsertOrUpdateExecutor<'_, I> {
        InsertOrUpdateExecutor {
            pool: &self.pool,
            models,
            _marker: std::marker::PhantomData,
        }
    }

    /// 插入或忽略记录 - 返回执行器（存在重复主键时忽略）
    pub fn insert_or_ignore<I: crate::model::Insertable>(
        &self,
        models: I,
    ) -> InsertOrIgnoreExecutor<'_, I> {
        InsertOrIgnoreExecutor {
            pool: &self.pool,
            models,
            _marker: std::marker::PhantomData,
        }
    }

    /// 批量插入记录，返回自增主键值（如果有自增主键）或 ()
    /// 对于批量插入，返回的是第一条插入记录的自增ID（即最小值）
    pub(crate) async fn insert_impl<T: Model>(
        &self,
        models: &[&T],
    ) -> crate::Result<T::AutoIncrementKeyType> {
        if models.is_empty() {
            return Ok(T::AutoIncrementKeyType::default());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        let (sql, all_values) = common_helpers::build_insert_statement::<T>(DbType::MySQL, models);
        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        // 获取自增ID（如果有自增主键）
        let has_auto_increment = T::COLUMN_SCHEMA.iter().any(|c| c.is_auto_increment);
        if has_auto_increment {
            let last_id = conn.last_insert_id().unwrap_or(0);
            let result =
                common_helpers::convert_auto_increment_key::<T::AutoIncrementKeyType>(last_id)?;
            Ok(result)
        } else {
            Ok(T::AutoIncrementKeyType::default())
        }
    }

    /// 批量插入或更新记录（遇到重复键时更新）
    pub async fn insert_or_update_batch<T: Model>(&self, models: &[&T]) -> crate::Result<()> {
        if models.is_empty() {
            return Ok(());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        // 构建批量插入或更新的 SQL: INSERT INTO table (cols) VALUES (...), (...) ON DUPLICATE KEY UPDATE ...
        let (mut sql, all_values) = common_helpers::build_batch_insert_statement::<T>(
            DbType::MySQL,
            "INSERT INTO",
            T::table_name_for_db(DbType::MySQL),
            T::COLUMNS,
            models,
            common_helpers::BatchInsertValuesMode::All,
        );

        // 添加 ON DUPLICATE KEY UPDATE 子句
        sql.push_str(" ON DUPLICATE KEY UPDATE ");
        let mut first = true;
        for col_name in T::COLUMNS.iter() {
            if !first {
                sql.push_str(", ");
            }
            sql.push_str(&common_helpers::quote_mysql_values_assignment(
                DbType::MySQL,
                col_name,
            ));
            first = false;
        }

        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        Ok(())
    }

    /// 批量插入或忽略记录（遇到重复键时忽略）
    pub async fn insert_or_ignore_batch<T: Model>(&self, models: &[&T]) -> crate::Result<()> {
        if models.is_empty() {
            return Ok(());
        }

        let mut conn = self.pool.get_conn().trace().await?;

        // 构建批量插入或忽略的 SQL: INSERT IGNORE INTO table (cols) VALUES (...), (...)
        let (sql, all_values) = common_helpers::build_batch_insert_statement::<T>(
            DbType::MySQL,
            "INSERT IGNORE INTO",
            T::table_name_for_db(DbType::MySQL),
            T::COLUMNS,
            models,
            common_helpers::BatchInsertValuesMode::All,
        );

        let params = values_to_params(&all_values)?;

        conn.exec_drop(&sql, params).trace().await?;

        Ok(())
    }

    /// 创建 Select 查询执行器
    pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
        SelectExecutor {
            select: Select::<T>::new(),
            pool: &self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建分组聚合查询执行器
    pub fn select_column<T: Model, V>(&self) -> GroupedSelectExecutor<'_, T, V> {
        GroupedSelectExecutor {
            select: GroupedSelect::<T, V>::new(),
            pool: &self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建 Delete 执行器
    pub fn delete<T: Model>(&self) -> DeleteExecutor<'_, T> {
        DeleteExecutor {
            filters: Vec::new(),
            pool: &self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建 Update 执行器
    pub fn update<T: Model>(&self) -> UpdateExecutor<'_, T> {
        UpdateExecutor {
            sets: Vec::new(),
            filters: Vec::new(),
            model_updates: Vec::new(),
            pool: &self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建 Related 查询执行器（关联查询）
    pub fn related<T: Model + 'static, R: Model>(&self) -> RelatedSelectExecutor<'_, T, R> {
        RelatedSelectExecutor {
            select: Select::<T>::new().from::<T, R>(),
            pool: &self.pool,
            _marker: PhantomData,
        }
    }

    /// 开始事务
    pub async fn begin(&self) -> crate::Result<Transaction<'_>> {
        let mut conn = self.pool.get_conn().trace().await?;

        conn.query_drop("START TRANSACTION").trace().await?;

        Ok(Transaction {
            conn: Some(conn),
            pool: &self.pool,
            committed: false,
            rolled_back: false,
        })
    }

    /// 删除表 - 返回执行器
    pub fn drop_table<T: Model>(&self) -> DropTableExecutor<'_, T> {
        DropTableExecutor {
            pool: &self.pool,
            _marker: std::marker::PhantomData,
        }
    }

    /// 执行原生非查询 SQL 并返回影响的行数
    pub async fn execute_sql(&self, sql: impl IntoRawSql) -> crate::Result<u64> {
        let sql = sql.into_raw_sql();
        let (sql, params) = sql.render(DbType::MySQL)?;
        self.exec_raw(&sql, params).await
    }

    pub(crate) async fn select_raw<V, C>(&self, sql: &str, params: Vec<Value>) -> crate::Result<C>
    where
        V: crate::model::FromRowValues,
        C: FromIterator<V>,
    {
        let mut conn = self.pool.get_conn().trace().await?;
        let mysql_params = values_to_params(&params)?;
        let rows: Vec<mysql_async::Row> = if mysql_params.is_empty() {
            conn.query(sql).trace().await?
        } else {
            conn.exec(sql, mysql_async::Params::Positional(mysql_params))
                .trace()
                .await?
        };

        let mut results = Vec::new();
        for row in rows {
            results.push(common_helpers::decode_row_values_from_indexed_values(
                row.columns_ref().len(),
                |i| convert_mysql_value(&row, i),
            )?);
        }
        Ok(results.into_iter().collect())
    }

    pub(crate) async fn exec_raw(&self, sql: &str, params: Vec<Value>) -> crate::Result<u64> {
        let mut conn = self.pool.get_conn().trace().await?;
        let mysql_params = values_to_params(&params)?;
        if mysql_params.is_empty() {
            conn.query_drop(sql).trace().await?;
        } else {
            conn.exec_drop(sql, mysql_async::Params::Positional(mysql_params))
                .trace()
                .await?;
        }
        Ok(conn.affected_rows())
    }

    pub(crate) async fn migration_history(&self) -> crate::Result<Vec<(u64, String, u64)>> {
        let mut conn = self.pool.get_conn().trace().await?;
        let rows: Vec<mysql_async::Row> = conn
            .query("SELECT version, name, checksum FROM __ormer_migrations ORDER BY version")
            .trace()
            .await?;
        rows.into_iter()
            .map(|row| {
                let version = row
                    .get::<u64, _>(0)
                    .ok_or_else(|| crate::ormer_error!("Migration version is NULL"))?;
                let name = row.get::<String, _>(1).unwrap_or_default();
                let checksum = row
                    .get::<String, _>(2)
                    .ok_or_else(|| crate::ormer_error!("Migration checksum is NULL"))?
                    .parse::<u64>()
                    .map_err(|_| crate::ormer_error!("Migration checksum is invalid"))?;
                Ok((version, name, checksum))
            })
            .collect()
    }

    pub(crate) async fn schema_columns(
        &self,
        table_name: &str,
    ) -> crate::Result<Option<Vec<SchemaColumn>>> {
        let mut conn = self.pool.get_conn().trace().await?;
        let exists: Option<u64> = conn
            .exec_first(
                "SELECT COUNT(*) FROM information_schema.tables \
                 WHERE table_schema = DATABASE() AND table_name = ?",
                (table_name,),
            )
            .trace()
            .await?;
        if exists.unwrap_or(0) == 0 {
            return Ok(None);
        }
        let rows: Vec<mysql_async::Row> = conn
            .exec(
                "SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE, COLUMN_KEY \
                 FROM information_schema.columns \
                 WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? \
                 ORDER BY ORDINAL_POSITION",
                (table_name,),
            )
            .trace()
            .await?;
        let columns = rows
            .into_iter()
            .map(|row| {
                let name: String = row.get(0).unwrap_or_default();
                let type_name: String = row.get(1).unwrap_or_default();
                let nullable: String = row.get(2).unwrap_or_default();
                let key: String = row.get(3).unwrap_or_default();
                schema_column(name, type_name, nullable == "YES", key == "PRI")
            })
            .collect();
        Ok(Some(columns))
    }

    /// 检查连接是否有效
    pub async fn is_valid(&self) -> bool {
        if let Ok(mut conn) = self.pool.get_conn().trace().await {
            conn.query_drop("SELECT 1").trace().await.is_ok()
        } else {
            false
        }
    }
}

/// MySQL 事务对象
pub struct Transaction<'a> {
    conn: Option<mysql_async::Conn>,
    pool: &'a Pool,
    committed: bool,
    rolled_back: bool,
}

/// 事务中的插入执行器 - 现在持有连接的引用而不是所有权  
pub struct TransactionInsertExecutor<'a, I: crate::model::Insertable> {
    conn: &'a mut Option<mysql_async::Conn>,
    models: I,
    conflict: Option<InsertConflict>,
    _marker: std::marker::PhantomData<&'a ()>,
}

impl_insert_conflict_methods!(TransactionInsertExecutor);

impl<'a, I: crate::model::Insertable + Send + Sync> TransactionInsertExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }
        let (sql, all_values) = common_helpers::build_insert_statement_with_conflict::<I::Model>(
            DbType::MySQL,
            &refs,
            self.conflict.as_ref(),
        )?;

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<<I::Model as Model>::AutoIncrementKeyType> {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor
    for TransactionInsertExecutor<'a, I>
{
    type Output = <I::Model as Model>::AutoIncrementKeyType;

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        TransactionInsertExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(<<I::Model as Model>::AutoIncrementKeyType>::default());
        }
        let hook_ctx = HookContext::new(HookOperation::Insert).transaction();
        self.models.run_before_insert(hook_ctx).await?;
        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;

        let result = if let Some(conn) = self.conn.as_mut() {
            conn.exec_drop(&statement.sql, params).trace().await?;

            let has_auto_increment = I::Model::COLUMN_SCHEMA.iter().any(|c| c.is_auto_increment);
            if has_auto_increment {
                let last_id = conn.last_insert_id().unwrap_or(0);
                common_helpers::convert_auto_increment_key::<
                    <I::Model as Model>::AutoIncrementKeyType,
                >(last_id)
            } else {
                Ok(<<I::Model as Model>::AutoIncrementKeyType>::default())
            }
        } else {
            Ok(<<I::Model as Model>::AutoIncrementKeyType>::default())
        }?;

        self.models.run_after_insert(hook_ctx).await?;
        Ok(result)
    }
}

/// 事务中的插入或更新执行器 - 现在持有连接的引用而不是所有权  
pub struct TransactionInsertOrUpdateExecutor<'a, I: crate::model::Insertable> {
    conn: &'a mut Option<mysql_async::Conn>,
    models: I,
    _marker: std::marker::PhantomData<&'a ()>,
}

impl<'a, I: crate::model::Insertable + Send + Sync> TransactionInsertOrUpdateExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }
        let (mut sql, all_values) = common_helpers::build_batch_insert_statement::<I::Model>(
            DbType::MySQL,
            "INSERT INTO",
            <I::Model as Model>::table_name_for_db(DbType::MySQL),
            I::Model::COLUMNS,
            &refs,
            common_helpers::BatchInsertValuesMode::All,
        );

        sql.push_str(" ON DUPLICATE KEY UPDATE ");
        let mut first = true;
        for col_name in I::Model::COLUMNS.iter() {
            if !first {
                sql.push_str(", ");
            }
            sql.push_str(&common_helpers::quote_mysql_values_assignment(
                DbType::MySQL,
                col_name,
            ));
            first = false;
        }

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor
    for TransactionInsertOrUpdateExecutor<'a, I>
{
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        TransactionInsertOrUpdateExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(());
        }
        let hook_ctx = HookContext::new(HookOperation::Insert).transaction();
        self.models.run_before_insert(hook_ctx).await?;
        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;

        if let Some(conn) = self.conn.as_mut() {
            conn.exec_drop(&statement.sql, params).trace().await?;
        }

        self.models.run_after_insert(hook_ctx).await?;
        Ok(())
    }
}

impl<'a> Transaction<'a> {
    pub(crate) async fn exec_raw(&mut self, sql: &str, params: Vec<Value>) -> crate::Result<u64> {
        let conn = self
            .conn
            .as_mut()
            .ok_or_else(|| crate::ormer_error!("Transaction connection is unavailable"))?;
        let mysql_params = values_to_params(&params)?;
        if mysql_params.is_empty() {
            conn.query_drop(sql).trace().await?;
        } else {
            conn.exec_drop(sql, mysql_async::Params::Positional(mysql_params))
                .trace()
                .await?;
        }
        Ok(conn.affected_rows())
    }

    pub(crate) async fn select_raw<V, C>(
        &mut self,
        sql: &str,
        params: Vec<Value>,
    ) -> crate::Result<C>
    where
        V: crate::model::FromRowValues,
        C: FromIterator<V>,
    {
        let conn = self
            .conn
            .as_mut()
            .ok_or_else(|| crate::ormer_error!("Transaction connection is unavailable"))?;
        let mysql_params = values_to_params(&params)?;
        let rows: Vec<mysql_async::Row> = if mysql_params.is_empty() {
            conn.query(sql).trace().await?
        } else {
            conn.exec(sql, mysql_async::Params::Positional(mysql_params))
                .trace()
                .await?
        };

        let mut results = Vec::new();
        for row in rows {
            results.push(common_helpers::decode_row_values_from_indexed_values(
                row.columns_ref().len(),
                |i| convert_mysql_value(&row, i),
            )?);
        }
        Ok(results.into_iter().collect())
    }

    /// 提交事务
    pub async fn commit(mut self) -> crate::Result<()> {
        if self.committed || self.rolled_back {
            return Err(crate::ormer_error!(
                "Transaction already committed or rolled back".to_string(),
            ));
        }
        if let Some(mut conn) = self.conn.take() {
            conn.query_drop("COMMIT").trace().await?;
        }
        self.committed = true;
        Ok(())
    }

    /// 回滚事务
    pub async fn rollback(mut self) -> crate::Result<()> {
        if self.committed || self.rolled_back {
            return Err(crate::ormer_error!(
                "Transaction already committed or rolled back".to_string(),
            ));
        }
        if let Some(mut conn) = self.conn.take() {
            conn.query_drop("ROLLBACK").trace().await?;
        }
        self.rolled_back = true;
        Ok(())
    }

    /// 创建 Select 查询执行器
    pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
        SelectExecutor {
            select: Select::<T>::new(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建分组聚合查询执行器
    pub fn select_column<T: Model, V>(&self) -> GroupedSelectExecutor<'_, T, V> {
        GroupedSelectExecutor {
            select: GroupedSelect::<T, V>::new(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建 Delete 执行器
    pub fn delete<T: Model>(&self) -> DeleteExecutor<'_, T> {
        DeleteExecutor {
            filters: Vec::new(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 创建 Update 执行器
    pub fn update<T: Model>(&self) -> UpdateExecutor<'_, T> {
        UpdateExecutor {
            sets: Vec::new(),
            filters: Vec::new(),
            model_updates: Vec::new(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 插入记录 - 返回执行器  
    pub fn insert<I: crate::model::Insertable>(
        &mut self,
        models: I,
    ) -> TransactionInsertExecutor<'_, I> {
        TransactionInsertExecutor {
            conn: &mut self.conn,
            models,
            conflict: None,
            _marker: std::marker::PhantomData,
        }
    }

    /// 插入或更新记录 - 返回执行器  
    pub fn insert_or_update<I: crate::model::Insertable>(
        &mut self,
        models: I,
    ) -> TransactionInsertOrUpdateExecutor<'_, I> {
        TransactionInsertOrUpdateExecutor {
            conn: &mut self.conn,
            models,
            _marker: std::marker::PhantomData,
        }
    }

    /// 插入或忽略记录 - 返回执行器（存在重复主键时忽略）
    pub fn insert_or_ignore<I: crate::model::Insertable>(
        &mut self,
        models: I,
    ) -> TransactionInsertOrIgnoreExecutor<'_, I> {
        TransactionInsertOrIgnoreExecutor {
            conn: &mut self.conn,
            models,
            _marker: std::marker::PhantomData,
        }
    }

    /// 批量插入或更新记录（遇到重复键时更新）
    pub async fn insert_or_update_batch<T: Model>(&mut self, models: &[&T]) -> crate::Result<()> {
        if models.is_empty() {
            return Ok(());
        }

        // 构建批量插入或更新的 SQL: INSERT INTO table (cols) VALUES (...), (...) ON DUPLICATE KEY UPDATE ...
        let (mut sql, all_values) = common_helpers::build_batch_insert_statement::<T>(
            DbType::MySQL,
            "INSERT INTO",
            T::table_name_for_db(DbType::MySQL),
            T::COLUMNS,
            models,
            common_helpers::BatchInsertValuesMode::All,
        );

        // 添加 ON DUPLICATE KEY UPDATE 子句
        sql.push_str(" ON DUPLICATE KEY UPDATE ");
        let mut first = true;
        for col_name in T::COLUMNS.iter() {
            if !first {
                sql.push_str(", ");
            }
            sql.push_str(&common_helpers::quote_mysql_values_assignment(
                DbType::MySQL,
                col_name,
            ));
            first = false;
        }

        let params = values_to_params(&all_values)?;

        if let Some(ref mut conn) = self.conn {
            conn.exec_drop(&sql, params).trace().await?;
        }

        Ok(())
    }
}

/// 事务中的插入或忽略执行器
pub struct TransactionInsertOrIgnoreExecutor<'a, I: crate::model::Insertable> {
    conn: &'a mut Option<mysql_async::Conn>,
    models: I,
    _marker: std::marker::PhantomData<I::Model>,
}

impl<'a, I: crate::model::Insertable + Send + Sync> TransactionInsertOrIgnoreExecutor<'a, I> {
    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let refs = self.models.as_refs();
        if refs.is_empty() {
            return Ok(SqlStatement::batch(DbType::MySQL, Vec::new()));
        }
        let (sql, all_values) = common_helpers::build_batch_insert_statement::<I::Model>(
            DbType::MySQL,
            "INSERT IGNORE INTO",
            <I::Model as Model>::table_name_for_db(DbType::MySQL),
            I::Model::COLUMNS,
            &refs,
            common_helpers::BatchInsertValuesMode::All,
        );

        Ok(SqlStatement::single(DbType::MySQL, sql, all_values))
    }

    pub async fn execute(self) -> crate::Result<()> {
        <Self as SqlExecutor>::execute(self).await
    }
}

impl<'a, I: crate::model::Insertable + Send + Sync> SqlExecutor
    for TransactionInsertOrIgnoreExecutor<'a, I>
{
    type Output = ();

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        TransactionInsertOrIgnoreExecutor::to_sql(self)
    }

    async fn execute_with_sql(mut self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(());
        }
        let hook_ctx = HookContext::new(HookOperation::Insert).transaction();
        self.models.run_before_insert(hook_ctx).await?;
        let statement = &sql.statements[0];
        let params = values_to_params(&statement.params)?;

        if let Some(conn) = self.conn.as_mut() {
            conn.exec_drop(&statement.sql, params).trace().await?;
        }

        self.models.run_after_insert(hook_ctx).await?;
        Ok(())
    }
}

/// LEFT JOIN 查询执行器
pub struct LeftJoinedSelectExecutor<'a, T: Model, J: Model> {
    select: LeftJoinedSelect<T, J>,
    pool: &'a Pool,
    _marker: PhantomData<(T, J)>,
}

/// INNER JOIN 查询执行器
pub struct InnerJoinedSelectExecutor<'a, T: Model, J: Model> {
    select: InnerJoinedSelect<T, J>,
    pool: &'a Pool,
    _marker: PhantomData<(T, J)>,
}

/// RIGHT JOIN 查询执行器
pub struct RightJoinedSelectExecutor<'a, T: Model, J: Model> {
    select: RightJoinedSelect<T, J>,
    pool: &'a Pool,
    _marker: PhantomData<(T, J)>,
}

/// Select 查询执行器
pub struct SelectExecutor<'a, T: Model> {
    select: Select<T>,
    pool: &'a Pool,
    _marker: PhantomData<T>,
}

/// 映射查询结果执行器
pub struct MappedSelectExecutor<'a, T: Model, V> {
    select: crate::query::builder::MappedSelect<T, V>,
    pool: &'a Pool,
    _marker: PhantomData<(T, V)>,
}

/// 分组聚合查询执行器
pub struct GroupedSelectExecutor<'a, T: Model, V> {
    select: GroupedSelect<T, V>,
    pool: &'a Pool,
    _marker: PhantomData<(T, V)>,
}

impl<'a, T: Model, V> MappedSelectExecutor<'a, T, V> {
    /// 生成子查询SQL和参数
    pub fn to_subquery_sql(&self) -> (String, Vec<crate::model::Value>) {
        self.select.to_sql_with_params(DbType::MySQL)
    }

    /// 执行查询并收集结果
    pub fn collect<C: FromIterator<V> + 'static>(&self) -> MappedCollectFuture<'a, T, V, C>
    where
        T: 'static,
        V: crate::model::FromRowValues + 'static,
    {
        MappedCollectFuture {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 克隆executor（保持相同的pool引用）
    pub fn clone_with_pool(&self) -> Self {
        Self {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }
}

/// 映射查询收集Future
pub struct MappedCollectFuture<'a, T: Model, V, C> {
    select: crate::query::builder::MappedSelect<T, V>,
    pool: &'a Pool,
    _marker: PhantomData<(T, V, C)>,
}

impl<
    'a,
    T: Model + 'static + Send,
    V: crate::model::FromRowValues + 'static + Send,
    C: FromIterator<V> + 'static,
> std::future::IntoFuture for MappedCollectFuture<'a, T, V, C>
{
    type Output = crate::Result<C>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move {
            let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

            let mut conn = self.pool.get_conn().trace().await?;

            // 将ormer::Value转换为mysql_async::Params
            let mysql_params: Vec<mysql_async::Value> = params
                .into_iter()
                .map(|v| mysql_value_from_ormer_value(&v))
                .collect();

            let rows: Vec<mysql_async::Row> = if mysql_params.is_empty() {
                conn.query(&sql).trace().await?
            } else {
                conn.exec(&sql, mysql_async::Params::Positional(mysql_params))
                    .trace()
                    .await?
            };

            let mut results = Vec::new();
            for row in rows {
                let v = common_helpers::decode_row_values_from_indexed_values(
                    row.columns_ref().len(),
                    |i| convert_mysql_value(&row, i),
                )?;
                results.push(v);
            }

            Ok(results.into_iter().collect())
        })
    }
}

impl_backend_executor_methods!(SelectExecutor, pool, &'a Pool, Select);

impl<'a, T: Model> SelectExecutor<'a, T> {
    pub(crate) fn select_model<R: Model>(&self) -> SelectExecutor<'a, R> {
        SelectExecutor {
            select: Select::new(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 克隆executor（保持相同的pool引用）
    pub fn clone_with_pool(&self) -> Self {
        Self {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加 LEFT JOIN 查询
    pub fn left_join<J: Model>(
        self,
        f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
    ) -> LeftJoinedSelectExecutor<'a, T, J> {
        LeftJoinedSelectExecutor {
            select: self.select.left_join::<J>(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加 INNER JOIN 查询
    pub fn inner_join<J: Model>(
        self,
        f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
    ) -> InnerJoinedSelectExecutor<'a, T, J> {
        InnerJoinedSelectExecutor {
            select: self.select.inner_join::<J>(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加 RIGHT JOIN 查询
    pub fn right_join<J: Model>(
        self,
        f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
    ) -> RightJoinedSelectExecutor<'a, T, J> {
        RightJoinedSelectExecutor {
            select: self.select.right_join::<J>(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 映射查询结果到自定义类型
    pub fn map_to<F, M>(self, f: F) -> MappedSelectExecutor<'a, T, M::Output>
    where
        F: FnOnce(T::Where) -> M,
        M: crate::query::builder::MapToResult,
    {
        let mapped_select = self.select.map_to(f);
        MappedSelectExecutor {
            select: mapped_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 忽略指定字段，查询时用默认常量替代真实列值
    pub fn ignore<F, M>(self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> M,
        M: crate::query::builder::MapToResult,
    {
        Self {
            select: self.select.ignore(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 选择列（支持聚合函数）- 转换为分组查询
    pub fn select_column<F, V>(self, f: F) -> GroupedSelectExecutor<'a, T, V>
    where
        F: FnOnce(T::Where) -> V,
        V: crate::query::builder::SelectColumnResult,
    {
        let grouped_select = self.select.select_column(f);
        GroupedSelectExecutor {
            select: grouped_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 执行查询并收集结果
    pub fn collect<C: FromIterator<T> + 'static>(&self) -> CollectFuture<'a, T, C> {
        CollectFuture {
            executor: self.clone_with_pool(),
            _marker: PhantomData,
        }
    }

    /// 执行查询并返回第一条记录
    pub fn first(self) -> FirstFuture<'a, T> {
        FirstFuture { executor: self }
    }

    /// COUNT 聚合函数
    pub fn count<F, C>(self, f: F) -> AggregateFuture<'a, T, usize>
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
    {
        let aggregate_select = self.select.count(f);
        AggregateFuture {
            aggregate_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// SUM 聚合函数
    pub fn sum<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
        C: crate::query::builder::AggregateResultType + 'static,
    {
        let aggregate_select = self.select.sum(f);
        AggregateFuture {
            aggregate_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// AVG 聚合函数
    pub fn avg<F, C>(self, f: F) -> AggregateFuture<'a, T, Option<f64>>
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
        C: crate::query::builder::AggregateResultType + 'static,
    {
        let aggregate_select = self.select.avg(f);
        AggregateFuture {
            aggregate_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// MAX 聚合函数
    pub fn max<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
        C: crate::query::builder::AggregateResultType + 'static,
    {
        let aggregate_select = self.select.max(f);
        AggregateFuture {
            aggregate_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// MIN 聚合函数
    pub fn min<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C, T>,
        C: crate::query::builder::AggregateResultType + 'static,
    {
        let aggregate_select = self.select.min(f);
        AggregateFuture {
            aggregate_select,
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加关联表查询（支持2个泛型参数，第一个必须与T相同）
    /// select::<User>().from::<User, Role>()
    pub fn from<T2, R: Model>(self) -> RelatedSelectExecutor<'a, T, R>
    where
        T2: Model + 'static,
    {
        RelatedSelectExecutor {
            select: self.select.from::<T2, R>(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加关联表查询（支持3个表）
    /// select::<User>().from3::<User, Role, Permission>()
    pub fn from3<T2, R1: Model, R2: Model>(self) -> MultiTableSelectExecutor<'a, T, R1, R2>
    where
        T2: Model + 'static,
    {
        MultiTableSelectExecutor {
            select: self.select.from3::<T2, R1, R2>(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加关联表查询（支持4个表）
    /// select::<User>().from4::<User, Role, Permission, Department>()
    pub fn from4<T2, R1: Model, R2: Model, R3: Model>(
        self,
    ) -> FourTableSelectExecutor<'a, T, R1, R2, R3>
    where
        T2: Model + 'static,
    {
        FourTableSelectExecutor {
            select: self.select.from4::<T2, R1, R2, R3>(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }
}

/// Collect future - 允许 .collect::<Vec<_>>().await 语法
pub struct CollectFuture<'a, T: Model, C: FromIterator<T>> {
    executor: SelectExecutor<'a, T>,
    _marker: PhantomData<C>,
}

/// First future for单条记录查询
pub struct FirstFuture<'a, T: Model> {
    executor: SelectExecutor<'a, T>,
}

/// Aggregate future for聚合函数执行
pub struct AggregateFuture<'a, T: Model, R> {
    aggregate_select: crate::query::builder::AggregateSelect<T, R>,
    pool: &'a Pool,
    _marker: PhantomData<(T, R)>,
}

impl<'a, T: Model + 'static + Send, R: crate::model::FromValue + 'static + Send>
    std::future::IntoFuture for AggregateFuture<'a, T, R>
{
    type Output = crate::Result<R>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move {
            let (sql, params) = self.aggregate_select.to_sql_with_params(DbType::MySQL);

            // 将ormer::Value转换为mysql_async Value
            let mut conn = self.pool.get_conn().trace().await?;

            // 构建参数
            let mysql_params: Vec<mysql_async::Value> = params
                .into_iter()
                .map(|v| mysql_value_from_ormer_value(&v))
                .collect();

            let mut exec_result = conn.exec_iter(&sql, mysql_params).trace().await?;

            if let Some(row) = exec_result.next().trace().await? {
                // 获取第一个列的值
                let value: Option<mysql_async::Value> = row.get(0);
                let value = value.unwrap_or(mysql_async::Value::NULL);

                // 调试输出
                println!(
                    "DEBUG MySQL aggregate raw value type: {:?}",
                    std::mem::discriminant(&value)
                );

                // 将mysql_async::Value转换为ormer::Value
                let ormer_value = match value {
                    mysql_async::Value::Int(i) => crate::model::Value::Integer(i),
                    mysql_async::Value::UInt(u) => crate::model::Value::Integer(u as i64),
                    mysql_async::Value::Float(f) => crate::model::Value::Real(f as f64),
                    mysql_async::Value::Double(d) => crate::model::Value::Real(d),
                    mysql_async::Value::Bytes(b) => {
                        // 尝试将字节解析为数值（MySQL聚合函数可能返回字符串形式的数值）
                        if let Ok(s) = String::from_utf8(b.clone()) {
                            // 尝试解析为整数
                            if let Ok(i) = s.parse::<i64>() {
                                crate::model::Value::Integer(i)
                            } else if let Ok(f) = s.parse::<f64>() {
                                // 尝试解析为浮点数
                                crate::model::Value::Real(f)
                            } else {
                                // 作为文本处理
                                crate::model::Value::Text(s)
                            }
                        } else {
                            crate::model::Value::Null
                        }
                    }
                    mysql_async::Value::Date(_, _, _, _, _, _, _)
                    | mysql_async::Value::Time(_, _, _, _, _, _) => crate::model::Value::Null,
                    mysql_async::Value::NULL => crate::model::Value::Null,
                };

                // 调试输出
                println!("DEBUG MySQL aggregate ormer_value: {:?}", ormer_value);

                // 使用 FromValue 转换为目标类型
                R::from_value(&ormer_value)
            } else {
                // 如果没有结果，返回 NULL 的转换
                R::from_value(&crate::model::Value::Null)
            }
        })
    }
}

impl<'a, T: Model + 'static + Send, C: FromIterator<T> + 'static> std::future::IntoFuture
    for CollectFuture<'a, T, C>
{
    type Output = crate::Result<C>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

impl<'a, T: Model + 'static + Send + std::marker::Sync> std::future::IntoFuture
    for FirstFuture<'a, T>
{
    type Output = crate::Result<Option<T>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move {
            let results: Vec<T> = self.executor.collect_inner().await?;
            Ok(results.into_iter().next())
        })
    }
}

impl<'a, T: Model> SelectExecutor<'a, T> {
    async fn collect_inner<C: FromIterator<T>>(self) -> crate::Result<C> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mut conn = self.pool.get_conn().trace().await?;

        let mysql_params = values_to_params(&params)?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();

        for row in rows {
            let model = common_helpers::decode_model_from_indexed_values::<T, _>(0, |i| {
                convert_mysql_value(&row, i)
            })?;
            results.push(model);
        }

        Ok(results.into_iter().collect())
    }
}

/// Delete 执行器
pub struct DeleteExecutor<'a, T: Model> {
    filters: Vec<FilterExpr>,
    pool: &'a Pool,
    _marker: PhantomData<T>,
}

impl<'a, T: Model> DeleteExecutor<'a, T> {
    /// 添加 WHERE 条件
    pub fn filter<F>(mut self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> WhereExpr,
    {
        let where_obj = T::Where::default();
        let expr = f(where_obj);
        self.filters.push(expr.into());
        self
    }

    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let (sql, params) = self.build_sql_with_params();
        Ok(SqlStatement::single(DbType::MySQL, sql, params))
    }

    /// 执行删除操作并返回影响的行数
    pub async fn execute(self) -> crate::Result<u64> {
        <Self as SqlExecutor>::execute(self).await
    }

    pub async fn returning(self) -> crate::Result<Vec<T>> {
        Err(crate::ormer_error!(
            "MySQL does not support RETURNING clause"
        ))
    }

    fn build_sql_with_params(&self) -> (String, Vec<Value>) {
        let mut sql = format!(
            "DELETE FROM {}",
            common_helpers::quote_table_name::<T>(DbType::MySQL)
        );
        let mut params = Vec::new();

        if !self.filters.is_empty() {
            sql.push_str(" WHERE ");
            let mut param_idx: usize = 1;
            for (i, filter) in self.filters.iter().enumerate() {
                if i > 0 {
                    sql.push_str(" AND ");
                }
                let _ = common_helpers::format_filter_with_params(
                    filter,
                    &mut sql,
                    &mut param_idx,
                    &mut params,
                    DbType::MySQL,
                );
            }
        }

        (sql, params)
    }
}

impl<'a, T: Model> SqlExecutor for DeleteExecutor<'a, T> {
    type Output = u64;

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        DeleteExecutor::to_sql(self)
    }

    async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
        if sql.statements.is_empty() {
            return Ok(0);
        }

        let statement = &sql.statements[0];
        let mysql_params = values_to_params(&statement.params)?;
        let mut conn = self.pool.get_conn().trace().await?;
        conn.exec_drop(&statement.sql, mysql_params).trace().await?;
        Ok(conn.affected_rows())
    }
}

impl<'a, T: Model + 'static + Send> std::future::IntoFuture for DeleteExecutor<'a, T> {
    type Output = crate::Result<u64>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.execute().await })
    }
}

/// Update 执行器
pub struct UpdateExecutor<'a, T: Model> {
    sets: Vec<UpdateAssignment>,
    filters: Vec<FilterExpr>,
    model_updates: ModelUpdateBatch,
    pool: &'a Pool,
    _marker: PhantomData<T>,
}

impl<'a, T: Model> UpdateExecutor<'a, T> {
    /// 添加 WHERE 条件
    pub fn filter<F>(mut self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> WhereExpr,
    {
        let where_obj = T::Where::default();
        let expr = f(where_obj);
        self.filters.push(expr.into());
        self
    }

    /// 设置要更新的字段
    pub fn set<F>(mut self, f: F) -> Self
    where
        F: FnOnce(&mut T::Update),
    {
        let mut update = T::Update::default();
        f(&mut update);
        self.sets
            .extend(<T::Update as crate::query::update::UpdateFields>::assignments(&update));
        self
    }

    /// 从模型实例设置所有非主键字段，并自动添加主键作为 WHERE 条件
    ///
    /// ```ignore
    /// let user = User { id: 1, name: "Bob".into(), age: 25, email: Some("bob@test.com".into()) };
    /// db.update::<User>().set_model(&user).execute().await?;
    /// ```
    pub fn set_model(mut self, model: &T) -> Self {
        let mut model_sets = Vec::new();
        for (col_name, value) in model.non_pk_field_values() {
            model_sets.push((col_name.to_string(), value));
        }
        let pk_columns = T::primary_key_columns();
        let pk_values = model.primary_key_values();
        let mut model_filters = Vec::new();
        for (col, val) in pk_columns.iter().zip(pk_values) {
            let filter_val =
                crate::abstract_layer::common::common_helpers::value_to_filter_value(&val);
            model_filters.push(crate::query::filter::FilterExpr::Comparison {
                column: col.to_string(),
                operator: "=".to_string(),
                value: filter_val,
            });
        }
        self.model_updates.push((model_sets, model_filters));
        self
    }

    pub fn set_model_fields(mut self, model: &T, fields: &[String]) -> Self {
        let model_sets = model
            .non_pk_field_values_for_columns(fields)
            .into_iter()
            .map(|(col_name, value)| (col_name.to_string(), value))
            .collect::<Vec<_>>();
        let pk_columns = T::primary_key_columns();
        let pk_values = model.primary_key_values();
        let model_filters = pk_columns
            .iter()
            .zip(pk_values)
            .map(|(col, val)| crate::query::filter::FilterExpr::Comparison {
                column: col.to_string(),
                operator: "=".to_string(),
                value: crate::abstract_layer::common::common_helpers::value_to_filter_value(&val),
            })
            .collect();

        if !model_sets.is_empty() {
            self.model_updates.push((model_sets, model_filters));
        }
        self
    }

    pub fn to_sql(&self) -> crate::Result<SqlStatement> {
        let statements = self.build_all_sql()?;
        Ok(SqlStatement::batch(
            DbType::MySQL,
            statements
                .into_iter()
                .map(|(sql, params)| SingleSqlStatement::new(sql, params))
                .collect(),
        ))
    }

    /// 执行更新操作
    pub async fn execute(self) -> crate::Result<u64> {
        <Self as SqlExecutor>::execute(self).await
    }

    pub async fn returning(self) -> crate::Result<Vec<T>> {
        Err(crate::ormer_error!(
            "MySQL does not support RETURNING clause"
        ))
    }

    fn build_all_sql(&self) -> crate::Result<Vec<(String, Vec<crate::model::Value>)>> {
        let mut statements = Vec::new();

        // Base UPDATE from sets/filters
        if !self.sets.is_empty() || (self.model_updates.is_empty() && !self.filters.is_empty()) {
            let mut sql = format!(
                "UPDATE {} SET ",
                common_helpers::quote_table_name::<T>(DbType::MySQL)
            );
            let mut params = Vec::new();
            let mut first = true;
            for assignment in &self.sets {
                if !first {
                    sql.push_str(", ");
                }
                sql.push_str(&common_helpers::format_update_assignment(
                    DbType::MySQL,
                    assignment,
                    &mut params,
                ));
                first = false;
            }
            if !self.filters.is_empty() {
                sql.push_str(" WHERE ");
                let mut param_idx = params.len() + 1;
                for (i, filter) in self.filters.iter().enumerate() {
                    if i > 0 {
                        sql.push_str(" AND ");
                    }
                    let _ = common_helpers::format_filter_with_params(
                        filter,
                        &mut sql,
                        &mut param_idx,
                        &mut params,
                        DbType::MySQL,
                    );
                }
            }
            statements.push((sql, params));
        }

        // Model UPDATE statements
        for (model_sets, model_filters) in &self.model_updates {
            let mut sql = format!(
                "UPDATE {} SET ",
                common_helpers::quote_table_name::<T>(DbType::MySQL)
            );
            let mut params = Vec::new();
            let mut first = true;
            for (col_name, value) in model_sets {
                if !first {
                    sql.push_str(", ");
                }
                sql.push_str(&common_helpers::quote_assignment(
                    DbType::MySQL,
                    col_name,
                    "?",
                ));
                params.push(value.clone());
                first = false;
            }
            if !model_filters.is_empty() {
                sql.push_str(" WHERE ");
                let mut param_idx = params.len() + 1;
                for (i, filter) in model_filters.iter().enumerate() {
                    if i > 0 {
                        sql.push_str(" AND ");
                    }
                    let _ = common_helpers::format_filter_with_params(
                        filter,
                        &mut sql,
                        &mut param_idx,
                        &mut params,
                        DbType::MySQL,
                    );
                }
            }
            statements.push((sql, params));
        }

        Ok(statements)
    }
}

impl<'a, T: Model> SqlExecutor for UpdateExecutor<'a, T> {
    type Output = u64;

    fn to_sql(&self) -> crate::Result<SqlStatement> {
        UpdateExecutor::to_sql(self)
    }

    async fn execute_with_sql(self, sql: SqlStatement) -> crate::Result<Self::Output> {
        let mut conn = self.pool.get_conn().trace().await?;
        let mut total: u64 = 0;
        for statement in &sql.statements {
            let mysql_params = values_to_params(&statement.params)?;
            let result = conn.exec_iter(&statement.sql, mysql_params).trace().await?;
            total += result.affected_rows();
        }
        Ok(total)
    }
}

impl<'a, T: Model + 'static + Send> std::future::IntoFuture for UpdateExecutor<'a, T> {
    type Output = crate::Result<u64>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.execute().await })
    }
}

/// 将 ormer Value 转换为 mysql_async 参数
fn values_to_params(values: &[crate::model::Value]) -> crate::Result<Vec<mysql_async::Value>> {
    let mut params: Vec<mysql_async::Value> = Vec::new();

    for value in values {
        params.push(mysql_value_from_ormer_value(value));
    }

    Ok(params)
}

/// Related 查询执行器（支持2表关联查询）
pub struct RelatedSelectExecutor<'a, T: Model, R: Model> {
    select: RelatedSelect<T, R>,
    pool: &'a Pool,
    _marker: PhantomData<(T, R)>,
}

/// SelectStream - 流式查询执行器 (MySQL)
pub struct SelectStream<'a, T: Model> {
    select: Select<T>,
    pool: &'a mysql_async::Pool,
    _marker: std::marker::PhantomData<&'a T>,
}

impl<'a, T: Model> SelectExecutor<'a, T> {
    /// 创建流式查询执行器
    pub fn stream(self) -> SelectStream<'a, T> {
        SelectStream {
            select: self.select,
            pool: self.pool,
            _marker: std::marker::PhantomData,
        }
    }
}

impl<'a, T: Model + 'static> SelectStream<'a, T> {
    /// 返回异步迭代器 (真正的流式查询)
    ///
    /// 使用 mysql_async 的 Query::stream() 实现真正的流式查询，
    /// 逐行读取数据而不是一次性加载所有结果到内存中。
    pub async fn into_iter(self) -> crate::Result<SelectStreamIterator<'a, T>> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        // 将参数转换为 mysql_async::Value
        let mysql_params: Vec<mysql_async::Value> =
            params.iter().map(mysql_value_from_ormer_value).collect();

        // 获取连接
        let conn = self.pool.get_conn().trace().await?;

        // 使用 Query::stream() 实现真正的流式查询
        // 该方法返回 'static 的流，流拥有连接的所有权
        use mysql_async::prelude::Query;
        let stream = sql
            .with(mysql_params)
            .stream::<mysql_async::Row, _>(conn)
            .trace()
            .await?;

        Ok(SelectStreamIterator {
            stream: Some(stream),
            _marker: std::marker::PhantomData,
        })
    }
}

/// 将 MySQL Row 解析为 Model
fn parse_mysql_row<T: Model>(row: &mysql_async::Row) -> crate::Result<T> {
    common_helpers::decode_model_from_indexed_values::<T, _>(0, |i| convert_mysql_value(row, i))
}

/// SelectStreamIterator - 真正的流式查询迭代器 (MySQL)
///
/// 使用 mysql_async 的 Query::stream() 实现真正的流式查询，
/// 逐行读取数据而不是一次性加载所有结果到内存中。
///
/// 该流拥有连接的所有权，当流被消费完毕或丢弃时，
/// 连接会自动释放回连接池。
pub struct SelectStreamIterator<'a, T: Model> {
    stream: Option<
        mysql_async::ResultSetStream<
            'static,
            'static,
            'static,
            mysql_async::Row,
            mysql_async::BinaryProtocol,
        >,
    >,
    _marker: std::marker::PhantomData<&'a T>,
}

impl<'a, T: Model + 'static> SelectStreamIterator<'a, T> {
    /// 获取下一行数据 (真正的流式查询)
    ///
    /// 逐行从数据库中读取数据，内存占用为 O(1)。
    pub async fn next(&mut self) -> Option<crate::Result<T>> {
        use futures::StreamExt;

        let stream = self.stream.as_mut()?;

        match stream.next().await {
            Some(Ok(row)) => {
                // 解析行数据
                match parse_mysql_row::<T>(&row) {
                    Ok(model) => Some(Ok(model)),
                    Err(e) => Some(Err(e)),
                }
            }
            Some(Err(e)) => Some(Err(crate::ormer_error!(
                "mysql_async::ResultSetStream::next failed: {e}"
            ))),
            None => None,
        }
    }
}

impl_backend_related_executor_methods_with_lifetime!(
    RelatedSelectExecutor,
    pool,
    &'a Pool,
    RelatedSelect
);

impl<'a, T: Model, R: Model> RelatedSelectExecutor<'a, T, R> {
    pub async fn collect<C: FromIterator<T>>(self) -> crate::Result<C> {
        let results = self.collect_inner().trace().await?;
        Ok(results.into_iter().collect())
    }

    pub(crate) fn into_collect_future(self) -> RelatedCollectFuture<'a, T, R> {
        RelatedCollectFuture { executor: self }
    }

    async fn collect_inner(self) -> crate::Result<Vec<T>> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mysql_params: Vec<mysql_async::Value> =
            params.iter().map(mysql_value_from_ormer_value).collect();

        let mut conn = self.pool.get_conn().trace().await?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        for row in rows {
            let mut data = HashMap::new();
            for (i, col_name) in T::COLUMNS.iter().enumerate() {
                let column_info = &T::COLUMN_SCHEMA[i];
                let rust_type = column_info.rust_type;
                let is_nullable = column_info.is_nullable;

                let ormer_value = if is_nullable {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i).unwrap_or(None);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => crate::model::Value::Null,
                                _ => crate::model::Value::Null,
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported nullable column type: {rust_type}"
                            )));
                        }
                    }
                } else {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected integer type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected String type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected float type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected bool type)",
                                        col_name
                                    )));
                                }
                                _ => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (invalid bool value)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported column type: {rust_type}"
                            )));
                        }
                    }
                };
                data.insert(col_name.to_string(), ormer_value);
            }
            let ormer_row = Row::new(data);
            let model = T::from_row(&ormer_row)?;
            results.push(model);
        }
        Ok(results)
    }
}

pub struct RelatedCollectFuture<'a, T: Model, R: Model> {
    executor: RelatedSelectExecutor<'a, T, R>,
}

impl<'a, T: Model + 'static + Send, R: Model + 'static + Send> std::future::IntoFuture
    for RelatedCollectFuture<'a, T, R>
{
    type Output = crate::Result<Vec<T>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

/// Multi-Table 查询执行器（支持3表关联查询）
pub struct MultiTableSelectExecutor<'a, T: Model, R1: Model, R2: Model> {
    select: MultiTableSelect<T, R1, R2>,
    pool: &'a Pool,
    _marker: PhantomData<(T, R1, R2)>,
}

impl_backend_multi_table_executor_methods_with_lifetime!(
    MultiTableSelectExecutor,
    pool,
    &'a Pool,
    MultiTableSelect
);

impl<'a, T: Model, R1: Model, R2: Model> MultiTableSelectExecutor<'a, T, R1, R2> {
    async fn collect_inner(self) -> crate::Result<Vec<T>> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mysql_params: Vec<mysql_async::Value> =
            params.iter().map(mysql_value_from_ormer_value).collect();

        let mut conn = self.pool.get_conn().trace().await?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        for row in rows {
            let mut data = HashMap::new();
            for (i, col_name) in T::COLUMNS.iter().enumerate() {
                let column_info = &T::COLUMN_SCHEMA[i];
                let rust_type = column_info.rust_type;
                let is_nullable = column_info.is_nullable;

                let ormer_value = if is_nullable {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i).unwrap_or(None);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => crate::model::Value::Null,
                                _ => crate::model::Value::Null,
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported nullable column type: {rust_type}"
                            )));
                        }
                    }
                } else {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected integer type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected String type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected float type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected bool type)",
                                        col_name
                                    )));
                                }
                                _ => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (invalid bool value)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported column type: {rust_type}"
                            )));
                        }
                    }
                };
                data.insert(col_name.to_string(), ormer_value);
            }
            let ormer_row = Row::new(data);
            let model = T::from_row(&ormer_row)?;
            results.push(model);
        }
        Ok(results)
    }
}

pub struct MultiTableCollectFuture<'a, T: Model, R1: Model, R2: Model> {
    executor: MultiTableSelectExecutor<'a, T, R1, R2>,
}

impl<'a, T: Model + 'static + Send, R1: Model + 'static + Send, R2: Model + 'static + Send>
    std::future::IntoFuture for MultiTableCollectFuture<'a, T, R1, R2>
{
    type Output = crate::Result<Vec<T>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

/// FourTable 查询执行器（支持4表关联查询）
pub struct FourTableSelectExecutor<'a, T: Model, R1: Model, R2: Model, R3: Model> {
    select: FourTableSelect<T, R1, R2, R3>,
    pool: &'a Pool,
    _marker: PhantomData<(T, R1, R2, R3)>,
}

impl_backend_four_table_executor_methods_with_lifetime!(
    FourTableSelectExecutor,
    pool,
    &'a Pool,
    FourTableSelect
);

impl<'a, T: Model, R1: Model, R2: Model, R3: Model> FourTableSelectExecutor<'a, T, R1, R2, R3> {
    async fn collect_inner(self) -> crate::Result<Vec<T>> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);
        let mut conn = self.pool.get_conn().trace().await?;

        let mysql_params: Vec<mysql_async::Value> =
            params.iter().map(mysql_value_from_ormer_value).collect();

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        for row in rows {
            let mut data = HashMap::new();
            for (i, col_name) in T::COLUMNS.iter().enumerate() {
                let column_info = &T::COLUMN_SCHEMA[i];
                let rust_type = column_info.rust_type;
                let is_nullable = column_info.is_nullable;

                let ormer_value = if is_nullable {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i).unwrap_or(None);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => crate::model::Value::Null,
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i).unwrap_or(None);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => crate::model::Value::Null,
                                _ => crate::model::Value::Null,
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported nullable column type: {rust_type}"
                            )));
                        }
                    }
                } else {
                    match rust_type {
                        "i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
                            let v: Option<i64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Integer(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected integer type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "String" => {
                            let v: Option<String> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Text(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected String type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "f32" | "f64" => {
                            let v: Option<f64> = row.get(i);
                            match v {
                                Some(val) => crate::model::Value::Real(val),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected float type)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        "bool" => {
                            let v: Option<i8> = row.get(i);
                            match v {
                                Some(1) => crate::model::Value::Integer(1),
                                Some(0) => crate::model::Value::Integer(0),
                                None => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (expected bool type)",
                                        col_name
                                    )));
                                }
                                _ => {
                                    return Err(crate::ormer_error!(format!(
                                        "Failed to parse non-nullable column '{}' (invalid bool value)",
                                        col_name
                                    )));
                                }
                            }
                        }
                        _ => {
                            return Err(crate::ormer_error!(format!(
                                "Unsupported column type: {rust_type}"
                            )));
                        }
                    }
                };
                data.insert(col_name.to_string(), ormer_value);
            }
            let ormer_row = Row::new(data);
            let model = T::from_row(&ormer_row)?;
            results.push(model);
        }
        Ok(results)
    }
}

pub struct FourTableCollectFuture<'a, T: Model, R1: Model, R2: Model, R3: Model> {
    executor: FourTableSelectExecutor<'a, T, R1, R2, R3>,
}

impl<
    'a,
    T: Model + 'static + Send,
    R1: Model + 'static + Send,
    R2: Model + 'static + Send,
    R3: Model + 'static + Send,
> std::future::IntoFuture for FourTableCollectFuture<'a, T, R1, R2, R3>
{
    type Output = crate::Result<Vec<T>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

impl_backend_join_executor_methods_with_lifetime!(
    LeftJoinedSelectExecutor,
    pool,
    &'a Pool,
    LeftJoinedSelect
);
impl_backend_join_executor_methods_with_lifetime!(
    InnerJoinedSelectExecutor,
    pool,
    &'a Pool,
    InnerJoinedSelect
);
impl_backend_join_executor_methods_with_lifetime!(
    RightJoinedSelectExecutor,
    pool,
    &'a Pool,
    RightJoinedSelect
);

/// LeftJoinedSelectExecutor 实现
impl<'a, T: Model, J: Model> LeftJoinedSelectExecutor<'a, T, J> {
    /// 克隆executor（保持相同的pool引用）
    pub fn clone_with_pool(&self) -> Self {
        Self {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    pub fn collect<C: FromIterator<(T, Option<J>)> + 'static>(
        &self,
    ) -> LeftJoinCollectFuture<'a, T, J> {
        LeftJoinCollectFuture {
            executor: self.clone_with_pool(),
            _marker: PhantomData,
        }
    }
}

/// LEFT JOIN Collect future
pub struct LeftJoinCollectFuture<'a, T: Model, J: Model> {
    executor: LeftJoinedSelectExecutor<'a, T, J>,
    _marker: PhantomData<(T, J)>,
}

impl<'a, T: Model + 'static + Send, J: Model + 'static + Send> std::future::IntoFuture
    for LeftJoinCollectFuture<'a, T, J>
{
    type Output = crate::Result<Vec<(T, Option<J>)>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

impl<'a, T: Model, J: Model> LeftJoinedSelectExecutor<'a, T, J> {
    async fn collect_inner<C: FromIterator<(T, Option<J>)>>(self) -> crate::Result<C> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mysql_params = values_to_params(&params)?;

        let mut conn = self.pool.get_conn().trace().await?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        let t_col_count = T::COLUMNS.len();

        for row in rows {
            let mut t_data = HashMap::new();
            for (i, col_name) in T::COLUMNS.iter().enumerate() {
                let rust_type = T::COLUMN_SCHEMA[i].rust_type;
                let ormer_value = match rust_type {
                    "i32" | "i64" | "u32" | "u64" => {
                        let v: i64 = row.get(i).unwrap_or(0);
                        crate::model::Value::Integer(v)
                    }
                    "String" => {
                        let v: String = row.get(i).unwrap_or(String::new());
                        crate::model::Value::Text(v)
                    }
                    "f32" | "f64" => {
                        let v: f64 = row.get(i).unwrap_or(0.0);
                        crate::model::Value::Real(v)
                    }
                    _ => {
                        return Err(crate::ormer_error!(format!(
                            "Unsupported column type: {rust_type}"
                        )));
                    }
                };
                t_data.insert(col_name.to_string(), ormer_value);
            }
            let t_model = T::from_row(&Row::new(t_data))?;

            // 尝试读取 J 的列
            let mut j_data = HashMap::new();
            let mut j_is_null = true;
            for (i, col_name) in J::COLUMNS.iter().enumerate() {
                let idx = t_col_count + i;
                let rust_type = J::COLUMN_SCHEMA[i].rust_type;
                let ormer_value = match rust_type {
                    "i32" | "i64" | "u32" | "u64" => {
                        // 先检查是否为 NULL
                        match row.get_opt::<i64, usize>(idx) {
                            None | Some(Err(_)) => crate::model::Value::Integer(0),
                            Some(Ok(v)) => {
                                if v != 0 {
                                    j_is_null = false;
                                }
                                crate::model::Value::Integer(v)
                            }
                        }
                    }
                    "String" => {
                        // 先检查是否为 NULL
                        match row.get_opt::<String, usize>(idx) {
                            None | Some(Err(_)) => crate::model::Value::Text(String::new()),
                            Some(Ok(v)) => {
                                if !v.is_empty() {
                                    j_is_null = false;
                                }
                                crate::model::Value::Text(v)
                            }
                        }
                    }
                    "f32" | "f64" => {
                        // 先检查是否为 NULL
                        match row.get_opt::<f64, usize>(idx) {
                            None | Some(Err(_)) => crate::model::Value::Real(0.0),
                            Some(Ok(v)) => {
                                if v != 0.0 {
                                    j_is_null = false;
                                }
                                crate::model::Value::Real(v)
                            }
                        }
                    }
                    _ => {
                        return Err(crate::ormer_error!(format!(
                            "Unsupported column type: {rust_type}"
                        )));
                    }
                };
                j_data.insert(col_name.to_string(), ormer_value);
            }

            if j_is_null {
                results.push((t_model, None));
            } else {
                let j_model = J::from_row(&Row::new(j_data))?;
                results.push((t_model, Some(j_model)));
            }
        }

        Ok(results.into_iter().collect())
    }
}

/// InnerJoinedSelectExecutor 实现
impl<'a, T: Model, J: Model> InnerJoinedSelectExecutor<'a, T, J> {
    /// 克隆executor（保持相同的pool引用）
    pub fn clone_with_pool(&self) -> Self {
        Self {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    pub fn collect<C: FromIterator<(T, J)> + 'static>(&self) -> InnerJoinCollectFuture<'a, T, J> {
        InnerJoinCollectFuture {
            executor: self.clone_with_pool(),
            _marker: PhantomData,
        }
    }
}

/// INNER JOIN Collect future
pub struct InnerJoinCollectFuture<'a, T: Model, J: Model> {
    executor: InnerJoinedSelectExecutor<'a, T, J>,
    _marker: PhantomData<(T, J)>,
}

impl<'a, T: Model + 'static + Send, J: Model + 'static + Send> std::future::IntoFuture
    for InnerJoinCollectFuture<'a, T, J>
{
    type Output = crate::Result<Vec<(T, J)>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

impl<'a, T: Model, J: Model> InnerJoinedSelectExecutor<'a, T, J> {
    async fn collect_inner<C: FromIterator<(T, J)>>(self) -> crate::Result<C> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mysql_params = values_to_params(&params)?;

        let mut conn = self.pool.get_conn().trace().await?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        let t_col_count = T::COLUMNS.len();

        for row in rows {
            let mut t_data = HashMap::new();
            for (i, col_name) in T::COLUMNS.iter().enumerate() {
                let rust_type = T::COLUMN_SCHEMA[i].rust_type;
                let ormer_value = match rust_type {
                    "i32" | "i64" | "u32" | "u64" => {
                        let v: i64 = row.get(i).unwrap_or(0);
                        crate::model::Value::Integer(v)
                    }
                    "String" => {
                        let v: String = row.get(i).unwrap_or(String::new());
                        crate::model::Value::Text(v)
                    }
                    "f32" | "f64" => {
                        let v: f64 = row.get(i).unwrap_or(0.0);
                        crate::model::Value::Real(v)
                    }
                    _ => {
                        return Err(crate::ormer_error!(format!(
                            "Unsupported column type: {rust_type}"
                        )));
                    }
                };
                t_data.insert(col_name.to_string(), ormer_value);
            }
            let t_model = T::from_row(&Row::new(t_data))?;

            let mut j_data = HashMap::new();
            for (i, col_name) in J::COLUMNS.iter().enumerate() {
                let idx = t_col_count + i;
                let rust_type = J::COLUMN_SCHEMA[i].rust_type;
                let ormer_value = match rust_type {
                    "i32" | "i64" | "u32" | "u64" => {
                        let v: i64 = row.get(idx).unwrap_or(0);
                        crate::model::Value::Integer(v)
                    }
                    "String" => {
                        let v: String = row.get(idx).unwrap_or(String::new());
                        crate::model::Value::Text(v)
                    }
                    "f32" | "f64" => {
                        let v: f64 = row.get(idx).unwrap_or(0.0);
                        crate::model::Value::Real(v)
                    }
                    _ => {
                        return Err(crate::ormer_error!(format!(
                            "Unsupported column type: {rust_type}"
                        )));
                    }
                };
                j_data.insert(col_name.to_string(), ormer_value);
            }

            let j_model = J::from_row(&Row::new(j_data))?;
            results.push((t_model, j_model));
        }

        Ok(results.into_iter().collect())
    }
}

/// RightJoinedSelectExecutor 实现
impl<'a, T: Model, J: Model> RightJoinedSelectExecutor<'a, T, J> {
    /// 克隆executor（保持相同的pool引用）
    pub fn clone_with_pool(&self) -> Self {
        Self {
            select: self.select.clone(),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    pub fn collect<C: FromIterator<(Option<T>, J)> + 'static>(
        &self,
    ) -> RightJoinCollectFuture<'a, T, J> {
        RightJoinCollectFuture {
            executor: self.clone_with_pool(),
            _marker: PhantomData,
        }
    }
}

/// RIGHT JOIN Collect future
pub struct RightJoinCollectFuture<'a, T: Model, J: Model> {
    executor: RightJoinedSelectExecutor<'a, T, J>,
    _marker: PhantomData<(T, J)>,
}

/// Grouped Collect future（分组聚合查询）
pub struct GroupedCollectFuture<'a, T: Model, V, C> {
    executor: GroupedSelectExecutor<'a, T, V>,
    _marker: PhantomData<(T, V, C)>,
}

impl<'a, T: Model + 'static + Send, J: Model + 'static + Send> std::future::IntoFuture
    for RightJoinCollectFuture<'a, T, J>
{
    type Output = crate::Result<Vec<(Option<T>, J)>>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move { self.executor.collect_inner().await })
    }
}

impl<'a, T: Model, J: Model> RightJoinedSelectExecutor<'a, T, J> {
    async fn collect_inner<C: FromIterator<(Option<T>, J)>>(self) -> crate::Result<C> {
        let (sql, params) = self.select.to_sql_with_params(DbType::MySQL);

        let mysql_params = values_to_params(&params)?;

        let mut conn = self.pool.get_conn().trace().await?;

        let rows: Vec<mysql_async::Row> = conn.exec(&sql, mysql_params).trace().await?;

        let mut results = Vec::new();
        let t_col_count = T::COLUMNS.len();

        for row in rows {
            let t_model =
                common_helpers::decode_optional_model_from_indexed_values::<T, _>(0, |i| {
                    convert_mysql_value(&row, i)
                })?;
            let j_model =
                common_helpers::decode_model_from_indexed_values::<J, _>(t_col_count, |i| {
                    convert_mysql_value(&row, i)
                })?;
            results.push((t_model, j_model));
        }

        Ok(results.into_iter().collect())
    }
}

/// 将 MySQL 行中的值转换为 ormer Value
fn convert_mysql_value(row: &mysql_async::Row, index: usize) -> crate::Result<crate::model::Value> {
    use mysql_async::Value;
    use mysql_async::consts::ColumnType;

    // 获取原始值
    let value = row.get::<Option<Value>, _>(index).unwrap_or(None);

    // 检查列类型是否为二进制类型（BLOB / TINYBLOB / MEDIUMBLOB / LONGBLOB / BINARY / VARBINARY）
    let is_binary_col = row.columns().get(index).is_some_and(|col| {
        matches!(
            col.column_type(),
            ColumnType::MYSQL_TYPE_TINY_BLOB
                | ColumnType::MYSQL_TYPE_BLOB
                | ColumnType::MYSQL_TYPE_MEDIUM_BLOB
                | ColumnType::MYSQL_TYPE_LONG_BLOB
                | ColumnType::MYSQL_TYPE_STRING
                | ColumnType::MYSQL_TYPE_VAR_STRING
        ) && col.character_set() == 63 // 63 = binary charset
    });

    match value {
        Some(Value::NULL) | None => Ok(crate::model::Value::Null),
        Some(Value::Int(i)) => Ok(crate::model::Value::Integer(i)),
        Some(Value::UInt(u)) => Ok(crate::model::Value::Integer(u as i64)),
        Some(Value::Float(f)) => Ok(crate::model::Value::Real(f as f64)),
        Some(Value::Double(d)) => Ok(crate::model::Value::Real(d)),
        Some(Value::Date(year, month, day, hour, minute, second, micros)) => {
            let date = chrono::NaiveDate::from_ymd_opt(year as i32, month as u32, day as u32)
                .ok_or_else(|| {
                    crate::ormer_error!("Invalid MySQL DATE value at index {}", index)
                })?;

            if hour == 0 && minute == 0 && second == 0 && micros == 0 {
                Ok(crate::model::Value::Date(date))
            } else {
                let datetime = date
                    .and_hms_micro_opt(hour as u32, minute as u32, second as u32, micros)
                    .ok_or_else(|| {
                        crate::ormer_error!("Invalid MySQL DATETIME value at index {}", index)
                    })?;
                Ok(crate::model::Value::DateTime(
                    chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(
                        datetime,
                        chrono::Utc,
                    ),
                ))
            }
        }
        Some(Value::Time(negative, days, hours, minutes, seconds, micros)) => {
            if !negative && days == 0 {
                let time = chrono::NaiveTime::from_hms_micro_opt(
                    hours as u32,
                    minutes as u32,
                    seconds as u32,
                    micros,
                )
                .ok_or_else(|| {
                    crate::ormer_error!("Invalid MySQL TIME value at index {}", index)
                })?;
                Ok(crate::model::Value::Time(time))
            } else {
                let total_hours = days.saturating_mul(24).saturating_add(hours as u32);
                let sign = if negative { "-" } else { "" };
                let value = if micros == 0 {
                    format!("{sign}{total_hours:02}:{minutes:02}:{seconds:02}")
                } else {
                    format!("{sign}{total_hours:02}:{minutes:02}:{seconds:02}.{micros:06}")
                };
                Ok(crate::model::Value::Text(value))
            }
        }
        Some(Value::Bytes(b)) if is_binary_col => {
            // 二进制列（BLOB/BINARY）直接作为 Bytes 返回
            Ok(crate::model::Value::Bytes(b))
        }
        Some(Value::Bytes(b)) => {
            // 文本列：尝试将字节解析为数值（MySQL聚合函数可能返回字符串形式的数值）
            if let Ok(s) = String::from_utf8(b.clone()) {
                if let Ok(i) = s.parse::<i64>() {
                    Ok(crate::model::Value::Integer(i))
                } else if let Ok(f) = s.parse::<f64>() {
                    Ok(crate::model::Value::Real(f))
                } else {
                    Ok(crate::model::Value::Text(s))
                }
            } else {
                // 非 UTF-8 字节，作为二进制数据处理
                Ok(crate::model::Value::Bytes(b))
            }
        }
    }
}

impl<'a, T: Model, V> GroupedSelectExecutor<'a, T, V> {
    /// 执行查询并收集结果
    pub fn collect<C: FromIterator<V> + 'static>(&self) -> GroupedCollectFuture<'a, T, V, C>
    where
        T: 'static,
        V: crate::model::FromRowValues + 'static,
    {
        GroupedCollectFuture {
            executor: GroupedSelectExecutor {
                select: self.select.clone(),
                pool: self.pool,
                _marker: PhantomData,
            },
            _marker: PhantomData,
        }
    }

    /// 添加 GROUP BY 字段
    pub fn group_by<F, G>(self, f: F) -> Self
    where
        F: FnOnce(<T as Model>::Where) -> G,
        G: crate::query::builder::GroupByColumns,
    {
        Self {
            select: self.select.group_by(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加 HAVING 条件
    pub fn having<F>(self, f: F) -> Self
    where
        F: FnOnce(<T as Model>::Where) -> crate::query::builder::WhereExpr,
    {
        Self {
            select: self.select.having(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }

    /// 添加 WHERE 条件（分组前过滤）
    pub fn filter<F>(self, f: F) -> Self
    where
        F: FnOnce(T::Where) -> crate::query::builder::WhereExpr,
    {
        Self {
            select: self.select.filter(f),
            pool: self.pool,
            _marker: PhantomData,
        }
    }
}

impl<
    'a,
    T: Model + 'static + Send,
    V: crate::model::FromRowValues + 'static + Send,
    C: FromIterator<V> + 'static,
> std::future::IntoFuture for GroupedCollectFuture<'a, T, V, C>
{
    type Output = crate::Result<C>;
    type IntoFuture =
        std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + Send + 'a>>;

    fn into_future(self) -> Self::IntoFuture {
        Box::pin(async move {
            let (sql, params) = self.executor.select.build_sql(DbType::MySQL);
            let mut conn = self.executor.pool.get_conn().trace().await?;

            // 将ormer::Value转换为mysql_async::Params
            let mysql_params: Vec<mysql_async::Value> = params
                .into_iter()
                .map(|v| mysql_value_from_ormer_value(&v))
                .collect();

            let rows: Vec<mysql_async::Row> = if mysql_params.is_empty() {
                conn.query(&sql).trace().await?
            } else {
                conn.exec(&sql, mysql_async::Params::Positional(mysql_params))
                    .trace()
                    .await?
            };

            let mut results = Vec::new();
            let column_count = self.executor.select.column_count();
            for row in rows {
                let v = common_helpers::decode_row_values_from_indexed_values(column_count, |i| {
                    convert_mysql_value(&row, i)
                })?;
                results.push(v);
            }

            Ok(results.into_iter().collect())
        })
    }
}